大模型开发手记（九）：LangChain Agent 中间件-提升Agent的可靠性与可控性
目录
前言
- 在构建复杂的LangChain Agent时，我们常常面临上下文超限、成本失控、工具调用失败、敏感信息泄露等问题。LangChain提供了一套生产就绪的预构建中间件，像乐高积木一样，让我们轻松为Agent添加对话总结、人工审核、调用限流、自动重试等能力。
- 这篇博客主要讲述中间件的作用、有哪些内置的中间件、怎么自己定义中间等
一、中间件是什么？
- 中间件（Middleware）是LangChain Agent执行流程中的钩子函数，可以在模型调用前、工具执行前后、状态更新时插入自定义逻辑。预构建中间件覆盖了从成本控制到安全合规的常见场景，且支持灵活配置。
所有中间件通过create_agent的middleware参数传入：
from langchain.agents import create_agent agent = create_agent( model="gpt-4.1", tools=[...], middleware=[Middleware1(...), Middleware2(...)]# 按顺序执行)二、通用中间件详解（Provider-agnostic）
langchain为我们提供了多个常用的现成的中间件，我们可以根据任务选择使用,这里我讲下比较基础的一些中间件，一些比较复杂的中间件，我会放后面博客内容降价
2.1 对话总结（SummarizationMiddleware）
- 场景：长对话易超Token限制，自动总结旧消息，保留最新上下文。
- 配置：
- trigger：触发总结的条件（如token数≥4000或消息数≥6）
- keep：总结后保留的内容（如保留最近20条消息）
- model：用于生成总结的模型（通常用更便宜的模型）
from langchain.agents import create_agent from langchain.agents.middleware import SummarizationMiddleware agent = create_agent( model="gpt-4.1", tools=[weather_tool, calculator_tool], middleware=[ SummarizationMiddleware( model="gpt-4.1-mini",# 用迷你模型做总结 trigger=[("tokens",4000),("messages",6)],# 任一条件触发 keep=("messages",20)# 保留最近20条原始消息)])2.2 人工介入（HumanInTheLoopMiddleware）
- 场景：高风险操作（如发送邮件、修改数据库）需人工审批，这个中间件后续我会详细讲解，在这大家有个印象就可以，知道有这么个中间件能实现人工干预agent执行过程即可。
- interrupt_on：指定哪些工具需要人工介入，及允许的决策（approve/edit/reject）
- 必须搭配checkpointer（如InMemorySaver）实现状态暂停/恢复
配置：
from langchain.agents import create_agent from langchain.agents.middleware import HumanInTheLoopMiddleware from langgraph.checkpoint.memory import InMemorySaver agent = create_agent( model="gpt-4.1", tools=[read_email_tool, send_email_tool], checkpointer=InMemorySaver(), middleware=[ HumanInTheLoopMiddleware( interrupt_on={"send_email_tool":{"allowed_decisions":["approve","edit","reject"]},"read_email_tool":False# 无需人工介入})])2.3 模型调用限流（ModelCallLimitMiddleware）
- 场景：防止Agent无限循环或API费用失控。
- thread_limit：整个会话的大模型最大调用次数，一个会话包含多次对话（一个对话就是一次invoke，可以看下我的短期记忆讲的内容，checkpoint）
- run_limit：单次用户请求内的最大调用次数，一次对话，一个invoke
- exit_behavior：超出限制时，是"end"（优雅终止）还是"error"（抛异常）
配置：
from langchain.agents.middleware import ModelCallLimitMiddleware agent = create_agent( model="gpt-4.1", checkpointer=InMemorySaver(),# 跨多轮对话限制需要checkpointer tools=[], middleware=[ ModelCallLimitMiddleware( thread_limit=10,# 整个对话最多调10次模型 run_limit=5,# 单次请求最多调5次 exit_behavior="end")])2.4 工具调用限流（ToolCallLimitMiddleware）
- 场景：针对特定工具设置调用上限（如搜索引擎、数据库查询）。
- tool_name：指定工具名称（不指定则全局限制）
- thread_limit / run_limit：同模型限流
- exit_behavior：支持"continue"（返回错误让Agent继续）、“error”、“end”
配置：
from langchain.agents.middleware import ToolCallLimitMiddleware agent = create_agent( model="gpt-4.1", tools=[search_tool, db_tool], middleware=[ ToolCallLimitMiddleware(thread_limit=20, run_limit=10),# 全局限制 ToolCallLimitMiddleware(tool_name="search", thread_limit=5, run_limit=3)# 特定工具])2.5 模型降级（ModelFallbackMiddleware）
场景：主模型故障时自动切换到备用模型，提升可靠性。配置的备用模型支持字符串或BaseChatModel实例
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"# 第二备用)])2.6 工具重试（ToolRetryMiddleware）
- 场景：工具调用因网络波动失败时，自动指数退避重试。
- max_retries：最大重试次数
- retry_on：指定哪些异常需重试（如ConnectionError）
- on_failure：重试耗尽后，“return_message”（返回错误消息）或"raise"（抛异常）
配置：
from langchain.agents.middleware import ToolRetryMiddleware agent = create_agent( model="gpt-4.1", tools=[api_tool], middleware=[ ToolRetryMiddleware( max_retries=3, backoff_factor=2.0,# 指数退避：1s, 2s, 4s retry_on=(ConnectionError, TimeoutError), on_failure="return_message")])2.7 模型重试（ModelRetryMiddleware）
- 场景：模型API调用失败（如限流、超时）时自动重试。
配置：与工具重试类似，支持自定义异常过滤和失败处理。
from langchain.agents.middleware import ModelRetryMiddleware agent = create_agent( model="gpt-4.1", tools=[], middleware=[ ModelRetryMiddleware( max_retries=4, retry_on=lambda e:hasattr(e,"status_code")and e.status_code ==429,# 仅重试限流 on_failure="continue"# 返回错误消息让Agent继续)])三、自定义中间件
- 上述是langchain官方提供的一些通用的中间件，这些中间件会在固定的流程节点执行固定的程序，我们也可以定义开发自己的中间件，无非确定两个点，在哪个节点（位置），执行什么程序。
- langchain 提供了两种实现方式：基于装饰器（快速简洁）和基于类（功能强大）的方式定义我们的中间件。
3.1 核心概念：钩子（Hook）与执行点
- 钩子（Hook）：代理执行流程中预定义的位置，钩子就是执行点，中间件可以在这里挂载代码（钩子函数）。LangChain 中间件主要提供两类钩子：
- 节点式钩子（Node-style）：在固定的顺序点运行，如“模型调用前”、“模型调用后”。多个中间件的同类型节点式钩子会按添加顺序依次执行。
- 包装式钩子（Wrap-style）：包裹整个调用过程（如一次模型调用），可以完全控制是否执行、执行几次以及如何处理结果。
- 状态（AgentState,）：一个字典，存储当前代理运行的全部上下文，包括消息列表、中间变量等。中间件可以通过读取和修改状态来影响后续流程。
静态运行时上下文（Runtime）：一个对象，提供代理运行时的环境信息，如当前线程 ID、用户上下文等。
3.2 基于装饰器的中间件
装饰器方式适用于只需要一个钩子的简单中间件。LangChain 提供了多个装饰器，分别对应不同的执行点。
3.2.1 节点式装饰器
- 返回值说明
- 返回 None：正常流程，继续执行下一个中间件或实际处理器。
- 返回一个字典：该字典会与当前状态合并（state.update(returned_dict)），可用于添加或修改状态中的字段。特殊键 “jump_to” 可以强制跳转到指定节点（如 “end” 提前结束代理）。
中间件返回的message列表，跟工具是一样的，可以兼具同时删除消息，和添加新消息到messages列表后面（默认）。
from langchain.agents.middleware import after_model from langchain.messages import AIMessage @after_model(can_jump_to=["end"])# 声明允许跳转到 end 节点defblock_sensitive_output(state, runtime): last_msg = state["messages"][-1]if"password"in last_msg.content.lower():# 修改最后一条消息，并跳转到结束节点return{"messages":[AIMessage("I cannot reveal that information.")],"jump_to":"end"}returnNone这些装饰器定义的函数会在特定的执行点（位置、钩子）被调用，不能控制是否调用下游中间件或实际处理器，只能读取/修改状态。
| 装饰器 | 执行时机 | 函数签名 | 典型用途 |
|---|---|---|---|
| @before_agent | 整个代理调用开始时（仅一次） | (state: AgentState, runtime: Runtime) -> dict | None |
| @before_model | 每次模型调用之前 | (state: AgentState, runtime: Runtime) -> dict | None |
| @after_model | 每次模型调用之后 | (state: AgentState, runtime: Runtime) -> dict | None |
| @after_agent | 整个代理调用结束时（最多一次） | (state: AgentState, runtime: Runtime) -> dict | None |
3.2.2 包装式装饰器
包装式装饰器，我们可以包裹实际的调用处理程序（handler），完全控制其执行。
| 装饰器 | 作用范围 | 函数签名 | 典型用途 |
|---|---|---|---|
| @wrap_model_call | 每次模型调用 | (request: ModelRequest, handler: Callable) -> ModelResponse | 重试、缓存、动态模型切换 |
| @wrap_tool_call | 每次工具调用 | (request: ToolRequest, handler: Callable) -> ToolResponse | 重试、模拟、权限检查 |
| @dynamic_prompt | 生成动态系统提示词 | (request: ModelRequest) -> str | 个性化提示词、根据上下文调整系统消息 |
参数说明：
- request：包含调用所需的信息，如 request.messages（消息列表）、request.runtime（运行时）、request.model（当前模型）等。
handler：下一个中间件或实际处理器的可调用对象。调用 handler(request) 会继续执行包装链，最终返回响应。
from langchain.agents.middleware import wrap_model_call import time @wrap_model_calldefretry_with_backoff(request, handler): max_retries =3for attempt inrange(max_retries):try:return handler(request)except Exception as e:if attempt == max_retries -1:raise# 最后一次失败则向上抛出异常 wait =(2** attempt)# 指数退避: 1, 2, 4 秒 time.sleep(wait)print(f"Retry {attempt+1}/{max_retries} after {wait}s")使用注意
- 装饰器函数本身会作为中间件对象，直接传入 middleware 列表。
- 多个中间件的执行顺序：节点式钩子按列表顺序执行；包装式钩子则按列表顺序从外到内包装（即列表最后一个包装器最先接触 handler，最内层执行）。
3.3 基于类的中间件
当中间件需要多个钩子、复杂配置或复用性时，应继承 AgentMiddleware 基类。
3.3.1 节点式钩子方法
在子类中重写以下方法即可，那这样一个类中间件实例化对象就包含了多个钩子
from langchain.agents.middleware import AgentMiddleware classMyMiddleware(AgentMiddleware):defbefore_agent(self, state, runtime):# 返回 None 或字典passdefbefore_model(self, state, runtime):passdefafter_model(self, state, runtime):passdefafter_agent(self, state, runtime):pass3.3.2 包装式钩子方法
重写 wrap_model_call 和 wrap_tool_call 方法，包装式方法中，你必须显式调用 handler(request) 来继续执行链，否则下游中间件和实际调用将被跳过。你可以调用多次（重试），也可以根据条件不调用（短路）
classMyMiddleware(AgentMiddleware):defwrap_model_call(self, request, handler):# 在调用前可以修改 request# 可以选择是否调用 handler，以及调用几次 response = handler(request)# 在调用后可以修改 responsereturn response defwrap_tool_call(self, request, handler):# 类似 wrap_model_call，但 request 是 ToolRequest 类型pass3.3.3 类中间件的初始化
可以通过 init 接收配置参数，然后保存为实例变量，在钩子方法中使用。
classRetryMiddleware(AgentMiddleware):def__init__(self, max_retries=3): self.max_retries = max_retries super().__init__()四、深入理解 AgentState 和 Runtime
4.1 AgentState
agent全局维护的状态对象，它是一个 TypedDict，至少包含以下字段：
- messages：List[BaseMessage]，当前对话的所有消息（包括人类消息、AI 消息、工具消息等）。
- intermediate_steps：记录工具执行的历史。
- 还可能包含其他自定义字段（如 todos、jump_to 等）。
中间件可以通过返回字典来更新状态，例如添加新字段或修改现有字段。但需注意：某些字段（如 messages）的修改需要遵循不可变原则或正确追加。
4.2 Runtime
提供运行时的上下文信息，常用属性：
- runtime.context：一个字典，包含了调用 agent.invoke(…, config={“configurable”: {…}}) 时传入的自定义上下文数据。
- runtime.thread_id：当前线程 ID，可用于区分不同会话。
