1. 从一次踩坑说起为什么中间件才是 Agent 的胜负手去年年底我接手了一个内部知识库问答 Agent 的活儿需求听起来不复杂能查文档、能调接口、能记住上下文、能多轮追问。我一开始用的是最朴素的写法一个AgentExecutor加几个 Tool跑通 demo 只花了半天。结果上线第三天就出事了——用户连续追问五轮之后Agent 开始胡言乱语把上一轮的检索结果当成这一轮的事实依据还煞有介事地编了个不存在的接口返回值。排查了两天才定位到根因不是模型不行也不是 Prompt 写得烂而是上下文管理、工具调用、状态持久化这三件事全糊在一个循环里没有任何一层做拦截和治理。这就好比你家水管、电线、燃气管全捆在一起走平时没事一旦哪根出问题整栋楼都得停。后来我把这套逻辑重构成了带中间件层的结构用的就是DeepAgents 的中间件机制底层依托 LangChain 的createMiddleware体系。重构之后同样的问题再没复现过而且新增一个敏感词过滤能力只花了二十分钟——写一个中间件挂上去就行完全没动主流程代码。这篇就围绕AI Agent 中间件这个主题把 DeepAgents 中间件到底解决什么问题、核心机制怎么运转、怎么落地写一个自己的中间件、以及我踩过的那些坑一次性讲透。不管你是刚接触 LangChain 的新手还是已经搭过几套 Agent 的老手应该都能从里面抠出点能直接抄的东西。先给个一句话定位DeepAgents 中间件是夹在 Agent 主循环和具体执行动作之间的一层可插拔治理层负责在请求进入模型前、工具调用前后、响应返回前这些关键节点做拦截、改写、记录和兜底。它不改变 Agent 的核心决策逻辑但决定了这个 Agent 能不能上生产。2. DeepAgents 中间件到底在解决什么问题2.1 没有中间件的 Agent 长什么样先看一个典型的裸奔 Agent 结构。用 LangChain 搭的话大概是这么个流程# 伪代码展示裸奔结构 while not done: response llm.invoke(messages) # 调模型 if response.has_tool_call: result execute_tool(response.tool) # 调工具 messages.append(result) # 塞回上下文 else: done True这段代码能跑但问题一大堆。我列一下实际生产中会遇到的上下文无限膨胀每轮工具返回都往messages里塞十轮之后 token 直接爆掉要么报错要么被截断截断之后模型就失忆了。工具调用没有权限校验模型说调delete_user就真调了没有任何拦截。失败没有重试和降级工具超时了整个 Agent 就卡死或者直接抛异常给用户。没有可观测性出了事你根本不知道是哪一轮、哪个工具、什么参数导致的。敏感信息裸奔用户手机号、身份证号直接进模型上下文合规上过不去。这些问题有个共同点它们都不是决策问题而是治理问题。模型该不该调工具是决策调之前要不要检查权限是治理。把治理逻辑硬塞进主循环代码会迅速腐烂成一坨意大利面。2.2 中间件层的核心价值定位DeepAgents 中间件的思路本质上是把治理从主循环里抽出来做成一个个独立的、可组合的钩子。你可以把它理解成 Web 开发里的 Express/Koa 中间件——请求进来一层层过每层都能决定是放行、改写还是直接返回。它带来的直接好处有这么几个维度无中间件有中间件上下文管理手动拼接易爆中间件统一裁剪、摘要权限控制散落在各 Tool 里集中拦截一处配置可观测性靠打日志零散统一埋点全链路追踪失败处理各写各的统一重试/降级策略扩展新能力改主流程挂一个中间件我特别想强调最后一行。中间件最大的价值不是解决当下问题而是让未来加需求这件事变得便宜。你想想产品经理今天要加个回答必须带引用来源明天要加个超过三次工具调用就转人工后天要加个输出前做一次敏感词过滤——如果每次都要动主循环你迟早会把主循环改成一个谁都不敢碰的黑盒。而中间件模式下这些都是独立文件加一个挂一个互不干扰。2.3 和 LangChain 原生能力的边界这里得说清楚一个容易混淆的点LangChain 本身有 Callback、有Runnable的before/after那 DeepAgents 中间件和它们是什么关系我的理解是LangChain 的 Callback 偏观察中间件偏干预。Callback 主要是让你知道发生了什么打日志、上报指标它不太方便去改写输入输出、中断流程。而中间件是可以在链路中间动手的——改 Prompt、改工具参数、直接短路返回、抛异常终止。DeepAgents 的createMiddleware就是在这个基础上把干预点标准化了。它定义了几个明确的钩子位置你只需要实现对应的方法框架会在正确的时机调用你。这比自己去 hack Callback 要干净得多也更可控。3. 核心机制拆解中间件的钩子与执行顺序3.1 四个关键拦截点DeepAgents 中间件最核心的设计是把 Agent 的一次完整执行拆成了几个可拦截的节点。我按执行顺序梳理一下这也是理解整个机制的钥匙请求前置beforeModel在消息发给模型之前触发。这里能拿到完整的 messages 列表可以裁剪、可以注入系统提示、可以做敏感信息脱敏。模型响应后afterModel模型返回之后、工具执行之前触发。这里能拿到模型的原始输出可以校验它要调的工具是否合法、参数是否合规。工具执行前beforeTool真正调用工具之前触发。可以做权限校验、参数二次加工、限流。工具执行后afterTool工具返回之后触发。可以做结果清洗、错误包装、缓存写入。这四个点基本覆盖了 Agent 一次循环里所有值得干预的位置。你写中间件就是选一个或多个点去实现。注意不同版本的 DeepAgents 钩子命名可能略有差异有的叫wrapModelCall、wrapToolCall但语义是一致的。核心是理解前置/后置这个二分法命名只是壳。3.2 洋葱模型与执行顺序多个中间件叠加时执行顺序遵循经典的洋葱模型。假设你挂了 A、B、C 三个中间件那么请求进入 → A前置 → B前置 → C前置 → 模型/工具 → C后置 → B后置 → A后置 → 返回这个顺序非常关键直接决定了你的中间件能不能正确工作。举个实际例子如果你有一个脱敏中间件和一个日志中间件脱敏必须在前置阶段先执行日志才能记录到脱敏后的内容。如果你把日志挂在脱敏外层那日志里就会记下原始敏感信息等于白脱敏。我踩过一次这个坑把限流中间件挂在了重试中间件的外层结果重试的每一次都被算作一次独立请求限流阈值瞬间被打满。后来把顺序调过来——限流在外、重试在内——才符合预期。顺序错了逻辑再对也是错的。3.3 中间件的状态传递中间件之间怎么共享数据比如脱敏中间件想把这次请求涉及的用户 ID传给后面的审计中间件。DeepAgents 一般会提供一个 context 对象贯穿整条链路。def my_middleware(): def before_model(state, context): # 从 context 里读或者往里写 context[user_id] extract_user_id(state[messages]) return state return {beforeModel: before_model}这个 context 是单次请求级别的请求结束就销毁不会跨请求污染。这点很重要——我见过有人把 context 写成了全局变量结果并发一上来A 用户的上下文串到了 B 用户那里出了严重的数据泄露。永远不要把请求级状态放到全局。4. 手把手写一个生产级中间件4.1 环境准备与依赖确认动手之前先把环境理清楚。DeepAgents 中间件依赖 LangChain 的核心包版本上要留意不同大版本 API 差异不小。pip install langchain langchain-core deepagents # 确认版本建议锁死 pip show langchain-core我一般会在项目里建一个requirements.txt把版本钉死因为 LangChain 生态迭代快今天能跑的代码下个月可能就报ImportError。这不是危言耸听我被坑过不止一次。langchain0.3.x langchain-core0.3.x deepagents0.x.x提示如果你的项目还要接 Redis 做状态存储热词里提到的 redis 做中间件其实指的是会话状态持久化和这里的治理中间件是两个概念别搞混那还要额外装redis和langchain-redis。4.2 第一个中间件上下文裁剪这是我认为最该优先做的中间件没有之一。因为 token 爆炸是 Agent 上生产后第一个会撞上的墙。思路很简单在beforeModel阶段检查 messages 的总长度超阈值就做裁剪。裁剪策略我一般用保留系统提示 最近 N 轮 对早期内容做摘要。from langchain_core.messages import SystemMessage, trim_messages def context_trim_middleware(max_tokens4000, keep_last6): def before_model(state, context): messages state[messages] # 先做 token 估算粗略按字符数 / 3 估 estimated sum(len(str(m.content)) for m in messages) // 3 if estimated max_tokens: return state # 超了就裁剪保留系统消息和最近几轮 trimmed trim_messages( messages, max_tokensmax_tokens, strategylast, token_counterlen, include_systemTrue, ) context[trimmed] True return {**state, messages: trimmed} return {beforeModel: before_model}这里有几个细节值得说。第一token 估算我用的是字符数除以 3这种土办法因为精确计算要调 tokenizer每次请求都算一遍开销不小。粗略估算留 20% 余量就够了。第二include_systemTrue一定要开否则系统提示被裁掉Agent 直接失忆行为会变得莫名其妙。第三我在 context 里打了个trimmed标记方便后面审计中间件知道这次请求被裁剪过。实测下来这个中间件能把长对话场景的 token 消耗压下来 60% 以上而且因为保留了最近几轮用户几乎感知不到失忆。4.3 第二个中间件工具权限校验工具权限这事我强烈建议不要写在每个 Tool 内部。原因很简单Tool 是给模型看的模型可能被诱导去调不该调的工具而权限是业务规则应该独立于 Tool 存在。TOOL_PERMISSIONS { search_docs: [user, admin], query_database: [admin], delete_record: [admin], } def tool_permission_middleware(get_user_role): def before_tool(tool_name, tool_input, context): role get_user_role(context) allowed TOOL_PERMISSIONS.get(tool_name, []) if role not in allowed: # 直接短路返回一个拒绝结果给模型 return { blocked: True, reason: f当前角色 {role} 无权调用 {tool_name}, } return None # 返回 None 表示放行 return {beforeTool: before_tool}这个中间件的关键设计是短路返回。当权限不足时不是抛异常终止整个 Agent而是返回一个结构化的拒绝结果让模型自己决定下一步——比如告诉用户这个操作需要管理员权限。这样用户体验是连贯的而不是突然报错。注意短路返回的内容一定要清晰告诉模型发生了什么否则模型可能会反复重试同一个被拒的工具陷入死循环。我一般会在拒绝信息里明确写请勿重试直接告知用户。4.4 第三个中间件可观测性埋点这个中间件不改变任何行为但它是你排查问题的命根子。我要求团队里所有 Agent 项目必须挂这个。import time import logging logger logging.getLogger(agent.trace) def observability_middleware(): def before_model(state, context): context[model_start] time.time() context[trace_id] generate_trace_id() return state def after_model(response, context): cost time.time() - context.get(model_start, time.time()) logger.info({ trace_id: context[trace_id], stage: model, latency: round(cost, 3), output_len: len(str(response)), }) return response def after_tool(tool_name, result, context): logger.info({ trace_id: context.get(trace_id), stage: tool, tool: tool_name, result_len: len(str(result)), }) return result return { beforeModel: before_model, afterModel: after_model, afterTool: after_tool, }埋点数据我一般会打到结构化日志里再接到监控平台。有了 trace_id一次请求从进到出经过哪些中间件、每个环节耗时多少、调了哪些工具全都串得起来。出问题的时候直接拿 trace_id 一搜比翻半天日志高效太多。4.5 组装与注册三个中间件写好了怎么挂上去DeepAgents 的createMiddleware一般支持传入一个中间件列表按顺序执行。from deepagents import create_agent, createMiddleware agent create_agent( modelllm, tools[search_docs, query_database], middlewarecreateMiddleware([ observability_middleware(), # 最外层记录全链路 context_trim_middleware(), # 上下文治理 tool_permission_middleware(get_user_role), # 权限 ]), )顺序上我的经验是可观测性放最外层治理类放内层。因为可观测性要记录的是最终实际执行的内容放外层能捕获到所有内层中间件处理后的结果。而权限校验要尽量靠近工具执行避免被其他中间件干扰。5. 常见问题与排查实录5.1 中间件不生效的几种典型情况这是新手最容易卡住的地方。我整理了一个速查表现象可能原因排查方法中间件完全没被调用钩子名写错打印中间件注册日志确认 key 名前置生效后置不生效前置里抛了异常检查前置返回值是否合法顺序和预期不符列表顺序反了记住洋葱模型外层先执行改了 state 但没生效没返回新 state前置必须 return 修改后的 state并发下数据串了用了全局变量改用 context 传请求级状态我重点说下第一条。钩子名这东西不同版本真的会变。我遇到过beforeModel和before_model两种写法写错了框架不报错就是静默不调用特别坑。建议在中间件里加一行日志确认它真的被触发了。5.2 上下文裁剪把关键信息裁没了这个坑我踩得很深。有一次用户问我刚才提到的那个订单号是多少结果裁剪中间件把包含订单号的那轮对话裁掉了Agent 一脸茫然。解决办法是给裁剪加锚点保护。具体做法是在裁剪前扫描 messages把包含关键实体订单号、用户 ID、金额等的消息标记为不可裁裁剪时跳过它们。def is_anchor_message(msg): # 简单实现包含数字 ID 模式的消息视为锚点 import re return bool(re.search(r\b\d{6,}\b, str(msg.content)))这个逻辑不复杂但能极大提升裁剪后的对话连贯性。你也可以做得更精细比如用一个小模型来判断哪条消息重要但那样开销就上去了看场景取舍。5.3 工具重试导致的重复副作用重试中间件是个好东西但用在有副作用的工具上会出大事。比如下单工具超时了你重试一次用户就被下了两单。我的处理原则是只对幂等工具开启自动重试。在工具定义里加一个idempotent标记重试中间件只对标记为 True 的工具生效。IDEMPOTENT_TOOLS {search_docs, query_database} def retry_middleware(max_retry2): def after_tool(tool_name, result, context): if tool_name not in IDEMPOTENT_TOOLS: return result if is_error(result) and context.get(retry_count, 0) max_retry: context[retry_count] context.get(retry_count, 0) 1 return {retry: True} return result return {afterTool: after_tool}提示非幂等工具的重试一定要配合业务侧的幂等键比如订单号去重否则迟早出事故。5.4 中间件性能开销中间件多了每次请求都要过一遍开销是实打实的。我实测过三个轻量中间件裁剪、权限、埋点加起来大概增加 5-15ms基本可以忽略。但如果你在中间件里做了同步的网络调用比如每次都查一次数据库校验权限那开销就上去了。优化思路有两个一是缓存权限这类变化不频繁的数据可以缓存几分钟二是异步化埋点日志这种不阻塞主流程的操作扔到后台队列去写。6. 中间件设计的几条经验法则写了这么多中间件我总结出几条自己一直遵守的原则分享出来供参考。第一一个中间件只做一件事。我见过有人把裁剪、脱敏、埋点全塞一个中间件里结果改一处影响三处维护起来痛不欲生。拆开之后每个文件职责单一测试也好写。第二中间件要能独立开关。通过配置控制每个中间件是否启用这样出问题的时候可以快速关掉某个中间件定位问题而不用改代码重新部署。第三中间件不能假设自己一定被调用。框架版本升级、配置错误都可能导致中间件失效所以核心业务逻辑不能依赖中间件来保证正确性。中间件是增强不是必需。第四给中间件写测试。尤其是权限和裁剪这种逻辑一定要有单元测试覆盖边界情况。我一般会构造一批恶意输入来测权限中间件比如让模型尝试调用不存在的工具、传超长参数等。第五日志要能区分中间件。每个中间件打日志时带上自己的名字排查时一眼就能看出是哪一层出的问题。最后再分享一个我最近在用的技巧把中间件的执行情况也纳入 Agent 的自我评估。比如裁剪中间件触发了就在最终响应里加一个隐式的标记让评估系统知道这次回答是基于裁剪后的上下文生成的准确率评估时要打个折扣。这个思路有点绕但确实能提升整体评估的准确性。这套中间件体系搭下来我的 Agent 项目从能跑 demo到敢上生产之间少走了至少两个月的弯路。如果你现在正卡在 Agent 稳定性问题上不妨从上下文裁剪和权限校验这两个中间件开始动手收益最直接。