Agent中间件实战：从线上事故到生产级智能体防护体系

1. Agent中间件到底在解决什么问题1.1 从一次线上事故说起去年冬天我负责的一个智能体项目在凌晨两点突然开始疯狂调用外部接口短短二十分钟烧掉了将近三百块。排查日志发现模型在某个环节陷入了循环推理反复调用同一个工具而整个链路里没有任何一层能拦住它。那次之后我才真正意识到Agent中间件不是锦上添花的东西而是决定一个智能体能不能上生产的关键基础设施。很多人刚接触Agent开发时注意力都放在提示词怎么写、工具怎么定义、模型选哪个上这些当然重要。但当你的智能体从demo走向真实业务你会发现真正让人头疼的是另一类问题模型调用超预算了怎么办、工具执行失败了要不要重试、多轮对话里上下文越来越长怎么裁剪、敏感操作要不要人工确认、整个执行链路怎么追踪。这些问题靠提示词是解决不了的必须靠中间件。1.2 中间件在Agent架构里的位置用一句话概括Agent中间件是横切在模型调用、工具执行、状态流转这些核心环节之间的一层可插拔逻辑。它不改变Agent的核心决策能力但它在每个关键节点上提供了拦截、增强、观测和控制的能力。打个比方Agent的核心循环就像一条流水线模型是大脑工具是手脚。中间件就是流水线上的质检工位、计数器、急停按钮和监控摄像头。没有它们流水线也能转但你不敢让它转太久也不敢让它处理贵重物料。在LangChain这类框架里中间件通常以**钩子Hook**的形式存在。所谓钩子就是在Agent执行的特定时机自动触发的回调函数。比如模型调用前、模型调用后、工具执行前、工具执行后、整个Agent开始和结束时你都可以挂上自己的逻辑。create_agent这类工厂函数在构建Agent时允许你把一组中间件传进去框架会在合适的时机依次调用它们。1.3 谁需要认真对待中间件如果你只是写个玩具Agent玩玩中间件确实可以先放一放。但只要你符合下面任意一条中间件就是必修课智能体会调用付费接口或消耗token较多的模型智能体会执行有副作用的操作比如写数据库、发消息、下单智能体需要跑在无人值守的环境里比如定时任务、后台服务你需要知道Agent每一步到底干了什么用于调试或审计你的Agent要处理多用户请求需要隔离和限流我见过太多团队在demo阶段一路顺畅一上生产就各种翻车根子往往就出在缺少中间件这层防护。下面我会把中间件的设计思路、核心类型、实操写法和踩坑经验完整拆一遍。2. 中间件的核心类型与设计思路2.1 按执行时机分类的钩子体系中间件最本质的分类维度是执行时机。不同框架叫法不同但核心时机就那么几个。我把它整理成一张表方便你对照理解。钩子时机触发点典型用途before_agentAgent整体开始前初始化上下文、鉴权、加载用户配置before_model每次模型调用前裁剪上下文、注入系统提示、限流after_model每次模型返回后解析输出、校验格式、记录token消耗before_tool每次工具执行前参数校验、权限检查、人工确认after_tool每次工具返回后结果清洗、错误处理、缓存写入after_agentAgent整体结束后汇总统计、持久化、清理资源理解这张表的关键在于Agent的执行是一个循环模型和工具会交替调用多次。所以before_model和before_tool这类钩子会被触发很多次而before_agent和after_agent只触发一次。写中间件时一定要搞清楚自己的逻辑应该挂在哪个时机挂错了要么不生效要么重复执行。2.2 为什么用钩子而不是改源码有人会问我直接改Agent的执行逻辑不就行了为什么要用钩子这个问题我在项目初期也纠结过。后来想明白了钩子机制解决的是关注点分离的问题。Agent的核心逻辑是思考-行动-观察的循环这是相对稳定的。而限流、日志、鉴权、缓存这些是横切关注点它们和核心逻辑正交。如果把这些都塞进核心循环里代码会迅速变成一团乱麻而且每加一个功能就要动一次核心代码风险极高。钩子机制让你可以在不改动核心逻辑的前提下通过组合不同的中间件来定制Agent行为。这就像给手机装壳和贴膜手机本身不用变但你可以根据场景选择不同的保护方案。这种设计带来的另一个好处是可测试性每个中间件都是独立的可以单独写单元测试不用启动整个Agent。2.3 中间件的执行顺序与优先级多个中间件挂在一起时执行顺序就变得很重要。这里有个容易踩的坑不同框架对中间件顺序的处理规则不一样。有的框架按注册顺序正序执行有的在前钩子上正序、后钩子上倒序形成类似洋葱模型的结构。洋葱模型是比较好理解的一种设计。想象你有一层层洋葱最外层是最先注册的中间件。请求进来时从外往里穿过每一层响应回来时从里往外再穿一遍。这样before类钩子按注册顺序执行after类钩子按相反顺序执行。这种设计的好处是资源分配和释放能配对比如你在before里开了个计时器在after里关掉顺序对了才不会乱。我的建议是在写中间件之前先花十分钟确认你用的框架到底怎么处理顺序。可以写个最简单的中间件只打印日志注册两三个跑一次看输出顺序。这个时间绝对值得花我见过因为顺序搞反导致限流失效、日志错乱的案例不止一次。2.4 中间件与Agent记忆、编排的关系热词里频繁出现agent记忆agent框架与编排这些概念和中间件关系密切但容易混淆。简单说记忆解决的是记住什么编排解决的是按什么流程走中间件解决的是在流程的每个点上做什么。记忆系统通常需要中间件来配合比如在after_model钩子里把重要信息写入长期记忆在before_model钩子里从记忆里检索相关内容注入上下文。编排则更多体现在Agent的整体结构上比如是单Agent还是多Agent协作中间件负责在每个Agent的执行点上做统一处理。LangChain和LangGraph的区别也常被问到。粗略讲LangChain更偏向链式调用和工具集成LangGraph更偏向用图结构描述复杂的状态流转。中间件的概念在两者里都有体现但LangGraph因为显式建模了状态和节点中间件的挂载点会更清晰。选哪个取决于你的Agent复杂度简单任务LangChain够用复杂多步流转LangGraph更合适。3. 核心中间件的实操写法3.1 限流与预算控制中间件这是我认为最应该优先实现的中间件。没有它你的Agent就是一个随时可能失控的烧钱机器。核心思路是在before_model和before_tool钩子里做计数和判断。class BudgetMiddleware: def __init__(self, max_tokens100000, max_tool_calls50): self.max_tokens max_tokens self.max_tool_calls max_tool_calls self.token_used 0 self.tool_calls 0 def before_model(self, state): if self.token_used self.max_tokens: raise BudgetExceededError( fToken预算已耗尽: {self.token_used}/{self.max_tokens} ) def after_model(self, state, response): # 从response的元数据里取实际消耗 usage response.get(usage, {}) self.token_used usage.get(total_tokens, 0) def before_tool(self, state, tool_name): self.tool_calls 1 if self.tool_calls self.max_tool_calls: raise BudgetExceededError( f工具调用次数超限: {self.tool_calls} )这段代码的关键点在于预算的判断要放在调用之前消耗的累加要放在调用之后。判断放前面才能拦住累加放后面才能拿到真实数据。另外要注意token消耗不一定每次都能从响应里拿到有些模型或框架不返回usage字段这时候你得用估算的方式兜底比如按字符数除以一个经验系数。提示预算阈值不要设得太死。我一般会设一个软阈值和一个硬阈值软阈值触发告警日志硬阈值才真正中断。这样既能提前发现问题又不会因为偶发的长回复误杀正常请求。3.2 工具权限与人工确认中间件当Agent能执行写操作时权限控制就是刚需。我设计过一个中间件把工具分成三档只读、可写、危险。只读工具直接放行可写工具记录日志危险工具必须经过确认。class PermissionMiddleware: def __init__(self, dangerous_tools, confirm_handler): self.dangerous_tools set(dangerous_tools) self.confirm_handler confirm_handler def before_tool(self, state, tool_name, tool_args): if tool_name in self.dangerous_tools: approved self.confirm_handler( tool_name, tool_args, state ) if not approved: return ToolRejected( f工具 {tool_name} 未获授权已拦截 )confirm_handler是一个回调可以对接多种确认方式控制台询问、发消息到审批群、写一条待办记录等。在无人值守场景下我通常会让它把请求挂起并通知人工人工确认后再恢复执行。这里有个细节要注意拦截工具调用后要给模型一个明确的反馈告诉它这个操作被拒绝了否则模型可能会反复尝试同一个工具陷入死循环。3.3 上下文裁剪中间件多轮对话跑久了上下文会越来越长既费token又可能超出模型窗口。上下文裁剪中间件挂在before_model上在每次调用模型前对消息列表做处理。裁剪策略有好几种我常用的组合是保留系统提示 保留最近N轮 对更早的内容做摘要。具体实现时先判断总长度是否超过阈值没超过就不动超过了再裁剪。裁剪时要注意保持消息的完整性不能把一条工具调用和它的返回结果拆开否则模型会困惑。class ContextTrimMiddleware: def __init__(self, max_messages20, keep_recent10): self.max_messages max_messages self.keep_recent keep_recent def before_model(self, state): messages state[messages] if len(messages) self.max_messages: return system_msgs [m for m in messages if m.role system] recent messages[-self.keep_recent:] older messages[len(system_msgs):-self.keep_recent] summary self.summarize(older) state[messages] system_msgs [summary] recentsummarize可以用一个小模型来做也可以用规则提取关键信息。实测下来用规则提取用户提过的关键实体和已完成的动作往往比让模型总结更稳定因为模型总结有时会丢关键信息或引入幻觉。3.4 可观测性中间件调试Agent最痛苦的就是不知道它内部到底发生了什么。可观测性中间件在每个钩子上打点把执行链路完整记录下来。我一般会记录这些字段时间戳、钩子类型、模型输入输出摘要、工具名和参数、耗时、token消耗、错误信息。这些数据可以写到本地文件也可以发到日志系统。关键是要有一个统一的trace_id串起整个执行链路这样你才能把一次Agent运行的所有事件关联起来。我踩过的坑是早期没加trace_id日志混在一起根本没法排查后来补上之后效率提升非常明显。注意记录日志时一定要对敏感信息做脱敏。工具参数里可能包含用户隐私、密钥、内部地址等直接落盘是安全隐患。我一般会维护一个敏感字段列表记录前统一替换成占位符。4. 中间件组合与实战踩坑记录4.1 多个中间件如何协同真实项目里中间件从来不是单独用的而是组合使用。组合时最大的挑战是错误处理。一个中间件抛异常是应该中断整个Agent还是应该被捕获后继续这取决于中间件的性质。我的经验是分两类保护性中间件限流、权限抛异常应该中断因为继续下去有风险增强性中间件日志、缓存抛异常应该被捕获并降级不能因为记日志失败就让整个Agent挂掉。所以在组合时我会给每个中间件标注它的错误处理策略框架层面统一处理。另一个协同问题是状态共享。多个中间件可能需要读写同一份状态比如限流中间件记录了token消耗可观测性中间件想把它写进日志。这时候需要一个共享的上下文对象所有中间件都从它里面读写。设计这个上下文对象时要注意线程安全和并发隔离多用户场景下不能串数据。4.2 常见问题速查表下面这张表是我在实际项目里积累的问题清单基本覆盖了中间件开发中八成以上的坑。问题现象可能原因排查方向中间件不生效钩子挂错时机确认钩子名和框架版本匹配限流没拦住顺序问题或计数未累加打印执行顺序检查after钩子上下文越裁越长裁剪后又被追加检查裁剪是否在每轮都执行工具被反复调用拒绝后未反馈给模型拦截时返回明确错误消息日志缺字段钩子未覆盖全部路径补全before/after配对并发下数据串了共享状态未隔离按会话ID隔离上下文性能明显下降中间件里有同步阻塞耗时操作改异步或缓存4.3 性能与开销的平衡中间件不是免费的每个钩子都会增加执行开销。我做过一次压测一个包含五个中间件的Agent相比裸Agent单次执行耗时增加了大约百分之十五。这个开销在大多数场景下可以接受但如果你的Agent本身对延迟敏感就得优化。优化的方向有几个把不必要每次都跑的中间件改成条件触发比如日志中间件只在调试模式开启把耗时操作异步化比如写日志、发通知不要阻塞主流程对中间件本身做缓存比如权限检查结果可以缓存一段时间。我一般会在开发环境全开中间件方便调试生产环境只保留必要的几个。4.4 从中间件到Agent安全热词里agent安全agent记忆防御这些话题其实和中间件强相关。很多安全策略最终都要落到中间件上实现。比如防止提示词注入可以在before_model里对输入做检测防止工具被滥用可以在before_tool里做参数白名单校验防止记忆被污染可以在写入记忆前做内容审核。我个人的体会是Agent安全不是一个独立模块而是一组分布在各个执行点上的中间件策略。你不可能在某个单点解决所有安全问题但你可以通过在每个关键节点挂上对应的检查把风险控制在一个可接受的范围内。这个思路和传统后端的纵深防御是一致的。5. 中间件的测试与迭代5.1 怎么给中间件写测试中间件最大的好处之一就是好测试。因为它不依赖真实的模型和工具你可以用假的state和response来驱动它。我一般会为每个中间件写三类测试正常路径、边界条件、异常路径。正常路径验证基本功能比如限流中间件在预算充足时放行。边界条件验证临界值比如token刚好等于阈值时应该放行还是拦截。异常路径验证错误处理比如响应里没有usage字段时会不会崩。这三类测试写下来中间件的可靠性就有基本保障了。5.2 迭代节奏与灰度中间件上线不要一次性全量。我的做法是先在测试环境跑通然后在一个低风险的Agent上灰度观察一段时间再推广。灰度期间重点看两个指标中间件自身的错误率和对Agent整体成功率的影响。如果中间件错误率高说明实现有问题如果Agent成功率下降说明中间件的策略可能过于激进。迭代时还要注意版本兼容。中间件和框架版本、模型版本都可能耦合升级任何一方之前都要回归测试。我吃过一次亏框架小版本升级后钩子的参数签名变了中间件静默失效直到线上出问题才发现。从那以后我养成了习惯升级后第一件事就是跑一遍中间件的测试用例。5.3 一些值得长期投入的方向中间件这层做扎实之后能延伸出很多有价值的能力。比如基于中间件收集的执行数据做Agent评估agent evals用真实运行数据来量化Agent的表现比如基于中间件的拦截能力做成本优化动态选择更便宜的模型处理简单任务再比如基于中间件的编排能力做多Agent协作让不同Agent通过中间件交换信息。我现在维护的中间件库已经积累了大十几个覆盖限流、权限、日志、缓存、重试、脱敏、评估等场景。每次新项目启动直接挑几个组合起来省下的时间非常可观。这套东西的价值会随着项目数量增加而放大越早投入越划算。最后分享一个我踩过的小坑中间件里的异常一定要用自定义异常类型不要直接抛通用的Exception。因为框架或上层代码可能会捕获通用异常做兜底处理导致你的中间件异常被吞掉问题被掩盖。用自定义异常既能精确控制处理逻辑又方便在日志里快速定位。这个细节看起来小但在排查线上问题时能帮你省下大量时间。