企业官网建设流程全解析

从 0 到 1 打造 AI 提示流编排器：别让 Agent 刷爆你的信用卡！运行时看门狗与死循环熔断器设计（开源系列 13）

如果你也搞过稍微复杂一点的 Agent 应用，一定遇到过这种场景：凌晨三点被手机震醒，打开一看，GPT API 的消耗账单在半小时内飙了上百美元。排查日志发现，某个 Agent 在工具调用和模型返回之间陷入了一个逻辑死循环——它以为自己每一步都在“推进”，实际上只是在同一个错误结果上反复重试，每次都要完整地调用一次大模型接口。等我手动 kill 进程的时候，它已经默默跑了上千轮。

这种问题，本质上是“失控的循环”没有被及时发现和终止。我在做提示流编排器（Prompt Flow Orchestrator）的时候，把这个问题当成头等大事来处理，最终沉淀出两个关键组件：运行时看门狗（Runtime Watchdog）和死循环熔断器（Loop Circuit Breaker）。这一篇就把完整的设计思路、实现原理和踩坑过程分享出来，针对的就是所有做 Agent 开发、接大模型 API 后需要控制成本和保障稳定性的朋友。

1. 提示流编排器为什么需要看门狗和熔断器

1.1 Agent 失控的底层原因分析

先说清楚一个容易被忽视的事实：Agent 本身并不知道自己在“循环”。它只是在执行“根据当前状态，调用工具/模型，更新状态”这个动作。真正的问题出在：状态的变化可能是无效的，但 Agent 无法感知这种无效性。

举个例子。我有一个很常见的工具函数，用来从数据库里读取用户订单。这个函数内部做了重试，网络抖动时会抛异常，Agent 捕获异常后决定“再试一次”。第一次异常，第二次异常，第三次还是异常。第四次，Agent 换了个说法，把请求参数里的时间范围扩大了，但它不知道这个扩大后的时间范围其实不合法，数据库直接返回空结果。Agent 看到空结果认为是“查无数据”，于是又换了个条件继续查……每一步都有足够的日志，每一步都看起来像“在执行任务”，但整体来看就是在一个无效路径上反复打转。

为什么会这样？因为大模型是一个概率系统，它不保证每次输出都收敛。当 prompt 里的“意图”和“上下文”存在轻微歧义时，模型完全可能选出一条与目标不相关的路径，然后沿着这条路径持续输出。这也是为什么很多 Agent 框架只做了任务调度的编排，但对“执行质量”的感知几乎为零。

1.2 成本失控的数学原理

算一笔账就明白问题的严重性。假设一个 Agent 每轮循环需要调用一次大模型接口，单次消耗 2000 个 token，按当前主流模型的定价折算约 0.01 美元。看起来不多，但如果循环没人管，每分钟跑 10 轮，一小时就是 600 轮，直接烧掉 6 美元。如果是更复杂的任务链，每轮还附带工具调用、向量检索等多重费用，这个数字要乘以 3 到 5 倍。

更隐蔽的是，很多 API 是按 token 预扣费的，循环一旦失控，账单是即时滚动的，根本来不及人工干预。所以，必须在运行时层面内置终止机制，不能指望模型自己“幡然醒悟”。

1.3 看门狗与熔断器的职责划分

看门狗和熔断器听起来有点像，但职责完全不同。

• 看门狗：负责监控“执行过程是否健康”，检测指标包括单轮执行时长、总轮数、token 消耗速率、状态变更有效性。它管的是“度量”。
• 熔断器：负责在检测到异常后执行“止损动作”，包括暂停当前 Agent 流、切换降级策略、发送告警。它管的是“反应”。

简单类比：看门狗是体检医生，熔断器是急救医生。一个负责发现毛病，一个负责动手抢救。两者必须配合，否则只有检测没有反应，或者只有熔断没有诊断，效果都会大打折扣。

2. 运行时看门狗的整体设计

2.1 看门狗的核心指标设计

我在实现看门狗时，参考了硬件看门狗电路的设计思路：硬件看门狗会定期“喂狗”，如果超时未喂狗就强制复位系统。软件层面不能直接“复位”，但可以设计一套可量化的指标，实时判断当前 Agent 运行是否处于健康状态。

看门狗需要监控的指标，我归纳为四类：

第一类：运行时长相关

• 单轮执行时间（单次模型调用 + 工具调用 + 状态更新的总耗时）
• 单步执行时间（工具调用前/后的耗时分解）
• 累计运行时间

第二类：调用频次相关

• 模型调用总次数
• 工具调用总次数
• 多 Agent 协作场景下的消息传递次数

第三类：成本消耗相关

• 累计输入 token 数
• 累计输出 token 数
• 预估费用（可按模型单价测算）

第四类：状态健康相关

• 状态变更 delta（每轮执行之后，Agent 的核心状态是否发生变化）
• 结果熵值（连续多轮返回的结果是否具有足够的分辨度）
• 无效操作比例（比如失败重试、空结果、异常分支的占比）

这四个维度里，状态变更 delta 是最容易被忽视但最有价值的信号。很多时候 Agent 明明在“跑”，但它的核心状态（比如“当前已获取的信息列表”）已经连续几轮没有任何新增了，这就是典型的“原地踏步”。单看轮数或者耗时，根本发现不了这种问题。

2.2 实时检测与刻度采样策略

看门狗不能每轮都做全量统计，那样性能损耗太大。我采用的策略是双刻度采样：

• 细刻度（每轮采样）：只更新计数器和累计值，不做复杂判断。每轮结束后，更新模型调用次数、token 累计数、状态哈希值。
• 粗刻度（每 N 轮或每 5 秒采样）：基于细刻度累积的数据做健康度评估，包括计算状态变化率、token 消耗速率等。

这个设计的核心权衡在于：细刻度要保持极低的延迟和资源消耗，粗刻度才能做有意义的决策。每轮都计算熵值在逻辑上没问题，但实际跑起来会发现，高频率的哈希计算和条件分支对 Python 这类解释型语言来说性能损耗不小，尤其在长流程里会放大。

2.3 看门狗数据模型设计

看门狗的数据结构要能支撑多轮状态追溯，我用了一个轻量级的快照模型：

@dataclass class StepSnapshot: step_index: int status_hash: str token_input: int token_output: int tool_calls: int elapsed_ms: int result_digest: str @dataclass class WatchdogState: agent_id: str snapshots: list[StepSnapshot] total_tokens: int = 0 total_cost: float = 0.0 last_trigger: str | None = None

每个快照只保留核心字段，方便后续做趋势判断。status_hash是我单独计算的状态指纹，会把 Agent 当前的关键状态（比如所有已收集信息的排序列表、当前正在等待的结果集合）做一次稳定哈希。

这样在检测“状态是否有变化”时，只需要比较相邻快照的哈希值，而不需要真的去理解状态内容。这个技巧在处理复杂 Agent 状态时非常实用，因为状态往往是一个嵌套字典，直接比对重量大，也不方便持久化存储。

3. 死循环熔断器的核心机制

3.1 三种熔断模式的触发阈值设计

就算有看门狗持续采样，没有熔断逻辑去执行止损还是白搭。熔断器必须预设几种明确的触发模式，每类模式对应不同的阈值和止损策略。

我实际的触发阈值表如下：

这里面，值得单独聊的是“重复结果”检测。很多 Agent 死循环并不是状态完全不变，而是状态在微小变化，比如每次都往一个列表里追加一个元素，但追加的内容几乎一样。这种表面上的“动态”是伪装性最强的。我用了一个简单的相似度判断：把每次模型输出的摘要向量化，计算相邻结果的余弦相似度，超过阈值就判定为重复。

3.2 断路器状态机：关闭、开启与半开

熔断器本身也是一个有限状态机，我实现了标准的“关闭 → 开启 → 半开”三态模型：

• 关闭（CLOSED）：一切正常，请求正常通过，看门狗持续计数。
• 开启（OPEN）：熔断生效，所有请求直接短路拒绝，快速失败。
• 半开（HALF_OPEN）：熔断达到冷却时间后，允许少量探测请求通过，验证是否恢复。

在 Agent 编排场景里，我做了两处调整：

一是冷却时间动态调整。标准熔断器的冷却时间通常是固定值，但在 Agent 场景里，简单问题冷却 30 秒就够了，复杂任务可能需要 5 分钟。我根据触发熔断时的任务复杂度动态调整冷却时间：如果熔断发生时已执行超过 20 轮，冷却时间翻倍。

二是半开状态的探测请求不是真实业务请求，而是用一个最小化的“ping prompt”验证模型是否恢复正常。这样既不影响业务正确性，又能快速探活。

3.3 熔断器的恢复策略：不要一刀切

熔断之后怎么恢复正常？我的实践结论是：不要自动全量恢复。很多 Agent 任务是有状态的，直接全量重跑不仅浪费成本，还可能再次触发同样的死循环。

我采用的恢复策略分三档：

• 冷恢复：适合无状态短任务，熔断结束后直接从初始状态重跑。
• 温恢复：适合有状态的中间任务，从最后有效的 checkpoint 恢复，跳过已完成的部分。
• 降级执行：熔断触发后，不重启完整 Agent 链，而是退化为最简单的一问一答模式，无论确率优先于完整度——这个在处理用户紧急诉求时特别重要，总比让用户干等强。

默认选择温恢复，因为大部分真实场景里，失败前已经拿到不少有效信息，直接扔掉太可惜。

4. 编排器的整体架构与集成实现

4.1 编排器的运行流程设计

在加入看门狗和熔断器之后，编排器的运行时流程变成了这样：

class PromptFlowOrchestrator: def __init__(self, name, steps, model_client, watchdog, breaker): self.name = name self.steps = steps self.model_client = model_client self.watchdog = watchdog self.breaker = breaker async def run(self, initial_context): # 熔断器先做前置检查：如果已经处于 OPEN 状态，直接拒绝 if not self.breaker.is_accessible(): return {"status": "blocked", "reason": "circuit_open"} context = initial_context result = None # 进入编排主循环，每一步都向看门狗上报数据 for step_index in range(len(self.steps)): # 运行前先检查熔断器状态，避免已开启时继续消耗资源 if not self.breaker.is_accessible(): await self.watchdog.record_breach(step_index, "circuit_open") break step = self.steps[step_index] step_start = time.perf_counter() try: result = await self._execute_step(step, context) step_success = True except ToolExecutionError as e: result = {"error": str(e), "partial": e.partial_data} step_success = False step_elapsed_ms = time.perf_counter() - step_start # 核心环节：向看门狗上报本轮快照 should_break = await self.watchdog.check_step( agent_id=self.name, step_index=step_index, context=context, result=result, step_elapsed_ms=step_elapsed_ms, step_success=step_success, ) if should_break: break # 更新上下文，进入下一步 context = self._merge_context(context, result) # 最后统一上报，让看门狗生成完整报告 await self.watchdog.finalize(self.name) return {"status": "completed", "result": result}

核心要点在于：看门狗的检查是同步嵌入到主循环里的，每走一步都会检查。熔断器的检查则在两个位置执行：主循环入口和每一步开始时。这样即使某一步执行时间特别长，也不会跳过熔断判断。

4.2 看门狗触发判定逻辑的实现

看门狗的核心是check_step方法。这个方法要综合判断多种信号，但性能要足够快，不能成为编排器的瓶颈。我的精简实现如下：

class RuntimeWatchdog: def __init__(self, max_steps=50, max_seconds=600, max_cost=5.0, stall_step_threshold=8): self.max_steps = max_steps self.max_seconds = max_seconds self.max_cost = max_cost self.stall_step_threshold = stall_step_threshold self.state = {} async def check_step(self, agent_id, step_index, context, result, step_elapsed_ms, step_success): snap = StepSnapshot( step_index=step_index, status_hash=self._hash_state(context), token_input=self._estimate_tokens(context), token_output=self._estimate_tokens(result), tool_calls=1 if step_success else 0, elapsed_ms=int(step_elapsed_ms), result_digest=self._digest(result), ) self.state[agent_id].snapshots.append(snap) # 基础指标判断 if step_index >= self.max_steps: self.trigger(agent_id, "max_steps_exceeded") return True elapsed_total = sum(item.elapsed_ms for item in self.state[agent_id].snapshots) if elapsed_total / 1000 >= self.max_seconds: self.trigger(agent_id, "max_seconds_exceeded") return True # 成本判断：根据 token 估算费用 total_tokens = sum(item.token_input + item.token_output for item in self.state[agent_id].snapshots) est_cost = self._estimate_cost(total_tokens) if est_cost >= self.max_cost: self.trigger(agent_id, "max_cost_exceeded") return True # 状态停滞检测 recent = self.state[agent_id].snapshots[-self.stall_step_threshold:] if len(recent) == self.stall_step_threshold: hashes = [item.status_hash for item in recent] if len(set(hashes)) == 1: self.trigger(agent_id, "state_stalled") return True # 重复结果检测 digests = [item.result_digest for item in self.state[agent_id].snapshots[-6:]] if len(digests) == 6 and len(set(digests)) <= 2: self.trigger(agent_id, "result_repetition") return True return False def _hash_state(self, context): serialized = json.dumps(context, sort_keys=True, ensure_ascii=False) return hashlib.sha256(serialized.encode()).hexdigest()[:16] def _digest(self, result): try: return json.dumps(result, sort_keys=True, ensure_ascii=False)[:200] except (TypeError, ValueError): return str(result)[:200] def _estimate_cost(self, total_tokens): # 粗略估算，按输入 0.001 美元/1k token 计算 return total_tokens * 0.001 / 1000

这里有几个细节要解释一下。

关于状态哈希：_hash_state对 context 做了 sort_keys 处理再哈希。这样做的目的是保证同一个状态的两种不同 key 顺序也能得到相同的哈希值，避免因为 Python 字典遍历顺序不同而产生的误判。我用的是 sha256 的前 16 位，碰撞概率在实际场景里可以忽略。

关于成本估算：_estimate_cost用了极其粗糙的估算模型。实际项目里应该根据具体模型单价分输入/输出 token 分别计算，我这里为了保持代码简洁做成了统一的粗略估算。如果你接的是 OpenAI 的 API，建议把input和output分开算，因为两者的单价差了好几倍。

关于重复结果检测：我取的最近 6 轮的 result digest，如果这 6 轮里出现相同摘要的次数不超过 2 种，就判定为重复。这个启发式规则来自我的观察：正常 Agent 每轮输出即使结论相似，摘要也会有形态差异。真正的死循环输出通常会高度雷同。

4.3 熔断器状态机的完整实现

熔断器我在项目中封装成了独立类，便于复用：

class CircuitBreaker: CLOSED = "closed" OPEN = "open" HALF_OPEN = "half_open" def __init__(self, failure_threshold=3, recovery_timeout=60, half_open_max_calls=1): self.failure_threshold = failure_threshold self.recovery_timeout = recovery_timeout self.half_open_max_calls = half_open_max_calls self.state = self.CLOSED self.failure_count = 0 self.last_opened_at = None self.half_open_calls = 0 def is_accessible(self) -> bool: if self.state == self.CLOSED: return True if self.state == self.HALF_OPEN: return self.half_open_calls < self.half_open_max_calls # OPEN 状态判断是否超过冷却期，自动转 HALF_OPEN if self.last_opened_at and (time.time() - self.last_opened_at) >= self.recovery_timeout: self.state = self.HALF_OPEN self.half_open_calls = 0 return True return False def record_success(self): if self.state == self.HALF_OPEN: self.half_open_calls += 1 # 连续成功探测后，判定系统恢复 if self.half_open_calls >= self.half_open_max_calls: self.state = self.CLOSED self.failure_count = 0 self.half_open_calls = 0 elif self.state == self.CLOSED: self.failure_count = max(0, self.failure_count - 1) def record_failure(self): if self.state == self.CLOSED: self.failure_count += 1 if self.failure_count >= self.failure_threshold: self.trip() elif self.state == self.HALF_OPEN: self.trip() def trip(self): self.state = self.OPEN self.last_opened_at = time.time() self.failure_count = 0 self.half_open_calls = 0

这个熔断器的关键技巧是：OPEN 状态本身不做定时唤醒。它在每次被查询时，通过时间戳判断是否已经过了冷却期，然后自动迁移到 HALF_OPEN。这样省去了额外的心跳线程或定时器，实现非常轻量。

HALF_OPEN 状态下，我会限制最多放行 1 个探测请求。如果这个探测请求失败了，熔断器会直接再次进入 OPEN 状态，冷却时间重新计算。如果成功了，说明系统恢复，闭合并重新计数。

4.4 看门狗与熔断器的联动时机

这两个组件不是各自独立工作，而是通过事件机制联动的。看门狗在触发异常时，会把异常类型传给熔断器，然后由熔断器决定是快速失败还是半开探测。

class WatchdogCircuitBridge: def __init__(self, watchdog, breaker): self.watchdog = watchdog self.breaker = breaker async def verify_step(self, agent_id, step_index, context, result, elapsed_ms, step_success): should_break = await self.watchdog.check_step( agent_id=agent_id, step_index=step_index, context=context, result=result, step_elapsed_ms=elapsed_ms, step_success=step_success, ) if should_break: # 看门狗发现异常，通知熔断器记录一次失败 self.breaker.record_failure() else: self.breaker.record_success() # 如果熔断器已经 OPEN，强制终止 if self.breaker.state == CircuitBreaker.OPEN: return True return False

这种设计的好处是，即使某轮看门狗认为一切正常，但之前的连续异常已经让熔断器进入了半开或开启状态，也能被及时拦截。

5. 工具调用层的特殊防护

5.1 工具超时与并发限制

编排器的看门狗主要管 Agent 主循环，但工具调用层本身也要有独立防护。工具是 Agent 的“手”，也是最容易出问题的环节：慢工具、死锁工具、异常消耗资源的工具，任何一个卡住都会让整条 Agent 流停滞。

我实现的工具调用防护有三层：

第一层是单次超时。每个工具调用都必须设定超时时间，默认 30 秒。超时后直接抛出异常，把控制权交还给编排器。这里有一个坑：asyncio.wait_for 在 Python 里虽然好用，但要注意它取消的是协程而不是底层线程。如果你的工具封装了同步的阻塞调用（比如 requests.get），协程取消不会中断底层的 HTTP 请求，会造成孤儿线程。我用asyncio.to_thread把同步调用放到线程池后，配合wait_for限制等待时间，这样超时后的线程池任务仍会跑完但结果直接被丢弃，至少不会阻塞主循环。

第二层是并发上限。同一 Agent 内同时执行的工具调用数不能超过 2 个。超过的话直接排队等待，而不是全部放行。我之前踩过坑：一个多 Agent 协作编排器，某个步骤里有 5 个工具都可以并行调用，结果 5 个请求同时出去了，日志看着效率很高，实际对下游 API 的限流命中率暴涨，经常返回 429 导致重试风暴。

第三层是幂等控制。对于可能产生副作用的工具（比如发送邮件、扣减库存、写数据库），必须要求实现幂等 key。编排器在调用这类工具前，会检查幂等 key 是否已存在，如果存在说明是重试请求，直接返回上一次的结果，不再真正执行。这在配合熔断器做温恢复时特别重要，保证了重新执行流程不会产生重复副作用。

5.2 工具失败重试策略

工具失败时，不能盲目重试。我的策略是基于错误类型区分处理：

class RetryPolicy: RETRYABLE_ERRORS = (TimeoutError, ConnectionError, RateLimitError) NON_RETRYABLE_ERRORS = (InvalidArgumentError, PermissionDeniedError, NotFoundError) def should_retry(self, error): return isinstance(error, self.RETRYABLE_ERRORS) def get_delay(self, attempt): # 指数退避，最多等待 10 秒 return min(2 ** attempt, 10)

为什么这套策略值得参考？因为 Agent 场景里，工具失败的后果可能要等好几轮以后才能暴露。如果每次都立刻重试，不仅消耗 token，还可能放大故障面。我建议把重试次数限制在 2 次以内，2 次都失败就直接让编排器做降级处理，不再执着于这个工具。

6. 实际测试方法与压测记录

6.1 模拟死循环的三种测试样例

要把熔断器做扎实，必须用真实场景测试。我在测试阶段构造了三种经典的死循环样本：

样例一：无效状态更新。Agent 需要搜集用户偏好，但工具每次返回相同的数据。由于代码逻辑里有一个“如果搜到数据就加入列表”的 bug，列表每轮都在变，但内容其实是重复的。这个用例用来测试重复结果检测是否有效。

样例二：同工具连续失败触发降级。某个工具 5 次连续抛异常，熔断器需要把它暂时禁用，然后走降级分支。注意这里熔断的不是整个 Agent，而是针对单一工具的资源级熔断，我专门为这类场景做了工具级熔断器，规则与全局熔断器类似，但作用域更小。

样例三：状态停滞但输出变化。Agent 每轮都输出不同的话术，比如“让我再看看”“换个思路”“再试试”，但 core context（比如已获取的订单列表）完全没有变化。这个用例专门测试状态哈希是否有效——只看输出文本的话，每轮都不一样，容易误判为正常。

这个样例三是跳坑最多的地方。因为大模型的输出天然有随机性，即使 agent 在做重复的事，输出措辞也会产生变化。只看输出相似度很容易漏判，只有结合状态哈希才能确认 Agent 确实没有推进。

6.2 压测数据与调参记录

我跑了一组基础压测，给一个需要调用 3 个工具链的 Agent 注入了死循环场景，记录不同配置下熔断器的表现：

实测结论：循环次数上限设 50、状态停滞阈值设 8，是当前任务类型下比较平衡的配置。如果阈值设太紧，一些复杂的推理任务（尤其是涉及多步工具调度的）会被误杀，用户会抱怨“莫名其妙被中断”。这也提醒了我：阈值参数应该支持按任务类型自动调整，不同类型的 Agent 要有不同的默认值。

7. 常见问题与排查技巧实录

7.1 熔断器误触发导致正常任务被打断

现象：一个耗时长但完全正常的多步推理任务，跑了 40 多轮，结果被成本熔断器拦下。

排查过程：查日志发现，预估费用计算方式有误，我给 token 估算时把缓存 token 也算了进去，导致费用被高估。另一个原因是模型输出特别长，单轮输出就 4000 token，30 轮下来累计费用确实相当可观。

解决方案：把缓存 token 单独计量，不计入熔断成本。同时给成本熔断器增加了“时间窗口”判断，只看最近 5 分钟的消耗速率，而不是累计总消耗。因为累计消耗过高可能只是因为任务本身复杂，但消耗速率过高才是真正的失控信号。

7.2 Agent 卡死但看门狗没有触发

现象：Agent 在等待某个外部服务的回调，主循环一直阻塞在 await 上，看门狗的单轮超时设置没生效。

排查过程：接的是流式响应（streaming），模型首 token 很快返回，但 Agent 方在等待完整响应时，流已经断了，客户端却不知道，一直挂在那里等。

解决方案：给流式读取也加上独立的超时时间。流式响应看似简单，但“连接建立”和“全部数据到达”之间可能隔了非常久，需要一个单独的空闲超时（idle timeout）。我把这个参数暴露成配置项，默认 60 秒无新数据即判定超时，强制断开流式会话。

7.3 熔断恢复后任务重复执行

现象：温恢复模式下，Agent 从 checkpoint 继续跑，但之前已经发送过的邮件又发送了一遍。

排查过程：checkpoint 保存的是上下文状态，但工具调用的副作用记录没有同步保存。恢复后编排器重新跑了前一步的工具调用，而恰好该工具没有实现幂等。

解决方案：checkpoint 必须包含已执行工具的幂等记录。恢复前先查询幂等表，已执行的直接跳过。这也是我前面强调工具层幂等设计重要的原因，没有这层防护，任何恢复策略都可能造成业务事故。

7.4 看门狗在大并发下性能退化

现象：同时跑 10 个 Agent 编排任务，看门狗每轮都做状态哈希和摘要计算，导致整体吞吐下降明显。

排查过程：profile 发现瓶颈在json.dumps(context, sort_keys=True)。复杂任务的 context 可能非常大（包含多轮对话历史、工具返回结果），dump 一次要花几十毫秒。

解决方案：改为只对 context 的“摘要子集”做哈希，比如只取最近 3 轮的关键字段，而不是全量序列化。这个子集虽然信息不完整，但对于检测“状态是否变化”已经足够。代价是某些极端情况下哈希没变化但实际内容变了——我可以接受，因为看门狗本来就是一个“不完美但足够好”的护栏。

8. 开源化过程中的经验与扩展方向

8.1 从自用脚本到开源库的架构演进

这套看门狗和熔断器最初只是写在一个项目里的几个函数，后面越用越顺手，才决定抽成独立模块开源。抽取过程中我最大的体会是：组件边界一定要清晰。看门狗不直接调用 API，熔断器不直接读日志，它们通过事件接口通信。这样使用者可以自由替换实现。

我在开源库里保持了这种解耦：

• Watchdog只负责采集和判定，触发时发出事件。
• CircuitBreaker只负责状态机和放行决策，不关心业务细节。
• Orchestrator负责串起来，并向外暴露配置入口。

8.2 社区反馈驱动的三项改进

开源后陆续收到一些反馈，挺有启发的。印象最深的三条：

有用户提出，看门狗的单轮耗时检测不能只看总耗时，要关注“模型调用耗时”和“工具调用耗时”的分布。因为有时候总耗时完全正常，但模型调用只花了 200ms、工具调用花了 8 秒，这通常说明工具侧有问题。

另一个用户建议增加“结果相似度”的自适应阈值，因为他的业务里正常结果就是高度雷同的（比如查询天气，几百个城市返回的格式都类似）。我后来实现了基于滑动窗口的自适应阈值，把默认相似度阈值从固定值改成了“历史均值的 1.2 倍”。

还有一位做客服系统的朋友建议把告警接入企业微信/钉钉机器人，这样就算 Agent 在凌晨自己重启了，也能及时收到通知。这属于扩展功能，我已经在 Roadmap 里排上。

8.3 后续可以扩展的方向

这套机制后续值得补的方向，我自己的判断是：

第一，预算管理与多项目配额。现在只是单 Agent 级别的熔断，还没做团队级别的预算池管理。如果你在一个团队里同时跑几十个 Agent，需要一个总体的预算分配机制，某个 Agent 超预算不能直接影响其他 Agent。

第二，基于分数的最终答案可靠性验证。熔断器只解决“停止”问题，没解决“答案质量”问题。一些失控 Agent 虽然没陷入死循环，但最终输出的答案明显偏离用户意图。这类问题需要一套独立的“答案可靠性评分”机制来兜底。

第三，编排器持久化与可观测性集成。看门狗的快照数据如果能直接对接 Prometheus 或 ClickHouse，就可以做长时间的离线分析，找出某个提示词模板长期诱发死循环的模式。这是很有价值的优化方向。

我个人在项目演进中的体会是，AI Agent 的开发不能只盯着“正确率”和“效果指标”，运行时的稳定性和成本可控性同样决定了一个产品能不能真正常态化跑下去。正确率再高的 Agent，如果会偶发失控烧钱，产品团队就不敢放开让用户大量使用。看门狗和熔断器这套机制，本质上是在给 Agent 上一道安全保险，让它可以更放心地去执行那些长链路、多工具的任务。而且你会发现，这些稳定性措施设计得越早，后面的迭代成本就越低。真要等线上出了问题再补，那种被迫救火的压力，谁经历谁知道。