1. 一次实测事故:用户喊“刹车”,Agent却先查完了网站
1.1 完整复现:“先斩后奏”的查询流程
前两天我在调一个多步AI Agent,任务是让它“整理一下研发费用加计扣除的最新申报材料要求,顺便看有没有新政策修正”。Agent按计划分了三步:先调用搜索工具确认政策主题,再打开对应的政务公开页面,最后把申报材料清单提取出来生成表格。
前两步跑得很顺,但问题出在第二步和第三步之间。我盯着日志发现Agent准备继续读取页面里的附件列表,而这项操作可能会多花十几秒,我下意识在对话里发了句“刹车,先停一下,我还没确认搜索范围”。结果呢?Agent完全没有停,它在大约1秒后照样把页面内容抓取回来,然后继续生成下一步计划,甚至还在总结里写了一句“已获取相关页面数据,正准备解析”。
我把这个过程原样记了下来,时间线是这样的:
- 11:02:31,Agent发起对政务公开信息页的HTTP请求;
- 11:02:33,用户在对话里发送“刹车,先停一下”;
- 11:02:34,Agent仍然拿到了页面响应,并把结果追加到上下文;
- 11:02:35,Agent继续规划下一步,完全没理会上一条“刹车”指令。
说它是“不听话”其实不公平。这个现象的本质是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函数可以检查中断标志,但外部系统调用就不好说了。
第三个环节是结果回填。工具执行完之后,结果会作为一条消息追加回对话,模型在下一轮继续推理。如果工具执行被强制中断,那回填什么、怎么标记失败,又成了一个需要设计的问题。
“不可回收”的问题主要出在第二个环节。我在生产环境里遇到过几种情况,有些是工具自身的限制,有些是框架设计导致的问题。比如你调用一个外部API,API已经接受了请求,你在本地怎么取消都只是“假装取消”,远程那边照样执行。再比如你用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(f"domain not allowed: {parsed.netloc}") cached = snapshot_cache.get(url) if cached: return cached html = raw_fetch_with_timeout(url, timeout=8) markdown = html2markdown(html) # 截断超长文本,防止上下文爆炸 markdown = markdown[:6000] snapshot_cache.put(url, markdown, ttl=300) 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_token=cancel_token) assert future_calls == []这里有一个容易写错的地方:很多人只检查“工具调用被标记为取消”,忽略了“已经发出的工具调用可能有副作用”。我在测试里会专门用一个mock工具记录调用次数,然后在断言里对比“取消前已经执行的次数”和“取消后新增的次数”,确保新增次数为零。
6.2 用观测数据复盘这类问题
排查“喊完刹车还在执行”这类问题时,没有观测数据基本靠猜。我现在给所有Agent工具调用加了一套统一的结构化日志,字段如下:
| 字段 | 示例值 | 说明 |
|---|---|---|
| trace_id | task_00123 | 任务链路ID,贯穿整个任务 |
| step_index | 2 | 当前是第几步 |
| tool | fetch_url | 工具名称 |
| input_summary | url=https://... | 入参摘要 |
| output_summary | markdown length=5432 | 出参摘要 |
| status | cancelled / ok / timeout | 最终状态 |
| duration_ms | 1234 | 耗时 |
有了这些日志,当用户说“我喊停了但它还在跑”时,我可以直接查看到底是中断信号没有生效,还是工具已经处于执行中,抑或是中断信号传到了但执行器没处理。大部分“刹不住”的问题其实都能在日志里找到对应的竞态点。
我还建议把Agent的执行轨迹接入可观测系统,用链路追踪把模型调用、工具调用、用户消息串起来。这样既能复盘单次问题,也能从全局统计里发现高频中断发生在哪些工具上——通常这些工具就是要优先加沙箱和超时的对象。
6.3 实测中的几个反直觉经验
最后说几个我在实操中踩过的坑。
第一个坑是:在系统提示里写“如果用户说停,你必须停,否则后果严重”,基本没用。模型不会把这个口头指令当成硬性控制信号,它只会把“停”当成对话内容的一部分,然后礼貌地回应你,再继续干活。控制必须是系统层面的事件,不能靠模型自觉。
第二个坑是:读取型工具和高危工具的取消策略要分开。读取型工具即使已经执行,后果也不大,直接丢弃结果即可;但发送邮件、写入数据这类工具,即使收到取消信号,命令可能已经到达目标系统。我对这类工具坚持“先确认再执行”的强审批模式,宁可在速度上慢一点,也不能让Agent自作主张。
第三个坑是:不要用“全部打断”去处理所有中断。用户喊“刹车”不一定是要终止整个任务,有时候只是想换个方向。比较好的做法是把中断状态设计成“取消当前步骤”和“取消整个任务”两级,让上层根据用户意图选择。实际体验上,“取消当前步骤并等待新指令”比“整个任务灰飞烟灭”要友好得多。
我做Agent项目到现在,最大的感触就是把“刹车”当成头等公民来设计。你与其花大量时间调prompt去让模型理解“停止”的重要性,不如先在工程上把控制信号独立出来、把工具调用管起来、把超时和审计装好。一个可控的Agent,不是因为它足够聪明,而是因为它在关键位置上装了足够多的物理开关。