跳到主要内容 LangChain 智能体中间件如何参与 Agent、Model 和 Tool 交互 | 极客日志
Python AI 算法
LangChain 智能体中间件如何参与 Agent、Model 和 Tool 交互 LangChain 中间件是基于 AgentMiddleware 的可插拔钩子系统,允许在不修改核心逻辑的情况下拦截 Agent 执行流程。主要分为生命周期拦截器和调用包装器。通过为 Pregel 对象添加节点和通道实现拦截,支持同步和异步方法。中间件可封装工具,并在模型或工具调用前后修改请求状态,解决无绑定可执行对象的工具调用问题。
指针猎手 发布于 2026/3/21 更新于 2026/4/18 3 浏览LangChain 智能体本质论:中间件是如何参与 Agent、Model 和 Tool 三者交互的?
LangChain 的中间件(Middleware)是围绕 Agent 执行流程构建的'可插拔钩子系统'。它允许开发者在不修改核心逻辑的情况下,在执行的关键节点(如输入处理、模型调用前后、输出解析等)对数据流进行拦截、修改或验证。中间件类型以 AgentMiddleware 为基类。
1. AgentMiddleware
AgentMiddleware 是一个泛型类型,两个泛型参数分别代表状态和静态上下文的类型,我们可以利用 state_schema 字段得到状态类型。它的 name 属性返回中间件的名称,默认返回的是当前的类名。
class AgentMiddleware (Generic [StateT, ContextT]):
state_schema: type [StateT] = cast("type[StateT]" , _DefaultAgentState)
tools: Sequence [BaseTool]
@property
def name (self ) -> str :
return self .__class__.__name__
def before_agent (self, state: StateT, runtime: Runtime[ContextT] ) -> dict [str , Any ] | None :
pass
async def abefore_agent (self, state: StateT, runtime: Runtime[ContextT] ) -> dict [str , Any ] | None :
pass
def before_model (self, state: StateT, runtime: Runtime[ContextT] ) -> dict [str , Any ] | None :
pass
( ) -> [ , ] | :
( ) -> [ , ] | :
( ) -> [ , ] | :
( ) -> [ , ] | :
( ) -> [ , ] | :
( ) -> ModelCallResult:
NotImplementedError(...)
( ) -> ModelCallResult:
NotImplementedError(...)
( ) -> ToolMessage | Command[ ]:
NotImplementedError(...)
( ) -> ToolMessage | Command[ ]:
NotImplementedError(...)
微信扫一扫,关注极客日志 微信公众号「极客日志」,在微信中扫描左侧二维码关注。展示文案:极客日志 zeeklog
相关免费在线工具 加密/解密文本 使用加密算法(如AES、TripleDES、Rabbit或RC4)加密和解密文本明文。 在线工具,加密/解密文本在线工具,online
RSA密钥对生成器 生成新的随机RSA私钥和公钥pem证书。 在线工具,RSA密钥对生成器在线工具,online
Mermaid 预览与可视化编辑 基于 Mermaid.js 实时预览流程图、时序图等图表,支持源码编辑与即时渲染。 在线工具,Mermaid 预览与可视化编辑在线工具,online
curl 转代码 解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online
Base64 字符串编码/解码 将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online
Base64 文件转换器 将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online
async
def
abefore_model
self, state: StateT, runtime: Runtime[ContextT]
dict
str
Any
None
pass
def
after_model
self, state: StateT, runtime: Runtime[ContextT]
dict
str
Any
None
pass
async
def
aafter_model
self, state: StateT, runtime: Runtime[ContextT]
dict
str
Any
None
pass
def
after_agent
self, state: StateT, runtime: Runtime[ContextT]
dict
str
Any
None
pass
async
def
aafter_agent
self, state: StateT, runtime: Runtime[ContextT]
dict
str
Any
None
pass
def
wrap_model_call
self,
request: ModelRequest,
handler: Callable [[ModelRequest], ModelResponse],
raise
async
def
awrap_model_call
self,
request: ModelRequest,
handler: Callable [[ModelRequest], Awaitable[ModelResponse]],
raise
def
wrap_tool_call
self,
request: ToolCallRequest,
handler: Callable [[ToolCallRequest], ToolMessage | Command[Any ]],
Any
raise
async
def
awrap_tool_call
self,
request: ToolCallRequest,
handler: Callable [[ToolCallRequest], Awaitable[ToolMessage | Command[Any ]]],
Any
raise
通过前面的介绍我们知道,在调用 create_agent 函数时可以利用 tools 参数进行工具注册,其实工具也可以利用 tools 字段封装到中间件中。中间件被注册时,其封装的工具也会一并予以注册。换句话说,create_agent 方法内部会读取所有注册中间件的 tools 字段存储的工具,连同利用 tools 参数直接注册的工具一起处理。虽然 Agent 定义了众多方法,但我们可以将它们划分为如下两类:
生命周期拦截器 :在 Agent 和 Model 执行前后调用,包括 before_agent/before_model/after_agent/after_model 及其异步版本。
调用包装器 :对 Model 和 Tool 的调用进行包装;包括 wrap_model_call/wrap_tool_call 及其异步版本。
2. 生命周期拦截器 对于一个利用 create_agent 函数创建的 Agent,在没有任何中间件注册的情况下,它本质上是由 model 和 tools 两个核心节点组成的 Pregel 对象。注册中间件的生命周期拦截器方法针对 Agent 和 Model 调用前后的拦截,是通过为 Pregel 对象添加额外节点和通道来实现的。
from langchain.agents import create_agent
from dotenv import load_dotenv
from langchain.agents.middleware.types import AgentState
from langchain_openai import ChatOpenAI
from PIL import Image as PILImage
from langchain.agents.middleware import AgentMiddleware
from typing import Any
from langgraph.runtime import Runtime
import io
class FooMiddleware (AgentMiddleware ):
def before_agent (self, state: AgentState[Any ], runtime: Runtime[None ] ) -> dict [str , Any ] | None :
return super ().before_agent(state, runtime)
def before_model (self, state: AgentState[Any ], runtime: Runtime[None ] ) -> dict [str , Any ] | None :
return super ().before_model(state, runtime)
def after_agent (self, state: AgentState[Any ], runtime: Runtime[None ] ) -> dict [str , Any ] | None :
return super ().after_agent(state, runtime)
def after_model (self, state: AgentState[Any ], runtime: Runtime[None ] ) -> dict [str , Any ] | None :
return super ().after_model(state, runtime)
load_dotenv()
def test_tool ():
"""A test tool"""
pass
agent = create_agent(
model=ChatOpenAI(model="gpt-5.2-chat" ),
tools=[test_tool],
middleware=[FooMiddleware()]
)
payload = agent.get_graph(xray=True ).draw_mermaid_png()
PILImage.open (io.BytesIO(payload)).show()
print ("channels:" )
for (name, chan) in agent.channels.items():
print (f"\t[{chan.__class__.__name__} ]{name} " )
print ("trigger_to_nodes" )
for (name, nodes) in agent.trigger_to_nodes.items():
print (f"\t{name} : {nodes} " )
从上图可以看出,注册的中间件为 Agent 添加了四个节点,这四个节点对应于我们重写的四个方法,节点所在的位置体现四个方法的执行顺序:FooMiddleware.before_agent -> FooMiddleware.before_model -> FooMiddleware.after_model -> FooMiddleware.after_agent。而且 FooMiddleware.before_model 可以实现针对'tools'节点的跳转,'tools'执行结束后又会被 FooMiddleware.before_model 拦截。
演示程序还输出了通道列表,以及节点与订阅通道之间的映射关系。从如下的输出结果可以看出,上述四个节点各自有独立定义的通道。
channels:
[BinaryOperatorAggregate ]messages
[EphemeralValue ]jump_to
[LastValue ]structured_response
[EphemeralValue ]__start__
[Topic ]__pregel_tasks
[EphemeralValue ]branch:to:model
[EphemeralValue ]branch:to:tools
[EphemeralValue ]branch:to:FooMiddleware.before_agent
[EphemeralValue ]branch:to:FooMiddleware.before_model
[EphemeralValue ]branch:to:FooMiddleware.after_model
[EphemeralValue ]branch:to:FooMiddleware.after_agent
trigger_to_nodes
__start__: ['__start__' ]
branch:to:model: ['model' ]
branch:to:tools: ['tools' ]
branch:to:FooMiddleware.before_agent: ['FooMiddleware.before_agent' ]
branch:to:FooMiddleware.before_model: ['FooMiddleware.before_model' ]
branch:to:FooMiddleware.after_model: ['FooMiddleware.after_model' ]
branch:to:FooMiddleware.after_agent: ['FooMiddleware.after_agent' ]
如果我们采用如下的方式再注册一个中间件 BarMiddleware:
class BarMiddleware (AgentMiddleware ):
def before_agent (self, state: AgentState[Any ], runtime: Runtime[None ] ) -> dict [str , Any ] | None :
return super ().before_agent(state, runtime)
agent = create_agent(
model=ChatOpenAI(model="gpt-5.2-chat" ),
tools=[test_tool],
middleware=[FooMiddleware(), BarMiddleware()]
)
在 Agent 新的拓扑结构中,优化多出四个针对 BarMiddleware 的节点。
3. 调用包装器 AgentMiddleware 提供了四个方法(wrap_model_call、awrap_model_call、wrap_agent_call 和 awrap_agent_call),分别用于包装针对 Model 和 Tool 的同步和异步调用。对于作为 Pregel 的 Agent 来说,针对模型和工具的调用是由'model'和'tools'节点发出的,所以利用中间件对调用的封装也在这两个节点中完成。
3.1 针对模型调用的包装 常规的模型调用会返回一个 AIMessage 消息。如果采用基于 ToolStrategy 的结构化输出,除了返回格式化的输出外,还会涉及格式化工具生成的 ToolMessage,它们被封装在一个 ModelResponse 对象里。所以表示模型调用结果的 ModelResult 类型是 ModelResponse 和 AIMessage 这两个类型的联合。
@dataclass
class ModelResponse :
result: list [BaseMessage]
structured_response: Any = None
ModelCallResult: TypeAlias = ModelResponse | AIMessage
ModelRequest 表示模型调用的请求,我们从中可以得到 Chat 模型组件、请求消息列表、系统指令、注册的工具以及针对工具选择策略、结构化输出 Schema、状态、运行时和针对模型的设置。在绝大部分情况下,我们通过自定义中间件包装模型调用的目的都是为了更新上述的某一个或者多个请求元素,ModelRequest 利用 override 方法将一切变得简单。
@dataclass(init=False )
class ModelRequest :
model: BaseChatModel
messages: list [AnyMessage]
system_message: SystemMessage | None
tool_choice: Any | None
tools: list [BaseTool | dict [str , Any ]]
response_format: ResponseFormat[Any ] | None
state: AgentState[Any ]
runtime: Runtime[ContextT]
model_settings: dict [str , Any ] = field(default_factory=dict )
@property
def system_prompt (self ) -> str | None :
...
def override (self, **overrides: Unpack[_ModelRequestOverrides] ) -> ModelRequest:
class _ModelRequestOverrides (TypedDict, total=False ):
model: BaseChatModel
system_message: SystemMessage | None
messages: list [AnyMessage]
tool_choice: Any | None
tools: list [BaseTool | dict [str , Any ]]
response_format: ResponseFormat[Any ] | None
model_settings: dict [str , Any ]
state: AgentState[Any ]
由于模型调用的输入和输出类型分别是 ModelRequest 和 ModelResponse,所以被封装的针对模型的同步调用和异步调用可以表示成 Callable[[ModelRequest], ModelResponse] 和 Callable[[ModelRequest], Awaitable[ModelResponse]] 对象,wrap_model_call/awrap_model_call 方法的 handler 参数分别返回的就是这两个对象。
3.2 针对工具调用的包装 表示工具调用请求的 ToolRequest 类型定义如下,请求携带了模型生成的用于调用目标工具的 ToolCall 对象,代表工具自身的 BaseTool 对象,以及当前状态和工具运行时。ToolCallRequest 也提供了 override 方法实现针对这些请求元素的更新。
@dataclass
class ToolCallRequest :
tool_call: ToolCall
tool: BaseTool | None
state: Any
runtime: ToolRuntime
def override (self, **overrides: Unpack[_ToolCallRequestOverrides] ) -> ToolCallRequest:
class _ToolCallRequestOverrides (TypedDict, total=False ):
tool_call: ToolCall
tool: BaseTool
state: Any
调用工具执行的结构可以封装成一个 ToolCallRequest 反馈给模型,也可以返回一个 Command 对象实现对状态的更新和跳转,所以 wrap_tool_call/awrap_tool_call 方法中表示针对工具原始调用的 handler 参数分别是一个 Callable[[ToolCallRequest], ToolMessage|Command[Any]] 和 Callable[[ToolCallRequest], Awaitable[ToolMessage|Command[Any]]] 对象。
在介绍 create_agent 方法针对工具的注册时,我们曾经说过:除了以可执行对象或者 BaseTool 对象标识注册的工具外,我们还可以指定一个表示注册工具 JSON Schema 的字典。但是以这种方式注册的工具并没有绑定一个具体的可执行对象,所以默认是无法被调用的。我们可以采用中间件的方式来解决这个问题。
from langchain.agents import create_agent
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import ToolMessage
from langchain.agents.middleware import AgentMiddleware, ToolCallRequest
from langgraph.types import Command
from typing import Any , Callable
load_dotenv()
tool = {
"name" : "get_weather" ,
"description" : "Get weather information for given city" ,
"parameters" : {
"type" : "object" ,
"properties" : {"city" : {"type" : "string" }},
"required" : ["city" ]
}
}
class WeatherMiddleware (AgentMiddleware ):
def wrap_tool_call (
self,
request: ToolCallRequest,
handler: Callable [[ToolCallRequest], ToolMessage | Command[Any ]],
) -> ToolMessage | Command[Any ]:
tool_call = request.tool_call
if tool_call["name" ] == "get_weather" :
city = tool_call["args" ]["city" ]
return ToolMessage(content=f"It's sunny in {city} ." , tool_call_id=tool_call["id" ])
else :
return handler(request)
agent = create_agent(
model=ChatOpenAI(model="gpt-5.2-chat" ),
tools=[tool],
middleware=[WeatherMiddleware()],
)
result = agent.invoke(input ={"messages" : [{"role" : "user" , "content" : "What is the weather like in Suzhou?" }]})
for message in result["messages" ]:
message.pretty_print()
如上面的演示程序所示,我们注册的工具是一个字典,它表示注册工具的 JSON Schema。其中提供了工具的名称('get_weather')和参数结构(包含一个必需的名为'city'的字符串成员)。注册的 WeatherMiddleware 通过重写的 wrap_tool_call 实现了针对工具调用的拦截。如果是针对工具 get_weather 的调用,我们将天气信息封装成返回的 ToolMessage。程序执行后会以如下的方式输出消息历史:
================================ Human Message =================================
What is the weather like in Suzhou?
================================= Ai Message ==================================
Tool Calls: get_weather (call_LjastyaYNrovwMhSmvoJMcNz)
Call ID: call_LjastyaYNrovwMhSmvoJMcNz
Args: city: Suzhou
================================= Tool Message =================================
It's sunny in Suzhou.
================================= Ai Message ==================================
The weather in **Suzhou** is **sunny**. ☀️