LangChain Agent 中间件全解与实战

m0_64581050

448人浏览 · 2026-07-30 20:06:30

m0_64581050 · 2026-07-30 20:06:30 发布

前言：为什么 Agent 需要中间件？

如果你用过 LangChain 构建过 AI Agent，大概率遇到过这样的困境：测试阶段一切正常，部署到生产环境后却问题频发——上下文管理混乱、Agent 行为不可预测、工具调用失控……最后不得不写一堆自定义代码来补救。

问题根源在于：Agent 的可靠性取决于上下文工程。模型接收什么信息，决定了它输出什么结果。而过多的配置参数、散落在各处的自定义逻辑，让 Agent 变得难以维护和扩展。

LangChain 1.0 引入的 Agent 中间件（Middleware） 正是为了解决这个问题。它借鉴了 Web 框架（如 FastAPI）的中间件模式，在 Agent 执行流程的关键节点提供标准化的钩子（Hooks），让开发者能系统化地控制上下文、注入业务逻辑、增强生产可靠性。

本文将带你全面理解 Agent 中间件的设计理念、核心钩子、内置实现，并通过实战案例掌握自定义中间件的开发。

一、中间件是什么？Agent 的“信息调度中心”

1.1 核心思想

中间件本质上是一组可组合的执行钩子，在 Agent 循环的各个阶段运行。它不改变 Agent 的核心逻辑（模型调用 → 工具执行 → 循环），而是在这些步骤前后插入自定义逻辑，实现：

输入预处理：PII 脱敏、提示词动态生成、对话摘要
执行控制：工具权限校验、人工审批（HITL）、重试与降级
输出后处理：结果验证、日志记录、状态更新

1.2 中间件 vs 传统参数配置

在 v1.0 之前，Agent 的定制依赖大量参数和手写循环，代码臃肿且难以复用。例如旧版 AgentExecutor 需要配置 max_iterations、handle_parsing_errors、trim_intermediate_steps 等十几个参数。

中间件的优势在于：

模块化：每个中间件独立负责一个功能（如摘要、缓存、日志）
可组合：多个中间件按顺序叠加，互不干扰
可复用：写好的中间件可在不同项目间共享

二、中间件的核心钩子（Hooks）

LangChain 中间件提供了 6 个核心钩子，覆盖 Agent 的完整生命周期：

钩子	执行时机	典型用途
`before_agent`	Agent 启动前（仅一次）	加载外部记忆、初始化资源、输入校验
`before_model`	每次模型调用前	动态更新提示词、对话摘要、PII 脱敏
`wrap_model_call`	包裹模型调用	重试机制、模型降级、动态工具绑定
`after_model`	模型响应后、工具执行前	人工审批（HITL）、输出格式校验
`wrap_tool_call`	包裹工具调用	工具权限控制、参数修改、缓存结果
`after_agent`	Agent 完成后（仅一次）	保存结果、释放资源、发送通知

钩子分为两种风格：

节点式钩子（Node-style）：顺序执行，通过返回 dict 更新 Agent 状态。如 before_model、after_model
包裹式钩子（Wrap-style）：包裹核心调用，可控制是否执行、重试或短路。如 wrap_model_call、wrap_tool_call

三、开箱即用的内置中间件

LangChain 提供了多个生产级中间件，可直接引入使用。

3.1 对话摘要中间件（SummarizationMiddleware）

长对话容易超出模型上下文窗口。该中间件在 before_model 阶段检测消息数量或 Token 阈值，自动生成摘要替换历史消息。

python

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

agent = create_agent(
    model="claude-sonnet-4-6",
    tools=[...],
    middleware=[
        SummarizationMiddleware(
            model="claude-sonnet-4-6",
            trigger={"tokens": 500}  # 超过 500 token 触发摘要
        )
    ]
)

3.2 人工审批中间件（HumanInTheLoopMiddleware）

敏感操作（如发送邮件、删除数据）需人工确认。该中间件在 after_model 阶段拦截指定工具调用，挂起执行等待人工决策。

python

from langchain.agents.middleware import HumanInTheLoopMiddleware

agent = create_agent(
    model="claude-sonnet-4-6",
    tools=[send_email, read_email],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "send_email": {
                    "allowed_decisions": ["approve", "edit", "reject"]
                }
            }
        )
    ],
    checkpointer=MemorySaver()  # 必须配合 checkpointer 实现暂停与恢复
)

恢复挂起的执行：

python

from langgraph.types import Command

agent.invoke(
    Command(resume={"decisions": [{"type": "approve"}]}),
    config={"configurable": {"thread_id": "session-123"}}
)

3.3 PII 脱敏中间件（PIIRedactionMiddleware）

在模型调用前自动识别并脱敏敏感信息（邮箱、电话、身份证号等），防止数据泄露。

python

from langchain.agents.middleware import PIIRedactionMiddleware

agent = create_agent(
    model="claude-sonnet-4-6",
    middleware=[
        PIIRedactionMiddleware(
            patterns=["email", "phone", "ssn"],
            strategy="redact"  # 或 "mask", "block"
        )
    ]
)

3.4 Anthropic 专属中间件

针对 Claude 模型，langchain-anthropic 提供了专用中间件：

Prompt Caching：缓存系统提示词和工具定义，降低成本和延迟
Bash Tool：提供生产级的 Bash 命令执行（支持 Docker 隔离）
Text Editor：文件编辑工具（支持状态持久化和文件系统两种模式）

python

from langchain_anthropic.middleware import (
    AnthropicPromptCachingMiddleware,
    ClaudeBashToolMiddleware
)

agent = create_agent(
    model=ChatAnthropic(model="claude-sonnet-4-6"),
    middleware=[
        AnthropicPromptCachingMiddleware(ttl="5m"),
        ClaudeBashToolMiddleware(
            workspace_root="/workspace",
            execution_policy=DockerExecutionPolicy(image="python:3.11")
        )
    ]
)

四、实战：自定义中间件开发

内置中间件无法覆盖所有场景，自定义中间件才是真正的杀手锏。我们通过两个实战案例演示如何开发。

4.1 装饰器式中间件（快速开发）

适合单钩子、逻辑简单的场景。使用 @before_model、@wrap_tool_call 等装饰器。

需求：为所有工具调用添加重试机制，失败时返回友好提示。

python

from langchain.agents.middleware import wrap_tool_call, ToolCallRequest
from langchain_core.messages import ToolMessage
from typing import Callable

@wrap_tool_call
def retry_on_tool_failure(
    request: ToolCallRequest,
    handler: Callable[[ToolCallRequest], ToolMessage]
) -> ToolMessage:
    for attempt in range(3):
        try:
            return handler(request)
        except Exception as e:
            print(f"工具 {request.tool_call['name']} 第 {attempt+1} 次尝试失败: {e}")
            if attempt == 2:
                return ToolMessage(
                    content=f"工具执行失败，请重试或换一种方式。错误：{e}",
                    tool_call_id=request.tool_call["id"]
                )
    # 不会执行到这里

使用中间件：

python

agent = create_agent(
    model="gpt-4",
    tools=[search, calculator],
    middleware=[retry_on_tool_failure]
)

4.2 类式中间件（复杂场景）

适合多钩子、需维护状态的场景。继承 AgentMiddleware 并实现对应钩子方法。

需求：根据用户技术水平动态调整模型和工具。专家用户使用高级模型 + 复杂工具，新手使用轻量模型 + 基础工具。

python

from langchain.agents.middleware import AgentMiddleware, ModelRequest, ModelResponse
from typing import Callable, Any
from pydantic import BaseModel

# 1. 定义上下文 Schema（运行时传入，不持久化）
class UserContext(BaseModel):
    user_expertise: str = "beginner"  # "beginner" 或 "expert"

# 2. 自定义中间件
class ExpertiseBasedToolMiddleware(AgentMiddleware):
    context_schema = UserContext

    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse]
    ) -> ModelResponse:
        # 从运行时上下文读取用户级别
        user_level = request.runtime.context.get("user_expertise", "beginner")

        if user_level == "expert":
            # 专家用户：切换更强大的模型 + 高级工具
            request = request.override(
                model="gpt-4",
                tools=[advanced_search, data_analysis, calculator]
            )
        else:
            # 新手用户：轻量模型 + 基础工具
            request = request.override(
                model="gpt-3.5-turbo",
                tools=[simple_search, basic_calculator]
            )

        return handler(request)

使用中间件，并在调用时传入上下文：

python

agent = create_agent(
    model="gpt-4",  # 默认模型，会被中间件覆盖
    tools=[simple_search, advanced_search, basic_calculator, data_analysis],
    middleware=[ExpertiseBasedToolMiddleware()]
)

# 专家用户调用
result = agent.invoke(
    {"messages": [HumanMessage("分析这份销售数据")]},
    config={"context": {"user_expertise": "expert"}}
)

4.3 中间件状态管理

如果中间件需要跨轮次持久化数据（如调用计数、Token 用量），可扩展 Agent 的状态 Schema。

python

from langchain.agents.middleware import AgentMiddleware, AgentState
from typing import NotRequired

class TrackingState(AgentState):
    model_call_count: NotRequired[int]

class UsageTrackingMiddleware(AgentMiddleware):
    state_schema = TrackingState

    def after_model(self, state: TrackingState, runtime) -> dict[str, Any] | None:
        # 每次模型调用后更新计数
        current = state.get("model_call_count", 0)
        return {"model_call_count": current + 1}

五、进阶：中间件组合与执行顺序

多个中间件按注册顺序执行，类似洋葱模型：

进入阶段（before_model）：按注册顺序执行
包裹阶段（wrap_model_call）：外层 → 内层
退出阶段（after_model）：按注册顺序的逆序执行

python

agent = create_agent(
    model="gpt-4",
    middleware=[
        LoggingMiddleware(),      # ① 先记录日志
        RetryMiddleware(),        # ② 再添加重试
        PIIRedactionMiddleware()  # ③ 最后脱敏
    ]
)

执行顺序：Logging.before_model → Retry.before_model → PII.before_model → 模型调用 → PII.after_model → Retry.after_model → Logging.after_model

六、中间件 + LangGraph：在复杂工作流中使用

中间件并非仅限单 Agent 使用。create_agent 返回的 Agent 本身是一个 LangGraph 图，可嵌入更大的工作流中，中间件钩子依然有效。

python

from langgraph.graph import StateGraph, START
from langchain.agents import create_agent, AgentState

# 创建带中间件的 Agent
email_agent = create_agent(
    model="claude-sonnet-4-6",
    tools=[read_email, send_email],
    middleware=[HumanInTheLoopMiddleware(interrupt_on={"send_email": True})]
)

# 嵌入更大的工作流
graph = (
    StateGraph(AgentState)
    .add_node("classify", classify_node)      # 分类节点
    .add_node("email_agent", email_agent)     # Agent 节点
    .add_edge(START, "classify")
    .add_conditional_edges("classify", route) # 动态路由到不同 Agent
    .compile()
)

这种模式适合多 Agent 协作或预处理-处理-后处理的流水线场景。