代码仓库ChainReaction

中间件这个词已经被用滥了。Web 框架拿它做鉴权，Django 拿它记请求日志。放到 agent 上，意思没变：在不改核心循环的前提下，在固定位置插一段自己的代码。

create_agent 返回的是一张编译好的 LangGraph 图，循环逻辑写死在图里。你想加一句”每次调用模型前打印当前消息数”，有两个选择。自己用 StateGraph 重画一遍循环，或者把这段逻辑做成中间件塞进 middleware=[...]。前者的代价是官方每次调整循环你都得跟着改，后者只多一个参数。

这篇文章把 middleware 这个参数讲清楚。钩子挂在哪些位置，装饰器和类两种写法各适合什么场景，状态怎么在钩子之间传，内置中间件有哪些能直接拿来用，多个中间件叠在一起时谁先谁后。最后有两个完整例子：按对话长度动态换模型，以及工具抛异常之后怎么让 agent 活下来。

所有代码都在 DeepSeek 上跑过，输出是从终端直接复制的，脚本放在 Middleware/。

中间件挂在循环的哪一步

先看循环本身。模型读一遍消息，决定要不要调工具。要调就执行工具，把结果作为 ToolMessage 塞回消息列表，再问一次模型。直到某次回复里没有 tool_calls，循环结束。

中间件做的事，是在这个循环的每个转折点放一个钩子。

六个钩子里，四个是节点式的，位置固定，按顺序执行。两个是包裹式的，包住一次模型调用或一次工具调用。

有件事容易忽略：中间件不是独立的运行时。钩子就跑在 create_agent 编译出来的那张图里。你可以把整个 agent 当成节点或子图塞进更大的 StateGraph，HITL 中断、摘要、PII 脱敏、重试这些行为会跟着 agent 一起走，不用重新配。

最小用法就是往 middleware 里塞实例：

1
2
3
4
5
6
7
8
9
10
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
    model=model,
    tools=[search],
    middleware=[
        SummarizationMiddleware(model=model, trigger=("tokens", 4000), keep=("messages", 20)),
    ],
)

中间件能插手的地方，官方文档列了四类：追踪 agent 行为（日志、埋点、调试），改写提示词和工具选择，加重试、降级和提前终止，加限流、护栏和 PII 检测。前两类是看和改，后两类是控。

六个钩子，两种脾气

钩子 类型 触发时机
before_agent 节点式 agent 开始前，一次调用执行一次
before_model 节点式 每次模型调用前
after_model 节点式 每次模型回复后
after_agent 节点式 agent 结束后，一次调用执行一次
wrap_model_call 包裹式 包住每次模型调用
wrap_tool_call 包裹式 包住每次工具调用

节点式钩子的签名是 (state, runtime)，返回一个 dict 或 None。包裹式钩子的签名是 (request, handler)，拿到请求，自己决定 handler 什么时候调、调几次。

差别就落在”调几次”上。节点式钩子拦不住模型调用，想提前退出只能返回 jump_to，这一点后面单独说。wrap_model_call 可以一次都不调 handler，直接返回一个自己造的 ModelResponse；也可以循环调三次，这就是重试中间件的实现方式。

钩子签名里的 runtime 是本次调用的运行时对象，runtime.context 里放着通过 context_schema 声明的每轮数据，比如 user_id、feature flag。包裹式钩子从 request.runtime 拿到的是同一个对象。要做”按用户等级决定能不能调这个工具”这类逻辑，读它就够了。

状态更新的路径也不一样：

  • 节点式钩子返回 dict，dict 按图的 reducer 合并进 state
  • wrap_model_call 要注入状态，得返回 ExtendedModelResponse(model_response=..., command=Command(update={...}))
  • wrap_tool_call 直接返回 Command

多个中间件都返回 ExtendedModelResponse 时，command 会依次应用。非 reducer 字段是内层先写、外层后写，键冲突时外层赢。外层如果做了重试、handler 被调了多次，前面几次的 command 会被丢掉，不会污染最终状态。

装饰器还是类

两种写法，先各来一份。

装饰器式，一个函数就是一个钩子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from typing import Any
from langchain.agents.middleware import AgentState, before_model, after_model
from langgraph.runtime import Runtime


@before_model
def log_before_model(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    print(f"即将调用模型，当前消息数 {len(state['messages'])}")
    return None


@after_model
def log_after_model(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    print(f"模型返回：{state['messages'][-1].content}")
    return None

类式，一个类可以挂多个钩子，还能带构造参数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from typing import Any
from langchain.agents.middleware import (
    AgentMiddleware, AgentState, ModelRequest, ModelResponse,
)
from langgraph.runtime import Runtime


class LoggingMiddleware(AgentMiddleware):
    def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        print(f"即将调用模型，当前消息数 {len(state['messages'])}")
        return None

    def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        print(f"模型返回：{state['messages'][-1].content}")
        return None

    async def abefore_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        return None
装饰器式 类式
适合 单个钩子、逻辑十行以内、原型阶段 多个钩子、需要构造参数、同步异步都要
写法 @before_model 修饰函数 继承 AgentMiddleware，覆盖方法
声明状态 @before_model(state_schema=MyState) class MyMW(AgentMiddleware[MyState]): state_schema = MyState
声明可跳转 @before_model(can_jump_to=["end"]) 方法上加 @hook_config(can_jump_to=["end"])
附带工具 装饰器传 tools=[...] 类属性 tools = [...]

AgentMiddleware 子类还能声明三个类属性，工厂在编译期会读：state_schema 扩展状态字段，tools 注册随中间件一起提供的工具，transformers 注册流式转换器。TodoListMiddleware 给 agent 加的 write_todos 工具就是通过 tools 注册的。

选哪个没有硬规则。我的习惯是：一个钩子、逻辑短，用装饰器；出现第二个钩子，或者需要构造函数传阈值、传模型，就换类。需要在 invoke 和 ainvoke 两条路径上都跑，也只能用类，因为装饰器生成的中间件没有独立的异步实现。

自定义状态

钩子之间要传数据，靠扩展 agent 的 state。用 state_schema 声明新字段，字段用 NotRequired 包一层，钩子返回 dict 时键名对得上就自动合并。

下面这个脚本给状态加了一个 model_call_count，四个钩子各打印一行，看看计数在什么时候变。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
import sys, os
sys.stdout.reconfigure(encoding="utf-8")

from typing import Any
from typing_extensions import NotRequired

from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain.agents.middleware import (
    AgentState, before_agent, before_model, after_model, after_agent,
)
from langchain.messages import HumanMessage
from langchain.tools import tool
from langgraph.runtime import Runtime


class CustomState(AgentState):
    model_call_count: NotRequired[int]


events: list[str] = []


@before_agent(state_schema=CustomState)
def on_start(state: CustomState, runtime: Runtime) -> dict[str, Any] | None:
    events.append("before_agent   -> 初始化 model_call_count=0")
    return {"model_call_count": 0}


@before_model(state_schema=CustomState)
def before(state: CustomState, runtime: Runtime) -> dict[str, Any] | None:
    n = state.get("model_call_count", 0)
    events.append(f"before_model   -> 第 {n + 1} 次模型调用前，当前消息数 {len(state['messages'])}")
    return None


@after_model(state_schema=CustomState)
def after(state: CustomState, runtime: Runtime) -> dict[str, Any] | None:
    n = state.get("model_call_count", 0)
    last = state["messages"][-1]
    events.append(f"after_model    -> 计数 {n} -> {n + 1}，最后一条是 {type(last).__name__}")
    return {"model_call_count": n + 1}


@after_agent(state_schema=CustomState)
def on_end(state: CustomState, runtime: Runtime) -> dict[str, Any] | None:
    events.append(f"after_agent    -> 本轮共调用模型 {state.get('model_call_count', 0)} 次")
    return None


@tool
def add(a: int, b: int) -> int:
    """计算两个整数之和。"""
    return a + b


model = ChatOpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
    model="deepseek-chat",
    temperature=0.1,
    max_tokens=1000,
)

agent = create_agent(model=model, tools=[add], middleware=[on_start, before, after, on_end])

result = agent.invoke(
    {"messages": [HumanMessage("请调用 add 工具计算 17 + 25，然后告诉我结果。")],
     "model_call_count": 0}
)

真实输出：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
=== 钩子执行轨迹 ===
before_agent   -> 初始化 model_call_count=0
before_model   -> 第 1 次模型调用前，当前消息数 1
after_model    -> 计数 0 -> 1，最后一条是 AIMessage
before_model   -> 第 2 次模型调用前，当前消息数 3
after_model    -> 计数 1 -> 2，最后一条是 AIMessage
after_agent    -> 本轮共调用模型 2 次

=== 最终状态 ===
model_call_count = 2
messages 条数 = 4
  [HumanMessage] 请调用 add 工具计算 17 + 25，然后告诉我结果。
  [AIMessage] I'll calculate that for you.
  [ToolMessage] 42
  [AIMessage] 17 + 25 = **42**

几个细节。after_model 返回 {"model_call_count": n + 1}，下一次 before_model 读到的就是新值，中间不用自己维护外部变量。第一次模型调用前消息数是 1，第二次变成 3，多出来的两条是模型回复和工具结果。计数器停在 2，说明这轮对话模型被调了两次：第一次决定调工具，第二次读着工具结果给出答案。最后 after_agent 拿到的还是完整状态。

状态字段的初始化是可选的。这里传了 model_call_count: 0，不传也行，因为钩子里用的是 state.get(..., 0)。

内置中间件清单

官方预置了一批，都是 provider 无关的，DeepSeek 也能用。下面这张表按用途分类。

名字 解决什么问题 关键参数
SummarizationMiddleware 对话快撑爆上下文时把旧消息压成摘要 model、trigger、keep、token_counter、summary_prompt、trim_tokens_to_summarize
HumanInTheLoopMiddleware 工具执行前暂停，等人审批、修改或拒绝 interrupt_on={工具名: True} 或 {工具名: {"allowed_decisions": [...]}}，需要 checkpointer
PIIMiddleware 检测并处理邮箱、卡号、IP 等敏感信息 pii_type、strategy（block/redact/mask/hash）、detector、apply_to_input、apply_to_output、apply_to_tool_results
ModelRetryMiddleware 模型调用失败时按指数退避重试 max_retries、retry_on、on_failure、backoff_factor、initial_delay、max_delay、jitter
ToolRetryMiddleware 工具调用失败时按指数退避重试 同上，多一个 tools 限定生效范围
ToolErrorMiddleware 把工具异常转成 error ToolMessage，让模型自己纠正 on_error(exc, request)、aon_error、tools
ModelFallbackMiddleware 主模型挂了就按顺序换备用模型 位置参数依次是备用模型，字符串或模型实例都行
ModelCallLimitMiddleware 限制模型调用次数，防死循环和账单 thread_limit、run_limit、exit_behavior
ToolCallLimitMiddleware 限制某个工具或全部工具的调用次数 tool_name、thread_limit、run_limit、exit_behavior
TodoListMiddleware 给 agent 一个 write_todos 工具做任务规划 system_prompt、tool_description
LLMToolSelectorMiddleware 工具太多时先让模型挑出相关的几个 model、max_tools、always_include
ContextEditingMiddleware 上下文超限时清掉旧的工具输出 edits=[ClearToolUsesEdit(trigger=100000, keep=3)]
@dynamic_prompt 每次调用模型前重新生成系统提示 函数返回 str 或 SystemMessage，可读 request.messages、request.runtime.context
@wrap_model_call 换模型 按消息长度、用户等级等条件切换模型 request.override(model=...)

SummarizationMiddleware 的触发条件写法值得单独看一眼。单个元组 ("tokens", 4000) 表示达到阈值就触发；传列表 [("tokens", 3000), ("messages", 6)] 是或的关系；传字典 {"tokens": 4000, "messages": 10} 是与的关系。keep 指定摘要后保留多少上下文，("messages", 20) 是保留最近 20 条。trigger 不传就不会自动触发。

HITL 那个中间件必须配 checkpointer，不然中断之后状态没地方存。

PII 的用法看一眼就明白：

1
2
3
4
5
6
7
8
agent = create_agent(
    model=model,
    tools=[],
    middleware=[
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        PIIMiddleware("credit_card", strategy="mask", apply_to_input=True),
    ],
)

内置类型有 email、credit_card、ip、mac_address、url。要检测自家系统的 API key，传 detector=r"sk-[a-zA-Z0-9]{32}" 或者一个返回 PIIMatch 列表的函数就行。

上面这张表里有一个我到目前都没跑通：LLMToolSelectorMiddleware。加上 Gemini 之后，一共在四条通路上试过，错法分成两类。它内部调的是 selection_request.model.with_structured_output(schema)，method 用 provider 的默认值，也就是 json_schema。前三条通路（middleware_selector_native.py）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
=== 1. 中间件内部走的是哪条路 ===
    schema = type_adapter.json_schema()
    structured_model = selection_request.model.with_structured_output(schema)

=== 2. LLMToolSelectorMiddleware 在三条通路上的表现 ===
--- DeepSeek deepseek-chat ---
  OpenAIInvalidRequestError: Error code: 400 - {'error': {'message': 'This response_format type is unavailable now (request_id: 91c07733-...)'}}
  结果: 失败

--- Ollama 云端 OpenAI 兼容端点 gpt-oss:120b ---
  OutputParserException: Invalid json output: weather For troubleshooting, visit: https://docs.langchain.com/oss/python/langchain/errors/OUTPUT_PARSING_FAILURE
  结果: 失败

--- Ollama 云端原生端点 gpt-oss:120b（temperature=0） ---
  OutputParserException: Invalid json output: weather For troubleshooting, visit: https://docs.langchain.com/oss/python/langchain/errors/OUTPUT_PARSING_FAILURE
  结果: 失败

--- Ollama 云端原生端点 gpt-oss:120b（默认温度） ---
  OutputParserException: Invalid json output: browse For troubleshooting, visit: https://docs.langchain.com/oss/python/langchain/errors/OUTPUT_PARSING_FAILURE
  结果: 失败

DeepSeek 死得最早，400 里明说这个 response_format 类型现在不给用。两条 Ollama 通路都是请求发得出去，模型回一个词就完事，JsonOutputParser 拿到非 JSON 直接抛。回的那个词每次还不一样，weather、browse 都见过。这三条属于 provider 层面不支持结构化输出，跟结构化输出那篇的结论对得上。

Gemini 是第二类，也是这篇文章要改口的地方。同一个 key 上，create_agent 加 ProviderStrategy 已经能拿回对象，但塞进这个中间件照样失败，报错换成了一个更具体的东西（gemini_provider_strategy.py）：

1
2
3
4
5
6
7
=== 2. Middleware 篇：LLMToolSelectorMiddleware ===
  候选工具名: ['get_weather', 'get_stock', 'send_email', 'book_flight']
  --- 2a. 直接看选择器模型的原始输出 ---
  原始输出: {'tools': ['get_weather']}
  --- 2b. 中间件实际行为 ---
  第1次: 失败 ValueError: Model selected invalid tools: ['Get weather for a city.']
  第2次: 失败 ValueError: Model selected invalid tools: ['Get weather for a city.']

Get weather for a city. 是 get_weather 这个工具的 description。模型没把 schema 里的 Literal["get_weather"] 当约束，把 description 当值吐了出来，LangChain 对不上工具名就抛 ValueError。这跟 provider 支不支持结构化输出已经没关系，Gemini 的结构化输出本身是好的。

换 system_prompt 能不能把它掰回来。我试了四种，每种跑两次（gemini_selector_schema_probe.py）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
=== 1. 四种 system_prompt，各跑 2 次 ===
  --- 默认 system_prompt ---
      第1次: 失败 ValueError: Model selected invalid tools: ['Get weather for a city.']
      第2次: 失败 ValueError: Model selected invalid tools: ['Get weather for a city.']
  --- 只返回工具名 ---
      system_prompt = 从候选工具里挑出与用户问题最相关的一个，只返回工具名。
      第1次: 失败 ValueError: Model selected invalid tools: ['Get weather for a city.']
      第2次: 失败 ValueError: Model selected invalid tools: ['Get weather for a city.']
  --- 只返回工具名 + 不要描述 ---
      system_prompt = 从候选工具里挑出与用户问题最相关的一个。只返回工具名，不要返回工具描述，不要解释。
      第1次: 失败 ValueError: Model selected invalid tools: ['Get weather for a city.']
      第2次: 失败 ValueError: Model selected invalid tools: ['Get weather for a city.']
  --- 只返回工具名 + 举例 get_weather ---
      system_prompt = 从候选工具里挑出与用户问题最相关的一个。只返回工具名本身（例如 get_weather），不要返回工具描述，不要解释。
      第1次: 失败 GoogleRateLimitError: Error calling model 'gemini-3.6-flash' (RESOURCE_EXHAUSTED): 429 ...
      第2次: OK 调用工具=['get_weather']

前三种都失败，明写着”不要返回工具描述”也没用。唯一通过的那版，prompt 里直接写了 例如 get_weather，等于把答案喂给了模型，换个查询就不成立。所以”换 system_prompt 修不好”这个结论保留，只是原因从 provider 不支持换成了 schema 形状不友好。

原因在源码里，三行就能看明白：

1
2
3
tool_selection.py:85  Annotated[Literal[tool.name], Field(description=tool.description)] for tool in tools
tool_selection.py:94  tools: Annotated[list[selected_tool_type], Field(description=description)]  # type: ignore[valid-type]
tool_selection.py:336  msg = f"Model selected invalid tools: {invalid_tool_selections}"

第 85 行给每个工具造一个 Literal["工具名"]，但 Field(description=...) 填的是工具描述。第 94 行把这些 Literal 组成 Union 再塞进一个 list。生成的 JSON Schema 长这样：

1
schema.tools.items = {"anyOf": [{"const": "get_weather", "description": "Get weather for a city.", "type": "string"}, ...]}

const 是工具名，description 是描述。Gemini 在这个形状下把 description 当成了候选值。约束写在 schema 里，不在提示词里，换 system_prompt 补不上这一刀。第 336 行就是最后对不上名字时抛错的地方。

模型本身有能力做这件事。同一份工具清单，不走中间件的 schema，直接问它，回答是对的（同一份脚本）：

1
2
3
4
5
=== 2. 对照实验：问题出在 schema 形状 ===
  --- 2a. 不走中间件的 schema，自由问答，明确只要工具名 ---
      模型回答: 'get_weather'
  --- 2b. 用中间件自己那份 schema（with_structured_output）直接问 ---
      模型返回: {'tools': ['Get weather for a city.']}

2a 和 2b 用同一个模型、同一份工具清单、同一个问题，差别只有 schema。自由问答回工具名，套上中间件那份 schema 就回描述。问题在 schema 形状，不在模型。

实验条件补一句。gemini-3.5-flash 的免费额度是每天 20 次，写这篇时已经用完，四种 system_prompt 那组是在同代的 gemini-3.6-flash 上跑的，默认 prompt 的失败在 3.5-flash 上单独复现过两次。上面第三组里那条 429 是并发太快撞了分钟级限流，重跑一次就过了。

替代品不难写。ModelRequest 里有 tools 字段，request.override(tools=...) 是官方的替换入口。让模型用普通对话挑一个工具名，自己对一遍名字，对不上就全部保留：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from collections.abc import Callable

from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse


@wrap_model_call
def select_tools(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    names = [t.name for t in request.tools]
    if len(names) <= 2:
        return handler(request)
    catalog = "\n".join(f"- {t.name}: {t.description}" for t in request.tools)
    picked = request.model.invoke([
        {"role": "system", "content": f"从下面这些工具里挑出与用户问题最相关的一个，只回工具名，不要解释。\n{catalog}"},
        {"role": "user", "content": str(request.messages[-1].content)},
    ]).content.strip().strip("。.")
    selected = [t for t in request.tools if t.name == picked]
    if not selected:
        return handler(request)
    return handler(request.override(tools=selected))

真实输出（middleware_tool_selector_probe.py）：

1
2
3
4
5
6
7
8
9
=== 2. 替代方案：@wrap_model_call 自己筛工具，跑在 DeepSeek 上 ===
  [select] 候选 ['get_weather', 'get_stock', 'send_email', 'book_flight'] -> 选中 ['get_weather']
  [select] 候选 ['get_weather', 'get_stock', 'send_email', 'book_flight'] -> 选中 ['get_weather']

消息序列:
  HumanMessage: tools=[] content='帮我查一下西安的天气'
  AIMessage: tools=['get_weather'] content=''
  ToolMessage: tools=[] content='西安 今天晴，24 度。'
  AIMessage: tools=[] content='西安今天天气晴朗，气温 24 度，挺舒适的，适合外出活动。'

四个工具进，一个工具出，模型调对了工具。日志打了两遍，是因为这一轮 agent 调了两次模型，筛选钩子每次模型调用都会跑一次。代价是每轮多一次模型调用，挑得准不准全看那段 prompt，没有 schema 兜底。工具筛选本来就是启发式，能接受。其余几个（摘要、重试、动态模型）都正常。

中间件顺序

多个中间件叠在一起时的规则，文档写得很清楚：

三条规则：before 正序，after 倒序，wrap 嵌套（第一个中间件包住所有后面的）。

我写了一个三类中间件的脚本验证，每个钩子打印自己的名字，wrap 钩子用缩进表示嵌套深度：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
class Named(AgentMiddleware):
    def before_model(self, state, runtime):
        log(f"{self.name}.before_model")
        return None

    def after_model(self, state, runtime):
        log(f"{self.name}.after_model")
        return None

    def wrap_model_call(self, request, handler):
        log(f"{self.name}.wrap_model_call 进入")
        depth["n"] += 1
        try:
            return handler(request)
        finally:
            depth["n"] -= 1
            log(f"{self.name}.wrap_model_call 返回")


class M1(Named): pass
class M2(Named): pass
class M3(Named): pass


agent = create_agent(model=model, tools=[add], middleware=[M1(), M2(), M3()])

真实输出（省略了后半段重复的循环）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
=== middleware=[M1, M2, M3] 的实际执行顺序 ===
M1.before_agent
M2.before_agent
M3.before_agent
M1.before_model
M2.before_model
M3.before_model
M1.wrap_model_call 进入
  M2.wrap_model_call 进入
    M3.wrap_model_call 进入
    M3.wrap_model_call 返回
  M2.wrap_model_call 返回
M1.wrap_model_call 返回
M3.after_model
M2.after_model
M1.after_model
M1.wrap_tool_call 进入 (add)
  M2.wrap_tool_call 进入 (add)
    M3.wrap_tool_call 进入 (add)
    M3.wrap_tool_call 返回
  M2.wrap_tool_call 返回
M1.wrap_tool_call 返回

这里有个坑，我第一次运行直接报错：

1
AssertionError: Please remove duplicate middleware instances.

原因是我写了三个 Named() 实例，而 create_agent 里有一句校验 len({m.name for m in middleware}) != len(middleware) 就抛异常。name 默认取类名，三个实例的类名都是 Named，于是撞名。改成三个子类就过了。给装饰器传 name="..." 也能绕开。

排序的经验，按重要性排：

  • 会短路、做校验的放前面。后面的钩子直接不跑，省掉无谓开销。
  • 日志类的放前面，能看到最原始的入参。放在后面看到的是被别的中间件改过的版本。
  • 重试类的放里面。官方文档里 ToolErrorMiddleware 和 ToolRetryMiddleware 的组合就是 error 在外、retry 在内，而且 retry 要配 on_failure="error"，让重试次数用尽后的异常冒到外层去兜。

短路

想提前结束，返回的 dict 里带 jump_to。三个目标：

  • 'end'：跳到 agent 结束，或者第一个 after_agent 钩子
  • 'tools'：跳到工具节点
  • 'model'：跳到模型节点，或者第一个 before_model 钩子

返回 jump_to 的钩子必须声明 can_jump_to。装饰器写法是在参数里写 @before_model(can_jump_to=["end"])，类写法是在方法上加 @hook_config(can_jump_to=["end"])。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
model_calls = {"n": 0}


@before_model(can_jump_to=["end"])
def block_secret(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    text = state["messages"][-1].content
    if "内部代号" in text:
        print("  [guard] 命中敏感词，返回 jump_to=end，本次不调用模型")
        return {
            "messages": [AIMessage("这个信息我不能提供。")],
            "jump_to": "end",
        }
    return None


@after_model
def count_calls(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    model_calls["n"] += 1
    print(f"  [count] 模型实际被调用了 {model_calls['n']} 次")
    return None

真实输出：

1
2
3
4
5
6
7
8
9
### A. 普通问题
  [count] 模型实际被调用了 1 次
回答：Agent 是一个能感知环境、自主决策并采取行动以达成目标的智能实体。
模型调用次数：1

### B. 触发短路的问题
  [guard] 命中敏感词，返回 jump_to=end，本次不调用模型
回答：这个信息我不能提供。
模型调用次数：1（与 A 相同，说明这次没有走模型）

B 这次没有走模型，回复直接由钩子给出。注意返回的 AIMessage 会真的进入消息历史，模型下一轮能看到它，所以别把中间状态塞进去。

'model' 这个目标用得少一些。一个场景是在 after_model 里发现回复格式不对，把最后一条替换成新的 AIMessage，再跳回模型节点让它重说一遍。

例子一：按对话长度切换模型

Agent 那篇里带过一句”可以动态选模型”，这里给完整版本。规则很简单：消息少于 4 条用 deepseek-chat，超过就换 deepseek-reasoner。同时加一个 @dynamic_prompt 生成系统提示，再用一个包裹式钩子把模型实际收到的提示打印出来，验证提示真的换掉了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
import sys, os
sys.stdout.reconfigure(encoding="utf-8")

from collections.abc import Callable
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain.agents.middleware import (
    wrap_model_call, dynamic_prompt, ModelRequest, ModelResponse,
)
from langchain.messages import HumanMessage, AIMessage

simple_model = ChatOpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
    model="deepseek-chat",
    temperature=0.1,
    max_tokens=300,
)
complex_model = ChatOpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
    model="deepseek-reasoner",
    temperature=0.1,
    max_tokens=300,
)

picked: list[str] = []


@wrap_model_call
def dynamic_model(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    if len(request.messages) > 3:
        model, name = complex_model, "deepseek-reasoner"
    else:
        model, name = simple_model, "deepseek-chat"
    picked.append(name)
    print(f"  [router] 消息数={len(request.messages)} -> 选择 {name}")
    return handler(request.override(model=model))


@dynamic_prompt
def state_aware_prompt(request: ModelRequest) -> str:
    base = "你是一个简洁的助手，回答不超过一句话。"
    if len(request.messages) > 3:
        base += " 这是一段很长的对话，请比平时更简短。"
    print(f"  [prompt] 生成系统提示，长度 {len(base)} 字符")
    return base


@wrap_model_call
def log_prompt(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    text = "".join(
        b.get("text", "") for b in request.system_message.content_blocks
        if isinstance(b, dict)
    )
    print(f"  [inspect] 模型实际收到的系统提示：{text}")
    return handler(request)


agent = create_agent(
    model=simple_model,
    tools=[],
    middleware=[dynamic_model, state_aware_prompt, log_prompt],
)

真实输出：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
  [router] 消息数=1 -> 选择 deepseek-chat
  [prompt] 生成系统提示，长度 19 字符
  [inspect] 模型实际收到的系统提示：你是一个简洁的助手，回答不超过一句话。
  [router] 消息数=5 -> 选择 deepseek-reasoner
  [prompt] 生成系统提示，长度 38 字符
  [inspect] 模型实际收到的系统提示：你是一个简洁的助手，回答不超过一句话。 这是一段很长的对话，请比平时更简短。

### A. 短对话：走 deepseek-chat
实际使用的模型：deepseek-chat
回答：中间件是位于操作系统和应用软件之间的软件层，用于协调不同系统或组件之间的通信与数据管理。

### B. 长对话：自动切到 deepseek-reasoner
实际使用的模型：deepseek-reasoner
回答：中间件是夹在请求与核心处理逻辑之间、用于统一拦截并处理日志、鉴权、限流等横切关注点的可复用组件。
响应里的 reasoning_tokens：190

几个实际踩到的点。

request.override(model=...) 返回一个新的 request，不要原地改属性。ModelRequest 是 dataclass，字段有 model、messages、system_message、tools、tool_choice、response_format、state、runtime、model_settings，override 是官方的修改入口。

判断条件可以读 request.messages（就是 request.state["messages"] 的快捷方式），也可以读 request.runtime.context 里的用户信息。路由到哪个模型完全由你决定，中间件本身不关心。

验证路由别信 response_metadata["model_name"]。DeepSeek 对 deepseek-chat 和 deepseek-reasoner 都返回 deepseek-flash，我一开始就被这个字段骗了。能区分的是 completion_tokens_details.reasoning_tokens：走 reasoner 时非空（上面是 190），走 chat 时是 None。

@dynamic_prompt 和 wrap_model_call 都是包裹式钩子，谁在里面谁先看到请求。我把 log_prompt 放在列表最后，它就在最内层，看到的是 @dynamic_prompt 改过之后的系统提示。顺序反过来，打印出来的就是原始提示。

例子二：工具抛异常之后

工具函数抛异常，默认行为是整个 agent run 直接崩掉。用户看到的是一个 traceback，而不是一句”这个查询失败了”。wrap_tool_call 可以在异常冒到图之前截住它，把它转成一条 status="error" 的 ToolMessage 塞回消息历史，模型读到之后有机会换个参数重试，或者换个工具。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from collections.abc import Callable
from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage
from langchain.tools.tool_node import ToolCallRequest
from langgraph.types import Command


@tool
def divide(a: float, b: float) -> float:
    """计算 a 除以 b。"""
    if b == 0:
        raise ValueError("除数不能为 0")
    return a / b


@wrap_tool_call
def catch_tool_error(
    request: ToolCallRequest,
    handler: Callable[[ToolCallRequest], ToolMessage | Command],
) -> ToolMessage | Command:
    name = request.tool_call["name"]
    print(f"  -> 拦截到工具调用 {name}，参数 {request.tool_call['args']}")
    try:
        result = handler(request)
        print(f"  <- {name} 正常返回")
        return result
    except Exception as e:
        print(f"  !! {name} 抛出 {type(e).__name__}: {e}")
        return ToolMessage(
            content=f"工具 {name} 执行失败（{type(e).__name__}），请检查参数后重试。",
            tool_call_id=request.tool_call["id"],
            name=name,
            status="error",
        )

给 agent 的指令是”先用 divide 算 10 除以 0，失败就改用 add 算 10 + 0”。真实输出：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
  -> 拦截到工具调用 divide，参数 {'a': 10, 'b': 0}
  !! divide 抛出 ValueError: 除数不能为 0
  -> 拦截到工具调用 add，参数 {'a': 10, 'b': 0}
  <- add 正常返回

### A. 自定义 wrap_tool_call 兜住异常
[HumanMessage] 请用 divide 工具计算 10 除以 0。如果这个工具失败了，就改用 add 工具计算 10 + 0，然后告诉我最终数字。
[AIMessage] I'll try the divide tool first as requested.
[ToolMessage(status=error)] 工具 divide 执行失败（ValueError），请检查参数后重试。
[AIMessage] The divide tool failed as expected (division by zero). Falling back to the add t
[ToolMessage(status=success)] 10
[AIMessage] 结果如下：
1. **divide 工具**：计算 10 ÷ 0 失败，报错 `ValueError`（除数为零，无法计算）。
2. **add 工具**（备...

模型看到 error 之后自己换了工具，run 没断。

同样的事，内置的 ToolErrorMiddleware 一行配置就够，不用自己写 handler：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from langchain.agents.middleware import ToolErrorMiddleware


def on_error(exc: Exception, request: ToolCallRequest) -> str | None:
    if isinstance(exc, ValueError):
        return f"工具 {request.tool_call['name']} 参数不合法，请修正后重试。"
    return None


agent = create_agent(
    model=model,
    tools=[divide, add],
    middleware=[ToolErrorMiddleware(on_error=on_error)],
)

on_error 的返回值控制信息披露。返回字符串，异常就变成 error ToolMessage；返回 None 或干脆不 return，异常继续上抛，run 会终止。文档建议返回异常类型而不是原始异常消息，因为原始消息里可能带着内部细节，比如数据库连接串。ToolErrorMiddleware 需要 langchain>=1.3.14，本地的 1.4.3 没问题。

再补一个重试的实测。ToolRetryMiddleware 包住工具调用，失败后按 initial_delay * backoff_factor ** retry_number 等待再试。我造了一个前两次必失败、第三次成功的工具：

1
2
3
4
5
6
7
8
9
retry = ToolRetryMiddleware(
    max_retries=3,
    retry_on=(ConnectionError,),
    initial_delay=0.5,
    backoff_factor=2.0,
    jitter=False,
    on_failure="continue",
)
agent = create_agent(model=model, tools=[flaky], middleware=[retry])

真实输出：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
  [flaky] 第 1 次调用，距上次 1.03s
  [flaky] 第 2 次调用，距上次 0.50s
  [flaky] 第 3 次调用，距上次 1.00s

### A. 前两次失败、第三次成功的工具
总耗时 3.43s，flaky 实际被调用 3 次
[ToolMessage(status=success)] 查询成功：middleware 的结果是 42
[AIMessage] 调用成功。结果如下：
- **查询内容**：middleware
- **返回结果**：42

### B. 重试耗尽后交给模型处理
dead 实际被调用 3 次（1 次初始 + 2 次重试）
[AIMessage]
[ToolMessage(status=error)] Tool 'dead' failed after 3 attempts with ConnectionError: 服务彻底不可用. Please try again.
[AIMessage] 调用失败了。
- 工具：`dead`
- 结果：重试 3 次后仍失败，报错 `ConnectionError: 服务彻底不可用`

间隔 0.50s 和 1.00s 正好是 0.5 乘 2 的等比。jitter 默认开着，会加正负 25% 的随机抖动，我为了看清楚数字关掉了；生产环境建议留着，避免一堆请求同时重试。

B 组把 max_retries=2 用尽，on_failure="continue" 让中间件返回一条 error ToolMessage，内容是 Tool 'dead' failed after 3 attempts with ConnectionError，模型读到之后直接告诉用户失败。如果把 on_failure 改成 "error"，异常会重新抛出，run 立刻结束。

SummarizationMiddleware 的效果也顺手验了，trigger=("messages", 6)、keep=("messages", 2)，送进去 11 条消息：

1
2
3
4
5
6
7
8
### 送入前：11 条消息
### 返回后：4 条消息
[HumanMessage] Here is a summary of the conversation to date:
## SESSION INTENT
用户正在学习 LangChain 的中间件（middleware），并逐轮要求记住一系列要点...
[AIMessage] 收到，要点 5 已记下。
[HumanMessage] 把前面几轮的要点合并成一句话告诉我。
[AIMessage] 目前记录到的要点 1 至要点 5 都只是编号...

11 条压成 4 条：一条摘要 HumanMessage，加上 keep 保留的最近两条，再加本次新回复。摘要本身是一次额外的模型调用，所以 model 参数通常配个便宜的小模型。

小结

  • 中间件是挂在 agent 循环固定位置的钩子。节点式钩子管顺序和状态，包裹式钩子管 handler 被调几次。
  • 执行顺序是 before 正序、after 倒序、wrap 嵌套。同名中间件会触发 AssertionError，用子类或 name= 区分。
  • 自定义状态用 state_schema 加 NotRequired 字段，钩子返回的 dict 自动合并，计数、标记、审计信息都能挂在上面。
  • 内置中间件覆盖了重试、兜错、摘要、HITL、PII、限流、任务规划。动态提示用 @dynamic_prompt，动态模型用 @wrap_model_call 加 request.override(model=...)。
  • LLMToolSelectorMiddleware 四条通路都跑不通，但原因分两类。DeepSeek 报 400，Ollama 云端无论兼容端点还是原生端点都是解析器拿到非 JSON 后抛 OutputParserException，这是 provider 不支持结构化输出。Gemini 上 provider 支持了，中间件却因为选择 schema 把工具描述写进了 Literal 的 description，模型把描述当值返回，抛 ValueError: Model selected invalid tools。换 system_prompt 补不上，因为约束在 schema 里。工具筛选用 @wrap_model_call 加 request.override(tools=...) 自己写，实测可行。摘要、重试、动态模型、工具兜错都正常。