![[LangChain智能体本质论-08]中间件是如何参与Agent、Model和Tool三者交互的？](http://pic.xiahunao.cn/yaotu/[LangChain智能体本质论-08]中间件是如何参与Agent、Model和Tool三者交互的？)
LangChain的中间件是围绕Agent执行流程构建的可插拔钩子系统。它允许开发者在不修改核心逻辑的情况下在执行的关键节点如输入处理、模型调用前后、输出解析等对数据流进行拦截、修改或验证。中间件类型以AgentMiddleware为基类。1. AgentMiddlewareAgentMiddleware是一个泛型类型两个泛型参数分别代表状态和静态上下文的类型我们可以利用state_schema字段得到状态类型。它的name属性返回中间件的名称默认返回的是当前的类名。classAgentMiddleware(Generic[StateT,ContextT]):state_schema:type[StateT]cast(type[StateT],_DefaultAgentState)tools:Sequence[BaseTool]propertydefname(self)-str:returnself.__class__.__name__defbefore_agent(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|None:passasyncdefabefore_agent(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|None:passdefbefore_model(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|None:passasyncdefabefore_model(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|None:passdefafter_model(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|None:passasyncdefaafter_model(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|None:passdefafter_agent(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|None:passasyncdefaafter_agent(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|None:passdefwrap_model_call(self,request:ModelRequest,handler:Callable[[ModelRequest],ModelResponse],)-ModelCallResult:msg(Synchronous implementation of wrap_model_call is not available. You are likely encountering this error because you defined only the async version (awrap_model_call) and invoked your agent in a synchronous context (e.g., using stream() or invoke()). To resolve this, either: (1) subclass AgentMiddleware and implement the synchronous wrap_model_call method, (2) use the wrap_model_call decorator on a standalone sync function, or (3) invoke your agent asynchronously using astream() or ainvoke().)raiseNotImplementedError(msg)asyncdefawrap_model_call(self,request:ModelRequest,handler:Callable[[ModelRequest],Awaitable[ModelResponse]],)-ModelCallResult:msg(Asynchronous implementation of awrap_model_call is not available. You are likely encountering this error because you defined only the sync version (wrap_model_call) and invoked your agent in an asynchronous context (e.g., using astream() or ainvoke()). To resolve this, either: (1) subclass AgentMiddleware and implement the asynchronous awrap_model_call method, (2) use the wrap_model_call decorator on a standalone async function, or (3) invoke your agent synchronously using stream() or invoke().)raiseNotImplementedError(msg)defwrap_tool_call(self,request:ToolCallRequest,handler:Callable[[ToolCallRequest],ToolMessage|Command[Any]],)-ToolMessage|Command[Any]:msg(Synchronous implementation of wrap_tool_call is not available. You are likely encountering this error because you defined only the async version (awrap_tool_call) and invoked your agent in a synchronous context (e.g., using stream() or invoke()). To resolve this, either: (1) subclass AgentMiddleware and implement the synchronous wrap_tool_call method, (2) use the wrap_tool_call decorator on a standalone sync function, or (3) invoke your agent asynchronously using astream() or ainvoke().)raiseNotImplementedError(msg)asyncdefawrap_tool_call(self,request:ToolCallRequest,handler:Callable[[ToolCallRequest],Awaitable[ToolMessage|Command[Any]]],)-ToolMessage|Command[Any]:msg(Asynchronous implementation of awrap_tool_call is not available. You are likely encountering this error because you defined only the sync version (wrap_tool_call) and invoked your agent in an asynchronous context (e.g., using astream() or ainvoke()). To resolve this, either: (1) subclass AgentMiddleware and implement the asynchronous awrap_tool_call method, (2) use the wrap_tool_call decorator on a standalone async function, or (3) invoke your agent synchronously using stream() or invoke().)raiseNotImplementedError(msg)通过前面的介绍我们知道在调用create_agent函数时可以利用tools参数进行工具注册其实工具也可以利用tools字段封装到中间件中。中间件被注册时其封装的工具也会一并予以注册。换句话说create_agent方法内部会读取所有注册中间件的tools字段存储的工具连同利用tools参数直接注册的工具一起处理。虽然Agent定义了众多方法但我们可以将它们划分为如下两类生命周期拦截器在Agent和Model执行前后调用包括before_agent/before_model/after_agent/after_modelabefore_agent/abefore_model/aafter_agent/aafter_model调用包装器对Model和Tool的调用进行包装wrap_model_call/wrap_tool_callawrap_model_call/awrap_tool_call2. 生命周期拦截器对于一个利用create_agent函数创建的Agent在没有任何中间件注册的情况下它本质上是由model和tools两个核心节点组成的Pregel对象。注册中间件的生命周期拦截器方法针对Agent和Model调用前后的拦截是通过为Pregel对象添加额外节点和通道来实现的。我们可以利用如下这个简单的实例来验证fromlangchain.agentsimportcreate_agentfromdotenvimportload_dotenvfromlangchain.agents.middleware.typesimportAgentStatefromlangchain_openaiimportChatOpenAIfromPILimportImageasPILImagefromlangchain.agents.middlewareimportAgentMiddlewarefromtypingimportAnyfromlanggraph.runtimeimportRuntimeimportioclassFooMiddleware(AgentMiddleware):defbefore_agent(self,state:AgentState[Any],runtime:Runtime[None])-dict[str,Any]|None:returnsuper().before_agent(state,runtime)defbefore_model(self,state:AgentState[Any],runtime:Runtime[None])-dict[str,Any]|None:returnsuper().before_model(state,runtime)defafter_agent(self,state:AgentState[Any],runtime:Runtime[None])-dict[str,Any]|None:returnsuper().after_agent(state,runtime)defafter_model(self,state:AgentState[Any],runtime:Runtime[None])-dict[str,Any]|None:returnsuper().after_model(state,runtime)load_dotenv()deftest_tool():A test toolagentcreate_agent(modelChatOpenAI(modelgpt-5.2-chat),tools[test_tool],middleware[FooMiddleware()])payloadagent.get_graph(xrayTrue).draw_mermaid_png()PILImage.open(io.BytesIO(payload)).show()print(channels:)for(name,chan)inagent.channels.items():print(f\t[{chan.__class__.__name__}]{name})print(trigger_to_nodes)for(name,nodes)inagent.trigger_to_nodes.items():print(f\t{name}:{nodes})在如上的演示程序中我们创建了自定义中间件类型FooMiddleware并重写它的before_agent、before_model、after_agent和after_model四个方法我们通过注册此中间件调用create_agent函数创建了一个Agent并将他的拓扑结构以PNG图片的形式呈现出来呈现效果如下所示。从上图可以看出注册的中间件为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:classBarMiddleware(AgentMiddleware):defbefore_agent(self,state:AgentState[Any],runtime:Runtime[None])-dict[str,Any]|None:returnsuper().before_agent(state,runtime)defbefore_model(self,state:AgentState[Any],runtime:Runtime[None])-dict[str,Any]|None:returnsuper().before_model(state,runtime)defafter_agent(self,state:AgentState[Any],runtime:Runtime[None])-dict[str,Any]|None:returnsuper().after_agent(state,runtime)defafter_model(self,state:AgentState[Any],runtime:Runtime[None])-dict[str,Any]|None:returnsuper().after_model(state,runtime)agentcreate_agent(modelChatOpenAI(modelgpt-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节点发出的所以利用中间件对调用的封装也在这两个节点中完成。classAgentMiddleware(Generic[StateT,ContextT]):defwrap_model_call(self,request:ModelRequest,handler:Callable[[ModelRequest],ModelResponse],)-ModelCallResultasyncdefawrap_model_call(self,request:ModelRequest,handler:Callable[[ModelRequest],Awaitable[ModelResponse]],)-ModelCallResultdefwrap_tool_call(self,request:ToolCallRequest,handler:Callable[[ToolCallRequest],ToolMessage|Command[Any]],)-ToolMessage|Command[Any]asyncdefawrap_tool_call(self,request:ToolCallRequest,handler:Callable[[ToolCallRequest],Awaitable[ToolMessage|Command[Any]]],)-ToolMessage|Command[Any]3.1 针对模型调用的包装常规的模型调用会返回一个AIMessage消息。如果采用基于ToolStrategy的结构化输出除了返回格式化的输出外还会涉及格式化工具生成的ToolMessage它们被封装在一个ModelResponse对象里。所以表示模型调用结果的ModelResult类型是ModelResponse和AIMessage这两个类型的联合。dataclassclassModelResponse:result:list[BaseMessage]structured_response:AnyNoneModelCallResult:TypeAliasModelResponse|AIMessageModelRequest表示模型调用的请求我们从中可以得到Chat模型组件、请求消息列表、系统指令、注册的工具以及针对工具选择策略、结构化输出Schema、状态、运行时和针对模型的设置。在绝大部情况下我们通过自定义中间件包装模型调用的目的都是为了更新上述的某一个或者多个请求元素ModelRequest利用override方法将一切变得简单。dataclass(initFalse)classModelRequest:model:BaseChatModel messages:list[AnyMessage]system_message:SystemMessage|Nonetool_choice:Any|Nonetools:list[BaseTool|dict[str,Any]]response_format:ResponseFormat[Any]|Nonestate:AgentState[Any]runtime:Runtime[ContextT]model_settings:dict[str,Any]field(default_factorydict)propertydefsystem_prompt(self)-str|None:defoverride(self,**overrides:Unpack[_ModelRequestOverrides])-ModelRequest:class_ModelRequestOverrides(TypedDict,totalFalse):model:BaseChatModel system_message:SystemMessage|Nonemessages:list[AnyMessage]tool_choice:Any|Nonetools:list[BaseTool|dict[str,Any]]response_format:ResponseFormat[Any]|Nonemodel_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方法实现针对这些请求元素的更新。dataclassclassToolCallRequest:tool_call:ToolCall tool:BaseTool|Nonestate:Any runtime:ToolRuntimedefoverride(self,**overrides:Unpack[_ToolCallRequestOverrides])-ToolCallRequest:class_ToolCallRequestOverrides(TypedDict,totalFalse):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的字典。但是以这种方式注册的工具并没有绑定一个具体的可执行对象所以默认是无法被调用的。我们可以采用中间件的方式来解决这个问题。fromlangchain.agentsimportcreate_agentfromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportToolMessagefromlangchain.agents.middlewareimportAgentMiddleware,ToolCallRequestfromlanggraph.typesimportCommandfromtypingimportAny,Callable load_dotenv()tool{name:get_weather,description:Get weather information for given city,parameters:{type:object,properties:{city:{type:string}},required:[city]}}classWeatherMiddleware(AgentMiddleware):defwrap_tool_call(self,request:ToolCallRequest,handler:Callable[[ToolCallRequest],ToolMessage|Command[Any]],)-ToolMessage|Command[Any]:tool_callrequest.tool_calliftool_call[name]get_weather:citytool_call[args][city]returnToolMessage(contentfIts sunny in{city}.,tool_call_idtool_call[id])else:returnhandler(request)agentcreate_agent(modelChatOpenAI(modelgpt-5.2-chat),tools[tool],middleware[WeatherMiddleware()],)resultagent.invoke(input{messages:[{role:user,content:What is the weather like in Suzhou?}]})formessageinresult[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 Its sunny in Suzhou. Ai Message The weather in **Suzhou** is **sunny**. ☀️ 
长河君 · 2026/10/6 7:41:12 · 阅读 1.2k [LangChain智能体本质论-08]中间件是如何参与Agent、Model和Tool三者交互的？
LangChain的中间件是围绕Agent执行流程构建的可插拔钩子系统。它允许开发者在不修改核心逻辑的情况下在执行的关键节点如输入处理、模型调用前后、输出解析等对数据流进行拦截、修改或验证。中间件类型以AgentMiddleware为基类。1. AgentMiddlewareAgentMiddleware是一个泛型类型两个泛型参数分别代表状态和静态上下文的类型我们可以利用state_schema字段得到状态类型。它的name属性返回中间件的名称默认返回的是当前的类名。classAgentMiddleware(Generic[StateT,ContextT]):state_schema:type[StateT]cast(type[StateT],_DefaultAgentState)tools:Sequence[BaseTool]propertydefname(self)-str:returnself.__class__.__name__defbefore_agent(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|None:passasyncdefabefore_agent(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|None:passdefbefore_model(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|None:passasyncdefabefore_model(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|None:passdefafter_model(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|None:passasyncdefaafter_model(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|None:passdefafter_agent(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|None:passasyncdefaafter_agent(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|None:passdefwrap_model_call(self,request:ModelRequest,handler:Callable[[ModelRequest],ModelResponse],)-ModelCallResult:msg(Synchronous implementation of wrap_model_call is not available. You are likely encountering this error because you defined only the async version (awrap_model_call) and invoked your agent in a synchronous context (e.g., using stream() or invoke()). To resolve this, either: (1) subclass AgentMiddleware and implement the synchronous wrap_model_call method, (2) use the wrap_model_call decorator on a standalone sync function, or (3) invoke your agent asynchronously using astream() or ainvoke().)raiseNotImplementedError(msg)asyncdefawrap_model_call(self,request:ModelRequest,handler:Callable[[ModelRequest],Awaitable[ModelResponse]],)-ModelCallResult:msg(Asynchronous implementation of awrap_model_call is not available. You are likely encountering this error because you defined only the sync version (wrap_model_call) and invoked your agent in an asynchronous context (e.g., using astream() or ainvoke()). To resolve this, either: (1) subclass AgentMiddleware and implement the asynchronous awrap_model_call method, (2) use the wrap_model_call decorator on a standalone async function, or (3) invoke your agent synchronously using stream() or invoke().)raiseNotImplementedError(msg)defwrap_tool_call(self,request:ToolCallRequest,handler:Callable[[ToolCallRequest],ToolMessage|Command[Any]],)-ToolMessage|Command[Any]:msg(Synchronous implementation of wrap_tool_call is not available. You are likely encountering this error because you defined only the async version (awrap_tool_call) and invoked your agent in a synchronous context (e.g., using stream() or invoke()). To resolve this, either: (1) subclass AgentMiddleware and implement the synchronous wrap_tool_call method, (2) use the wrap_tool_call decorator on a standalone sync function, or (3) invoke your agent asynchronously using astream() or ainvoke().)raiseNotImplementedError(msg)asyncdefawrap_tool_call(self,request:ToolCallRequest,handler:Callable[[ToolCallRequest],Awaitable[ToolMessage|Command[Any]]],)-ToolMessage|Command[Any]:msg(Asynchronous implementation of awrap_tool_call is not available. You are likely encountering this error because you defined only the sync version (wrap_tool_call) and invoked your agent in an asynchronous context (e.g., using astream() or ainvoke()). To resolve this, either: (1) subclass AgentMiddleware and implement the asynchronous awrap_tool_call method, (2) use the wrap_tool_call decorator on a standalone async function, or (3) invoke your agent synchronously using stream() or invoke().)raiseNotImplementedError(msg)通过前面的介绍我们知道在调用create_agent函数时可以利用tools参数进行工具注册其实工具也可以利用tools字段封装到中间件中。中间件被注册时其封装的工具也会一并予以注册。换句话说create_agent方法内部会读取所有注册中间件的tools字段存储的工具连同利用tools参数直接注册的工具一起处理。虽然Agent定义了众多方法但我们可以将它们划分为如下两类生命周期拦截器在Agent和Model执行前后调用包括before_agent/before_model/after_agent/after_modelabefore_agent/abefore_model/aafter_agent/aafter_model调用包装器对Model和Tool的调用进行包装wrap_model_call/wrap_tool_callawrap_model_call/awrap_tool_call2. 生命周期拦截器对于一个利用create_agent函数创建的Agent在没有任何中间件注册的情况下它本质上是由model和tools两个核心节点组成的Pregel对象。注册中间件的生命周期拦截器方法针对Agent和Model调用前后的拦截是通过为Pregel对象添加额外节点和通道来实现的。我们可以利用如下这个简单的实例来验证fromlangchain.agentsimportcreate_agentfromdotenvimportload_dotenvfromlangchain.agents.middleware.typesimportAgentStatefromlangchain_openaiimportChatOpenAIfromPILimportImageasPILImagefromlangchain.agents.middlewareimportAgentMiddlewarefromtypingimportAnyfromlanggraph.runtimeimportRuntimeimportioclassFooMiddleware(AgentMiddleware):defbefore_agent(self,state:AgentState[Any],runtime:Runtime[None])-dict[str,Any]|None:returnsuper().before_agent(state,runtime)defbefore_model(self,state:AgentState[Any],runtime:Runtime[None])-dict[str,Any]|None:returnsuper().before_model(state,runtime)defafter_agent(self,state:AgentState[Any],runtime:Runtime[None])-dict[str,Any]|None:returnsuper().after_agent(state,runtime)defafter_model(self,state:AgentState[Any],runtime:Runtime[None])-dict[str,Any]|None:returnsuper().after_model(state,runtime)load_dotenv()deftest_tool():A test toolagentcreate_agent(modelChatOpenAI(modelgpt-5.2-chat),tools[test_tool],middleware[FooMiddleware()])payloadagent.get_graph(xrayTrue).draw_mermaid_png()PILImage.open(io.BytesIO(payload)).show()print(channels:)for(name,chan)inagent.channels.items():print(f\t[{chan.__class__.__name__}]{name})print(trigger_to_nodes)for(name,nodes)inagent.trigger_to_nodes.items():print(f\t{name}:{nodes})在如上的演示程序中我们创建了自定义中间件类型FooMiddleware并重写它的before_agent、before_model、after_agent和after_model四个方法我们通过注册此中间件调用create_agent函数创建了一个Agent并将他的拓扑结构以PNG图片的形式呈现出来呈现效果如下所示。从上图可以看出注册的中间件为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:classBarMiddleware(AgentMiddleware):defbefore_agent(self,state:AgentState[Any],runtime:Runtime[None])-dict[str,Any]|None:returnsuper().before_agent(state,runtime)defbefore_model(self,state:AgentState[Any],runtime:Runtime[None])-dict[str,Any]|None:returnsuper().before_model(state,runtime)defafter_agent(self,state:AgentState[Any],runtime:Runtime[None])-dict[str,Any]|None:returnsuper().after_agent(state,runtime)defafter_model(self,state:AgentState[Any],runtime:Runtime[None])-dict[str,Any]|None:returnsuper().after_model(state,runtime)agentcreate_agent(modelChatOpenAI(modelgpt-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节点发出的所以利用中间件对调用的封装也在这两个节点中完成。classAgentMiddleware(Generic[StateT,ContextT]):defwrap_model_call(self,request:ModelRequest,handler:Callable[[ModelRequest],ModelResponse],)-ModelCallResultasyncdefawrap_model_call(self,request:ModelRequest,handler:Callable[[ModelRequest],Awaitable[ModelResponse]],)-ModelCallResultdefwrap_tool_call(self,request:ToolCallRequest,handler:Callable[[ToolCallRequest],ToolMessage|Command[Any]],)-ToolMessage|Command[Any]asyncdefawrap_tool_call(self,request:ToolCallRequest,handler:Callable[[ToolCallRequest],Awaitable[ToolMessage|Command[Any]]],)-ToolMessage|Command[Any]3.1 针对模型调用的包装常规的模型调用会返回一个AIMessage消息。如果采用基于ToolStrategy的结构化输出除了返回格式化的输出外还会涉及格式化工具生成的ToolMessage它们被封装在一个ModelResponse对象里。所以表示模型调用结果的ModelResult类型是ModelResponse和AIMessage这两个类型的联合。dataclassclassModelResponse:result:list[BaseMessage]structured_response:AnyNoneModelCallResult:TypeAliasModelResponse|AIMessageModelRequest表示模型调用的请求我们从中可以得到Chat模型组件、请求消息列表、系统指令、注册的工具以及针对工具选择策略、结构化输出Schema、状态、运行时和针对模型的设置。在绝大部情况下我们通过自定义中间件包装模型调用的目的都是为了更新上述的某一个或者多个请求元素ModelRequest利用override方法将一切变得简单。dataclass(initFalse)classModelRequest:model:BaseChatModel messages:list[AnyMessage]system_message:SystemMessage|Nonetool_choice:Any|Nonetools:list[BaseTool|dict[str,Any]]response_format:ResponseFormat[Any]|Nonestate:AgentState[Any]runtime:Runtime[ContextT]model_settings:dict[str,Any]field(default_factorydict)propertydefsystem_prompt(self)-str|None:defoverride(self,**overrides:Unpack[_ModelRequestOverrides])-ModelRequest:class_ModelRequestOverrides(TypedDict,totalFalse):model:BaseChatModel system_message:SystemMessage|Nonemessages:list[AnyMessage]tool_choice:Any|Nonetools:list[BaseTool|dict[str,Any]]response_format:ResponseFormat[Any]|Nonemodel_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方法实现针对这些请求元素的更新。dataclassclassToolCallRequest:tool_call:ToolCall tool:BaseTool|Nonestate:Any runtime:ToolRuntimedefoverride(self,**overrides:Unpack[_ToolCallRequestOverrides])-ToolCallRequest:class_ToolCallRequestOverrides(TypedDict,totalFalse):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的字典。但是以这种方式注册的工具并没有绑定一个具体的可执行对象所以默认是无法被调用的。我们可以采用中间件的方式来解决这个问题。fromlangchain.agentsimportcreate_agentfromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportToolMessagefromlangchain.agents.middlewareimportAgentMiddleware,ToolCallRequestfromlanggraph.typesimportCommandfromtypingimportAny,Callable load_dotenv()tool{name:get_weather,description:Get weather information for given city,parameters:{type:object,properties:{city:{type:string}},required:[city]}}classWeatherMiddleware(AgentMiddleware):defwrap_tool_call(self,request:ToolCallRequest,handler:Callable[[ToolCallRequest],ToolMessage|Command[Any]],)-ToolMessage|Command[Any]:tool_callrequest.tool_calliftool_call[name]get_weather:citytool_call[args][city]returnToolMessage(contentfIts sunny in{city}.,tool_call_idtool_call[id])else:returnhandler(request)agentcreate_agent(modelChatOpenAI(modelgpt-5.2-chat),tools[tool],middleware[WeatherMiddleware()],)resultagent.invoke(input{messages:[{role:user,content:What is the weather like in Suzhou?}]})formessageinresult[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 Its sunny in Suzhou. Ai Message The weather in **Suzhou** is **sunny**. ☀️