维护过一个跑了大半年、接了十几个外部工具、背后还有两三个模型供应商的 Agent 系统之后你最怕的往往不是模型效果差而是改不动。想统计每个模型调用的耗时和 Token 消耗得去 Agent 主循环里翻代码想对工具调用做限流又得在每个工具函数里塞重复的守卫逻辑某天突然来了一个审计需求要改十几个文件。后来我把这些横切逻辑统一收敛到 AgentMiddleware 的 wrap_model_call 和 wrap_tool_call 两个事件钩子上用一套中间件机制解决了日志、追踪、限流、重试、护栏五类问题。下面我会把这套机制的执行模型、生产级实现、高频踩坑和测试编排完整讲一遍适合正在自研 Agent 运行时、或者想在现有框架里做统一控制的开发者参考。1. 先搞清楚痛点Agent 运行时里为什么偏偏是这两个事件1.1 Agent 执行链路上真正可统一拦截的两个节点Agent 运行时的主循环归根结底只有两件事让模型想调用模型服务生成下一步行动让工具做执行某个具体函数并把结果喂回模型。中间穿插的状态管理、记忆读写、用户交互本质上都是围绕想和做两个节点展开的。因此如果要在 Agent 链路上做横切控制比如日志、追踪、鉴权、限流、重试、内容过滤、费用控制只需要抓住模型调用和工具调用这两个交叉点就能覆盖 90% 以上的需求。有人会问为什么不拦截每一轮对话或每次 Agent 执行这种更粗粒度的事件我的经验是粗粒度事件确实简单但代价是信息粒度太粗。你无法区分这一轮里到底调了几次模型、哪一次工具调用是重试命中还是首次执行也就没法做精确的配额控制和故障定位。反过来只拦截工具调用、不拦截模型调用也不行——模型调用往往是最贵的一环费用治理和提示词审计全在这一环漏掉它等于拦了个寂寞。1.2 从散落埋点到统一拦截面一次真实的重构我们早期实现过一台能跑的 Agent23 个工具函数每个函数开头都贴一段耗时统计中间穿插各自的重试逻辑鉴权代码甚至写在调用方。后来要做按租户的 Token 配额我数了一下涉及 17 个文件、34 处改动点还漏改了两处。那次之后我下决心做统一拦截面抽出一个 AgentMiddleware 基类暴露 wrap_model_call 和 wrap_tool_call 两个方法所有横切逻辑以中间件形式注册到运行时。重构之后新增一个配额策略只需要新写一个中间件类、注册一行完全不会碰到任何工具函数和主循环代码。那次重构让我意识到一个关键点中间件的价值不是少写两行代码而是让横切关注点和业务逻辑彻底解耦让后来接手的人一眼就能看出这个系统在哪些环节被干预过、干预的方式是什么。这个心智模型比具体的代码实现更重要。2. wrap_model_call 与 wrap_tool_call 的执行模型回调语义、洋葱结构与短路2.1 钩子签名与参数含义先看我们组件里最基本的定义具体框架写法可能略有差异但语义是通用的from abc import ABC from typing import Any, Awaitable, Callable # next_call 代表继续向下执行可能是下一个中间件也可能是真实调用 ModelCallNext Callable[[ModelRequest], Awaitable[ModelResponse]] ToolCallNext Callable[[str, dict[str, Any]], Awaitable[ToolResult]] class AgentMiddleware(ABC): async def wrap_model_call( self, ctx: RunContext, next_call: ModelCallNext, request: ModelRequest, ) - ModelResponse: # 模型调用前的钩子逻辑 response await next_call(request) # 模型调用后的钩子逻辑 return response async def wrap_tool_call( self, ctx: RunContext, next_call: ToolCallNext, tool_name: str, args: dict[str, Any], ) - ToolResult: # 工具调用前的钩子逻辑 result await next_call(tool_name, args) # 工具调用后的钩子逻辑 return result三个参数各司其职。ctx 是本次运行上下文贯穿一次 Agent 执行的完整生命周期里面的 run_id、trace_id、user_id、租户信息、Token 累积用量都是中间件之间共享的关键状态。next_call 是往下走的入口所有不打算在此处终结的处理都要调用它。request或 tool_name args是当前事件的具体内容中间件可以读取甚至修改其中的字段后继续向下传——这才是拦截的意义不只是观察还能干预。注意 wrap_ 这个前缀的含义。它不是在事件发生点插入一个监听器而是把整个调用包裹起来你有机会在事件发生前做动作调用 next_call 让事件真正发生再在返回路径上做动作。理解这个包裹语义后面所有的顺序问题就都好解释了。2.2 洋葱式调用链进入顺序与退出顺序多个中间件同时注册时调用链是一个洋葱结构。假设按注册顺序 A、B 依次注册实际执行是这样的agent.execute() └→ A.wrap_model_call(request) └→ B.wrap_model_call(request) └→ 真实模型服务 ← B 处理响应之后继续执行 ← A 处理响应之后继续执行进入时 A 先执行、B 后执行拿到响应往回退时恰好反过来B 先处理响应A 后处理响应。如果你在两个方法里把调用前逻辑和调用后逻辑写在 next_call 两侧那么调用前逻辑按注册顺序执行调用后逻辑按注册逆序执行。这个特性在实际排错时非常重要第 4 节我会专门展开说。2.3 短路不调用 next_call直接决定结果中间件最有威力的能力是短路——不调用 next_call自己直接返回值或者抛异常从而终止底层调用。典型的适用场景有三个。第一是请求缓存命中把模型名加消息哈希作为 key命中缓存时直接在 wrap_model_call 里返回缓存响应连模型服务都不需要请求成本和延迟同时降下来。第二是静态内容拦截输入里命中敏感词或违规指令在 wrap_model_call 阶段直接返回一段预设的拒绝文案避免把风险内容发到模型服务。第三是配额不足用户 Token 余额耗尽时直接抛配额异常让上层以 429 语义反馈而不是真实发起调用再被计费。短路的代价是所有内层中间件都不会执行。所以审计、日志这类即使失败也要有记录的中间件必须放在最外层否则一个短路就会导致该次请求完全没有日志。这一点我先埋个伏笔后面踩坑部分还会再强调。3. 生产级实现日志、全链路追踪、限流熔断与内容护栏3.1 日志与用量统计中间件第一个要写的中间件永远是日志。别急着写花哨功能先把谁在什么时候调了什么、花了多少钱、耗时多久记清楚import logging import time logger logging.getLogger(agent.middleware) class ObservabilityMiddleware(AgentMiddleware): async def wrap_model_call(self, ctx, next_call, request): start time.perf_counter() try: response await next_call(request) return response finally: cost_ms (time.perf_counter() - start) * 1000 ctx.usage.add_model_usage( modelrequest.model, input_tokensestimate_tokens(request.messages), output_tokensresponse.usage.output_tokens if response.usage else 0, cost_mscost_ms, ) logger.info( model_call_finished, extra{ run_id: ctx.run_id, trace_id: ctx.trace_id, model: request.model, cost_ms: round(cost_ms, 2), }, )注意我故意用 try/finally 而不是 try/except。finally 保证即使底层模型抛异常日志照打而异常本身要原样往外抛不要让中间件吞掉。这行代码是后面所有排查的基础——没有这条日志你连问题发生在哪一层都不知道。工具调用同样要记但要额外记工具名、参数摘要、返回状态async def wrap_tool_call(self, ctx, next_call, tool_name, args): start time.perf_counter() try: result await next_call(tool_name, args) logger.info( tool_call_finished, extra{ tool: tool_name, args_snapshot: summarize(args), status: ok, cost_ms: round((time.perf_counter() - start) * 1000, 2), }, ) return result except Exception as exc: logger.warning( tool_call_failed, extra{ tool: tool_name, error: repr(exc), cost_ms: round((time.perf_counter() - start) * 1000, 2), }, ) raise参数摘要 summarize 很重要完整参数可能包含大段文本甚至二进制内容直接全量打日志会把日志系统压垮。我一般截断到 200 字符同时对疑似敏感字段做掩码处理既保可用性又保安全。3.2 Trace ID 注入与异步上下文传递Agent 执行链路往往跨多个服务入口网关、Agent 运行时、模型服务、内部工具服务。没有统一的 trace_id出问题时要靠时间戳拼链路非常痛苦。中间件是注入 trace_id 最合适的位置一次 run 开始时就生成放进 ctx然后通过两条路径传播——给模型服务的请求头带上给工具服务的调用参数带上同时塞进所有日志的 extra 字段。这里有一个容易翻车的点异步上下文。我们用的是 contextvars 保存 trace_id工具函数内部的协程直接读 contextvars 通常没问题但一旦工具内部用了线程池 executor或者用 asyncio.create_task 派生出独立任务trace_id 就可能丢失。线程池里的线程是全新的上下文create_task 默认复制的是创建那一刻的上下文不会跟随运行时后续写入的值。这种问题修起来很隐蔽日志里一半请求有 trace_id一半没有查代码又看不出明显错误。我的建议是双保险contextvars 只作为读的捷径真正的权威信息放 ctx 对象里在需要异步派生的地方显式把 ctx或至少 trace_id作为参数传下去。不要信任隐式传播显式传递永远最稳。3.3 限流熔断给模型调用上闸给工具调用上保险模型调用是计费资源必须有限流。按租户分桶的令牌桶是最常见的方案class RateLimitMiddleware(AgentMiddleware): def __init__(self, bucket_factory): # 例如 redis 上的分布式令牌桶 self._buckets bucket_factory async def wrap_model_call(self, ctx, next_call, request): bucket self._buckets.get_bucket(keyfmodel:{ctx.tenant_id}) if not await bucket.try_acquire(1): raise RateLimitExceeded( codeMODEL_RATE_LIMIT, messageftenant {ctx.tenant_id} model call rate exceeded, ) return await next_call(request)注意这里的异常语义限流不是返回一个错误文本让模型自己消化而是让上层运行时直接以 429 类状态结束本轮并把这个状态暴露给调用方。如果中间件偷偷把限流异常转成再试一次的普通响应Agent 会拿着这个假响应继续编排可能造成无限重试。工具调用侧我通常做三件套超时、重试、熔断。工具是外部依赖外部服务抖动是常态。重试中间件只对瞬时错误重试永久错误参数校验失败、权限不足直接放行class ToolResilienceMiddleware(AgentMiddleware): def __init__(self, max_retries2, base_delay0.5): self._max_retries max_retries self._base_delay base_delay async def wrap_tool_call(self, ctx, next_call, tool_name, args): last_exc None for attempt in range(self._max_retries 1): try: result await next_call(tool_name, args) if last_exc: logger.info( tool_call_recovered, extra{tool: tool_name, attempt: attempt}, ) return result except TransientToolError as exc: last_exc exc if attempt self._max_retries: await asyncio.sleep(self._base_delay * (2**attempt)) raise last_exc指数退避的 base_delay 我一般给 0.5 秒最多重试两次避免雪崩。熔断放在重试外层统计最近窗口内工具失败率超过阈值直接短路该工具让 Agent 尽快切换别的工具而不是死磕一个坏掉的服务。3.4 内容护栏与敏感信息过滤安全需求往往以突然出现的方式到来某天合规要求所有发往模型服务的提示词必须先过滤身份证号、手机号等敏感信息返回给用户的内容也要做脱敏。这种需求散落在各个工具和编排逻辑里几乎不可维护放在 wrap_model_call 里刚好输入侧在 next_call 前扫描并脱敏输出侧在拿到响应后、返回给编排器之前处理。输入护栏和输出护栏的策略不一样。输入过滤做拒绝或替换——命中禁用内容直接拦截命中敏感信息则替换成掩码输出过滤做静默修改——把模型返回里的敏感实体打码后再交给下游。前者往往需要抛异常让链路中断后者只是改一个字符串字段两者在同一组中间件里各司其职即可。实现上有一点要提醒正则和海量规则匹配是有成本的每次模型调用都跑一遍全量规则延迟可能增加几十毫秒。规则集要在中间件启动时编译好不要每次匹配都重新编译必要时对提示词做长度截断后再扫避免超长文本拖垮过滤性能。4. 高频踩坑清单异步上下文丢失、递归重入与中间件顺序反直觉4.1 异步上下文丢失的定位过程前面说过线程池和 create_task 都会导致 trace_id 丢失但真正定位到根因的过程比想象中曲折。第一次发现异常是监控里看到某类日志的 trace_id 字段接近一半为空而且空的记录集中在某些工具相关的日志里。刚开始以为是日志参数传错了查了半天才发现这些工具内部用了 asyncio.to_thread 包装同步代码线程里的 contextvars 拿到的是空上下文。定位方法其实很简单写个小脚本在工具内部打印 contextvars 里的 trace_id再对比主链路里的值很快就能确认是不是上下文丢失。修复时注意contextvars.copy_context() 只能解决创建一个新任务时提前拷贝的场景没法解决线程池这种运行时值已经改变的情况。所以最可靠的还是把 ctx 显式传进工具工具内部需要 trace_id 时直接读参数根本不去依赖隐式传播。4.2 工具内部再调 Agent 引发的递归重入有一个工具专门做总结当前会话它内部会再次调用主 Agent。于是工具调用中间件内部又嵌套发起一次新的 run新 run 又走整套中间件。此时如果外层限流中间件按租户令牌桶限流嵌套调用可能被自己的外层调用吃掉额度甚至触发互斥外层模型调用在等工具返回工具内部的模型调用又被同一个配额拒绝整个链路卡死。处理方式是在 ctx 里记录 nesting_depth嵌套 run 默认跳过配额类中间件或者使用独立的嵌套配额。同时要防重入风暴限制最大嵌套层数超过则直接拒绝工具调用并给模型一个明确错误说明。这部分逻辑一定要放在最外层的中间件里因为它需要看到完整的嵌套结构。我也见过有人用全局计数器防重入但全局计数器在并发下会互相污染必须用 ctx 里的游标。4.3 注册顺序与执行顺序相反中间件排列的真实语义我见过不止一次有人把日志中间件注册在限流中间件里面结果被限流的请求完全没有日志排查限流误伤时两眼一抹黑。回顾洋葱模型注册在外的中间件包住注册在里的中间件所以调用前逻辑外面先执行。限流短路时短路发生在外层内层根本不会执行。于是你期望的所有请求都有日志实际变成了只有通过的请求有日志。正确的排列规则是日志、追踪这类观测性中间件放最外层业务控制类限流、熔断、鉴权放中间层重试、超时这类靠近实际调用的放内层。这样被任何内层拦截的请求外层都能留下完整记录。异常兜底中间件同样必须放最外层它才能接住所有内部异常。每次新增中间件先想清楚它属于哪一层别把所有逻辑一股脑堆在同一个层级。4.4 异常吞掉问题谁负责兜底谁来放行新手写中间件最容易踩的坑是过度防护except Exception: return ModelResponse(content抱歉出了点问题)。看起来安全实际上会让 Agent 编排器把错误文案当成模型的正常回复继续分析可能拿这个假回复再去调工具浪费 Token 还掩盖根因。我的原则是中间件里除了明确的兜底中间件其他中间件一律只记录不吞异常兜底中间件即使返回友好错误也要在响应里打标记让编排器识别出来这不是模型真实输出并对下游做补偿。错误处理最怕的就是看起来成功了。宁可让错误显式冒泡导致一次运行快速失败也不要让错误变成一条被正常解析的文本。5. 测试与编排让中间件成为可持续演进的拦截面5.1 不跑完整 Agent 也能单测中间件中间件机制的另一个红利是测试友好。你不需要启动整个 Agent 运行时、连上真实模型服务就能把单个中间件测明白。核心思路构造一个假的 ctx给一个假的 next_call直接调用中间件的 wrap_ 方法断言调用前后的行为async def test_observability_logs_and_passes_through(): mw ObservabilityMiddleware() ctx RunContext(run_idrun-1, trace_idtrace-1) async def fake_next(request): return ModelResponse(usageUsage(input_tokens10, output_tokens5)) resp await mw.wrap_model_call( ctx, fake_next, ModelRequest(modeltest-model, messages[hi]) ) assert resp.usage.output_tokens 5 assert ctx.usage.output_tokens {test-model: 5}短路中间件则测不调用 next_call路径传入一个会抛异常的 fake_next如果中间件正确短路异常不会触发。限流中间件测试更直接把令牌桶换成计数为 0 的 mock断言抛出 RateLimitExceeded。我的经验是中间件因为逻辑聚焦单测覆盖率比主循环好做得多务必把正常放行、短路路径、异常路径三条分支都覆盖到。5.2 按请求条件编排中间件不同租户、不同场景需要的控制强度不一样内部调试环境可以跳过限流和护栏企业租户要多一层内容审计普通用户用默认配置。实现上不要在中间件内部写大量 if 分支而是在注册层维护一份可过滤的中间件列表class MiddlewareRegistry: def __init__(self, middlewares: list[AgentMiddleware]): self._all middlewares def resolve_for(self, ctx) - list[AgentMiddleware]: if ctx.is_internal_debug: return [ m for m in self._all if not isinstance(m, (RateLimitMiddleware, GuardrailMiddleware)) ] if ctx.tenant_plan enterprise: return [self._audit_middleware] self._all return self._all这样每个中间件只关心自己的事组合策略集中在入口处。每次需求变化先问自己这是新中间件还是旧中间件的配置变化尽量避免一个中间件同时干四五件事。职责单一后面所有的顺序调整、热插拔才做得起来。5.3 性能注意事项与保持无状态中间件每多一层就多一层 await 跳转虽然单次跳转开销很小但模型调用一轮可能要串行调用多个工具乘起来就不可忽视了。我一般把中间件总数控制在 5 层以内每层只做最必要的事正则、规则扫描这类重活只在该中间件确实需要时才启用不要全局盲目堆规则。另一个容易踩的坑是中间件实例状态。中间件实例通常是单例、被所有并发请求共享如果你在 self 上存了当前请求的 trace_id这种临时状态两个并发请求就会互相污染。金色一原则所有请求级状态一律放 ctx中间件类的属性只允许是不可变配置比如重试次数、阈值、规则集引用。最后分享一点个人体会中间件方案刚上线那会儿我一度想把所有控制逻辑都做成中间件结果光看注册列表就已经头晕。后来收敛下来真正长期在线的只有观测、追踪、限流、重试、护栏五个加上一个按需启用的兜底。wrap_model_call 和 wrap_tool_call 这两个钩子之所以好用是因为它们正好卡在 Agent 的两条命脉上——思考与行动把横切逻辑收在两道闸门里系统才谈得上可维护、可观测、可治理。如果你也在重构自己的 Agent 运行时建议从日志和追踪两个中间件起步跑通顺序和上下文的语义之后再逐步把限流、护栏加进来会比一上来就铺满十几个中间件稳妥得多。