LangChain 08 - 中间件学习导航与实战

模块 1：你只需要记住这些

1. 什么是中间件

Middleware（中间件），简单说就是 Agent 执行过程中的钩子函数。

钩子（Hook）是框架或系统在某些关键执行点暴露的扩展接口。开发者可以"挂上"自己的逻辑，在那些点上插入、修改或替换行为，而无需改变主流程代码。

在 LangChain 的 Agent 执行循环中，中间件可以在"模型调用前"、“模型调用后”、"工具调用前后"等关键位置设置钩子，实现策略与治理。

create_agent 的中间件用法：

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware, HumanInTheLoopMiddleware

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

没有中间件的 Agent 架构是线性的：用户输入 → 拼接消息 → 调用模型 → 调用工具 → 返回结果。添加中间件后，每个关键节点都可以被拦截和增强。

2. 为什么需要中间件

真实项目中，Agent 会遇到很多额外需求：

根据问题复杂度动态切换模型
限制某些用户只能调用部分工具
工具报错时自动重试或返回兜底结果
在模型调用前插入额外的系统提示
记录每一步的执行日志
在敏感信息出现时阻断执行
在正式执行工具前增加人工审批

这些需求有一个共同特点：它们不是 Agent 的核心业务逻辑，但会影响执行过程。如果把它们全塞进主流程：

主流程迅速变乱
横切逻辑难以复用（日志、鉴权等需求多个 Agent 都需要）
流程控制粒度不够细
后期维护成本高

中间件的价值就在于把这些与业务无关、但与执行过程强相关的横切逻辑分离出来：

中间件 = 拦截流程 + 修改流程 + 增强流程

💡 核心功能：

功能	说明
日志与分析	追踪行为、调试、性能监控
转换	修改提示词、工具选择、输出格式
容错	重试、降级、早期终止
安全	限流、守护规则、PII 检测

3. 中间件的分类

根据来源分类：

自定义中间件：开发者自行实现
内置中间件：LangChain 提供的，共 16 个预置中间件
模型供应商定制的中间件：依赖特定模型服务

LangChain 提供的与模型供应商无关的内置中间件分为六大类别：

类型 1：成本与资源控制类 — 控成本、控配额、避免无限调用

ModelCallLimitMiddleware：限制模型调用次数
ToolCallLimitMiddleware：限制工具调用次数
SummarizationMiddleware：上下文快满时自动总结历史
ContextEditingMiddleware：裁剪上下文、清理工具调用痕迹

类型 2：稳定性与容错保障类 — 保证服务不中断、失败后自动恢复

ModelFallbackMiddleware：主模型失败时切换备用模型
ModelRetryMiddleware：模型调用失败后自动重试
ToolRetryMiddleware：工具调用失败后自动重试

类型 3：安全与合规风控类 — 让 Agent 可控、可审、合规

HumanInTheLoopMiddleware：关键工具调用前暂停，等人工审批
PIIMiddleware：检测和处理个人敏感信息

类型 4：决策增强与智能编排类 — 提升决策质量和任务拆解能力

TodoListMiddleware：给 Agent 增加任务规划和进度跟踪能力
LLMToolSelectorMiddleware：工具太多时用子模型筛选最相关的几个
SubagentMiddleware：生成子 Agent 拆分复杂任务

类型 5：执行能力扩展类 — 给 Agent 更多"手脚"

ShellToolMiddleware：给 Agent 持久 shell，执行命令
FilesystemFileSearchMiddleware：给 Agent 文件搜索能力（Glob/Grep）
FilesystemMiddleware：给 Agent 文件系统读写能力

类型 6：开发调试与测试辅助类 — 方便开发、测试、验证

LLMToolEmulatorMiddleware：用 LLM 模拟工具执行

4. SummarizationMiddleware — 摘要中间件

作用：对历史消息列表进行摘要总结，压缩上下文。

原理：在达到触发条件时，调用大模型对历史消息进行摘要，将摘要结果作为 HumanMessage 放到消息列表最前面的位置。

核心参数：

参数	说明
`model`	用于摘要的模型（字符串或对象）
`trigger`	摘要触发条件列表，任一满足即触发
`keep`	摘要时保留的原始消息（同一时间只接收一种条件）
`token_counter`	统计 token 数量的函数（默认估算）
`summary_prompt`	自定义摘要提示词（需包含 `{messages}` 占位符）
`trim_token_to_summarize`	摘要时历史消息最大 token 数（默认 4000）

trigger 条件的三种度量：

("tokens", 100) — 历史 token 累计达到 100 触发
("messages", 6) — 历史消息条数达到 6 触发
("fraction", 0.0001) — 历史 token 达到 max_input_tokens × fraction 触发

⚠️ 注意：使用 fraction 需要模型的 profile 包含 max_input_tokens，DeepSeek 模型需要手动添加。

keep 条件的三种度量：

("tokens", N) — 保留 N 个 token
("messages", N) — 保留 N 条历史消息
("fraction", F) — 保留 max_input_tokens × F 个 token

使用示例：

from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
    model=model,
    middleware=[
        SummarizationMiddleware(
            model=model,
            trigger=[
                ("tokens", 100),
                ("messages", 6),
            ],
            keep=("messages", 2),
            summary_prompt="对历史消息摘要，消息列表如下\n{messages}"
        )
    ]
)

5. HumanInTheLoopMiddleware — 人工审核中间件

作用：在工具调用前中断 Agent 运行，等待用户决策。

三种决策：

决策	说明
`approve`	同意执行
`edit`	编辑调用配置后执行
`reject`	拒绝执行

核心参数：

参数	说明
`interrupt_on`	工具名和中断策略的映射
`description_prefix`	自定义中断描述（默认 “Tool execution requires approval”）

interrupt_on 的配置方式：

interrupt_on={
    "get_weather": True,           # 所有决策都可选
    "read_email_tool": False,      # 不中断，直接执行
    "send_email_tool": {           # 精细控制
        "allowed_decisions": ["approve", "reject"],  # 不允许 edit
        "description": "发送邮件中断啦",
    },
}

使用流程：

第一次 invoke：Agent 会在需要审批的工具前中断
获取中断信息：response.get("__interrupt__", [])
传入决策并恢复：agent.invoke(Command(resume=decisions), config=config)

关键点：需要配置 checkpointer（如 InMemorySaver()），通过相同的 thread_id 加载记忆来恢复中断的 Agent。

决策格式：

# approve
{"type": "approve"}

# edit（修改参数后执行）
{"type": "edit", "edited_action": {"name": "get_weather", "args": {"city": "北京", "is_forcast": True}}}

# reject
{"type": "reject"}

决策的顺序必须和中断请求的顺序一致。

6. PIIMiddleware — 敏感信息保护中间件

作用：检测和处理对话中的个人身份信息（PII）。

核心参数：

参数	说明
`pii_type`	检测的 PII 数据类型
`strategy`	处理策略
`detector`	自定义检测函数或正则表达式
`apply_to_input`	是否在调用模型前检测（默认 True）
`apply_to_output`	是否在模型调用后检测（默认 False）
`apply_to_tool_results`	是否在工具调用后检测其输出（默认 False）

内置 PII 类型：email、credit_card、url、mac_address、ip

四种处理策略：

策略	说明	示例
`redact`	用 `[REDACTED_类型]` 替换	`[REDACTED_EMAIL]`
`mask`	用 `***` 遮蔽部分信息	`**--**-5100`
`hash`	用哈希值替代	`<email_hash:a1b2c3d4>`
`block`	检测到直接抛异常	抛出异常

使用示例 — 内置检测器：

from langchain.agents.middleware import PIIMiddleware

agent = create_agent(
    model=model,
    tools=[],
    middleware=[
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        PIIMiddleware("credit_card", strategy="mask", apply_to_input=True),
        PIIMiddleware("ip", strategy="block", apply_to_input=True),
    ]
)

自定义检测器：

import re

# 自定义检测函数
def detect_phone_number(content: str):
    return [
        {
            "text": m.group(0),
            "start": m.start(),
            "end": m.end()
        } for m in re.finditer(r"[0-9]{11}", content)
    ]

# 使用正则表达式作为 detector
PIIMiddleware("api_key", strategy="hash", detector=r"sk-[a-zA-Z0-9]+")

# **使用自**定义函数作为 detector
PIIMiddleware("phone_number", strategy="mask", detector=detect_phone_number)

自定义检测函数返回一个列表，每个元素包含 text（匹配文本）、start（起始位置）、end（结束位置）。

7. TodoListMiddleware — 任务规划中间件

作用：给 Agent 增加任务规划和进度跟踪能力。

原理：通过 write_todos 工具创建和维护待办事项列表，将计划挂在全局状态里，时刻提醒 Agent “下一步该干什么”。

核心参数：

参数	说明
`system_prompt`	自定义指导 todo 列表使用的提示词（通常不必提供）
`tool_description`	自定义 write_todos 工具的描述信息（通常不必提供）

待办事项状态：

状态	说明
`in_progress`	正在进行
`completed`	已完成
`pending`	待执行

适用场景判断：

任务是否需要拆解？
  ├── 否（问答、翻译、单次函数调用）──> 不需要，浪费算力
  └── 是（多文件工程）
      └── 步骤是否多变且需要应对失败？
          ├── 否（步骤完全固定 A→B→C）──> 传统 LangGraph **线性节点**即可
          └── 是（AI 需要边做边调计划）──> 引入 TodoListMiddleware

使用示例：

from langchain.agents.middleware import TodoListMiddleware

agent = create_agent(
    model=model,
    tools=[list_files, read_file, write_file, run_tests],
    middleware=[TodoListMiddleware()],
    system_prompt="你是一个代码修复助手。遇到多步骤任务时，先使用 write_todos 制定待办事项..."
)

Agent 会在执行过程中自动调用 write_todos 工具，更新状态中的 todos 字段。最终结果可以通过 final_state["todos"] 查看。

8. 其它内置中间件速览

ModelCallLimitMiddleware — 限制模型调用次数

参数	说明
`thread_limit`	每个线程最多调用次数
`run_limit`	每次运行最多调用次数
`exit_behavior`	达到限制后的行为：`end`（优雅退出）、`error`（抛异常）、`continue`（继续运行）

需要配置 checkpointer（thread_limit 依赖线程记忆）。

ToolCallLimitMiddleware — 限制工具调用次数

参数同 ModelCallLimitMiddleware，但限制的是工具调用。可以避免 Agent 陷入无限循环。

ModelFallbackMiddleware — 模型故障转移

from langchain.agents.middleware import ModelFallbackMiddleware

middleware=[
    ModelFallbackMiddleware(
        fallback_models=[
            init_chat_model("openai:gpt-4o-mini"),
            init_chat_model("anthropic:claude-3-haiku")
        ]
    )
]

主模型失败时，依次尝试备用模型。

LLMToolSelectorMiddleware — 智能工具筛选

参数	说明
`model`	用于工具筛选的子模型
`max_tools`	限定可以调用的工具总数
`always_include`	指定的工具不被计数（始终保留）

当工具太多时（如 100 个），用子模型筛选最相关的几个交给主模型。

ToolRetryMiddleware — 工具调用重试

基于指数退避算法。

参数	说明
`max_retries`	最大重试次数（不含初始调用，总调用 = 1 + max_retries）
`backoff_factor`	指数退避因子
`initial_delay`	初始等待时间
`max_delay`	最大等待延迟上限
`jitter`	是否开启抖动（默认 True）
`retry_on`	仅对指定异常类型重试
`on_failure`	达到最大重试仍失败时的行为：`continue`/`error`

指数退避等待时间公式：delay = min(initial_delay * (backoff_factor ** retry_number), max_delay)

jitter（抖动）的作用：

jitter=True（默认）：重试间隔加入随机性，防止高并发下的"惊群效应"
jitter=False：重试间隔严格固定，适合本地调试

惊群效应：1000 个请求同时失败后，如果都严格等待相同时间重试，会在同一时刻再次冲击服务器，形成恶性循环。抖动让重试时间错开，削峰填谷。

ModelRetryMiddleware — 模型调用重试

参数和策略同 ToolRetryMiddleware，但作用于模型调用。on_failure 支持 continue（将错误信息塞回对话历史让 Agent 继续）和 error（抛异常）。

LLMToolEmulatorMiddleware — 工具模拟

用 LLM 模拟工具执行结果，适合工具尚未开发完成时的测试。

from langchain.agents.middleware import LLMToolEmulator

middleware=[LLMToolEmulator(model=model_in)]

ContextEditingMiddleware — 上下文编辑

通过裁剪发送给模型的消息列表来控制成本。

from langchain.agents.middleware import ContextEditingMiddleware, ClearToolUsesEdit

middleware=[
    ContextEditingMiddleware(
        edits=[
            ClearToolUsesEdit(trigger=50, keep=0),
        ],
    )
]

ClearToolUsesEdit 会在 token 数量达到 trigger 时，清理工具调用痕迹，只保留 keep 个 token 的内容。不修改实际消息列表，只修改发送给模型的内容。

效果：多轮对话中 input_tokens 明显降低。

FilesystemFileSearchMiddleware — 文件搜索

给 Agent 添加 Glob 和 Grep 文件搜索能力。

from langchain.agents.middleware import FilesystemFileSearchMiddleware

middleware=[
    FilesystemFileSearchMiddleware(
        root_path="../todo_workspace",
        use_ripgrep=True,
        max_file_size_mb=10
    )
]

9. 多个中间件的执行顺序 — 洋葱模型

非常重要：多个中间件的书写顺序非常重要。

执行规则：

before_* 钩子：从前到后执行（正序）
after_* 钩子：从后往前执行（逆序）
wrap_* 钩子：洋葱架构，前面的包裹后面的

图示：

中间件1 → 中间件2 → 中间件3 → [模型/工具] → 中间件3 → 中间件2 → 中间件1
  before      before      before                   after      after      after

这就是"洋葱模型"：外层先进后出。

组合示例：

agent = create_agent(
    model=model,
    tools=[get_weather, get_news],
    middleware=[
        PIIMiddleware(strategy="redact"),                    # 1. 最先检查 PII
        ModelCallLimitMiddleware(run_limit=10),              # 2. 限制调用次数
        SummarizationMiddleware(max_tokens_before_summary=500),  # 3. 总结历史
        ToolRetryMiddleware(max_retries=3),                  # 4. 重试工具
    ]
)

10. 自定义中间件 — Hook 函数体系

LangChain 暴露了六种 Hook 函数，分为两类：

Node-style hooks（节点风格） — 在流程特定节点运行，适合顺序逻辑：

Hook	触发时机	典型场景
`before_agent`	Agent 开始运行之前	初始化、权限检查
`before_model`	模型调用之前	消息修剪、PII 脱敏、输入验证
`after_model`	模型调用之后	输出验证、格式化响应、状态更新
`after_agent`	Agent 流程全部完成后	清理资源、最终日志

Wrap-style hooks（包装风格） — 在调用前后运行，适合控制流：

Hook	触发时机	典型场景
`wrap_model_call`	包裹模型调用	重试、缓存、修改系统提示
`wrap_tool_call`	包裹工具调用	监控、重试、修改工具执行

11. Node-style hooks 的实现

方式 1：装饰器实现

from langchain.agents.middleware import before_model, after_model, before_agent, after_agent, AgentState
from langgraph.runtime import Runtime
from typing import Any

@before_model
def before_model_middleware(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    state["messages"][-1].content += " -> before_model <- "
    return None

@after_model
def after_model_middleware(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    state["messages"][-1].content += " -> after_model <- "
    return None

agent = create_agent(
    model=model,
    middleware=[before_model_middleware, after_model_middleware]
)

方式 2：类实现

from langchain.agents.middleware import AgentMiddleware, AgentState
from langgraph.runtime import Runtime
from typing import Any

class MyMiddleware(AgentMiddleware):
    def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        state["messages"][-1].content += " -> before_model <- "
        return None

    def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        state["messages"][-1].content += " -> after_model <- "
        return None

agent = create_agent(
    model=model,
    middleware=[MyMiddleware()]
)

关键规则：

类必须继承 AgentMiddleware
方法名固定（before_model、after_model 等）
类名随意

两种方式的统一：装饰器底层会创建一个 AgentMiddleware 子类的实例，重写对应方法。所以两者本质相同。

12. Node-style hooks 的参数和返回值

两个参数：

参数	说明
`state`	`AgentState` 实例，维护 Agent 运行状态（包括消息列表）
`runtime`	`Runtime` 实例，维护运行时上下文（长期记忆等）

返回值：

返回值	说明
`None`	不修改状态，继续流程
字典	更新状态
`{"jump_to": "..."}`	控制流程跳转

jump_to 目标：

目标	说明
`"__end__"`	结束 Agent
`"tools"`	跳到工具节点
`"model"`	跳到模型节点

13. can_jump_to — 流程控制参数

Hook 函数可以改变 Agent 的运行轨迹。can_jump_to 决定了钩子函数可以跳转到哪些位置。

可取值：

值	说明
`end`	跳转至流程末尾，终止
`tools`	跳转至工具节点
`model`	跳转至模型节点

装饰器写法：

@before_model(can_jump_to=["tools"])
def force_tool_first(state, runtime):
    # 检测到关键词，直接跳到工具节点，跳过模型思考
    if "direct tool" in state["messages"][-1].content.lower():
        fake_tool_call = AIMessage(
            content="人工构造的消息",
            tool_calls=[{"name": "get_news", "args": {}, "id": "call_001"}],
        )
        return {"messages": [fake_tool_call], "jump_to": "tools"}
    return None

@before_model(can_jump_to=["end"])
def overflow_processor(state, runtime):
    # 检测到溢出标记，直接终止
    if "overflow" in state["messages"][-1].content:
        return {"messages": [AIMessage("上下文窗口溢出，终止")], "jump_to": "end"}
    return None

类写法 — 需要用 @hook_config 装饰器传参：

class MyMiddleware(AgentMiddleware):
    @hook_config(can_jump_to=["tools", "end"])
    def before_model(self, state, runtime):
        ...

    @hook_config(can_jump_to=["model"])
    def after_model(self, state, runtime):
        ...

14. Wrap-style hooks 的实现

wrap_model_call — 包裹模型调用

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

@wrap_model_call
def wrap_model_call_middleware(
    request: ModelRequest,      # 包含发送给模型的所有请求数据
    handler: Callable[[ModelRequest], ModelResponse]  # 实际调用模型的函数
) -> ModelResponse | None:
    # 模型调用前：修改请求
    request.messages[-1].content += " -> before <- "
    
    # 调用模型
    response = handler(request)
    
    # 模型调用后：修改响应
    response.result[0].content += " -> after <- "
    
    return response

使用场景：

重试逻辑：在 handler 外加 try/except + 指数退避
响应缓存：用 request 内容生成 cache_key，命中则直接返回缓存
修改系统提示：通过 request.override(system_message=new_msg) 动态修改

wrap_tool_call — 包裹工具调用

from langchain.agents.middleware import wrap_tool_call
from langchain.tools.tool_node import ToolCallRequest
from typing import Callable

@wrap_tool_call
def monitor_tool(
    request: ToolCallRequest,
    handler: Callable[[ToolCallRequest], ToolMessage | Command]
) -> ToolMessage | Command:
    tool_name = request.tool_call["name"]
    print(f"🔧 开始执行工具：{tool_name}")
    start_time = time.time()
    try:
        result = handler(request)
        print(f"✅ 耗时：{time.time() - start_time:.2f}秒")
        return result
    except Exception as e:
        print(f"❌ 失败：{e}")
        raise

15. 装饰器 vs 类写法的选择

场景	推荐	原因
单个 Hook	装饰器	最简单，快速原型
多个 Hook	类	集中管理，结构清晰
复杂配置	类	类的 `__init__` 传参更自然，可自省
跨项目复用	类	可实例化、可封装、可测试

总结：单钩子用装饰器，多钩子/复杂配置/跨项目复用用类。

16. Hook 函数执行顺序（重要）

三种 Hook 的执行顺序规则：

before_* 钩子：从前到后执行（正序）—— 按中间件列表顺序
after_* 钩子：从后往前执行（逆序）—— 按中间件列表逆序
wrap_* 钩子：洋葱架构 —— 前面的包裹后面的

执行顺序只和创建 Agent 时传递中间件的顺序有关，和定义顺序无关。

示例输出：

# middleware=[mw1, mw2, mw3] 时

before_model-1
before_model-2
before_model-3
wrap_model-before-1
wrap_model-before-2
wrap_model-before-3
[模型调用]
wrap_model-after-3
wrap_model-after-2
wrap_model-after-1
after_model-3
after_model-2
after_model-1

模块 2：带教式理解

1. 为什么中间件叫"钩子函数"

想象一条流水线：原料 → 加工 → 检测 → 包装 → 出厂。主流程就是流水线的传送带。

如果想在"加工"后加一个"额外检查"，你不应该拆掉传送带重新组装——你只需要在加工后"挂上"一个检查器。

这个"挂上去"的检查器就是钩子（Hook）。它在流水线的特定位置自动被触发，你不需要在主流程代码里手动调用它。

LangChain 的中间件就是在 Agent 执行流程的特定位置（模型调用前、模型调用后、工具调用前、工具调用后）预留了"插槽"，你把逻辑挂上去就行。

2. 为什么 trigger 用列表而 keep 只接受一种条件

trigger 是"或"关系——任一条件满足就触发摘要，所以是列表。

keep 是"精确指定"——摘要时保留多少原始消息，只能有一种度量方式，所以只接收一种条件。

3. HumanInTheLoopMiddleware 为什么需要 checkpointer

因为 Agent 在工具调用前被中断了，整个执行状态（包括 AI 决定调用哪些工具、传什么参数）需要被保存下来。当你传入决策恢复执行时，Agent 需要从上次中断的位置继续——这就需要记忆。

checkpointer 就是这个记忆存储，thread_id 就是会话标识。同一个 thread_id 的调用会共享记忆，从而实现"中断 → 恢复"。

4. PIIMiddleware 的 apply_to_input 为什么默认 True 而 apply_to_output 默认 False

PII 检测的主要目的是避免把敏感信息发送给模型服务，防止信息泄露到外部 API。

apply_to_input=True：在消息发给模型之前检测脱敏——这是最重要的防线
apply_to_output=False：模型返回的内容通常不包含用户原始的 PII，所以默认不检测

如果你担心模型自己生成了敏感信息（比如模型有记忆能力），可以开启 apply_to_output=True。

5. TodoListMiddleware 为什么要通过工具调用来维护 todos

因为 LLM 的"记忆"有限。如果不把计划外化为一个显式的列表，LLM 在执行到第 3 步时可能就忘了最初的目标。

通过 write_todos 工具，计划被写入 Agent 的全局状态（state["todos"]），每一步执行时 LLM 都能看到当前进度，相当于给它一个"备忘录"。

这就像人类工作时写 CheckList——不是因为你记不住，而是写下来后更不容易遗漏，而且可以随时回顾。

6. 指数退避为什么不能直接固定间隔重试

如果服务器因为流量过大崩溃了，所有客户端都每秒重试一次，等于在服务器最脆弱的时候又给它持续 DDoS。

指数退避让每次重试等待时间加倍（1s → 2s → 4s → 8s…），给服务器恢复的时间。

但光有指数退避还不够——如果 1000 个请求同时失败，它们仍然会在 1s、2s、4s 的精准时间点同时重试。jitter（抖动）通过加入随机性，让这 1000 个请求的重试时间分散开，避免"惊群效应"。

7. 洋葱模型为什么 before 正序、after 逆序

想象你穿衣服：先穿内衣（中间件1 before），再穿衬衫（中间件2 before），再穿外套（中间件3 before）。脱衣服时反过来：先脱外套（中间件3 after），再脱衬衫（中间件2 after），最后脱内衣（中间件1 after）。

这就是洋葱模型——外层先进入、最后退出。before 是"穿"的过程（正序），after 是"脱"的过程（逆序）。

对于 wrap 类型，更明显：handler(request) 是"核心操作"，wrap_before 是进入核心前的操作，wrap_after 是离开核心后的操作。最外层的 wrap 最先进入、最后离开。

8. can_jump_to 为什么需要显式声明

安全考虑。如果不限制跳转目标，一个 before_model 钩子可能跳到任意位置，导致不可预期的行为。

通过 can_jump_to=["tools", "end"] 显式声明，框架可以：

在运行时校验跳转目标是否合法
在开发阶段提供更好的错误提示
防止意外的流程控制

9. 装饰器和类底层为什么是统一的

以 @after_model 为例，装饰器底层代码等价于：

# 创建一个 AgentMiddleware 子类，重写 after_model 方法
return type(
    "after_model_middleware",       # 类名
    (AgentMiddleware,),             # 继承 AgentMiddleware
    {
        "state_schema": AgentState,
        "tools": [],
        "after_model": func(state, runtime),  # 重写方法
    },
)()  # 实例化

所以装饰器最终返回的也是一个 AgentMiddleware 子类对象，和类写法本质完全相同。装饰器只是语法糖。

10. LLMToolSelectorMiddleware 的 max_tools=0 是什么意思

max_tools=0 表示"不限制筛选的工具数量"，但 always_include 指定的工具始终保留。

这看起来矛盾（不限制为什么还要用这个中间件），但实际上它的作用是只保留 always_include 指定的工具，屏蔽其他所有工具。

这在测试场景中很有用：你想验证 Agent 在只有特定工具时是否能正确完成任务。

这一章结束后，你应该能自己回答

1、中间件解决的核心问题是什么？

中间件解决的是"横切逻辑与业务逻辑耦合"的问题。没有中间件时，日志、鉴权、重试、风控等逻辑会被写死在 Agent 主流程中，导致主流程臃肿、代码难以复用、维护成本高。中间件把这些与业务无关但与执行过程强相关的逻辑分离出来，实现"拦截流程、修改流程、增强流程"。

2、LangChain 内置中间件分为哪六大类别？

成本与资源控制类（ModelCallLimit、ToolCallLimit、Summarization、ContextEditing）
稳定性与容错保障类（ModelFallback、ModelRetry、ToolRetry）
安全与合规风控类（HumanInTheLoop、PII）
决策增强与智能编排类（TodoList、LLMToolSelector、Subagent）
执行能力扩展类（ShellTool、FilesystemFileSearch、Filesystem）
开发调试与测试辅助类（LLMToolEmulator）

3、HumanInTheLoopMiddleware 的完整使用流程是什么？

① 创建 Agent 时配置 checkpointer 和 HumanInTheLoopMiddleware（设置 interrupt_on 映射）
② 第一次 invoke，Agent 会在需要审批的工具前中断
③ 从 response 的 __interrupt__ 字段获取中断信息（action_requests 和 review_configs）
④ 构造 decisions 字典，每个决策是 approve/edit/reject 之一
⑤ 用 agent.invoke(Command(resume=decisions), config=config) 恢复执行，config 必须使用相同的 thread_id

4、PIIMiddleware 的四种处理策略分别是什么？

redact（用 [REDACTED_类型] 替换，适合日志清洗）、mask（用 *** 遮蔽部分信息，适合前端显示）、hash（用哈希值替代，适合分析统计）、block（直接抛异常，适合极高安全要求场景）。

5、自定义中间件的六种 Hook 函数分别是什么？分为哪两类？

Node-style hooks（节点风格，在特定节点运行）：before_agent、before_model、after_model、after_agent。Wrap-style hooks（包装风格，在调用前后运行）：wrap_model_call、wrap_tool_call。

6、多个中间件的执行顺序规则是什么？

before_* 钩子从前到后执行（正序，按列表顺序）；after_* 钩子从后往前执行（逆序）；wrap_* 钩子是洋葱架构（前面的包裹后面的）。执行顺序只和创建 Agent 时传递中间件的顺序有关，和定义顺序无关。

7、装饰器和类写法分别在什么场景下使用？

单个 Hook + 逻辑简单 → 装饰器（快速原型）。多个 Hook 组合 / 复杂配置 / 跨项目复用 → 类写法（结构清晰、可维护性好、可自省）。两者底层实现统- 装饰器最终也会创建 AgentMiddleware 子类实例。

小测试

1. SummarizationMiddleware 的 trigger 和 keep 参数有什么区别？

trigger 是触发摘要的条件列表（支持 tokens/messages/fraction 三种度量），任一满足即触发，是"或"关系。keep 是摘要时保留的原始消息条件，同一时间只接收一种条件。trigger 决定"什么时候摘要"，keep 决定"摘要时保留多少原始消息"。

2. ModelCallLimitMiddleware 的 exit_behavior 有哪些选项？分别是什么效果？

end（优雅退出，Agent 返回一条提示消息如 “Model call limits exceeded: run limit (3/3)”）、error（抛出 ModelCallLimitExceededError 异常）、continue（继续运行 Agent，但不再调用模型）。thread_limit 需要 checkpointer 支持。

3. ToolRetryMiddleware 的 jitter 参数设为 True 和 False 有什么区别？

jitter=True（默认）在重试等待时间中加入随机性，防止高并发下的"惊群效应"，适合生产环境。jitter=False 使重试间隔严格固定（1s、2s、4s、8s、10s…），适合本地调试验证重试逻辑是否生效。

4. wrap_model_call 的 handler 参数是什么？

handler 是一个 Callable 函数，代表"下一个中间件或最终的模型调用服务"。调用 handler(request) 会真正执行模型调用（或流转到下一个 wrap 中间件），产生真实的 Token 消耗并等待响应。你可以在 handler 前后插入自定义逻辑，也可以不调用 handler 来阻止模型调用（如缓存命中时直接返回）。

5. before_model 钩子返回 `{"jump_to": "tools"}` 会发生什么？

Agent 会跳过本次模型调用，直接进入工具执行节点。通常配合 can_jump_to=["tools"] 使用，在 before_model 中构造一个带 tool_calls 的 AIMessage，然后返回 {"messages": [fake_tool_call], "jump_to": "tools"}，让系统"误以为"模型已经决定调用工具，直接执行。

LangChain 08 - 中间件学习导航与实战

模块 1：你只需要记住这些

1. 什么是中间件

2. 为什么需要中间件

3. 中间件的分类

4. SummarizationMiddleware — 摘要中间件

5. HumanInTheLoopMiddleware — 人工审核中间件

6. PIIMiddleware — 敏感信息保护中间件

7. TodoListMiddleware — 任务规划中间件

8. 其它内置中间件速览

9. 多个中间件的执行顺序 — 洋葱模型

10. 自定义中间件 — Hook 函数体系

11. Node-style hooks 的实现

12. Node-style hooks 的参数和返回值

13. can_jump_to — 流程控制参数

14. Wrap-style hooks 的实现

15. 装饰器 vs 类写法的选择

16. Hook 函数执行顺序（重要）

模块 2：带教式理解

1. 为什么中间件叫"钩子函数"

2. 为什么 trigger 用列表而 keep 只接受一种条件

3. HumanInTheLoopMiddleware 为什么需要 checkpointer

4. PIIMiddleware 的 apply_to_input 为什么默认 True 而 apply_to_output 默认 False

5. TodoListMiddleware 为什么要通过工具调用来维护 todos

6. 指数退避为什么不能直接固定间隔重试

7. 洋葱模型为什么 before 正序、after 逆序

8. can_jump_to 为什么需要显式声明

9. 装饰器和类底层为什么是统一的

10. LLMToolSelectorMiddleware 的 max_tools=0 是什么意思

这一章结束后，你应该能自己回答

1、中间件解决的核心问题是什么？

2、LangChain 内置中间件分为哪六大类别？

3、HumanInTheLoopMiddleware 的完整使用流程是什么？

4、PIIMiddleware 的四种处理策略分别是什么？

5、自定义中间件的六种 Hook 函数分别是什么？分为哪两类？

6、多个中间件的执行顺序规则是什么？

7、装饰器和类写法分别在什么场景下使用？

小测试

1. SummarizationMiddleware 的 trigger 和 keep 参数有什么区别？

2. ModelCallLimitMiddleware 的 exit_behavior 有哪些选项？分别是什么效果？

3. ToolRetryMiddleware 的 jitter 参数设为 True 和 False 有什么区别？

4. wrap_model_call 的 handler 参数是什么？

5. before_model 钩子返回 {"jump_to": "tools"} 会发生什么？

5. before_model 钩子返回 `{"jump_to": "tools"}` 会发生什么？