1. 项目概述先说个真实场景我在公司内部跑过一版带联网搜索的Agent原本设计好的流程是用户提问 - 检索资料 - 归纳总结 - 输出结论。看起来很简单对吧结果上线第三天那个Agent因为一个工具返回了异常格式的数据在一个解析函数里疯狂重试四十分钟调了上万次大模型接口。等到我发现的时候那个月的API预算已经烧掉了接近三成。这还只是内部测试环境如果跑的是面向用户的付费服务这一晚上赔出去的钱可能就够买一辆代步车了。所以这个开源系列做到第13期我下定决心要做一件事给提示流编排器装上安全带和保险丝——也就是标题里说的运行时看门狗Runtime Watchdog和死循环熔断器Loop Circuit Breaker。简单来说前者盯着Agent运行时的资源消耗和健康状态后者专门拦截那些一旦跑起来就停不下来的循环调用链。目标是让Agent在失控的时候能够被快速发现、快速降级、快速止损而不是等账单出来才发现自己成了信用卡受害者。这套东西适合谁看如果你正在开发Agent应用、搭过工作流编排工具、或者只是对大模型API的成本控制有焦虑那这篇文章值得你花二十分钟读一遍。我不会扯太玄乎的理论全程都是可以直接落到代码里的方案包括我踩过的坑和改了三版才定下来的细节设计。2. 为什么提示流编排器需要看门狗和熔断器2.1 Agent失控的几种典型姿势先说几个我真实遇到过的Agent失控场景你可以对照一下自己有没有踩过类似的坑。第一种叫重试风暴。Agent在调用外部API的时候收到一个异常响应按照代码逻辑它会重试。但问题在于很多Agent框架的重试机制是按次数设计的重试三次、五次就停。可如果调用链比较深A调用B、B调用C、C失败导致B重试、B重试又触发A重试这种级联放大很容易让调用次数指数级增长。第二种叫逻辑死循环。Agent在规划下一步行动的时候因为提示词写得不够严谨或者上下文里出现了矛盾的指令它会反复执行同一个动作搜索同一个关键词、调用同一个工具、输出几乎一样的中间结果然后继续规划、继续执行。每一步看起来都有逻辑但整体上就是在原地打转。第三种叫上下文膨胀。Agent在一次会话中不断往上下文里追加历史记录token数越滚越大。表面上每次调用都正常返回了但延迟在飙升、成本在涨甚至上下文有可能溢出报错。这类失控是最隐蔽的因为从功能上看它没出Bug纯粹是资源消耗失控。无论哪一种症结都是同一个使用方缺少对运行过程的实时观测和主动干预能力。普通的工作流引擎只要流程定了就不会跑偏但基于大模型的Agent天然带有随机性和不确定性你不能指望提示词写得好它就不会失控。这就像开车不能只靠技术好还得装刹车和保险杠。2.2 看门狗和熔断器的职责切分在设计的时候我把这两个组件定位成不同层级的保护机制。运行时看门狗是持续监控器它的动作是观察、预警、干预。它盯着每一步的执行耗时、Token消耗、调用频率、上下文占用一旦发现指标超过阈值要么告警要么自动触发降级操作。它防的是慢性病——成本慢慢失控、延迟慢慢恶化这一类。死循环熔断器则是模式识别器它的动作是识别、切断、隔离。它分析执行轨迹里的循环特征如果发现相同或高度相似的步骤反复出现就直接熔断掉整条执行链。它防的是急性病——逻辑死循环、重试风暴这一类一旦发作就会迅速烧钱的故障。打个比方看门狗是气象台实时监测风速和降雨量预警台风熔断器是防洪闸门当水位超过临界线时一刀切下去保住下游。两者配合才能在Agent真正失控之前把损失控制住。3. 运行时看门狗的设计与实现3.1 需要盯住的核心指标设计看门狗的第一步是确定监控指标。我根据实际踩坑经验筛选了四个核心维度其他指标都是在这四个维度上的延伸。指标维度具体指标建议阈值参考监控目标耗时单步执行耗时、整链总耗时单步不超过30秒整链不超过5分钟识别慢调用和挂死Token消耗每一步的输入/输出Token数、累计Token数单轮任务的Token消耗上限控制成本调用频率单位时间内的API调用次数每分钟不超过20次就是一个合理警戒值拦截重试风暴上下文占用上下文消息条数、Token占用率上下文Token占用率不超过80%防止上下文膨胀举个例子我设置的 Token 警戒线是单轮任务累计消耗不超过 10 万 Token。这个数字不是拍脑袋定的而是根据业务量估算出来的一个典型的复杂任务大概需要 20 到 30 步推理每步平均消耗 3000 到 5000 Token10 万的上限已经留出了差不多一倍的缓冲空间。如果超过这个用量大概率不是正常执行而是什么环节出了问题。3.2 看门狗的工作流程从观察到干预看门狗的完整工作流程我分成了五个阶段每个阶段都有明确的动作和出口条件。第一步是埋点采集。在编排器内部定义一个统一的执行器接口每一次步骤执行开始、结束、失败都会发出事件。事件会带上步骤ID、执行器类型、耗时、Token用量这些元数据。这一步的关键是一个字全。宁可多采集一些暂时用不上的数据也不要漏掉关键节点。第二步是数据聚合。事件流会进入一个轻量级的滑动窗口计数器我实现的是一个环形缓冲区固定窗口大小为 60 秒每秒一个时间桶存储该秒内的调用次数、Token消耗、平均耗时。之所以用滑动窗口而不是简单计数器是因为滑动窗口能真实反映最近的节奏而简单计数器容易把过去的波动掺进当前判断里。第三步是指标计算。每隔一段时间就会基于滑动窗口算一次指标快照跟阈值做比较。这里我用了一个小技巧对耗时指标做指数移动平均EMA公式是ema alpha * current (1 - alpha) * emaalpha 取 0.3。这样做的好处是过滤掉偶发的毛刺比如某个步骤因为网络抖动多花了三秒钟EMA不会因为这一个局部波动就误报警。第四步是状态判定。看门狗分三级状态绿色表示安全黄色表示需要注意红色表示危险。黄色状态下只告警不干预只有红色状态才触发降级动作。第五步才是干预动作。干预分两档一档是软降级比如强制调整Agent的temperature参数降低随机性、切换到更便宜的模型、暂停非关键的旁路工具调用另一档是硬熔断直接终止当前执行链并返回提示信息。注意硬熔断不是Lane逻辑的回滚而是直接中止整个任务宁可让任务失败也不让它继续烧钱。3.3 冷静期与自动恢复熔断不是结束后面还要有恢复策略。打个比方电路跳闸之后你肯定会先检查故障点再合闸。Agent的熔断恢复也是类似我给编排器设计了冷却时间机制默认熔断后 30 秒内禁止同一类型的任务重新发起执行这个冷却时间会根据熔断次数指数退避——第一次30秒、第二次60秒、第三次120秒上限是10分钟。冷却期过去之后系统允许少量任务以灰度模式放行也就是同时使用正常模型和降级模型各跑一次对比两者的结果质量。如果灰度任务在连续五次执行中都没有触发熔断就自动取消降级恢复正常模式。这背后是一个简单的统计置信度逻辑连续成功的次数越多说明故障原因被解决的可能性越大可以逐步恢复信心。看门狗组件里还有一个容易被忽视的细节代码总是先记录、后判断。换句话说即使报警信息没有触发任何熔断动作日志也必须完整落盘。因为很多问题在发生的那一刻看起来无关紧要复盘的时候才发现是早期信号。如果你连信号都没留下排查就只能靠瞎猜了。4. 死循环熔断器的设计与实现4.1 如何识别这步之前做过了看门狗管的是总量失控但死循环熔断器要解决的是逻辑卡死。这两者的核心差异在于看门狗判断是不是太多了用的是数值阈值熔断器判断是不是在重复用的是模式匹配。我的实现思路分三层。第一层是步骤指纹提取。每一次步骤执行结束时根据(执行器类型, 输入参数的哈希值, 返回结果的摘要值)生成一个指纹字符串。所谓输入参数哈希指的是把传给执行器的核心参数序列化之后做一个 SHA-256 摘要这样哪怕参数只有细微差别指纹也会完全不同而返回结果的摘要则是为了防止一种情况——两个不同的输入产生相同的结果那其实也算是收敛了不该算重复。第二层是循环模式判定。在编排器内部维护一个执行轨迹栈每生成一个新的步骤指纹就推入栈中。判定的规则是如果当前指纹跟栈里前面的某个指纹相同并且中间的步骤数少于一个预设值我设的是 5就判定为循环。中间步骤数也被限制的原因很好理解如果两次相同操作之间隔了五十个不同的步骤那更像是一个正常的复杂流程而不是死循环。第三层是次数累计与熔断。如果同一个指纹对在最近 10 个步骤内出现超过 3 次熔断器直接置位。为了记录这些数据我用了一个基于一致性哈希的环形缓冲——它的作用是把最近执行过的步骤指纹均匀地分布到固定长度的缓冲区里既能快速查找又不会因为追踪历史信息越来越多而让内存无限膨胀。4.2 用状态机设计熔断器生命周期熔断器不能只是简单的断与不断那样太粗糙了。我把它设计成了一个带状态迁移的小型状态机总共有四个状态。第一个是CLOSED状态也就是一切正常的闭合状态。此时熔断器只做记录不做拦截。第二个是OPEN状态熔断器的断开状态一旦进入这个状态所有新任务请求都会被直接拒绝连执行入口都不会打开直接返回提示信息。第三个是HALF_OPEN状态也就是半开状态是熔断后试探性的放行少量任务状态用来验证故障是否已经解除。第四个是HALF_OPEN_TIMEOUT如果半开状态下放行的任务在超时时间内没有完成则说明故障环境仍然存在会直接拉回 OPEN 状态。这四个状态之间的迁移条件是硬编码在代码里的允许一定的配置自由度。状态机实现的好处是不会出现已经熔断了但还有漏网任务在执行的尴尬情况——因为所有入口都只有一处在判定状态。4.3 熔断器的触发阈值计算死循环熔断器的核心痛点在于阈值设多少合适。设得太小正常的多步骤任务可能被误杀设得太大又起不到保护作用。我自己的经验是先从最小循环长度入手。一个最小循环是两步执行A、执行B、再执行A。所以我在指纹匹配时设的最低循环长度是 2 步。但考虑到底层大模型偶尔会有非常相近的中间结果我额外要求最近 10 步内同一子序列出现 3 次才算熔断条件。换句话说允许 2 次偶然重复第 3 次就基本可以认定是卡死了。这个 3 次的数字是根据我自己的线上数据统计来的正常任务的重复率极少超过 1 次而真正死循环的任务从第 2 或第 3 次重复开始后续动作几乎必然完全一致。如果你的业务场景对误杀特别敏感建议阈值调到 4 到 5 次再熔断代价是故障期间多烧 30% 左右的费用。这是一个成本和误判率的权衡没有标准答案得看你的钱袋子有多厚。4.4 熔断动作的执行细节熔断器触发之后不是简单地停住就完事了。我在实现里加了两个动作来让熔断更彻底。第一个动作是状态冻结。被熔断的执行链会保留当时的完整上下文快照包括步骤轨迹、输入输出、Token消耗等。这样做不只是为了复盘更是为了下次恢复执行时可以拿到精准的断点位置而不用从零开始重跑。第二个动作是依赖隔离。如果熔断是由某个具体工具触发的比如那个工具连续三次返回同样的异常结果熔断器会把该工具标记为不可用后续的执行链会自动绕开它。这一步很关键因为它能防止熔断之后系统自动走了另一条路结果那条路上也有同一个有问题的工具等于换了个地方继续烧钱。这两个动作加在一起才勉强称得上熔断——毕竟熔断的语义本来就是切掉整段电路不是只切掉一个灯泡。5. 完整实操从配置到代码5.1 配置项设计为了让看门狗和熔断器真正能用起来我把所有参数都做成了配置项而不是写死在代码里。下面这份配置可以直接粘到你的项目里作为起点结合自身场景微调。WATCHDOG_CONFIG { window_seconds: 60, step_timeout: 30, # 单步超时 total_timeout: 300, # 总超时 max_tokens_per_task: 100000, # 单任务Token上限 max_calls_per_minute: 20, # 每分钟调用上限 context_ratio_threshold: 0.8, # 上下文占用比例阈值 ema_alpha: 0.3, # 指数移动平均系数 cool_down_seconds: 30, # 初始冷静期 cool_down_max: 600, # 冷静期上限 } CIRCUIT_BREAKER_CONFIG { min_loop_length: 2, same_subsequence_threshold: 3, recent_steps_buffer_size: 10, half_open_max_tasks: 2, half_open_timeout: 30, }这里有一个值得说明的参数same_subsequence_threshold设成了 3对应的含义是最近 10 步里同一个子序列出现 3 次就算死循环。如果你希望更精确一些可以把这个阈值定得更高但那样熔断器的反应速度会更慢属于少止损和少误杀之间的取舍。5.2 核心代码骨架我把这个系统简化成了一个可以直接跑的版本核心逻辑大约两百行。你可以把它嵌入到任何一个支持事件驱动的编排器里。import hashlib import json import time import threading from collections import deque from enum import Enum class CircuitState(Enum): CLOSED 1 OPEN 2 HALF_OPEN 3 HALF_OPEN_TIMEOUT 4 class StepEvent: def __init__(self, step_id, executor_type, inputs, outputs, elapsed_ms, token_usage): self.step_id step_id self.executor_type executor_type self.inputs inputs self.outputs outputs self.elapsed_ms elapsed_ms self.token_usage token_usage self.fingerprint self._build_fingerprint() def _build_fingerprint(self): input_digest hashlib.sha256( json.dumps(self.inputs, sort_keysTrue, defaultstr).encode() ).hexdigest()[:16] output_digest hashlib.sha256( json.dumps(self.outputs, sort_keysTrue, defaultstr).encode() ).hexdigest()[:16] return f{self.executor_type}:{input_digest}:{output_digest} class SlidingWindowCounter: def __init__(self, window_seconds60): self.window_seconds window_seconds self.buckets {} self.lock threading.Lock() def add(self, value, bucket_keyNone): bucket_key bucket_key or int(time.time()) with self.lock: self.buckets[bucket_key] self.buckets.get(bucket_key, 0) value cutoff bucket_key - self.window_seconds self.buckets {k: v for k, v in self.buckets.items() if k cutoff} def sum(self, current_timeNone): current_time current_time or int(time.time()) cutoff current_time - self.window_seconds return sum(v for k, v in self.buckets.items() if k cutoff) class RuntimeWatchdog: def __init__(self, config): self.config config self.call_counter SlidingWindowCounter(config[window_seconds]) self.token_counter SlidingWindowCounter(config[window_seconds]) self.ema_step_time None self.state GREEN self.trigger_limit config[max_tokens_per_task] def on_step_completed(self, event): # 记录指标 self.call_counter.add(1) self.token_counter.add(event.token_usage) if self.ema_step_time is None: self.ema_step_time event.elapsed_ms else: alpha self.config[ema_alpha] self.ema_step_time alpha * event.elapsed_ms (1 - alpha) * self.ema_step_time # 状态判定 total_tokens self.token_counter.sum() calls_per_minute self.call_counter.sum() step_timeout self.config[step_timeout] * 1000 if total_tokens self.trigger_limit or calls_per_minute self.config[max_calls_per_minute]: self.state RED return BREAK if self.ema_step_time step_timeout: self.state YELLOW return WARN self.state GREEN return OK class LoopCircuitBreaker: def __init__(self, config): self.config config self.state CircuitState.CLOSED self.recent_steps deque(maxlenconfig[recent_steps_buffer_size]) self.sequence_count {} self.action_lock threading.Lock() def _extract_subsequences(self): sequences [] recent_list list(self.recent_steps) if len(recent_list) self.config[min_loop_length]: return sequences step self.config[min_loop_length] for i in range(len(recent_list) - step 1): sequences.append(tuple(recent_list[i:i step])) return sequences def on_step_completed(self, event): fingerprint event.fingerprint with self.action_lock: if self.state CircuitState.OPEN: return BREAK self.recent_steps.append(fingerprint) for seq in self._extract_subsequences(): self.sequence_count[seq] self.sequence_count.get(seq, 0) 1 if self.sequence_count[seq] self.config[same_subsequence_threshold]: self.state CircuitState.OPEN return BREAK return OK这段代码是演示用的但已经能跑通核心逻辑。真正接入生产环境的时候你需要把on_step_completed挂到编排器的执行回调上通常是在每一步执行完await返回的位置上调用。5.3 把两者接入编排器看门狗和熔断器是独立的组件但实际要协同工作。我在编排器里用一个名为ExecutorGuard的外层包装器来统一调用它们。class ExecutorGuard: def __init__(self, watchdog, breaker): self.watchdog watchdog self.breaker breaker async def execute(self, step_id, executor_type, inputs, executor_fn): event_start time.time() try: # 检查断路器状态 if self.breaker.state CircuitState.OPEN: return {status: rejected, reason: circuit_open} # 执行真正的步骤 result await executor_fn(inputs) event StepEvent( step_idstep_id, executor_typeexecutor_type, inputsinputs, outputsresult, elapsed_ms(time.time() - event_start) * 1000, token_usageresult.get(token_usage, 0), ) # 先过熔断器 breaker_decision self.breaker.on_step_completed(event) # 再看门狗 watchdog_decision self.watchdog.on_step_completed(event) if breaker_decision BREAK or watchdog_decision BREAK: return {status: terminated, reason: guard_triggered} return {status: ok, data: result} except Exception as e: return {status: error, message: str(e)}通过这个包装器所有执行步骤都汇聚到同一个保护伞下。这里有一个细节值得注意breaker和watchdog是顺序判断的先判断是不是循环再判断是不是超限。这个顺序有讲究——如果先做资源判断可能会因为一个循环任务瞬时消耗过多资源而直接把任务杀掉导致你想在日志里观察循环轨迹都来不及。先让熔断器识别出这是一个循环再让看门狗确认这个循环确实在烧钱这样生产环境里能留下更完整的故障证据链。5.4 在异步事件循环里防重入写Agent编排器的时候最让人头疼的一个细节是异步重入。大模型调用大多是异步的这意味着on_step_completed会被并发触发。如果熔断器的sequence_count不是线程安全的多个回调同时写同一个字典轻则丢数据重则死锁。我的做法是在两个组件里都加了threading.Lock但这还不够。内层锁可以保护共享数据却解决不了检查-再执行之间的竞态条件。比如两个并发回调同时通过了breaker.state OPEN的检查然后一起进入了真实执行逻辑。这会让熔断器形同虚设。要真正防住并发穿透需要在ExecutorGuard.execute入口处加一个跨步骤的信号量Semaphore把真正能同时进入执行逻辑的任务数限制在一到两个。我选了一宁可让执行过程串行也不冒并发穿透的风险。如果追求吞吐可以放宽到二但那样还需要更复杂的锁机制来保证状态一致。这个就是典型的用吞吐换安全的取舍。6. 从生产环境里趟出来的问题和排查思路6.1 问题一重试风暴伪装成正常调用有一次我监控到看门狗频繁触发每分钟调用数超限的告警但熔断器毫无反应。查了执行轨迹才发现这个任务的步骤指纹每次都不同因为每次重试的时候Agent 都会在输入参数里加一个尝试次数的字段。从语义上看明明就是同一个操作不停重来但从指纹哈希的角度来说每次都不同熔断器自然识别不出来。排查之后我改了规则指纹的输入部分只取业务核心参数的哈希明确排除重试次数、时间戳这类噪声字段。更优雅的做法是给执行器定义指纹关键字段的白名单把哪些字段参与哈希计算的决定权下放给各个执行器自己管理。这样每个执行器都清楚自己的哪些输入是业务相关、哪些只是控制参数。6.2 问题二回调函数污染了主流程第一次集成熔断器的时候我直接把判断逻辑塞进了 Agent 推理循环的核心路径上。后来发现一个隐蔽的坑熔断器自身抛出的异常会干扰原来的业务逻辑。比如某个步骤超时了看门狗直接把任务中断但因为中断的异常类型和业务里的超时异常长得一样下游的兜底逻辑把它当成正常超时处理又触发了一次重试。等于保护机制自己给故障续了一命。解决方案是把保护逻辑做成旁路模式全部通过事件订阅和回调触发不直接跟业务代码共享异常语义。Guard 的返回值统一走{status: terminated}这样的协议结构业务代码只要按这个协议响应即可不会导致异常类型混乱。6.3 问题三误杀正常的长链任务熔断器上线之后我收到过一个有点棘手的投诉——一个正常的调研摘要任务执行了 12 步在中间某一步被熔断了。日志显示任务里有两个步骤恰好都调用了同一个搜索工具搜索关键词还类似于是指纹判定它们构成子序列重复触发了熔断。这个问题的根源是min_loop_length2太敏感了。解决办法是引入了区分度判断两个步骤不仅指纹要一致还要求后续的行为轨迹也要一致。我加了一个展望窗口的逻辑——只有当[A, B, A]这个三连模式出现并且在展望窗口内确认 A 之后的下一步也是 B才真正判定为循环。这样把单点重复变成序列重复误杀率就明显下降了。6.4 排查工具与调试技巧生产环境的排查不能靠瞎猜。我在编排器里加了一个跟踪快照功能每当任一防护组件被触发就会自动生成一个结构化的事件记录包含触发时间、步骤轨迹、阈值配置、当前状态等。这个快照会单独存一份 JSON 文件方便事后用任何 JSON 工具分析。排查的时候我的步骤固定为先看触发类型是超时、Token超限还是循环再顺着快照里的执行轨迹回溯看故障点是工具外部异常还是模型返回异常最后根据故障点决定调参放宽阈值、改实现修正指纹逻辑还是加白名单允许某些操作重复执行。7. 一些补充心得和后续规划写到这里方法论部分基本讲完了。再说几句我在落地过程中的体会。这套看门狗和熔断器本质上是在和大模型的不确定性做对抗。你要接受一个现实无论提示词写得多么精密Agent 依然可能做出出乎意料的动作。这不是提示词工程能完全解决的它需要系统级的安全机制来做收底。所以与其花大量时间去琢磨怎么让提示词永远不出错不如把这些防护机制当成基础设施的一部分一开始就规划进去。我给开源版的后续规划里还有两个可以重点说的方向第一个是成本预测模块基于历史Token消耗数据的线性回归来预判当前任务最终的消耗上限把保护机制从事后熔断推进到事前预测第二个是多Agent协同保护不是每个Agent各自带一套防护而是在编排层做全局的熔断汇总防止多个Agent之间互相触发导致系统级雪崩。如果你打算在自己项目里落地这套机制我的建议是从最小可用版本开始先装一个看门狗盯住耐心阈值单步超时总时长等跑了一段时间有数据了再逐步加上Token预算和循环熔断。别一开始就追求大而全否则你会在调参上耗费大量时间反而错过了真正该观察的故障模式。项目后续的代码和迭代记录会继续同步到开源仓库里感兴趣的朋友可以关注下一期——到时候我会专门讲讲成本预测模块的算法选型以及那套多Agent协同保护框架的架构设计。 
长河君 · 2026/11/4 5:04:05 · 阅读 1.2k 运行时看门狗与死循环熔断器：防止Agent提示流编排器失控的成本安全机制
1. 项目概述先说个真实场景我在公司内部跑过一版带联网搜索的Agent原本设计好的流程是用户提问 - 检索资料 - 归纳总结 - 输出结论。看起来很简单对吧结果上线第三天那个Agent因为一个工具返回了异常格式的数据在一个解析函数里疯狂重试四十分钟调了上万次大模型接口。等到我发现的时候那个月的API预算已经烧掉了接近三成。这还只是内部测试环境如果跑的是面向用户的付费服务这一晚上赔出去的钱可能就够买一辆代步车了。所以这个开源系列做到第13期我下定决心要做一件事给提示流编排器装上安全带和保险丝——也就是标题里说的运行时看门狗Runtime Watchdog和死循环熔断器Loop Circuit Breaker。简单来说前者盯着Agent运行时的资源消耗和健康状态后者专门拦截那些一旦跑起来就停不下来的循环调用链。目标是让Agent在失控的时候能够被快速发现、快速降级、快速止损而不是等账单出来才发现自己成了信用卡受害者。这套东西适合谁看如果你正在开发Agent应用、搭过工作流编排工具、或者只是对大模型API的成本控制有焦虑那这篇文章值得你花二十分钟读一遍。我不会扯太玄乎的理论全程都是可以直接落到代码里的方案包括我踩过的坑和改了三版才定下来的细节设计。2. 为什么提示流编排器需要看门狗和熔断器2.1 Agent失控的几种典型姿势先说几个我真实遇到过的Agent失控场景你可以对照一下自己有没有踩过类似的坑。第一种叫重试风暴。Agent在调用外部API的时候收到一个异常响应按照代码逻辑它会重试。但问题在于很多Agent框架的重试机制是按次数设计的重试三次、五次就停。可如果调用链比较深A调用B、B调用C、C失败导致B重试、B重试又触发A重试这种级联放大很容易让调用次数指数级增长。第二种叫逻辑死循环。Agent在规划下一步行动的时候因为提示词写得不够严谨或者上下文里出现了矛盾的指令它会反复执行同一个动作搜索同一个关键词、调用同一个工具、输出几乎一样的中间结果然后继续规划、继续执行。每一步看起来都有逻辑但整体上就是在原地打转。第三种叫上下文膨胀。Agent在一次会话中不断往上下文里追加历史记录token数越滚越大。表面上每次调用都正常返回了但延迟在飙升、成本在涨甚至上下文有可能溢出报错。这类失控是最隐蔽的因为从功能上看它没出Bug纯粹是资源消耗失控。无论哪一种症结都是同一个使用方缺少对运行过程的实时观测和主动干预能力。普通的工作流引擎只要流程定了就不会跑偏但基于大模型的Agent天然带有随机性和不确定性你不能指望提示词写得好它就不会失控。这就像开车不能只靠技术好还得装刹车和保险杠。2.2 看门狗和熔断器的职责切分在设计的时候我把这两个组件定位成不同层级的保护机制。运行时看门狗是持续监控器它的动作是观察、预警、干预。它盯着每一步的执行耗时、Token消耗、调用频率、上下文占用一旦发现指标超过阈值要么告警要么自动触发降级操作。它防的是慢性病——成本慢慢失控、延迟慢慢恶化这一类。死循环熔断器则是模式识别器它的动作是识别、切断、隔离。它分析执行轨迹里的循环特征如果发现相同或高度相似的步骤反复出现就直接熔断掉整条执行链。它防的是急性病——逻辑死循环、重试风暴这一类一旦发作就会迅速烧钱的故障。打个比方看门狗是气象台实时监测风速和降雨量预警台风熔断器是防洪闸门当水位超过临界线时一刀切下去保住下游。两者配合才能在Agent真正失控之前把损失控制住。3. 运行时看门狗的设计与实现3.1 需要盯住的核心指标设计看门狗的第一步是确定监控指标。我根据实际踩坑经验筛选了四个核心维度其他指标都是在这四个维度上的延伸。指标维度具体指标建议阈值参考监控目标耗时单步执行耗时、整链总耗时单步不超过30秒整链不超过5分钟识别慢调用和挂死Token消耗每一步的输入/输出Token数、累计Token数单轮任务的Token消耗上限控制成本调用频率单位时间内的API调用次数每分钟不超过20次就是一个合理警戒值拦截重试风暴上下文占用上下文消息条数、Token占用率上下文Token占用率不超过80%防止上下文膨胀举个例子我设置的 Token 警戒线是单轮任务累计消耗不超过 10 万 Token。这个数字不是拍脑袋定的而是根据业务量估算出来的一个典型的复杂任务大概需要 20 到 30 步推理每步平均消耗 3000 到 5000 Token10 万的上限已经留出了差不多一倍的缓冲空间。如果超过这个用量大概率不是正常执行而是什么环节出了问题。3.2 看门狗的工作流程从观察到干预看门狗的完整工作流程我分成了五个阶段每个阶段都有明确的动作和出口条件。第一步是埋点采集。在编排器内部定义一个统一的执行器接口每一次步骤执行开始、结束、失败都会发出事件。事件会带上步骤ID、执行器类型、耗时、Token用量这些元数据。这一步的关键是一个字全。宁可多采集一些暂时用不上的数据也不要漏掉关键节点。第二步是数据聚合。事件流会进入一个轻量级的滑动窗口计数器我实现的是一个环形缓冲区固定窗口大小为 60 秒每秒一个时间桶存储该秒内的调用次数、Token消耗、平均耗时。之所以用滑动窗口而不是简单计数器是因为滑动窗口能真实反映最近的节奏而简单计数器容易把过去的波动掺进当前判断里。第三步是指标计算。每隔一段时间就会基于滑动窗口算一次指标快照跟阈值做比较。这里我用了一个小技巧对耗时指标做指数移动平均EMA公式是ema alpha * current (1 - alpha) * emaalpha 取 0.3。这样做的好处是过滤掉偶发的毛刺比如某个步骤因为网络抖动多花了三秒钟EMA不会因为这一个局部波动就误报警。第四步是状态判定。看门狗分三级状态绿色表示安全黄色表示需要注意红色表示危险。黄色状态下只告警不干预只有红色状态才触发降级动作。第五步才是干预动作。干预分两档一档是软降级比如强制调整Agent的temperature参数降低随机性、切换到更便宜的模型、暂停非关键的旁路工具调用另一档是硬熔断直接终止当前执行链并返回提示信息。注意硬熔断不是Lane逻辑的回滚而是直接中止整个任务宁可让任务失败也不让它继续烧钱。3.3 冷静期与自动恢复熔断不是结束后面还要有恢复策略。打个比方电路跳闸之后你肯定会先检查故障点再合闸。Agent的熔断恢复也是类似我给编排器设计了冷却时间机制默认熔断后 30 秒内禁止同一类型的任务重新发起执行这个冷却时间会根据熔断次数指数退避——第一次30秒、第二次60秒、第三次120秒上限是10分钟。冷却期过去之后系统允许少量任务以灰度模式放行也就是同时使用正常模型和降级模型各跑一次对比两者的结果质量。如果灰度任务在连续五次执行中都没有触发熔断就自动取消降级恢复正常模式。这背后是一个简单的统计置信度逻辑连续成功的次数越多说明故障原因被解决的可能性越大可以逐步恢复信心。看门狗组件里还有一个容易被忽视的细节代码总是先记录、后判断。换句话说即使报警信息没有触发任何熔断动作日志也必须完整落盘。因为很多问题在发生的那一刻看起来无关紧要复盘的时候才发现是早期信号。如果你连信号都没留下排查就只能靠瞎猜了。4. 死循环熔断器的设计与实现4.1 如何识别这步之前做过了看门狗管的是总量失控但死循环熔断器要解决的是逻辑卡死。这两者的核心差异在于看门狗判断是不是太多了用的是数值阈值熔断器判断是不是在重复用的是模式匹配。我的实现思路分三层。第一层是步骤指纹提取。每一次步骤执行结束时根据(执行器类型, 输入参数的哈希值, 返回结果的摘要值)生成一个指纹字符串。所谓输入参数哈希指的是把传给执行器的核心参数序列化之后做一个 SHA-256 摘要这样哪怕参数只有细微差别指纹也会完全不同而返回结果的摘要则是为了防止一种情况——两个不同的输入产生相同的结果那其实也算是收敛了不该算重复。第二层是循环模式判定。在编排器内部维护一个执行轨迹栈每生成一个新的步骤指纹就推入栈中。判定的规则是如果当前指纹跟栈里前面的某个指纹相同并且中间的步骤数少于一个预设值我设的是 5就判定为循环。中间步骤数也被限制的原因很好理解如果两次相同操作之间隔了五十个不同的步骤那更像是一个正常的复杂流程而不是死循环。第三层是次数累计与熔断。如果同一个指纹对在最近 10 个步骤内出现超过 3 次熔断器直接置位。为了记录这些数据我用了一个基于一致性哈希的环形缓冲——它的作用是把最近执行过的步骤指纹均匀地分布到固定长度的缓冲区里既能快速查找又不会因为追踪历史信息越来越多而让内存无限膨胀。4.2 用状态机设计熔断器生命周期熔断器不能只是简单的断与不断那样太粗糙了。我把它设计成了一个带状态迁移的小型状态机总共有四个状态。第一个是CLOSED状态也就是一切正常的闭合状态。此时熔断器只做记录不做拦截。第二个是OPEN状态熔断器的断开状态一旦进入这个状态所有新任务请求都会被直接拒绝连执行入口都不会打开直接返回提示信息。第三个是HALF_OPEN状态也就是半开状态是熔断后试探性的放行少量任务状态用来验证故障是否已经解除。第四个是HALF_OPEN_TIMEOUT如果半开状态下放行的任务在超时时间内没有完成则说明故障环境仍然存在会直接拉回 OPEN 状态。这四个状态之间的迁移条件是硬编码在代码里的允许一定的配置自由度。状态机实现的好处是不会出现已经熔断了但还有漏网任务在执行的尴尬情况——因为所有入口都只有一处在判定状态。4.3 熔断器的触发阈值计算死循环熔断器的核心痛点在于阈值设多少合适。设得太小正常的多步骤任务可能被误杀设得太大又起不到保护作用。我自己的经验是先从最小循环长度入手。一个最小循环是两步执行A、执行B、再执行A。所以我在指纹匹配时设的最低循环长度是 2 步。但考虑到底层大模型偶尔会有非常相近的中间结果我额外要求最近 10 步内同一子序列出现 3 次才算熔断条件。换句话说允许 2 次偶然重复第 3 次就基本可以认定是卡死了。这个 3 次的数字是根据我自己的线上数据统计来的正常任务的重复率极少超过 1 次而真正死循环的任务从第 2 或第 3 次重复开始后续动作几乎必然完全一致。如果你的业务场景对误杀特别敏感建议阈值调到 4 到 5 次再熔断代价是故障期间多烧 30% 左右的费用。这是一个成本和误判率的权衡没有标准答案得看你的钱袋子有多厚。4.4 熔断动作的执行细节熔断器触发之后不是简单地停住就完事了。我在实现里加了两个动作来让熔断更彻底。第一个动作是状态冻结。被熔断的执行链会保留当时的完整上下文快照包括步骤轨迹、输入输出、Token消耗等。这样做不只是为了复盘更是为了下次恢复执行时可以拿到精准的断点位置而不用从零开始重跑。第二个动作是依赖隔离。如果熔断是由某个具体工具触发的比如那个工具连续三次返回同样的异常结果熔断器会把该工具标记为不可用后续的执行链会自动绕开它。这一步很关键因为它能防止熔断之后系统自动走了另一条路结果那条路上也有同一个有问题的工具等于换了个地方继续烧钱。这两个动作加在一起才勉强称得上熔断——毕竟熔断的语义本来就是切掉整段电路不是只切掉一个灯泡。5. 完整实操从配置到代码5.1 配置项设计为了让看门狗和熔断器真正能用起来我把所有参数都做成了配置项而不是写死在代码里。下面这份配置可以直接粘到你的项目里作为起点结合自身场景微调。WATCHDOG_CONFIG { window_seconds: 60, step_timeout: 30, # 单步超时 total_timeout: 300, # 总超时 max_tokens_per_task: 100000, # 单任务Token上限 max_calls_per_minute: 20, # 每分钟调用上限 context_ratio_threshold: 0.8, # 上下文占用比例阈值 ema_alpha: 0.3, # 指数移动平均系数 cool_down_seconds: 30, # 初始冷静期 cool_down_max: 600, # 冷静期上限 } CIRCUIT_BREAKER_CONFIG { min_loop_length: 2, same_subsequence_threshold: 3, recent_steps_buffer_size: 10, half_open_max_tasks: 2, half_open_timeout: 30, }这里有一个值得说明的参数same_subsequence_threshold设成了 3对应的含义是最近 10 步里同一个子序列出现 3 次就算死循环。如果你希望更精确一些可以把这个阈值定得更高但那样熔断器的反应速度会更慢属于少止损和少误杀之间的取舍。5.2 核心代码骨架我把这个系统简化成了一个可以直接跑的版本核心逻辑大约两百行。你可以把它嵌入到任何一个支持事件驱动的编排器里。import hashlib import json import time import threading from collections import deque from enum import Enum class CircuitState(Enum): CLOSED 1 OPEN 2 HALF_OPEN 3 HALF_OPEN_TIMEOUT 4 class StepEvent: def __init__(self, step_id, executor_type, inputs, outputs, elapsed_ms, token_usage): self.step_id step_id self.executor_type executor_type self.inputs inputs self.outputs outputs self.elapsed_ms elapsed_ms self.token_usage token_usage self.fingerprint self._build_fingerprint() def _build_fingerprint(self): input_digest hashlib.sha256( json.dumps(self.inputs, sort_keysTrue, defaultstr).encode() ).hexdigest()[:16] output_digest hashlib.sha256( json.dumps(self.outputs, sort_keysTrue, defaultstr).encode() ).hexdigest()[:16] return f{self.executor_type}:{input_digest}:{output_digest} class SlidingWindowCounter: def __init__(self, window_seconds60): self.window_seconds window_seconds self.buckets {} self.lock threading.Lock() def add(self, value, bucket_keyNone): bucket_key bucket_key or int(time.time()) with self.lock: self.buckets[bucket_key] self.buckets.get(bucket_key, 0) value cutoff bucket_key - self.window_seconds self.buckets {k: v for k, v in self.buckets.items() if k cutoff} def sum(self, current_timeNone): current_time current_time or int(time.time()) cutoff current_time - self.window_seconds return sum(v for k, v in self.buckets.items() if k cutoff) class RuntimeWatchdog: def __init__(self, config): self.config config self.call_counter SlidingWindowCounter(config[window_seconds]) self.token_counter SlidingWindowCounter(config[window_seconds]) self.ema_step_time None self.state GREEN self.trigger_limit config[max_tokens_per_task] def on_step_completed(self, event): # 记录指标 self.call_counter.add(1) self.token_counter.add(event.token_usage) if self.ema_step_time is None: self.ema_step_time event.elapsed_ms else: alpha self.config[ema_alpha] self.ema_step_time alpha * event.elapsed_ms (1 - alpha) * self.ema_step_time # 状态判定 total_tokens self.token_counter.sum() calls_per_minute self.call_counter.sum() step_timeout self.config[step_timeout] * 1000 if total_tokens self.trigger_limit or calls_per_minute self.config[max_calls_per_minute]: self.state RED return BREAK if self.ema_step_time step_timeout: self.state YELLOW return WARN self.state GREEN return OK class LoopCircuitBreaker: def __init__(self, config): self.config config self.state CircuitState.CLOSED self.recent_steps deque(maxlenconfig[recent_steps_buffer_size]) self.sequence_count {} self.action_lock threading.Lock() def _extract_subsequences(self): sequences [] recent_list list(self.recent_steps) if len(recent_list) self.config[min_loop_length]: return sequences step self.config[min_loop_length] for i in range(len(recent_list) - step 1): sequences.append(tuple(recent_list[i:i step])) return sequences def on_step_completed(self, event): fingerprint event.fingerprint with self.action_lock: if self.state CircuitState.OPEN: return BREAK self.recent_steps.append(fingerprint) for seq in self._extract_subsequences(): self.sequence_count[seq] self.sequence_count.get(seq, 0) 1 if self.sequence_count[seq] self.config[same_subsequence_threshold]: self.state CircuitState.OPEN return BREAK return OK这段代码是演示用的但已经能跑通核心逻辑。真正接入生产环境的时候你需要把on_step_completed挂到编排器的执行回调上通常是在每一步执行完await返回的位置上调用。5.3 把两者接入编排器看门狗和熔断器是独立的组件但实际要协同工作。我在编排器里用一个名为ExecutorGuard的外层包装器来统一调用它们。class ExecutorGuard: def __init__(self, watchdog, breaker): self.watchdog watchdog self.breaker breaker async def execute(self, step_id, executor_type, inputs, executor_fn): event_start time.time() try: # 检查断路器状态 if self.breaker.state CircuitState.OPEN: return {status: rejected, reason: circuit_open} # 执行真正的步骤 result await executor_fn(inputs) event StepEvent( step_idstep_id, executor_typeexecutor_type, inputsinputs, outputsresult, elapsed_ms(time.time() - event_start) * 1000, token_usageresult.get(token_usage, 0), ) # 先过熔断器 breaker_decision self.breaker.on_step_completed(event) # 再看门狗 watchdog_decision self.watchdog.on_step_completed(event) if breaker_decision BREAK or watchdog_decision BREAK: return {status: terminated, reason: guard_triggered} return {status: ok, data: result} except Exception as e: return {status: error, message: str(e)}通过这个包装器所有执行步骤都汇聚到同一个保护伞下。这里有一个细节值得注意breaker和watchdog是顺序判断的先判断是不是循环再判断是不是超限。这个顺序有讲究——如果先做资源判断可能会因为一个循环任务瞬时消耗过多资源而直接把任务杀掉导致你想在日志里观察循环轨迹都来不及。先让熔断器识别出这是一个循环再让看门狗确认这个循环确实在烧钱这样生产环境里能留下更完整的故障证据链。5.4 在异步事件循环里防重入写Agent编排器的时候最让人头疼的一个细节是异步重入。大模型调用大多是异步的这意味着on_step_completed会被并发触发。如果熔断器的sequence_count不是线程安全的多个回调同时写同一个字典轻则丢数据重则死锁。我的做法是在两个组件里都加了threading.Lock但这还不够。内层锁可以保护共享数据却解决不了检查-再执行之间的竞态条件。比如两个并发回调同时通过了breaker.state OPEN的检查然后一起进入了真实执行逻辑。这会让熔断器形同虚设。要真正防住并发穿透需要在ExecutorGuard.execute入口处加一个跨步骤的信号量Semaphore把真正能同时进入执行逻辑的任务数限制在一到两个。我选了一宁可让执行过程串行也不冒并发穿透的风险。如果追求吞吐可以放宽到二但那样还需要更复杂的锁机制来保证状态一致。这个就是典型的用吞吐换安全的取舍。6. 从生产环境里趟出来的问题和排查思路6.1 问题一重试风暴伪装成正常调用有一次我监控到看门狗频繁触发每分钟调用数超限的告警但熔断器毫无反应。查了执行轨迹才发现这个任务的步骤指纹每次都不同因为每次重试的时候Agent 都会在输入参数里加一个尝试次数的字段。从语义上看明明就是同一个操作不停重来但从指纹哈希的角度来说每次都不同熔断器自然识别不出来。排查之后我改了规则指纹的输入部分只取业务核心参数的哈希明确排除重试次数、时间戳这类噪声字段。更优雅的做法是给执行器定义指纹关键字段的白名单把哪些字段参与哈希计算的决定权下放给各个执行器自己管理。这样每个执行器都清楚自己的哪些输入是业务相关、哪些只是控制参数。6.2 问题二回调函数污染了主流程第一次集成熔断器的时候我直接把判断逻辑塞进了 Agent 推理循环的核心路径上。后来发现一个隐蔽的坑熔断器自身抛出的异常会干扰原来的业务逻辑。比如某个步骤超时了看门狗直接把任务中断但因为中断的异常类型和业务里的超时异常长得一样下游的兜底逻辑把它当成正常超时处理又触发了一次重试。等于保护机制自己给故障续了一命。解决方案是把保护逻辑做成旁路模式全部通过事件订阅和回调触发不直接跟业务代码共享异常语义。Guard 的返回值统一走{status: terminated}这样的协议结构业务代码只要按这个协议响应即可不会导致异常类型混乱。6.3 问题三误杀正常的长链任务熔断器上线之后我收到过一个有点棘手的投诉——一个正常的调研摘要任务执行了 12 步在中间某一步被熔断了。日志显示任务里有两个步骤恰好都调用了同一个搜索工具搜索关键词还类似于是指纹判定它们构成子序列重复触发了熔断。这个问题的根源是min_loop_length2太敏感了。解决办法是引入了区分度判断两个步骤不仅指纹要一致还要求后续的行为轨迹也要一致。我加了一个展望窗口的逻辑——只有当[A, B, A]这个三连模式出现并且在展望窗口内确认 A 之后的下一步也是 B才真正判定为循环。这样把单点重复变成序列重复误杀率就明显下降了。6.4 排查工具与调试技巧生产环境的排查不能靠瞎猜。我在编排器里加了一个跟踪快照功能每当任一防护组件被触发就会自动生成一个结构化的事件记录包含触发时间、步骤轨迹、阈值配置、当前状态等。这个快照会单独存一份 JSON 文件方便事后用任何 JSON 工具分析。排查的时候我的步骤固定为先看触发类型是超时、Token超限还是循环再顺着快照里的执行轨迹回溯看故障点是工具外部异常还是模型返回异常最后根据故障点决定调参放宽阈值、改实现修正指纹逻辑还是加白名单允许某些操作重复执行。7. 一些补充心得和后续规划写到这里方法论部分基本讲完了。再说几句我在落地过程中的体会。这套看门狗和熔断器本质上是在和大模型的不确定性做对抗。你要接受一个现实无论提示词写得多么精密Agent 依然可能做出出乎意料的动作。这不是提示词工程能完全解决的它需要系统级的安全机制来做收底。所以与其花大量时间去琢磨怎么让提示词永远不出错不如把这些防护机制当成基础设施的一部分一开始就规划进去。我给开源版的后续规划里还有两个可以重点说的方向第一个是成本预测模块基于历史Token消耗数据的线性回归来预判当前任务最终的消耗上限把保护机制从事后熔断推进到事前预测第二个是多Agent协同保护不是每个Agent各自带一套防护而是在编排层做全局的熔断汇总防止多个Agent之间互相触发导致系统级雪崩。如果你打算在自己项目里落地这套机制我的建议是从最小可用版本开始先装一个看门狗盯住耐心阈值单步超时总时长等跑了一段时间有数据了再逐步加上Token预算和循环熔断。别一开始就追求大而全否则你会在调参上耗费大量时间反而错过了真正该观察的故障模式。项目后续的代码和迭代记录会继续同步到开源仓库里感兴趣的朋友可以关注下一期——到时候我会专门讲讲成本预测模块的算法选型以及那套多Agent协同保护框架的架构设计。