使用子图¶

本指南介绍如何使用subgraphs（子图）的机制。子图的一个常见应用是构建多智能体系统。

在添加子图时，你需要定义父图与子图的通信方式：

共享状态架构 — 父图与子图在其状态架构中拥有共享的状态键
不同的状态架构 — 父图与子图的架构之间没有任何共享的状态键

设置¶

pip install -U langgraph

为 LangGraph 开发设置 LangSmith

注册 LangSmith，即可快速发现问题并提升 LangGraph 项目的性能。LangSmith 允许你使用追踪数据来调试、测试和监控基于 LangGraph 构建的 LLM 应用——点击这里了解如何开始。

共享状态架构¶

常见场景是：父图与子图通过状态架构中的某个共享状态键（通道）通信。例如，在多智能体系统中，各个智能体通常通过共享的messages键进行通信。

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

定义子图工作流（下方示例中的 subgraph_builder）并进行编译
在定义父图工作流时，将已编译的子图传给 .add_node 方法

^{API Reference: StateGraph}

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

完整示例：共享状态架构

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

# 定义子图
class SubgraphState(TypedDict):
    foo: str  # (1)! 
    bar: str  # (2)!

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)

此键与父图的状态共享
此键是 SubgraphState 的私有键，对父图不可见

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

不同的状态架构¶

对于更复杂的系统，你可能希望定义与父图完全不同架构的子图（无任何共享键）。例如，在多智能体系统中，你或许希望为每个智能体保留各自私有的消息历史。

如果你的应用属于这种情况，你需要定义一个节点函数来调用子图。该函数在调用子图之前需要把输入（父图）状态转换为子图状态，在返回节点的状态更新前再把结果转换回父图状态。

^{API Reference: StateGraph}

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_output = subgraph.invoke({"bar": state["foo"]})  # (1)!
    return {"foo": subgraph_output["bar"]}  # (2)!

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

将状态转换为子图状态
将返回结果转换回父图状态

完整示例：不同的状态架构

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):
    response = subgraph.invoke({"bar": state["foo"]})  # (1)!
    return {"foo": response["bar"]}  # (2)!


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"}, subgraphs=True):
    print(chunk)

将状态转换为子图状态
将返回结果转换回父图状态

((), {'node_1': {'foo': 'hi! foo'}})
(('node_2:9c36dd0f-151a-cb42-cbad-fa2f851f9ab7',), {'grandchild_1': {'my_grandchild_key': 'hi Bob, how are you'}})
(('node_2:9c36dd0f-151a-cb42-cbad-fa2f851f9ab7',), {'grandchild_2': {'bar': 'hi! foobaz'}})
((), {'node_2': {'foo': 'hi! foobaz'}})

完整示例：不同的状态架构（两层子图）

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

# 孙图
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:
    # 注意：在此处无法访问子图或父图的键
    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) -> ChildState:
    # 注意：在此处无法访问父图或孙图的键
    grandchild_graph_input = {"my_grandchild_key": state["my_child_key"]}  # (1)!
    grandchild_graph_output = grandchild_graph.invoke(grandchild_graph_input)
    return {"my_child_key": grandchild_graph_output["my_grandchild_key"] + " today?"}  # (2)!

child = StateGraph(ChildState)
child.add_node("child_1", call_grandchild_graph)  # (3)!
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) -> ParentState:
    # 注意：在此处无法访问子图或孙图的键
    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"]}  # (4)!
    child_graph_output = child_graph.invoke(child_graph_input)
    return {"my_key": child_graph_output["my_child_key"]}  # (5)!

parent = StateGraph(ParentState)
parent.add_node("parent_1", parent_1)
parent.add_node("child", call_child_graph)  # (6)!
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"}, subgraphs=True):
    print(chunk)

将状态从子图的状态通道（my_child_key）转换为孙图的状态通道（my_grandchild_key）
将状态从孙图的状态通道（my_grandchild_key）转换回子图的状态通道（my_child_key）
这里传入的是一个函数，而不是已编译的图（grandchild_graph）
将状态从父图的状态通道（my_key）转换为子图的状态通道（my_child_key）
将状态从子图的状态通道（my_child_key）转换回父图的状态通道（my_key）
这里传入的是一个函数，而不是已编译的图（child_graph）

((), {'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!'}})

添加持久化¶

你只需要在编译父图时提供 checkpointer。LangGraph 会自动将该 checkpointer 传播到子图。

^{API Reference: START | StateGraph | MemorySaver}

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

class State(TypedDict):
    foo: str

# 子图

def subgraph_node_1(state: State):
    return {"foo": state["foo"] + "bar"}

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 = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

如果你希望子图拥有它自己的记忆，可以在编译子图时传入相应的 checkpointer 选项。这在多智能体系统中很有用，例如你希望各个智能体维护各自的内部消息历史：

subgraph_builder = StateGraph(...)
subgraph = subgraph_builder.compile(checkpointer=True)

查看子图状态¶

当你启用持久化后，可以通过相应的方法检查图状态（检查点）。若要查看子图状态，你可以使用 subgraphs 选项。

你可以通过 graph.get_state(config) 检查图状态。若要查看子图状态，可以使用 graph.get_state(config, subgraphs=True)。

仅在被中断时可用

子图状态只能在子图被中断时查看。一旦你恢复图运行，就无法再访问子图状态。

查看被中断的子图状态

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

# 子图

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

# 父图

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)
parent_state = graph.get_state(config)
subgraph_state = graph.get_state(config, subgraphs=True).tasks[0].state  # (1)!

# 恢复子图
graph.invoke(Command(resume="bar"), config)

只有当子图被中断时才可用。一旦恢复图运行，就无法再访问子图状态。

流式传输子图输出¶

若要在流式输出中包含子图的输出，可以在父图的 stream 方法中设置 subgraphs 选项。这样会同时流式输出父图和任意子图的结果。

for chunk in graph.stream(
    {"foo": "foo"},
    subgraphs=True, # (1)!
    stream_mode="updates",
):
    print(chunk)

设置 subgraphs=True，以流式输出子图结果。

从子图进行流式输出

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"},
    stream_mode="updates",
    subgraphs=True, # (1)!
):
    print(chunk)

设置 subgraphs=True，以流式输出子图结果。

((), {'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'}})