1. 为什么要在 DeepAgents 里插一层中间件DeepAgents 是 LangChain 团队在 2025 年推出的智能体框架它把「规划、文件系统、子智能体、上下文压缩」这些能力做成了默认装配的中间件栈。你调用create_deep_agent()的时候框架已经在背后挂了一串AgentMiddleware只是大多数人没注意到。问题也出在这里默认栈能跑通 demo但一旦进入真实业务你马上会遇到三个绕不开的需求——每次工具调用要留审计日志、敏感工具要加人工审批、模型调用要限流防止账单爆炸。这些逻辑如果写进tools[]里LLM 得自己决定什么时候调用不可靠写进业务代码里又会和 Agent 的循环执行纠缠在一起。AgentMiddleware就是为这类「每次自动介入」的场景设计的。它挂在 Agent 运行链路上在 LLM 思考之前、工具执行前后自动触发不需要 LLM 主动决定。你可以把它理解成 Web 框架里的中间件请求进来先过一层鉴权出去再过一层日志业务逻辑本身不用关心这些横切关注点。DeepAgents 的中间件钩子有四个关键位置——before_agent、after_agent、wrap_model_call、wrap_tool_call分别对应整轮开始前、整轮结束后、每次调 LLM 前、每次调工具前后。这篇文章面向已经跑过 LangChain Agent、想给 DeepAgents 加可插拔逻辑的开发者。我会先讲清楚中间件在链路里的位置和钩子执行顺序然后给出可复制的注册配置接着把模型 endpoint 切到 TaoToken 统一走 Key/API 通道最后用一次本地运行验证钩子确实按预期触发并整理几个我实际踩过的报错。全程代码可直接复制不需要你从零搭环境。2. TaoToken 前置把模型 endpoint 统一到一条通道在写中间件之前先把模型调用这条链路理顺。DeepAgents 本身不绑定模型供应商它通过 LangChain 的init_chat_model或直接传 model 字符串来调 LLM。默认情况下你可能用的是各家厂商的直连地址但一旦中间件里要做限流、审计、成本统计调用入口分散在多个 endpoint 就很难集中观测。我的做法是把所有模型请求收敛到 TaoToken 的 API 通道这样中间件里打印的请求日志、统计的 token 消耗都对应同一个出口。TaoToken 提供 OpenAI 兼容的接口Base URL 是https://taotoken.net/api你只需要一个 API Key 就能调用多个模型。对于 DeepAgents 这种需要频繁切换模型做对比测试的场景统一通道的好处很明显换模型只改一个 Model ID不用改 base_url 和鉴权逻辑。下面是我实际用的配置方式通过环境变量注入避免 Key 硬编码进代码。# 在 shell 里设置或写进 .env 由 python-dotenv 加载 export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api然后在 Python 里用init_chat_model构造模型对象。注意model_provider要设成openai因为 TaoToken 走的是 OpenAI 兼容协议base_url指向 TaoToken 的 API 地址api_key从环境变量读。这样构造出来的 model 对象可以直接传给create_deep_agent中间件里拿到的 request 也会带上这条通道的信息。import os from langchain.chat_models import init_chat_model model init_chat_model( modelclaude-sonnet-4-5, # 换成你在 TaoToken 控制台看到的 Model ID model_provideropenai, api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL), )如果你更习惯用ChatOpenAI显式构造效果一样from langchain_openai import ChatOpenAI model ChatOpenAI( modelclaude-sonnet-4-5, api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL), temperature0, )这里有个容易忽略的点DeepAgents 的SummarizationMiddleware在触发上下文压缩时会额外调一次 LLM 做摘要。如果这个摘要调用没走同一条通道你的成本统计就会漏掉一块。用init_chat_model构造的 model 对象会被框架复用摘要调用也走同一个 endpoint这点比手动传字符串 model 更可控。Key 的获取和模型列表可以在 TaoToken 控制台的 API Keys 页面看到接入文档里有各语言的完整示例。3. 可复制配置注册你的第一个 AgentMiddlewareDeepAgents 的中间件注册入口是create_deep_agent(middleware[...])。这个参数接收一个AgentMiddleware实例序列或者被wrap_tool_call装饰过的 callable。关键要记住middleware是追加不是替换。默认栈TodoList、Filesystem、Summarization、Patch、AnthropicCache始终在你传进去的中间件插在 Patch 之后、Memory/HITL 之前。先看最小可运行的配置。下面这段代码注册了一个工具调用日志中间件每次工具执行前后打印一行同时把模型指向 TaoToken。import os from deepagents import create_deep_agent from langchain.agents.middleware import wrap_tool_call from langchain.chat_models import init_chat_model model init_chat_model( modelclaude-sonnet-4-5, model_provideropenai, api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL), ) wrap_tool_call def audit_log(request, handler): 拦截每个工具调用执行前打印工具名和参数执行后打印结束标记。 print(f [MW] 调用工具: {request.name}, 参数{request.args}) try: result handler(request) return result finally: print(f [MW] 工具结束: {request.name}) agent create_deep_agent( modelmodel, middleware[audit_log], # 追加不替换默认栈 )如果你需要更复杂的逻辑比如跨轮次维护状态、修改 system prompt、动态过滤工具列表就得用AgentMiddleware子类。下面这个例子在每轮 Agent 启动前把轮次计数写进 graph state这是官方推荐的存状态方式比改self.x安全得多——后者在子智能体或并行工具调用时会串状态。from langchain.agents.middleware import AgentMiddleware class TurnCounterMiddleware(AgentMiddleware): 每轮 Agent 启动前轮次 1 写入 graph state。 def before_agent(self, state, runtime): n state.get(turn_count, 0) 1 print(f[MW] 第 {n} 轮 Agent 启动) return {turn_count: n} agent create_deep_agent( modelmodel, middleware[audit_log, TurnCounterMiddleware()], )中间件在链路里的插入位置可以用一张顺序图说清楚。用户消息进入后先过PatchToolCallsMiddleware修复悬空 tool_call然后是你的中间件接着是 Profile extras、AnthropicCache、Memory、HITL。这意味着你的中间件看到的是已经修复过的干净历史但看不到 Memory 注入后的 prompt——除非你在wrap_model_call里自己读 state。这个顺序决定了你能拦截什么、不能拦截什么写逻辑前一定要心里有数。子智能体的中间件要单独配。主 Agent 的middleware不会继承给子智能体你需要在 SubAgent 字典里单独写middleware: [...]。这点和很多人直觉相反我第一次用的时候以为会继承结果子智能体的工具调用完全没进日志排查了半天。agent create_deep_agent( modelmodel, middleware[audit_log], # 仅主智能体 subagents[ { name: coder, description: 写代码的子智能体, system_prompt: 你是程序员, middleware: [audit_log], # 子智能体单独挂 }, ], )4. 验证请求本地跑一次看钩子执行顺序配置写完得实际跑一次确认钩子按预期触发。我用的验证脚本会构造一个简单任务让 Agent 调用文件系统工具写一个文件观察wrap_tool_call的打印顺序。先确认依赖装好pip install deepagents langchain langchain-openai python-dotenv然后写验证脚本verify_mw.py。这里我故意让 Agent 执行一个需要多步的任务——先列目录再写文件这样能观察到多次工具调用的钩子触发。import os from dotenv import load_dotenv from deepagents import create_deep_agent from langchain.agents.middleware import wrap_tool_call from langchain.chat_models import init_chat_model load_dotenv() model init_chat_model( modelclaude-sonnet-4-5, model_provideropenai, api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL), ) wrap_tool_call def audit_log(request, handler): print(f [MW] 开始: {request.name}) result handler(request) print(f [MW] 结束: {request.name}) return result agent create_deep_agent( modelmodel, middleware[audit_log], ) result agent.invoke({ messages: [{role: user, content: 在当前目录创建一个 hello.txt内容写 middleware works}] }) print( 最终消息 ) print(result[messages][-1].content)运行python verify_mw.py你会看到类似下面的输出。注意 [MW]开头的行就是中间件打印的它们夹在 Agent 的工具调用之间说明钩子确实在每次工具执行前后触发了。 [MW] 开始: ls [MW] 结束: ls [MW] 开始: write_file [MW] 结束: write_file 最终消息 已创建 hello.txt内容为 middleware works。如果你看到 [MW]只出现一次或者完全没出现通常是两个原因一是middleware参数没传对二是子智能体在执行工具而你没给子智能体挂中间件。验证的时候先用主智能体直接调工具排除子智能体的干扰。再验证一下before_agent钩子的执行顺序。把TurnCounterMiddleware加进去连续调用两次agent.invoke观察轮次计数是否递增。这里要注意graph state 是按 thread 隔离的如果你没配 checkpointer每次 invoke 都是新 thread计数会从 1 重新开始。想看到递增得配一个InMemorySaver。from langgraph.checkpoint.memory import InMemorySaver agent create_deep_agent( modelmodel, middleware[audit_log, TurnCounterMiddleware()], checkpointerInMemorySaver(), ) config {configurable: {thread_id: test-1}} agent.invoke({messages: [{role: user, content: 你好}]}, config) agent.invoke({messages: [{role: user, content: 再来一次}]}, config)两次调用应该分别打印「第 1 轮」和「第 2 轮」。如果第二次还是「第 1 轮」检查 checkpointer 是否真的传进去了以及thread_id是否一致。这个验证能帮你确认before_agent钩子和 graph state 的配合是正常的。5. 本篇常见错排查401、local proxy failed 与钩子不触发中间件跑起来之后报错基本集中在两类模型通道问题和钩子配置问题。我把实际遇到过的几个整理出来对照着排查能省不少时间。401 Unauthorized / invalid api key。这个几乎都是 Key 或 Base URL 配错。先确认TAOTOKEN_API_KEY环境变量真的被读到了在脚本里加一行print(os.getenv(TAOTOKEN_API_KEY)[:8])看前缀。然后确认base_url是https://taotoken.net/api不要多加/v1或漏掉/api。如果你用的是init_chat_modelmodel_provider必须是openai写成别的会走错协议。还有一种情况是 Key 复制时带了空格或换行用strip()处理一下。local proxy failed / connection refused。这个报错说明请求根本没发出去通常是本地网络配置问题。检查你的HTTP_PROXY/HTTPS_PROXY环境变量是否指向了一个不可用的地址临时unset掉再试。另外确认base_url没有写成localhost或内网地址。如果公司网络有出口限制换一个网络环境验证。reading choices / KeyError: choices。这个报错说明返回的 JSON 结构里没有choices字段通常是 endpoint 返回了错误页或非 OpenAI 格式的响应。先单独用 curl 测一下通道curl -s https://taotoken.net/api/chat/completions \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -H Content-Type: application/json \ -d {model:claude-sonnet-4-5,messages:[{role:user,content:hi}]} | head -c 500如果 curl 返回正常但 Python 报错检查model字段的 Model ID 是否和 TaoToken 控制台里的一致大小写和连字符都要对上。OAuth / authentication_error。如果你之前用过 Claude Code 或 Codex 的 OAuth 登录环境里可能残留了ANTHROPIC_API_KEY或OPENAI_API_KEYLangChain 会优先读这些。在脚本开头显式覆盖或者用init_chat_model时明确传api_key参数避免被环境变量劫持。中间件钩子不触发。这个最隐蔽。先确认middleware参数真的传了打印一下agent.middleware看列表长度。然后确认你调的工具是主智能体在执行——如果任务被派给了子智能体而子智能体没挂中间件钩子就不会触发。最后检查wrap_tool_call装饰的函数签名必须是(request, handler)两个参数返回handler(request)的结果少一个参数或忘了 return 都会静默失效。HITL 暂停后无法 resume。HumanInTheLoopMiddleware或permissions modeinterrupt触发暂停后必须配 checkpointer 才能恢复。没配 checkpointer 的话暂停状态无处存储resume 会报找不到 thread。用InMemorySaver做本地测试生产环境换成持久化的 saver。6. 把中间件用到真实链路从日志到限流验证通过之后中间件真正的价值在于组合使用。我现在的项目里主智能体挂了三个中间件一个wrap_tool_call做审计日志一个AgentMiddleware子类做轮次计数和成本累计一个 LangChain 预置的ToolCallLimitMiddleware做工具调用次数上限。这三个各管一摊互不干扰加新逻辑只需要往middleware列表里追加。成本累计这个中间件值得展开说。它在wrap_model_call里读 state 里累计的 token 数超过阈值就打印告警。注意wrap_model_call能改 prompt 和工具列表但改 state 要通过返回值。下面是一个简化版from langchain.agents.middleware import AgentMiddleware class CostGuardMiddleware(AgentMiddleware): 累计 token 消耗超过阈值打印告警。 def __init__(self, threshold100000): self.threshold threshold def wrap_model_call(self, request, handler): response handler(request) usage getattr(response, usage_metadata, None) if usage: total usage.get(total_tokens, 0) print(f[MW] 本次消耗 {total} tokens) return response限流用 LangChain 预置的ToolCallLimitMiddleware更省事直接传max_calls参数。它会在工具调用次数达到上限后拦截后续调用返回一个提示信息而不是真的执行。这个中间件默认不安装需要手动加到middleware列表里。from langchain.agents.middleware import ToolCallLimitMiddleware agent create_deep_agent( modelmodel, middleware[ audit_log, TurnCounterMiddleware(), ToolCallLimitMiddleware(max_calls20), ], )这里有个坑要避开SummarizationMiddleware和TodoListMiddleware默认已经装了你再手动加一遍会冲突。判断方法是看create_deep_agent的参数——skills、memory、subagents、permissions、interrupt_on这些参数会自动挂载对应的中间件你不需要也不应该再手动加。只有 LangChain 预置的那批ModelCallLimitMiddleware、ToolRetryMiddleware、PIIMiddleware等才需要手动追加。最后说一下子智能体的中间件策略。我的做法是主智能体挂全局审计和成本统计子智能体只挂它自己需要的严格审计。比如coder子智能体要写文件就给它单独挂一个记录文件写入的中间件researcher子智能体只读不写就不挂。这样日志不会太吵每个子智能体的行为边界也清晰。整套配置跑下来你得到的是一个可观测、可限流、可审计的 DeepAgents 链路模型调用统一走 TaoToken 的 Key/API 通道中间件在关键节点自动介入。后续要加新逻辑往middleware列表里追加一个函数或类就行不用动 Agent 的核心循环。