DeepAgents中间件实战：从胶水代码到可维护的Agent工程化架构

1. 从“胶水代码”到“中间件”：DeepAgents 到底在解决什么问题

如果你最近半年在折腾 AI Agent，大概率经历过这样一个阶段：一开始用 LangChain 的AgentExecutor跑通一个能调用工具的对话机器人，感觉挺爽；然后想加个“对话历史压缩”“工具调用重试”“敏感词过滤”“Token 用量统计”，于是开始往主流程里塞各种if-else和回调函数。再往后，业务方要求“同一个 Agent 在不同渠道要有不同的行为”“工具调用前要过审批”“长任务要能断点续跑”，你发现自己的agent.py已经膨胀到八百行，改一处崩三处。

这就是典型的“胶水代码地狱”。而DeepAgents 中间件这个项目，本质上就是冲着这个痛点来的——它把 Agent 执行过程中那些横切的、可复用的、与具体业务逻辑无关的能力，抽成一层可插拔的中间件管道。你可以把它理解成 Web 框架里的 middleware：请求进来，先过日志、再过鉴权、再过限流，最后才到业务 handler；Agent 也一样，用户输入进来，先过上下文管理、再过工具权限校验、再过 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(f"enter: {context.step}") result = next_handler(context) logger.info(f"exit: {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_ratio=0.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；参数校验用pydantic（LangChain 本身就依赖它）。

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_handler=agent_core )

为什么日志放最外层？因为它要记录整条链路的耗时，包括所有中间件的开销。为什么预算放次外层？因为它要在工具校验之前就拦住超支请求，避免白跑一遍校验。为什么上下文管理放最内层？因为它最贴近核心执行器，改动频率最高，放内层方便单独替换。

这个顺序不是绝对的，但每次调整都要问自己：这个中间件需要看到“原始请求”还是“处理后的请求”？需要拦截“所有下游”还是“部分下游”？想清楚这两个问题，顺序自然就定了。

4.4 接入 Redis 做状态持久化

长任务场景下，中间件的执行状态必须能跨进程恢复。我用 Redis 存两类数据：一是会话级的上下文快照，二是审批挂起时的待办队列。

import json import redis r = redis.Redis(host="localhost", port=6379, db=0) def save_context(session_id, context): # 只序列化可变状态和 scratch，请求元信息可重建 payload = { "state": context.state, "scratch": context.scratch, } r.setex(f"agent:ctx:{session_id}", 3600, json.dumps(payload)) def load_context(session_id, user_input): raw = r.get(f"agent: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 存了不可序列化对象	只存基础类型，复杂对象存 ID

5.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(f"agent: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 再到中间件管道，本质上是一回事。

最后分享一个小技巧：如果你刚开始引入中间件，别一上来就搞七八个。先从日志和预算两个开始，跑通管道、建立直觉，再逐步往里加。每加一个中间件，都问自己“它解决的是哪类问题、放在哪一层、失败了怎么兜底”。想清楚这三个问题再加，比事后返工省事得多。