LangChain - Middleware

 

Agent 的核心循环就两步：调模型、调工具，来回转。

Middleware（中间件）就是在这个循环的关键节点上插入的"拦截器"——请求进来前、模型调用前后、工具调用前后，都可以挂上自己的逻辑，而不用改 Agent 或工具本身的代码。

熟悉 Web 开发的话可以直接类比：它和 Express/Koa 的中间件是同一个思想，只不过拦截的不是 HTTP 请求，而是 Agent 的执行流程。

上图是一次 Agent 调用的完整生命周期：请求进来先过 before_agent，然后进入"模型 ↔ 工具"的循环（每次模型调用被 before_model / after_model 夹住，模型调用和工具调用本身还能被 wrap_model_call / wrap_tool_call 包裹），循环结束后走 after_agent 返回结果。

中间件能干什么

凡是"我想在模型/工具调用的前后做点事"的需求，都是中间件的活，大致四类：

类别	干的事	典型例子
监视	跟踪 Agent 行为	日志记录、分析、调试
调整	改造进出模型的内容	转换提示词、筛选工具、调整输出格式
控制	干预执行流程	重试、回退、提前终止
防护	加安全和成本约束	速率限制、护栏、PII 检测

内置中间件

常见需求 LangChain 已经内置好了，拿来即用：

中间件	作用
Summarization 摘要	接近 Token 上限时，自动把对话历史压缩成摘要
Human-in-the-loop 人工审核	暂停执行，等人工批准工具调用后再继续
Model call limit 模型调用限制	限制模型调用次数，防止成本失控
Tool call limit 工具调用限制	限制工具调用次数
Model fallback 模型回退	主模型挂了自动切换到备用模型
PII detection PII 检测	检测并处理个人身份信息
To-do list 待办清单	给 Agent 加任务规划和跟踪能力
LLM tool selector 工具选择器	工具太多时，先用一个 LLM 筛出相关工具再调主模型
Tool retry 工具重试	工具调用失败后按指数退避自动重试
LLM tool emulator 工具模拟器	用 LLM 模拟工具的执行结果，方便测试
Context editing 上下文编辑	裁剪或清除历史中的工具调用记录，给上下文瘦身
Shell tool Shell 工具	给 Agent 一个持久的 shell 会话执行命令
File search 文件搜索	提供 Glob / Grep 文件搜索工具

自定义中间件

内置的不够用时就自己写。

两类钩子

Node 风格（节点式）：在执行流程的固定节点上按顺序运行，做完自己的事就放行。

• before_agent —— Agent 启动前（每次调用只跑一次）
• before_model —— 每次模型调用前
• after_model —— 每次模型响应后
• after_agent —— Agent 结束后（每次调用只跑一次）

Wrap 风格（包裹式）：把模型/工具调用整个包起来，拿到一个 handler，你决定调不调、调几次、怎么处理结果——重试、缓存、改写请求响应都靠它。

• wrap_model_call —— 包裹每次模型调用
• wrap_tool_call —— 包裹每次工具调用

两者的区别：Node 风格是"到点执行"，Wrap 风格是"全权接管"。

两种写法

写法	适合
装饰器	只需要一个钩子的简单场景，一个函数搞定
继承类	需要多个钩子配合、或要维护状态的复杂中间件

装饰器写法，给模型调用前加个日志：

from langchain.agents.middleware import before_model
from langchain.agents import create_agent

@before_model
def log_before_model(state, runtime):
    print(f"即将调用模型,当前消息数: {len(state['messages'])}")
    return None  # 返回 None 表示不修改状态,直接放行

agent = create_agent(
    model="gpt-4o",
    tools=[...],
    middleware=[log_before_model],
)

类写法，多个钩子组合：

from langchain.agents.middleware import AgentMiddleware

class MyMiddleware(AgentMiddleware):
    def before_model(self, state, runtime):
        ...  # 模型调用前

    def wrap_tool_call(self, request, handler):
        return handler(request)  # 可在前后加逻辑,或干脆不调 handler

写好后统一通过 create_agent(middleware=[...]) 挂载，多个中间件按列表顺序依次执行。

案例

源码仓库

自定义脱敏中间件案例，用于把邮箱和手机号进行脱敏处理