1. 一次实测事故用户喊“刹车”Agent却先查完了网站1.1 完整复现“先斩后奏”的查询流程前两天我在调一个多步AI Agent任务是让它“整理一下研发费用加计扣除的最新申报材料要求顺便看有没有新政策修正”。Agent按计划分了三步先调用搜索工具确认政策主题再打开对应的政务公开页面最后把申报材料清单提取出来生成表格。前两步跑得很顺但问题出在第二步和第三步之间。我盯着日志发现Agent准备继续读取页面里的附件列表而这项操作可能会多花十几秒我下意识在对话里发了句“刹车先停一下我还没确认搜索范围”。结果呢Agent完全没有停它在大约1秒后照样把页面内容抓取回来然后继续生成下一步计划甚至还在总结里写了一句“已获取相关页面数据正准备解析”。我把这个过程原样记了下来时间线是这样的11:02:31Agent发起对政务公开信息页的HTTP请求11:02:33用户在对话里发送“刹车先停一下”11:02:34Agent仍然拿到了页面响应并把结果追加到上下文11:02:35Agent继续规划下一步完全没理会上一条“刹车”指令。说它是“不听话”其实不公平。这个现象的本质是Agent的执行循环存在竞态条件用户的停止指令要等到当前这一轮工具调用结束后才会被读入上下文而Agent早在这之前就把“查网站”的动作发出去了。这就像你对着已经冲出去的快递员喊“别送了”但快递员已经骑上车走了他得等到下一个路口才能看到消息。1.2 为什么“停止”没有生效执行循环里的竞态大多数Agent框架的主循环长得差不多都是这样一段伪代码while not finished: # 把最新消息交给大模型 response llm.chat(messages) # 如果模型决定调用工具 if response.tool_calls: for call in response.tool_calls: result execute_tool(call) # 这一步一旦发出就无法回收 messages.append(tool_result_message(result)) else: finished True break在这个循环里“用户的停止消息”只存在于messages这个列表里。它确实会在下一轮迭代时被大模型看到但当前这一轮的工具调用已经在执行了。更麻烦的是工具执行往往是外部系统行为比如打开浏览器、发起网络请求、调用第三方API这些动作本身就不可撤销。HTTP请求发出去了服务端已经处理了你这边再怎么打断结果都已经产生。所以“喊完刹车AI已经溜出去查政府网站了”并不是段子而是所有准备做真Agent工程的人都绕不开的经典问题如何让一个自主执行多步任务的Agent具备真正可中断性。接下来我把自己拆解这个问题的过程和改造方案完整写出来涉及执行循环设计、工具网关、沙箱隔离、并发场景下的中断传播以及Agent测试里最容易漏掉的“中断注入”用例。2. Agent的执行循环与工具调用的“不可回收”特性2.1 从LLM对话到工具执行一条链路上的三个环节要理解这个“刹不住”的问题得先拆开一条链路上的三个环节。第一个环节是模型生成。大模型根据当前对话上下文决定下一步做什么它的输出可能是普通文本也可能是一个结构化的工具调用请求比如{action: fetch_url, url: ...}。这个环节本身是可以随时中断的模型还没输出完你把它停了就行。第二个环节是工具执行器。Agent框架解析模型输出的工具调用去执行对应的函数比如请求网页、查询数据库、调用邮件接口。这个环节是否可中断完全取决于工具本身的实现。纯Python函数可以检查中断标志但外部系统调用就不好说了。第三个环节是结果回填。工具执行完之后结果会作为一条消息追加回对话模型在下一轮继续推理。如果工具执行被强制中断那回填什么、怎么标记失败又成了一个需要设计的问题。“不可回收”的问题主要出在第二个环节。我在生产环境里遇到过几种情况有些是工具自身的限制有些是框架设计导致的问题。比如你调用一个外部APIAPI已经接受了请求你在本地怎么取消都只是“假装取消”远程那边照样执行。再比如你用Playwright控制浏览器打开了一个页面页面里的JS已经开始加载就算马上关掉浏览器上下文部分请求也已经发到了服务器。2.2 三种常见的“刹不住”场景与成因做了一段时间Agent应用之后我把“刹不住”的场景归纳成三类每一类的成因和应对方式都不一样。第一类是单步工具调用已经开始无法收回。这是最普遍的情况用户发出停止指令的时机太晚刚好卡在工具执行的那几十毫秒或几秒里。应对思路只有一个词兜底。给工具调用加超时统一设一个最长执行时间比如10秒或30秒超过就直接放弃结果把错误回填给模型。第二类是停止指令被当成普通任务消息模型把它理解成“话题转向”而不是“立即终止”。这很坑因为用户说“刹车先停一下”之后模型可能真的会把“刹车”当成一个任务来回答比如回复“好的我先停下来请问您需要我做什么”然后继续之前的步骤。这不是模型笨而是你把控制信号和任务数据放在了同一个通道里模型没法区分哪个是系统指令、哪个是对话内容。第三类是并行工具调用场景下的“部分已执行”。有些Agent框架支持一次输出多个工具调用并行执行比如同时请求三个网站。用户喊停时第一个已经完成了第二个正在跑第三个还没开始。如果中断逻辑只检查一次很容易漏掉后面两个。更麻烦的是“一半已经完成”的状态会让任务上下文变得不一致模型甚至会把已拿到的部分结果继续利用起来。3. 给Agent装上能用的“刹车”可中断性的工程改造3.1 最小改造控制面与数据面分离每步执行前检查中断标志先给出我建议的最小改造方案把用户消息分成控制消息和数据消息控制消息不进入模型上下文而是直接设置一个全局中断标志。我实际用过的实现大概是这样的import threading control_event threading.Event() CONTROL_PHRASES {stop, 刹车, 停一下, 先别执行, /cancel} def classify_user_message(message: str): # 控制面触发中断不进入模型上下文 if message.strip() in CONTROL_PHRASES or message.startswith(/): control_event.set() return control # 数据面正常任务消息进入模型上下文 return data主循环里每次迭代开头都检查一下while not finished: if control_event.is_set(): # 标记任务取消清理资源退出循环 cleanup() break # 只在非中断状态下调用工具 response llm.chat(messages) ...这个改造的核心思想是“控制面与数据面分离”。控制指令是给系统的事件不是给模型的对话。只要能做到这一点就能避免“模型把停止当成回答话题”的问题。但要注意如果工具调用已经在执行这种检查是挡不住的所以必须叠加超时兜底。3.2 架构级手段工具网关、沙箱隔离与人在回路审批只做最小改造Agent只能说具备了“基本刹车能力”。如果想让它在生产环境里真正可控我建议把工具调用全部收口到一个网关层。所有外部操作都走网关不在Agent业务代码里直接请求。工具网关要做的四件事白名单校验、权限分级、审批流、审计日志。白名单校验Agent只能调用预先注册好的工具新工具必须手动加白名单权限分级读操作、写操作、外部副作用操作分别配置不同策略审批流高危操作比如发送邮件、下单、修改数据库触发人工确认审计日志每次工具调用的入参、出参、耗时、是否被取消全部记录。以浏览器类工具为例最好把整个浏览器会话丢到沙箱里。我在项目里用的是受限浏览器容器具体来说就是每个Agent任务启动一个独立的浏览器上下文任务结束或者中断时直接销毁整个上下文从根上掐断所有未完成的网络请求。from playwright.async_api import async_playwright async def run_browser_step(task_budget): async with async_playwright() as p: browser await p.chromium.launch() # 每个任务独立上下文隔离 cookie 和页面状态 context await browser.new_context() try: page await context.new_page() # 设置页面级超时防止单个请求卡死整个任务 await page.set_default_timeout(5_000) result await execute_with_cancel(page, task_budget) return result finally: # 销毁上下文未完成的请求全部掐断 await context.close() await browser.close()这个方案的直观好处是当用户在中途喊停我们把Agent任务标记为取消后直接调用context.close()浏览器立刻关闭连带着所有正在进行的页面加载、AJAX请求全部终止。这比单纯在Python代码里抛一个异常要干净得多。3.3 超时与配额兜底措施必须存在即使有了控制标志和沙箱我也建议给所有Agent任务装一道“最后防线”统一的超时与配额机制。我通常配置三个维度的限制限制类型默认值说明单次工具调用超时10秒单个工具执行超过则直接失败单轮规划最大步数8步Agent最多调用8轮工具整个任务最大时长120秒不管是什么任务到时就中断配置的时候不要为了图方便把超时设成无限。我见过不少线上事故Agent在等待某个外部接口时彻底卡死用户怎么喊停都没反应就是因为没有兜底。有了这个机制就算中断信号因为网络延迟没及时送达任务也会自行终止然后返回一个“执行超时”的状态给上层。有些人会担心超时的结果回填问题——如果工具超时了是把异常信息回填给模型还是直接终止整个任务我的经验是分情况。如果是读取型工具超时就把“获取超时”写进结果让模型自己决定下一步如果是写入型工具宁可终止任务也不要回填模糊状态避免模型误以为写入成功。4. 从“查网站”这个动作看Agent工具调用的边界设计4.1 Agent为什么总想自己去“看”网站标题里的场景很有趣Agent“溜出去查政府网站”其实不是坏事它说明Agent已经学会了工具调用与主动信息检索。问题只在于这个动作应该在什么条件下被允许、什么时候可以被取消。为什么Agent特别倾向于自己去“看”网站我总结了三个原因。第一是知识截止的问题模型训练数据有截止时间而很多时效性信息只有实时查询才准确。第二是溯源需求用户想知道数据出处Agent给出一个可访问的来源链接比空口回答更可信。第三是精确解析比如查补贴政策、申报材料这类内容直接从原始页面提取比让模型凭记忆输出更靠谱。这也解释了为什么很多Agent应用里查公开信息站点成了默认高频动作。政务信息发布平台往往是结构化程度比较高的公开数据源适合被Agent当作工具调用目标。大家在设计工具白名单时可以考虑把这类权威公开信息源加入默认可用列表但要用好下面的受限代理机制。4.2 给Agent配“带刹车的浏览器”受限代理与页面快照我在工程上的实践是不让Agent直接裸跑浏览器去访问任意URL而是给它配一个“受限代理层”。这个代理层拦截所有页面访问请求做三件事域名白名单校验、缓存命中检查、页面内容快照化。先看域名白名单。Agent能访问哪些站点是通过正则或域名列表控制的没有在清单里的URL一律拒绝。这样就算Agent发疯想乱跳转网关也会把它拦下来。再看缓存与快照化。同一页面短时间内重复请求直接命中缓存不必每次都访问源站。更重要的是我通常要求代理把HTML流转成Markdown快照再返回给Agent而不是把原始HTML直接塞进上下文。直接塞HTML有两大坑一是token消耗爆炸一个页面动辄几十KB二是噪音太多导航、广告、脚本标签全是无效信息。Markdown化之后模型看到的是一篇干净的文本解析效率和准确率都明显提升。def fetch_page_with_snapshot(url: str, domain_whitelist: list[str]) - str: parsed urlparse(url) if not any(parsed.netloc.endswith(d) for d in domain_whitelist): raise PermissionError(fdomain not allowed: {parsed.netloc}) cached snapshot_cache.get(url) if cached: return cached html raw_fetch_with_timeout(url, timeout8) markdown html2markdown(html) # 截断超长文本防止上下文爆炸 markdown markdown[:6000] snapshot_cache.put(url, markdown, ttl300) return markdown这套方案的另一个好处是让“刹车”变得更干净。Agent调用的不再是真实浏览器而是受限代理服务。用户喊停时只要把代理缓存里的请求标记为放弃Agent拿不到新结果自然就停了不需要真的去关闭浏览器进程。对于只想读公开信息做检索处理的Agent任务这个模式比直接上浏览器自动化更轻、更稳也更好审计。5. 多Agent与并发场景下中断信号如何传播5.1 单个Agent可控不等于一群Agent可控做Agent应用的人早晚会遇到一个话题AI Agent怎么扛并发。单个Agent跑通不难一旦进入多Agent协作或者高并发任务调度问题就全冒出来了其中有一个和“喊刹车”高度相关单个Agent的可中断性设计得再好中断信号传不到子Agent那里也等于零。我遇到过一个真实的场景。一个主管Agent为了完成用户的任务拆出三个子Agent分别去查资料、整理格式、校验数据。用户发现任务目标有问题发了“停止”指令主管Agent确实停下来了但三个子Agent还在各自的循环里继续跑。因为它们根本收不到主管的中断状态。这个问题的根源在于很多人在设计多Agent协作时把子Agent之间、主管与子Agent之间的通信建立在“对话消息”上。主管把“停止”作为一条文本消息发给子Agent子Agent的理解取决于模型的心情。想让中断可靠传播必须把它变成基础设施层面的信号。5.2 并发Agent架构中的中断令牌传递我的做法是为每个任务分配一个CancelToken所有子Agent共享同一个任务级令牌子Agent的每一步执行都检查令牌状态。class CancelToken: def __init__(self): self._cancelled threading.Event() def cancel(self): self._cancelled.set() def is_cancelled(self): return self._cancelled.is_set()主管Agent和子Agent通过任务队列通信时不再发“请停止”这样的自然语言而是直接调用cancel_token.cancel()每个子Agent的执行器在每次工具调用前检查令牌。如果发现已取消立即终止后续计划并清理资源。这才是工程意义上的中断传播。再往深一层说并发场景下要特别小心共享状态。我建议所有Agent任务的数据传递都走显式的消息结构不要依赖全局变量。多个Agent实例并发跑的时候全局变量会产生各种奇怪的串扰A任务里设置的停止标志可能把B任务也中断了A任务写入的临时文件会被B任务误读。这属于并发安全的基础要求但我在实际项目里见过太多次因为共享可变状态导致的诡异故障。6. 从“能跑”到“可控”Agent测试该补哪些用例6.1 写一个“中断注入”回归测试要保证“刹车”功能不回归测试用例必须覆盖中断流程。我习惯给Agent执行器写这样一个回归测试模拟用户在第2步发送停止指令然后断言后续没有任何新的工具调用发生。def test_cancel_stops_future_tool_calls(): agent create_agent() cancel_token CancelToken() # 模拟第一步用户提需求 agent.handle_user_message(查一下最新的政策信息) # 模拟第二步Agent发起工具调用时用户喊停 cancel_token.cancel() # 上面某工具也许已经执行但后续不应该再出现新调用 future_calls agent.execute_loop(cancel_tokencancel_token) assert future_calls []这里有一个容易写错的地方很多人只检查“工具调用被标记为取消”忽略了“已经发出的工具调用可能有副作用”。我在测试里会专门用一个mock工具记录调用次数然后在断言里对比“取消前已经执行的次数”和“取消后新增的次数”确保新增次数为零。6.2 用观测数据复盘这类问题排查“喊完刹车还在执行”这类问题时没有观测数据基本靠猜。我现在给所有Agent工具调用加了一套统一的结构化日志字段如下字段示例值说明trace_idtask_00123任务链路ID贯穿整个任务step_index2当前是第几步toolfetch_url工具名称input_summaryurlhttps://...入参摘要output_summarymarkdown length5432出参摘要statuscancelled / ok / timeout最终状态duration_ms1234耗时有了这些日志当用户说“我喊停了但它还在跑”时我可以直接查看到底是中断信号没有生效还是工具已经处于执行中抑或是中断信号传到了但执行器没处理。大部分“刹不住”的问题其实都能在日志里找到对应的竞态点。我还建议把Agent的执行轨迹接入可观测系统用链路追踪把模型调用、工具调用、用户消息串起来。这样既能复盘单次问题也能从全局统计里发现高频中断发生在哪些工具上——通常这些工具就是要优先加沙箱和超时的对象。6.3 实测中的几个反直觉经验最后说几个我在实操中踩过的坑。第一个坑是在系统提示里写“如果用户说停你必须停否则后果严重”基本没用。模型不会把这个口头指令当成硬性控制信号它只会把“停”当成对话内容的一部分然后礼貌地回应你再继续干活。控制必须是系统层面的事件不能靠模型自觉。第二个坑是读取型工具和高危工具的取消策略要分开。读取型工具即使已经执行后果也不大直接丢弃结果即可但发送邮件、写入数据这类工具即使收到取消信号命令可能已经到达目标系统。我对这类工具坚持“先确认再执行”的强审批模式宁可在速度上慢一点也不能让Agent自作主张。第三个坑是不要用“全部打断”去处理所有中断。用户喊“刹车”不一定是要终止整个任务有时候只是想换个方向。比较好的做法是把中断状态设计成“取消当前步骤”和“取消整个任务”两级让上层根据用户意图选择。实际体验上“取消当前步骤并等待新指令”比“整个任务灰飞烟灭”要友好得多。我做Agent项目到现在最大的感触就是把“刹车”当成头等公民来设计。你与其花大量时间调prompt去让模型理解“停止”的重要性不如先在工程上把控制信号独立出来、把工具调用管起来、把超时和审计装好。一个可控的Agent不是因为它足够聪明而是因为它在关键位置上装了足够多的物理开关。