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等十几个参数。中间件的优势在于模块化每个中间件独立负责一个功能如摘要、缓存、日志可组合多个中间件按顺序叠加互不干扰可复用写好的中间件可在不同项目间共享二、中间件的核心钩子HooksLangChain 中间件提供了 6 个核心钩子覆盖 Agent 的完整生命周期钩子执行时机典型用途before_agentAgent 启动前仅一次加载外部记忆、初始化资源、输入校验before_model每次模型调用前动态更新提示词、对话摘要、PII 脱敏wrap_model_call包裹模型调用重试机制、模型降级、动态工具绑定after_model模型响应后、工具执行前人工审批HITL、输出格式校验wrap_tool_call包裹工具调用工具权限控制、参数修改、缓存结果after_agentAgent 完成后仅一次保存结果、释放资源、发送通知钩子分为两种风格节点式钩子Node-style顺序执行通过返回dict更新 Agent 状态。如before_model、after_model包裹式钩子Wrap-style包裹核心调用可控制是否执行、重试或短路。如wrap_model_call、wrap_tool_call三、开箱即用的内置中间件LangChain 提供了多个生产级中间件可直接引入使用。3.1 对话摘要中间件SummarizationMiddleware长对话容易超出模型上下文窗口。该中间件在before_model阶段检测消息数量或 Token 阈值自动生成摘要替换历史消息。pythonfrom langchain.agents import create_agent from langchain.agents.middleware import SummarizationMiddleware agent create_agent( modelclaude-sonnet-4-6, tools[...], middleware[ SummarizationMiddleware( modelclaude-sonnet-4-6, trigger{tokens: 500} # 超过 500 token 触发摘要 ) ] )3.2 人工审批中间件HumanInTheLoopMiddleware敏感操作如发送邮件、删除数据需人工确认。该中间件在after_model阶段拦截指定工具调用挂起执行等待人工决策。pythonfrom langchain.agents.middleware import HumanInTheLoopMiddleware agent create_agent( modelclaude-sonnet-4-6, tools[send_email, read_email], middleware[ HumanInTheLoopMiddleware( interrupt_on{ send_email: { allowed_decisions: [approve, edit, reject] } } ) ], checkpointerMemorySaver() # 必须配合 checkpointer 实现暂停与恢复 )恢复挂起的执行pythonfrom langgraph.types import Command agent.invoke( Command(resume{decisions: [{type: approve}]}), config{configurable: {thread_id: session-123}} )3.3 PII 脱敏中间件PIIRedactionMiddleware在模型调用前自动识别并脱敏敏感信息邮箱、电话、身份证号等防止数据泄露。pythonfrom langchain.agents.middleware import PIIRedactionMiddleware agent create_agent( modelclaude-sonnet-4-6, middleware[ PIIRedactionMiddleware( patterns[email, phone, ssn], strategyredact # 或 mask, block ) ] )3.4 Anthropic 专属中间件针对 Claude 模型langchain-anthropic提供了专用中间件Prompt Caching缓存系统提示词和工具定义降低成本和延迟Bash Tool提供生产级的 Bash 命令执行支持 Docker 隔离Text Editor文件编辑工具支持状态持久化和文件系统两种模式pythonfrom langchain_anthropic.middleware import ( AnthropicPromptCachingMiddleware, ClaudeBashToolMiddleware ) agent create_agent( modelChatAnthropic(modelclaude-sonnet-4-6), middleware[ AnthropicPromptCachingMiddleware(ttl5m), ClaudeBashToolMiddleware( workspace_root/workspace, execution_policyDockerExecutionPolicy(imagepython:3.11) ) ] )四、实战自定义中间件开发内置中间件无法覆盖所有场景自定义中间件才是真正的杀手锏。我们通过两个实战案例演示如何开发。4.1 装饰器式中间件快速开发适合单钩子、逻辑简单的场景。使用before_model、wrap_tool_call等装饰器。需求为所有工具调用添加重试机制失败时返回友好提示。pythonfrom 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]} 第 {attempt1} 次尝试失败: {e}) if attempt 2: return ToolMessage( contentf工具执行失败请重试或换一种方式。错误{e}, tool_call_idrequest.tool_call[id] ) # 不会执行到这里使用中间件pythonagent create_agent( modelgpt-4, tools[search, calculator], middleware[retry_on_tool_failure] )4.2 类式中间件复杂场景适合多钩子、需维护状态的场景。继承AgentMiddleware并实现对应钩子方法。需求根据用户技术水平动态调整模型和工具。专家用户使用高级模型 复杂工具新手使用轻量模型 基础工具。pythonfrom 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( modelgpt-4, tools[advanced_search, data_analysis, calculator] ) else: # 新手用户轻量模型 基础工具 request request.override( modelgpt-3.5-turbo, tools[simple_search, basic_calculator] ) return handler(request)使用中间件并在调用时传入上下文pythonagent create_agent( modelgpt-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。pythonfrom 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按注册顺序的逆序执行pythonagent create_agent( modelgpt-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 图可嵌入更大的工作流中中间件钩子依然有效。pythonfrom langgraph.graph import StateGraph, START from langchain.agents import create_agent, AgentState # 创建带中间件的 Agent email_agent create_agent( modelclaude-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 协作或预处理-处理-后处理的流水线场景。七、总结与最佳实践何时使用中间件生产级 Agent需要日志、重试、监控、降级复杂上下文管理动态提示词、对话摘要、PII 脱敏安全与合规工具权限控制、人工审批、数据隐私多租户或个性化根据用户角色动态调整模型和工具