涵盖所有预构建中间件的使用场景、代码示例和配置说明，适合开发者快速集成到 Agent 项目中。

---

一、概述

LangChain 和 Deep Agents 提供了一系列生产级预构建中间件，覆盖 Agent 开发中的常见场景（如会话总结、权限控制、错误重试等），所有中间件支持灵活配置，可直接集成到基于 LangChain 的 Agent 项目中。

中间件分为两类：

• 通用中间件：支持所有 LLM 提供商（OpenAI、Anthropic、AWS 等）
• 特定提供商中间件：针对特定 LLM 优化的专属功能

---

二、通用中间件（Provider-agnostic）

2.1 会话总结（Summarization）

功能说明

当会话接近 Token 限制时，自动总结历史对话，保留最近消息并压缩旧上下文，适用于长对话、多轮交互场景。

API 参考

SummarizationMiddleware

基础使用示例

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

agent = create_agent(
    model="gpt-4.1",
    tools=[your_weather_tool, your_calculator_tool],  # 替换为实际工具
    middleware=[
        SummarizationMiddleware(
            model="gpt-4.1-mini",  # 用于生成总结的模型
            trigger=("tokens", 4000),  # 触发总结的 Token 阈值
            keep=("messages", 20),  # 保留最近 20 条消息
        ),
    ],
)

高级配置示例（多触发条件/比例阈值）

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

# 自定义模型配置（适用于 langchain>=1.1）
custom_profile = {
    "max_input_tokens": 100_000,  # 自定义最大输入 Token
}
model = init_chat_model("gpt-4.1", profile=custom_profile)

# 多条件触发：Token>=3000 或 消息数>=6
agent2 = create_agent(
    model=model,
    tools=[your_weather_tool, your_calculator_tool],
    middleware=[
        SummarizationMiddleware(
            model="gpt-4.1-mini",
            trigger=[("tokens", 3000), ("messages", 6)],  # 多条件 OR 逻辑
            keep=("fraction", 0.3),  # 保留 30% 的上下文
            summary_prompt="请简洁总结以下对话历史：{messages}",  # 自定义总结提示词
        ),
    ],
)

核心配置参数

参数	类型	说明	默认值
model	string / BaseChatModel	总结用模型（必填）	-
trigger	ContextSize / list[ContextSize]	触发条件（tokens/messages/fraction）	-
keep	ContextSize	总结后保留的上下文	(“messages”, 20)
summary_prompt	string	自定义总结提示词	内置模板
trim_tokens_to_summarize	number	总结时最大 Token 限制	4000

---

2.2 人工介入（Human-in-the-loop）

功能说明

在工具调用前暂停执行，等待人工审批、编辑或拒绝，适用于高风险操作（如数据库写入、财务交易）、合规流程。

API 参考

HumanInTheLoopMiddleware

使用示例

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

# 模拟工具
def your_read_email_tool(email_id: str) -> str:
    """读取指定 ID 的邮件"""
    return f"邮件内容（ID：{email_id}）：XXX"

def your_send_email_tool(recipient: str, subject: str, body: str) -> str:
    """发送邮件（高风险操作）"""
    return f"已发送邮件至 {recipient}，主题：{subject}"

agent = create_agent(
    model="gpt-4.1",
    tools=[your_read_email_tool, your_send_email_tool],
    checkpointer=InMemorySaver(),  # 必须：用于维护中断状态
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "your_send_email_tool": {  # 对发送邮件工具启用人工介入
                    "allowed_decisions": ["approve", "edit", "reject"],  # 允许的操作
                },
                "your_read_email_tool": False,  # 读取邮件工具跳过人工介入
            }
        ),
    ],
)

关键说明

• 必须配置 checkpointer（如 InMemorySaver）以保存状态
• 支持工具级别的精细控制（部分工具启用介入）
• 官方文档：Human-in-the-loop 详细指南
• 视频演示：Human-in-the-loop 中间件用法（替换为原始链接）

---

2.3 模型调用限制（Model call limit）

功能说明

限制模型调用次数，防止无限循环或过高成本，适用于生产环境成本控制、测试场景。

API 参考

ModelCallLimitMiddleware

使用示例

from langchain.agents import create_agent
from langchain.agents.middleware import ModelCallLimitMiddleware
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model="gpt-4.1",
    checkpointer=InMemorySaver(),  # 必须：用于跨会话计数
    tools=[],
    middleware=[
        ModelCallLimitMiddleware(
            thread_limit=10,  # 单个会话最大调用次数
            run_limit=5,  # 单次请求最大调用次数
            exit_behavior="end",  # 达到限制时的行为：end（优雅终止）/ error（抛异常）
        ),
    ],
)

核心配置参数

参数	类型	说明	默认值
thread_limit	number	跨会话最大调用次数	无限制
run_limit	number	单次请求最大调用次数	无限制
exit_behavior	string	达到限制行为	“end”

---

2.4 工具调用限制（Tool call limit）

功能说明

限制工具调用次数（全局或特定工具），防止过度调用外部 API、控制成本。

API 参考

ToolCallLimitMiddleware

使用示例（全局+工具级限制）

from langchain.agents import create_agent
from langchain.agents.middleware import ToolCallLimitMiddleware

# 全局限制：所有工具共 20 次会话调用，单次请求 10 次
global_limiter = ToolCallLimitMiddleware(thread_limit=20, run_limit=10)
# 工具级限制：search 工具最多 5 次会话调用，单次请求 3 次
search_limiter = ToolCallLimitMiddleware(tool_name="search", thread_limit=5, run_limit=3)
# 严格限制：scrape_webpage 工具单次请求最多 2 次，超限制抛异常
strict_limiter = ToolCallLimitMiddleware(
    tool_name="scrape_webpage", 
    run_limit=2, 
    exit_behavior="error"
)

agent = create_agent(
    model="gpt-4.1",
    tools=[search_tool, database_tool, scraper_tool],  # 替换为实际工具
    middleware=[global_limiter, search_limiter, strict_limiter],
)

核心配置参数

参数	类型	说明	默认值
tool_name	string	目标工具名称（不指定则全局生效）	None
thread_limit	number	跨会话最大调用次数（需 checkpointer）	None
run_limit	number	单次请求最大调用次数	None
exit_behavior	string	超限制行为：continue/error/end	“continue”

---

2.5 模型降级（Model fallback）

功能说明

当主模型调用失败时，自动切换到备用模型，提升系统容错性、优化成本。

API 参考

ModelFallbackMiddleware

使用示例

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

agent = create_agent(
    model="gpt-4.1",  # 主模型
    tools=[],
    middleware=[
        ModelFallbackMiddleware(
            "gpt-4.1-mini",  # 第一备用模型
            "claude-3-5-sonnet-20241022",  # 第二备用模型（支持多个）
        ),
    ],
)

关键说明

• 支持任意 LLM 提供商的模型组合（OpenAI + Anthropic + AWS 等）
• 按参数顺序依次尝试降级
• 视频演示：Model Fallback 中间件用法（替换为原始链接）

---

2.6 PII 检测（PII detection）

功能说明

检测并处理对话中的个人身份信息（PII），支持内置类型和自定义类型，适用于合规场景（医疗、金融）。

API 参考

PIIMiddleware

基础使用示例（内置 PII 类型）

from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware

agent = create_agent(
    model="gpt-4.1",
    tools=[],
    middleware=[
        # 检测邮箱并脱敏
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        # 检测信用卡号并掩码
        PIIMiddleware("credit_card", strategy="mask", apply_to_input=True),
    ],
)

高级使用示例（自定义 PII 类型）

from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware
import re

# 方法 1：正则表达式匹配 API Key
agent1 = create_agent(
    model="gpt-4.1",
    tools=[],
    middleware=[
        PIIMiddleware(
            "api_key",
            detector=r"sk-[a-zA-Z0-9]{32}",  # 正则模式
            strategy="block",  # 检测到则阻止
        ),
    ],
)

# 方法 2：自定义检测函数（带验证逻辑）
def detect_valid_ssn(content: str) -> list[dict[str, str | int]]:
    """检测并验证 SSN（美国社会安全号）"""
    matches = []
    pattern = r"\d{3}-\d{2}-\d{4}"
    for match in re.finditer(pattern, content):
        ssn = match.group(0)
        first_three = int(ssn[:3])
        # 排除无效前缀（000、666、900-999）
        if first_three not in [0, 666] and not (900 <= first_three <= 999):
            matches.append({
                "text": ssn,
                "start": match.start(),
                "end": match.end(),
            })
    return matches

agent2 = create_agent(
    model="gpt-4.1",
    tools=[],
    middleware=[
        PIIMiddleware(
            "ssn",
            detector=detect_valid_ssn,  # 自定义检测函数
            strategy="hash",  # 检测到则哈希处理
            apply_to_output=True,  # 同时检查 AI 输出
        ),
    ],
)

核心配置参数

参数	类型	说明	默认值
pii_type	string	PII 类型（内置/自定义）	必填
strategy	string	处理策略：block/redact/mask/hash	“redact”
detector	function / regex	检测逻辑（默认用内置检测器）	None
apply_to_input	boolean	检查用户输入	True
apply_to_output	boolean	检查 AI 输出	False

内置 PII 类型

• email：邮箱地址
• credit_card：信用卡号
• ip：IP 地址
• mac_address：MAC 地址
• url：网址

---

2.7 任务清单（To-do list）

功能说明

为 Agent 提供任务规划和跟踪能力，适用于复杂多步骤任务、长流程协作。

API 参考

TodoListMiddleware

使用示例

from langchain.agents import create_agent
from langchain.agents.middleware import TodoListMiddleware

agent = create_agent(
    model="gpt-4.1",
    tools=[read_file, write_file, run_tests],  # 替换为实际工具
    middleware=[
        TodoListMiddleware(
            system_prompt="请根据用户需求创建任务清单，按优先级执行并更新进度",  # 自定义提示词
            tool_description="write_todos：创建或更新任务清单，格式为：- [ ] 任务描述",  # 自定义工具说明
        ),
    ],
)

关键说明

• 自动注入 write_todos 工具
• 内置任务规划提示词，支持自定义
• 视频演示：To-do List 中间件用法（替换为原始链接）

---

2.8 工具选择器（LLM tool selector）

功能说明

通过 LLM 智能筛选相关工具后再调用主模型，适用于工具数量多（10+）、减少 Token 消耗的场景。

API 参考

LLMToolSelectorMiddleware

使用示例

from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolSelectorMiddleware

# 假设有 10+ 个工具
tools = [tool1, tool2, tool3, tool4, tool5, tool6, tool7, tool8, tool9, tool10]

agent = create_agent(
    model="gpt-4.1",
    tools=tools,
    middleware=[
        LLMToolSelectorMiddleware(
            model="gpt-4.1-mini",  # 用于筛选工具的轻量模型
            max_tools=3,  # 最多选择 3 个相关工具
            always_include=["search"],  # 始终包含 search 工具（不计入 max_tools）
            system_prompt="根据用户查询，从提供的工具中选择最相关的，无需解释",  # 自定义筛选提示词
        ),
    ],
)

---

2.9 工具重试（Tool retry）

功能说明

工具调用失败时自动重试（支持指数退避），适用于网络依赖型工具、临时故障场景。

API 参考

ToolRetryMiddleware

使用示例

from langchain.agents import create_agent
from langchain.agents.middleware import ToolRetryMiddleware

agent = create_agent(
    model="gpt-4.1",
    tools=[search_tool, database_tool, api_tool],
    middleware=[
        ToolRetryMiddleware(
            max_retries=3,  # 最大重试次数（含首次共 4 次）
            backoff_factor=2.0,  # 指数退避系数（1s → 2s → 4s）
            initial_delay=1.0,  # 首次重试延迟 1s
            max_delay=60.0,  # 最大延迟 60s
            jitter=True,  # 增加随机抖动（避免并发重试冲突）
            tools=["api_tool"],  # 仅对 api_tool 启用重试
            retry_on=(ConnectionError, TimeoutError),  # 仅重试特定异常
            on_failure="return_message",  # 重试失败后返回错误信息
        ),
    ],
)

---

2.10 模型重试（Model retry）

功能说明

模型调用失败时自动重试（支持指数退避），适用于模型 API 临时不可用、限流场景。

API 参考

ModelRetryMiddleware

高级使用示例（自定义异常筛选）

from langchain.agents import create_agent
from langchain.agents.middleware import ModelRetryMiddleware

# 自定义异常筛选函数
def should_retry(error: Exception) -> bool:
    """仅对限流和超时异常重试"""
    # 检查 HTTP 状态码（如 429 限流、503 服务不可用）
    if hasattr(error, "status_code"):
        return error.status_code in (429, 503)
    # 检查异常类型
    return isinstance(error, (TimeoutError, ConnectionError))

agent = create_agent(
    model="gpt-4.1",
    tools=[search_tool],
    middleware=[
        ModelRetryMiddleware(
            max_retries=4,
            retry_on=should_retry,  # 自定义筛选逻辑
            backoff_factor=1.5,  # 退避系数 1.5
            initial_delay=2.0,
            on_failure=lambda e: f"模型调用失败：{e}，请稍后重试",  # 自定义错误信息
        ),
    ],
)

---

2.11 工具模拟（LLM tool emulator）

功能说明

用 LLM 生成模拟工具响应，替代真实工具调用，适用于测试、工具未就绪场景。

API 参考

LLMToolEmulator

使用示例

from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolEmulator
from langchain.tools import tool

# 定义真实工具（但用模拟响应）
@tool
def get_weather(location: str) -> str:
    """获取指定城市天气"""
    return f"{location} 天气：晴"

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件"""
    return "邮件发送成功"

# 仅模拟 get_weather 工具，send_email 用真实逻辑
agent = create_agent(
    model="gpt-4.1",
    tools=[get_weather, send_email],
    middleware=[
        LLMToolEmulator(
            tools=["get_weather"],  # 仅模拟指定工具
            model="claude-sonnet-4-6",  # 用于生成模拟响应的模型
        ),
    ],
)

---

2.12 上下文编辑（Context editing）

功能说明

当 Token 超限，清理旧工具调用输出，保留最近结果，适用于长对话、多工具调用场景。

API 参考

ContextEditingMiddleware、ClearToolUsesEdit

使用示例

from langchain.agents import create_agent
from langchain.agents.middleware import ContextEditingMiddleware, ClearToolUsesEdit

agent = create_agent(
    model="gpt-4.1",
    tools=[search_tool, calculator_tool, database_tool],
    middleware=[
        ContextEditingMiddleware(
            edits=[
                ClearToolUsesEdit(
                    trigger=2000,  # Token 超过 2000 时触发清理
                    keep=3,  # 保留最近 3 个工具结果
                    clear_tool_inputs=False,  # 不清理工具调用参数
                    exclude_tools=["database_tool"],  # 不清理 database_tool 的输出
                    placeholder="[已清理旧工具输出]",  # 清理后的占位符
                ),
            ],
        ),
    ],
)

---

2.13 Shell 工具（Shell tool）

功能说明

为 Agent 提供持久化 Shell 会话，支持执行系统命令、文件操作、自动化脚本。

安全注意

• 建议使用隔离执行策略（Docker/CodexSandbox）
• 避免在未授权环境中使用 HostExecutionPolicy

API 参考

ShellToolMiddleware

使用示例（Docker 隔离模式）

from langchain.agents import create_agent
from langchain.agents.middleware import (
    ShellToolMiddleware,
    DockerExecutionPolicy,
    RedactionRule,
)

agent = create_agent(
    model="gpt-4.1",
    tools=[search_tool],
    middleware=[
        ShellToolMiddleware(
            workspace_root="/workspace",  # 工作目录
            startup_commands=["pip install requests", "export PYTHONPATH=/workspace"],  # 启动命令
            execution_policy=DockerExecutionPolicy(
                image="python:3.11-slim",  # Docker 镜像
                command_timeout=60.0,  # 命令超时 60s
            ),
            # 输出脱敏（隐藏 API Key）
            redaction_rules=[
                RedactionRule(pii_type="api_key", detector=r"sk-[a-zA-Z0-9]{32}"),
            ],
        ),
    ],
)

执行策略选项

策略	说明	适用场景
HostExecutionPolicy	直接在主机执行	可信环境、已隔离容器
DockerExecutionPolicy	启动独立 Docker 容器	需隔离、多环境兼容
CodexSandboxExecutionPolicy	Codex CLI 沙箱	严格安全限制、 syscall 管控

---

2.14 文件搜索（File search）

功能说明

提供 Glob 文件名匹配和 Grep 内容搜索工具，适用于代码分析、大文件检索场景。

API 参考

FilesystemFileSearchMiddleware

使用示例

from langchain.agents import create_agent
from langchain.agents.middleware import FilesystemFileSearchMiddleware
from langchain.messages import HumanMessage

agent = create_agent(
    model="gpt-4.1",
    tools=[],
    middleware=[
        FilesystemFileSearchMiddleware(
            root_path="/workspace",  # 搜索根目录
            use_ripgrep=True,  # 使用 ripgrep 加速（未安装则 fallback 到 Python 正则）
            max_file_size_mb=10,  # 跳过大于 10MB 的文件
        ),
    ],
)

# 示例：搜索所有 Python 文件中的 async 函数
result = agent.invoke({
    "messages": [HumanMessage("查找 /workspace 下所有 Python 文件中包含 'async def' 的代码")]
})

内置工具

• glob_search(pattern)：文件名匹配（如 **/*.py）
• grep_search(pattern, include)：内容搜索（支持正则，include 过滤文件类型）

---

2.15 文件系统（Filesystem）

功能说明

提供文件读写工具，支持短期/长期存储，解决上下文窗口溢出问题（将长文本存储到文件而非上下文）。

依赖说明

需安装 deepagents：pip install deepagents

API 参考

FilesystemMiddleware

基础使用示例

from langchain.agents import create_agent
from deepagents.middleware.filesystem import FilesystemMiddleware

agent = create_agent(
    model="claude-sonnet-4-6",
    middleware=[
        FilesystemMiddleware(
            # 自定义工具说明
            custom_tool_descriptions={
                "ls": "列出指定目录下的文件",
                "read_file": "读取文件内容，支持按行读取（line_start/line_end 参数）",
                "write_file": "写入文件，覆盖模式（需指定 file_path 和 content）",
                "edit_file": "编辑文件，追加或替换内容",
            },
        ),
    ],
)

长期存储配置（跨会话持久化）

from langchain.agents import create_agent
from deepagents.middleware import FilesystemMiddleware
from deepagents.backends import CompositeBackend, StateBackend, StoreBackend
from langgraph.store.memory import InMemoryStore

# 初始化持久化存储
store = InMemoryStore()

agent = create_agent(
    model="claude-sonnet-4-6",
    store=store,
    middleware=[
        FilesystemMiddleware(
            # 复合存储：/memories/ 路径下的文件持久化，其他路径临时存储
            backend=lambda rt: CompositeBackend(
                default=StateBackend(rt),
                routes={"/memories/": StoreBackend(rt)}
            ),
        ),
    ],
)

# 示例：写入持久化文件
agent.invoke({
    "messages": [HumanMessage("将 '用户偏好：喜欢深色模式' 写入 /memories/user_preferences.txt")]
})

---

2.16 子代理（Subagent）

功能说明

允许主 Agent 启动子 Agent 处理细分任务，隔离上下文，适用于复杂任务拆分、专业化处理。

依赖说明

需安装 deepagents：pip install deepagents

API 参考

SubAgentMiddleware

基础使用示例（定义子代理）

from langchain.tools import tool
from langchain.agents import create_agent
from deepagents.middleware.subagents import SubAgentMiddleware

# 子代理专用工具
@tool
def get_weather(city: str) -> str:
    """获取指定城市天气"""
    return f"{city} 天气：晴，25℃"

agent = create_agent(
    model="claude-sonnet-4-6",
    middleware=[
        SubAgentMiddleware(
            default_model="claude-sonnet-4-6",  # 子代理默认模型
            default_tools=[],  # 子代理默认工具
            subagents=[
                {
                    "name": "weather_agent",  # 子代理名称
                    "description": "专门处理天气查询的子代理",  # 子代理描述
                    "system_prompt": "仅使用 get_weather 工具查询天气，直接返回结果",  # 子代理提示词
                    "tools": [get_weather],  # 子代理专属工具
                    "model": "gpt-4.1",  # 子代理自定义模型
                    "middleware": [],  # 子代理专属中间件
                }
            ],
        )
    ],
)

# 示例：主 Agent 调用子代理
result = agent.invoke({
    "messages": [HumanMessage("查询北京的天气")]
})

高级使用示例（自定义 LangGraph 子代理）

from langchain.agents import create_agent
from deepagents.middleware.subagents import SubAgentMiddleware, CompiledSubAgent
from langgraph.graph import StateGraph

# 1. 构建自定义 LangGraph 工作流
def create_weather_graph():
    """创建天气查询专用工作流"""
    from langgraph.graph import StateGraph, MessagesState

    def weather_node(state: MessagesState):
        """处理天气查询的节点"""
        city = state["messages"][-1].content  # 提取城市名
        return {"messages": [get_weather(city)]}

    workflow = StateGraph(MessagesState)
    workflow.add_node("weather", weather_node)
    workflow.set_entry_point("weather")
    return workflow.compile()

# 2. 包装为 CompiledSubAgent
weather_graph = create_weather_graph()
weather_subagent = CompiledSubAgent(
    name="weather_graph_agent",
    description="天气查询专用子代理（基于 LangGraph）",
    runnable=weather_graph
)

# 3. 集成到主 Agent
agent = create_agent(
    model="claude-sonnet-4-6",
    middleware=[
        SubAgentMiddleware(
            default_model="claude-sonnet-4-6",
            default_tools=[],
            subagents=[weather_subagent],
        )
    ],
)

---

三、特定提供商中间件（Provider-specific）

以下中间件针对特定 LLM 提供商优化，需结合对应平台使用：

提供商	支持的中间件	说明
Anthropic（Claude）	提示词缓存、Bash 工具、文本编辑器、内存管理、文件搜索	专为 Claude 模型优化，支持长上下文
AWS（Bedrock）	提示词缓存	适配 Amazon Bedrock 模型，降低调用成本
OpenAI	内容审核	集成 OpenAI 内容审核 API，合规过滤

详细使用说明请参考各提供商官方文档：

• Anthropic：Anthropic 中间件文档
• AWS：AWS Bedrock 中间件文档
• OpenAI：OpenAI 中间件文档

---

四、结语

LangChain 预构建中间件覆盖了 Agent 开发的核心场景（容错、成本控制、合规、任务拆分等），通过模块化设计可快速集成到项目中。使用建议：

1. 生产环境必选：ModelFallbackMiddleware（容错）、ModelCallLimitMiddleware（成本控制）、PIIMiddleware（合规）
2. 长对话场景：SummarizationMiddleware + ContextEditingMiddleware
3. 高风险操作：HumanInTheLoopMiddleware
4. 测试/开发：LLMToolEmulator（无需真实工具）

所有中间件支持灵活扩展，可通过自定义函数、配置参数适配具体业务需求。官方文档持续更新，建议关注 LangChain 中间件官网 获取最新功能。

---