内置中间件库：开箱即用的 Agent 能力扩展

内置中间件库：开箱即用的 Agent 能力扩展

学习目标

本章要解决什么问题：构建生产级 Agent 需要哪些常见的横切关注点？LangChain 内置中间件如何覆盖这些需求？

读者将学到：

• 17 个内置中间件的分类和功能概览
• 几个典型中间件的实现分析
• 如何组合多个中间件

前置知识

本章涉及中间件链的通用原理，建议先阅读：

• 工具调用中间件链

项目实践

内置中间件一览

位于 libs/langchain_v1/langchain/agents/middleware/ 目录：

中间件文件	挂钩子	功能
model_retry.py	wrap_model_call	模型调用失败时重试
model_fallback.py	wrap_model_call	主模型失败时切换到备用模型
model_call_limit.py	wrap_model_call	限制最大模型调用次数
tool_retry.py	wrap_tool_call	工具执行失败时重试
tool_call_limit.py	wrap_tool_call	限制工具调用次数
tool_selection.py	wrap_model_call	动态启用/禁用工具
tool_emulator.py	wrap_tool_call	将普通函数模拟为工具调用格式
human_in_the_loop.py	before_model	人工审批/编辑/拒绝
pii.py	wrap_model_call	PII（个人身份信息）检测和脱敏
shell_tool.py	—	安全的 Shell 工具实现
summarization.py	before_model	上下文过长时自动摘要压缩
context_editing.py	before_model	消息编辑（删除、替换）
file_search.py	—	文件搜索工具
todo.py	—	Todo 列表管理工具
_retry.py	—	重试逻辑的共享实现
_execution.py	—	执行逻辑的共享实现
_redaction.py	—	信息脱敏的共享实现

典型中间件分析

模型重试（model_retry）

# 伪代码
@wrap_model_call
def retry_model_call(request, handler):
    max_retries = 3
    for attempt in range(max_retries):
        try:
            return handler(request)
        except Exception as exc:
            if attempt == max_retries - 1:
                raise
            # 指数退避
            time.sleep(2 ** attempt)

设计要点：使用指数退避（exponential backoff）避免频繁重试对 API 造成压力。

上下文摘要压缩（summarization）

当消息数量接近 Token 限制时，summarization.py 中间件在 before_model 钩子中压缩历史消息：

# 伪代码
@before_model
def summarize_if_needed(state, runtime) -> dict | None:
    if estimate_tokens(state["messages"]) > threshold:
        summary = summarize_messages(state["messages"])
        return {"messages": [summary] + state["messages"][-keep_recent:]}
    return None

设计要点：保留最近 N 条消息不压缩（保留对话细节），将早期消息摘要为单条总结。

工具选择（tool_selection）

根据上下文动态启用/禁用工具：

# 伪代码
@wrap_model_call
def select_tools(request, handler):
    # 根据当前上下文过滤工具列表
    available_tools = filter_tools_by_context(request.tools, request.state)
    request = request.override(tools=available_tools)
    return handler(request)

设计要点：减少工具列表可以降低 Token 消耗，同时避免模型在不相关时调用工具。

中间件组合

多个中间件可以叠加使用：

# 伪代码
from langchain.agents import create_agent
from langchain.agents.middleware import (
    RetryMiddleware,
    ModelFallbackMiddleware,
    SummarizationMiddleware,
    HITLMiddleware,
)

agent = create_agent(
    model="openai:gpt-5.5",
    tools=[search, calculate],
    middleware=[
        RetryMiddleware(max_retries=3),
        ModelFallbackMiddleware(fallback_models=["anthropic:claude-opus-4-7"]),
        SummarizationMiddleware(token_limit=8000),
        HITLMiddleware(review_configs=[...]),
    ],
)

执行顺序：Retry（最外层）→ Fallback → Summarization → HITL（最内层）→ 模型调用。

问题与规避

中间件执行开销

问题：每个中间件都增加一次函数调用开销，中间件过多可能影响性能。

对策：中间件的开销通常远小于模型 API 调用延迟（几百 ms vs 几秒）。只有在极端高频调用场景下才需要关注。

中间件之间的隐式依赖

问题：一个中间件可能依赖另一个中间件的输出（如 Summarization 依赖 Retry 确保消息格式正确）。

对策：中间件注册顺序决定了执行顺序，需要仔细排列。框架不提供依赖声明机制，需要开发者手动管理。

设计取舍

内置中间件 vs 用户自定义

方案	优势	代价
内置中间件	开箱即用、经过充分测试、最佳实践	可能不覆盖所有场景
用户自定义	完全灵活	需要自行测试和维护

LangChain 提供内置中间件覆盖最常见场景，同时保留完整的装饰器和基类机制供用户扩展。

参考来源

• LangChain Middleware 文档