1. 从“胶水代码”到“中间件”DeepAgents 到底在解决什么问题如果你最近半年在折腾 AI Agent大概率经历过这样一个阶段一开始用 LangChain 的AgentExecutor跑通一个能调用工具的对话机器人感觉挺爽然后想加个“对话历史压缩”“工具调用重试”“敏感词过滤”“Token 用量统计”于是开始往主流程里塞各种if-else和回调函数。再往后业务方要求“同一个 Agent 在不同渠道要有不同的行为”“工具调用前要过审批”“长任务要能断点续跑”你发现自己的agent.py已经膨胀到八百行改一处崩三处。这就是典型的“胶水代码地狱”。而DeepAgents 中间件这个项目本质上就是冲着这个痛点来的——它把 Agent 执行过程中那些横切的、可复用的、与具体业务逻辑无关的能力抽成一层可插拔的中间件管道。你可以把它理解成 Web 框架里的 middleware请求进来先过日志、再过鉴权、再过限流最后才到业务 handlerAgent 也一样用户输入进来先过上下文管理、再过工具权限校验、再过 Token 预算控制最后才到模型推理。我最早接触这个概念是在 LangChain 生态里看到createMiddleware这个 API 设计当时第一反应是“这不就是 Koa 的洋葱模型吗”。后来实际用 DeepAgents 跑了一个多月的生产任务才真正体会到中间件化对 Agent 工程化的价值。它解决的不是“能不能跑通”的问题而是“能不能维护、能不能观测、能不能安全上线”的问题。这篇文章适合三类人看一是已经用 LangChain 或类似框架搭过 Agent、但被维护成本折磨的开发者二是正在做 Agent 平台化、需要给多个业务方提供统一能力的团队三是对 Agent 架构演进感兴趣、想了解“中间件”这层抽象到底怎么落地的人。我会从设计思路、核心机制、实操搭建、踩坑排查四个维度展开尽量把每个“为什么这么设计”讲透而不是只丢一段能跑的代码。提示本文涉及的代码示例基于 LangChain 生态的通用模式DeepAgents 的具体 API 名称可能随版本变化重点是理解中间件的分层思想和实现套路迁移到其他框架如 Dify、CrewAI 的插件机制也是相通的。2. 中间件分层设计为什么不是“写个装饰器就完事”2.1 从洋葱模型理解 Agent 中间件的执行顺序很多人第一次听到“Agent 中间件”会下意识觉得“不就是给工具函数加个装饰器吗”。这个理解只对了一半。装饰器解决的是单个函数的增强而中间件解决的是整条执行链路的编排。区别在哪我举个例子。假设你的 Agent 一次完整执行是这样的接收用户消息 → 加载历史 → 拼接 Prompt → 调用 LLM → 解析工具调用 → 执行工具 → 把结果回填 → 再次调用 LLM → 返回最终答案。如果只用装饰器你只能分别增强“调用 LLM”和“执行工具”这两个点但像“Token 预算在整条链路累计消耗”“上下文在多次 LLM 调用之间动态压缩”这类跨阶段逻辑装饰器就无能为力了。DeepAgents 的中间件采用的是洋葱模型也就是每个中间件都有“进入”和“退出”两个切面。请求从最外层中间件进入逐层向内到达核心执行器后再逐层向外返回。这个设计的好处是你可以在“进入”时做前置校验比如检查用户配额在“退出”时做后置处理比如统计本次消耗、写入审计日志而且外层中间件能拿到内层所有阶段的汇总结果。# 中间件洋葱模型的伪代码示意 class Middleware: def before(self, context): # 进入时执行从外到内 pass def after(self, context, result): # 退出时执行从内到外 pass # 执行顺序A.before - B.before - core - B.after - A.after pipeline [MiddlewareA(), MiddlewareB()]这个顺序非常关键。我踩过一个坑把“Token 统计中间件”放在了最内层结果发现它只能统计到单次 LLM 调用的消耗拿不到工具调用产生的额外 Token比如某些工具内部也会调模型。后来把它移到最外层才拿到了完整的链路消耗。所以中间件的位置本身就是一种语义放错位置等于逻辑写错。2.2 核心中间件类型拆解哪些能力值得抽出来不是所有逻辑都值得做成中间件。我的判断标准是是否横切多个 Agent、是否与业务无关、是否需要独立配置。满足这三条的才值得抽。基于这个标准DeepAgents 这类项目通常会内置以下几类中间件。中间件类型核心职责典型触发时机是否可省略上下文管理历史压缩、摘要生成、窗口裁剪每次 LLM 调用前短对话可省工具权限校验白名单、参数校验、审批流工具执行前生产环境必开Token 预算控制累计消耗统计、超限熔断全链路按量付费必开重试与降级失败重试、模型切换、兜底回复LLM/工具调用失败时建议开启可观测性日志、Trace、指标上报全链路建议开启内容安全输入输出过滤、敏感词拦截输入后、输出前合规必开这张表是我自己在项目里总结的不一定和官方文档完全一致但覆盖了 90% 的实战场景。你会发现一个规律越靠近“核心推理”的中间件越偏技术越靠近“外层”的中间件越偏业务和合规。这个分层不是随便定的它决定了你的中间件栈应该怎么排。2.3 为什么用createMiddleware而不是继承基类LangChain 生态里createMiddleware这种工厂函数的设计一开始我觉得多此一举——直接写个类继承BaseMiddleware不香吗用了一段时间才明白它的用意降低心智负担鼓励组合而非继承。继承的问题在于一旦你的中间件需要复用另一段逻辑就得处理多继承或者把逻辑塞进基类最后基类越来越胖。而工厂函数返回的是一个配置对象你可以用函数组合的方式把多个小能力拼起来。比如“带重试的日志中间件”就是“重试中间件”和“日志中间件”的组合而不是一个继承了两者的新类。# 工厂函数风格组合优于继承 def create_logging_middleware(logger): def handler(context, next_handler): logger.info(fenter: {context.step}) result next_handler(context) logger.info(fexit: {context.step}) return result return handler def create_retry_middleware(max_retries): def handler(context, next_handler): for i in range(max_retries): try: return next_handler(context) except Exception as e: if i max_retries - 1: raise return None return handler这种写法还有个隐性好处测试极其方便。每个工厂函数返回的都是纯函数你可以单独 mocknext_handler来验证前置逻辑不用启动整个 Agent。我在做 Token 预算中间件的单元测试时就是靠这个特性把覆盖率做到了 90% 以上。3. 核心机制深挖中间件管道是怎么跑起来的3.1 执行上下文Context的设计与传递中间件之间靠什么通信答案是执行上下文。这是整个中间件体系里最容易被低估、也最容易设计错的部分。上下文本质上是一个贯穿整条链路的数据袋里面装着用户输入、历史消息、当前步骤、累计消耗、临时变量等等。设计上下文时有两个极端要避免。一是上下文太瘦只放一个messages列表结果每个中间件都要自己去查数据库、调接口拿信息重复劳动且不一致。二是上下文太胖把所有东西都塞进去导致序列化困难、内存暴涨、调试时打印出来几千行。我的经验是遵循“按需注入、分层存放”原则。把上下文分成三层不可变的请求元信息用户 ID、会话 ID、渠道、可变的执行状态当前步骤、重试次数、累计 Token、以及中间件私有的临时数据用命名空间隔离避免键冲突。class ExecutionContext: def __init__(self, user_input, session_id): # 不可变层 self.request {input: user_input, session_id: session_id} # 可变状态层 self.state {step: 0, tokens_used: 0, retries: 0} # 中间件私有层按中间件名隔离 self.scratch {} def get_scratch(self, namespace): return self.scratch.setdefault(namespace, {})这个scratch命名空间的设计是我踩坑后加的。早期所有中间件共用一个字典结果两个中间件都用了temp这个键互相覆盖排查了半天才发现。加上命名空间后每个中间件只能访问自己的区域冲突问题彻底消失。3.2 工具调用拦截权限、参数与审批流工具调用是 Agent 最容易出安全事故的地方。模型可能被诱导调用删除数据的工具也可能传入非法参数导致下游服务崩溃。中间件在这里的价值就是在“模型决定调用”和“工具真正执行”之间插一道闸门。DeepAgents 的工具拦截中间件通常做三件事。第一是白名单校验检查工具名是否在当前 Agent 的允许列表里。第二是参数 Schema 校验用 Pydantic 或 JSON Schema 验证参数类型和范围。第三是审批流对于高危工具比如发邮件、改数据库先挂起等待人工确认。def create_tool_guard_middleware(allowed_tools, high_risk_tools): def handler(context, next_handler): tool_call context.state.get(pending_tool_call) if tool_call: if tool_call[name] not in allowed_tools: raise PermissionError(f工具 {tool_call[name]} 不在白名单) if tool_call[name] in high_risk_tools: # 挂起等待审批实际项目中会写入审批队列 context.state[awaiting_approval] True return {status: pending_approval} return next_handler(context) return handler这里有个实操细节审批挂起后Agent 的状态要能持久化。否则用户审批完进程重启上下文丢了任务就断了。我在项目里是把上下文序列化后存 Redis审批回调时再反序列化恢复。这也是为什么热词里会出现“redis 做中间件”——它不只是缓存更是 Agent 长任务状态的中转站。3.3 Token 预算与熔断别让一次对话烧掉一个月预算Token 消耗是 Agent 成本的大头尤其是带工具循环的场景一次任务可能调用十几次 LLM。没有预算控制的 Agent就像没有限额的信用卡迟早出事。Token 预算中间件的核心逻辑是在每次 LLM 调用前预估消耗调用后累计实际消耗超过阈值就熔断。预估消耗有个简单公式预估 Token 输入字符数 / 4 最大输出 Token。这个 4 是英文的平均字符 Token 比中文大概是 1.5 到 2 个字符一个 Token所以中文场景要调整系数。我一般会留 20% 的缓冲避免预估偏低导致超支。def create_token_budget_middleware(max_tokens, buffer_ratio0.2): def handler(context, next_handler): used context.state[tokens_used] estimated estimate_tokens(context) * (1 buffer_ratio) if used estimated max_tokens: # 熔断返回兜底回复而不是抛异常 return {status: budget_exceeded, message: 本次任务已达预算上限} result next_handler(context) context.state[tokens_used] result.get(tokens, 0) return result return handler注意熔断时返回兜底回复而不是抛异常是因为抛异常会中断整个 Agent 循环用户看到的是报错而兜底回复能让用户知道“任务被截断了”体验更好。这个细节在官方文档里通常不会写但生产环境很关键。4. 从零搭建一条中间件管道完整实操4.1 环境准备与依赖选型动手之前先把环境理清楚。DeepAgents 本身是 LangChain 生态的延伸所以基础依赖是langchain和langchain-core。如果你要用 Redis 做状态存储加redis和redis-py要做可观测性加langfuse或opentelemetry参数校验用pydanticLangChain 本身就依赖它。pip install langchain langchain-core redis pydantic # 可观测性可选 pip install langfuse opentelemetry-sdk版本上有个坑要提醒LangChain 的 API 在 0.1 到 0.3 之间变动很大AgentExecutor在 0.3 之后逐渐被 LangGraph 取代。如果你看的教程和你的版本对不上先pip show langchain确认版本别硬套。我建议新手直接用较新的版本因为中间件相关的抽象在新版本里更成熟。4.2 定义第一个中间件请求日志与耗时统计第一个中间件我建议从最简单的日志开始目的是验证管道能跑通同时建立对执行顺序的直觉。import time import logging logger logging.getLogger(agent.middleware) def create_logging_middleware(): def handler(context, next_handler): step context.state[step] start time.time() logger.info(f[step {step}] 进入中间件输入长度{len(context.request[input])}) try: result next_handler(context) elapsed time.time() - start logger.info(f[step {step}] 执行完成耗时{elapsed:.2f}s) return result except Exception as e: elapsed time.time() - start logger.error(f[step {step}] 执行失败耗时{elapsed:.2f}s错误{e}) raise return handler这段代码看起来平平无奇但它建立了一个重要模式中间件要同时处理成功和失败两条路径。很多新手只写成功路径结果一出错日志就断了排查时两眼一抹黑。把try-except加上失败时也记录耗时和错误这才是生产级日志。4.3 组装管道顺序决定行为中间件的组装顺序直接决定行为这是整个实操里最需要动脑的地方。我的一般原则是可观测性在最外层安全校验在次外层业务逻辑相关的在内层核心执行器在最内。def build_pipeline(middlewares, core_handler): # 从右到左包裹最左边的最先执行 handler core_handler for mw in reversed(middlewares): handler mw(handler) return handler pipeline build_pipeline( middlewares[ create_logging_middleware(), # 最外层日志 create_token_budget_middleware(100000), # 次外层预算 create_tool_guard_middleware(allowed, high_risk), # 内层工具校验 create_context_middleware(), # 最内层上下文管理 ], core_handleragent_core )为什么日志放最外层因为它要记录整条链路的耗时包括所有中间件的开销。为什么预算放次外层因为它要在工具校验之前就拦住超支请求避免白跑一遍校验。为什么上下文管理放最内层因为它最贴近核心执行器改动频率最高放内层方便单独替换。这个顺序不是绝对的但每次调整都要问自己这个中间件需要看到“原始请求”还是“处理后的请求”需要拦截“所有下游”还是“部分下游”想清楚这两个问题顺序自然就定了。4.4 接入 Redis 做状态持久化长任务场景下中间件的执行状态必须能跨进程恢复。我用 Redis 存两类数据一是会话级的上下文快照二是审批挂起时的待办队列。import json import redis r redis.Redis(hostlocalhost, port6379, db0) def save_context(session_id, context): # 只序列化可变状态和 scratch请求元信息可重建 payload { state: context.state, scratch: context.scratch, } r.setex(fagent:ctx:{session_id}, 3600, json.dumps(payload)) def load_context(session_id, user_input): raw r.get(fagent:ctx:{session_id}) ctx ExecutionContext(user_input, session_id) if raw: payload json.loads(raw) ctx.state payload[state] ctx.scratch payload[scratch] return ctx这里有个坑scratch里如果存了不可序列化的对象比如数据库连接、函数引用json.dumps会直接报错。我的做法是在中间件里约定scratch只存基础类型和字典列表需要存复杂对象时用 ID 引用恢复时再查回来。这个约定要写进团队规范否则迟早有人往里塞奇怪的东西。5. 常见问题与排查技巧实录5.1 中间件不生效先查这三个地方中间件写了但没执行是最常见的问题。我总结了三个排查方向按顺序查基本能定位。第一检查组装顺序。build_pipeline是从右到左包裹的如果你把中间件列表理解反了可能你以为在最外层的实际在最内层被其他中间件短路了。打印一下最终 handler 的包裹结构或者在最外层中间件打个日志看是否被调用。第二检查短路逻辑。某个中间件如果提前return了后面的中间件和核心执行器都不会跑。比如预算中间件熔断后直接返回工具校验中间件自然就不会执行。这是设计如此但如果你忘了就会误以为“中间件没生效”。第三检查异常吞没。有些中间件用try-except包住了next_handler但没重新抛出异常导致错误被静默吞掉表现为“中间件好像跑了但没效果”。记住捕获异常后要么处理要么重新抛出绝不能默默吞掉。现象可能原因排查方法中间件日志完全不打印组装顺序错误或未加入管道打印 handler 包裹链部分中间件不执行上游中间件短路返回检查各中间件的 return 分支中间件执行了但结果不对上下文键冲突或数据被覆盖用 scratch 命名空间隔离报序列化错误scratch 存了不可序列化对象只存基础类型复杂对象存 ID5.2 Token 统计对不上可能是这三个原因Token 统计不准是另一个高频问题。我遇到过三种情况。一是只统计了 LLM 调用漏了工具内部的模型调用。有些工具比如摘要工具内部也会调 LLM这部分消耗如果不算进去总账就对不上。解决办法是把统计中间件放最外层或者让工具也走同一套统计逻辑。二是流式输出的 Token 统计时机不对。流式场景下Token 是逐步产生的如果你在流开始时就统计拿到的是 0在流结束后统计又可能因为异常中断而漏统计。正确做法是监听流的on_llm_end回调在回调里累加。三是不同模型的 Token 计算方式不同。同样一段中文不同厂商的 Tokenizer 切出来的 Token 数可能差 20% 以上。如果你的 Agent 支持多模型切换统计逻辑要按模型分别处理不能用一个系数套所有。5.3 审批挂起后任务丢失状态持久化的坑审批流是中间件里最容易出问题的场景。用户提交任务中间件挂起等待审批结果服务重启任务没了。或者审批通过后恢复执行时上下文对不上工具参数丢了。我的解决方案是双写 幂等恢复。挂起时把上下文快照和待审批的工具调用一起写入 Redis同时给这个任务分配一个唯一 ID。审批回调时用这个 ID 找回上下文重新注入管道从挂起点继续执行。关键是恢复时要跳过已经执行过的中间件否则日志会重复、预算会重复扣。def resume_from_approval(task_id, approved): payload json.loads(r.get(fagent:task:{task_id})) if not approved: return {status: rejected} ctx rebuild_context(payload) ctx.state[resume_from] payload[checkpoint] # 从检查点恢复跳过已执行的中间件 return pipeline_resume(ctx)这个checkpoint机制是我踩了两次坑之后加的。第一次没做检查点恢复时从头跑结果工具被调用了两次差点造成数据重复。第二次做了检查点但没做幂等网络重试导致恢复执行了两次。第三次才把幂等键加上用任务 ID 步骤号做唯一约束彻底解决。5.4 性能优化中间件太多导致延迟上升中间件不是越多越好。我做过一个测试在一条链路上加到 8 个中间件后单次请求的额外延迟到了 200ms 以上其中大部分是序列化和日志 IO 的开销。优化方向有三个。一是合并同类中间件。比如“日志”和“指标上报”可以合并成一个可观测性中间件共享一次上下文遍历。二是异步化 IO 操作。日志写文件、指标上报这些不阻塞主流程的操作丢到后台队列异步处理。三是按需启用。不是所有请求都需要全量中间件比如内部调试请求可以跳过安全校验短对话可以跳过上下文压缩。# 按需启用根据请求特征动态选择中间件 def select_middlewares(context): mws [logging_mw, budget_mw] if context.request.get(channel) ! internal_debug: mws.append(tool_guard_mw) if len(context.request[input]) 2000: mws.append(context_compress_mw) return mws这个动态选择的设计让我们的平均延迟降了 40% 左右。核心思路是中间件是能力不是负担按需加载才是正解。6. 中间件化之后Agent 工程还能往哪走把中间件这层抽象跑通之后我发现很多之前觉得棘手的问题都有了新的解法。比如多租户场景不同租户要不同的工具权限和预算以前是给每个租户写一套 Agent现在是同一套 Agent 配不同的中间件组合配置化解决。再比如 A/B 测试想对比两种上下文压缩策略的效果以前要改代码重新部署现在是两个中间件实现按流量比例挂载线上直接对比。还有一个我觉得很有潜力的方向是中间件的可观测性反哺。当所有关键逻辑都走中间件后你天然获得了一条完整的执行轨迹每个中间件进入退出、每步耗时、每次 Token 消耗、每次工具调用。把这些数据沉淀下来可以做成本归因、可以做异常检测、甚至可以做自动调优——比如发现某个工具调用失败率高自动降级到备用工具。我个人在实际操作中的体会是中间件这套东西的价值不在于“少写代码”而在于“把变化点集中管理”。Agent 这个领域变化太快了模型在换、工具在加、合规要求在变如果每次变化都要动核心逻辑系统迟早失控。中间件让你把变化挡在外围核心执行器保持稳定。这个思路和当年 Web 框架从“面条式代码”演进到 MVC 再到中间件管道本质上是一回事。最后分享一个小技巧如果你刚开始引入中间件别一上来就搞七八个。先从日志和预算两个开始跑通管道、建立直觉再逐步往里加。每加一个中间件都问自己“它解决的是哪类问题、放在哪一层、失败了怎么兜底”。想清楚这三个问题再加比事后返工省事得多。