LangChain Agent 中间件全解与实战-平芜编程栈

前言：为什么 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 协作或预处理-处理-后处理的流水线场景。