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
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],
) -> ModelCallResult:
# ... implementation details ...
raise NotImplementedError(...)
async def awrap_model_call(
self,
request: ModelRequest,
handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
) -> ModelCallResult:
# ... implementation details ...
raise NotImplementedError(...)
def wrap_tool_call(
self,
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], ToolMessage | Command[Any]],
) -> ToolMessage | Command[Any]:
# ... implementation details ...
raise NotImplementedError(...)
async def awrap_tool_call(
self,
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]],
) -> ToolMessage | Command[Any]:
# ... implementation details ...
raise NotImplementedError(...)
通过前面的介绍我们知道，在调用 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)
# ... other methods ...
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**. ☀️
