第 9 章Subgraph 与多 Agent 系统文章目录第 9 章Subgraph 与多 Agent 系统9.1 本章目标9.2 核心概念子图是什么两种集成方式Supervisor 多 Agent 架构子图持久化模式9.3 实战实战 1子图作为节点共享 State实战 2Supervisor 多 Agent 系统9.4 API 速查9.5 错误与避坑指南坑 1子图 State 和父图 State 不匹配坑 2Supervisor 无限循环坑 3多 Agent 共享 State 时 key 冲突9.6 最佳实践总结9.1 本章目标学完本章你将能够理解子图的两种集成方式节点内调用 vs 作为节点掌握 Supervisor 多 Agent 模式学会 Agent 间 State 共享和隔离能够设计多 Agent 协作系统9.2 核心概念子图是什么子图Subgraph是一个编译好的图可以嵌入到另一个图中。它让复杂的 Agent 系统可以拆分为多个独立的、可复用的子模块。两种集成方式方式适用场景State 要求方法节点内调用子图父图和子图 State Schema 不同需要手动转换在节点函数中subgraph.invoke(...)子图作为节点父图和子图共享 State keys直接兼容builder.add_node(name, compiled_subgraph)Supervisor 多 Agent 架构需要搜索需要计算需要编程完成STARTSupervisor任务分派搜索 Agent搜索知识库计算 Agent数学计算编程 Agent代码生成END子图持久化模式模式行为使用场景Per-invocation默认每次调用重新开始继承父 checkpointer无状态子任务Per-thread状态跨调用累积子图有自己的 thread有状态的子 AgentStateless无 checkpoint纯函数调用纯计算任务9.3 实战实战 1子图作为节点共享 StatefromtypingimportTypedDict,Annotatedimportoperatorfromlanggraph.graphimportStateGraph,START,END# # 定义共享 State父图和子图使用相同的 key# classSharedState(TypedDict):data:Annotated[list[str],operator.add]# # 子图 A数据清洗# defclean_data(state:SharedState)-dict:return{data:[[子图A] 数据已清洗]}sub_a_builderStateGraph(SharedState)sub_a_builder.add_node(clean,clean_data)sub_a_builder.add_edge(START,clean)sub_a_builder.add_edge(clean,END)sub_graph_asub_a_builder.compile()# # 子图 B数据分析# defanalyze_data(state:SharedState)-dict:return{data:[[子图B] 数据已分析]}sub_b_builderStateGraph(SharedState)sub_b_builder.add_node(analyze,analyze_data)sub_b_builder.add_edge(START,analyze)sub_b_builder.add_edge(analyze,END)sub_graph_bsub_b_builder.compile()# # 父图编排子图# deffinal_summary(state:SharedState)-dict:all_data, .join(state.get(data,[]))return{data:[f[父图] 汇总:{all_data}]}parent_builderStateGraph(SharedState)parent_builder.add_node(step_clean,sub_graph_a)parent_builder.add_node(step_analyze,sub_graph_b)parent_builder.add_node(summary,final_summary)parent_builder.add_edge(START,step_clean)parent_builder.add_edge(step_clean,step_analyze)parent_builder.add_edge(step_analyze,summary)parent_builder.add_edge(summary,END)parent_graphparent_builder.compile()resultparent_graph.invoke({data:[]})print(处理结果:)fordinresult[data]:print(f{d})实战 2Supervisor 多 Agent 系统fromtypingimportTypedDict,Annotated,Literalfromlanggraph.graphimportStateGraph,START,END,add_messagesfromlangchain_core.messagesimportBaseMessage,HumanMessage,AIMessageclassSupervisorState(TypedDict):messages:Annotated[list[BaseMessage],add_messages]next_agent:strdefsearch_expert(state:SupervisorState)-dict:last_msgstate[messages][-1]return{messages:[AIMessage(contentf[搜索专家] 关于「{last_msg.content}」找到以下信息...,namesearch_expert)],}defmath_expert(state:SupervisorState)-dict:return{messages:[AIMessage(content[数学专家] 计算结果42,namemath_expert)],}defgeneral_expert(state:SupervisorState)-dict:return{messages:[AIMessage(content[通用专家] 这是对您问题的回答...,namegeneral_expert)],}defsupervisor(state:SupervisorState)-dict:last_msgstate[messages][-1].contentif搜索inlast_msgor查找inlast_msg:return{next_agent:search}elif计算inlast_msgorinlast_msg:return{next_agent:math}elif再见inlast_msgor结束inlast_msg:return{next_agent:END}else:return{next_agent:general}defsupervisor_route(state:SupervisorState)-Literal[search,math,general,END]:agentstate[next_agent]ifagentEND:returnENDreturnagent builderStateGraph(SupervisorState)builder.add_node(supervisor,supervisor)builder.add_node(search,search_expert)builder.add_node(math,math_expert)builder.add_node(general,general_expert)builder.add_edge(START,supervisor)builder.add_conditional_edges(supervisor,supervisor_route,{search:search,math:math,general:general,END:END,})builder.add_edge(search,supervisor)builder.add_edge(math,supervisor)builder.add_edge(general,supervisor)graphbuilder.compile()# 测试resultgraph.invoke({messages:[HumanMessage(content搜索 LangGraph 教程)],next_agent:,})formsginresult[messages]:ifhasattr(msg,name)andmsg.name:print(f [{msg.name}]{msg.content})9.4 API 速查API完整签名入参说明返回值说明add_node(name, subgraph)add_node(name: str, subgraph: CompiledGraph)name: 节点名;subgraph: 编译后的子图Self子图作为节点subgraph.invoke(input, config)invoke(input, config)input: 子图输入;config: 配置子图输出节点内调用子图Command(graphCommand.PARENT)Command(graph: str)graph: 目标图Command子图控制父图流程compile(checkpointer模式)compile(checkpointer)None(默认)/True(Per-thread)/False(Stateless)CompiledGraph控制子图持久化9.5 错误与避坑指南坑 1子图 State 和父图 State 不匹配# ❌ 错误父图 State 有 query子图 State 有 search_queryclassParentState(TypedDict):query:strclassSubState(TypedDict):search_query:str# 字段名不同# ✅ 正确节点内调用时做转换defcall_subgraph(state:ParentState)-dict:resultsubgraph.invoke({search_query:state[query]})return{query:result[search_query]}坑 2Supervisor 无限循环# ❌ 错误没有退出条件defsupervisor(state:State)-dict:return{next_agent:search}# 始终返回 search → 无限循环# ✅ 正确设置退出条件defsupervisor(state:State)-dict:ifstate.get(iteration,0)5:return{next_agent:END}if搜索instate[messages][-1].content:return{next_agent:search,iteration:state.get(iteration,0)1}return{next_agent:END}坑 3多 Agent 共享 State 时 key 冲突# ❌ 错误多个 Agent 共用 result 字段classState(TypedDict):result:str# Agent A 和 Agent B 都写这个字段互相覆盖# ✅ 正确使用不同前缀或 ReducerclassState(TypedDict):search_result:strmath_result:strresults:Annotated[list[str],operator.add]# 累加9.6 最佳实践总结简单分层用子图作为节点复杂转换用节点内调用按 State 兼容性选择Supervisor 模式适合任务明确、分工清晰的场景一个主管 多个专家子图应独立可测试不依赖父图的具体实现可以单独 invoke多 Agent 系统先设计通信协议再实现定义好 State 结构和消息格式设置最大调度次数防止无限循环Supervisor 路由中必须包含退出条件
