AI Agent循环调用如何止损?四道熔断闸门与兜底方案实战

发布时间:2026/9/26 9:42:53
AI Agent循环调用如何止损?四道熔断闸门与兜底方案实战 上个月底我盯着后台账单页看了好几分钟一行数字跳出来的时候心里凉了半截。一套普普通通的 Agent 调度服务没接任何昂贵的商业模型套餐也没跑大规模批量任务一天之内烧掉了平时一周的预算。翻日志才发现某个子任务从凌晨开始就一直在循环调用大模型接口每隔几秒发一次请求每次返回的内容还都差不多像台复读机在拼命给自己加戏。这就是 Agent 开发中最经典的隐形资产杀手循环调用而很多人直到账单爆雷才意识到需要给 Agent 设计兜底方案。这篇文章想聊的就是循环调用为什么会发生、怎么在设计阶段减少它、以及当它真的发生时如何用一套可靠的兜底机制及时熔断把损失控制在可接受范围内。适合正在用 LangChain、AutoGen、自研 Agent 框架或者任何接大模型 API 做自动化任务的开发者参考也适合刚接触 AI Agent、还没被账单毒打过的新手提前避坑。1. 账单爆雷那天Agent 到底在后台干了什么1.1 一次被循环调用掏空的典型事故现场我那次事故的日志链路其实特别简单简单到有点可笑。一个负责“整理销售数据”的子 Agent把一份 CSV 文件路径传给了工具函数工具函数正常返回了数据然后 Agent 觉得数据“不够干净”又调用了一次清洗工具。清洗工具返回结果后Agent 的下一轮回复居然还在重复请求同一个清洗工具参数几乎一模一样只是路径字符串后面多了一个没意义的空格。大模型没有记忆中的“刚才已经干过这件事了”这种自觉它会根据当前上下文继续推断下一步动作。只要上下文里没有明确的“停止”信号它就会一直产出工具调用指令。那次循环持续了六个多小时每分钟调用七八次累计消耗了接近两百万 token账单直接爆掉。更离谱的是没有一条监控告警触发因为 API 调用是成功的HTTP 状态码全是 200所谓“异常检测”根本没覆盖到这种业务层面的死循环。这件事给了我一个非常深刻的教训Agent 的失败模式和大模型本身的失败模式完全不同后者最多是单次输出质量差前者会把一个错误决策指数级放大直到账户余额耗尽。所以循环调用兜底不是可选项是上线前的必选项。1.2 大模型不会主动叫停三类最常见的死循环为什么大模型不会自己停下来因为它本质上是一个逐 token 生成概率的引擎每一轮都只根据当前上下文预测下一个最合理的动作。它没有“我花太多钱了吗”“我是不是已经在重复劳动”这种内省能力。除非提示词里明确写了“不要重复调用”否则它感知不到自己在循环。从实际跑过的项目来看循环调用大致可以归为三类每一类的触发原因都不一样。第一类是自说自话型。Agent 在单线程内反复调用同一个工具参数不断微调甚至不变。常见于数据清洗、网页抓取、文件处理这类任务。触发原因是系统提示词没有强调“一次完成不要重复”或者工具返回的格式太复杂Agent 误以为任务还有后续。第二类是工具互踢皮球型。两个或两个以上的工具互为输入输出比如 Agent 先调用 A 工具生成一份报告再调用 B 工具把报告格式化接着又调用 A 工具“优化”然后 B 又收到“优化后的报告”再次格式化。每个工具都成功执行但整体没有任何产出纯粹在空转。第三类是重试风暴型。工具调用失败后返回错误信息Agent 自动发起重试。如果错误是外部依赖导致的比如数据库连接池满了、第三方接口限流Agent 反复重试只会加重故障。更麻烦的是有些框架内置了自动重试机制框架一层、Agent 一层叠加起来就是指数级的请求量。这三类循环有一个共同特征单次调用的成本都不高但循环的次数没有上限最终成本完全不可预测。理解了这个本质后面设计兜底方案时思路就会清晰很多。2. 设计阶段的止损让循环在源头就没有生长土壤兜底方案再完善也只是事后补救。真正高效的做法是在 Agent 的任务链路设计阶段就把循环产生的土壤去掉。这一节不是讲空泛的架构原则而是讲三个可以直接落地的设计手段。2.1 给 Agent 一个明确的终态状态机设计我见过太多 Agent 项目的状态管理就是“拿到模型输出解析工具调用继续循环”完全没有终态概念。这种写法的问题在于只要模型连续产生几个工具调用指令循环就停不下来。一个可靠的实践是给 Agent 定义一个最小状态集合init、processing、waiting_tool、completed、failed。每一轮循环结束后必须显式判断下一步是留在processing还是跳转到completed。跳转条件不能依赖大模型的“自觉”而要在代码里硬编码。比如当工具执行成功后判断该工具是否是任务链路的最后一步。如果是直接强制状态为completed不管模型下一轮还想调用什么都不给执行机会。这种硬约束虽然看起来有点“粗暴”但在生产环境里非常有效它从根本上切断了“模型永远觉得还有下一步”的可能。2.2 工具返回结构把“结束”当成一等公民很多工具函数的返回就是一段文本或者一个 JSON 数据然后把这个东西原封不动塞回给模型当上下文。这里有个隐藏问题模型不知道这个结果是不是最终结果它会倾向于继续“加工”。我现在的做法是统一约定工具返回结构{ status: succeeded, is_final: false, data: 这里是工具执行结果, next_action_hint: optional }其中is_final字段尤其重要。如果工具执行者明确知道某个操作已经是任务终点就置为trueAgent 运行器看到is_finaltrue后无论模型下一轮输出什么工具调用指令都强制结束循环。这相当于把“停止权”从大模型手里收回来交给确定性的代码逻辑。next_action_hint则用来给模型一个“正路”的引导减少它自己瞎猜的概率。比如清洗工具执行完后hint 可以写“清洗完成可以生成最终报告”模型看到这个提示后就不会再去重复调用清洗工具了。2.3 提示词层的软约束除了硬性的代码约束系统提示词里也应该明确写清楚循环禁忌。不要写那种含糊的“请尽可能高效地完成任务”要写具体的。我常用的写法是你在执行任务时必须遵循以下规则每个工具只尝试一次除非收到明确的失败信号。如果工具返回结果正常不允许对同一目标再次发起相同调用。完成核心目标后立即输出最终回答不要补充额外的工具调用。如果发现自己连续执行了 5 轮且没有新的信息输入直接输出“任务已完成”并终止。这些规则模型不一定每次都遵守但它们确实能显著降低循环触发概率。它就像给一个容易冲动的人提前打了预防针不一定百分百有效但至少能减少冲动次数。更重要的价值是当兜底方案触发时日志里能明确看到是哪条规则被违反了方便后续优化。3. 兜底方案的骨架四道互相独立的熔断闸门任何 Agent 任务链路都不该以“模型觉得结束”作为唯一结束条件必须有确定性代码兜底。我把这套兜底拆成四道闸门轮次闸门、时间闸门、成本闸门、语义闸门。为什么要四道而不是一道因为每种失控方式不同单靠计数或单靠超时都有漏洞四道互相独立才能做到完整覆盖。3.1 轮次闸门设置调用上限最基础的一道闸门给单次 Agent 任务设置最大调用轮数。这里的“轮”指的是启动一次 Agent 运行到最终返回之间的大模型调用次数不是工具调用次数。比如设置为 10 轮那不管任务完成没有跑到第 11 轮必须强行终止并返回错误。这个值的设置需要根据任务复杂度调整一个简单查询任务可能 3 轮内就该结束复杂的数据分析可能允许 15 到 20 轮。宁可设宽一点也不要在正常任务跑一半的时候被误杀。我被误杀过太多次了所以现在都是先跑一遍无限制的测试任务统计正常轮数再乘以 1.5 到 2 作为上限。实现层面就是一个简单的计数器在每次调用大模型之前检查if self.round_count self.max_rounds: raise AgentLoopError(超过最大调用轮数已强制终止)这个检查必须放在调用大模型 API 之前而不是之后。放在之后意味着你这一轮的钱已经花了放之前才能止损。3.2 时间闸门硬超时轮次闸门解决的是“调了太多轮”的问题但有时候每轮调用本身很慢或者工具在等待外部响应轮数不多时间却很吓人。这时候需要时间闸门。时间闸门有两种粒度。一种是单轮超时比如规定一次模型调用必须在 60 秒内返回超时就按失败处理。另一种是整体任务超时比如整个 Agent 运行最长不超过 5 分钟超时直接终止。单轮超时通常通过 HTTP 客户端的 timeout 参数来设置比较简单。整体超时稍微复杂一点因为你可能需要中断正在进行的模型流式响应。Python 里可以用concurrent.futures或asyncio.wait_for来实现也可以用信号量在进程层面做强制中断。整体超时的时间设计要参考任务特性。如果任务大量依赖外部 API就要预留充足时间如果只是模型内部推理可以设紧一些。我在生产环境一般设置为轮次上限和平均单轮耗时的乘积再留出 20% 的缓冲。3.3 成本闸门token 与预算熔断轮次和时间都不敏感的场景下真正让你钱包出血的是单轮调用的 token 数量。有些复杂的上下文场景每轮调用都会携带越来越长的历史消息token 消耗是指数级增长的。你可能只调了 8 轮但每轮都比上一轮多塞了几千 token 的上下文总成本远高于预算。成本闸门就是在运行时累加每次调用的 token 用量一旦超过设定阈值立即熔断。大多数模型 API 的响应体里都会返回usage.prompt_tokens、usage.completion_tokens、usage.total_tokens读取后累加即可。这里有个很多人忽略的点Prompt 层级的 token 计数。如果用的模型支持max_tokens参数那不是真正的成本上限模型通常只限制单次输出长度不限制输入长度。要监控的其实是每次请求的prompt_tokens因为循环调用时真正暴涨的是输入侧的上下文。成本闸门的阈值设置我一般按“单任务最高可接受成本”计算。比如一个任务预期花 5 毛钱我可以设 5 元作为硬上限允许它有 10 倍偏差但绝不允许无限制涨。这样既能容忍正常的复杂度波动又能避免极端失控。3.4 语义闸门循环检测最后一道闸门也是最难的一道语义循环检测。它解决的是轮次不多、时间不长、token 也没超但模型在反复输出相似内容或重复调用类似工具的问题。这类循环最隐蔽因为它从计数上看完全正常实际上在空转。我的实现思路是维护一个最近 N 轮输出的文本列表每一轮结束后计算当前输出与之前输出的文本相似度。如果相似度超过阈值比如 0.9就判定为循环。相似度计算可以用简单的方式不一定要上向量模型。用字符串级别的编辑距离Levenshtein或者 hash 去重就够了。更有效的做法是提取“工具调用签名”也就是模型输出里所有工具调用的函数名和参数序列比较相邻几轮的签名是否高度一致。如果连续 3 轮都在请求同一个函数、参数也大同小异基本可以断定进入了循环。这个方法的灵感来自我在日志里看到的那个事故每次调用的参数确实有细微差别多一个空格、换一个变量名但工具签名本质上是一样的。字符串相似度可能只有 0.85工具签名相似度能达到 0.99。所以语义闸门更专注于“动作”的重复检测而不是“文本”的重复检测。4. 写一个可复用的循环防护基座附代码这一节直接给一套精简但可用的代码。它不是完整框架而是一个可以嵌入到你现有 Agent 运行器里的防护基座。我把它封装成一个类包含前面说的四道闸门轮次限制、时间预算、token 熔断、语义循环检测。import time import threading from typing import List class AgentLoopGuard: Agent 循环调用防护基座四道闸门合并控制 def __init__( self, max_rounds: int 10, max_duration: float 120.0, max_tokens: int 80000, similarity_threshold: float 0.9, window_size: int 4 ): self.max_rounds max_rounds self.max_duration max_duration self.max_tokens max_tokens self.similarity_threshold similarity_threshold self.window_size window_size self.round_count 0 self.total_tokens 0 self.start_time time.time() self._lock threading.Lock() self.recent_signatures: List[str] [] def before_call(self) - None: 调用大模型前执行检查不通过则抛出 RuntimeError with self._lock: if self.round_count self.max_rounds: raise RuntimeError(f熔断超过最大轮次 {self.max_rounds}) if time.time() - self.start_time self.max_duration: raise RuntimeError(f熔断超过最大时长 {self.max_duration}s) if self.total_tokens self.max_tokens: raise RuntimeError(f熔断超过 token 预算 {self.max_tokens}) def after_call(self, token_usage: int, action_signature: str) - None: 调用结束后记录用量检查语义循环 with self._lock: self.round_count 1 self.total_tokens token_usage self._check_loop(action_signature) def _check_loop(self, signature: str) - None: 基于工具调用签名做循环检测 self.recent_signatures.append(signature) if len(self.recent_signatures) self.window_size: self.recent_signatures.pop(0) n len(self.recent_signatures) if n 3: # 最近三次签名完全一致判定为循环 recent self.recent_signatures[-3:] if recent[0] recent[1] recent[2]: raise RuntimeError(熔断检测到工具调用签名连续重复疑似死循环)使用方式很简单。假设你原来是这样调 Agentdef run_agent(task): response call_llm(task) while response.get(tool_calls): result execute_tool(response[tool_calls]) response call_llm(result) return response改造后是这样def run_agent_with_guard(task): guard AgentLoopGuard() response call_llm(task) guard.after_call(response_usage(response), extract_signature(response)) while response.get(tool_calls): guard.before_call() result execute_tool(response[tool_calls]) response call_llm(result) guard.after_call(response_usage(response), extract_signature(response)) return responseextract_signature函数负责从模型输出中提取每个工具调用的函数名和参数摘要拼接成字符串。比如def extract_signature(response) - str: if not response.get(tool_calls): return no_tool_call sig_parts [] for call in response[tool_calls]: func call.get(function, {}).get(name, ) args call.get(function, {}).get(arguments, ) sig_parts.append(f{func}({args[:200]})) return |.join(sig_parts)注意参数只取前 200 个字符避免因为参数里带上无意义的随机串导致签名变化过大反而检测不出循环。这套代码的核心价值不在于实现多复杂而在于它把四道闸门的检查时机放对了before_call在花钱之前拦after_call在花钱之后记账。我踩过的坑是有人把轮次检查放在after_call里结果第一轮调用已经付费了才想起来检查根本没有止损意义。接入现有框架时LangChain 可以在RunnableSequence外层包一个中间件或者直接继承AgentExecutor重写_call方法自研框架就更灵活直接套在run_agent入口即可。核心原则只有一个检查必须在每次大模型 API 调用之前执行。5. 循环已经发生用日志和追踪还原现场兜底方案再完备也不可能消灭所有循环总有一些漏网的。问题发生后最怕的不是花钱而是花了一晚上排查也定位不到根因。所以这一节聊聊我日常排查循环调用现场的方法。5.1 结构化日志要打哪些字段我要求所有 Agent 运行日志必须是结构化 JSON一行一个事件至少包含这些字段task_id: 任务唯一 ID run_id: 单次运行 ID round: 当前轮次 event: run_start / llm_request / tool_call / llm_response / guard_trigger timestamp: 精确到毫秒 token_usage: 本轮 token 用量 tool_name: 本轮调用工具名 tool_args_hash: 工具参数的 hash便于快速比对 action_signature: 上述工具签名 error_message: 如果有错误其中tool_args_hash和action_signature是关键。循环调用最明显的特征就是这两个字段反复出现相同值。排查时直接对这两个字段做 group by一眼就能看到某个函数被调了多少次、每次参数是不是一样的。5.2 一个线上排查案例的完整链路有一次小伙伴反馈某个生产任务“偶尔卡死”但系统没有告警。我看了一眼日志先按task_id过滤出完整的运行链路然后看event序列。结果发现某个 Agent 的日志序列是这样的llm_request round3 tool_namesearch_files tool_call round3 tool_namesearch_files args_hasha1b2c3 llm_response round3 next_toolsearch_files llm_request round4 tool_namesearch_files tool_call round4 tool_namesearch_files args_hasha1b2c3d4 ...看起来 args_hash 每次不完全相同很容易误判为“参数在变不是死循环”。但我把action_signature拉出来一对比发现每次都是search_files(path/data, queryreport)只是后面的时间参数在变。也就是说模型每次搜索的路径和关键字完全一样唯一变化的是一个随机时间戳。这就是典型的伪装循环。定位到之后我加了一条规则参数 hash 不一定可靠工具签名中的函数名加核心参数才是判断依据。修复方式是调整系统提示词明确告诉模型“如果搜索路径和关键字相同不要重复搜索”并且在工具维度加了一个防护同一任务内相同路径同名查询最多执行一次第二次直接返回上次结果。5.3 常用可观测性工具简单的日志部署可以利用现有日志平台完成。如果要从头搭可以关注这几个OpenTelemetry 是目前最通用的方案给大模型调用链路打 span 后可以在 Jaeger 里可视化看到每一轮的调用关系循环会表现为一条链路上反复出现相同节点视觉上非常明显。LangSmith 这类专用可观测平台比较省事直接接入后能看到每一次运行的所有调用记录但要注意这是付费服务而且如果你的调用量非常大这部分的费用也得纳入成本考量。自建方案的话SQLite 加一个轻量查询页面其实就够用了。我的习惯是把关键日志事件写入本地 SQLite每天一份统计报表按工具名汇总调用次数、token 消耗、平均耗时。一旦某天的报表里某个工具调用次数异常基本就能提前预警不用等账单出来才后知后觉。6. 最后聊聊几个容易被忽略的细节兜底方案看起来是几条规则的事真正落地时其实有不少细节会在关键时刻坑你一把。这几条是我在多次账单惊吓后总结出来的分享出来希望能给大家省点学费。6.1 “兜底”是最后防线不是免费保险防护基座代码会拦住异常循环但它不会告诉你“这个任务应该用更少的轮数跑完”。我看过不少团队把轮次上限设得非常宽松比如 50 轮上限然后正常的任务只要跑 15 轮就能完成。结果兜底闸门常年不触发也没有人去优化提示词和工具链GPU 和 API 费用就在合理范围内持续膨胀。更好的做法是把防护熔断触发的次数当作一个核心监控指标。如果这个指标经常为零不一定是好消息反而说明你设置的阈值过宽。每周复盘一次触发的熔断类型和对应任务才能把成本逼到真正的合理区间。我以前一个月才看一次现在改成每周成本直接砍掉了将近三成。6.2 熔断之后的恢复策略比熔断本身更重要熔断不是把这个任务扔掉就完了。你得告诉客户端或者用户这个任务为什么失败以及是否允许重试。这里我又踩过一个坑一开始熔断后直接抛异常上游任务捕获异常后自动重试结果同一个死循环任务被重试了三次每次都消耗了将近一半的预算最后总成本比不熔断还要高。现在我的做法是在熔断错误里携带一个结构化错误码比如GUARD_ROUND_LIMIT、GUARD_TIME_LIMIT、GUARD_TOKEN_LIMIT、GUARD_LOOP_DETECTED。上游任务根据错误码决定是否允许重试轮次超限和语义循环不建议重试时间超时和 token 超限可以重试一次但重试前必须重置防护计数并清空上下文。6.3 成本监控要成为日常习惯最后一条其实和前文讲的技术关系不大但杀伤力极大。不要只在账单爆雷的时候才去看成本要养成每天瞄一眼成本仪表盘的习惯。不需要搞特别复杂的系统一个简单的表格就好日期、任务类型、Agent 名称、总调用次数、总 token 消耗、估算成本、熔断次数每天早上花三十秒看一眼这个表格一旦发现某个 Agent 的熔断次数从 0 变成了 5那就意味着提示词或者工具设计出了问题不必等周四复盘才发现。我现在的个人习惯是把这份表格自动发到工作群里渲染成一个普通文本图表不用任何特殊处理成本异常基本当天就能发现完全不会再出现月度账单爆雷这种事。