LangGraph 12：子图

发表于 2026-08-20 分类于 ai

精选子图四种模式：节点嵌入、checkpointer 共享、子图 interrupt，以及跨图 Command goto 与流式输出。

前言

复杂 Agent 不宜把所有节点塞进一张巨图。
子图把「清洗文本」「多轮对话」「专项工具环」封装成可复用单元，再挂到父图。
本文不逐课展开，只保留四条最强模式：当节点嵌入、checkpointer 共享、子图内 HITL，以及跨图 Command 跳转（并附流式要点）。
示例需要处对接 火山方舟 Coding Plan，模型用 ark-code-latest。
下文需要 Python 3.12+，依赖用 uv 管理。

依赖

建议使用 Python 3.12 及以上。

uv init langgraph-subgraphs
cd langgraph-subgraphs
uv venv --python 3.12
uv add "langgraph>=1.0,<2.0" "langchain>=1.0,<2.0" langchain-openai python-dotenv rich

模式一：子图当节点

子图先 compile()，再 parent.add_node("subgraph_node", subgraph)。
父子共用同一状态 schema 时，字段会自动透传；schema 不同时，可在父节点函数里 subgraph.invoke(...) 做字段映射。

from typing import TypedDict

from langgraph.graph import END, START, StateGraph
from rich import print as rprint

class OverAllState(TypedDict):
    raw_text: str
    stripped_text: str
    punctuated_text: str

def strip_node(state: OverAllState) -> OverAllState:
    return {"stripped_text": state["raw_text"].strip()}

def punctuate_node(state: OverAllState) -> OverAllState:
    return {"punctuated_text": state["stripped_text"] + "。"}

sub = StateGraph(state_schema=OverAllState)
sub.add_node("strip", strip_node)
sub.add_node("punctuate", punctuate_node)
sub.add_edge(START, "strip")
sub.add_edge("strip", "punctuate")
sub.add_edge("punctuate", END)
subgraph = sub.compile()

parent = StateGraph(state_schema=OverAllState)
parent.add_node("subgraph_node", subgraph)
parent.add_edge(START, "subgraph_node")
parent.add_edge("subgraph_node", END)
parent_graph = parent.compile()

rprint(parent_graph.invoke({"raw_text": "   langgraph 真有意思   "}))

函数式调用适合「父状态字段名与子状态不一致」：在 call_subgraph 里组装输入、再取子图输出写回父状态。
get_graph(xray=True) 可把子图内部展开到可视化里。

模式二：checkpointer 共享

父图挂 InMemorySaver（或其它 saver）后，子图可写 compile(checkpointer=True)，表示复用父线程的持久化策略（per-thread）。
默认 compile() 多为 per-invocation；checkpointer=False 则无持久化，子图内 interrupt 通常不可用。

from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from rich import print as rprint

class SubgraphState(TypedDict):
    raw_text: str
    stripped_text: str
    punctuated_text: str

def strip_node(state: SubgraphState) -> SubgraphState:
    return {"stripped_text": state["raw_text"].strip()}

def punctuate_node(state: SubgraphState) -> SubgraphState:
    return {"punctuated_text": state["stripped_text"] + "。"}

sub = StateGraph(state_schema=SubgraphState)
sub.add_node("strip", strip_node)
sub.add_node("punctuate", punctuate_node)
sub.add_edge(START, "strip")
sub.add_edge("strip", "punctuate")
sub.add_edge("punctuate", END)
subgraph = sub.compile(checkpointer=True)

class ParentState(TypedDict):
    input_text: str
    cleaned_text: str

def call_subgraph(state: ParentState) -> ParentState:
    res = subgraph.invoke({"raw_text": state["input_text"]})
    return {"cleaned_text": res["punctuated_text"]}

parent = StateGraph(state_schema=ParentState)
parent.add_node("call_subgraph", call_subgraph)
parent.add_edge(START, "call_subgraph")
parent.add_edge("call_subgraph", END)
parent_graph = parent.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "parent-1"}}
rprint(parent_graph.invoke({"input_text": "  hello  "}, config=config))

需要查子图快照时，对父状态用 get_state(..., subgraphs=True)，或经 tasks[0].state 拿到子图 config 再 get_state_history。
命名空间字段 checkpoint_ns 用来区分不同子图实例。

模式三：子图内 interrupt

子图节点里照常 interrupt()；父图必须有 checkpointer，子图建议 checkpointer=True。
恢复时仍对父图 invoke(Command(resume=...))，同一 thread_id。

from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
from rich import print as rprint

class OverAllState(TypedDict):
    raw_text: str
    cleaned_text: str

def strip_node(state: OverAllState) -> OverAllState:
    return {"cleaned_text": state["raw_text"].strip()}

def punctuate_node(state: OverAllState) -> OverAllState:
    mark = interrupt("句尾标点选哪个？[。/！/？]")
    return {"cleaned_text": state["cleaned_text"] + mark}

sub = StateGraph(state_schema=OverAllState)
sub.add_node("strip", strip_node)
sub.add_node("punctuate", punctuate_node)
sub.add_edge(START, "strip")
sub.add_edge("strip", "punctuate")
sub.add_edge("punctuate", END)
subgraph = sub.compile(checkpointer=True)

parent = StateGraph(state_schema=OverAllState)
parent.add_node("subgraph_node", subgraph)
parent.add_edge(START, "subgraph_node")
parent.add_edge("subgraph_node", END)
parent_graph = parent.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "sub-hitl"}}
paused = parent_graph.invoke({"raw_text": "  LangGraph  "}, config=config)
rprint(paused["__interrupt__"])
done = parent_graph.invoke(Command(resume="！"), config=config)
rprint(done)

同一子图节点内多次 interrupt，就多次 Command(resume=...)，规则与顶层 HITL 一致。

模式四：跨图 goto

子图节点可返回 Command(goto="父图节点名", graph=Command.PARENT)，执行完后跳到父图其它节点。
父图 add_node 时用 destinations=(...) 声明可能目标，便于可视化与类型提示。

from typing import TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Command
from rich import print as rprint

class SubState(TypedDict):
    data: str

def sub_node(state: SubState) -> Command:
    return Command(
        update={"data": state["data"] + " → 子图"},
        goto="node_b",
        graph=Command.PARENT,
    )

sub = StateGraph(state_schema=SubState)
sub.add_node("sub_node", sub_node)
sub.add_edge(START, "sub_node")
sub_graph = sub.compile()

class ParentState(TypedDict):
    data: str

def node_a(state: ParentState) -> ParentState:
    return {"data": state["data"] + " → A"}

def node_b(state: ParentState) -> ParentState:
    return {"data": state["data"] + " → B"}

parent = StateGraph(state_schema=ParentState)
parent.add_node("sub_graph", sub_graph, destinations=("node_a", "node_b"))
parent.add_node("node_a", node_a)
parent.add_node("node_b", node_b)
parent.add_edge(START, "sub_graph")
parent.add_edge("node_a", "node_b")
parent.add_edge("node_b", END)
parent_graph = parent.compile()

rprint(parent_graph.invoke({"data": "初始"}))

流式附注

对父图 stream(..., subgraphs=True, stream_mode=[...]) 可把子图内部 updates / messages 一并冒泡。
函数节点里调用 subgraph.invoke 时，子图 LLM 的 token 流是否可见，取决于封装方式；需要细粒度 token 时优先「子图当节点」或在文档推荐的 stream 组合下试验。
多轮记忆场景让子图 checkpointer=True，父线程固定 thread_id，子图 messages 才能跨次调用延续。

总结

复用单元：编译子图后 add_node，或函数内 invoke 做字段映射。
要持久化 / HITL：父图挂 saver，子图 compile(checkpointer=True)。
子图 interrupt 仍用父图 Command(resume=...) 恢复。
Command.PARENT + goto 实现子图决定父图下一跳；流式加 subgraphs=True。