一、中间件概述
1.1 是什么
中间件（Middleware）是 LangChain 中一个基于拦截器模式的钩子机制，允许你在 Agent 的整个生命周期中插入自定义逻辑。它是 Agent 架构中最重要的设计之一，提供了强大的扩展点。
1.2 解决的问题
可观测性：记录 Agent 每一步的状态，便于监控和排查；
流程控制：在模型调用、工具调用前后"插入一脚"；
动态行为：动态注入提示词、切换模型、强制跳转节点；
复用性：把通用逻辑打包成可复用的中间件组件。
1.3 底层原理
中间件本质是围绕 LangGraph 的 pregel 图模型，在节点周围做包装（wrap）和钩子（hook）。LangChain 提供了两大类接口：
Wrap-style 中间件：
wrap_model_call/wrap_tool_call，在核心操作"前后"都做事；Hook-style 中间件：
before_model/after_model，在操作"前"或"后"单独做事。
1.4 两种写法
每个中间件都可以用 装饰器 或 类（继承 AgentMiddleware） 两种方式实现：
装饰器：
@wrap_model_call、@wrap_tool_call、@before_model、@after_model类：继承
AgentMiddleware，实现同名方法
两种写法底层等价——装饰器内部也会创建一个
AgentMiddleware实例。
二、内置中间件（开箱即用）
LangChain 自带了一些常用中间件，直接导入即可使用。
2.1 TodoListMiddleware（任务清单）
在模型调用时，自动插入一张"待办事项清单"提示词，要求模型在回答前先规划步骤、逐条勾选，让推理过程更结构化、透明。
from langchain.agents import create_agent
from langchain.agents.middleware import TodoListMiddleware
agent = create_agent(
model=model,
middleware=[TodoListMiddleware()]
)
response = agent.invoke({
"messages": [{"role": "user", "content": "写一首关于春天的诗"}]
})模型会先输出一张任务清单（如 "1. 列意象 → 2. 组织语言 → 3. 成诗"），再给出正文。
2.2 其他内置中间件类型
课件中还提到 LangChain 提供了处理错误/重试、上下文管理等场景的内置中间件。使用前建议查阅官方文档确认最新列表：
三、自定义中间件（核心）
自定义中间件分为两大类、四种钩子，覆盖 Agent 生命周期的四大关键节点。
3.1 Hook-style：before_model / after_model
分别在模型调用前/后触发，用于日志、提示词注入、流程控制等。
装饰器实现（审计日志示例）：
from langchain.agents.middleware import before_model, after_model, AgentState
from langgraph.runtime import Runtime
from loguru import logger
@before_model
def before_log(state: AgentState, runtime: Runtime) -> dict | None:
logger.info("调用模型前消息数量: {}", len(state["messages"]))
return None
@after_model
def after_log(state: AgentState, runtime: Runtime) -> dict | None:
logger.info("调用模型后消息数量：{}", len(state["messages"]))
return None
agent = create_agent(model=model, middleware=[before_log, after_log])类实现：
from langchain.agents.middleware import AgentMiddleware, AgentState
class AuditMiddleware(AgentMiddleware):
def __init__(self, logger):
super().__init__()
self.logger = logger
def before_model(self, state: AgentState, runtime: Runtime) -> dict | None:
self.logger.info("before_model: {}", len(state["messages"]))
return None
def after_model(self, state: AgentState, runtime: Runtime) -> dict | None:
self.logger.info("after_model: {}", len(state["messages"]))
return None3.2 流程控制：jump_to 强制跳转
Hook 钩子有个高级能力——通过返回值里的 jump_to 字段，强制把执行流程跳到指定节点，不用经过正常推理链路。常用目标节点："model"、"tools"、"end"。
典型场景演示（三个中间件 + 一个对照）：
from langchain.agents.middleware import before_model, after_model
from langchain.messages import AIMessage, SystemMessage
@before_model
def force_tool_first(state, runtime) -> dict | None:
text = state["messages"][-1].content
if "direct tool" in text.lower():
print("[MIDDLEWARE] before_model: jump_to='tools'")
fake_tool_call = AIMessage(
content="人工构造的消息",
tool_calls=[{"name": "get_news", "args": {}, "id": "call_force_001"}],
)
return {"messages": [fake_tool_call], "jump_to": "tools"}
return None
@after_model
def retry_with_extra_instruction(state, runtime) -> dict | None:
# 找到用户输入，判断条件并防止无限重跳
if "retry model" in user_text.lower() and not already_injected:
return {
"messages": [SystemMessage("你必须以【二次回答】开头，并且只用一句话回答。")],
"jump_to": "model",
}
return None
@before_model
def overflow_context_processor(state, runtime) -> dict | None:
if "overflow" in state["messages"][-1].content:
return {"messages": [AIMessage("上下文窗口溢出，终止")], "jump_to": "end"}
return None
agent = create_agent(
model=model,
tools=[get_news],
middleware=[force_tool_first, retry_with_extra_instruction, overflow_context_processor],
)基于类实现时，需要用
@hook_config(can_jump_to=[...])装饰器为jump_to传参：
class MyMiddleware(AgentMiddleware):
@hook_config(can_jump_to=["tools", "end"])
def before_model(self, state, runtime) -> dict | None:
...3.3 Wrap-style：wrap_model_call（包裹模型调用）
在模型调用的前后都能做事，是最灵活的一类中间件。wrap（包裹）意味着你同时拿到"请求"和"响应"，可以在中间做拦截、修改、重试、缓存。
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
@wrap_model_call
def wrap_model_call_middleware(
request: ModelRequest, # 即将发给大模型的所有请求数据
handler, # 下一个中间件或真正的大模型调用
) -> ModelResponse | None:
# 调用前：动态篡改最后一条消息
request.messages[-1].content += " -> wrap_model_call_before <- "
# 真正调用模型（产生真实 Token 消耗）
response = handler(request)
# 调用后：篡改返回内容
response.result[0].content += " -> wrap_model_call_after <- "
return response使用场景：拦截、重试、缓存模型调用。
场景 1：重试逻辑（指数退避）
@wrap_model_call
def retry_model(request, handler) -> ModelResponse:
max_retries = 3
for attempt in range(max_retries):
try:
return handler(request)
except Exception as e:
if attempt == max_retries - 1:
raise
time.sleep(2 ** attempt) # 指数退避场景 2：响应缓存（按请求内容生成 md5 键，命中则直接返回）
class ModelCache:
def __init__(self):
self.cache = {}
def create_hook(self):
@wrap_model_call
def cache_model(request, handler) -> ModelResponse:
cache_key = hashlib.md5(
json.dumps({"messages": [str(m) for m in request.messages],
"system": str(request.system_message)}).encode()
).hexdigest()
if cache_key in self.cache:
return self.cache[cache_key] # 缓存命中
response = handler(request)
self.cache[cache_key] = response
return response
return cache_model场景 3：动态修改系统提示（用
request.override()优雅地替换）
@wrap_model_call
def add_context(request, handler) -> ModelResponse:
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
new_system_message = SystemMessage(
content=f"{request.system_message.content or ''}\n当前时间：{current_time}\n语言偏好：中文"
)
modified_request = request.override(system_message=new_system_message)
return handler(modified_request)3.4 Wrap-style：wrap_tool_call（包裹工具调用）
在工具调用的前后做事，用于监控、重试、修改工具执行/参数。
from langchain.agents.middleware import wrap_tool_call
from langchain.tools.tool_node import ToolCallRequest
@wrap_tool_call
def wrap_tool_call_middleware(request: ToolCallRequest, handler) -> ToolMessage | Command:
result = handler(request) # 第一次调用（原始参数）
request.tool_call["args"]["is_forcast"] = True # 篡改参数
result = handler(request) # 第二次调用（修改后参数）
return result监控工具执行示例：
@wrap_tool_call
def monitor_tool(request, handler):
tool_name = request.tool_call["name"]
start_time = time.time()
try:
result = handler(request)
print(f"✅ {tool_name} 成功，耗时 {time.time()-start_time:.2f}s")
return result
except Exception as e:
print(f"❌ {tool_name} 失败：{e}")
raise3.5 参数说明（Wrap-style 通用）
request：被封装的请求对象，可能是模型请求（ModelRequest）或工具请求（ToolCallRequest）；handler：处理器，用于继续执行请求并返回结果，本质上代表"下一个中间件或最终的服务调用"。
四、装饰器 vs 类：怎么选？
装饰器也能用"工厂函数返回多个装饰器"的方式实现多钩子，但会把一个逻辑上属于同一中间件的行为拆成多个独立函数，可维护性不如类写法。
五、中间件执行顺序（重要！）
中间件是乱序定义、顺序生效的——执行顺序只取决于传入 create_agent 的列表顺序。具体规律如"洋葱模型"：
before_model：按列表声明的顺序执行（1 → 2 → 3）；after_model：按列表声明顺序倒序执行（3 → 2 → 1）；wrap_model_call：先声明的包在最外层（洋葱结构）。
以三个中间件（1、2、3）为例，最终消息内容的追加顺序是：
原始消息 → before_model-1 → before_model-2 → before_model-3
→ wrap_model-before-1 → wrap_model-before-2 → wrap_model-before-3
【大模型调用】
→ wrap_model-after-3 → wrap_model-after-2 → wrap_model-after-1
→ after_model-3 → after_model-2 → after_model-1记忆口诀：before 正序、after 倒序、wrap 先包外。
六、总结
中间件 = LangChain 1.0 的拦截器钩子，扩展 Agent 的四大扩展点：模型前、模型后、模型包裹、工具包裹。
两类接口：Wrap-style（
wrap_model_call/wrap_tool_call，前后都管）和 Hook-style（before_model/after_model，只前或只后）。两种写法：装饰器（单钩子首选）与类（多钩子、复杂配置首选）。
高阶能力
jump_to：在 hook 中强制跳转到model/tools/end，实现"省思考调用、二次生成、熔断终止"等流程控制。执行顺序：严格遵循传入顺序，before 正序、after 倒序、wrap 洋葱式包裹。
典型应用：日志审计、重试、缓存、动态注入提示词、工具监控、流程熔断。
评论区