一、什么是子图1.1 通俗解释想象你在搭建一个大型乐高城堡没有子图 你需要一块一块地搭建整个城堡 每次都要从零开始很难复用 有子图 你可以先搭建好塔楼、城墙、大门等组件 然后像搭积木一样组合这些组件 每个组件都可以独立使用或复用子图就是把一个完整的图当作另一个图的节点来使用父图Parent Graph ├── 节点A ├── 子图B本身就是一个完整的图 │ ├── 节点B1 │ ├── 节点B2 │ └── 节点B3 └── 节点C 执行顺序A → B(B1→B2→B3) → C1.2 子图的作用作用说明例子构建多代理系统每个代理是一个子图研究代理 写作代理 审核代理代码复用同一套逻辑多处使用多个地方都需要搜索总结流程模块化开发不同团队独立开发团队A开发子图A团队B开发子图B封装复杂性隐藏内部实现细节外部只需知道输入输出1.3 核心概念┌─────────────────────────────────────────────────────────────┐ │ 子图架构示意 │ ├─────────────────────────────────────────────────────────────┤ │ │ │ 父图 │ │ ┌─────────────────────────────────────────────────────┐ │ │ │ │ │ │ │ 节点1 ──▶ 子图节点 ──▶ 节点3 │ │ │ │ │ │ │ │ │ ▼ │ │ │ │ ┌─────────┐ │ │ │ │ │ 子图 │ │ │ │ │ │ 节点A │ │ │ │ │ │ ↓ │ │ │ │ │ │ 节点B │ │ │ │ │ │ ↓ │ │ │ │ │ │ 节点C │ │ │ │ │ └─────────┘ │ │ │ │ │ │ │ └─────────────────────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────────────────────┘二、父图与子图的通信方式2.1 两种通信方式子图的关键问题是父图和子图如何交换数据方式一共享状态模式 父图和子图有相同的状态键如 messages 数据自动传递无需转换 方式二不同状态模式 父图和子图的状态完全不同 需要手动转换数据2.2 方式对比对比项共享状态模式不同状态模式状态定义父子图有共同的状态键父子图状态完全不同数据传递自动传递手动转换实现方式直接添加子图为节点在节点函数中调用子图适用场景多代理共享消息历史每个子图需要私有状态复杂度简单较复杂三、共享状态模式3.1 什么是共享状态父图状态{foo: str, bar: str} 子图状态{foo: str, baz: str} 共享状态键foo 私有状态键bar父图独有、baz子图独有3.2 使用方法步骤定义子图并编译将编译后的子图直接添加为父图的节点from typing_extensions import TypedDict from langgraph.graph.state import StateGraph, START # 定义共享状态 class State(TypedDict): foo: str # 子图定义 def subgraph_node_1(state: State): return {foo: hi! state[foo]} subgraph_builder StateGraph(State) subgraph_builder.add_node(subgraph_node_1) subgraph_builder.add_edge(START, subgraph_node_1) subgraph subgraph_builder.compile() # 父图定义 builder StateGraph(State) builder.add_node(node_1, subgraph) # 直接添加子图 builder.add_edge(START, node_1) graph builder.compile()3.3 完整示例from typing_extensions import TypedDict from langgraph.graph.state import StateGraph, START # 子图状态 class SubgraphState(TypedDict): foo: str # 与父图共享 bar: str # 子图私有 def subgraph_node_1(state: SubgraphState): return {bar: bar} def subgraph_node_2(state: SubgraphState): # 使用子图私有的 bar更新共享的 foo return {foo: state[foo] state[bar]} subgraph_builder StateGraph(SubgraphState) subgraph_builder.add_node(subgraph_node_1) subgraph_builder.add_node(subgraph_node_2) subgraph_builder.add_edge(START, subgraph_node_1) subgraph_builder.add_edge(subgraph_node_1, subgraph_node_2) subgraph subgraph_builder.compile() # 父图状态 class ParentState(TypedDict): foo: str # 与子图共享 def node_1(state: ParentState): return {foo: hi! state[foo]} builder StateGraph(ParentState) builder.add_node(node_1, node_1) builder.add_node(node_2, subgraph) # 添加子图 builder.add_edge(START, node_1) builder.add_edge(node_1, node_2) graph builder.compile() # 执行 for chunk in graph.stream({foo: foo}): print(chunk)输出{node_1: {foo: hi! foo}} {node_2: {foo: hi! foobar}}3.4 执行流程图输入: {foo: foo} │ ▼ node_1: {foo: hi! foo} │ ▼ 子图 node_2: │ ├── subgraph_node_1: {bar: bar} │ └── subgraph_node_2: {foo: hi! foobar} │ ▼ 输出: {foo: hi! foobar}四、不同状态模式4.1 什么是不同状态父图状态{foo: str} 子图状态{bar: str, baz: str} 没有共享状态键 需要在调用时手动转换数据4.2 使用方法步骤定义子图并编译创建一个节点函数在其中调用子图在节点函数中进行数据转换from typing_extensions import TypedDict from langgraph.graph.state import StateGraph, START # 子图状态与父图完全不同 class SubgraphState(TypedDict): bar: str def subgraph_node_1(state: SubgraphState): return {bar: hi! state[bar]} subgraph_builder StateGraph(SubgraphState) subgraph_builder.add_node(subgraph_node_1) subgraph_builder.add_edge(START, subgraph_node_1) subgraph subgraph_builder.compile() # 父图状态 class State(TypedDict): foo: str # 调用子图的节点函数 def call_subgraph(state: State): # 转换父图状态 → 子图状态 subgraph_input {bar: state[foo]} # 调用子图 subgraph_output subgraph.invoke(subgraph_input) # 转换子图状态 → 父图状态 return {foo: subgraph_output[bar]} builder StateGraph(State) builder.add_node(node_1, call_subgraph) builder.add_edge(START, node_1) graph builder.compile()4.3 完整示例from typing_extensions import TypedDict from langgraph.graph.state import StateGraph, START # 子图状态 class SubgraphState(TypedDict): bar: str # 子图私有 baz: str # 子图私有 def subgraph_node_1(state: SubgraphState): return {baz: baz} def subgraph_node_2(state: SubgraphState): return {bar: state[bar] state[baz]} subgraph_builder StateGraph(SubgraphState) subgraph_builder.add_node(subgraph_node_1) subgraph_builder.add_node(subgraph_node_2) subgraph_builder.add_edge(START, subgraph_node_1) subgraph_builder.add_edge(subgraph_node_1, subgraph_node_2) subgraph subgraph_builder.compile() # 父图状态 class ParentState(TypedDict): foo: str # 父图私有 def node_1(state: ParentState): return {foo: hi! state[foo]} def node_2(state: ParentState): # 转换输入 subgraph_input {bar: state[foo], baz: } # 调用子图 response subgraph.invoke(subgraph_input) # 转换输出 return {foo: response[bar]} builder StateGraph(ParentState) builder.add_node(node_1, node_1) builder.add_node(node_2, node_2) builder.add_edge(START, node_1) builder.add_edge(node_1, node_2) graph builder.compile() # 执行显示子图输出 for chunk in graph.stream({foo: foo}, subgraphsTrue): print(chunk)输出((), {node_1: {foo: hi! foo}}) ((node_2:xxx,), {subgraph_node_1: {baz: baz}}) ((node_2:xxx,), {subgraph_node_2: {bar: hi! foobaz}}) ((), {node_2: {foo: hi! foobaz}})4.4 数据转换流程父图状态: {foo: hi! foo} │ ▼ 转换输入 子图输入: {bar: hi! foo, baz: } │ ▼ 执行子图 subgraph_node_1: {baz: baz} subgraph_node_2: {bar: hi! foobaz} │ ▼ 转换输出 父图状态: {foo: hi! foobaz}五、多级子图5.1 什么是多级子图父图 └── 子图 └── 孙子图 三层嵌套5.2 完整示例from typing_extensions import TypedDict from langgraph.graph.state import StateGraph, START, END # 孙子图 class GrandChildState(TypedDict): my_grandchild_key: str def grandchild_1(state: GrandChildState): return {my_grandchild_key: state[my_grandchild_key] , how are you} grandchild StateGraph(GrandChildState) grandchild.add_node(grandchild_1, grandchild_1) grandchild.add_edge(START, grandchild_1) grandchild.add_edge(grandchild_1, END) grandchild_graph grandchild.compile() # 子图 class ChildState(TypedDict): my_child_key: str def call_grandchild_graph(state: ChildState): # 转换输入 grandchild_input {my_grandchild_key: state[my_child_key]} # 调用孙子图 grandchild_output grandchild_graph.invoke(grandchild_input) # 转换输出 return {my_child_key: grandchild_output[my_grandchild_key] today?} child StateGraph(ChildState) child.add_node(child_1, call_grandchild_graph) child.add_edge(START, child_1) child.add_edge(child_1, END) child_graph child.compile() # 父图 class ParentState(TypedDict): my_key: str def parent_1(state: ParentState): return {my_key: hi state[my_key]} def parent_2(state: ParentState): return {my_key: state[my_key] bye!} def call_child_graph(state: ParentState): # 转换输入 child_input {my_child_key: state[my_key]} # 调用子图 child_output child_graph.invoke(child_input) # 转换输出 return {my_key: child_output[my_child_key]} parent StateGraph(ParentState) parent.add_node(parent_1, parent_1) parent.add_node(child, call_child_graph) parent.add_node(parent_2, parent_2) parent.add_edge(START, parent_1) parent.add_edge(parent_1, child) parent.add_edge(child, parent_2) parent.add_edge(parent_2, END) parent_graph parent.compile() # 执行 for chunk in parent_graph.stream({my_key: Bob}, subgraphsTrue): print(chunk)输出((), {parent_1: {my_key: hi Bob}}) ((child:xxx, child_1:yyy), {grandchild_1: {my_grandchild_key: hi Bob, how are you}}) ((child:xxx,), {child_1: {my_child_key: hi Bob, how are you today?}}) ((), {child: {my_key: hi Bob, how are you today?}}) ((), {parent_2: {my_key: hi Bob, how are you today? bye!}})六、添加持久化6.1 父图持久化只需在编译父图时提供 checkpointer会自动传播到子图from langgraph.graph import START, StateGraph from langgraph.checkpoint.memory import InMemorySaver # 子图 subgraph_builder StateGraph(State) subgraph_builder.add_node(subgraph_node_1) subgraph_builder.add_edge(START, subgraph_node_1) subgraph subgraph_builder.compile() # 父图 builder StateGraph(State) builder.add_node(node_1, subgraph) builder.add_edge(START, node_1) # 只需在父图添加 checkpointer checkpointer InMemorySaver() graph builder.compile(checkpointercheckpointer)6.2 子图独立持久化如果子图需要独立的记忆如多代理系统中每个代理跟踪自己的消息历史# 子图独立持久化 subgraph_builder StateGraph(...) subgraph subgraph_builder.compile(checkpointerTrue)七、查看子图状态7.1 获取子图状态# 获取父图状态 parent_state graph.get_state(config) # 获取子图状态需要 subgraphsTrue state_with_subgraphs graph.get_state(config, subgraphsTrue) subgraph_state state_with_subgraphs.tasks[0].state7.2 注意事项重要子图状态只能在中断时查看 一旦恢复执行子图状态将无法访问。7.3 完整示例from langgraph.graph import START, StateGraph from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import interrupt, Command # 子图包含中断 def subgraph_node_1(state: State): value interrupt(请输入值) return {foo: state[foo] value} subgraph_builder StateGraph(State) subgraph_builder.add_node(subgraph_node_1) subgraph_builder.add_edge(START, subgraph_node_1) subgraph subgraph_builder.compile() # 父图 builder StateGraph(State) builder.add_node(node_1, subgraph) builder.add_edge(START, node_1) checkpointer InMemorySaver() graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: 1}} # 执行会中断 graph.invoke({foo: }, config) # 查看子图状态 subgraph_state graph.get_state(config, subgraphsTrue).tasks[0].state print(subgraph_state) # 恢复执行 graph.invoke(Command(resumebar), config)八、流式传输子图输出8.1 启用子图流式输出for chunk in graph.stream( {foo: foo}, subgraphsTrue, # 启用子图输出 stream_modeupdates, ): print(chunk)8.2 输出格式((), {node_1: {foo: hi! foo}}) # 父图节点 ((node_2:xxx,), {subgraph_node_1: {bar: bar}}) # 子图节点 ((node_2:xxx,), {subgraph_node_2: {foo: hi! foobar}}) # 子图节点 ((), {node_2: {foo: hi! foobar}}) # 父图节点格式说明()空元组父图节点(node_2:xxx,)子图路径子图节点8.3 完整示例from typing_extensions import TypedDict from langgraph.graph.state import StateGraph, START # 子图 class SubgraphState(TypedDict): foo: str bar: str def subgraph_node_1(state: SubgraphState): return {bar: bar} def subgraph_node_2(state: SubgraphState): return {foo: state[foo] state[bar]} subgraph_builder StateGraph(SubgraphState) subgraph_builder.add_node(subgraph_node_1) subgraph_builder.add_node(subgraph_node_2) subgraph_builder.add_edge(START, subgraph_node_1) subgraph_builder.add_edge(subgraph_node_1, subgraph_node_2) subgraph subgraph_builder.compile() # 父图 class ParentState(TypedDict): foo: str def node_1(state: ParentState): return {foo: hi! state[foo]} builder StateGraph(ParentState) builder.add_node(node_1, node_1) builder.add_node(node_2, subgraph) builder.add_edge(START, node_1) builder.add_edge(node_1, node_2) graph builder.compile() # 流式输出包含子图 for chunk in graph.stream( {foo: foo}, stream_modeupdates, subgraphsTrue, ): print(chunk)九、实际应用场景9.1 多代理系统# 研究代理子图 research_agent create_react_agent(model, tools[search]) # 写作代理子图 writing_agent create_react_agent(model, tools[write_file]) # 审核代理子图 review_agent create_react_agent(model, tools[send_email]) # 主流程父图 builder StateGraph(MessagesState) builder.add_node(research, research_agent) builder.add_node(writing, writing_agent) builder.add_node(review, review_agent) builder.add_edge(START, research) builder.add_edge(research, writing) builder.add_edge(writing, review)9.2 模块化工作流# 数据处理模块子图 data_processing StateGraph(DataState) data_processing.add_node(clean, clean_data) data_processing.add_node(transform, transform_data) data_processing_graph data_processing.compile() # 分析模块子图 analysis StateGraph(AnalysisState) analysis.add_node(analyze, analyze_data) analysis.add_node(visualize, visualize_data) analysis_graph analysis.compile() # 主流程父图 builder StateGraph(MainState) builder.add_node(process, data_processing_graph) builder.add_node(analyze, analysis_graph)十、常见问题Q1: 什么时候用共享状态什么时候用不同状态共享状态 - 多代理共享消息历史 - 状态结构相似 - 简单场景 不同状态 - 每个子图需要私有状态 - 状态结构完全不同 - 需要数据转换Q2: 子图可以有多少层理论上没有限制但建议不超过3层 推荐父图 → 子图 → 孙子图最多3层 不推荐父图 → 子图 → 孙子图 → 曾孙图 → ...Q3: 如何调试子图# 方法1流式输出 for chunk in graph.stream(input, subgraphsTrue): print(chunk) # 方法2查看状态 state graph.get_state(config, subgraphsTrue) # 方法3可视化 from IPython.display import Image Image(subgraph.get_graph().draw_mermaid_png())Q4: 子图的 checkpointer 如何配置# 方式1父图统一管理推荐 graph builder.compile(checkpointerInMemorySaver()) # 方式2子图独立管理 subgraph subgraph_builder.compile(checkpointerTrue)十一、API 速查表11.1 添加子图方式代码说明共享状态builder.add_node(name, subgraph)直接添加不同状态builder.add_node(name, call_subgraph)在函数中调用11.2 数据转换操作代码转换输入subgraph_input {bar: state[foo]}调用子图output subgraph.invoke(subgraph_input)转换输出return {foo: output[bar]}11.3 查看状态方法代码父图状态graph.get_state(config)子图状态graph.get_state(config, subgraphsTrue)11.4 流式输出参数说明subgraphsTrue包含子图输出stream_modeupdates更新模式十二、延伸阅读LangGraph 官方文档 - 子图多代理系统持久化总结子图的核心要点定义子图像普通图一样定义和编译添加到父图共享状态直接添加不同状态需要转换函数数据传递共享状态自动传递不同状态手动转换持久化在父图配置即可自动传播子图的作用构建多代理系统代码复用模块化开发封装复杂性一句话总结子图就是把一个完整的图当作节点使用像搭积木一样构建复杂系统。
