AI Agent 运行时安全:提示流编排器与死循环熔断机制设计 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 运行时看门狗的逻辑类似但监控的指标更复杂。我目前监控的核心指标有六个指标含义典型阈值轮次计数当前对话循环执行了多少轮单次任务不超过 25 轮工具调用频次单位时间内工具被调用的次数每分钟不超过 15 次重复调用率相同工具相同参数的调用占比超过 40% 触发警告Token 消耗速率每分钟消耗的 token 数根据预算动态调整单轮耗时一次完整推理工具执行的耗时超过 90 秒标记异常输出相似度连续两轮输出的语义相似度超过 0.92 判定为原地打转这六个指标不是拍脑袋定的。轮次计数来自经验——大部分正常任务在 15 轮内能完成超过 25 轮基本可以判定有问题。工具调用频次和 token 消耗速率是成本控制的核心。重复调用率和输出相似度是检测死循环的关键。2.2 指标采集的实现方式采集逻辑放在监控层每一轮循环结束后执行一次。核心思路是维护一个滑动窗口窗口大小根据指标类型不同而不同。from collections import deque import time import hashlib class WatchdogMonitor: def __init__(self, window_size20): self.window_size window_size self.tool_calls deque(maxlenwindow_size) self.round_outputs deque(maxlen5) self.token_usage deque(maxlenwindow_size) self.round_timestamps deque(maxlenwindow_size) self.round_count 0 def record_round(self, output_text, tool_nameNone, tool_argsNone, tokens0): 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 主循环的执行逻辑主循环是编排器的心脏。每一轮的执行顺序是组装当前轮的提示系统提示 历史 当前状态调用 LLM 获取输出解析输出判断是工具调用还是最终答案如果是工具调用执行工具并记录结果调用看门狗采集指标调用熔断器评估状态根据熔断状态决定下一步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_textoutput, tool_nameaction.tool_name if action.type tool_call else None, tool_argsaction.args if action.type tool_call else None, tokensself._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_keysTrue)再哈希问题解决。第二个坑相似度计算的时机。一开始我在 LLM 输出后立刻算相似度但那时候输出还没解析包含了很多格式标记。后来改成解析后再算准确率明显提升。第三个坑熔断状态的恢复。有次熔断触发后我没重置状态导致下一个任务一开始就是熔断态。后来加了任务级别的状态隔离每个新任务都重新初始化熔断器。第四个坑token 计数的准确性。不同模型的 tokenizer 不一样我一开始用估算误差很大。后来针对每个模型用对应的 tokenizer预算控制才准。这套东西我从去年冬天开始做到现在迭代了七八个版本。最开始只有简单的轮次限制后来逐步加上重复检测、相似度分析、分级熔断。每次线上出问题就回来加一个监控指标或者调一个阈值。现在这套机制已经拦下了十几次潜在的成本事故最惊险的一次是在两轮之内就识别出了死循环苗头直接一级警告让 Agent 自己跳出来了。如果你也在做 Agent 开发强烈建议在早期就把运行时安全机制加上。等账单来了再补代价太大。