从消息交换的角度来看Agent提供一个用于LLM请求和相应的管道。不我们属性的Web框架的处理管道不太一样的是这个管道涉及的消息交换模式并非简单的请求-响应模式内部会涉及ReAct循环。接下来我们LangChain和MAF的Agent管道在设计上有什么异同。1. LangChain对于LangGraph创建的Agent来说根本没有所谓的管道一说因为在这种编程模式下我们会利用StateGraph根据推理流程构建一个对应的图编译生成的Agent会严格按照这个图来执行。我们所谓的管道指的是当我们使用create_agent函数来创建Agent是由注册的中间件来构建的一个用于调用LLM和工具的消息处理管道。1.1 基于AgentState的双节点状态图create_agent函数根据指定的模型和工具集创建的Agent底层使用的依然还是LangGraph的那一套。构建的图默认使用AgentState作为其状态类型表示对话历史的messages是它的核心状态成员。状态图具有两个节点一个用于封装模型节点默认名称为model另一个承载所有的工具(节点默认名称为tools)。模型节点与工具节点之间具有一条动态条件边,当LLM返回的AIMessage包含工具调用时被激活的这条边会路由到tools节点完成工具调用反之则意味着AIMessage携带的就是最终的结果整个推理过程就此结束。工具节点与模型节点有一条静态边所以工具执行后会再次回到模型节点后者在新的状态下完成下一步推理。LangChain利用create_agent工厂函数创建Agent一个最简单的Agent只需要指定模型即可但一般情况我们都需要注册相应的工具。下面的代码利用create_agent函数创建了一个由ChatOpenAI模型和两个注册工具组成的Agentfromlangchain.agentsimportcreate_agentfromlangchain.toolsimporttoolfromlangchain_openaiimportChatOpenAIfromPILimportImagefromdotenvimportload_dotenvimportio load_dotenv()toolasyncdeffoo():test tool footoolasyncdefbar():test tool baragentcreate_agent(modelChatOpenAI(namegpt-5.2-chat),tools[foo,bar])Image.open(io.BytesIO(agent.get_graph().draw_mermaid_png())).show()当这个Agent被转换成图后具有如下的结构。这是一个包含两个节点的状态图一个是用于决策和推理的model节点作为决策执行者的工具全部被封装到tools节点中。1.2 中间件的作用这种AgentState 双节点的结构虽然简单但却能满足常规的推理任务。对于更复杂的推理任务一方面我们可以扩展AgentState提供更多的状态成员来承载更多的上下文信息另一方面我们也可以通过注册AgentMiddleware来添加更多的节点来完善工作流。具体来说注册的AgentMiddleware提供了如下的功能添加状态字段如果AgentMiddleware涉及到针对状态更新对应的状态字段会定义在state_schema字段返回的状态类型中。此状态状态类型通常是AgentState的子类定义其中的字段最终会转换成通道用于注册工具当中间件被注册到创建的Agent上时存储在其tools字段中的工具会自动注册到Agent上。这相当于提供了一种模块化的工具开发和注册的方式添加节点当中间件重写了before_agent/abefore_agent、before_model/abefore_model、after_model/aafter_model、after_agent/aafter_agent方法都会在状态图中相应的位置添加一个节点。Middleware相当于利用此方式完善了Agent的工作流包装模型和工具调用中间件利用重写的wrap_model_call/awrap_model_call、wrap_tool_call/awrap_tool_call方法对模型和工具的调用进行包装将AOP引入到模型和工具的调用中使得在调用前后添加一些额外的操作变得非常简单。比如很多中间件都具有各自的系统提示词它们基本上都是利用重写的wrap_model_call/awrap_model_call方法的方式实现针对系统提示词的注入classAgentMiddleware(Generic[StateT,ContextT]):defbefore_agent(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|Noneasyncdefabefore_agent(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|Nonedefbefore_model(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|Noneasyncdefabefore_model(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|Nonedefafter_model(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|Noneasyncdefaafter_model(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|Nonedefafter_agent(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|Noneasyncdefaafter_agent(self,state:StateT,runtime:Runtime[ContextT])-dict[str,Any]|Nonedefwrap_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]1.3 Agent管道的构建在下面的程序中我们定义了一个FooMiddleware和BarMiddleware并将其注册到Agent上fromlangchain.agentsimportcreate_agentfromdotenvimportload_dotenvfromlangchain.agents.middleware.typesimportAgentState,ExtendedModelResponse,ModelRequest,ModelResponsefromlangchain_openaiimportChatOpenAIfromlangchain.agents.middlewareimportAgentMiddlewarefromlanggraph.prebuilt.tool_nodeimportToolCallRequestfromlanggraph.runtimeimportRuntimefromtypingimportAny,Callablefromlangchain_core.messagesimportAIMessage,ToolMessagefromlangchain_core.toolsimporttoolfromlanggraph.typesimportCommandfromPILimportImageimportio load_dotenv()tooldefget_weather(city:str)-str:Get weather information for given cityreturnfIts sunny today in{city}.classBaseMiddleware(AgentMiddleware):defbefore_agent(self,state:AgentState[Any],runtime:Runtime[None])-dict[str,Any]|None:print(f{self.name}.before_agent)returnsuper().before_agent(state,runtime)defbefore_model(self,state:AgentState[Any],runtime:Runtime[None])-dict[str,Any]|None:print(f{self.name}.before_model)returnsuper().before_model(state,runtime)defafter_agent(self,state:AgentState[Any],runtime:Runtime[None])-dict[str,Any]|None:print(f{self.name}.after_agent)returnsuper().after_agent(state,runtime)defafter_model(self,state:AgentState[Any],runtime:Runtime[None])-dict[str,Any]|None:print(f{self.name}.after_model)returnsuper().after_model(state,runtime)defwrap_tool_call(self,request:ToolCallRequest,handler:Callable[[ToolCallRequest],ToolMessage|Command[Any]])-ToolMessage|Command[Any]:print(f{self.name}.wrap_tool_call)returnhandler(request)defwrap_model_call(self,request:ModelRequest[None],handler:Callable[[ModelRequest[None]],ModelResponse[Any]])-ModelResponse[Any]|AIMessage|ExtendedModelResponse[Any]:print(f{self.name}.wrap_model_call)returnhandler(request)classFooMiddleware(BaseMiddleware):passclassBarMiddleware(BaseMiddleware):passagentcreate_agent(modelChatOpenAI(modelgpt-5.2-chat),tools[get_weather],middleware[FooMiddleware(),BarMiddleware()])resultagent.invoke(input{messages:[{role:user,content:What is the weather like in Suzhou?}]})Image.open(io.BytesIO(agent.get_graph().draw_mermaid_png())).show()print(result[messages][-1].content)我们调用Agent的get_graph方法将它转换成一个Graph对象并调用Graph对象的draw_mermaid_png方法将图以Mermaid格式绘制成PNG图片。从下图可以看出由于我们注册了两个中间件所以会在状态图中添加八个节点至于整张图前后的两个节点对应中间件的before_agent和after_agent方法位于模型节点前后的四个节点对应中间件的before_model和after_model方法。中间件的wrap_tool_call和wrap_model_call方法会用于包装现有针对模型和工具的调用。在定义注册中间件的基类时我们在每个方法上都添加了输出语句以便我们在调用Agent时能够清晰地看到每个方法的调用顺序。我们的调用会产生如下的输出FooMiddleware.before_agent BarMiddleware.before_agent FooMiddleware.before_model BarMiddleware.before_model FooMiddleware.wrap_model_call BarMiddleware.wrap_model_call BarMiddleware.after_model FooMiddleware.after_model FooMiddleware.wrap_tool_call BarMiddleware.wrap_tool_call FooMiddleware.before_model BarMiddleware.before_model FooMiddleware.wrap_model_call BarMiddleware.wrap_model_call BarMiddleware.after_model FooMiddleware.after_model BarMiddleware.after_agent FooMiddleware.after_agent It’s sunny in Suzhou today. ☀️我们可以简单分析一下这个输出体现的执行流程对于每次Agent的调用中间件重写的before_agent方法会在Agent执行前被调用after_agent方法会在Agent执行后被调用涉及两次针对LLM的调用所以中间件的before_model、after_model和wrap_model_call方法会被调用两次涉及一次工具调用所以中间件的wrap_tool_call方法会被调用一次2. MAFMAF具有不同类型的AIAgent类型这里我们只关注最常用的ChatClientAgent。它具有如下图所示的管道结构从右到左看整个管道由三个部分组成IChatClient管道与LLM交互的是由LLM客户端构建的IChatClient对象我们可以采用装饰器的设计模式为它添加任意作为装饰器的DelegatingChatClient对象。DelegatingChatClient就是IChatClient的中间件整个IChatClient管道就是由这些中间件装饰而成的一个链式结构输入输出增强ChatClientAgent利用注册的AIContextProvider为输入输出增强提供了一个非常灵活的机制我们可以利用注册的AIContextProvider在调用IChatClient之前对输入消息列表、注册的工具集以及调用LLM的系统指令进行加工同样的在得到LLM的响应后我们也可以利用AIContextProvider对响应消息进行加工处理。输入输出增强机制为我们提供了一个非常灵活的方式来完善Agent的工作流Agent中间件和构建IChatClient管道的中间件类似我们也可以利用DelegatingAIAgent作为ChatClientAgent的中间件来完善Agent的工作流。下面的演示程序演示了如何使用上面介绍的这些对象来构建一个ChatClientAgent的管道。我们自定义了一个FakeChatClient来模拟与LLM交互的IChatClient并定义了ChatClientMiddleware和AgentMiddleware作为IChatClient管道和Agent管道的中间件NamedAIContextProvider作为输入输出增强的AIContextProvider。我们将它们组合在一起构建了一个ChatClientAgent并调用RunAsync方法来运行这个Agent。usingAzure.AI.Projects;usingMicrosoft.Agents.AI;usingMicrosoft.Extensions.AI;varagentnewFakeChatClient().AsBuilder().Use(innernewChatClientMiddleware(inner,FooChatClientMiddleware)).Use(innernewChatClientMiddleware(inner,BarChatClientMiddleware)).Build().AsAIAgent(newChatClientAgentOptions{AIContextProviders[newNamedAIContextProvider(FooAIContextProvider),newNamedAIContextProvider(BarAIContextProvider)]}).AsBuilder().Use(innernewAgentMiddleware(inner,FooAgentMiddleware)).Use(innernewAgentMiddleware(inner,BarAgentMiddleware)).Build();awaitagent.RunAsync();classFakeChatClient:IChatClient{publicvoidDispose(){}publicTaskChatResponseGetResponseAsync(IEnumerableChatMessagemessages,ChatOptions?optionsnull,CancellationTokencancellationTokendefault){Console.WriteLine(FakeChatClient.GetResponseAsync);returnTask.FromResult(newChatResponse(new[]{newChatMessage(ChatRole.Assistant,Fake response)}));}publicobject?GetService(TypeserviceType,object?serviceKeynull)null;publicIAsyncEnumerableChatResponseUpdateGetStreamingResponseAsync(IEnumerableChatMessagemessages,ChatOptions?optionsnull,CancellationTokencancellationTokendefault)thrownewNotImplementedException();}classChatClientMiddleware(IChatClientinnerClient,stringname):DelegatingChatClient(innerClient){publicoverrideTaskChatResponseGetResponseAsync(IEnumerableChatMessagemessages,ChatOptions?optionsnull,CancellationTokencancellationTokendefault){Console.WriteLine(${name}.GetResponseAsync[before]);varresultbase.GetResponseAsync(messages,options,cancellationToken);Console.WriteLine(${name}.GetResponseAsync[after]);returnresult;}}classAgentMiddleware(AIAgentinnerAgent,stringname):DelegatingAIAgent(innerAgent){protectedoverrideTaskAgentResponseRunCoreAsync(IEnumerableChatMessagemessages,AgentSession?sessionnull,AgentRunOptions?optionsnull,CancellationTokencancellationTokendefault){Console.WriteLine(${name}.RunAsync[before]);varresultbase.RunCoreAsync(messages,session,options,cancellationToken);Console.WriteLine(${name}.RunAsync[after]);returnresult;}}classNamedAIContextProvider(stringname):AIContextProvider{protectedoverrideValueTaskAIContextInvokingCoreAsync(InvokingContextcontext,CancellationTokencancellationTokendefault){Console.WriteLine(${name}.InvokingAsync);returnbase.InvokingCoreAsync(context,cancellationToken);}protectedoverrideValueTaskInvokedCoreAsync(InvokedContextcontext,CancellationTokencancellationTokendefault){Console.WriteLine(${name}.InvokedAsync);returnbase.InvokedCoreAsync(context,cancellationToken);}publicoverrideIReadOnlyListstringStateKeys[${name}.{GetType().Name}];}程序运行后会产生如下的输出。对照一下上面的管道图我们会发现输出的顺序完全符合管道图所示的执行流程FooAgentMiddleware.RunAsync[before] BarAgentMiddleware.RunAsync[before] FooAIContextProvider.InvokingAsync BarAIContextProvider.InvokingAsync FooChatClientMiddleware.GetResponseAsync[before] BarChatClientMiddleware.GetResponseAsync[before] FakeChatClient.GetResponseAsync BarChatClientMiddleware.GetResponseAsync[after] FooChatClientMiddleware.GetResponseAsync[after] FooAIContextProvider.InvokedAsync BarAIContextProvider.InvokedAsync BarAgentMiddleware.RunAsync[after] FooAgentMiddleware.RunAsync[after]我的博文系列“MAF的Agent管道详解”提供了对MAF的Agent管道的详细解读感兴趣的读者可以前往阅读。
