主题
子图
子图在以下方面很有用:
- 构建多智能体系统
- 在多个图中复用一组节点
- 分布式开发:当你希望不同团队独立处理图的不同部分时,你可以将每一部分定义为一个子图,只要遵守子图接口(输入和输出模式),父图就可以在不知道子图任何细节的情况下构建
设置
bash
pip install -U langgraphbash
uv add langgraphTIP
为 LangGraph 开发设置 LangSmith 注册 LangSmith,以快速发现问题并提升 LangGraph 项目的性能。LangSmith 让你可以使用追踪数据来调试、测试和监控用 LangGraph 构建的 LLM 应用——阅读更多关于如何开始使用 LangSmith的内容。
定义子图通信
添加子图时,你需要定义父图和子图如何进行通信:
| 模式 | 何时使用 | 状态模式 |
|---|---|---|
| 在节点内调用子图 | 父图和子图具有不同的状态模式(没有共享键),或者你需要在其间转换状态 | 你编写一个包装函数,将父图状态映射到子图输入,并将子图输出映射回父图状态 |
| 将子图添加为节点 | 父图和子图共享状态键——子图读写与父图相同的通道 | 你将编译后的子图直接传递给 add_node——无需包装函数 |
在节点内调用子图
当父图和子图具有不同的状态模式(没有共享键)时,在节点函数内部调用子图。当你希望为多智能体系统中的每个智能体保留私有的消息历史时,这很常见。
节点函数在调用子图之前将父图状态转换为子图状态,并在返回之前将结果转换回父图状态。
python
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"]})
# 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()完整示例:不同的状态模式
python
from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START
from langgraph.stream import UpdatesTransformer
# 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", transformers=[UpdatesTransformer])
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'}}完整示例:不同的状态模式(两层子图)
这是一个包含两层子图的示例:parent -> child -> grandchild。
python
# Grandchild graph
from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START, END
from langgraph.stream import UpdatesTransformer
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", transformers=[UpdatesTransformer])
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!'}}将子图添加为节点
当父图和子图共享状态键时,你可以将编译后的子图直接传递给 add_node。无需包装函数——子图会自动读写父图的状态通道。例如,在多智能体系统中,各智能体通常通过共享的 messages 键进行通信。

如果你的子图与父图共享状态键,你可以按照以下步骤将其添加到你的图中:
- 定义子图工作流(下面示例中的
subgraph_builder)并编译它 - 定义父图工作流时,将编译后的子图传递给
add_node方法
python
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)
builder.add_edge(START, "node_1")
graph = builder.compile()完整示例:共享状态模式
python
from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START
from langgraph.stream import UpdatesTransformer
# 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", transformers=[UpdatesTransformer])
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'}}子图持久化
当你使用子图时,你需要决定其内部数据在多次调用之间如何处理。考虑一个将任务委派给专业子智能体的客户支持机器人:“账单专家”子智能体应该记住客户之前的问题,还是每次被调用时都从头开始?
.compile() 上的 checkpointer 参数控制子图持久化:
| 模式 | checkpointer= | 行为 |
|---|---|---|
| 每次调用 | None(默认) | 每次调用都从头开始,并继承父图的检查点器,以在单次调用内支持中断和持久化执行。 |
| 按线程 | True | 状态在同一线程上的多次调用之间累积。每次调用都从上次离开的地方继续。 |
| 无状态 | False | 完全不做检查点保存——像普通函数调用一样运行。没有中断或持久化执行。 |
对于大多数应用来说,每次调用是合适的选择,包括子智能体处理独立请求的多智能体系统。当子智能体需要多轮对话记忆时(例如,一个在多次交流中逐步构建上下文的研究助手),请使用按线程。
NOTE
父图必须使用检查点器编译,子图持久化功能(中断、状态检查、按线程记忆)才能工作。请参阅持久化。
INFO
下面的示例使用 LangChain 的 create_agent,这是构建智能体的一种常见方式。create_agent 在底层会生成一个 LangGraph 图,因此所有子图持久化概念都直接适用。如果你使用原生 LangGraph StateGraph 构建,相同的模式和配置选项同样适用——详见 Graph API。
有状态
有状态子图继承父图的检查点器,从而支持中断、持久化和状态检查。这两种有状态模式的区别在于状态保留的时长不同。
每次调用(默认)
当对子图的每次调用都是独立的,且子智能体不需要记住之前调用的任何内容时,请使用每次调用持久化。这是最常见的模式,特别是对于子智能体处理诸如“查找此客户的订单”或“总结此文档”这类一次性请求的多智能体系统。
省略 checkpointer 或将其设置为 None。每次调用都从头开始,但在单次调用内,子图会继承父图的检查点器,并可以使用 interrupt() 来暂停和恢复。
下面的示例使用了两个子智能体(水果专家、蔬菜专家),它们被包装为外部智能体的工具:
python
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(),
)每次调用都可以使用 interrupt() 来暂停和恢复。在工具函数中添加 interrupt(),以在继续之前要求用户批准:
python
@tool
def fruit_info(fruit_name: str) -> str:
"""Look up fruit info."""
interrupt("continue?")
return f"Info about {fruit_name}"python
from langgraph.types import Command
config = {"configurable": {"thread_id": "1"}}
# Stream events - the subagent's tool calls interrupt()
stream = agent.stream_events(
{"messages": [{"role": "user", "content": "Tell me about apples"}]},
config=config,
version="v3",
)
output = stream.output # drive the stream to completion
# stream.interrupts contains pending interrupts (and stream.interrupted is True)
# Resume - approve the interrupt
resumed = agent.stream_events(Command(resume=True), config=config, version="v3")
final = resumed.output查看示例追踪
打开此示例的公开 LangSmith 运行记录。
按线程
当子智能体需要记住之前的交互时,请使用按线程持久化。例如,一个在多次交流中逐步构建上下文的研究助手,或一个跟踪已编辑过哪些文件的编码助手。子智能体的对话历史和数据会在同一线程上的多次调用之间累积。每次调用都从上次离开的地方继续。
使用 checkpointer=True 编译以启用此行为。
WARNING
按线程子图不支持并行工具调用。当 LLM 可以访问作为工具的按线程子智能体时,它可能会尝试并行多次调用该工具(例如,同时询问水果专家关于苹果和香蕉的信息)。这会导致检查点冲突,因为两次调用写入同一个命名空间。
下面的示例使用 LangChain 的 ToolCallLimitMiddleware 来防止这种情况。如果你使用纯 LangGraph StateGraph 构建,则需要自行防止并行工具调用——例如,通过配置模型禁用并行工具调用,或添加逻辑以确保同一个子图不会被并行调用多次。
下面的示例使用一个用 checkpointer=True 编译的水果专家子智能体:
python
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,
)
# 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=[
ToolCallLimitMiddleware(tool_name="ask_fruit_expert", run_limit=1),
],
checkpointer=MemorySaver(),
)按线程创建的子智能体与按调用创建的子智能体一样支持 interrupt()。在工具函数中添加 interrupt() 即可要求用户批准:
python
@tool
def fruit_info(fruit_name: str) -> str:
"""Look up fruit info."""
interrupt("continue?")
return f"Info about {fruit_name}"python
from langgraph.types import Command
config = {"configurable": {"thread_id": "1"}}
# Stream events - the subagent's tool calls interrupt()
stream = agent.stream_events(
{"messages": [{"role": "user", "content": "Tell me about apples"}]},
config=config,
version="v3",
)
output = stream.output # drive the stream to completion
# stream.interrupts contains pending interrupts (and stream.interrupted is True)
# Resume - approve the interrupt
resumed = agent.stream_events(Command(resume=True), config=config, version="v3")
final = resumed.output查看示例追踪
打开此示例的公开 LangSmith 运行记录。
无状态
当你希望像普通函数调用一样运行子智能体、且没有检查点开销时,请使用此方式。子图无法暂停/恢复,也无法从持久化执行中获益。使用 checkpointer=False 进行编译。
WARNING
如果没有检查点,子图就没有持久化执行。如果进程在运行中途崩溃,子图将无法恢复,必须从头重新运行。
python
subgraph_builder = StateGraph(...)
subgraph = subgraph_builder.compile(checkpointer=False) 检查点器参考
通过 .compile() 上的 checkpointer 参数控制子图持久化:
python
subgraph = builder.compile(checkpointer=False) # or True / None| 功能 | 按调用(默认) | 按线程 | 无状态 |
|---|---|---|---|
checkpointer= | None | True | False |
| 中断(HITL) | ✅ | ✅ | ❌ |
| 多轮记忆 | ❌ | ✅ | ❌ |
| 多次调用(不同的子图) | ✅ |
⚠️
| ✅ | | 多次调用(同一子图) | ✅ | ❌ | ✅ | | 状态检查 |
⚠️
| ✅ | ❌ |
- 中断(HITL):子图可以使用 interrupt() 暂停执行并等待用户输入,然后从暂停处恢复。
- 多轮记忆:子图在同一线程内的多次调用之间保留其状态。每次调用都会从上次中断处继续,而不是重新开始。
- 多次调用(不同的子图):可以在单个节点内调用多个不同的子图实例,而不会产生检查点命名空间冲突。
- 多次调用(同一子图):同一个子图实例可以在单个节点内被多次调用。在有状态持久化下,这些调用会写入同一个检查点命名空间并产生冲突——请改用按调用的持久化。
- 状态检查:可通过
get_state(config, subgraphs=True)获取子图的状态,用于调试和监控。
查看子图状态
当你启用持久化时,可以使用 subgraphs 选项检查子图状态。在无状态检查点(checkpointer=False)下,不会保存任何子图检查点,因此子图状态不可用。
NOTE
查看子图状态要求 LangGraph 能够 静态发现 该子图——也就是说,它被作为节点添加,或者在节点内被调用。当子图在工具函数或其他间接调用中(例如子智能体模式)被调用时,此功能不适用。无论嵌套层级如何,中断仍会传播到顶层图。
仅返回 当前调用 的子图状态。每次调用都从头开始。
python
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
# Resume the subgraph
graph.invoke(Command(resume="bar"), config)流式输出子图输出
要观察嵌套图的执行情况,我们推荐使用事件流式传输:stream.subgraphs 投影会发现每个嵌套运行,并公开其 path、messages 和 values,无需解析命名空间字符串。
python
stream = graph.stream_events({"foo": "foo"}, version="v3")
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"] 过滤:
python
from langgraph.stream import UpdatesTransformer
stream = graph.stream_events({"foo": "foo"}, version="v3", transformers=[UpdatesTransformer])
for event in stream:
if event["method"] == "updates":
print(event["params"]["namespace"], event["params"]["data"])从子图流式输出
python
from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START
from langgraph.stream import UpdatesTransformer
# 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", transformers=[UpdatesTransformer])
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'}}