LangChain Middleware 技术解析：从“插槽机制”到 Agent 运行时控制

根据 LangChain 官方文档，Middleware是 LangChain agent 运行时里的一个“拦截层 / 扩展层”，用来在 agent 执行的各个阶段插入控制逻辑。官方给它的定位很明确：它让你可以更精细地控制 agent 内部发生的事情，比如日志追踪、prompt 改写、工具选择、输出格式、重试、fallback、限流、guardrails，以及 PII 检测。官方还强调，Middleware 是create_agent的核心特性之一，也是做context engineering的关键机制。(LangChain 文档)

从执行模型看，LangChain 的 agent loop 本质上是：调用模型 → 模型决定是否调用工具 → 执行工具 → 再回到模型，直到结束。Middleware 就暴露在这些关键节点前后，因此你可以在调用模型前改写上下文、在模型返回后检查输出、在工具调用前后插入审批或重试逻辑，甚至直接改变执行流。官方文档还提到，middleware 不只是“看一眼”，它还能更新上下文，以及跳转到 agent 生命周期中的其他步骤。(LangChain 文档)

官方把 Middleware 分成两大类：

1. Node-style hooks

这类 hook 在固定执行点顺序运行，适合做日志、校验、状态更新。官方列出的节点有：

• before_agent：agent 整体开始前，只执行一次
• before_model：每次调用模型前
• after_model：每次模型返回后
• after_agent：agent 完成后，只执行一次 (LangChain 文档)

2. Wrap-style hooks

这类 hook 是“包裹式”的，直接围绕模型调用或工具调用执行，控制力更强。官方说明它适合做retry、cache、transformation，并且你可以决定底层 handler 被调用0 次、1 次或多次，也就是可以做 short-circuit、正常放行或者重试逻辑。对应两个 hook：

• wrap_model_call
• wrap_tool_call(LangChain 文档)

这两类 hook 的差别，可以这么理解：

• Node-style更像“在某个固定生命周期节点插一段逻辑”
• Wrap-style更像“把模型/工具调用整个包起来接管”

所以，像“打印日志”“做输入校验”“根据 state 改一些字段”，更适合 node-style；像“失败自动重试”“缓存结果”“替换模型调用策略”“拦截工具异常”，更适合 wrap-style。这个区分在官方文档里说得很清楚。(LangChain 文档)

Middleware 能解决什么问题

官方总览页和内置中间件页给出的典型用途主要有这几类：

• 可观测性：logging、analytics、debugging
• 上下文改造：改 prompt、改 tool selection、改 output formatting
• 鲁棒性：retry、fallback、early termination
• 安全与治理：rate limits、guardrails、PII detection (LangChain 文档)

如果用更工程化的语言说，Middleware 适合处理那些横切关注点（cross-cutting concerns）：这些逻辑不是某一个 tool 本身的业务职责，但又需要贯穿 agent 的多个阶段，比如成本控制、敏感信息处理、人工审批、上下文压缩等。(LangChain 文档)

官方内置了哪些 Middleware

LangChain 官方提供了一批预置 middleware，常见的包括：

• SummarizationMiddleware：上下文接近 token 限制时自动总结历史消息
• HumanInTheLoopMiddleware：敏感工具调用前暂停，等待人工批准
• ModelCallLimit：限制模型调用次数，防止成本失控
• ToolCallLimit：限制工具调用次数
• ModelFallback：主模型失败时自动切换备用模型
• PII detection / PIIMiddleware：检测和处理敏感个人信息
• To-do list：给 agent 增加任务规划和跟踪能力
• LLM tool selector：先用一个 LLM 选工具，再调用主模型
• Tool retry/Model retry：工具或模型失败时做指数退避重试
• LLM tool emulator：用 LLM 模拟工具执行，便于测试
• Context editing：裁剪或清理上下文里的工具使用痕迹 (LangChain 文档)

这意味着在很多实际项目里，你未必需要一开始就自己写中间件。很多常见需求，官方已经给了现成实现。(LangChain 文档)

怎么挂到 agent 上

官方推荐的方式很直接：在create_agent(...)时，把 middleware 列表传进去。示例写法如下：

fromlangchain.agentsimportcreate_agentfromlangchain.agents.middlewareimportSummarizationMiddleware,HumanInTheLoopMiddleware agent=create_agent(model="gpt-4.1",tools=[...],middleware=[SummarizationMiddleware(...),HumanInTheLoopMiddleware(...)],)

这说明 middleware 是 agent runtime 的一等配置项，不是额外挂在外面的 hack。(LangChain 文档)

自定义 Middleware 怎么写

官方文档给了两种方式：

1. 装饰器方式

适合快速写轻量逻辑，例如在before_model做检查、在after_model做日志记录。官方例子里甚至演示了：当消息数超过限制时，在before_model里直接返回一条 AIMessage，并通过jump_to: "end"提前结束执行。(LangChain 文档)

2. 类方式

继承AgentMiddleware，把逻辑写成类方法，适合参数化、更复杂、可复用的中间件。官方示例里的MessageLimitMiddleware就是这个思路：构造函数接收max_messages，然后在before_model中检查长度，超限就结束。(LangChain 文档)

这两个接口说明了 LangChain Middleware 设计上的一个重点：它不是只能“观察”，而是能参与决策。你可以在 hook 里返回状态更新，甚至改变执行路径。(LangChain 文档)

State 更新机制

官方专门说明了 middleware 如何更新 agent state：

• Node-style hooks：直接返回一个dict，这个字典会通过 graph reducer 合并进 agent state
• Wrap-style hooks：
  • model call 里返回ExtendedModelResponse，并通过Command注入状态更新
  • tool call 里直接返回Command(LangChain 文档)

这背后其实和 LangChain v1 基于LangGraph的 agent runtime 有关。也就是说，middleware 并不是一个简单的 callback 系统，而是和图执行、状态流转、生命周期跳转深度绑定的。(LangChain 文档)

适合在什么场景用

结合官方文档，比较典型的落地场景有：

1. 成本与稳定性控制

   给模型/工具加调用次数限制、失败重试、fallback。(LangChain 文档)

2. 安全治理

   对输入做 PII redact / block，对高风险工具调用做人审。(LangChain 文档)

3. 上下文治理

   对超长对话做 summarization，按 state 动态裁剪消息或调整 system prompt。(LangChain 文档)

4. 工具编排优化

   动态筛选 tools、限制工具暴露范围、在工具调用前后做监控或补偿逻辑。(LangChain 文档)

LangChain 官方的 Middleware的插槽机制

官方提供的不是一个统一的大型 middleware pipeline，而是一组预定义的可插入 hook / wrap 点。你把自己的函数或类方法挂到这些点上，agent 运行到那里时就会执行你的逻辑。create_agent本身构建的是一个基于LangGraph的图式 runtime，agent 在 model 节点、tools 节点和 middleware 之间流转；middleware 就是在这些生命周期节点上暴露出来的扩展接口。(LangChain 文档)

LangChain 官方先定义好 agent 生命周期中的若干插槽点；开发者把自定义逻辑注册到这些插槽上；运行时按固定顺序调度这些插槽。

1）官方到底提供了哪些“插槽”

官方把插槽分成两类。

A. Node-style hooks：固定节点插槽

这类插槽是在确定的生命周期节点上顺序执行的，官方列出的 4 个点是：

• before_agent：agent 启动前，整次调用只执行一次
• before_model：每次调用模型前
• after_model：每次模型返回后
• after_agent：agent 完成后，整次调用只执行一次 (LangChain 文档)

这类最像你说的“插槽”：

运行到这里，就把控制权临时交给你。

A. Node-style hooks 例子：在每次调用模型前，检查消息长度

• 这个例子演示before_model。

  fromlangchain.agentsimportcreate_agentfromlangchain.agents.middlewareimportbefore_modelfromlangchain.messagesimportAIMessage# 这是一个 Node-style hook：固定在“调用模型前”执行@before_modeldefmessage_limit_guard(state,runtime):# state["messages"] 是当前对话消息iflen(state["messages"])>10:return{"messages":[AIMessage(content="消息太多了，我先停止这次执行。")],"jump_to":"end",# 直接结束 agent}# 返回 None 表示不拦截，正常继续returnNoneagent=create_agent(model="openai:gpt-4.1-mini",tools=[],middleware=[message_limit_guard],)result=agent.invoke({"messages":[{"role":"user","content":"你好，帮我总结一下今天的工作安排"}]})print(result)

效果是：

• 每次 agent 准备调用模型前
• 先进入这个“固定节点插槽”
• 如果消息太多，就直接结束，不再继续往下跑

官方文档里把before_model归为 node-style hook，并给过类似“消息超限就结束”的示例。(LangChain 文档)

这个例子怎么理解

这里的before_model就是一个固定插槽：

Agent 运行 -> before_model -> 真正调用模型 -> after_model

你的逻辑只是在“调用模型前”这个固定点执行一次。

B. Wrap-style hooks：包裹式插槽

这类不是“到了某点执行一下”，而是把一次 model/tool 调用整个包起来。官方给了两个：

• wrap_model_call
• wrap_tool_call(LangChain 文档)

它的语义更像：

“这里有一个标准调用过程，你可以在外面套一层壳，决定是否放行、修改请求、重试、替换模型、捕获异常、改返回值。”

所以从实现风格看：

• Node-style =事件点插槽
• Wrap-style =调用链包裹插槽(LangChain 文档)

B. Wrap-style hooks 例子：给工具调用加统一异常处理

• 这个例子演示wrap_tool_call。

  fromlangchain.agentsimportcreate_agentfromlangchain.agents.middlewareimportwrap_tool_callfromlangchain.toolsimporttoolfromlangchain.messagesimportToolMessage@tooldefdivide(a:float,b:float)->float:"""计算 a / b"""returna/b# 这是一个 Wrap-style hook：包裹整个工具调用过程@wrap_tool_calldefhandle_tool_error(request,handler):try:# 真正执行工具调用returnhandler(request)exceptExceptionase:# 如果工具报错，返回一个友好的 ToolMessage 给模型returnToolMessage(content=f"工具执行失败：{str(e)}。请检查参数后再试。",tool_call_id=request.tool_call["id"],)agent=create_agent(model="openai:gpt-4.1-mini",tools=[divide],middleware=[handle_tool_error],)result=agent.invoke({"messages":[{"role":"user","content":"请用 divide 工具计算 10 / 0"}]})print(result)

效果是：

• agent 调工具时
• 不直接调用工具
• 而是先进入你的 wrapper
• wrapper 再决定是否调用原工具、怎么处理异常、要不要改返回值

官方文档把wrap_tool_call定义为围绕每次工具调用执行，适合做错误处理、重试、缓存这类逻辑；官方在 agents 文档里还给了一个工具报错时返回ToolMessage的示例。(LangChain 文档)

这个例子怎么理解

这里的wrap_tool_call不是“在工具调用前做一下检查”那么简单，

而是：

Agent 要调工具 -> 先进入你的 wrapper -> wrapper 内部决定是否调用 handler(request) -> 也可以 try/except -> 也可以重试 -> 也可以直接返回，不调用 handler

所以它更像“给工具调用外面套了一层壳”。

两段代码的本质区别

Node-style：固定节点执行

像这样：

@before_modeldefxxx(state,runtime):...

特点是：

• 运行时机固定
• 到点就执行
• 常用于校验、日志、状态更新、消息裁剪

官方把这类 hook 定义为在特定执行点顺序运行。(LangChain 文档)

Wrap-style：包住一次调用

像这样：

@wrap_tool_calldefxxx(request,handler):...returnhandler(request)

特点是：

• 你拿到的是“被包裹的调用”
• 你能决定调不调handler
• 还能调一次、多次，或者一次都不调
• 常用于重试、缓存、统一异常处理

官方明确把它描述为“around each model/tool call”，并指出 handler 可以被调用 0 次、1 次或多次。(LangChain 文档)

2）它的“具体实现思路”可以怎么理解

你可以把官方实现抽象成下面这个伪代码：

defrun_agent(state):# 1. before_agent slotsformwinmiddleware:state=mw.before_agent(state)orstatewhilenotfinished(state):# 2. before_model slotsformwinmiddleware:state=mw.before_model(state)orstate# 3. wrap_model_call chainresponse=call_model_with_wrappers(middleware,state)# 4. after_model slotsstate=apply_model_response(state,response)formwinreversed(middleware):state=mw.after_model(state)orstate# 5. 如果模型要求调工具ifneed_tool_call(state):result=call_tool_with_wrappers(middleware,state)state=apply_tool_result(state,result)# 6. after_agent slotsformwinreversed(middleware):state=mw.after_agent(state)orstatereturnstate

这和官方文档给出的执行顺序是对齐的：

• before_*按 middleware 列表顺序执行
• wrap_*像函数嵌套一样包起来
• after_*逆序执行 (LangChain 文档)

官方示例明确写了顺序：

1. middleware1.before_agent()

2. middleware2.before_agent()

3. middleware3.before_agent()

4. middleware1.before_model()

5. middleware2.before_model()

6. middleware3.before_model()

7. middleware1.wrap_model_call()→middleware2.wrap_model_call()→middleware3.wrap_model_call()→ model

8. middleware3.after_model()

9. middleware2.after_model()

10. middleware1.after_model()

    ……最后after_agent()也是逆序。(LangChain 文档)

这就是“插槽机制”的调度器实现本质。