【langgraph 从入门到精通graphApi 篇】Subgraph 与多 Agent 系统

第 9 章：Subgraph 与多 Agent 系统

9.1 本章目标

学完本章你将能够：

1. 理解子图的两种集成方式（节点内调用 vs 作为节点）
2. 掌握 Supervisor 多 Agent 模式
3. 学会 Agent 间 State 共享和隔离
4. 能够设计多 Agent 协作系统

9.2 核心概念

子图是什么？

子图（Subgraph）是一个编译好的图，可以嵌入到另一个图中。它让复杂的 Agent 系统可以拆分为多个独立的、可复用的子模块。

两种集成方式

Supervisor 多 Agent 架构

子图持久化模式

9.3 实战

实战 1：子图作为节点（共享 State）

fromtypingimportTypedDict,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_builder=StateGraph(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_a=sub_a_builder.compile()# ============================================# 子图 B：数据分析# ============================================defanalyze_data(state:SharedState)->dict:return{"data":["[子图B] 数据已分析"]}sub_b_builder=StateGraph(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_b=sub_b_builder.compile()# ============================================# 父图：编排子图# ============================================deffinal_summary(state:SharedState)->dict:all_data=", ".join(state.get("data",[]))return{"data":[f"[父图] 汇总:{all_data}"]}parent_builder=StateGraph(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_graph=parent_builder.compile()result=parent_graph.invoke({"data":[]})print("处理结果:")fordinresult["data"]:print(f"{d}")

实战 2：Supervisor 多 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_msg=state["messages"][-1]return{"messages":[AIMessage(content=f"[搜索专家] 关于「{last_msg.content}」，找到以下信息：...",name="search_expert")],}defmath_expert(state:SupervisorState)->dict:return{"messages":[AIMessage(content="[数学专家] 计算结果：42",name="math_expert")],}defgeneral_expert(state:SupervisorState)->dict:return{"messages":[AIMessage(content="[通用专家] 这是对您问题的回答...",name="general_expert")],}defsupervisor(state:SupervisorState)->dict:last_msg=state["messages"][-1].contentif"搜索"inlast_msgor"查找"inlast_msg:return{"next_agent":"search"}elif"计算"inlast_msgor"+"inlast_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"]:agent=state["next_agent"]ifagent=="END":return"END"returnagent builder=StateGraph(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")graph=builder.compile()# 测试result=graph.invoke({"messages":[HumanMessage(content="搜索 LangGraph 教程")],"next_agent":"",})formsginresult["messages"]:ifhasattr(msg,"name")andmsg.name:print(f" [{msg.name}]{msg.content}")

9.4 API 速查

9.5 错误与避坑指南

坑 1：子图 State 和父图 State 不匹配

# ❌ 错误：父图 State 有 "query"，子图 State 有 "search_query"classParentState(TypedDict):query:strclassSubState(TypedDict):search_query:str# 字段名不同！# ✅ 正确：节点内调用时做转换defcall_subgraph(state:ParentState)->dict:result=subgraph.invoke({"search_query":state["query"]})return{"query":result["search_query"]}

坑 2：Supervisor 无限循环

# ❌ 错误：没有退出条件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 最佳实践总结

1. 简单分层用子图作为节点，复杂转换用节点内调用：按 State 兼容性选择
2. Supervisor 模式适合任务明确、分工清晰的场景：一个主管 + 多个专家
3. 子图应独立可测试：不依赖父图的具体实现，可以单独 invoke
4. 多 Agent 系统先设计通信协议再实现：定义好 State 结构和消息格式
5. 设置最大调度次数防止无限循环：Supervisor 路由中必须包含退出条件