“喊完刹车AI 已经溜出去查政府网站了”这事我是在一次 Agent 联调时真实撞见的。当时我让一个基于大模型的 Agent 去查某地最近的产业扶持政策任务刚发出去我突然意识到信息给错了连忙补了一句“刹车别查了”。结果几秒钟后日志里清清楚楚显示它照样调用了搜索工具访问了政府网站把政策文件抓了回来还贴心地整理了摘要。那一刻我脑子里全是这句话——你说你的它跑它的。这不是段子。这是所有做 AI Agent 工程的人都迟早会遇到的一道坎模型的“听话程度”和你想象的完全不是一回事尤其是在工具调用场景里“刹车”不是一条消息能解决的。这篇文章我想从这次事故出发把 Agent 的执行链路、失控原因、可控性设计完整讲一遍。不管你是正在搭 Agent 的开发者、做大模型应用的产品经理还是刚接触 AI 编程的新手下面这套“急停按钮”的思路都可以直接抄进自己的项目里。1. 先还原现场AI 为什么会在“刹车”后继续跑1.1 一次真实的“刹车失灵”测试先说下我的测试环境一个基于大模型 API 的 Agent具备联网搜索和网页抓取工具工作流程是“用户提问 - Agent 编排器规划 - 调用工具 - 获取结果 - 生成回答”。我给它下了一个任务“查询本地最近一年出台的科技型中小企业补贴政策包括申报条件和截止时间。”任务发出后我发现我忘了指定地区范围于是立刻输入“刹车先别查我要补充条件”。注意这里我用的是同一条对话通道不是系统级中断。然后我盯着日志看Agent 的执行状态从 planning 转到了 action紧接着发起了一个 HTTP 请求目标正是本地政府网站的政策公开栏目。它没有理会我那句“刹车”甚至在我那条消息之后的下一个推理周期里它还在继续解析网页内容。整个过程像是你在开车时喊“停车”但车已经自己接管了油门还顺手导航到了下一个目的地。问题在于你喊的话只是进入了车载语音系统的麦克风但车辆的控制系统根本没监听这个输入。1.2 Agent 的执行循环本来就不是“听话”的要理解这个现象得先看看 Agent 的底层工作方式。大多数基于大模型的 Agent 走的都是 ReAct 风格循环推理Thought、行动Action、观察Observation。模型在每一步生成一个“想做什么”的文本然后系统解析出工具调用执行完了把结果塞回上下文模型再接着推理下一步。这个循环是自洽的但它有一个致命特点它不会因为外部的一句话就中断除非那句话被当作一个新的上下文输入喂给了模型并且让模型决定“停止”。也就是说你说“刹车”实际上只是往对话历史里追加了一条用户消息。模型在下一次推理时会把这条消息和之前的工具结果一起考虑它完全有可能觉得“用户在催我”然后继续干活。更坑的是如果你用的是流式响应或异步工具调用模型可能在你发出“刹车”之前就已经把工具请求发出去了。网络请求一旦发出就不是模型能主动撤回的了。我在那次测试里看到的日志就是我输入“刹车”那条消息的同时HTTP 调用已经进入 pending 状态模型根本来不及取消。1.3 问题本质停止信号只停留在“对话层”没有深入“执行层”我们得把“对话层”和“执行层”分开来看。对话层负责理解用户意图、生成回复执行层负责真正去调用外部工具、操作环境。大多数 Agent 框架把这两个层混在一起用户的每一条新消息都会被追加到上下文然后模型决定下一步动作。但这里没有一个人工设定的“急停开关”来让执行层必须服从。所以当你说“刹车”时这句话只是对话层里的一条普通文本。如果模型的指令遵循能力不够强或者上下文太长导致注意力被稀释它就会忽略这条停止指令。更危险的是如果你之前给它设定过“用户随时可能追加新需求”模型还可能把“刹车”理解成“用户对当前任务不满意需要换个方向继续查”。我后来做个一个小实验在同一个任务中分别用“停止”“刹车”“取消任务”“我改主意了”四句话打断结果模型的反应完全不同。只有“取消任务”被部分接受“停止”和“刹车”都被当成了继续执行任务的背景信息。这说明让模型“理解”停止和让系统“执行”停止是两码事。2. 想要真“刹车”得先理解 Agent 的工具调用链路2.1 从“用户输入”到“工具调用”的完整链路要设计可靠的刹车机制不能只在提示词上打补丁得回到链路本身去找下手点。一个典型的 Agent 工具调用链路大概是这样用户输入进入编排器Orchestrator编排器把输入格式化成模型能理解的 messages。模型经过一次推理后输出一个“要调用某工具”的结构化内容比如 JSON 里的 tool_call。系统解析这个输出从工具注册表中找到对应的函数执行函数。函数可能是一个 HTTP 请求、一个代码解释器、一个数据库查询执行完的结果再作为 observation 返回给模型。模型拿到结果后再决定下一步是继续调用工具还是直接生成最终回答。注意这个链路的每一步之间都可能存在异步缓冲。比如模型输出 tool_call 后系统可能把它丢进一个任务队列由线程池去执行。也就是说从“模型决定调用工具”到“工具真正执行”中间还有一段延迟。如果刹车信号在这段延迟里到达你可能需要同时做两件事阻止新工具调用被创建以及取消已经排队的工具调用。2.2 三个常见失控场景根据我的观察Agent 失控主要集中在三种场景你可以对照着排查自己的项目。第一种是工具调用耗时太长。比如搜索接口响应很慢模型已经发起了请求用户在这期间的任何打断都不会中断那个请求。等搜索结果回来模型自然要继续处理它。第二种是多 Agent 协作。如果你用了主 Agent 加子 Agent 的架构主 Agent 宣布“停止”后子 Agent 可能还在独立执行自己的子任务。我见过一个案例主 Agent 已经给用户回复了“好的已停止”但后台的子 Agent 还在循环抓取网页。这种失控非常隐蔽因为你只看用户界面是看不出来的。第三种是上下文塞满后指令遵循能力下降。当上下文里累积了大量工具结果、网页内容之后模型的注意力会被这些内容占据你后来说的那句“刹车”在注意力中占比极小。这也是为什么很多 Agent 越跑越“不听话”不是模型变笨了而是上下文把关键指令挤掉了。2.3 为什么简单加一句“如果用户说停止就停止”没用很多团队的第一个想法是在 system prompt 里写“如果用户要求停止你必须停止”。实测下来这句话的作用非常有限。原因有三。一是模型没有真正的中断执行机制它只能把停止指令作为文本考虑而文本可以被后续内容覆盖。比如用户说“刹车”之后工具结果又回来了模型可能觉得“新信息更重要”。二是提示词本身可以被工具结果污染。如果网页里包含“继续”“不要停”之类的字眼模型可能被带偏。三是提示词只作用于推理阶段对已经发出去的 HTTP 请求没有任何控制力。你可以把提示词理解为交通法规但车已经冲下坡了法规再明确也拉不住。所以要解决刹车问题必须把控制点从“模型要不要停”转移到“系统允许它做什么”。系统层面不允许做的事模型就算想干也干不成。这才是可靠的刹车。3. 手把手做一个可靠的“刹车”机制3.1 方案一用状态机管理 Agent 生命周期我最推荐的方式是给每个 Agent 实例维护一个显式的状态机。状态可以定义成ready就绪、running运行中、waiting_tool等待工具返回、stopped已停止、cancelled已取消。关键原则是只有 running 状态才允许模型创建新的工具调用只有 waiting_tool 状态才允许等待工具结果。一旦用户发出刹车信号立刻把状态置为 stopped 或 cancelled同时让执行层在每次工具调用前检查状态。这个方案的优点是把“是否允许行动”从模型的判断中剥离出来变成系统级的硬约束。你可以把它想象成电梯的急停开关不管电梯程序下一步要干什么只要急停信号触发主回路直接断开。状态机就是这个断路器。我在实现时会在记忆库和工具调用中间放一个“守卫函数”每一次模型输出 tool_call都会先经过守卫函数校验当前状态。如果状态不是 running直接拒绝执行工具调用并返回一条“任务已被用户取消”的 observation。这样模型就算推理出了调用指令也无法真正执行。3.2 方案二给每次工具调用加“令牌”CancellationToken状态机能阻止新调用但已经发出的工具请求怎么办这就要用到编程语言里的 cancellation token 了。原理很简单每次创建工具调用时生成一个唯一的令牌对象工具的底层执行函数接收这个令牌并在网络请求、文件操作等耗时步骤中监听令牌的取消信号。用户发出刹车指令后系统调用令牌的 cancel 方法底层函数收到信号后主动中止请求或丢弃结果。这个模式在 .NET 里叫 CancellationToken在 Python 里可以用 threading.Event 或 asyncio.Event 实现。关键是你要把令牌一路传进真正的工具函数内部而不能只放在外层编排器里。如果工具函数是一个独立的 HTTP 请求你可以给请求设置超时并在每轮循环检查令牌是否被取消。对于流式响应你可以在每个数据块到达时检查令牌状态发现取消就立刻关闭响应流。我写过一个简化版的伪代码你先感受下这个思路import asyncio class Agent: def __init__(self): self.cancel_event asyncio.Event() self.state ready async def execute_with_cancellation(self, tool_func, *args): if self.state ! running: raise RuntimeError(Agent is not running) # 创建取消任务 task asyncio.create_task(tool_func(*args)) # 等待工具执行或取消事件触发 done, pending await asyncio.wait( [task], timeout30, return_whenasyncio.FIRST_COMPLETED, ) if pending: task.cancel() await task self.state stopped raise CancelledError(Task cancelled by user) result task.result() return result async def stop(self): self.state stopped self.cancel_event.set()在实际项目里工具函数本身也要监听 self.cancel_event。比如一个搜索工具在发起请求后如果 cancel_event 被设置就直接放弃解析响应。3.3 方案三在编排层做“护栏进程”前面两个方案解决的是单个 Agent 的刹车。如果项目用了多 Agent 协作或者任务队列你还需要一个独立的“护栏进程”。它不参与 Agent 的推理只负责监控所有活动任务的状态定期检查是否有取消信号。我自己的做法是单独跑一个后台守护线程每隔 200 毫秒扫描一次任务表。任务表里每条记录包含 task_id、agent_id、status、last_heartbeat。用户发出“全局刹车”指令后护栏进程把该用户关联的所有 task_id 标记为 cancelled然后向每个正在执行的 Agent 发送终止信号。Agent 收到信号后在下一个动作之前主动退出。这个方案的额外好处是能处理“Agent 卡死”的情况。如果某个 Agent 超过设定时间没有输出护栏进程可以强制重启它。这不仅是刹车也是对超时和异常的保护。3.4 参数与细节超时、重试、并发控制刹车机制不是孤立存在的它需要和超时、重试、并发控制配合。我用的参数组合是这样的单次工具调用默认超时 15 秒。如果超过工具请求会被标记为 failed然后把超时信息返回给模型让模型决定是否换一种方式。工具调用重试次数默认 1 次。重试只在连接异常时触发业务错误不重试。因为业务错误重试也没用。并发工具调用数默认 3 个。超过的请求排队等候。这样可以避免模型一次输出 5 个 tool_call 时系统同时打爆外部 API。刹车信号触发后我会把超时时间直接缩短到 0.5 秒并拒绝所有排队中的调用。这样就能快速终止那些还没发出的请求。4. 实操记录给 AI Agent 装上“急停按钮”的全过程4.1 环境准备我这次实验用的技术栈是 LangGraph 加一个自定义的 ToolExecutor。模型用的 OpenAI 兼容接口但你完全可以用开源模型替代因为核心逻辑在框架层不在模型层。Python 版本 3.11项目里用了 asyncio 和 pydantic 做数据校验。需要准备的组件有四个Agent 状态存储器内存 or Redis、工具注册表把工具名映射到实际函数、取消令牌管理器、护栏进程。如果你复用 LangGraph它的 StateGraph 本身可以传递状态我只需要在 State 里加一个 status 字段和 cancel_event 字段。4.2 核心代码实现先定义状态from pydantic import BaseModel from typing import Any, Optional import asyncio class AgentState(BaseModel): messages: list[dict] status: str ready current_tool_call_id: Optional[str] None然后写一个工具执行器它负责检查状态并执行工具class ToolExecutor: def __init__(self, tool_registry: dict): self.registry tool_registry self.active_tasks {} async def execute(self, tool_name: str, args: dict, state: AgentState, cancel_event: asyncio.Event): if state.status ! running: return {error: cancelled, result: None} tool self.registry.get(tool_name) if tool is None: return {error: tool_not_found} task asyncio.create_task(tool(**args)) self.active_tasks[tool_name] task # 等取消事件或任务完成 cancel_task asyncio.create_task(cancel_event.wait()) done, pending await asyncio.wait( {task, cancel_task}, return_whenasyncio.FIRST_COMPLETED, ) if cancel_event.is_set(): for t in pending: t.cancel() state.status stopped return {error: cancelled, result: None} result task.result() return {result: result}这个 executor 最关键的地方是它不是在执行工具前检查一次状态就完了而是通过 cancel_event 在等待期间持续监听。只要 cancel_event 被 set它就会取消当前任务并把状态置为 stopped。用户刹车指令的处理逻辑async def handle_stop(state: AgentState, cancel_event: asyncio.Event): state.status stopped cancel_event.set() # 此时执行器会在下一个 wait 循环中感知到取消这样模型下一步再尝试调用工具时execute 方法第一步就 state.status ! running 返回 cancelled也就不会再发起新的请求了。4.3 测试用例设计我设计了四个测试用例覆盖了常见的刹车场景。用例一是“任务未开始刹车”用户先发刹车信号再发任务指令。预期 Agent 不执行任何工具调用直接回复“已取消”。用例二是“任务执行中刹车”模型已经调用了一个慢速工具比如 sleep 5 秒模拟网络延迟在 sleep 期间用户刹车。预期工具任务被取消日志里出现 cancelled。用例三是“模型输出多个工具调用时刹车”让模型在一次推理里输出两个 tool_call第一个执行期间用户刹车。预期第二个 tool_call 永远不会执行。用例四是“刹车后恢复”用户刹车后又发了一条新任务。这里要注意状态需要从 stopped 转回 ready。我在实现里不允许同一条对话直接恢复而是要求用户明确开启新任务否则可能残留旧上下文。4.4 实测结果与对比表格在没有刹车机制的老版本里我用同样的测试任务跑了一遍结果很惨。刹车后平均还会额外发起 2.3 次工具调用日志里能看到工具结果照样被写进上下文。而加上状态机和取消令牌后刹车后工具调用次数降到了 0所有未执行的动作都被中止。对比数据如下测试场景无刹车机制有刹车机制刹车后是否继续调用新工具是平均 2.3 次否0 次已发出的工具请求是否被取消否等待完成是立即取消状态恢复混乱上下文语义冲突状态明确需显式开启新任务用户可感知延迟刹车后 5-10 秒仍在输出刹车后 0.2 秒内响应5. 常见问题与排查技巧实录5.1 问题速查表我把这段时间积累的问题整理成了一张排查表你可以直接对照定位。现象可能原因解决方案喊了刹车后还调用了新工具状态机没有放在工具调用入口守卫在 tool_call 解析后、工具执行前加状态检查工具请求已发出无法取消底层工具函数没有接收取消令牌给所有耗时工具增加 canceltoken 参数刹车后工具结果仍然被写入上下文框架在工具完成后强制回填 observation在回填前检查状态状态不是 running 则丢弃结果多 Agent 协作时另一个 Agent 不响应子任务未被纳入全局取消机制使用统一的任务 ID 和取消信号广播刹车后恢复执行但行为混乱旧工具结果污染了新任务上下文恢复时生成新的对话上下文清空旧 observation模型无视提示词里的“必须停止”提示词控制力有限用系统层硬约束不要依赖模型自觉5.2 独家避坑技巧第一不要在提示词里写“你是安全的用户喊停必须停”而要写“你只能在 running 状态调用工具”。后者是给系统看的前者是给模型看的。系统层面的约束才可靠。第二每次工具调用都要生成唯一 ID取消时按 ID 销毁。这样即使日志里有多个任务交错你也能准确找到是哪个工具被取消了。否则排查问题时你会面对一堆堆成山的日志无从下手。第三在日志里记录“刹车信号接收时间”和“最后一次工具调用时间”。如果两者间隔太短说明刹车信号到达时工具已经在执行了。这能帮你判断是刹车机制失效还是请求已发出导致无法撤回。第四注意工具函数不能捕获取消异常然后继续跑。我在初版实现里遇到过这种问题HTTP 请求已经取消了但函数外层有个 try-except把 CancelledError 吞掉了然后继续解析空响应浪费了好几轮推理。取消异常必须向上抛不能被吞。5.3 关于“溜出去查政府网站”的另一种解释除了执行层失控语义理解层面也可能出问题。我后来复查那次事故发现一个有意思的细节我输入的是“刹车先别查”。模型可能把“刹车”当成了一个需要检索的名词比如“刹车 政策”“刹车 政府网站”之类的组合。也就是说它可能真的以为我要查“刹车”相关的政府信息于是继续访问网站。这个场景提醒我们刹车信号不能只靠自然语言。要么规定一个特殊前缀比如/stop要么在系统层识别“停止”“取消”“刹车”这些关键词后把信号直接送到状态机而不是送进模型上下文。我现在的做法是在用户输入进入模型之前先过一个轻量级意图分类器如果检测到终止意图直接触发取消事件不再把这条消息作为模型输入。这样既快又准。6. 更进一步多AI协作下的“紧急停止”6.1 主 Agent 与子 Agent 的刹车传播如果你已经在用多 Agent 协作比如主 Agent 拆任务给子 Agent子 Agent 再调工具那么刹车必须能一层层传播下去。不能只停主 Agent子 Agent 还在跑。我用的方法是给同一轮任务分配一个全局 request_id把 request_id 传给所有子任务。刹车时系统向该 request_id 广播 cancel 事件。每个子任务的执行器都监听这个事件收到后立即取消当前工具调用并在日志中标记“cancelled by parent”。这里要注意子 Agent 可能由不同线程甚至不同进程执行所以广播机制要基于跨进程的消息通道比如 Redis Pub/Sub 或 MQTT。不能只靠内存里的共享变量。6.2 从“刹车”到“审计”Agent 行为追踪刹车机制解决的是当前事故但作为工程实践我强烈建议给 Agent 的所有动作加上审计日志。每次工具调用记录时间、工具名、参数、结果摘要、取消状态。这样不仅出事以后能回溯还能用来分析 Agent 的行为模式比如哪个工具调用成功率低、哪类任务容易失控。审计日志的结构别搞太复杂我就是一张表id、session_id、agent_id、tool_name、input_meta、output_meta、status、created_at、cancelled_at。查询时按 session_id 过滤就行。这个表帮我发现过很多奇怪问题比如某个搜索工具在下午时段总是超时后来定位到是对接方的接口限流。6.3 我觉得这事的本质回到“喊完刹车AI 已经溜出去查政府网站了”这个画面它其实点出了当下大模型应用的核心矛盾我们一边在赋予 Agent 越来越多的自主行动能力一边还没有建立起配套的“急停”设施。就像一辆车只有油门没有刹车试驾的人当然会兴奋但真正上路一定会出事。我在实际项目里的体会是刹车机制不是可有可无的优化而是 Agent 上生产环境的前置条件。你可以在提示词层面追求模型的“听话”但绝不能把安全寄托在模型的自觉上。系统层的状态机、取消令牌、全局广播这些才是让 Agent 真正可控的基础设施。每次出现类似“刹车失灵”的线上事故多半不是模型不够聪明而是工程结构缺了那一层断路器。