05 子图（Subgraph）

张

张建站

2026/6/30 1:52:37

10分钟阅读

05 子图Subgraph一、什么是子图子图Subgraph是 LangGraph 中的图嵌套机制允许将一个完整的图作为节点嵌入到另一个图中实现模块化设计。核心理念大图由多个小图组合而成每个小图负责一个独立的功能模块。概念说明类比父图包含子图的整体图主流程图子图作为节点嵌入的独立图子流程模块嵌套深度理论上无限制推荐 2-3 层流程图的子流程二、为什么需要子图2.1 代码组织问题当图变得复杂时所有节点堆在一个文件里会难以维护# ❌ 单一巨型图 - 难以维护graphStateGraph(State)graph.add_node(node_1,node_1)graph.add_node(node_2,node_2)# ... 50 个节点graph.add_node(node_50,node_50)2.2 子图的优势# ✅ 模块化设计 - 清晰的职责分离import_subgraphcreate_import_graph()# 导入模块search_subgraphcreate_search_graph()# 检索模块graphStateGraph(State)graph.add_node(import,import_subgraph)# 作为节点添加graph.add_node(search,search_subgraph)优势说明模块化每个子图独立封装一个业务模块可复用同一个子图可在多个父图中使用易测试子图可单独测试不依赖父图团队协作不同成员负责不同子图可视化图结构更清晰支持折叠展示三、子图的实现方式3.1 基本嵌套将编译后的子图直接作为节点添加到父图子图基本嵌套演示运行方式python 05_子图_基本嵌套.py fromtypingimportTypedDictfromlanggraph.graphimportStateGraph,START,END# 子图定义 classSubState(TypedDict):input:stroutput:strdefprocess_node(state:SubState):return{output:f处理:{state[input]}}sub_builderStateGraph(SubState)sub_builder.add_node(process,process_node)sub_builder.add_edge(START,process)sub_builder.add_edge(process,END)sub_graphsub_builder.compile()# 父图定义 classMainState(TypedDict):data:strresult:strdefprepare_node(state:MainState):return{data:准备完成}# 子图作为节点 - 需要适配状态defsubgraph_wrapper(state:MainState):# 调用子图并传递状态sub_resultsub_graph.invoke({input:state[data]})return{result:sub_result[output]}deffinalize_node(state:MainState):return{result:f最终结果:{state[result]}}main_builderStateGraph(MainState)main_builder.add_node(prepare,prepare_node)main_builder.add_node(sub_task,subgraph_wrapper)# 包装子图main_builder.add_node(finalize,finalize_node)main_builder.add_edge(START,prepare)main_builder.add_edge(prepare,sub_task)main_builder.add_edge(sub_task,finalize)main_builder.add_edge(finalize,END)main_graphmain_builder.compile()resultmain_graph.invoke({data:,result:})print(result)3.2 状态映射当父子图的状态结构不同时需要进行状态映射状态映射演示运行方式python 05_子图_状态映射.py fromtypingimportTypedDictfromlanggraph.graphimportStateGraph,START,END# 子图状态 - 专注于文档处理classDocState(TypedDict):doc_content:strdoc_metadata:dictparsed_result:str# 父图状态 - 包含更多业务字段classAppState(TypedDict):user_id:strdoc_path:strdoc_content:str# 需要映射到子图final_output:str# 子图节点defparse_doc(state:DocState):contentstate[doc_content]return{parsed_result:f解析完成:{content[:50]}...}doc_builderStateGraph(DocState)doc_builder.add_node(parse,parse_doc)doc_builder.add_edge(START,parse)doc_builder.add_edge(parse,END)doc_graphdoc_builder.compile()# 父图 - 包含状态映射逻辑defload_doc(state:AppState):return{doc_content:f文件内容来自:{state[doc_path]}}defcall_doc_parser(state:AppState):# 状态映射AppState - DocStatesub_input{doc_content:state[doc_content],doc_metadata:{}}sub_resultdoc_graph.invoke(sub_input)# 状态映射DocState - AppStatereturn{final_output:sub_result[parsed_result]}app_builderStateGraph(AppState)app_builder.add_node(load,load_doc)app_builder.add_node(parse,call_doc_parser)app_builder.add_edge(START,load)app_builder.add_edge(load,parse)app_builder.add_edge(parse,END)app_graphapp_builder.compile()resultapp_graph.invoke({user_id:u001,doc_path:test.pdf,doc_content:,final_output:})print(result)3.3 使用add_node直接添加子图LangGraph 支持直接将子图添加为节点无需手动包装直接添加子图演示运行方式python 05_子图_直接添加.py fromlanggraph.graphimportStateGraph,START,END# 子图defsub_node_1(state):return{value:step1}defsub_node_2(state):return{value:f{state[value]}- step2}sub_builderStateGraph(dict)sub_builder.add_node(s1,sub_node_1)sub_builder.add_node(s2,sub_node_2)sub_builder.add_edge(START,s1)sub_builder.add_edge(s1,s2)sub_builder.add_edge(s2,END)sub_graphsub_builder.compile()# 父图直接使用子图作为节点main_builderStateGraph(dict)main_builder.add_node(sub_process,sub_graph)# 直接添加编译后的子图main_builder.add_edge(START,sub_process)main_builder.add_edge(sub_process,END)main_graphmain_builder.compile()resultmain_graph.invoke({})print(result)# {value: step1 - step2}四、实际应用场景4.1 RAG 系统中的子图应用以掌柜智库项目为例导入流程和检索流程可以设计为独立子图 RAG 子图架构示例图结构 START ──→ import_subgraph ──→ search_subgraph ──→ END ├─ PDF解析 ├─ 意图识别 ├─ 文本切分 ├─ 向量检索 └─ 向量入库 └─ 答案生成 fromtypingimportTypedDict,Listfromlanggraph.graphimportStateGraph,START,END# 导入子图 classImportState(TypedDict):pdf_path:strmd_content:strchunks:List[str]import_status:strdefparse_pdf(state:ImportState):return{md_content:解析后的Markdown内容,import_status:parsing}defsplit_text(state:ImportState):return{chunks:[chunk1,chunk2,chunk3],import_status:splitting}defstore_vectors(state:ImportState):return{import_status:completed}import_builderStateGraph(ImportState)import_builder.add_node(parse,parse_pdf)import_builder.add_node(split,split_text)import_builder.add_node(store,store_vectors)import_builder.add_edge(START,parse)import_builder.add_edge(parse,split)import_builder.add_edge(split,store)import_builder.add_edge(store,END)import_graphimport_builder.compile()# 检索子图 classSearchState(TypedDict):query:strintent:strretrieved_docs:List[str]answer:strdefrecognize_intent(state:SearchState):return{intent:查询类}defvector_search(state:SearchState):return{retrieved_docs:[doc1,doc2]}defgenerate_answer(state:SearchState):return{answer:f基于检索结果的回答:{state[query]}}search_builderStateGraph(SearchState)search_builder.add_node(intent,recognize_intent)search_builder.add_node(search,vector_search)search_builder.add_node(generate,generate_answer)search_builder.add_edge(START,intent)search_builder.add_edge(intent,search)search_builder.add_edge(search,generate)search_builder.add_edge(generate,END)search_graphsearch_builder.compile()# 主图 - 组合两个子图 classMainState(TypedDict):mode:str# import 或 searchquery:strresult:strdefroute_mode(state:MainState):ifstate[mode]import:returnimport_processreturnsearch_processdefimport_process(state:MainState):sub_resultimport_graph.invoke({pdf_path:test.pdf,md_content:,chunks:[],import_status:})return{result:f导入完成:{sub_result[import_status]}}defsearch_process(state:MainState):sub_resultsearch_graph.invoke({query:state[query],intent:,retrieved_docs:[],answer:})return{result:sub_result[answer]}main_builderStateGraph(MainState)main_builder.add_node(import_process,import_process)main_builder.add_node(search_process,search_process)main_builder.add_conditional_edges(START,route_mode)main_builder.add_edge(import_process,END)main_builder.add_edge(search_process,END)main_graphmain_builder.compile()# 测试导入模式result1main_graph.invoke({mode:import,query:,result:})print(f导入结果:{result1})# 测试检索模式result2main_graph.invoke({mode:search,query:如何使用LangGraph,result:})print(f检索结果:{result2})4.2 多智能体系统子图非常适合构建多智能体协作系统多智能体协作示例图结构 START ──→ coordinator ──→ researcher ──→ writer ──→ END │ ↑ │ └────────────────┴──────────────┘ fromtypingimportTypedDict,Listfromlanggraph.graphimportStateGraph,START,END# 研究员子图classResearchState(TypedDict):topic:strfindings:List[str]defgather_info(state:ResearchState):return{findings:[f关于{state[topic]}的发现1,f关于{state[topic]}的发现2]}research_builderStateGraph(ResearchState)research_builder.add_node(gather,gather_info)research_builder.add_edge(START,gather)research_builder.add_edge(gather,END)research_graphresearch_builder.compile()# 作家子图classWriterState(TypedDict):findings:List[str]draft:strdefwrite_draft(state:WriterState):return{draft:f基于{len(state[findings])}个发现的初稿}writer_builderStateGraph(WriterState)writer_builder.add_node(write,write_draft)writer_builder.add_edge(START,write)writer_builder.add_edge(write,END)writer_graphwriter_builder.compile()# 协调器主图classCoordinatorState(TypedDict):topic:strfindings:List[str]draft:strstatus:strdefresearch_phase(state:CoordinatorState):resultresearch_graph.invoke({topic:state[topic]})return{findings:result[findings],status:研究完成}defwrite_phase(state:CoordinatorState):resultwriter_graph.invoke({findings:state[findings]})return{draft:result[draft],status:写作完成}coordinator_builderStateGraph(CoordinatorState)coordinator_builder.add_node(research,research_phase)coordinator_builder.add_node(write,write_phase)coordinator_builder.add_edge(START,research)coordinator_builder.add_edge(research,write)coordinator_builder.add_edge(write,END)coordinator_graphcoordinator_builder.compile()resultcoordinator_graph.invoke({topic:LangGraph子图,findings:[],draft:,status:})print(result)五、子图最佳实践5.1 设计原则原则说明示例单一职责每个子图只负责一个业务模块导入子图只处理文档导入接口清晰子图的状态结构就是它的接口ImportState定义输入输出独立可测子图可以脱离父图单独测试import_graph.invoke(...)状态最小化子图只接收必要的状态字段避免传递无关数据5.2 状态管理策略# 策略1: 共享状态 - 父子图使用相同的状态结构classSharedState(TypedDict):field_a:strfield_b:str# 策略2: 独立状态映射 - 子图有自己的状态通过函数映射classSubState(TypedDict):x:strdefwrapper(state:ParentState):sub_resultsub_graph.invoke({x:state[a]})return{b:sub_result[x]}5.3 调试技巧# 1. 单独测试子图sub_resultsub_graph.invoke({input:test})print(f子图输出:{sub_result})# 2. 查看子图结构print(sub_graph.get_graph().draw_ascii())# 3. 使用 LangSmith 追踪嵌套调用# 配置 LANGCHAIN_TRACING_V2true 即可自动追踪六、核心要点总结要点说明子图本质编译后的图可以作为节点添加到另一个图状态映射父子图状态不同时需要手动映射嵌套深度推荐 2-3 层过深影响可读性调试方式子图可独立测试LangSmith 支持嵌套追踪适用场景模块化设计、多智能体、复杂业务流程相关笔记[[01-LangGraph概述与快速入门]] · [[03-节点]] · [[04-边]]