使用子图 - LangChain 框架 --知识铺

使用子图¶

本指南解释了使用子图的机制。子图的一个常见应用是构建多代理系统。

添加子图时，需要定义父图和子图如何通信

• 共享状态模式 — 父图和子图在其状态模式中具有共享状态键
• 不同状态模式 — 父图和子图的模式中没有共享状态键

设置¶

为 LangGraph 开发设置 LangSmith

注册 LangSmith，快速发现问题并提高 LangGraph 项目的性能。LangSmith 允许您使用跟踪数据来调试、测试和监控您使用 LangGraph 构建的 LLM 应用——在此处了解更多入门信息。

常见情况是父图和子图通过模式中的共享状态键（通道）进行通信。例如，在多代理系统中，代理通常通过共享的消息键进行通信。

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

1. 定义子图工作流（在下面的示例中为subgraph_builder）并编译它
2. 在定义父图工作流时，将已编译的子图传递给.add_node方法

API 参考：StateGraph

<span id="__span-1-1">from typing_extensions import TypedDict
<span id="__span-1-2">from langgraph.graph.state import StateGraph, START
<span id="__span-1-3">
<span id="__span-1-4">class State(TypedDict):
<span id="__span-1-5">    foo: str
<span id="__span-1-6">
<span id="__span-1-7"># Subgraph
<span id="__span-1-8">
<span id="__span-1-9">def subgraph_node_1(state: State):
<span id="__span-1-10">    return {"foo": "hi! " + state["foo"]}
<span id="__span-1-11">
<span id="__span-1-12">subgraph_builder = StateGraph(State)
<span id="__span-1-13">subgraph_builder.add_node(subgraph_node_1)
<span id="__span-1-14">subgraph_builder.add_edge(START, "subgraph_node_1")
<span id="__span-1-15">subgraph = subgraph_builder.compile()
<span id="__span-1-16">
<span id="__span-1-17"># Parent graph
<span id="__span-1-18">
<span id="__span-1-19">builder = StateGraph(State)
<span id="__span-1-20">builder.add_node("node_1", subgraph)
<span id="__span-1-21">builder.add_edge(START, "node_1")
<span id="__span-1-22">graph = builder.compile()

完整示例：共享状态模式

<span id="__span-2-1">from typing_extensions import TypedDict
<span id="__span-2-2">from langgraph.graph.state import StateGraph, START
<span id="__span-2-3">
<span id="__span-2-4"># Define subgraph
<span id="__span-2-5">class SubgraphState(TypedDict):
<span id="__span-2-6">    foo: str  
<span id="__span-2-7">    bar: str  
<span id="__span-2-8">
<span id="__span-2-9">def subgraph_node_1(state: SubgraphState):
<span id="__span-2-10">    return {"bar": "bar"}
<span id="__span-2-11">
<span id="__span-2-12">def subgraph_node_2(state: SubgraphState):
<span id="__span-2-13">    # note that this node is using a state key ('bar') that is only available in the subgraph
<span id="__span-2-14">    # and is sending update on the shared state key ('foo')
<span id="__span-2-15">    return {"foo": state["foo"] + state["bar"]}
<span id="__span-2-16">
<span id="__span-2-17">subgraph_builder = StateGraph(SubgraphState)
<span id="__span-2-18">subgraph_builder.add_node(subgraph_node_1)
<span id="__span-2-19">subgraph_builder.add_node(subgraph_node_2)
<span id="__span-2-20">subgraph_builder.add_edge(START, "subgraph_node_1")
<span id="__span-2-21">subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
<span id="__span-2-22">subgraph = subgraph_builder.compile()
<span id="__span-2-23">
<span id="__span-2-24"># Define parent graph
<span id="__span-2-25">class ParentState(TypedDict):
<span id="__span-2-26">    foo: str
<span id="__span-2-27">
<span id="__span-2-28">def node_1(state: ParentState):
<span id="__span-2-29">    return {"foo": "hi! " + state["foo"]}
<span id="__span-2-30">
<span id="__span-2-31">builder = StateGraph(ParentState)
<span id="__span-2-32">builder.add_node("node_1", node_1)
<span id="__span-2-33">builder.add_node("node_2", subgraph)
<span id="__span-2-34">builder.add_edge(START, "node_1")
<span id="__span-2-35">builder.add_edge("node_1", "node_2")
<span id="__span-2-36">graph = builder.compile()
<span id="__span-2-37">
<span id="__span-2-38">for chunk in graph.stream({"foo": "foo"}):
<span id="__span-2-39">    print(chunk)

<span id="__span-3-1">{'node_1': {'foo': 'hi! foo'}}
<span id="__span-3-2">{'node_2': {'foo': 'hi! foobar'}}

```

不同状态模式¶

对于更复杂的系统，您可能希望定义与父图完全不同模式（没有共享键）的子图。例如，您可能希望为多代理系统中的每个代理保留私有消息历史。

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

API 参考：StateGraph

<span id="__span-4-1">from typing_extensions import TypedDict
<span id="__span-4-2">from langgraph.graph.state import StateGraph, START
<span id="__span-4-3">
<span id="__span-4-4">class SubgraphState(TypedDict):
<span id="__span-4-5">    bar: str
<span id="__span-4-6">
<span id="__span-4-7"># Subgraph
<span id="__span-4-8">
<span id="__span-4-9">def subgraph_node_1(state: SubgraphState):
<span id="__span-4-10">    return {"bar": "hi! " + state["bar"]}
<span id="__span-4-11">
<span id="__span-4-12">subgraph_builder = StateGraph(SubgraphState)
<span id="__span-4-13">subgraph_builder.add_node(subgraph_node_1)
<span id="__span-4-14">subgraph_builder.add_edge(START, "subgraph_node_1")
<span id="__span-4-15">subgraph = subgraph_builder.compile()
<span id="__span-4-16">
<span id="__span-4-17"># Parent graph
<span id="__span-4-18">
<span id="__span-4-19">class State(TypedDict):
<span id="__span-4-20">    foo: str
<span id="__span-4-21">
<span id="__span-4-22">def call_subgraph(state: State):
<span id="__span-4-23">    subgraph_output = subgraph.invoke({"bar": state["foo"]})  
<span id="__span-4-24">    return {"foo": subgraph_output["bar"]}  
<span id="__span-4-25">
<span id="__span-4-26">builder = StateGraph(State)
<span id="__span-4-27">builder.add_node("node_1", call_subgraph)
<span id="__span-4-28">builder.add_edge(START, "node_1")
<span id="__span-4-29">graph = builder.compile()

完整示例：不同状态模式

<span id="__span-5-1">from typing_extensions import TypedDict
<span id="__span-5-2">from langgraph.graph.state import StateGraph, START
<span id="__span-5-3">
<span id="__span-5-4"># Define subgraph
<span id="__span-5-5">class SubgraphState(TypedDict):
<span id="__span-5-6">    # note that none of these keys are shared with the parent graph state
<span id="__span-5-7">    bar: str
<span id="__span-5-8">    baz: str
<span id="__span-5-9">
<span id="__span-5-10">def subgraph_node_1(state: SubgraphState):
<span id="__span-5-11">    return {"baz": "baz"}
<span id="__span-5-12">
<span id="__span-5-13">def subgraph_node_2(state: SubgraphState):
<span id="__span-5-14">    return {"bar": state["bar"] + state["baz"]}
<span id="__span-5-15">
<span id="__span-5-16">subgraph_builder = StateGraph(SubgraphState)
<span id="__span-5-17">subgraph_builder.add_node(subgraph_node_1)
<span id="__span-5-18">subgraph_builder.add_node(subgraph_node_2)
<span id="__span-5-19">subgraph_builder.add_edge(START, "subgraph_node_1")
<span id="__span-5-20">subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
<span id="__span-5-21">subgraph = subgraph_builder.compile()
<span id="__span-5-22">
<span id="__span-5-23"># Define parent graph
<span id="__span-5-24">class ParentState(TypedDict):
<span id="__span-5-25">    foo: str
<span id="__span-5-26">
<span id="__span-5-27">def node_1(state: ParentState):
<span id="__span-5-28">    return {"foo": "hi! " + state["foo"]}
<span id="__span-5-29">
<span id="__span-5-30">def node_2(state: ParentState):
<span id="__span-5-31">    response = subgraph.invoke({"bar": state["foo"]})  
<span id="__span-5-32">    return {"foo": response["bar"]}  
<span id="__span-5-33">
<span id="__span-5-34">
<span id="__span-5-35">builder = StateGraph(ParentState)
<span id="__span-5-36">builder.add_node("node_1", node_1)
<span id="__span-5-37">builder.add_node("node_2", node_2)
<span id="__span-5-38">builder.add_edge(START, "node_1")
<span id="__span-5-39">builder.add_edge("node_1", "node_2")
<span id="__span-5-40">graph = builder.compile()
<span id="__span-5-41">
<span id="__span-5-42">for chunk in graph.stream({"foo": "foo"}, subgraphs=True):
<span id="__span-5-43">    print(chunk)

<span id="__span-6-1">((), {'node_1': {'foo': 'hi! foo'}})
<span id="__span-6-2">(('node_2:9c36dd0f-151a-cb42-cbad-fa2f851f9ab7',), {'subgraph_node_1': {'baz': 'baz'}})
<span id="__span-6-3">(('node_2:9c36dd0f-151a-cb42-cbad-fa2f851f9ab7',), {'subgraph_node_2': {'bar': 'hi! foobaz'}})
<span id="__span-6-4">((), {'node_2': {'foo': 'hi! foobaz'}})

完整示例：不同状态模式（两级子图）

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

<span id="__span-7-1"># Grandchild graph
<span id="__span-7-2">from typing_extensions import TypedDict
<span id="__span-7-3">from langgraph.graph.state import StateGraph, START, END
<span id="__span-7-4">
<span id="__span-7-5">class GrandChildState(TypedDict):
<span id="__span-7-6">    my_grandchild_key: str
<span id="__span-7-7">
<span id="__span-7-8">def grandchild_1(state: GrandChildState) -&gt; GrandChildState:
<span id="__span-7-9">    # NOTE: child or parent keys will not be accessible here
<span id="__span-7-10">    return {"my_grandchild_key": state["my_grandchild_key"] + ", how are you"}
<span id="__span-7-11">
<span id="__span-7-12">
<span id="__span-7-13">grandchild = StateGraph(GrandChildState)
<span id="__span-7-14">grandchild.add_node("grandchild_1", grandchild_1)
<span id="__span-7-15">
<span id="__span-7-16">grandchild.add_edge(START, "grandchild_1")
<span id="__span-7-17">grandchild.add_edge("grandchild_1", END)
<span id="__span-7-18">
<span id="__span-7-19">grandchild_graph = grandchild.compile()
<span id="__span-7-20">
<span id="__span-7-21"># Child graph
<span id="__span-7-22">class ChildState(TypedDict):
<span id="__span-7-23">    my_child_key: str
<span id="__span-7-24">
<span id="__span-7-25">def call_grandchild_graph(state: ChildState) -&gt; ChildState:
<span id="__span-7-26">    # NOTE: parent or grandchild keys won't be accessible here
<span id="__span-7-27">    grandchild_graph_input = {"my_grandchild_key": state["my_child_key"]}  
<span id="__span-7-28">    grandchild_graph_output = grandchild_graph.invoke(grandchild_graph_input)
<span id="__span-7-29">    return {"my_child_key": grandchild_graph_output["my_grandchild_key"] + " today?"}  
<span id="__span-7-30">
<span id="__span-7-31">child = StateGraph(ChildState)
<span id="__span-7-32">child.add_node("child_1", call_grandchild_graph)  
<span id="__span-7-33">child.add_edge(START, "child_1")
<span id="__span-7-34">child.add_edge("child_1", END)
<span id="__span-7-35">child_graph = child.compile()
<span id="__span-7-36">
<span id="__span-7-37"># Parent graph
<span id="__span-7-38">class ParentState(TypedDict):
<span id="__span-7-39">    my_key: str
<span id="__span-7-40">
<span id="__span-7-41">def parent_1(state: ParentState) -&gt; ParentState:
<span id="__span-7-42">    # NOTE: child or grandchild keys won't be accessible here
<span id="__span-7-43">    return {"my_key": "hi " + state["my_key"]}
<span id="__span-7-44">
<span id="__span-7-45">def parent_2(state: ParentState) -&gt; ParentState:
<span id="__span-7-46">    return {"my_key": state["my_key"] + " bye!"}
<span id="__span-7-47">
<span id="__span-7-48">def call_child_graph(state: ParentState) -&gt; ParentState:
<span id="__span-7-49">    child_graph_input = {"my_child_key": state["my_key"]}  
<span id="__span-7-50">    child_graph_output = child_graph.invoke(child_graph_input)
<span id="__span-7-51">    return {"my_key": child_graph_output["my_child_key"]}  
<span id="__span-7-52">
<span id="__span-7-53">parent = StateGraph(ParentState)
<span id="__span-7-54">parent.add_node("parent_1", parent_1)
<span id="__span-7-55">parent.add_node("child", call_child_graph)  
<span id="__span-7-56">parent.add_node("parent_2", parent_2)
<span id="__span-7-57">
<span id="__span-7-58">parent.add_edge(START, "parent_1")
<span id="__span-7-59">parent.add_edge("parent_1", "child")
<span id="__span-7-60">parent.add_edge("child", "parent_2")
<span id="__span-7-61">parent.add_edge("parent_2", END)
<span id="__span-7-62">
<span id="__span-7-63">parent_graph = parent.compile()
<span id="__span-7-64">
<span id="__span-7-65">for chunk in parent_graph.stream({"my_key": "Bob"}, subgraphs=True):
<span id="__span-7-66">    print(chunk)

<code tabindex="0"><span id="__span-8-1">((), {'parent_1': {'my_key': 'hi Bob'}})
<span id="__span-8-2">(('child:2e26e9ce-602f-862c-aa66-1ea5a4655e3b', 'child_1:781bb3b1-3971-84ce-810b-acf819a03f9c'), {'grandchild_1': {'my_grandchild_key': 'hi Bob, how are you'}})
<span id="__span-8-3">(('child:2e26e9ce-602f-862c-aa66-1ea5a4655e3b',), {'child_1': {'my_child_key': 'hi Bob, how are you today?'}})
<span id="__span-8-4">((), {'child': {'my_key': 'hi Bob, how are you today?'}})
<span id="__span-8-5">((), {'parent_2': {'my_key': 'hi Bob, how are you today? bye!'}})

添加持久化¶

您只需在编译父图时提供检查器。LangGraph 会自动将检查器传播到子子图。

API 参考：START | StateGraph | InMemorySaver

<span id="__span-9-1">from langgraph.graph import START, StateGraph
<span id="__span-9-2">from langgraph.checkpoint.memory import InMemorySaver
<span id="__span-9-3">from typing_extensions import TypedDict
<span id="__span-9-4">
<span id="__span-9-5">class State(TypedDict):
<span id="__span-9-6">    foo: str
<span id="__span-9-7">
<span id="__span-9-8"># Subgraph
<span id="__span-9-9">
<span id="__span-9-10">def subgraph_node_1(state: State):
<span id="__span-9-11">    return {"foo": state["foo"] + "bar"}
<span id="__span-9-12">
<span id="__span-9-13">subgraph_builder = StateGraph(State)
<span id="__span-9-14">subgraph_builder.add_node(subgraph_node_1)
<span id="__span-9-15">subgraph_builder.add_edge(START, "subgraph_node_1")
<span id="__span-9-16">subgraph = subgraph_builder.compile()
<span id="__span-9-17">
<span id="__span-9-18"># Parent graph
<span id="__span-9-19">
<span id="__span-9-20">builder = StateGraph(State)
<span id="__span-9-21">builder.add_node("node_1", subgraph)
<span id="__span-9-22">builder.add_edge(START, "node_1")
<span id="__span-9-23">
<span id="__span-9-24">checkpointer = InMemorySaver()
<span id="__span-9-25">graph = builder.compile(checkpointer=checkpointer)

如果您希望子图拥有自己的内存，您可以将其编译为with checkpointer=True。这在多代理系统中很有用，如果您希望代理跟踪其内部消息历史。

<span id="__span-10-1">subgraph_builder = StateGraph(...)
<span id="__span-10-2">subgraph = subgraph_builder.compile(checkpointer=True)

查看子图状态¶

当您启用持久化时，可以通过graph.get_state(config)检查图状态（检查点）。要查看子图状态，可以使用graph.get_state(config, subgraphs=True)。

仅在中断时可用

子图状态只能在子图中断时查看。一旦您恢复图，您将无法访问子图状态。

查看中断的子图状态

<span id="__span-11-1">from langgraph.graph import START, StateGraph
<span id="__span-11-2">from langgraph.checkpoint.memory import InMemorySaver
<span id="__span-11-3">from langgraph.types import interrupt, Command
<span id="__span-11-4">from typing_extensions import TypedDict
<span id="__span-11-5">
<span id="__span-11-6">class State(TypedDict):
<span id="__span-11-7">    foo: str
<span id="__span-11-8">
<span id="__span-11-9"># Subgraph
<span id="__span-11-10">
<span id="__span-11-11">def subgraph_node_1(state: State):
<span id="__span-11-12">    value = interrupt("Provide value:")
<span id="__span-11-13">    return {"foo": state["foo"] + value}
<span id="__span-11-14">
<span id="__span-11-15">subgraph_builder = StateGraph(State)
<span id="__span-11-16">subgraph_builder.add_node(subgraph_node_1)
<span id="__span-11-17">subgraph_builder.add_edge(START, "subgraph_node_1")
<span id="__span-11-18">
<span id="__span-11-19">subgraph = subgraph_builder.compile()
<span id="__span-11-20">
<span id="__span-11-21"># Parent graph
<span id="__span-11-22">
<span id="__span-11-23">builder = StateGraph(State)
<span id="__span-11-24">builder.add_node("node_1", subgraph)
<span id="__span-11-25">builder.add_edge(START, "node_1")
<span id="__span-11-26">
<span id="__span-11-27">checkpointer = InMemorySaver()
<span id="__span-11-28">graph = builder.compile(checkpointer=checkpointer)
<span id="__span-11-29">
<span id="__span-11-30">config = {"configurable": {"thread_id": "1"}}
<span id="__span-11-31">
<span id="__span-11-32">graph.invoke({"foo": ""}, config)
<span id="__span-11-33">parent_state = graph.get_state(config)
<span id="__span-11-34">subgraph_state = graph.get_state(config, subgraphs=True).tasks[0].state  
<span id="__span-11-35">
<span id="__span-11-36"># resume the subgraph
<span id="__span-11-37">graph.invoke(Command(resume="bar"), config)

流式传输子图输出¶

要在流式输出中包含来自子图的输出，您可以在父图的.stream()方法中设置subgraphs=True。这将同时流式传输父图和任何子图的输出。

<span id="__span-12-1">for chunk in graph.stream(
<span id="__span-12-2">    {"foo": "foo"},
<span id="__span-12-3">    subgraphs=True, 
<span id="__span-12-4">    stream_mode="updates",
<span id="__span-12-5">):
<span id="__span-12-6">    print(chunk)

从子图流式传输

<span id="__span-13-1">from typing_extensions import TypedDict
<span id="__span-13-2">from langgraph.graph.state import StateGraph, START
<span id="__span-13-3">
<span id="__span-13-4"># Define subgraph
<span id="__span-13-5">class SubgraphState(TypedDict):
<span id="__span-13-6">    foo: str
<span id="__span-13-7">    bar: str
<span id="__span-13-8">
<span id="__span-13-9">def subgraph_node_1(state: SubgraphState):
<span id="__span-13-10">    return {"bar": "bar"}
<span id="__span-13-11">
<span id="__span-13-12">def subgraph_node_2(state: SubgraphState):
<span id="__span-13-13">    # note that this node is using a state key ('bar') that is only available in the subgraph
<span id="__span-13-14">    # and is sending update on the shared state key ('foo')
<span id="__span-13-15">    return {"foo": state["foo"] + state["bar"]}
<span id="__span-13-16">
<span id="__span-13-17">subgraph_builder = StateGraph(SubgraphState)
<span id="__span-13-18">subgraph_builder.add_node(subgraph_node_1)
<span id="__span-13-19">subgraph_builder.add_node(subgraph_node_2)
<span id="__span-13-20">subgraph_builder.add_edge(START, "subgraph_node_1")
<span id="__span-13-21">subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
<span id="__span-13-22">subgraph = subgraph_builder.compile()
<span id="__span-13-23">
<span id="__span-13-24"># Define parent graph
<span id="__span-13-25">class ParentState(TypedDict):
<span id="__span-13-26">    foo: str
<span id="__span-13-27">
<span id="__span-13-28">def node_1(state: ParentState):
<span id="__span-13-29">    return {"foo": "hi! " + state["foo"]}
<span id="__span-13-30">
<span id="__span-13-31">builder = StateGraph(ParentState)
<span id="__span-13-32">builder.add_node("node_1", node_1)
<span id="__span-13-33">builder.add_node("node_2", subgraph)
<span id="__span-13-34">builder.add_edge(START, "node_1")
<span id="__span-13-35">builder.add_edge("node_1", "node_2")
<span id="__span-13-36">graph = builder.compile()
<span id="__span-13-37">
<span id="__span-13-38">for chunk in graph.stream(
<span id="__span-13-39">    {"foo": "foo"},
<span id="__span-13-40">    stream_mode="updates",
<span id="__span-13-41">    subgraphs=True, 
<span id="__span-13-42">):
<span id="__span-13-43">    print(chunk)

<span id="__span-14-1">((), {'node_1': {'foo': 'hi! foo'}})
<span id="__span-14-2">(('node_2:e58e5673-a661-ebb0-70d4-e298a7fc28b7',), {'subgraph_node_1': {'bar': 'bar'}})
<span id="__span-14-3">(('node_2:e58e5673-a661-ebb0-70d4-e298a7fc28b7',), {'subgraph_node_2': {'foo': 'hi! foobar'}})
<span id="__span-14-4">((), {'node_2': {'foo': 'hi! foobar'}})