PythonAI
LangChain Agent 中间件详解：提升可靠性与可控性
介绍 LangChain Agent 中间件机制，涵盖内置中间件如对话总结、人工介入、限流、重试及降级功能，并详解基于装饰器和类的自定义中间件开发方法。通过钩子（Hook）与执行点概念，结合 AgentState 和 Runtime 对象，帮助开发者构建高可靠、可控的大模型应用。
介绍 LangChain Agent 中间件机制，涵盖内置中间件如对话总结、人工介入、限流、重试及降级功能，并详解基于装饰器和类的自定义中间件开发方法。通过钩子（Hook）与执行点概念，结合 AgentState 和 Runtime 对象，帮助开发者构建高可靠、可控的大模型应用。
在构建复杂的 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(...)] # 按顺序执行
)
langchain 为我们提供了多个常用的现成的中间件，我们可以根据任务选择使用。这里讲解一些基础中间件，复杂中间件将在后续内容中讲解。
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 条原始消息
)
]
)
配置：
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 # 无需人工介入
}
)
]
)
配置：
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"
)
]
)
配置：
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) # 特定工具
]
)
场景：主模型故障时自动切换到备用模型，提升可靠性。配置的备用模型支持字符串或 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" # 第二备用
)
]
)
配置：
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"
)
]
)
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 提供了两种实现方式：基于装饰器（快速简洁）和基于类（功能强大）的方式定义我们的中间件。
静态运行时上下文（Runtime）：一个对象，提供代理运行时的环境信息，如当前线程 ID、用户上下文等。
装饰器方式适用于只需要一个钩子的简单中间件。LangChain 提供了多个装饰器，分别对应不同的执行点。
中间件返回的 message 列表，跟工具是一样的，可以兼具同时删除消息，和添加新消息到 messages 列表后面（默认）。
from langchain.agents.middleware import after_model
from langchain.messages import AIMessage
@after_model(can_jump_to=["end"]) # 声明允许跳转到 end 节点
def block_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"}
return None
这些装饰器定义的函数会在特定的执行点（位置、钩子）被调用，不能控制是否调用下游中间件或实际处理器，只能读取/修改状态。
| 装饰器 | 执行时机 | 函数签名 | 典型用途 |
|---|---|---|---|
| @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 |
包装式装饰器，我们可以包裹实际的调用处理程序（handler），完全控制其执行。
| 装饰器 | 作用范围 | 函数签名 | 典型用途 |
|---|---|---|---|
| @wrap_model_call | 每次模型调用 | (request: ModelRequest, handler: Callable) -> ModelResponse | 重试、缓存、动态模型切换 |
| @wrap_tool_call | 每次工具调用 | (request: ToolRequest, handler: Callable) -> ToolResponse | 重试、模拟、权限检查 |
| @dynamic_prompt | 生成动态系统提示词 | (request: ModelRequest) -> str | 个性化提示词、根据上下文调整系统消息 |
参数说明：
from langchain.agents.middleware import wrap_model_call
import time
@wrap_model_call
def retry_with_backoff(request, handler):
max_retries = 3
for attempt in range(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")
使用注意
当中间件需要多个钩子、复杂配置或复用性时，应继承 AgentMiddleware 基类。
在子类中重写以下方法即可，那这样一个类中间件实例化对象就包含了多个钩子
from langchain.agents.middleware import AgentMiddleware
class MyMiddleware(AgentMiddleware):
def before_agent(self, state, runtime): # 返回 None 或字典
pass
def before_model(self, state, runtime):
pass
def after_model(self, state, runtime):
pass
def after_agent(self, state, runtime):
pass
重写 wrap_model_call 和 wrap_tool_call 方法，包装式方法中，你必须显式调用 handler(request) 来继续执行链，否则下游中间件和实际调用将被跳过。你可以调用多次（重试），也可以根据条件不调用（短路）
class MyMiddleware(AgentMiddleware):
def wrap_model_call(self, request, handler):
# 在调用前可以修改 request
# 可以选择是否调用 handler，以及调用几次
response = handler(request)
# 在调用后可以修改 response
return response
def wrap_tool_call(self, request, handler):
# 类似 wrap_model_call，但 request 是 ToolRequest 类型
pass
可以通过 init 接收配置参数，然后保存为实例变量，在钩子方法中使用。
class RetryMiddleware(AgentMiddleware):
def __init__(self, max_retries=3):
self.max_retries = max_retries
super().__init__()
agent 全局维护的状态对象，它是一个 TypedDict，至少包含以下字段：
中间件可以通过返回字典来更新状态，例如添加新字段或修改现有字段。但需注意：某些字段（如 messages）的修改需要遵循不可变原则或正确追加。
提供运行时的上下文信息，常用属性：
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online