1. 为什么需要一个提示流编排器

1.1 从一次惨痛的账单说起

去年冬天的一个凌晨，我被手机短信震醒。打开一看，云服务账单提醒：过去六小时消耗了平时一个月的额度。爬起来查日志，发现是一个 ReAct 模式的 Agent 在某个边界条件下陷入了死循环——它反复调用同一个搜索工具，每次拿到相似但不完全一样的结果，然后基于这个结果再发起新一轮调用。六小时，三千多次工具调用，账单直接起飞。

这件事之后我花了整整两周时间，给手里的 Agent 框架加了一套运行时看门狗和死循环熔断机制。今天这篇就把这套东西的设计思路和实现细节完整拆开讲。

如果你正在做 AI Agent 开发，尤其是基于 ReAct 模式构建能思考与行动的智能体，那你一定知道：Agent 的自由度越高，失控的风险就越大。提示流编排器要解决的核心问题，就是在保留 Agent 自主性的同时，给它套上一个可靠的缰绳。

1.2 这个编排器到底管什么

简单说，提示流编排器是 Agent 运行时的调度中枢。它负责几件事：管理提示模板的组装和注入、控制工具调用的权限和频率、监控每一轮推理的状态、在检测到异常模式时执行熔断。

和市面上那些偏重"画流程图"的编排工具不同，我这套东西更关注运行时安全。你可以把它理解成 Agent 的"行车记录仪加刹车系统"——平时默默记录，关键时刻一脚刹住。

适合谁来参考？如果你已经写过基础的 Agent 循环，知道什么是 ReAct、什么是工具调用，但还没认真处理过运行时安全问题，那这篇就是写给你的。如果你刚开始学 Agent 开发，建议先把基础循环跑通再来看，不然有些设计决策你体会不到痛点。

1.3 整体架构长什么样

整个编排器分成四层，从下往上依次是：

• 执行层：负责单次 LLM 调用和工具执行，是最底层的原子操作
• 监控层：看门狗所在的位置，每一轮循环结束后采集状态指标
• 决策层：熔断器根据监控数据判断是否继续、降级还是终止
• 编排层：管理整个提示流的生命周期，包括初始化、循环、收尾

这四层之间通过一个共享的运行时上下文传递数据。上下文里记录了轮次计数、工具调用历史、token 消耗、时间戳等关键信息。每一层只读自己需要的字段，写自己负责的字段，避免耦合。

注意：不要一上来就把四层全部实现。我建议先做监控层和决策层，执行层和编排层用最简单的版本跑起来，等监控数据积累够了再优化。

2. 看门狗的核心设计

2.1 看门狗到底在看什么

看门狗这个词借自嵌入式系统里的硬件看门狗电路——那个东西的作用是定时喂狗，如果程序跑飞了没按时喂，硬件就强制复位。Agent 运行时看门狗的逻辑类似，但监控的指标更复杂。

我目前监控的核心指标有六个：

这六个指标不是拍脑袋定的。轮次计数来自经验——大部分正常任务在 15 轮内能完成，超过 25 轮基本可以判定有问题。工具调用频次和 token 消耗速率是成本控制的核心。重复调用率和输出相似度是检测死循环的关键。

2.2 指标采集的实现方式

采集逻辑放在监控层，每一轮循环结束后执行一次。核心思路是维护一个滑动窗口，窗口大小根据指标类型不同而不同。

from collections import deque import time import hashlib class WatchdogMonitor: def __init__(self, window_size=20): self.window_size = window_size self.tool_calls = deque(maxlen=window_size) self.round_outputs = deque(maxlen=5) self.token_usage = deque(maxlen=window_size) self.round_timestamps = deque(maxlen=window_size) self.round_count = 0 def record_round(self, output_text, tool_name=None, tool_args=None, tokens=0): self.round_count += 1 now = time.time() self.round_timestamps.append(now) self.token_usage.append((now, tokens)) self.round_outputs.append(output_text) if tool_name: call_signature = self._make_signature(tool_name, tool_args) self.tool_calls.append((now, call_signature)) def _make_signature(self, tool_name, tool_args): raw = f"{tool_name}:{tool_args}" return hashlib.md5(raw.encode()).hexdigest()[:12]

这里有个细节值得说：工具调用的签名我用的是 MD5 前 12 位。为什么不用完整哈希？因为窗口里最多存 20 条记录，12 位十六进制碰撞概率极低，而且省内存。参数序列化的时候要注意字典的键顺序，最好先排序再序列化，否则同样的参数因为键顺序不同会算出不同的签名。

2.3 重复调用率的计算逻辑

重复调用率是判断死循环最直接的指标。计算方式是在滑动窗口内，统计出现次数超过一次的签名占比。

def calc_repeat_rate(self): if len(self.tool_calls) < 4: return 0.0 signatures = [sig for _, sig in self.tool_calls] from collections import Counter counter = Counter(signatures) repeated = sum(count for count in counter.values() if count > 1) return repeated / len(signatures)

阈值定在 40% 是调出来的。太低会误伤——有些任务确实需要多次调用同一个工具查不同参数。太高又漏检——我实测过一个死循环案例，重复率稳定在 55% 左右。40% 这个位置能在第三轮重复出现时就报警，给熔断留出反应时间。

2.4 输出相似度怎么算才靠谱

输出相似度用语义层面的比较，不能简单用字符串匹配。我的做法是用一个轻量级的句子嵌入模型，把连续两轮输出转成向量，算余弦相似度。

如果不想引入额外模型，也有个偷懒的办法：用 Jaccard 相似度比较两轮输出的词集合。精度差一些，但对"原地打转"这种明显情况够用。

def jaccard_similarity(text_a, text_b): set_a = set(text_a.lower().split()) set_b = set(text_b.lower().split()) if not set_a or not set_b: return 0.0 intersection = set_a & set_b union = set_a | set_b return len(intersection) / len(union)

Jaccard 的阈值我设在 0.85。超过这个值，基本可以认为 Agent 在重复自己。注意要去掉停用词再算，否则"的"、"了"这些高频词会拉高相似度。

3. 死循环熔断器的实现

3.1 熔断策略的分级设计

熔断器不是简单的"触发就杀"，而是分三级响应：

• 一级（警告）：记录日志，在下一轮提示中注入提醒，比如"你似乎在进行重复操作，请检查是否陷入循环"
• 二级（降级）：限制工具调用权限，只允许调用还没用过的工具，或者强制切换到更简单的推理模式
• 三级（终止）：直接中断执行，返回当前已有结果并附带终止原因

分级的好处是给 Agent 自我纠正的机会。我实测下来，大约 60% 的潜在死循环在一级警告后就能自己跳出来。剩下 40% 里，又有一半在二级降级后能完成。真正需要三级终止的，不到 20%。

3.2 状态机实现

熔断器本质上是个状态机，三个状态：正常、警告、熔断。

class CircuitBreaker: STATE_NORMAL = "normal" STATE_WARNING = "warning" STATE_TRIPPED = "tripped" def __init__(self, monitor): self.monitor = monitor self.state = self.STATE_NORMAL self.warning_count = 0 self.max_warnings = 3 def evaluate(self): repeat_rate = self.monitor.calc_repeat_rate() similarity = self.monitor.calc_output_similarity() round_count = self.monitor.round_count risk_score = 0 if repeat_rate > 0.4: risk_score += 2 if similarity > 0.85: risk_score += 2 if round_count > 25: risk_score += 3 if self.monitor.calc_token_rate() > self.monitor.token_budget: risk_score += 3 if risk_score >= 6: self.state = self.STATE_TRIPPED elif risk_score >= 3: self.state = self.STATE_WARNING self.warning_count += 1 if self.warning_count >= self.max_warnings: self.state = self.STATE_TRIPPED else: self.state = self.STATE_NORMAL self.warning_count = max(0, self.warning_count - 1) return self.state

风险评分的权重是调出来的。轮次和 token 超预算权重最高，因为这两个直接对应成本。重复率和相似度权重稍低，因为它们有误报可能。

3.3 熔断后的收尾处理

三级熔断触发后，不能直接抛异常了事。要做三件事：

第一，保存现场。把完整的对话历史、工具调用记录、监控指标快照写入日志。这些数据是后续调优的宝贵素材。

第二，生成收尾提示。让 LLM 基于已有信息给出一个"尽力而为"的答案，而不是空手而归。

第三，释放资源。关闭所有打开的工具连接，清理临时文件，重置运行时上下文。

def handle_trip(self, context): snapshot = { "round_count": self.monitor.round_count, "repeat_rate": self.monitor.calc_repeat_rate(), "tool_calls": list(self.monitor.tool_calls), "timestamp": time.time() } self._persist_snapshot(snapshot) final_prompt = ( "由于检测到执行异常，任务被提前终止。" "请基于目前已有的信息，给出你能提供的最佳答案。" "如果信息不足，请明确说明缺少什么。" ) return self._generate_final_response(context, final_prompt)

4. 提示流编排的完整流程

4.1 初始化阶段

编排器启动时要做几件事：加载配置、初始化监控器和熔断器、构建系统提示、准备工具注册表。

配置里最关键的是预算参数。我通常设三个预算：token 总预算、工具调用总次数预算、最大轮次预算。这三个预算是硬约束，熔断器会参考它们做决策。

系统提示的构建有个技巧：把安全约束写进去。比如"如果你发现自己重复调用同一个工具超过两次，请停下来重新思考策略"。这种提示不能完全防止死循环，但能降低发生概率。

4.2 主循环的执行逻辑

主循环是编排器的心脏。每一轮的执行顺序是：

1. 组装当前轮的提示（系统提示 + 历史 + 当前状态）
2. 调用 LLM 获取输出
3. 解析输出，判断是工具调用还是最终答案
4. 如果是工具调用，执行工具并记录结果
5. 调用看门狗采集指标
6. 调用熔断器评估状态
7. 根据熔断状态决定下一步

def run(self, task): context = self._init_context(task) while True: prompt = self._build_prompt(context) output = self._call_llm(prompt) action = self._parse_output(output) if action.type == "final_answer": return action.content if action.type == "tool_call": result = self._execute_tool(action.tool_name, action.args) context.add_tool_result(result) self.monitor.record_round( output_text=output, tool_name=action.tool_name if action.type == "tool_call" else None, tool_args=action.args if action.type == "tool_call" else None, tokens=self._count_tokens(prompt, output) ) state = self.breaker.evaluate() if state == CircuitBreaker.STATE_TRIPPED: return self.breaker.handle_trip(context) elif state == CircuitBreaker.STATE_WARNING: context.inject_warning(self._build_warning_message())

4.3 工具调用的权限控制

工具注册表里每个工具除了函数本身，还带元数据：调用配额、超时时间、是否需要审批。

配额控制是防止单个工具被刷爆的关键。比如搜索工具设 10 次配额，文件写入设 5 次。配额用完后再调用直接返回错误，不消耗 LLM 调用。

超时时间也很重要。有些工具（比如网络请求）可能卡住，必须设超时。我一般设 30 秒，超过就中断并返回超时错误。

4.4 上下文窗口的管理

Agent 跑多轮之后，对话历史会越来越长，最终超出模型的上下文窗口。处理方式有两种：滑动窗口和摘要压缩。

滑动窗口简单粗暴：只保留最近 N 轮。缺点是可能丢掉早期的重要信息。

摘要压缩更优雅：当历史超过阈值时，调用 LLM 把早期对话压缩成一段摘要。缺点是额外消耗 token，而且摘要本身可能丢信息。

我的做法是混合：保留最近 5 轮完整对话，更早的压缩成摘要。摘要里重点保留工具调用结果和关键决策点。

5. 常见问题与排查技巧

5.1 误报太多怎么办

看门狗误报是最常见的问题。典型表现是正常任务被频繁警告甚至熔断。

排查思路：先看是哪个指标触发的。如果是重复调用率，检查是不是任务本身就需要反复调用同一个工具。如果是输出相似度，检查是不是 Agent 在正常地做总结性陈述。

调优方法：把阈值调高，或者给特定任务类型设置不同的阈值。我后来加了一个"任务类型"字段，不同类型的任务用不同的阈值配置。比如数据分析类任务允许更多轮次，问答类任务轮次限制更严。

5.2 熔断后结果质量差

三级熔断触发后，Agent 往往给不出好答案，因为它是被强行中断的。

改善方法：在熔断前多给几次警告机会，让 Agent 有机会调整。另外收尾提示要写得好，引导 LLM 基于已有信息做最大努力。

我试过在收尾提示里加上"请列出你已经完成的部分和未完成的部分"，这样即使答案不完整，用户也能知道进展到哪了。

5.3 监控本身的开销

看门狗和熔断器的计算本身也要消耗资源。如果每轮都做复杂的相似度计算，会拖慢整体速度。

优化方法：相似度计算可以异步做，不阻塞主循环。指标采集用滑动窗口，避免全量扫描。熔断评估用简单的阈值判断，不要引入复杂模型。

实测下来，这套监控逻辑增加的开销在 5% 以内，完全可以接受。

5.4 常见问题速查表

6. 几个实操中踩过的坑

第一个坑：工具调用的参数序列化。最开始我用str(args)做签名，结果字典键顺序不同导致同样的参数算出不同签名，重复调用率一直偏低，漏检了好几次死循环。后来改成先json.dumps(args, sort_keys=True)再哈希，问题解决。

第二个坑：相似度计算的时机。一开始我在 LLM 输出后立刻算相似度，但那时候输出还没解析，包含了很多格式标记。后来改成解析后再算，准确率明显提升。

第三个坑：熔断状态的恢复。有次熔断触发后，我没重置状态，导致下一个任务一开始就是熔断态。后来加了任务级别的状态隔离，每个新任务都重新初始化熔断器。

第四个坑：token 计数的准确性。不同模型的 tokenizer 不一样，我一开始用估算，误差很大。后来针对每个模型用对应的 tokenizer，预算控制才准。

这套东西我从去年冬天开始做，到现在迭代了七八个版本。最开始只有简单的轮次限制，后来逐步加上重复检测、相似度分析、分级熔断。每次线上出问题，就回来加一个监控指标或者调一个阈值。现在这套机制已经拦下了十几次潜在的成本事故，最惊险的一次是在两轮之内就识别出了死循环苗头，直接一级警告让 Agent 自己跳出来了。

如果你也在做 Agent 开发，强烈建议在早期就把运行时安全机制加上。等账单来了再补，代价太大。