[LangChain中的Multi-Agent模式-02]Router：多领域任务并行时的路由风险评估与分发策略

在路由架构中路由步骤会对输入进行分类并将其定向到专门的Agent其主要特征包括路由器将查询分解并行调用零个或多个专用Agent将结果综合成一个连贯的响应。当我们拥有不同的垂直领域各自独立的知识领域每个领域都需要自己的Agent、需要并行查询多个数据源并且希望将结果综合成一个统一的响应时特别适合使用路由模式。对于前面采用Sub-Agent模式实现的差旅助手如果切换成这种模式可以采用下图所示的路由方式。熟悉LangGraph的人应该知道上图本质上是由StateGraph构建的状态图。话句话说路由模式可以利用LangGraph来构建。1. 构建路由器路由会对输入进行分类并将其定向到专门的Agent。对于我们这个差旅助手来说这个路由器就是意图分析器。我们将这个路由器定义成如下这个router方法它最终会成为StateGraph状态图中的一个节点。IntentTypeLiteral[book_hotel,buy_airplane_ticket,both,none]classState(AgentState):intent:IntentType|NoneclassIntent(TypedDict):出行安排意图need_arrange_accommodation:bool是否涉及酒店住宿安排need_arrange_transportation:bool是否涉及以机票为主的交通安排modelChatOpenAI(modelgpt-5.2-chat)asyncdefrouter(state:State)-dict:result:Intentcast(Intent,awaitmodel.with_structured_output(schemaIntent).ainvoke((f请根据以下请求分析用户的出行安排意图判断是购买机票、预订酒店还是两者都有{state[messages][-1].content})))need_book_hotelresult[need_arrange_accommodation]need_buy_airplane_ticketresult[need_arrange_transportation]intent:IntentTypenoneifneed_book_hotelandneed_buy_airplane_ticket:intentbothelifneed_book_hotel:intentbook_hotelelse:intentbuy_airplane_ticketreturn{intent:intent}如上面的代码片段所示我们依然在继承自AgentState的状态类型State中添加了表示意图的字段intent。实现在router函数中的意图分析依然了利用LLM来完成并且同样利用结构化输出得到一个明确的意图。前面以工具形式实现的意图分析通过返回的Command将分析结果写入状态这里直接利用返回的字典达到相同的目的。2. 利用节点封装Sub-Agent现在我们直接利用LangGraph构建工作流直接将Sub-Agent封装在状态图的节点中。和前面一样我们利用定义的函数create_accommodation_agent和create_transportation_agent来创建分别用于安排住宿和交通出行的Sub-Agent后者利用MultiServerMCPClient链接MCP服务器得到所需的工具book_hotel和buy_airplane_ticket。MCP服务器的构建在“SubAgent集中编排视角下的上下文隔离与并行化实现”已经介绍过了这里不再赘述。clientMultiServerMCPClient(connections{server:{transport:stdio,command:python,args:[server.py]}},)asyncdefcreate_accommodation_agent():toolsawaitclient.get_tools(server_nameserver)selected_tools[toolfortoolintoolsiftool.namebook_hotel]agentcreate_agent(modelChatOpenAI(modelgpt-5.2-chat),toolsselected_tools,system_prompt(你是一个专注于住宿安排的差旅助理你唯一需要做的是在无需授权情况下直接调用注册的book_hotel工具预订酒店。完成预订是首要任务无需考虑其他任何信息。你可以完全自由地选择酒店、价位和房型等信息无需用户确认。))returnagentasyncdefcreate_transportation_agent():toolsawaitclient.get_tools(server_nameserver)selected_tools[toolfortoolintoolsiftool.namebuy_airplane_ticket]agentcreate_agent(modelChatOpenAI(modelgpt-5.2-chat),toolsselected_tools,system_prompt(你是一个专注于交通安排的差旅助理你唯一需要做的是在无需授权情况下调用注册的buy_airplane_ticket工具购买机票。完成购买是首要任务无需考虑其他任何信息。你可以完全自由地选择具体的航司、舱位等级和航班等无需用户确认。))returnagentasyncdefmain():accommodation_agentawaitcreate_accommodation_agent()transportation_agentawaitcreate_transportation_agent()asyncdefarrange_transportation(state:State)-dict:resultawaittransportation_agent.ainvoke({messages:state[messages]})# type: ignorereturn{messages:[result[messages][-1]]}asyncdefarrange_accommodation(state:State)-dict:resultawaitaccommodation_agent.ainvoke({messages:state[messages]})# type: ignorereturn{messages:[result[messages][-1]]}在main函数中定义的arrange_transportation和arrange_accommodation就是用于封装Sub-Agent的节点函数它们在完成Sub-Agent调用之后只会将最后的结果最后一条AIMessage写入消息历史所以Main-Agent的上下文窗口压力会大大减轻。3. 路由汇总路由将请求按照预定义的路径分发出去进行并行处理后一般还需要对这些结果进行汇总。对于我们的例子来说我们需要一个汇总节点对已经完成的交通出行和住宿安排进行整理后统一显示给用户。如下这个synthesize就是这么一个节点函数它直接将整理工作交给LLM来完成。asyncdefmain():...asyncdefsynthesize(state:State)-dict:messageHumanMessage(content整理已经制定好的出行安排。如果没有发现任何出行安排请回复用户意图不明无法执行。)resultawaitmodel.ainvoke([message,*state[messages]])return{messages:[result]}4. 构建Agent上图展示的状态图由如下的程序来构建。我们针对状态类型State创建了一个StateGraph对象并添加了四个函数router、arrange_transportation、arrange_accommodation和synthesize对应的节点。router和synthesize分别被设置成入口节点和完成节点。router和其余三个节点之间具有条件边条件函数route_condition根据状态存储的意图决定目标节点。最终的Agent通过编译StateGraph生成。asyncdefmain():defroute_condition(state:State)-list[str]:ifstate[intent]both:return[arrange_transportation,arrange_accommodation]elifstate[intent]book_hotel:return[arrange_accommodation]elifstate[intent]buy_airplane_ticket:return[arrange_transportation]else:return[synthesize]agent(StateGraph(State).add_node(router,router).add_node(arrange_transportation,arrange_transportation).add_node(arrange_accommodation,arrange_accommodation).add_node(synthesize,synthesize).set_entry_point(router).set_finish_point(synthesize).add_conditional_edges(router,lambdastate:route_condition(state),[arrange_transportation,arrange_accommodation,synthesize]).add_edge(arrange_transportation,synthesize).add_edge(arrange_accommodation,synthesize)).compile()5. 测试Agent我们采用与前面一样的测试用例供了四种输入来模拟四种情况同时包含酒店预订和机票购买需求、只包含酒店预订或者机票购买需求以及不涉及这两种需求对构建的Agent进行测试asyncdefmain():...inputs[我有明后两天(从上海)去成都的出差请帮我预订酒店明天入住后天离开和机票往返明天去后天回。,我有明后两天(从上海)去成都的出差请帮我预订酒店明天入住后天离开。,我有明后两天(从上海)去成都的出差请帮我预订机票往返明天去后天回。,随便聊聊。,]forinputininputs:resultawaitagent.ainvoke(input{messages:[HumanMessage(contentinput)]})print(f用户输入{input})print(f系统回复{result[messages][-1].text}\n\n)以下是对应的四段输出用户输入我有明后两天(从上海)去成都的出差请帮我预订酒店明天入住后天离开和机票往返明天出发后天返回。 系统回复 **已预订信息汇总** ** 酒店预订** - 城市成都 - 酒店名称如家酒店 - 入住时间2026-04-17 - 离店时间2026-04-18 **✈️ 机票信息往返** - **去程** - 上海 → 成都 - 出发日期2026-04-17 - 航班号MU5401 - **返程** - 成都 → 上海 - 出发日期2026-04-18 - 航班号MU5402 如需我继续帮您核对订单、调整行程或补充其他出差安排请告诉我。用户输入我有明后两天(从上海)去成都的出差请帮我预订酒店明天入住后天离开。 系统回复 **预订信息汇总** **酒店预订** - 酒店名称如家酒店 - 城市成都 - 入住时间2026年4月17日 - 离店时间2026年4月18日 **机票信息** - ✈️ 暂无机票预订信息用户输入我有明后两天(从上海)去成都的出差请帮我预订机票往返明天出发后天返回。 系统回复### 已有预订信息汇总 ✈️ **机票预订** - **行程**上海 ⇄ 成都往返 - **去程** - 日期2026-04-17 - 航班号MU5401 - **返程** - 日期2026-04-18 - 航班号MU5402 **酒店预订**无已忽略 如需补充酒店预订或调整航班信息请告知。用户输入随便说说 系统回复意图不明无法执行。对于第一个输入如图反映了Agent内部完整的调用链可以看出它将完整的流程走了一遍。6. 完整程序下面给出完整程序fromtypingimportLiteral,TypedDict,castfromlangchain.agentsimportcreate_agent,AgentStatefromlanggraph.graphimportStateGraphfromlangchain_openaiimportChatOpenAIfromlangchain_mcp_adapters.clientimportMultiServerMCPClientfromlangchain_core.messagesimportHumanMessageimportasynciofromdotenvimportload_dotenv load_dotenv()IntentTypeLiteral[book_hotel,buy_airplane_ticket,both,none]classState(AgentState):intent:IntentType|NoneclassIntent(TypedDict):出行安排意图need_arrange_accommodation:bool是否涉及酒店住宿安排need_arrange_transportation:bool是否涉及以机票为主的交通安排modelChatOpenAI(modelgpt-5.2-chat)asyncdefrouter(state:State)-dict:result:Intentcast(Intent,awaitmodel.with_structured_output(schemaIntent).ainvoke((f请根据以下请求分析用户的出行安排意图判断是购买机票、预订酒店还是两者都有{state[messages][-1].content})))need_book_hotelresult[need_arrange_accommodation]need_buy_airplane_ticketresult[need_arrange_transportation]intent:IntentTypenoneifneed_book_hotelandneed_buy_airplane_ticket:intentbothelifneed_book_hotel:intentbook_hotelelse:intentbuy_airplane_ticketreturn{intent:intent}clientMultiServerMCPClient(connections{server:{transport:stdio,command:python,args:[server.py]}},)asyncdefcreate_accommodation_agent():toolsawaitclient.get_tools(server_nameserver)selected_tools[toolfortoolintoolsiftool.namebook_hotel]agentcreate_agent(modelChatOpenAI(modelgpt-5.2-chat),toolsselected_tools,system_prompt(你是一个专注于住宿安排的差旅助理你唯一需要做的是在无需授权情况下直接调用注册的book_hotel工具预订酒店。完成预订是首要任务无需考虑其他任何信息。你可以完全自由地选择酒店、价位和房型等信息无需用户确认。))returnagentasyncdefcreate_transportation_agent():toolsawaitclient.get_tools(server_nameserver)selected_tools[toolfortoolintoolsiftool.namebuy_airplane_ticket]agentcreate_agent(modelChatOpenAI(modelgpt-5.2-chat),toolsselected_tools,system_prompt(你是一个专注于交通安排的差旅助理你唯一需要做的是在无需授权情况下调用注册的buy_airplane_ticket工具购买机票。完成购买是首要任务无需考虑其他任何信息。你可以完全自由地选择具体的航司、舱位等级和航班等无需用户确认。))returnagentasyncdefmain():accommodation_agentawaitcreate_accommodation_agent()transportation_agentawaitcreate_transportation_agent()asyncdefarrange_transportation(state:State)-dict:resultawaittransportation_agent.ainvoke({messages:state[messages]})# type: ignorereturn{messages:[result[messages][-1]]}asyncdefarrange_accommodation(state:State)-dict:resultawaitaccommodation_agent.ainvoke({messages:state[messages]})# type: ignorereturn{messages:[result[messages][-1]]}asyncdefsynthesize(state:State)-dict:messageHumanMessage(content汇总预订的酒店如果没有请忽略和购买机票信息如果没有请忽略。如果没有任何预订信息请回复意图不明无法执行。)resultawaitmodel.ainvoke([message,*state[messages]])return{messages:[result]}defroute_condition(state:State)-list[str]:ifstate[intent]both:return[arrange_transportation,arrange_accommodation]elifstate[intent]book_hotel:return[arrange_accommodation]elifstate[intent]buy_airplane_ticket:return[arrange_transportation]else:return[synthesize]agent(StateGraph(State).add_node(router,router).add_node(arrange_transportation,arrange_transportation).add_node(arrange_accommodation,arrange_accommodation).add_node(synthesize,synthesize).set_entry_point(router).set_finish_point(synthesize).add_conditional_edges(router,lambdastate:route_condition(state),[arrange_transportation,arrange_accommodation,synthesize]).add_edge(arrange_transportation,synthesize).add_edge(arrange_accommodation,synthesize)).compile()fromPILimportImageasPILImageimportio# payload agent.get_graph().draw_mermaid_png()# PILImage.open(io.BytesIO( payload)).show()inputs[我有明后两天(从上海)去成都的出差请帮我预订酒店明天入住后天离开和机票往返明天出发后天返回。,我有明后两天(从上海)去成都的出差请帮我预订酒店明天入住后天离开。,我有明后两天(从上海)去成都的出差请帮我预订机票往返明天出发后天返回。,随便说说]formessageininputs:reusltawaitagent.ainvoke(input{messages:[HumanMessage(contentmessage)]})# type: ignoreprint(f用户输入{message})print(f系统回复{reuslt[messages][-1].text}\n\n)asyncio.run(main())