引言:一次失控的实验,账单翻了 40 倍
先讲一个真实场景。
某个内部知识库问答 Agent 上线灰度,任务描述是"回答用户关于产品文档的问题,必要时检索"。上线第二天,监控告警:某个单会话在 3 小时内消耗了 180 万 token,发起 2400 多次检索调用。日志拉出来一看,Agent 的轨迹是这样的:
Thought: 我需要找到关于"退款周期"的说明 Action: search_docs("退款周期") Observation: 未找到相关文档 Thought: 检索结果不理想,我换个关键词再试 Action: search_docs("退款 周期") Observation: 未找到相关文档 Thought: 还是没有,再换一个说法 Action: search_docs("退款政策 时间") ...它没有崩溃,没有报错,只是在"努力"。这正是 LLM Agent 最危险的一种失效模式:它不是不能工作,而是不知道自己已经做完了,或者做不完了。
把这次事故拆开算一笔账会更直观。按主流模型每百万 input token 约 3 美元、output token 约 15 美元计价,180 万 token 里即便九成是输入,单会话的直接成本也在 30 美元量级——而正常会话的成本是 0.1 美元以内,这就是"翻了 40 倍"的来源。这还没算那 2400 次检索带来的三方 API 账单、向量库的 QPS 挤占,以及对同租户其他用户造成的延迟抖动。换句话说,一次失控循环的成本是三重叠加的:token 账单、外部系统负载、用户体验。
传统软件的死循环会卡死 CPU,很快被发现。Agent 的循环是"付费"的——每一轮都要调用一次大模型,投入一次工具调用,代价是 token 账单、延迟和外部系统的副作用。更糟的是,很多工具(写数据库、发邮件、下单)不是幂等的,重复执行是事故而不只是浪费。
本文不谈"如何让 Agent 更聪明",只谈一件工程上更关键的事:如何让 Agent 可靠地停下来。我们会从 ReAct 循环的控制流本质出发,拆解循环的几种典型形态,然后给出一套可落地的执行器设计:预算控制、调用指纹去重、振荡检测、超时与取消传播,最后聊聊那些只有踩过才知道的坑。
无论你是用 LangChain、LlamaIndex、LangGraph,还是自己写了一个while True调 API,下面这套机制都是通用的——因为它约束的对象不是某个框架,而是"LLM 驱动的循环"这个结构本身。
一、Agent 循环的本质:为什么它停不下来
1.1 ReAct 的控制流是一台"没有退出条件"的状态机
主流 Agent 范式是 ReAct:模型输出Thought → Action,运行时执行工具得到Observation,再把 Observation 塞回上下文,进入下一轮。终止信号只有一个:模型自己决定输出 Final Answer。
问题在于,这个终止条件完全由模型的主观判断驱动,而模型有三个固有缺陷:
- 概率性:同样的上下文,两次采样可能给出不同决策,"要不要再试一次"本质是采样出来的,不是算出来的。
- 无跨轮显式状态:它看不到"我已经用同样参数调用过 5 次搜索"这个事实,除非运行时主动告诉它。
- 目标漂移:当检索失败时,模型倾向于"换个说法再试",这个行为在有限次数内是合理的探索,但它没有"探索已经失败"的判定能力。
用控制论的话说,Agent 是一个开环系统:执行器(LLM)不知道系统的总成本,也不知道自己是否在收敛。要在开环系统上做可靠性,必须外挂一个反馈控制器。
这里有一个认知上的关键转变:终止不是模型的职责,而是运行时的职责。模型负责的是"在给定信息下,下一步做什么"这个局部决策;而"这个循环还该不该继续"是全局决策,全局信息(已花多少 token、还剩多少步、外部系统是否在抖)根本不在模型的上下文里,它不可能做出正确判断。所以,把终止权交给模型,等于让一个蒙着眼睛的人判断车该不该停。正确的分工是:模型提议,运行时裁决。
1.2 无限循环的五种典型形态
实践中,失控循环并不是一种,而是五种,需要不同的检测手段:
| 形态 | 表现 | 检测手段 |
|---|---|---|
| 重复调用 | 同工具同参数连续调用 | 调用指纹去重 |
| 振荡 | A→B→A→B 交替 | 滑动窗口模式匹配 |
| 重试风暴 | 工具报错后高频重试 | 错误计数 + 指数退避 |
| 探索发散 | 参数不断变化但无进展 | 观察结果哈希/语义相似度 |
| 上下文退化 | 上下文溢出后模型输出混乱 | token 预算 + 强制收敛 |
关键洞察是:前四种都能用"轨迹特征"检测,不需要理解语义。这意味着我们可以在不依赖第二个模型的前提下,用确定性的代码拦住绝大多数失控。
稍微展开一下每种形态的现场感,方便你对号入座:
- 重复调用最典型,往往发生在工具有缓存但模型不知道的情况下。它以为每次都拿到了新结果,实际上拿到的是一模一样的旧结果,于是"再试一次"。
- 振荡通常是模型在两个候选路径之间反复横跳:先
search_docs觉得不够,get_page看了一页又觉得该换搜索词,换完又回到原来那页。它每一步都"有道理",整体却零进展。 - 重试风暴的触发点往往是限流错误或 5xx。模型的默认反应是立刻重试,而没有任何退避概念——它没有时间感,一个循环里 0.1 秒和 10 秒对它是等价的。
- 探索发散最隐蔽,因为轨迹看起来"很健康":每步都在变,都在调用工具,都拿到了结果。但把 observation 的哈希集合拿出来一看,全是一样的空结果。
- 上下文退化是终局形态,一旦发生,模型输出的工具名可能都是不存在的,此时任何检测逻辑都要防止它把畸形调用打到真实系统上。
1.3 成本是超线性的,不是线性的
容易被忽略的一点:Agent 每一轮的成本不是常数。ReAct 把全部历史对话重新喂给模型,第 N 轮的 prompt token 大致是前 N-1 轮的总和。也就是说,一个跑了 20 轮的 Agent,其总成本接近 O(N²) 而非 O(N)。这就是为什么"多跑几轮兜底"这种想法在成本上极其危险——你以为多花 20%,实际可能多花 200%。
把这句话量化一下。设第 k 轮的 prompt 为 P_k,历史全量回灌意味着:
P_k ≈ P_base + Σ_{i<k} (O_i + Obs_i)假设每轮的输出加观察长度稳定为 L,那么P_k ≈ P_base + (k-1)·L,K 轮的总输入成本就是:
Total ≈ Σ_{k=1..K} P_k ≈ K·P_base + L·K(K-1)/2 ≈ O(K²)代入一组实测数字:某 Agent 的 base prompt 是 1200 token,每轮 action 加 observation 约 800 token。那么第 1 轮 prompt 是 1200,第 10 轮是 8400,第 20 轮是 16400。跑满 20 轮的累计输入约 176K token;而如果第 5 轮就收敛,只需要约 14K。轮数翻 4 倍,成本涨了 12 倍以上,这个非线性放大就是"不知不觉账单爆炸"的数学原因。
还有一个反直觉的推论:长会话比长任务更危险。如果 Agent 每轮都带完整的多轮对话历史,那么第 10 轮时它已经为前面 9 轮的"无效探索"重复付费了 9 次。这也是为什么"上下文压缩"和"预算控制"必须一起做——前者治本,后者兜底。
二、控制机制的核心原理
2.1 三层防御:执行前、执行中、执行后
一套可靠的终止机制应该分层,任何单点都不足以兜底:
- 执行前(Pre-flight):预算初始化、任务复杂度评估、工具白名单。决定"这次允许花多少"。具体动作包括:根据任务类型选择预算档位(单轮问答 3 步、多跳检索 8 步、代码修复 15 步);把工具清单收敛到当前任务真正需要的子集,减少模型"手滑"调用昂贵工具的概率;对高风险工具(写库、支付、发信)打上
requires_confirmation标记,走单独闸门。 - 执行中(In-flight):步数上限、token 预算、调用指纹、振荡检测、超时、取消信号。这是主战场。
- 执行后(Post-flight):轨迹归档、成本归因、失败模式归类,用于迭代 prompt 和阈值。
三层的关系不是"任一命中即可",而是逐层收紧:Pre-flight 决定资源池有多大,In-flight 负责在池子见底前优雅收场,Post-flight 负责让下一版跑得更省。缺少任何一层,系统都会在某个特定场景下失守——只有执行中检测,你会缺少成本归因,永远不知道为什么循环;只有执行前预算,你拦不住"预算很大但全打水漂"的浪费。
2.2 终止信号必须显式化
不要指望模型自然输出一个漂亮的 Final Answer。要做两件事:
- 结构化终止:要求模型在最后一步返回特定字段(如
stop_reason: "final"),而不是靠解析自然语言里的"答案如下"。推荐用一个专门的终结工具,比如submit_answer(answer: str, confidence: float),让终止变成一个可校验的 schema,而不是一段需要正则匹配的散文。 - 强制收敛:当预算只剩最后一步时,在请求中注入约束(例如
tool_choice="none"或提示"这是最后一步,必须给出当前最优答案"),让模型优雅降级,而不是被硬截断——硬截断会得到一个半成品的 Observation:工具只返回了一半的 JSON,或者模型的 Action 正写到一半就被掐掉,最终落库的是一条残缺记录;如果这一步恰好是有副作用的写操作,你连判断它是否提交成功的依据都没有。
优雅降级的具体做法有三种,可以叠加使用:
- 软提示:在每轮请求里附加一个系统消息,告知"你已使用 X/Y 步,剩余预算 Z token"。实测表明,仅仅把剩余步数告诉模型,就能让相当比例的循环在第 6~8 步自行收敛,因为它终于有了"我还有多少余量"这个信息。
- 硬约束:在最后一轮把
tools=[]或tool_choice="none"传进去,物理上断掉工具调用的可能,逼迫模型基于已有 Observation 作答。 - 状态兜底:运行时不依赖模型的输出格式,自己维护一个
best_effort_answer——每次拿到"看起来有效"的 Observation 就更新一次,超限时直接返回它,保证用户永远能得到点什么,而不是一个报错。
三、落地实现:一个可控的 Agent 执行器
原理讲完,下面给出一套可以直接改吧改吧用在生产里的执行器。语言用 Python,但设计对任何语言都成立。核心思路是:把"是否继续"这个决策从模型手里拿走,收归到一个确定性的控制器。
3.1 执行器骨架
importhashlibimportjsonimporttimefromdataclassesimportdataclass,fieldfromtypingimportAny,Callable,Optional@dataclassclassBudget:"""一次 Agent 运行的全部资源约束,所有阈值都在这里集中管理。"""max_steps:int=12# 硬步数上限max_tokens:int=200_000# 输入+输出 token 总量max_seconds:float=120.0# 墙钟时间上限max_tool_calls:int=30# 工具调用总次数(含并行展开后的计数)steps:int=0tokens:int=0tool_calls:int=0started_at:float=field(default_factory=time.monotonic)defexhausted(self)->Optional[str]:"""返回命中的约束名,全部未命中则返回 None。"""ifself.steps>=self.max_steps:return"max_steps"ifself.tokens>=self.max_tokens:return"max_tokens"ifself.tool_calls>=self.max_tool_calls:return"max_tool_calls"iftime.monotonic()-self.started_at>=self.max_seconds:return"timeout"returnNoneclassAgentRunError(Exception):"""预算耗尽或检测到失控时抛出,携带部分结果供上层降级。"""def__init__(self,reason:str,partial:Any=None):super().__init__(reason)self.reason=reason self.partial=partial注意exhausted()返回的是原因字符串而不是布尔值。原因会一路传到日志和用户提示里——"因步数超限而提前结束"和"因检测到重复调用而终止"是完全不同的两件事,前者需要调大预算,后者需要改 prompt 或工具描述。可观测性从数据结构的形状就开始了。
3.2 调用指纹:把"重试"和"重复"区分开
重复调用是最好抓的一种,但前提是你要先会归一化。直接用原始参数做 hash 会被这些噪音绕过:参数顺序变化、query 前后多了空格、全角/半角混用、模型每次加一个无关的request_id。
_VOLATILE_KEYS={"request_id","trace_id","timestamp","nonce"}def_normalize(value:Any)->Any:ifisinstance(value,str):# 归一化:去首尾空格、压缩连续空白、统一大小写return" ".join(value.split()).lower()ifisinstance(value,dict):return{k:_normalize(v)fork,vinsorted(value.items())ifknotin_VOLATILE_KEYS}ifisinstance(value,(list,tuple)):return[_normalize(v)forvinvalue]returnvaluedeftool_fingerprint(name:str,args:dict)->str:"""稳定指纹:同工具 + 语义等价的参数 → 同一个字符串。"""payload=json.dumps({"name":name,"args":_normalize(args)},sort_keys=True,ensure_ascii=False,separators=(",",":"),)returnhashlib.sha256(payload.encode("utf-8")).hexdigest()[:16]有了指纹,去重逻辑要区分两种场景:
- 成功过的调用再次出现→ 直接拦截。把上一次的 Observation 原样回灌给模型,并附加一句"该调用已执行过,结果同上,请勿重复调用"。这比单纯拒绝更有效——模型看到结果仍在,往往会转向下一策略。
- 失败过的调用再次出现→ 允许有限次重试(默认 1 次),但必须走退避,并且第二次失败后要把该指纹加入黑名单,后续命中直接返回结构化错误。
这里有个容易踩的坑:不要无条件放行所有"重试"。如果工具返回的是"未找到"这种语义失败(HTTP 200 但业务无结果),它和网络超时完全不同——后者重试有意义,前者重试一千次也一样。判断标准是错误里有没有retryable标记,让工具实现方来声明,而不是让控制器猜。
3.3 振荡检测:滑动窗口找周期
振荡的检测思路是:把最近 N 步压成一个符号序列(工具名 + 参数指纹),如果这个序列存在一个最小周期 p,且至少重复了两个完整周期,就判定为振荡。
classOscillationDetector:def__init__(self,window:int=8,max_period:int=4,min_cycles:int=2):self.window=window self.max_period=max_period self.min_cycles=min_cycles self.history:list[str]=[]defobserve(self,symbol:str)->Optional[int]:"""喂入一个符号,返回检测到的周期长度,未检测到则返回 None。"""self.history.append(symbol)iflen(self.history)>self.window:self.history.pop(0)seq=self.history n=len(seq)forpinrange(1,self.max_period+1):ifn<p*self.min_cycles:continue# 检查最后 p*min_cycles 个元素是否严格周期为 ptail=seq[-p*self.min_cycles:]ifall(tail[i]==tail[i-p]foriinrange(p,len(tail))):returnpreturnNoneA→B→A→B会在第 4 步命中 p=2;A→B→C→A→B→C在第 6 步命中 p=3。检测到振荡后的处理不应是简单抛错,而是注入打断:在下一轮请求里显式告诉模型"你正在重复 A 和 B 之间的切换,二者都未产生新信息,请换一个完全不同的策略,或直接给出当前最优答案"。绝大多数情况下,模型会立刻跳出循环。
3.4 进展度量:识别"看起来很忙"的发散
振荡检测抓的是"结构重复",但探索发散在结构上是变化的。这时候要看信息增益:每轮的 observation 到底带来了多少新东西。
classNoveltyTracker:def__init__(self,stall_patience:int=3):self.seen:set[str]=set()self.stall_patience=stall_patience self.stall_count=0defobserve(self,observation:str)->bool:"""返回 True 表示有进展,False 表示本轮无新信息。"""h=hashlib.sha256(observation.strip().encode("utf-8")).hexdigest()ifhinself.seen:self.stall_count+=1returnFalseself.seen.add(h)self.stall_count=0returnTruedefstalled(self)->bool:returnself.stall_count>=self.stall_patience生产环境里,纯哈希判重有点严——模型换个标点就会绕过。更稳的做法是哈希 + 轻量相似度:对 observation 做 token 化的 Jaccard 相似度,与最近 K 条比较,若最高相似度都超过 0.9,就当作"无新信息"。这一步不需要向量库,几十行代码就够。
把这两个检测器接进循环,就得到了执行器的核心:
defrun_agent(llm,tools,messages,budget:Budget):osc=OscillationDetector()novelty=NoveltyTracker()best_effort=NonewhileTrue:reason=budget.exhausted()ifreason:# 优雅降级:能返回什么就返回什么,绝不抛裸异常给用户raiseAgentRunError(reason,partial=best_effort)# 剩余步数不足 2 时,物理切断工具调用,强制收敛force_final=budget.max_steps-budget.steps<=2resp=llm.chat(messages,tools=[]ifforce_finalelsetools.specs(),tool_choice="none"ifforce_finalelse"auto",)budget.steps+=1budget.tokens+=resp.usage.total_tokensifresp.is_final:returnresp.answerifnotresp.tool_calls:# 模型既没调工具也没给答案,视作一次无效步,注入提醒后继续messages.append({"role":"system","content":"请给出答案或调用工具。"})continueforcallinresp.tool_calls:fp=tool_fingerprint(call.name,call.args)ifosc.observe(f"{call.name}:{fp}")isnotNone:messages.append({"role":"system","content":"检测到策略振荡,请更换思路或直接作答。"})break# ……执行工具、记录指纹、更新 novelty、回灌 observation3.5 超时与取消传播:别让一个卡住的工具拖垮全局
预算里的max_seconds只有在每个工具都真正响应取消时才有意义。现实中常见的情况是:控制器超时了,可那个正在跑的 HTTP 请求还挂在连接池里,占着线程,几分钟后返回,然后写出一个"迟到的副作用"。
正确做法是把取消信号做成一等公民,沿着调用链传下去:
classCancelToken:"""轻量取消令牌,沿调用链传递,工具实现方有义务定期检查。"""def__init__(self):self._cancelled=Falsedefcancel(self):self._cancelled=Truedefraise_if_cancelled(self):ifself._cancelled:raiseAgentRunError("cancelled",partial=None)实践要点:
- 每个工具函数签名里加
cancel: CancelToken,在循环体(分页拉取、流式读取)里定期raise_if_cancelled()。 - 对无法改造的三方调用,用带超时的封装器兜底,超时值取
min(工具自身超时, 预算剩余时间)。 - 取消要幂等且可重入:多个检测器可能同时触发取消,重复 cancel 不能报错。
- 取消后的清理(回滚事务、删除临时文件)要走
finally,不要写在正常路径上。
3.6 把成本归因做进收尾
Post-flight 最容易被省略,但它是唯一能让下一版更省的一环。每次运行结束时,落一条结构化记录:
{"run_id":"r-9f3c","termination":"max_steps","steps":12,"tokens":186400,"tool_calls":27,"duplicate_calls":9,"oscillation_hits":2,"stall_rounds":3,"cost_usd":0.62,"final_answer_present":true}有了这些字段,你就能回答"我们的预算是定小了还是 prompt 写坏了":如果termination以max_steps居多但duplicate_calls很低,说明任务确实复杂,预算要调大;如果duplicate_calls和stall_rounds很高,那是策略问题,继续加预算只会更贵。
四、常见踩坑与优化建议
下面这些坑,基本每一个
更多硬核网安与AI工具包,请扫码获取完整源码!