内置中间件

LangChain 提供了 18 个内置中间件，覆盖安全合规、成本控制、流程管理、容错重试、工具增强等横切关注点。本章逐一讲解三个核心中间件（PII / Summarization / HITL）的完整 API 和实战示例，末尾速览其余 15 个内置中间件，重点展开容错相关的 Model Retry / Tool Retry / Model Fallback。

先修知识

已阅读中间件概览
了解 Agent 实战指南中的 create_agent 用法

PIIMiddleware - PII 检测与处理

PIIMiddleware 用于在 Agent 的请求-响应流程中自动检测和处理个人身份信息（Personally Identifiable Information）。它支持对用户输入和模型输出的双向检测。

基本用法

python

import os
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.agents.middleware import PIIMiddleware

model = init_chat_model(os.environ["LLM_MODEL"])

agent = create_agent(
    model=model,
    tools=[search, send_email],
    middleware=[
        PIIMiddleware("email", strategy="redact"),
    ],
)

构造参数

参数	类型	默认值	说明
`pii_type`	`str`	必填	PII 类型标识，如 `"email"`、`"phone_number"`、`"ssn"`
`strategy`	`str`	`"redact"`	处理策略：`"redact"` / `"block"` / `"mask"` / `"hash"`
`detector`	`str` (regex)	内置正则	自定义正则表达式，覆盖内置检测规则
`apply_to_input`	`bool`	`True`	是否检测用户输入
`apply_to_output`	`bool`	`False`	是否检测模型输出
`apply_to_tool_results`	`bool`	`False`	是否检测工具返回结果

策略详解

strategy: "redact" - 脱敏替换

将匹配到的 PII 信息替换为占位符，请求继续执行：

python

# 输入: "我的邮箱是 test@example.com"
# 脱敏后: "我的邮箱是 ***@***.***"

PIIMiddleware("email", strategy="redact")

适用于：需要保留对话上下文但隐藏具体信息的场景。

strategy: "block" - 拦截请求

发现 PII 后立即拦截请求，返回错误信息，不继续执行后续 Middleware 或 Agent 调用：

python

# 输入: "我的身份证号是 110101199001011234"
# 结果: 请求被拦截，返回错误提示

PIIMiddleware("ssn", strategy="block")

适用于：严格合规要求，任何包含特定 PII 的请求都不应被处理。

strategy: "mask" / "hash"

mask 部分遮盖（如保留前几位后几位），hash 将 PII 替换为哈希值。适用于需要在日志中引用但不泄露原始值的场景。

自定义正则检测器

内置检测器覆盖了常见的 PII 类型。当内置规则不满足需求时，可以通过 detector 参数提供自定义正则：

python

# 检测中国大陆手机号
PIIMiddleware(
    "phone_number",
    detector=r"(?:(?:\+|00)86)?1[3-9]\d{9}",
    strategy="redact",
)

# 检测中国大陆身份证号
PIIMiddleware(
    "id_card",
    detector=r"[1-9]\d{5}(?:19|20)\d{2}(?:0[1-9]|1[0-2])(?:0[1-9]|[12]\d|3[01])\d{3}[\dXx]",
    strategy="block",
)

apply_to_output 与 apply_to_tool_results 选项

默认情况下，PIIMiddleware 只检测用户输入。如需检测模型输出或工具返回结果：

python

# 同时检测用户输入和模型输出
PIIMiddleware("email", strategy="redact", apply_to_input=True, apply_to_output=True)

# 检测工具返回结果中的 PII（如搜索 API 返回了他人邮箱）
PIIMiddleware("email", strategy="redact", apply_to_tool_results=True)

典型场景：用户主动提供自己的邮箱用于发送邮件，不应拦截输入，但需要确保模型不会在响应中泄露其他用户的邮箱。

完整示例：多类型 PII 检测

python

import os
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.agents.middleware import PIIMiddleware

def search_contacts(name: str) -> str:
    """搜索联系人信息"""
    return "张三, test@example.com, 13800138000"

def send_message(to: str, content: str) -> str:
    """发送消息"""
    return f"消息已发送至 {to}"

model = init_chat_model(os.environ["LLM_MODEL"])

agent = create_agent(
    model=model,
    tools=[search_contacts, send_message],
    middleware=[
        # 邮箱：脱敏处理（允许流程继续，但隐藏具体地址）
        PIIMiddleware("email", strategy="redact", apply_to_input=True),

        # 手机号：自定义正则，拦截请求
        PIIMiddleware(
            "phone_number",
            detector=r"(?:(?:\+|00)86)?1[3-9]\d{9}",
            strategy="block",
        ),

        # 身份证号：拦截请求
        PIIMiddleware(
            "id_card",
            detector=(
                r"[1-9]\d{5}(?:19|20)\d{2}"
                r"(?:0[1-9]|1[0-2])"
                r"(?:0[1-9]|[12]\d|3[01])"
                r"\d{3}[\dXx]"
            ),
            strategy="block",
        ),
    ],
)

# 运行 Agent
result = agent.invoke("帮我查一下张三的联系方式")
# 邮箱会被脱敏显示，手机号会被拦截

SummarizationMiddleware - 自动对话摘要

当对话历史过长时，SummarizationMiddleware 会自动调用 LLM 生成摘要，用摘要替换早期消息，从而控制 token 用量和成本。

基本用法

python

from langchain.agents.middleware import SummarizationMiddleware

SummarizationMiddleware(
    model=model,
    trigger=("tokens", 4000),  # 超过 4000 token 时触发摘要
    keep=("messages", 20),     # 保留最近 20 条消息不压缩
)

构造参数

参数	类型	默认值	说明
`model`	`BaseChatModel`	必填	用于生成摘要的模型实例
`trigger`	`tuple`	`None`	触发条件：`("tokens", N)` / `("messages", N)` / `("fraction", 0.5)` 或列表
`keep`	`tuple`	`("messages", 20)`	保留策略：`("messages", N)` / `("tokens", N)` / `("fraction", f)`
`token_counter`	callable	内置	自定义 token 计数函数
`summary_prompt`	`str`	内置	自定义摘要提示词
`trim_tokens_to_summarize`	`int`	`4000`	摘要时裁剪的最大 token 数

trigger 详解

trigger 决定何时触发摘要，支持多种维度：

python

# 按 token 数触发
trigger=("tokens", 4000)

# 按消息数触发
trigger=("messages", 50)

# 按上下文窗口占比触发（如占满 50% 时触发）
trigger=("fraction", 0.5)

# 多条件组合（任一满足即触发）
trigger=[("tokens", 4000), ("messages", 50)]

keep 详解

keep 决定摘要后保留哪些消息不被压缩：

python

# 保留最近 20 条消息
keep=("messages", 20)

# 保留最近 2000 token 的消息
keep=("tokens", 2000)

# 保留最近 30% 的消息
keep=("fraction", 0.3)

工作原理

每次模型调用前，统计当前对话历史的 token / 消息数
如果超过 trigger 阈值，触发摘要
使用指定模型将早期消息压缩为一条摘要消息
用摘要 + keep 保留的近期消息替换原始历史
将压缩后的消息列表传递给模型

前端类比

这类似于前端虚拟列表（Virtual List）的思路--不渲染所有 DOM 节点，只保留视口内的元素。SummarizationMiddleware 不保留所有历史消息，只保留"视口"（最近对话）加一个"摘要"（早期内容的压缩表示）。

不过需要注意：摘要会丢失细节信息，这是有损压缩，不像虚拟列表可以无损还原。

完整示例

python

import os
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.agents.middleware import SummarizationMiddleware

def search_docs(query: str) -> str:
    """搜索文档库"""
    return f"关于 '{query}' 的搜索结果..."

model = init_chat_model(os.environ["LLM_MODEL"])

agent = create_agent(
    model=model,
    tools=[search_docs],
    middleware=[
        SummarizationMiddleware(
            model=model,
            trigger=("tokens", 4000),  # 超过 4000 token 时自动摘要
            keep=("messages", 20),     # 保留最近 20 条消息
        ),
    ],
)

# 长对话场景
config = {"configurable": {"thread_id": "docs-1"}}
result = agent.invoke(
    {"messages": [{"role": "user", "content": "帮我查一下 Python 异步编程的最佳实践"}]},
    config=config,
)
# ... 多轮对话后，早期消息会被自动摘要
result = agent.invoke(
    {"messages": [{"role": "user", "content": "再查一下 asyncio 和 trio 的对比"}]},
    config=config,
)
# 此时如果历史 token 超过 4000，早期对话会被压缩

使用建议

摘要模型选择：可以使用与 Agent 相同的模型，也可以使用更轻量的模型降低摘要成本
阈值设置：根据模型的上下文窗口大小和对话长度预期来设定，通常设为上下文窗口的 30%-50%
信息丢失：摘要是有损压缩，对于需要精确回溯的场景（如法律咨询），应额外保存完整对话记录

旧版

max_tokens_before_summary、max_tokens、max_messages 参数已弃用。使用 trigger= 和 keep= 替代。

python

# 旧版（已弃用）
SummarizationMiddleware(model=..., max_tokens_before_summary=500)

# 当前写法
SummarizationMiddleware(model=model, trigger=("tokens", 500), keep=("messages", 20))

HumanInTheLoopMiddleware - 人工审批

HumanInTheLoopMiddleware 在 Agent 执行敏感工具调用前暂停流程，等待人工审批。这是实现合规审计和风险控制的关键 Middleware。

完整的 HITL 流程（中断与恢复、approve/edit/reject 语义、Command 恢复）请参考人机协作 HITL。

基本用法

python

from langchain.agents.middleware import HumanInTheLoopMiddleware

HumanInTheLoopMiddleware(
    interrupt_on={
        "send_email": {
            "allowed_decisions": ["approve", "edit", "reject"],
        },
    },
)

构造参数

参数	类型	说明
`interrupt_on`	`dict`	需要人工审批的工具配置，key 为工具名
`description_prefix`	`str`	中断描述前缀，默认 `"Tool execution requires approval"`

interrupt_on 的值可以是 True（启用全部决策）或一个配置字典：

python

# 简写：启用审批，默认允许 approve / edit / reject
interrupt_on={"send_email": True}

# 完整配置：指定允许的决策
interrupt_on={
    "send_email": {"allowed_decisions": ["approve", "edit", "reject"]},
}

allowed_decisions 支持三种决策：

决策	含义
`"approve"`	批准执行，不修改参数
`"edit"`	允许审批者修改工具参数后再执行
`"reject"`	拒绝执行，返回错误给 Agent

注意

HumanInTheLoopMiddleware 的参数是 interrupt_on=，不是 tools=。传 tools= 会抛出 TypeError。恢复执行需要使用 Command(resume=...)，不能用普通用户消息--详见 HITL 页面。

完整示例

python

import os
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver

def read_database(query: str) -> str:
    """查询数据库（只读，低风险）"""
    return f"查询结果: {query}"

def update_user_profile(user_id: str, data: dict) -> str:
    """更新用户资料（写操作，中风险）"""
    return f"用户 {user_id} 资料已更新"

def delete_account(user_id: str) -> str:
    """删除用户账户（不可逆，高风险）"""
    return f"用户 {user_id} 账户已删除"

def send_bulk_email(recipients: list, content: str) -> str:
    """群发邮件（影响范围大，高风险）"""
    return f"邮件已发送给 {len(recipients)} 位收件人"

model = init_chat_model(os.environ["LLM_MODEL"])

agent = create_agent(
    model=model,
    tools=[read_database, update_user_profile, delete_account, send_bulk_email],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                # 更新操作：允许批准或拒绝
                "update_user_profile": {
                    "allowed_decisions": ["approve", "reject"],
                },
                # 删除操作：只允许批准或拒绝（不允许编辑，因为操作本身不应修改）
                "delete_account": {
                    "allowed_decisions": ["approve", "reject"],
                },
                # 群发邮件：允许编辑内容后发送
                "send_bulk_email": {
                    "allowed_decisions": ["approve", "edit", "reject"],
                },
            },
        ),
    ],
    checkpointer=InMemorySaver(),  # HITL 需要 Checkpointer 保存中断状态
)

# read_database 不在 interrupt_on 中，直接执行
# update_user_profile / delete_account / send_bulk_email 会触发人工审批
config = {"configurable": {"thread_id": "ops-1"}}
result = agent.invoke(
    {"messages": [{"role": "user", "content": "帮我清理过期用户并通知他们"}]},
    config=config,
)

设计建议

按风险分级：只对中高风险工具启用审批，低风险的只读操作无需审批
精简 allowed_decisions：不可逆操作（如删除）不提供 "edit" 选项，避免误操作
结合 PII 检测：将 PIIMiddleware 放在 HumanInTheLoopMiddleware 前面，确保审批者看到的是脱敏后的内容

实战：组合使用内置 Middleware

下面是一个完整的生产级配置示例，展示三个核心中间件的协同工作：

python

import os
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.agents.middleware import (
    PIIMiddleware,
    SummarizationMiddleware,
    HumanInTheLoopMiddleware,
)
from langgraph.checkpoint.memory import InMemorySaver

def search_customers(query: str) -> str:
    """搜索客户信息"""
    return "客户: 李四, li4@company.com, 13912345678"

def send_notification(customer_id: str, message: str) -> str:
    """向客户发送通知"""
    return f"通知已发送给客户 {customer_id}"

def update_crm_record(customer_id: str, field: str, value: str) -> str:
    """更新 CRM 记录"""
    return f"已更新客户 {customer_id} 的 {field}"

model = init_chat_model(os.environ["LLM_MODEL"])

agent = create_agent(
    model=model,
    tools=[search_customers, send_notification, update_crm_record],
    middleware=[
        # 第一层：PII 脱敏（邮箱和手机号）
        PIIMiddleware("email", strategy="redact"),
        PIIMiddleware(
            "phone_number",
            detector=r"(?:(?:\+|00)86)?1[3-9]\d{9}",
            strategy="redact",
        ),

        # 第二层：对话摘要（控制 token 用量）
        SummarizationMiddleware(
            model=model,
            trigger=("tokens", 4000),
            keep=("messages", 20),
        ),

        # 第三层：人工审批（敏感操作）
        HumanInTheLoopMiddleware(
            interrupt_on={
                "send_notification": {
                    "allowed_decisions": ["approve", "edit", "reject"],
                },
                "update_crm_record": {
                    "allowed_decisions": ["approve", "reject"],
                },
            },
        ),
    ],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "crm-1"}}
result = agent.invoke(
    {"messages": [{"role": "user", "content": "帮我查一下李四的信息，然后通知他账户即将到期"}]},
    config=config,
)

执行流程：

用户请求
  -> PIIMiddleware: 脱敏邮箱和手机号
    -> SummarizationMiddleware: 检查是否需要摘要
      -> HumanInTheLoopMiddleware: send_notification 需要审批
        -> Agent / LLM 处理
      <- HumanInTheLoopMiddleware: 返回审批结果
    <- SummarizationMiddleware: 更新对话历史
  <- PIIMiddleware: 检测响应中的 PII
最终响应

其它内置中间件速览

除上述三个核心中间件外，LangChain 还提供了 15 个内置中间件。以下按功能分组速览，重点展开容错相关的三个。

容错与重试

ModelRetryMiddleware - 模型调用重试

模型调用失败时自动重试，支持配置重试次数和退避策略：

python

from langchain.agents.middleware import ModelRetryMiddleware

agent = create_agent(
    model=model,
    tools=[search],
    middleware=[
        ModelRetryMiddleware(max_retries=3),  # 最多重试 3 次
    ],
)

适用场景：网络不稳定、API 限流、临时性 Provider 故障。

ToolRetryMiddleware - 工具执行重试

工具执行失败时自动重试。与 ModelRetryMiddleware 的区别：一个重试模型调用，一个重试工具调用：

python

from langchain.agents.middleware import ToolRetryMiddleware

agent = create_agent(
    model=model,
    tools=[fetch_api_data, search_web],
    middleware=[
        ToolRetryMiddleware(max_retries=2),  # 工具失败最多重试 2 次
    ],
)

适用场景：工具依赖的外部 API 偶发超时、第三方服务短暂不可用。

ModelFallbackMiddleware - 模型降级

主模型失败时自动切换到备用模型，实现多 Provider 容灾：

python

import os
from langchain.chat_models import init_chat_model
from langchain.agents.middleware import ModelFallbackMiddleware

primary = init_chat_model(os.environ["LLM_MODEL"])          # 主模型
fallback = init_chat_model(os.environ["LLM_MODEL_FALLBACK"])  # 备用模型

agent = create_agent(
    model=primary,
    tools=[search],
    middleware=[
        ModelFallbackMiddleware(fallback_model=fallback),
    ],
)

适用场景：高可用要求的生产环境、多 Provider 容灾（如主用 Anthropic，备用 OpenAI）。

调用限制

中间件	功能
`ModelCallLimitMiddleware`	限制单次 `invoke` 中模型调用次数上限，超过则抛出错误。防止 Agent 陷入无限循环
`ToolCallLimitMiddleware`	限制单次 `invoke` 中工具调用次数上限。防止工具滥用或循环调用

python

from langchain.agents.middleware import ModelCallLimitMiddleware

agent = create_agent(
    model=model,
    tools=[search, calculator],
    middleware=[
        ModelCallLimitMiddleware(max_calls=10),  # 最多调用模型 10 次
    ],
)

工具与上下文增强

中间件	功能
`ContextEditingMiddleware`	在发送给模型前编辑 / 裁剪上下文消息，比 SummarizationMiddleware 更精细
`LLMToolSelectorMiddleware`	用 LLM 动态选择当前轮次可用的工具子集，适合工具数量多的场景
`LLMToolEmulatorMiddleware`	用 LLM 模拟工具执行结果，用于测试或无真实 API 时预演
`ProviderToolSearchMiddleware`	从 Provider（如 OpenAI）搜索可用工具并自动集成
`FilesystemMiddleware`	提供文件系统操作工具（读写、目录管理）
`FileSearchMiddleware`	提供文件搜索工具，适合大量文件中检索

高级能力

中间件	功能	状态
`SubagentMiddleware`	嵌套子 Agent，将子任务委托给独立的 Agent 执行	稳定
`TodoListMiddleware`	自动维护待办列表，追踪多步任务进度	稳定
`ShellToolMiddleware`	提供 Shell 命令执行工具	稳定
`RubricGradingMiddleware`	基于评分标准对 Agent 输出进行质量评估	`[beta]`

标注 [beta] 的中间件 API 尚不稳定，生产环境谨慎使用。

下一步

自定义中间件 - 当内置 Middleware 不满足需求时，学习从零编写自定义 Middleware
中间件概览 - 回顾 Middleware 的执行机制和排序策略
人机协作 HITL - HumanInTheLoopMiddleware 的完整中断与恢复流程
Agent 实战指南 - 了解 create_agent 的完整参数和工具定义

内置中间件 ​

先修知识 ​

PIIMiddleware - PII 检测与处理 ​

基本用法 ​

构造参数 ​

策略详解 ​

strategy: "redact" - 脱敏替换 ​

strategy: "block" - 拦截请求 ​

strategy: "mask" / "hash" ​

自定义正则检测器 ​

apply_to_output 与 apply_to_tool_results 选项 ​

完整示例：多类型 PII 检测 ​

SummarizationMiddleware - 自动对话摘要 ​

基本用法 ​

构造参数 ​

trigger 详解 ​

keep 详解 ​

工作原理 ​

完整示例 ​

使用建议 ​

HumanInTheLoopMiddleware - 人工审批 ​

基本用法 ​

构造参数 ​

完整示例 ​

设计建议 ​

实战：组合使用内置 Middleware ​

其它内置中间件速览 ​

容错与重试 ​

ModelRetryMiddleware - 模型调用重试 ​

ToolRetryMiddleware - 工具执行重试 ​

ModelFallbackMiddleware - 模型降级 ​

调用限制 ​

工具与上下文增强 ​

高级能力 ​

下一步 ​

内置中间件

先修知识

PIIMiddleware - PII 检测与处理

基本用法

构造参数

策略详解

strategy: "redact" - 脱敏替换

strategy: "block" - 拦截请求

strategy: "mask" / "hash"

自定义正则检测器

apply_to_output 与 apply_to_tool_results 选项

完整示例：多类型 PII 检测

SummarizationMiddleware - 自动对话摘要

基本用法

构造参数

trigger 详解

keep 详解

工作原理

完整示例

使用建议

HumanInTheLoopMiddleware - 人工审批

基本用法

构造参数

完整示例

设计建议

实战：组合使用内置 Middleware

其它内置中间件速览

容错与重试

ModelRetryMiddleware - 模型调用重试

ToolRetryMiddleware - 工具执行重试

ModelFallbackMiddleware - 模型降级

调用限制

工具与上下文增强

高级能力

下一步