前几天帮一个朋友排查他们某 AI 代理项目半夜跑崩的问题。一个自动整理行业资讯的小程序从下午四点多开始重复执行同一组搜索动作到第二天早上账单涨了近千块。日志拉出来一看同一个关键词搜了 47 次每次生成的摘要几乎一模一样前端页面还在傻等结果。这不是什么罕见事故。AI 代理本质上是一个自主决策进程每做一个动作都可能产生模型调用费用一旦失控它不是浪费算力是在直接烧钱。所以我现在做 agent 项目一定会加一层“实时监控 毫秒级叫停”的护栏。这篇文章不聊复杂理论就讲清楚这套护栏怎么设计、怎么实现、有哪些坑。核心就一句话让 AI 代理自由奔跑但给它装个油门刹车一体的安全阀。1. 先搞清楚钱是怎么“烧”掉的代理失控的几种典型场景要设计叫停机制得先弄明白失控长什么样。我见过的大多数“天价账单”都不是模型本身贵而是代理的运行路径出了问题。1.1 场景一低价值循环——同一个动作反复执行这是最常见、也最烧钱的一种。代理在某个决策点失去有效进展反复调用同一个工具。比如上面提到的资讯整理程序陷入循环的原因很简单提示词里没有“如果已经拿到该信息就立即停止”的约束工具响应的结果也没有做去重。于是模型每次都以为“这次应该能搜到新内容”结果每次都拿到同样结果再生成一遍几乎相同的摘要。这种循环的特征是工具调用序列里出现大量重复项。正常人的工作模式是A→B→C→D推进失控代理的模式是B→B→B→B。而且这类循环很难靠“设置 max_tokens”拦住因为每次单独调用都是合法的小请求问题出在调用次数上。1.2 场景二任务膨胀——单步请求越写越长聊天型 agent 有另一个隐蔽花钱点上下文滚动累加。每一轮对话都把历史记录原样重发给模型输入 token 随着轮数增长。一个 5 轮任务可能每轮只需几百 token跑到 20 轮的时候单次输入可能膨胀到几万 token费用翻几倍。更糟的是很多 agent 框架的“记忆管理”做得很粗糙把工具返回的完整 HTML、异常堆栈、中间日志一股脑塞进上下文。我见过一个案例代理只是做产品比价却把 3 万字的商品详情页原样带进下一轮 prompt一次调用的输入 token 直接顶别人十次。1.3 场景三级联失败的放大效应第三个典型失控路径是重试机制被滥用。某个工具调用超时 → 捕获异常 → 整体重试 → 又超时 → 再重试。如果重试次数不设上限或者退避策略变成“失败后升级到更大的模型再试一次”一个小故障就会变成账单放大器。更隐蔽的是“半成功”状态任务已经失败但 agent 还在继续生成解释性文本把错误信息反复总结、道歉、再尝试每一步都是成本。1.4 成本组成拆解一次性账单 vs 持续风险成本来源什么时候产生失控风险输入 token每次请求的 prompt 上下文上下文膨胀导致持续走高输出 token每次响应的内容话痨模式、重复生成工具调用附加费用调用外部服务 / API循环调用时成倍累积重试与降级失败后的再次请求指数级放大风险记忆/缓存写入长时任务中间态存储容易被忽略理解这张表之后再看实时监控不能只盯一个指标必须同时覆盖“量的膨胀”和“路径的异常”。2. 实时监控的四个核心信号没有指标就谈不上叫停监控不是写一句“盯紧点”就行要落到具体可计算的信号上。我在生产环境里长期用的信号主要有四个它们组合起来能覆盖绝大多数失控场景。2.1 信号一Token 消耗速率 / 预算消耗曲线这是最基础的信号监控的是“花钱的速度”。所有模型服务都会返回 usage 信息记录下来后可以维护两个指标累计成本整个任务从开始到现在的总花费单位是“元”或“美金”。滑动窗口速率过去 60 秒内花了多少。这个指标比累计值更灵敏能在流量刚抬头的时候就发现异常。为什么要用滑动窗口而不是固定窗口举个例子固定窗口从 0 秒到 60 秒统计如果任务正好从 59 秒开始爆发头一秒的消耗不会被计入等进入下一个窗口时已经晚了。滑动窗口每时每刻都在滚动统计能平滑掉这种边界割裂问题。2.2 信号二工具调用序列的“重复度”针对循环型失控我习惯给代理的每一次工具调用生成一个“指纹”工具名 参数摘要的哈希值。然后维护一个最近 N 步的调用队列计算当前动作在最近 N 步里出现了多少次。def is_repetition(recent_calls, threshold5): latest recent_calls[-1] count sum(1 for c in recent_calls[-threshold:] if c latest) return count threshold比如最近 10 步里有 5 次是同一个搜索动作那基本可以判定进入了低价值循环。这个计算是 O(N) 的毫秒级完成完全不会拖慢代理主流程。阈值不需要写死批处理场景可以严格一点交互式场景可以松一点。2.3 信号三单步请求体积与响应长度突变这个信号专门用来抓“任务膨胀”和“话痨模式”。正常任务的单次输入长度和输出长度通常在一个稳定区间内。一旦出现以下突变就要引起注意输入上下文突然比上一轮翻倍大概率是记忆管理出了问题把超大文本塞进了上下文。单次响应输出 token 远超该任务的合理长度模型可能进入了“车轱辘话”模式。某个工具返回的内容异常巨大后续所有轮次都会背着这个包袱费用会被反复放大。监控这个信号不需要机器学习设定一个“单次请求体积上限”和“单次响应长度上限”超出就告警就是有效方案。2.4 信号四执行路径偏离度最后这个信号适合流程相对固定的 agent。比如你预定义了一个五步执行流程每一步只能调用指定工具。执行路径偏离度就是实时比对“当前正在做的动作”和“预设动作清单”发现模型突然执行了清单之外的操作立刻标记风险。简单实现维护一个所有允许动作的集合以及一个可选的顺序期望。如果代理调用了不在集合内的方法或者跳过了必经步骤触发警告。这个判断规则简单直接误报率低适合合规要求高、流程标准化的任务。3. 毫秒级叫停的实现路径拦截、熔断与流式中断有了信号接下来解决最关键的落地问题怎么在毫秒级别叫停正在烧钱的代理。拖延半秒钟可能就多烧上千 token。3.1 在哪个环节叫停三层拦截架构我在工程上把叫停点分成三层各有各的用途接入层在 agent 与模型 API 之间加一层代理或网关统一记录所有请求和响应。这里能做全局限流比如总体预算耗尽后拒绝一切新请求。适合做账户级别的“总闸”。运行层在 agent 框架的执行循环里挂回调每次开始执行一步、每次工具调用返回时都触发检查。这一层能拿到完整的上下文适合做精准的路径判断。输出层在流式读取模型响应时逐块检查。这里能最快发现单次生成里的异常膨胀适合做“半路掐断”。实际项目里我推荐“运行层为主、接入层兜底、输出层补漏”。运行层负责判断“该不该停”接入层负责“绝对不超支”输出层负责“一旦开跑尽可能快速止血”。3.2 拦截器与回调钩子最小侵入方案大多数 agent 框架都支持回调或钩子机制。你不需要改框架源码只要注册几个事件处理器就行。伪代码大概是这样的agent.on(tool_call_start, check_tool_repetition) agent.on(step_end, check_budget) agent.on(response_chunk, check_stream_anomaly)关键注意点这些回调函数必须是轻量、同步、无阻塞的。不要在拦截器里发起新的网络请求、不要写大文件、不要调用另一个模型来判断“当前状况如何”。那会带来两倍延迟和两倍费用。正确的做法是所有要判断的指标都已经在内存里实时更新回调只是查一下计数器、比一下阈值立刻返回。3.3 预算熔断器阈值、窗口与状态机熔断器是叫停机制的核心组件。它不是简单的一次性判断而是一个带状态的状态机通常有三态关闭closed一切正常所有请求放行。打开open检测到预算超限或异常直接拒绝所有请求进入保护状态。半开half-open过一段时间后放开少量请求试探如果恢复正常则回到关闭否则继续保持打开。下面是一个简化的双阈值熔断器示例同时统计累计预算和窗口速率class BudgetGuard: def __init__(self, total_budget, rate_budget, window_seconds60): self.total_budget total_budget self.rate_budget rate_budget self.window_seconds window_seconds self.spent 0.0 self.window_start time.monotonic() self.window_spend 0.0 self.state closed def check(self, cost: float) - None: if self.state open: raise BudgetExceeded(circuit open, task stopped) now time.monotonic() if now - self.window_start self.window_seconds: self.window_start now self.window_spend 0.0 self.spent cost self.window_spend cost if self.spent self.total_budget or self.window_spend self.rate_budget: self.state open raise BudgetExceeded(budget or rate limit exceeded)为什么要双阈值累计预算解决的是“绝对不能超过总花费”的底线问题窗口速率解决的是“短时间异常爆发”的灵敏问题。两个结合既能防长期超支又能防瞬时爆炸。3.4 流式输出的“半路掐断”SSE 中断与资源回收大模型响应普遍走流式输出。很多人的监控只做在“整个响应返回之后”这是不够的。一个响应如果设定 max_tokens 为 8000模型可能真的生成 8000 token 才会停下来。如果中间第 3000 token 时已经能判定异常剩下 5000 token 的钱就白烧了。流式场景的正确做法是边收边查def guarded_stream(prompt, max_output2048): collected try: for chunk in model_api.stream(prompt): # 1. 单步输出检查超长立即掐断 if current_output_tokens chunk.token_count max_output: break # 2. 内容重复性检查连续 chunk 高度重复则掐断 if is_repetitive_chunk(chunk, recent_chunks): break # 3. 动态预算检查累计花费超预算则掐断 guard.check(estimate_chunk_cost(chunk)) collected chunk.text yield chunk finally: close_stream()这里的 finally 很重要。流式连接是一个长期占用的资源掐断后必须显式关闭连接、取消任务状态否则会出现连接泄漏导致后续任务排队阻塞反而衍生出新的问题。4. 一套可落地的实时监控原型从埋点到叫停光有概念不够下面给出一套我自己实际跑过的原型方案。它适合单机、中小并发的 agent 应用核心思路能平滑迁移到生产环境。4.1 原型架构与数据结构整个原型分四个模块模块职责存储状态存储保存每个任务的累计花费、调用序列、最近响应块内存 dict deque指标收集从模型响应中解析 token 用量、工具调用信息在回调里实时更新规则引擎执行四个信号的判断返回 stop / continue / warn纯函数式判断执行器根据规则结果决定中断、降级或继续调用框架的 stop 接口单机原型不需要引入消息队列和时序数据库一个defaultdict加一个deque就能跑起来。只要规则判断逻辑是纯内存计算毫秒级叫停就没有瓶颈。数据结构建议用任务 ID 作为维度task_state { task_id: task_001, total_cost: 0.0, window_start: 0.0, window_cost: 0.0, calls: deque(maxlen20), recent_chunks: deque(maxlen10), context_tokens: 0, output_tokens: 0, }4.2 核心实现规则引擎与流式中断的组合在上一节的 BudgetGuard 基础上我做了第二个关键实现——把四个信号统一收敛到一个should_stop()方法里def should_stop(state, event) - tuple[bool, str]: # 信号一预算与速率 if event.type cost: guard.check(event.cost) # 信号二调用序列重复 if event.type tool_call: recent list(state[calls]) if is_repetition(recent [event.fingerprint], threshold5): return True, tool_call_loop # 信号三单步响应膨胀 if event.type chunk: if state[output_tokens] MAX_OUTPUT_TOKENS: return True, output_too_long # 信号四路径偏离 if event.type action: if event.action not in allowed_actions: return True, path_deviation return False, 把这个方法挂到回调里之后代理每走一步都会经过检查。调用这个方法是毫秒级开销相对于一次动辄几百毫秒甚至几秒的模型调用完全可以忽略。4.3 多任务并发下的状态共享与锁一旦同时跑多个 agent 任务就不能用全局共享 dict 裸写。任务之间会互相污染还会出现竞态两个并发任务同时更新同一个计数器导致预算统计失真。原型里我建议按任务 ID 分片加锁每个任务一个独立的Lock对象更新状态前先获取锁。跨任务的全局预算用threading.Lock保护每次更新时做一次原子加操作。如果将来上生产用集中式存储比如 Redis存预算状态配INCRBY和EXPIRE做原子操作天然支持多进程。这里容易踩的坑是把锁粒度放得太大导致并发任务互相阻塞。合理做法是锁只保护“状态更新”那几行代码规则判断本身可以无锁执行因为判断只读当前快照。5. 实战中容易踩的坑误杀、漏放与监控本身的开销这章是血泪经验总结。一个监控系统做得不好比不做的危害更大——它会误杀正常任务或者给你虚假的安全感。5.1 误杀把正常任务当成失控最典型的是长报告生成任务。代理确实要连续输出大量内容比如生成一份万字行业分析输出 token 天生就大。如果阈值设得太紧系统会在任务进行到一半时误判为“异常膨胀”并强制叫停整个任务前功尽弃。解决办法是分级动作而不是一把梭第一级黄色预警。命中速率阈值后只记录日志、发送提醒不中断任务。第二级橙色限制。持续超过阈值后开始降低模型规格或压缩上下文让任务继续但消耗放缓。第三级红色熔断。超过更高级别的阈值比如预算耗尽、循环判定成立才强制叫停。这样既避免了误杀又保证了失控时能收住。你不能指望一个规则在所有场景都完美准确你只能让规则的分级足够合理。5.2 漏放监控滞后与采样盲区有些监控只在“完整响应返回后”执行。这等于在钱已经烧完后再看账单完全失去了叫停的意义。流式输出必须边收边查工具调用循环必须在每次调用返回时立刻检查而不是攒够十次再算。还有一类漏放来自“采样盲区”如果你只读取流的前 N 个 chunk 做异常检测后面的内容可能完全不可控。所以检查逻辑要贯穿整个流生命周期不能只查开头。5.3 监控本身的开销别把监控做成新的瓶颈监控代码写在回调里它不能反过来拖慢主流程。我在早期版本里犯过错误在回调里用正则深度解析完整的工具返回内容结果一个回调花了 80 毫秒代理每走一步都慢得离谱。经验做法用哈希指纹代替全文比对重复检测只比较哈希。用累计计数代替复杂统计除非必要不做概率计算。不许在回调里发起同步网络请求所有数据都放内存。日志异步写入或者只在触发告警时写详细日志。实测下来一个纯内存的规则检查基本在 1 毫秒以内对比模型调用本身动辄几百毫秒以上这个开销不值得担心。5.4 叫停之后怎么办审计、重启与善后叫停不是终点是救援的开始。每次触发熔断后至少要留下完整的审计信息触发时间、触发规则、当时的累计花费。最近 N 步的工具调用序列。被中断时的上下文快照和中间结果。当前断点位置以便后续恢复。对于有状态任务建议把中间结果持久化到磁盘或对象存储这样下一次可以从断点重启而不是从零开始。我的典型做法是触发熔断后先把任务标记为“interrupted”然后走一个自动恢复流程——如果任务可以降级就用小模型继续如果不行就通知人工接管。6. 不同场景下的叫停策略配置参考不同任务的失控形态不一样监控规则也要跟着调。下面是我在几个典型场景里用的配置参考你可以按需调整。6.1 场景对照表场景主要失控形态建议监控信号参考阈值叫停动作批处理数据处理重试放大、循环调用重试次数、单任务成本重试≥3次成本超预算80%熔断并人工复核客服问答代理长话痨、上下文膨胀单次输出 token、输入长度单次输出超5000上下文超限截断重写、降级小模型代码生成代理相同错误反复修复相同报错出现次数同一报错≥3次暂停、换修复策略长时研究型任务累计烧钱、路径漂移每分钟成本、调用重复度每10步≥5次重复动作断点暂停 通知人工这个表只是起点。更精确的阈值要基于你自己任务的数据来定。建议先跑几天不做自动叫停只把监控数据打点记录下来看到正常任务的实际消耗分布后再反过来设置阈值。6.2 如何根据预算反推叫停阈值假设你接到一个需求某个批处理任务单次预算上限是 5 元。模型服务报价大约是输入 token 1 元/百万输出 token 3 元/百万。根据运行历史平均每一步输入约 4000 token输出约 1500 token。那每一步的成本估算为输入成本4000 × 1 / 1000000 0.004 元输出成本1500 × 3 / 1000000 0.0045 元单步合计约 0.0085 元5 元预算大约能支撑 588 步。这时候不要把 588 直接设为硬阈值因为正常任务可能有上下波动。更稳的做法是累计预算硬上限设为 4 元留 1 元作为熔断后处理余量。窗口速率阈值设为 60 秒内不超过约 0.5 元折算下来大约每分钟不超过 58 步正常任务很难达到这个速率失控任务却能轻松撞线。这样配置的逻辑是累计上限兜底速率上限抓异常。两条线同时起作用比单纯设一个步数限制可靠得多。7. 最后分享一点我的实际体会做实时监控这套东西真正让我受益最深的一点是把“叫停”从简单的 break 升级成带断点恢复的 cancel 语义。第一次做的时候我只会在预算超限后粗暴终止结果任务丢失上下文重跑一次又花了大半预算。后来改成记录断点、序列化中间状态恢复效率高了很多。如果你也在做类似系统建议不要一上来就追求复杂的模型判断先用一个计数器加一个预算阈值把“发现异常→中断→留痕→复盘”这条链路跑通再逐步加入路径偏离、重复检测这些高级规则。监控的目标不是把 agent 管死而是让它敢跑、跑不丢。