子图 - 以编程方式使用文档

本指南解释了使用子图的机制。子图是一个图，用作节点在另一个图中。

子图可用于： - 构建多智能体系统 - 在多个图中重用一组节点 - 分布式开发：当您希望不同的团队独立处理图的不同部分时，可以将每个部分定义为一个子图，只要遵守子图接口（输入和输出模式），父图就可以在不了解子图任何细节的情况下构建

设置

pip install -U langgraph

uv add langgraph

定义子图通信

添加子图时，您需要定义父图与子图之间的通信方式：

模式	使用场景	状态模式
在节点内调用子图	父图和子图具有不同的状态模式（没有共享的键），或者需要在它们之间转换状态	您编写一个包装函数，将父状态映射到子图输入，并将子图输出映射回父状态
将子图添加为节点	父图和子图共享状态键—子图从父图读取并写入相同的通道	您直接将编译后的子图传递给 `add_node`—无需包装函数

<a id="invoke-a-graph-from-a-node"></a> ### 在节点内调用子图

当父图和子图具有 **不同的状态模式** （没有共享的键）时，在节点函数中调用子图。当您想要为每个代理保留私有消息历史时，这很常见 multi-agent system.

节点函数在调用子图之前将父状态转换为子图状态，并在返回之前将结果转换回父状态。

from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START

class SubgraphState(TypedDict):
    bar: str

# Subgraph

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()

# Parent graph

class State(TypedDict):
    foo: str

def call_subgraph(state: State):
    # Transform the state to the subgraph state
    subgraph_output = subgraph.invoke({"bar": state["foo"]})  # [!code highlight]
    # Transform response back to the parent state
    return {"foo": subgraph_output["bar"]}

builder = StateGraph(State)
builder.add_node("node_1", call_subgraph)
builder.add_edge(START, "node_1")
graph = builder.compile()

Full example: different state schemas

  from typing_extensions import TypedDict
  from langgraph.graph.state import StateGraph, START

  # Define subgraph
  class SubgraphState(TypedDict):
      # note that none of these keys are shared with the parent graph state
      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()

  # Define parent graph
  class ParentState(TypedDict):
      foo: str

  def node_1(state: ParentState):
      return {"foo": "hi! " + state["foo"]}

  def node_2(state: ParentState):
      # Transform the state to the subgraph state
      response = subgraph.invoke({"bar": state["foo"]})
      # Transform response back to the parent state
      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()

  stream = graph.stream_events({"foo": "foo"}, version="v3")
  for event in stream:
      if event["method"] == "updates":
          print(event["params"]["namespace"], event["params"]["data"])

  [] {'node_1': {'foo': 'hi! foo'}}
  ['node_2:577b710b-64ae-31fb-9455-6a4d4cc2b0b9'] {'subgraph_node_1': {'baz': 'baz'}}
  ['node_2:577b710b-64ae-31fb-9455-6a4d4cc2b0b9'] {'subgraph_node_2': {'bar': 'hi! foobaz'}}
  [] {'node_2': {'foo': 'hi! foobaz'}}

Full example: different state schemas (two levels of subgraphs)

这是一个具有两层子图的示例：父图 -> 子图 -> 孙图。

  # Grandchild graph
  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) -> GrandChildState:
      # NOTE: child or parent keys will not be accessible here
      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()

  # Child graph
  class ChildState(TypedDict):
      my_child_key: str

  def call_grandchild_graph(state: ChildState) -> ChildState:
      # NOTE: parent or grandchild keys won't be accessible here
      grandchild_graph_input = {"my_grandchild_key": state["my_child_key"]}
      grandchild_graph_output = grandchild_graph.invoke(grandchild_graph_input)
      return {"my_child_key": grandchild_graph_output["my_grandchild_key"] + " today?"}

  child = StateGraph(ChildState)
  # We're passing a function here instead of just compiled graph (`grandchild_graph`)
  child.add_node("child_1", call_grandchild_graph)
  child.add_edge(START, "child_1")
  child.add_edge("child_1", END)
  child_graph = child.compile()

  # Parent graph
  class ParentState(TypedDict):
      my_key: str

  def parent_1(state: ParentState) -> ParentState:
      # NOTE: child or grandchild keys won't be accessible here
      return {"my_key": "hi " + state["my_key"]}

  def parent_2(state: ParentState) -> ParentState:
      return {"my_key": state["my_key"] + " bye!"}

  def call_child_graph(state: ParentState) -> ParentState:
      child_graph_input = {"my_child_key": state["my_key"]}
      child_graph_output = child_graph.invoke(child_graph_input)
      return {"my_key": child_graph_output["my_child_key"]}

  parent = StateGraph(ParentState)
  parent.add_node("parent_1", parent_1)
  # We're passing a function here instead of just a compiled graph (`child_graph`)
  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()

  stream = parent_graph.stream_events({"my_key": "Bob"}, version="v3")
  for event in stream:
      if event["method"] == "updates":
          print(event["params"]["namespace"], event["params"]["data"])

  [] {'parent_1': {'my_key': 'hi Bob'}}
  ['child:2e26e9ce-602f-862c-aa66-1ea5a4655e3b', 'child_1:781bb3b1-3971-84ce-810b-acf819a03f9c'] {'grandchild_1': {'my_grandchild_key': 'hi Bob, how are you'}}
  ['child:2e26e9ce-602f-862c-aa66-1ea5a4655e3b'] {'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!'}}

<a id="add-a-graph-as-a-node"></a> ### 将子图添加为节点

当父图和子图 **共享状态键**，您可以直接将编译好的子图传递给 add_node。无需包装函数——子图会自动读写父级的状态通道。例如，在 multi-agent 系统中，代理通常通过共享的消息 key.

如果您的子图与父图共享状态键，您可以按照以下步骤将其添加到您的图中：

定义子图工作流 (subgraph_builder 在下面的示例中) 并编译它
将编译好的子图传递给 add_node 方法，在定义父图工作流时

from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START

class State(TypedDict):
    foo: str

# Subgraph

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()

# Parent graph

builder = StateGraph(State)
builder.add_node("node_1", subgraph)  # [!code highlight]
builder.add_edge(START, "node_1")
graph = builder.compile()

Full example: shared state schemas

  from typing_extensions import TypedDict
  from langgraph.graph.state import StateGraph, START

  # Define subgraph
  class SubgraphState(TypedDict):
      foo: str  # shared with parent graph state
      bar: str  # private to SubgraphState

  def subgraph_node_1(state: SubgraphState):
      return {"bar": "bar"}

  def subgraph_node_2(state: SubgraphState):
      # note that this node is using a state key ('bar') that is only available in the subgraph
      # and is sending update on the shared state key ('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()

  # Define parent graph
  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()

  stream = graph.stream_events({"foo": "foo"}, version="v3")
  for event in stream:
      if event["method"] == "updates" and not event["params"]["namespace"]:
          print(event["params"]["data"])

  {'node_1': {'foo': 'hi! foo'}}
  {'node_2': {'foo': 'hi! foobar'}}

子图持久化

使用子图时，需要决定其内部数据在调用之间的处理方式。考虑一个将请求委托给专业子代理的客户支持机器人："账单专家"子代理应该记住客户之前的问题，还是每次调用时都重新开始？

checkpointer 上的 .compile() 参数控制子图持久化：

模式	`checkpointer=`	行为
Per-invocation	`None` （默认）	每次调用都会重新开始，并继承父图的检查点以支持中断和持久执行在单次调用中。
Per-thread	`True`	状态在同一线程的调用之间累积。每次调用都会从上一次的结束位置继续。
无状态	`False`	完全不进行保存点检查——像普通函数调用一样运行。不支持中断或持久执行。

对于大多数应用程序来说，每次调用是正确选择，包括 multi-agent 子代理处理独立请求的系统。当子代理需要多轮对话记忆时使用按线程（例如，一个在多次交互中积累上下文的研究助手）。

有状态

有状态子图继承父图的检查点器，这使得中断, 持久化和状态检查成为可能。两种有状态模式的区别在于状态保留的时长。

每次调用（默认）

当每次调用子图都是独立的，且子代理不需要记住之前调用的任何信息时，请使用每次调用持久化。这是最常见的模式，特别是对于 multi-agent 子代理处理一次性请求的系统，如“查询此客户的订单”或“总结此文档”。

省略 checkpointer 或将其设置为 None。每次调用都是全新的，但在单次调用内，子图继承父图的检查点器，并可以使用 interrupt() 来暂停和恢复。

以下示例使用两个子代理（水果专家、蔬菜专家）作为工具包装，供外部代理调用：

from langchain.agents import create_agent
from langchain.tools import tool
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command, interrupt

@tool
def fruit_info(fruit_name: str) -> str:
    """Look up fruit info."""
    return f"Info about {fruit_name}"

@tool
def veggie_info(veggie_name: str) -> str:
    """Look up veggie info."""
    return f"Info about {veggie_name}"

# Subagents - no checkpointer setting (inherits parent)
fruit_agent = create_agent(
    model="gpt-5.4-mini",
    tools=[fruit_info],
    prompt="You are a fruit expert. Use the fruit_info tool. Respond in one sentence.",
)

veggie_agent = create_agent(
    model="gpt-5.4-mini",
    tools=[veggie_info],
    prompt="You are a veggie expert. Use the veggie_info tool. Respond in one sentence.",
)

# Wrap subagents as tools for the outer agent
@tool
def ask_fruit_expert(question: str) -> str:
    """Ask the fruit expert. Use for ALL fruit questions."""
    response = fruit_agent.invoke(
        {"messages": [{"role": "user", "content": question}]},
    )
    return response["messages"][-1].content

@tool
def ask_veggie_expert(question: str) -> str:
    """Ask the veggie expert. Use for ALL veggie questions."""
    response = veggie_agent.invoke(
        {"messages": [{"role": "user", "content": question}]},
    )
    return response["messages"][-1].content

# Outer agent with checkpointer
agent = create_agent(
    model="gpt-5.4-mini",
    tools=[ask_fruit_expert, ask_veggie_expert],
    prompt=(
        "You have two experts: ask_fruit_expert and ask_veggie_expert. "
        "ALWAYS delegate questions to the appropriate expert."
    ),
    checkpointer=MemorySaver(),
)

Interrupts

每次调用都可以使用 interrupt() 来暂停和恢复。添加 interrupt() 到工具函数中，以要求用户批准后才能继续：

  @tool
  def fruit_info(fruit_name: str) -> str:
      """Look up fruit info."""
      interrupt("continue?")  # [!code highlight]
      return f"Info about {fruit_name}"

Multi-turn

每次调用都以全新的子代理状态开始。子代理不记得之前的调用：

  config = {"configurable": {"thread_id": "1"}}

  # First call
  response = agent.invoke(
      {"messages": [{"role": "user", "content": "Tell me about apples"}]},
      config=config,
  )
  # Subagent message count: 4

  # Second call - subagent starts fresh, no memory of apples
  response = agent.invoke(
      {"messages": [{"role": "user", "content": "Now tell me about bananas"}]},
      config=config,
  )
  # Subagent message count: 4 (still fresh!)

Multiple subgraph calls

对同一子图的多次调用不会产生冲突，因为每次调用都有各自独立的检查点命名空间：

  config = {"configurable": {"thread_id": "1"}}

  # LLM calls ask_fruit_expert for both apples and bananas
  response = agent.invoke(
      {"messages": [{"role": "user", "content": "Tell me about apples and bananas"}]},
      config=config,
  )
  # Subagent message count: 4 (apples - fresh)
  # Subagent message count: 4 (bananas - fresh)

Per-thread

当子代理需要记住之前的交互时使用每线程持久化。例如，在多个交换中积累上下文的研究助手，或跟踪已编辑文件列表的编码助手。子代理的对话历史和数据在同线程的调用中累积。每次调用从上次停止的地方继续。

编译时使用 checkpointer=True 来启用此行为。

以下示例使用水果专家子代理进行编译 checkpointer=True:

from langchain.agents import create_agent
from langchain.agents.middleware import ToolCallLimitMiddleware
from langchain.tools import tool
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command, interrupt

@tool
def fruit_info(fruit_name: str) -> str:
    """Look up fruit info."""
    return f"Info about {fruit_name}"

# Subagent with checkpointer=True for persistent state
fruit_agent = create_agent(
    model="gpt-5.4-mini",
    tools=[fruit_info],
    prompt="You are a fruit expert. Use the fruit_info tool. Respond in one sentence.",
    checkpointer=True,  # [!code highlight]
)

# Wrap subagent as a tool for the outer agent
@tool
def ask_fruit_expert(question: str) -> str:
    """Ask the fruit expert. Use for ALL fruit questions."""
    response = fruit_agent.invoke(
        {"messages": [{"role": "user", "content": question}]},
    )
    return response["messages"][-1].content

# Outer agent with checkpointer
# Use ToolCallLimitMiddleware to prevent parallel calls to per-thread subagents,
# which would cause checkpoint conflicts.
agent = create_agent(
    model="gpt-5.4-mini",
    tools=[ask_fruit_expert],
    prompt="You have a fruit expert. ALWAYS delegate fruit questions to ask_fruit_expert.",
    middleware=[  # [!code highlight]
        ToolCallLimitMiddleware(tool_name="ask_fruit_expert", run_limit=1),  # [!code highlight]
    ],  # [!code highlight]
    checkpointer=MemorySaver(),
)

Interrupts

每线程子代理支持 interrupt() 与每次调用相同。添加 interrupt() 到工具函数以要求用户批准：

  @tool
  def fruit_info(fruit_name: str) -> str:
      """Look up fruit info."""
      interrupt("continue?")  # [!code highlight]
      return f"Info about {fruit_name}"

Multi-turn

状态在多次调用中累积——子代理会记住过去的对话：

  config = {"configurable": {"thread_id": "1"}}

  # First call
  response = agent.invoke(
      {"messages": [{"role": "user", "content": "Tell me about apples"}]},
      config=config,
  )
  # Subagent message count: 4

  # Second call - subagent REMEMBERS apples conversation
  response = agent.invoke(
      {"messages": [{"role": "user", "content": "Now tell me about bananas"}]},
      config=config,
  )
  # Subagent message count: 8 (accumulated!)

Multiple subgraph calls

当你有多个 **不同的** 每线程子图（例如，水果专家和蔬菜专家），每个子图都需要自己的存储空间，这样它们的检查点就不会相互覆盖。这被称为 **命名空间隔离**.

如果你在节点内调用子图，LangGraph会根据调用顺序分配命名空间（第一次调用、第二次调用等）。这意味着重新排序调用可能会混淆哪个子图加载哪个状态。为了避免这种情况，请将每个子代理包装在其自己的 StateGraph 中并使用唯一的节点名称——这为每个子图提供了稳定、唯一的命名空间：

  from langgraph.graph import MessagesState, StateGraph

  def create_sub_agent(model, *, name, **kwargs):
      """Wrap an agent with a unique node name for namespace isolation."""
      agent = create_agent(model=model, name=name, **kwargs)
      return (
          StateGraph(MessagesState)
          .add_node(name, agent)  # unique name → stable namespace  # [!code highlight]
          .add_edge("__start__", name)
          .compile()
      )

  fruit_agent = create_sub_agent(
      "gpt-5.4-mini", name="fruit_agent",
      tools=[fruit_info], prompt="...", checkpointer=True,
  )
  veggie_agent = create_sub_agent(
      "gpt-5.4-mini", name="veggie_agent",
      tools=[veggie_info], prompt="...", checkpointer=True,
  )

  config = {"configurable": {"thread_id": "1"}}

  # First call - LLM calls both fruit and veggie experts
  response = agent.invoke(
      {"messages": [{"role": "user", "content": "Tell me about cherries and broccoli"}]},
      config=config,
  )
  # Fruit subagent message count: 4
  # Veggie subagent message count: 4

  # Second call - both agents accumulate independently
  response = agent.invoke(
      {"messages": [{"role": "user", "content": "Now tell me about oranges and carrots"}]},
      config=config,
  )
  # Fruit subagent message count: 8 (remembers cherries!)
  # Veggie subagent message count: 8 (remembers broccoli!)

子图作为节点添加时已经自动获得基于名称的命名空间，因此不需要此包装器。

无状态

Use this when you want to run a subagent like a plain function call with no checkpointing overhead. The subgraph cannot pause/resume and does not benefit from 持久执行。使用以下方式编译 checkpointer=False.

subgraph_builder = StateGraph(...)
subgraph = subgraph_builder.compile(checkpointer=False)  # [!code highlight]

检查点引用

使用 checkpointer 上的参数控制子图持久化 .compile():

subgraph = builder.compile(checkpointer=False)  # or True / None

功能	每次调用（默认）	每线程	无状态
`checkpointer=`	`None`	`True`	`False`
中断（HITL）	✅	✅	❌
多轮记忆	❌	✅	❌
多次调用（不同子图）	✅	⚠️	✅
多次调用（相同子图）	✅	❌	✅
状态检查	⚠️	✅	❌

中断 (HITL)：子图可以使用 interrupt() 来暂停执行并等待用户输入，然后从中断处继续。
多轮记忆：子图在同一个线程内的多次调用中保留其状态。每次调用都从上一个的断点继续，而不是重新开始。
多次调用（不同子图）：多个不同的子图实例可以在单个节点内被调用，而不会产生检查点命名空间冲突。
多次调用（相同子图）：同一个子图实例可以在单个节点内被多次调用。使用有状态持久化时，这些调用会写入相同的检查点命名空间并产生冲突——应改用每次调用的持久化。
状态检查：子图的状态可通过以下方式获取 get_state(config, subgraphs=True) 用于调试和监控。

查看子图状态

当您启用持久化时，您可以使用 subgraphs 选项检查子图状态。使用无状态检查点（checkpointer=False）时，不会保存子图检查点，因此子图状态不可用。

Per-invocation

返回 **当前调用**的子图状态。每次调用都重新开始。

  from langgraph.graph import START, StateGraph
  from langgraph.checkpoint.memory import MemorySaver
  from langgraph.types import interrupt, Command
  from typing_extensions import TypedDict

  class State(TypedDict):
      foo: str

  # Subgraph
  def subgraph_node_1(state: State):
      value = interrupt("Provide value:")
      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()  # inherits parent checkpointer

  # Parent graph
  builder = StateGraph(State)
  builder.add_node("node_1", subgraph)
  builder.add_edge(START, "node_1")

  checkpointer = MemorySaver()
  graph = builder.compile(checkpointer=checkpointer)

  config = {"configurable": {"thread_id": "1"}}

  graph.invoke({"foo": ""}, config)

  # View subgraph state for the current invocation
  subgraph_state = graph.get_state(config, subgraphs=True).tasks[0].state  # [!code highlight]

  # Resume the subgraph
  graph.invoke(Command(resume="bar"), config)

Per-thread

返回 **累积的** 此线程上所有调用的子图状态。

  from langgraph.graph import START, StateGraph, MessagesState
  from langgraph.checkpoint.memory import MemorySaver

  # Subgraph with its own persistent state
  subgraph_builder = StateGraph(MessagesState)
  # ... add nodes and edges
  subgraph = subgraph_builder.compile(checkpointer=True)  # [!code highlight]

  # Parent graph
  builder = StateGraph(MessagesState)
  builder.add_node("agent", subgraph)
  builder.add_edge(START, "agent")

  checkpointer = MemorySaver()
  graph = builder.compile(checkpointer=checkpointer)

  config = {"configurable": {"thread_id": "1"}}

  graph.invoke({"messages": [{"role": "user", "content": "hi"}]}, config)
  graph.invoke({"messages": [{"role": "user", "content": "what did I say?"}]}, config)

  # View accumulated subgraph state (includes messages from both invocations)
  subgraph_state = graph.get_state(config, subgraphs=True).tasks[0].state  # [!code highlight]

流式输出子图

要观察嵌套图的执行，我们推荐使用事件流式传输： stream.subgraphs 投影会发现每个嵌套运行并暴露其 path, messages和 values ，而无需解析命名空间字符串。

stream = graph.stream_events({"foo": "foo"}, version="v3")  # [!code highlight]

for subgraph in stream.subgraphs:
    print(subgraph.graph_name, subgraph.path)

    for snapshot in subgraph.values:
        print(subgraph.path, snapshot)

如果你需要原始协议事件，直接迭代流并过滤 event["method"] 和 event["params"]["namespace"]:

stream = graph.stream_events({"foo": "foo"}, version="v3")
for event in stream:
    if event["method"] == "updates":
        print(event["params"]["namespace"], event["params"]["data"])

Stream from subgraphs

  from typing_extensions import TypedDict
  from langgraph.graph.state import StateGraph, START

  # Define subgraph
  class SubgraphState(TypedDict):
      foo: str
      bar: str

  def subgraph_node_1(state: SubgraphState):
      return {"bar": "bar"}

  def subgraph_node_2(state: SubgraphState):
      # note that this node is using a state key ('bar') that is only available in the subgraph
      # and is sending update on the shared state key ('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()

  # Define parent graph
  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()

  stream = graph.stream_events({"foo": "foo"}, version="v3")  # [!code highlight]
  for event in stream:
      if event["method"] == "updates":
          print(event["params"]["namespace"], event["params"]["data"])

  [] {'node_1': {'foo': 'hi! foo'}}
  ['node_2:e58e5673-a661-ebb0-70d4-e298a7fc28b7'] {'subgraph_node_1': {'bar': 'bar'}}
  ['node_2:e58e5673-a661-ebb0-70d4-e298a7fc28b7'] {'subgraph_node_2': {'foo': 'hi! foobar'}}
  [] {'node_2': {'foo': 'hi! foobar'}}