数据截至 2026-08-18(上游 commit 2019bf5ebe50)

第 2 章:中间件系统

这是 LangChain v1 最有辨识度的设计。本章讲:中间件是什么、六个钩子分别在循环的哪个工位、以及 wrap_* 那种"洋葱式"包装为什么能做重试和短路。

2.1 要解决的小问题

agent 循环本身很死板:问模型→调工具→再问。但真实需求千奇百怪:

历史太长了,先压缩再发给模型;
工具是"删库"这类危险操作,先让人审批;
模型偶尔超时,自动重试或换个模型;
限制最多调 10 次模型,防止失控烧钱。

如果每个需求都去改 create_agent 的内部,这函数会爆炸。LangChain 的答案:把所有这些抽象成 middleware——一个你可以 create_agent(..., middleware=[A, B, C]) 传进去的列表,每个中间件挑自己关心的"工位"挂钩子。

2.2 六个工位:钩子挂在循环的哪里

AgentMiddleware 基类定义了一组可覆写的钩子(middleware/types.py:385-662)。按它们在循环里的位置:

  START
    │
    ▼  before_agent      ← 整个 agent 跑之前(只一次)
    │
    ▼  before_model      ← 每次问模型之前(可多次,循环里)
    │
  ┌─────────────────────────────────┐
  │  wrap_model_call  (洋葱包住下面) │
  │        model.invoke()           │
  └─────────────────────────────────┘
    │
    ▼  after_model       ← 每次问完模型之后
    │
  ┌─────────────────────────────────┐
  │  wrap_tool_call  (洋葱包住下面)  │
  │        tool.run()               │
  └─────────────────────────────────┘
    │
    ▼  after_agent       ← 整个 agent 收工之后(只一次)
    ▼
   END

钩子	时机	典型用途
`before_agent` / `after_agent`	agent 启停时各一次	初始化、收尾、记账
`before_model` / `after_model`	每轮模型调用前后	压缩历史、注入提示、人工审批
`wrap_model_call`	包住模型调用	重试、模型 fallback、改请求/响应
`wrap_tool_call`	包住工具调用	工具重试、危险操作拦截、模拟工具

两类钩子的本质区别:

节点式钩子(before_* / after_*):返回一个 state 更新 dict,被装配成图里的一个真实节点(回顾第 1 章 factory.py:1549-1645 那批 graph.add_node(f"{m.name}.before_model", ...))。
洋葱式钩子(wrap_*):不是节点,而是把内层调用当参数 handler 收进来,自己决定调几次、调不调。

2.3 节点式钩子:before_model 实战(压缩历史)

SummarizationMiddleware 是 before_model 的代表(middleware/summarization.py:398-436):

def before_model(self, state, runtime) -> dict[str, Any] | None:
    messages = state["messages"]
    total_tokens = self.token_counter(messages)
    if not self._should_summarize(messages, total_tokens):
        return None                       # token 没超标 → 啥也不做
    cutoff_index = self._determine_cutoff_index(messages)
    messages_to_summarize, preserved = self._partition_messages(messages, cutoff_index)
    summary = self._create_summary(messages_to_summarize)
    return {
        "messages": [
            RemoveMessage(id=REMOVE_ALL_MESSAGES),   # 先清空
            *self._build_new_messages(summary),       # 放一条"摘要"
            *preserved,                               # 保留最近几条
        ]
    }

要点:

返回 None = 不改 state,这是"本轮不触发"的惯用法(summarization.py:418)。
它返回的 messages 以 RemoveMessage(REMOVE_ALL_MESSAGES) 开头——配合 add_messages reducer,这是"先全删再重建"的标准手法,把超长历史换成"一条摘要 + 最近 N 条"。

2.4 节点式钩子:after_model + interrupt(人工审批)

HumanInTheLoopMiddleware 用 after_model 在工具执行之前拦下来,弹给人审批(middleware/human_in_the_loop.py:399-450)。核心是它调用了 LangGraph 的 interrupt(...):

decisions = interrupt(hitl_request)["decisions"]   # human_in_the_loop.py:435

interrupt 会暂停整张图(配合 checkpointer 存档),把请求抛给外部;人给出 approve / edit / reject / respond 后,图从断点恢复继续跑。每种决定怎么改写工具调用见 _process_decision(human_in_the_loop.py:314 起):reject 会塞一条 status="error" 的 ToolMessage 告诉模型"用户拒绝了"。

这就是"agent = 图"带来的红利:人工审批不用你自己存状态、轮询、恢复,interrupt + checkpointer 直接给你。

2.5 洋葱式钩子:wrap_model_call 为什么能重试

节点式钩子只能"前后插话",没法"重来一次模型调用"。重试需要把模型调用握在手里——这就是 wrap_model_call 的设计:它收到一个 handler(代表"真正执行模型"),自己决定调几次(middleware/types.py:503-529)。

# 示意,非源码:一个重试中间件的 wrap_model_call
def wrap_model_call(self, request, handler):
    for attempt in range(3):
        try:
            return handler(request)        # 调一次模型
        except RateLimitError:
            continue                        # 失败就再调
    raise

想短路?不调 handler,直接返回一个 AIMessage。
想改请求?handler(request.override(model=other_model))——这正是 fallback 的玩法。
想改响应?拿到 handler 的返回再加工。

返回值可以是 ModelResponse / AIMessage / ExtendedModelResponse 三选一(middleware/types.py:315-325),框架统一归一化(factory.py:_normalize_to_model_response)。

2.6 多个洋葱怎么套:组合顺序

传 middleware=[A, B] 且两者都有 wrap_model_call 时,要定义"谁包谁"。规则:列表里靠前 = 最外层(factory.py:263-275,_chain_model_call_handlers docstring:"first in list becomes outermost layer")。

请求进 →  A.wrap (最外)
              └─ B.wrap
                    └─ 真正的 model.invoke()
              ┌─ B 拿到响应
响应出 ←  A 拿到响应(最后处理)

实现是经典的"两两折叠"(factory.py:313-350):compose_two(outer, inner) 把两个 handler 拼成一个,然后从右往左 reduce。每层还把各自产生的 Command 累积进列表(inner-first),保证状态更新顺序确定(factory.py:324-343)。

wrap_tool_call 同理,也是"first = outermost"的洋葱(factory.py:658 起 _chain_tool_call_wrappers)。

2.7 内置中间件目录

v1 自带 20+ 个中间件(middleware/__init__.py:6-30),覆盖常见需求:

中间件	主钩子	干什么
`SummarizationMiddleware`	`before_model`	token 超标时压缩历史
`HumanInTheLoopMiddleware`	`after_model` + `interrupt`	工具调用前人工审批
`ModelCallLimitMiddleware`	`before_model`	限制模型调用次数
`ToolCallLimitMiddleware`	—	限制工具调用次数
`ModelFallbackMiddleware`	`wrap_model_call`	主模型失败时换备用模型
`ModelRetryMiddleware` / `ToolRetryMiddleware`	`wrap_*`	出错重试
`PIIMiddleware`	—	检测/脱敏个人信息
`ContextEditingMiddleware`	—	清理过期工具调用上下文
`TodoListMiddleware`	—	给 agent 一个待办清单工具
`LLMToolSelectorMiddleware`	—	工具太多时先用 LLM 选子集
`LLMToolEmulator`	`wrap_tool_call`	测试时用 LLM 假装执行工具
`ShellToolMiddleware`	—	带沙箱策略的 shell 工具

（完整清单见 middleware/__init__.py:53-100 的 __all__。）

你也可以不写类,用 @before_model / @wrap_model_call 等装饰器把一个函数直接变成中间件(middleware/types.py:934-1008),适合一次性的小逻辑。

2.8 巧妙之处

"前后插话"和"包起来"分成两套机制,不混。节点式适合"改 state",洋葱式适合"控制要不要/重复执行"——重试这种需求节点式根本表达不了(middleware/types.py 的钩子签名差异)。
只有被覆写的钩子才装配成节点(factory.py:1576:比较 m.__class__.before_model is not AgentMiddleware.before_model)。没覆写就不进图,图保持精简。
request.override(...) 返回新对象而非原地改,让重试/fallback 里反复改请求是安全的(middleware/types.py 的 _ModelRequestOverrides)。

2.9 边界与局限

wrap_model_call 返回的 Command 不支持 goto/resume/graph,用了会抛 NotImplementedError(middleware/types.py:323-324)。
中间件改了 request.tools 加进新工具,但这些工具没在 create_agent(tools=...) 注册过,会报"Unknown tools"——除非你也实现 wrap_tool_call 自己执行(factory.py:119-143 的 DYNAMIC_TOOL_ERROR_TEMPLATE)。

2.10 代码地图

主题	文件路径	关键符号
中间件基类 + 六钩子	`libs/langchain_v1/langchain/agents/middleware/types.py`	`AgentMiddleware`
模型请求/响应对象	`libs/langchain_v1/langchain/agents/middleware/types.py`	`ModelRequest` / `ModelResponse` / `ExtendedModelResponse`
装饰器式中间件	`libs/langchain_v1/langchain/agents/middleware/types.py`	`before_model` / `wrap_model_call` / `hook_config`
wrap_model_call 组合	`libs/langchain_v1/langchain/agents/factory.py`	`_chain_model_call_handlers`
wrap_tool_call 组合	`libs/langchain_v1/langchain/agents/factory.py`	`_chain_tool_call_wrappers`
钩子→图节点装配	`libs/langchain_v1/langchain/agents/factory.py`	`graph.add_node` (1508-1590)
压缩历史	`libs/langchain_v1/langchain/agents/middleware/summarization.py`	`SummarizationMiddleware.before_model`
人工审批	`libs/langchain_v1/langchain/agents/middleware/human_in_the_loop.py`	`HumanInTheLoopMiddleware.after_model`
中间件目录	`libs/langchain_v1/langchain/agents/middleware/__init__.py`	`__all__`

2.1 要解决的小问题​

2.2 六个工位:钩子挂在循环的哪里​

2.3 节点式钩子:before_model 实战(压缩历史)​

2.4 节点式钩子:after_model + interrupt(人工审批)​

2.5 洋葱式钩子:wrap_model_call 为什么能重试​

2.6 多个洋葱怎么套:组合顺序​

2.7 内置中间件目录​

2.8 巧妙之处​

2.9 边界与局限​

2.10 代码地图​