企业官网建设流程全解析

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 需要盯住的核心指标

设计看门狗的第一步，是确定监控指标。我根据实际踩坑经验筛选了四个核心维度，其他指标都是在这四个维度上的延伸。

举个例子，我设置的 Token 警戒线是"单轮任务累计消耗不超过 10 万 Token"。这个数字不是拍脑袋定的，而是根据业务量估算出来的：一个典型的复杂任务大概需要 20 到 30 步推理，每步平均消耗 3000 到 5000 Token，10 万的上限已经留出了差不多一倍的缓冲空间。如果超过这个用量，大概率不是正常执行，而是什么环节出了问题。

3.2 看门狗的工作流程：从观察到干预

看门狗的完整工作流程我分成了五个阶段，每个阶段都有明确的动作和出口条件。

第一步是埋点采集。在编排器内部定义一个统一的执行器接口，每一次步骤执行开始、结束、失败都会发出事件。事件会带上步骤ID、执行器类型、耗时、Token用量这些元数据。这一步的关键是一个字："全"。宁可多采集一些暂时用不上的数据，也不要漏掉关键节点。

第二步是数据聚合。事件流会进入一个轻量级的滑动窗口计数器，我实现的是一个环形缓冲区，固定窗口大小为 60 秒，每秒一个时间桶，存储该秒内的调用次数、Token消耗、平均耗时。之所以用滑动窗口而不是简单计数器，是因为滑动窗口能真实反映"最近的节奏"，而简单计数器容易把过去的波动掺进当前判断里。

第三步是指标计算。每隔一段时间就会基于滑动窗口算一次指标快照，跟阈值做比较。这里我用了一个小技巧：对耗时指标做指数移动平均（EMA），公式是ema = alpha * current + (1 - alpha) * ema，alpha 取 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_keys=True, default=str).encode() ).hexdigest()[:16] output_digest = hashlib.sha256( json.dumps(self.outputs, sort_keys=True, default=str).encode() ).hexdigest()[:16] return f"{self.executor_type}:{input_digest}:{output_digest}" class SlidingWindowCounter: def __init__(self, window_seconds=60): self.window_seconds = window_seconds self.buckets = {} self.lock = threading.Lock() def add(self, value, bucket_key=None): 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_time=None): 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(maxlen=config["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_id=step_id, executor_type=executor_type, inputs=inputs, outputs=result, elapsed_ms=(time.time() - event_start) * 1000, token_usage=result.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_length=2太敏感了。解决办法是引入了"区分度"判断：两个步骤不仅指纹要一致，还要求后续的行为轨迹也要一致。我加了一个"展望窗口"的逻辑——只有当[A, B, A]这个三连模式出现，并且在展望窗口内确认 A 之后的下一步也是 B，才真正判定为循环。这样把单点重复变成序列重复，误杀率就明显下降了。

6.4 排查工具与调试技巧

生产环境的排查不能靠瞎猜。我在编排器里加了一个"跟踪快照"功能：每当任一防护组件被触发，就会自动生成一个结构化的事件记录，包含触发时间、步骤轨迹、阈值配置、当前状态等。这个快照会单独存一份 JSON 文件，方便事后用任何 JSON 工具分析。

排查的时候我的步骤固定为：先看触发类型，是超时、Token超限还是循环；再顺着快照里的执行轨迹回溯，看故障点是工具外部异常还是模型返回异常；最后根据故障点，决定调参（放宽阈值）、改实现（修正指纹逻辑）还是加白名单（允许某些操作重复执行）。

7. 一些补充心得和后续规划

写到这里，方法论部分基本讲完了。再说几句我在落地过程中的体会。

这套看门狗和熔断器，本质上是在和大模型的不确定性做对抗。你要接受一个现实：无论提示词写得多么精密，Agent 依然可能做出出乎意料的动作。这不是提示词工程能完全解决的，它需要系统级的安全机制来做收底。所以与其花大量时间去琢磨"怎么让提示词永远不出错"，不如把这些防护机制当成基础设施的一部分，一开始就规划进去。

我给开源版的后续规划里还有两个可以重点说的方向：第一个是"成本预测模块"，基于历史Token消耗数据的线性回归来预判当前任务最终的消耗上限，把保护机制从"事后熔断"推进到"事前预测"；第二个是"多Agent协同保护"，不是每个Agent各自带一套防护，而是在编排层做全局的熔断汇总，防止多个Agent之间互相触发导致系统级雪崩。

如果你打算在自己项目里落地这套机制，我的建议是从最小可用版本开始：先装一个看门狗盯住耐心阈值（单步超时+总时长），等跑了一段时间有数据了，再逐步加上Token预算和循环熔断。别一开始就追求大而全，否则你会在调参上耗费大量时间，反而错过了真正该观察的故障模式。

项目后续的代码和迭代记录会继续同步到开源仓库里，感兴趣的朋友可以关注下一期——到时候我会专门讲讲成本预测模块的算法选型，以及那套多Agent协同保护框架的架构设计。