9.1 引言：为什么 Agent 需要 Middleware

在前几章中，我们构建了能够调用工具、执行多步推理的 Agent。然而，生产环境中的 Agent 面临着更复杂的挑战：模型调用可能因网络波动失败、长对话会耗尽上下文窗口、敏感数据可能泄露给模型、高风险操作需要人工审批。这些横切关注点（Cross-Cutting Concerns）如果直接嵌入 Agent 主逻辑，会导致代码耦合严重、难以维护。

LangChain v1 引入了 Middleware 系统，采用经典的中间件模式，允许开发者在 Agent 执行的关键节点（模型调用前后、工具调用前后、Agent 启动/结束）插入可组合的拦截逻辑。本章将从源码层面深入解析：

• 架构设计：AgentMiddleware 基类、ModelRequest/ModelResponse 数据模型、6 大生命周期钩子
• 内置中间件全览：14 个开箱即用的中间件，覆盖容错、安全、监控、增强四大类场景
• 自定义中间件开发：通过类继承和装饰器两种方式创建中间件
• 组合模式：中间件栈的执行顺序、右到左合成算法、最佳实践

---

9.2 Middleware 架构总览

9.2.1 核心类型体系

Middleware 系统的类型定义集中在 types.py，包含三层抽象：

AgentMiddleware[StateT, ContextT, ResponseT]   ← 基类（定义生命周期钩子）
     ↑
ModelRequest[ContextT] / ModelResponse[ResponseT]  ← 数据模型（封装请求/响应）
     ↑
AgentState[ResponseT]                              ← 状态容器（消息 + 控制流）

公共 API 通过 __init__.py 统一导出，包括所有 14 个内置中间件类、基类、数据模型和 7 个装饰器。

9.2.2 AgentState：状态容器

AgentState 是所有中间件共享的状态类型，定义于 types.py：

class AgentState(TypedDict, Generic[ResponseT]):
    """State schema for the agent."""
    messages: Required[Annotated[list[AnyMessage], add_messages]]
    jump_to: NotRequired[Annotated[JumpTo | None, EphemeralValue, PrivateStateAttr]]
    structured_response: NotRequired[Annotated[ResponseT, OmitFromInput]]

三个字段的设计意图：

字段	类型	说明
messages	list[AnyMessage]	对话消息列表，使用 add_messages reducer 自动追加
jump_to	JumpTo | None	控制流跳转目标（"tools"/"model"/"end"），标记为 EphemeralValue（单次使用后清除）
structured_response	ResponseT	结构化输出，标记为 OmitFromInput（不出现在输入 schema 中）

注意 PrivateStateAttr（types.py）标记的字段对外不可见——它同时设置 input=True, output=True 的 OmitFromSchema，意味着该字段在输入和输出 schema 中均被排除。中间件可以利用这一机制存储内部状态而不污染公共接口。

9.2.3 自定义 State 扩展

中间件可以通过泛型参数扩展状态类型。例如 ModelCallLimitMiddleware 定义了自己的状态 schema（model_call_limit.py）：

class ModelCallLimitState(AgentState[ResponseT]):
    thread_model_call_count: NotRequired[Annotated[int, PrivateStateAttr]]
    run_model_call_count: NotRequired[Annotated[int, UntrackedValue, PrivateStateAttr]]

UntrackedValue 来自 LangGraph，表示该字段在 checkpoint 中不持久化——每次 Agent 调用（run）重新计数，而 thread_model_call_count 则在整个会话线程中累积。

---

9.3 AgentMiddleware 基类与生命周期钩子

9.3.1 基类定义

AgentMiddleware 是所有中间件的基类，定义于 types.py：

class AgentMiddleware(Generic[StateT, ContextT, ResponseT]):
    """Base middleware class for an agent."""

    state_schema: type[StateT] = cast("type[StateT]", _DefaultAgentState)
    """The schema for state passed to the middleware nodes."""

    tools: Sequence[BaseTool]
    """Additional tools registered by the middleware."""

    @property
    def name(self) -> str:
        return self.__class__.__name__

三个泛型参数控制类型安全：

泛型参数	约束	说明
StateT	bound=AgentState[Any]	中间件操作的状态类型，可自定义扩展字段
ContextT	无约束，默认 None	运行时上下文类型，通过 Runtime[ContextT] 传递
ResponseT	无约束，默认 Any	结构化输出类型

类属性 state_schema 告诉 Agent 框架该中间件需要的状态 schema，多个中间件的 schema 会在构建 Agent 图时自动合并。tools 属性允许中间件注册额外工具（如 FilesystemFileSearchMiddleware 注册文件搜索工具）。

9.3.2 六大生命周期钩子

AgentMiddleware 定义了 6 个生命周期钩子，覆盖 Agent 执行的完整流程。每个钩子都有同步和异步两个版本（如 before_agent / abefore_agent）：

Agent 调用开始
    │
    ▼
┌─────────────┐
│ before_agent │  ← 初始化、环境准备
└──────┬──────┘
       │
       ▼
┌──────────────┐     ┌─────────────────┐
│ before_model │────→│ wrap_model_call  │────→ 模型调用
└──────────────┘     └────────┬────────┘
                              │
                              ▼
                     ┌──────────────┐     ┌─────────────────┐
                     │  after_model │────→│  wrap_tool_call  │────→ 工具调用
                     └──────────────┘     └────────┬────────┘
                              │                     │
                              ├─────────────────────┘
                              │  (循环：模型调用 → 工具调用 → ...)
                              ▼
                     ┌──────────────┐
                     │  after_agent │  ← 清理、收尾
                     └──────────────┘

各钩子的签名与职责：

1. before_agent / abefore_agent（types.py）

def before_agent(self, state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None:

执行时机：Agent 开始执行前，仅执行一次。
典型用途：环境初始化（如 ShellToolMiddleware 启动 Shell 会话）、全局参数注入。

2. before_model / abefore_model（types.py）

def before_model(self, state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None:

执行时机：每次模型调用之前。
典型用途：上下文压缩（SummarizationMiddleware）、消息预处理、PII 检测。

3. after_model / aafter_model（types.py）

def after_model(self, state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None:

执行时机：每次模型调用完成之后。
典型用途：人工审批（HumanInTheLoopMiddleware）、输出后处理、PII 检测输出侧。

4. wrap_model_call / awrap_model_call（types.py）

def wrap_model_call(
    self,
    request: ModelRequest[ContextT],
    handler: Callable[[ModelRequest[ContextT]], ModelResponse[ResponseT]],
) -> ModelResponse[ResponseT] | AIMessage | ExtendedModelResponse[ResponseT]:

执行时机：包裹整个模型调用过程。
核心机制：接收一个 handler 回调，调用它即执行模型。中间件可以：

• 修改请求：handler(request.override(model=different_model))
• 重试：在 try/except 中多次调用 handler
• 短路：不调用 handler，直接返回结果
• 修改响应：调用 handler 后修改返回值

这是最强大的钩子——它控制模型调用的完整生命周期。

5. wrap_tool_call / awrap_tool_call（types.py）

def wrap_tool_call(
    self,
    request: ToolCallRequest,
    handler: Callable[[ToolCallRequest], ToolMessage | Command[Any]],
) -> ToolMessage | Command[Any]:

执行时机：包裹每个工具调用。
核心机制：与 wrap_model_call 类似的 handler 回调模式，但作用于工具调用。可以重试失败的工具调用、模拟工具执行、限制调用次数。

6. after_agent / aafter_agent（types.py）

def after_agent(self, state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None:

执行时机：Agent 执行结束后，仅执行一次。
典型用途：资源清理（如 ShellToolMiddleware 关闭 Shell 会话）、结果后处理。

9.3.3 钩子返回值与控制流

before_* / after_* 钩子的返回值遵循统一约定：

返回值	效果
None	不修改状态
dict[str, Any]	合并到 Agent 状态中
{"jump_to": "end"}	跳转到结束节点，终止 Agent
{"jump_to": "model"}	跳回模型调用节点
{"jump_to": "tools"}	跳到工具执行节点

使用 jump_to 需要通过 @hook_config 装饰器预先声明可跳转目标（types.py），这在 Agent 图的构建阶段就确定了条件边：

class MyMiddleware(AgentMiddleware):
    @hook_config(can_jump_to=["end", "model"])
    def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        if should_exit(state):
            return {"jump_to": "end"}
        return None

---

9.4 ModelRequest 与 ModelResponse：请求响应模型

9.4.1 ModelRequest：不可变请求封装

ModelRequest 是对模型调用请求的完整封装，定义于 types.py：

@dataclass(init=False)
class ModelRequest(Generic[ContextT]):
    model: BaseChatModel
    messages: list[AnyMessage]       # 不含 system message
    system_message: SystemMessage | None
    tool_choice: Any | None
    tools: list[BaseTool | dict[str, Any]]
    response_format: ResponseFormat[Any] | None
    state: AgentState[Any]
    runtime: Runtime[ContextT]
    model_settings: dict[str, Any]

关键设计特征：

1. 不可变模式：直接属性赋值会触发 DeprecationWarning（types.py），引导使用 override() 方法：

# 不推荐（触发警告）
request.model = different_model

# 推荐（返回新实例，原实例不变）
new_request = request.override(model=different_model)

override() 方法（types.py）内部使用 dataclasses.replace() 创建浅拷贝，修改指定字段后返回新实例。这种不可变模式确保中间件链中每个层看到的请求不会被其他层意外修改。

2. system_prompt 兼容性：system_prompt 属性（types.py）作为 system_message 的便捷访问器保留，但新代码应使用 system_message。

3. 携带完整上下文：state 和 runtime 字段让中间件在 wrap_model_call 中无需额外参数即可访问完整上下文。

9.4.2 ModelResponse：响应封装

ModelResponse 封装模型调用的返回结果（types.py）：

@dataclass
class ModelResponse(Generic[ResponseT]):
    result: list[AnyMessage]
    structured_response: ResponseT | None = None

result 通常包含单个 AIMessage，但如果模型使用工具进行结构化输出，可能额外包含一个 ToolMessage。

wrap_model_call 的返回值支持三种形式，框架会自动标准化：

返回类型	自动转换
ModelResponse	直接使用
AIMessage	包装为 ModelResponse(result=[ai_message])
ExtendedModelResponse	提取 model_response 和 command

ExtendedModelResponse（types.py）在 ModelResponse 基础上增加了 command 字段，允许中间件在返回响应的同时发出控制流指令。

---

9.5 内置 Middleware 全览

LangChain v1 提供了 14 个内置中间件，覆盖四大类场景。以下按功能分类逐一解析。

9.5.1 容错类中间件

ModelFallbackMiddleware —— 模型降级

源码位置：model_fallback.py

使用钩子：wrap_model_call

当主模型调用失败时，按顺序尝试备选模型，所有模型均失败则抛出最后一个异常：

class ModelFallbackMiddleware(AgentMiddleware[AgentState[ResponseT], ContextT, ResponseT]):
    def __init__(self, first_model: str | BaseChatModel, *additional_models: str | BaseChatModel):
        # 初始化所有备选模型
        ...

    def wrap_model_call(self, request, handler):
        # 先尝试主模型
        try:
            return handler(request)
        except Exception as e:
            last_exception = e

        # 依次尝试备选模型
        for fallback_model in self.models:
            try:
                return handler(request.override(model=fallback_model))
            except Exception as e:
                last_exception = e
                continue

        raise last_exception

核心逻辑非常精简（model_fallback.py）：利用 request.override(model=fallback_model) 创建新请求，传入同一个 handler。这样备选模型使用与主模型完全相同的 messages、tools、tool_choice 等参数。

使用示例：

from langchain.agents import create_agent
from langchain.agents.middleware import ModelFallbackMiddleware

agent = create_agent(
    model="openai:gpt-4o",            # 主模型
    tools=[search_tool],
    middleware=[
        ModelFallbackMiddleware(
            "openai:gpt-4o-mini",       # 第一备选
            "anthropic:claude-sonnet-4-20250514",   # 第二备选
        )
    ],
)

ModelRetryMiddleware —— 指数退避重试

源码位置：model_retry.py

使用钩子：wrap_model_call

对模型调用失败进行自动重试，支持指数退避、异常过滤和自定义错误处理：

from langchain.agents.middleware import ModelRetryMiddleware
from openai import APITimeoutError, RateLimitError

retry = ModelRetryMiddleware(
    max_retries=3,
    retry_on=(APITimeoutError, RateLimitError),  # 仅重试特定异常
    backoff_factor=1.5,                           # 指数因子
    initial_delay=0.5,                            # 初始延迟（秒）
)

retry_on 参数支持三种形式：

• 异常类型元组：(APITimeoutError, RateLimitError)
• 可调用函数：lambda exc: isinstance(exc, APIStatusError) and exc.status_code >= 500
• None（默认）：重试所有异常

ToolRetryMiddleware —— 工具调用重试

源码位置：tool_retry.py

使用钩子：wrap_tool_call

与 ModelRetryMiddleware 对称，但作用于工具调用。当外部 API（如搜索引擎、数据库）返回错误时自动重试：

from langchain.agents.middleware import ToolRetryMiddleware

tool_retry = ToolRetryMiddleware(
    max_retries=3,
    backoff_factor=1.0,
    initial_delay=1.0,
)

ModelCallLimitMiddleware —— 模型调用限制

源码位置：model_call_limit.py

使用钩子：before_model（配合 @hook_config(can_jump_to=["end"])）

跟踪模型调用次数并在超限时终止 Agent，防止无限循环消耗 Token。支持两个维度的限制：

from langchain.agents.middleware import ModelCallLimitMiddleware

limit = ModelCallLimitMiddleware(
    run_limit=20,     # 单次 Agent 调用的最大模型调用次数
    thread_limit=100, # 整个会话线程的累计上限
    exit_behavior="end",  # 超限行为："end" | "error" | "continue"
)

该中间件通过自定义 ModelCallLimitState（model_call_limit.py）扩展 AgentState，添加了 thread_model_call_count 和 run_model_call_count 两个私有字段来持久化计数。

ToolCallLimitMiddleware —— 工具调用限制

源码位置：tool_call_limit.py

使用钩子：wrap_tool_call（配合 after_model 和 @hook_config）

限制工具调用次数，支持按工具名称单独限制或全局限制。与 ModelCallLimitMiddleware 类似，使用自定义 ToolCallLimitState 跟踪每个工具的调用计数。

9.5.2 上下文管理类中间件

SummarizationMiddleware —— 上下文压缩

源码位置：summarization.py

使用钩子：before_model

当对话消息超过阈值时，自动调用一个总结模型压缩历史消息，保留最近的消息不变。这是处理长对话的关键中间件。

from langchain.agents.middleware import SummarizationMiddleware

summarization = SummarizationMiddleware(
    model="openai:gpt-4o-mini",       # 用于生成摘要的模型（可以用廉价模型）
    trigger=("fraction", 0.8),         # 达到模型最大输入 token 的 80% 时触发
    keep=("messages", 20),             # 保留最近 20 条消息不压缩
)

trigger 参数（summarization.py）支持三种阈值类型：

类型	语法	说明
ContextFraction	("fraction", 0.8)	模型最大输入 token 的百分比
ContextTokens	("tokens", 3000)	绝对 token 数
ContextMessages	("messages", 50)	消息条数

还可以传入列表 [("fraction", 0.8), ("messages", 100)]，表示任一条件满足即触发。

工作流程（summarization.py）：

1. 在 before_model 中统计当前消息的 token 数
2. 判断是否满足触发条件
3. 根据 keep 参数确定切割点，将旧消息送入总结模型
4. 用总结结果替换旧消息，保留近期消息

ContextEditingMiddleware —— 状态修改

源码位置：context_editing.py

使用钩子：wrap_model_call

在模型调用时动态编辑上下文。内置策略 ClearToolUsesEdit 会清除消息中的工具调用记录，减少 token 消耗：

from langchain.agents.middleware import ContextEditingMiddleware, ClearToolUsesEdit

editing = ContextEditingMiddleware(edits=[ClearToolUsesEdit()])

9.5.3 安全类中间件

PIIMiddleware —— 敏感数据脱敏

源码位置：pii.py

使用钩子：before_model（输入侧扫描）、after_model（输出侧扫描）

检测并处理对话中的个人敏感信息（PII），支持 5 种内置检测类型和 4 种处理策略：

内置检测类型：email、credit_card（Luhn 校验）、ip、mac_address、url

处理策略（pii.py）：

策略	效果	适用场景
block	检测到 PII 即抛出异常	严格合规
redact	替换为 [REDACTED_EMAIL] 等占位符	日志脱敏
mask	部分遮盖，如 ****-****-****-1234	客服界面
hash	替换为确定性哈希，如 <email_hash:a1b2c3d4>	数据分析

from langchain.agents.middleware import PIIMiddleware

agent = create_agent(
    model="openai:gpt-4o",
    middleware=[
        PIIMiddleware("credit_card", strategy="mask"),   # 信用卡号部分遮盖
        PIIMiddleware("email", strategy="redact"),        # 邮箱地址脱敏
        PIIMiddleware("ip", strategy="hash"),             # IP 地址哈希化
    ],
)

还支持自定义检测器——通过正则表达式或自定义函数检测业务特定的敏感数据：

# 检测自定义格式的 API Key
PIIMiddleware("api_key", detector=r"sk-[a-zA-Z0-9]{32}", strategy="block")

HumanInTheLoopMiddleware —— 人工审批

源码位置：human_in_the_loop.py

使用钩子：after_model

在模型发出工具调用后、工具实际执行前，暂停 Agent 并请求人工审批。基于 LangGraph 的 interrupt 机制实现。

from langchain.agents.middleware import HumanInTheLoopMiddleware

hitl = HumanInTheLoopMiddleware(
    interrupt_on={
        "delete_file": True,          # 所有操作都需审批（approve/edit/reject）
        "send_email": {
            "allowed_decisions": ["approve", "reject"],  # 只能批准或拒绝
            "description": "即将发送邮件，请确认",
        },
        "search": False,              # 自动批准，不中断
    },
)

审批流程（human_in_the_loop.py）：

1. after_model 检查最后一条 AIMessage 的 tool_calls
2. 为需要审批的工具调用构造 HITLRequest（包含 ActionRequest 和 ReviewConfig）
3. 调用 interrupt() 暂停 Agent，等待人工决策
4. 根据人工决策（ApproveDecision/EditDecision/RejectDecision）修改或取消工具调用

三种决策类型：

决策	效果
approve	原样执行工具调用
edit	用人工修改后的参数替换原始工具调用
reject	取消工具调用，返回带 status="error" 的 ToolMessage 给模型

ShellToolMiddleware —— 受控 Shell 执行

源码位置：shell_tool.py

使用钩子：before_agent（启动 Shell）、after_agent（关闭 Shell）

为 Agent 提供受控的 Shell 命令执行能力，支持多种执行策略：HostExecutionPolicy（宿主机直接执行）、DockerExecutionPolicy（Docker 容器隔离）、CodexSandboxExecutionPolicy（沙箱执行）。

9.5.4 增强类中间件

LLMToolSelectorMiddleware —— 智能工具过滤

源码位置：tool_selection.py

使用钩子：wrap_model_call

当 Agent 配置了大量工具时，在每次模型调用前使用另一个 LLM 智能选择最相关的工具子集，减少 token 消耗和工具选择错误：

from langchain.agents.middleware import LLMToolSelectorMiddleware

selector = LLMToolSelectorMiddleware(
    model="openai:gpt-4o-mini",     # 用廉价模型做工具选择
    top_k=5,                         # 最多选择 5 个工具
)

实现原理（tool_selection.py）：构造一个 ToolSelectionResponse 结构化输出 schema，每个工具名称作为 Literal 类型的 Union，LLM 返回最相关的工具名称列表，然后用 request.override(tools=selected_tools) 只传递选中的工具给主模型。

LLMToolEmulator —— 工具模拟

源码位置：tool_emulator.py

使用钩子：wrap_tool_call

使用 LLM 模拟工具执行结果，用于测试环境或工具不可用时的降级：

from langchain.agents.middleware import LLMToolEmulator

# 模拟所有工具
emulator = LLMToolEmulator()

# 只模拟特定工具
emulator = LLMToolEmulator(tools=["get_weather", "get_stock_price"])

# 使用自定义模型做模拟
emulator = LLMToolEmulator(tools=["get_weather"], model="openai:gpt-4o-mini")

TodoListMiddleware —— 任务管理

源码位置：todo.py

使用钩子：wrap_model_call、after_model

为 Agent 提供任务规划与追踪能力，在模型调用时注入任务列表上下文，在模型响应后更新任务状态。

FilesystemFileSearchMiddleware —— 文件搜索

源码位置：file_search.py

通过 tools 属性注册工具

为 Agent 内嵌文件搜索能力（glob 匹配和 grep 搜索），使 Agent 具备 RAG 能力而无需外部向量数据库。

9.5.5 内置中间件速查表

中间件	钩子	分类	核心功能
ModelFallbackMiddleware	wrap_model_call	容错	主模型失败时依次尝试备选模型
ModelRetryMiddleware	wrap_model_call	容错	指数退避重试模型调用
ModelCallLimitMiddleware	before_model	容错	限制模型调用次数防止无限循环
ToolRetryMiddleware	wrap_tool_call	容错	指数退避重试工具调用
ToolCallLimitMiddleware	wrap_tool_call	容错	限制工具调用次数
SummarizationMiddleware	before_model	上下文	长对话自动压缩摘要
ContextEditingMiddleware	wrap_model_call	上下文	运行时编辑消息上下文
PIIMiddleware	before_model/after_model	安全	敏感数据检测与脱敏
HumanInTheLoopMiddleware	after_model	安全	高风险操作人工审批
ShellToolMiddleware	before_agent/after_agent	安全	受控 Shell 命令执行
LLMToolSelectorMiddleware	wrap_model_call	增强	LLM 智能工具过滤
LLMToolEmulator	wrap_tool_call	增强	LLM 模拟工具执行
TodoListMiddleware	wrap_model_call/after_model	增强	任务规划与追踪
FilesystemFileSearchMiddleware	tools 注册	增强	文件 glob/grep 搜索

---

9.6 自定义 Middleware 开发

9.6.1 方式一：继承 AgentMiddleware

继承基类并覆写所需钩子是最灵活的方式：

from langchain.agents.middleware import AgentMiddleware, AgentState, ModelRequest, ModelResponse
from langgraph.runtime import Runtime

class RequestLoggingMiddleware(AgentMiddleware):
    """记录每次模型调用的请求和响应。"""

    def wrap_model_call(self, request: ModelRequest, handler):
        print(f"[LOG] 模型: {request.model._llm_type}")
        print(f"[LOG] 消息数: {len(request.messages)}")
        print(f"[LOG] 工具数: {len(request.tools)}")

        response = handler(request)

        print(f"[LOG] 响应消息数: {len(response.result)}")
        return response

带自定义状态的示例：

from typing import Any
from typing_extensions import NotRequired, Annotated

class TokenTrackingState(AgentState):
    """扩展状态以追踪 token 用量。"""
    total_tokens: NotRequired[Annotated[int, PrivateStateAttr]]

class TokenTrackingMiddleware(AgentMiddleware[TokenTrackingState, None, Any]):
    """追踪累计 token 用量。"""

    state_schema = TokenTrackingState

    def after_model(self, state: TokenTrackingState, runtime: Runtime) -> dict[str, Any]:
        last_msg = state["messages"][-1]
        usage = getattr(last_msg, "usage_metadata", None)
        current = state.get("total_tokens", 0)
        if usage:
            return {"total_tokens": current + usage.get("total_tokens", 0)}
        return {"total_tokens": current}

9.6.2 方式二：装饰器快捷创建

对于简单的钩子逻辑，使用装饰器更为简洁。LangChain 提供了 7 个装饰器（types.py）：

装饰器	对应钩子	定义位置
@before_model	before_model	types.py:910-1077
@after_model	after_model	types.py:1080-1235
@before_agent	before_agent	types.py:1238-1428
@after_agent	after_agent	types.py:1431-1587
@dynamic_prompt	wrap_model_call	types.py:1590-1733
@wrap_model_call	wrap_model_call	types.py:1736-1892
@wrap_tool_call	wrap_tool_call	types.py:1895-2027

装饰器的内部实现原理：以 @before_model 为例（types.py），装饰器动态创建一个 AgentMiddleware 子类：

def decorator(func):
    is_async = iscoroutinefunction(func)

    if is_async:
        async def async_wrapped(_self, state, runtime):
            return await func(state, runtime)

        middleware_name = name or getattr(func, "__name__", "BeforeModelMiddleware")
        return type(
            middleware_name,
            (AgentMiddleware,),
            {
                "state_schema": state_schema or AgentState,
                "tools": tools or [],
                "abefore_model": async_wrapped,
            },
        )()
    ...

核心技巧是使用 type(name, bases, dict) 动态创建类——这与手动定义 class MyMiddleware(AgentMiddleware) 等价，但一行代码即可完成。装饰器自动检测函数是同步还是异步，设置对应的钩子方法。

基本装饰器示例：

from langchain.agents.middleware import before_model, after_model, AgentState
from langgraph.runtime import Runtime

@before_model
def log_before(state: AgentState, runtime: Runtime) -> None:
    """在每次模型调用前打印消息数量。"""
    print(f"即将调用模型，当前消息数: {len(state['messages'])}")

@after_model
def log_after(state: AgentState, runtime: Runtime) -> None:
    """在每次模型调用后打印最新消息。"""
    print(f"模型返回: {state['messages'][-1].content[:100]}")

@dynamic_prompt 装饰器（types.py）是一个特化的 wrap_model_call，专门用于动态生成 system prompt：

from langchain.agents.middleware import dynamic_prompt, ModelRequest

@dynamic_prompt
def user_aware_prompt(request: ModelRequest) -> str:
    """根据运行时上下文动态生成 system prompt。"""
    user_name = request.runtime.context.get("user_name", "用户")
    msg_count = len(request.state["messages"])

    if msg_count > 20:
        return f"你正在与{user_name}进行长对话。请保持简洁。"
    return f"你是{user_name}的专属助手。"

agent = create_agent(
    model="openai:gpt-4o",
    middleware=[user_aware_prompt],
)

# 通过 runtime context 传递用户信息
result = agent.invoke(
    {"messages": [HumanMessage("你好")]},
    config={"configurable": {"context": {"user_name": "张三"}}},
)

带参数的装饰器：

@before_model(can_jump_to=["end"])
def check_budget(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """检查 token 预算，超限则终止 Agent。"""
    budget = runtime.context.get("token_budget", float("inf"))
    if len(state["messages"]) > budget:
        return {"jump_to": "end"}
    return None

@wrap_model_call 装饰器示例：

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

@wrap_model_call
def add_timestamp(request: ModelRequest, handler):
    """在每次模型调用的 system prompt 中注入当前时间。"""
    from datetime import datetime
    timestamp = datetime.now().isoformat()
    original_prompt = request.system_message.text if request.system_message else ""
    new_prompt = f"{original_prompt}\n\n当前时间: {timestamp}"
    response = handler(request.override(system_prompt=new_prompt))
    return response

9.6.3 同步与异步对偶

每个钩子都有同步和异步版本。关键规则：

• 如果只实现了同步版本（wrap_model_call），在异步调用（ainvoke/astream）时会抛出 NotImplementedError
• 如果只实现了异步版本（awrap_model_call），在同步调用（invoke/stream）时会抛出 NotImplementedError
• 建议同时实现两个版本，或使用装饰器（自动处理同步/异步转换）

源码中的错误提示非常明确（types.py）：

msg = (
    "Synchronous implementation of wrap_model_call is not available. "
    "You are likely encountering this error because you defined only the async version "
    "(awrap_model_call) and invoked your agent in a synchronous context "
    "(e.g., using `stream()` or `invoke()`). "
    "To resolve this, either: "
    "(1) subclass AgentMiddleware and implement the synchronous wrap_model_call method, "
    "(2) use the @wrap_model_call decorator on a standalone sync function, or "
    "(3) invoke your agent asynchronously using `astream()` or `ainvoke()`."
)
raise NotImplementedError(msg)

---

9.7 Middleware 组合模式

9.7.1 中间件栈与执行顺序

中间件通过 create_agent 的 middleware 参数按列表顺序传入：

agent = create_agent(
    model="openai:gpt-4o",
    tools=[web_search, code_executor, file_manager],
    middleware=[
        ToolCallLimitMiddleware(max_calls=10),                        # 第 1 层
        ModelFallbackMiddleware("anthropic:claude-sonnet-4-20250514"),   # 第 2 层
        HumanInTheLoopMiddleware(interrupt_on={"code_executor": True}),  # 第 3 层
        SummarizationMiddleware(model="openai:gpt-4o-mini"),           # 第 4 层
    ],
)

对于 before_model / after_model 等简单钩子，按列表顺序依次执行。

对于 wrap_model_call / wrap_tool_call，情况更复杂——它们采用洋葱模型组合。

9.7.2 右到左合成算法

wrap_model_call 的组合由 _chain_model_call_handlers() 函数实现（factory.py）。核心算法是从右到左折叠：

# 列表中第一个 = 最外层
composed_handler = compose_two(handlers[-2], handlers[-1])
for h in reversed(handlers[:-2]):
    composed_handler = compose_two(h, composed_handler)

compose_two 的内部逻辑（factory.py）：

def compose_two(outer, inner):
    def composed(request, handler):
        accumulated_commands = []

        def inner_handler(req):
            accumulated_commands.clear()    # 重试安全：每次清除
            inner_result = inner(req, handler)
            # 提取 inner 的 commands
            if isinstance(inner_result, _ComposedExtendedModelResponse):
                accumulated_commands.extend(inner_result.commands)
                return inner_result.model_response
            ...
            return _normalize_to_model_response(inner_result)

        outer_result = outer(request, inner_handler)
        return _to_composed_result(outer_result, extra_commands=accumulated_commands or None)

    return composed

以 [A, B, C] 三个 wrap_model_call 中间件为例，调用链形成嵌套结构：

A(request, B(request, C(request, actual_model_call)))

即 列表中第一个中间件（A）是最外层，最后一个（C）最接近实际的模型调用。这意味着：

• A 最先看到请求，最后看到响应
• C 最后看到请求，最先看到响应
• 如果 A 不调用 handler，B 和 C 都不会执行（短路）

Command 累积机制：accumulated_commands 列表从内层到外层累积 Command 对象。这使得内层中间件发出的跳转指令不会被外层中间件覆盖，而是共存——框架最终按顺序处理所有 commands。特别注意 accumulated_commands.clear() 的重试安全设计：如果外层中间件重试（多次调用 handler），内层的 commands 会被正确清除重建。

9.7.3 组合最佳实践

推荐的中间件排列策略——安全优先、增强其次、监控最后：

middleware = [
    # 1. 安全层（最外层，最先执行）
    PIIMiddleware("credit_card", strategy="block"),
    PIIMiddleware("email", strategy="redact"),
    ModelCallLimitMiddleware(run_limit=30),
    ToolCallLimitMiddleware(max_calls=20),

    # 2. 容错层
    ModelFallbackMiddleware("openai:gpt-4o-mini"),
    ModelRetryMiddleware(max_retries=3),
    ToolRetryMiddleware(max_retries=2),

    # 3. 增强层
    LLMToolSelectorMiddleware(model="openai:gpt-4o-mini", top_k=5),
    SummarizationMiddleware(model="openai:gpt-4o-mini", trigger=("fraction", 0.8)),

    # 4. 交互层（最内层）
    HumanInTheLoopMiddleware(interrupt_on={"dangerous_tool": True}),
]

排列逻辑解析：

1. PII 检测放在最外层：在任何处理之前先脱敏输入数据，确保后续中间件和模型不会看到原始敏感信息
2. 限流放在容错前面：超限直接终止，不会触发不必要的重试
3. Fallback 在 Retry 外层：先尝试重试当前模型，都失败后再切换备选模型
4. 工具选择在摘要之前：先过滤工具减少 token，再决定是否需要压缩上下文
5. 人工审批放在最内层：只有当所有自动化处理都完成后，才呈现给人工审批

---

9.8 实战：构建生产级 Agent

将本章所学组合起来，构建一个生产就绪的 Agent：

from langchain.agents import create_agent
from langchain.agents.middleware import (
    ModelFallbackMiddleware,
    ModelRetryMiddleware,
    ModelCallLimitMiddleware,
    SummarizationMiddleware,
    PIIMiddleware,
    HumanInTheLoopMiddleware,
    ToolCallLimitMiddleware,
    before_agent,
    after_agent,
    dynamic_prompt,
)
from langchain.agents.middleware import AgentState, ModelRequest
from langgraph.runtime import Runtime


# === 自定义中间件：使用装饰器 ===

@before_agent
async def setup_session(state: AgentState, runtime: Runtime) -> None:
    """Agent 启动时初始化会话。"""
    runtime.stream_writer({"type": "status", "message": "正在初始化..."})


@after_agent
async def cleanup_session(state: AgentState, runtime: Runtime) -> None:
    """Agent 结束时清理资源。"""
    msg_count = len(state["messages"])
    runtime.stream_writer({
        "type": "status",
        "message": f"会话结束，共 {msg_count} 条消息",
    })


@dynamic_prompt
def personalized_prompt(request: ModelRequest) -> str:
    """根据用户角色动态生成 system prompt。"""
    role = request.runtime.context.get("user_role", "general")
    prompts = {
        "developer": "你是一个资深软件工程师助手，擅长代码审查和架构设计。",
        "analyst": "你是一个数据分析师助手，擅长数据解读和可视化建议。",
        "general": "你是一个通用助手，乐于帮助用户解决各种问题。",
    }
    return prompts.get(role, prompts["general"])


# === 组装 Agent ===

agent = create_agent(
    model="openai:gpt-4o",
    tools=[web_search, code_executor, file_manager, database_query],
    middleware=[
        # 安全层
        PIIMiddleware("credit_card", strategy="block"),
        PIIMiddleware("email", strategy="redact"),
        ModelCallLimitMiddleware(run_limit=25, exit_behavior="end"),

        # 容错层
        ModelFallbackMiddleware("anthropic:claude-sonnet-4-20250514", "openai:gpt-4o-mini"),
        ModelRetryMiddleware(max_retries=3, backoff_factor=1.5),

        # 增强层
        personalized_prompt,
        SummarizationMiddleware(
            model="openai:gpt-4o-mini",
            trigger=[("fraction", 0.75), ("messages", 80)],
            keep=("messages", 15),
        ),

        # 交互层
        HumanInTheLoopMiddleware(interrupt_on={
            "code_executor": True,
            "file_manager": {"allowed_decisions": ["approve", "reject"]},
            "database_query": {"allowed_decisions": ["approve", "edit", "reject"]},
        }),
        ToolCallLimitMiddleware(max_calls=15),

        # 生命周期
        setup_session,
        cleanup_session,
    ],
)

# === 使用 Agent ===

async for mode, event in agent.astream(
    {"messages": [HumanMessage("分析项目代码质量并生成报告")]},
    config={"configurable": {"context": {"user_role": "developer"}}},
    stream_mode=["updates", "custom"],
):
    if mode == "custom":
        print(f"状态: {event}")
    elif mode == "updates":
        # 处理 Agent 状态更新
        ...

---

9.9 设计洞察与总结

9.9.1 架构设计亮点

1. 不可变请求模式：ModelRequest.override() 返回新实例而非原地修改，这确保了中间件链中每一层都基于干净的数据工作。这是函数式编程中 immutable data 的经典应用——在并发和重试场景下尤为重要。

2. 洋葱模型与 handler 回调：wrap_model_call 的 handler 回调模式直接借鉴了 Web 框架（如 Express.js、Koa）的中间件设计。与简单的 before/after 钩子相比，handler 回调模式可以实现更复杂的控制流——重试、短路、降级——这些在 before/after 模式下需要额外的协调机制。

3. 右到左合成的重试安全性：_chain_model_call_handlers 中 accumulated_commands.clear() 的设计细节看似简单，但解决了一个微妙问题：当外层中间件重试时，内层中间件的副作用（commands）必须被清除并重新生成，否则会产生重复的控制流指令。

4. 泛型与类型安全：三个泛型参数 StateT, ContextT, ResponseT 贯穿整个中间件系统，确保：

• 中间件只能访问它声明的状态字段
• Runtime[ContextT] 强制上下文类型一致
• ModelResponse[ResponseT] 保持结构化输出类型不丢失

5. 装饰器的动态类创建：通过 type(name, bases, dict) 在运行时创建 AgentMiddleware 子类，实现了"一个函数 = 一个中间件"的极简 API，同时保持与类继承方式完全兼容的内部机制。

9.9.2 与上下文的联系

• 第五章（回调系统）：Middleware 的生命周期钩子与 Callback 系统的 on_llm_start / on_llm_end 等事件类似，但层次不同——Callback 是底层的事件通知机制，Middleware 是上层的行为修改机制
• 第六章（Agent 架构）：Middleware 扩展了 Agent 的 ReAct 循环，在不修改核心循环逻辑的前提下注入横切关注点
• 第八章（LangSmith）：LangChainTracer 是通过 Callback 系统实现追踪的，而 Middleware 可以在更高层面控制追踪粒度

9.9.3 下一章预告

下一章将深入 Prompt 工程与输出解析，探索 PromptTemplate、ChatPromptTemplate、FewShotPromptTemplate 的模板体系，以及 OutputParser 如何将 LLM 的自由文本转换为结构化数据——这些能力与本章的 @dynamic_prompt 装饰器形成互补，共同构成 LLM 应用的输入输出控制层。