Skip to content

如何为子图添加线程级持久性

前置条件

本指南假设您熟悉以下内容：

本指南展示了如何为使用子图的图添加线程级持久化。

设置环境

首先，让我们安装所需的包

pip install -U langgraph

为LangGraph开发设置LangSmith

注册LangSmith可以快速发现并解决您的LangGraph项目中的问题，并提高性能。LangSmith允许您使用跟踪数据来调试、测试和监控使用LangGraph构建的LLM应用程序——更多关于如何开始的信息，请参阅这里。

使用持久性定义图

要为包含子图的图形添加持久性，您所需要做的就是在**编译父图时**传递一个检查点器。LangGraph会自动将检查点器传播到子图中。

提示

在编译子图时，你不应该提供检查点器（checkpointer）。相反，你需要定义一个**单一**的检查点器，并将其传递给parent_graph.compile()，LangGraph会自动将该检查点器传播到子图中。如果你将检查点器传递给subgraph.compile()，它将被忽略。这同样适用于你在添加一个调用子图的节点函数时的情况。

让我们定义一个简单的图，其中包含一个子图节点，以展示如何做到这一点。

API Reference: START | StateGraph | MemorySaver

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


# subgraph


class SubgraphState(TypedDict):
    foo: str  # note that this key is shared with the parent graph state
    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()


# parent graph


class State(TypedDict):
    foo: str


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


builder = StateGraph(State)
builder.add_node("node_1", node_1)
# note that we're adding the compiled subgraph as a node to the parent graph
builder.add_node("node_2", subgraph)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")

<langgraph.graph.state.StateGraph at 0x106d2fa10>

我们现在可以使用内存检查点保存器（MemorySaver）来编译图并进行内存检查。

checkpointer = MemorySaver()
# You must only pass checkpointer when compiling the parent graph.
# LangGraph will automatically propagate the checkpointer to the child subgraphs.
graph = builder.compile(checkpointer=checkpointer)

验证持久化功能是否正常工作

现在让我们运行这个图，并检查父图和子图的持久化状态，以验证持久化功能是否正常工作。我们应该期望在state.values中看到父图和子图的最终执行结果。

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


for _, chunk in graph.stream({"foo": "foo"}, config, subgraphs=True):
    print(chunk)

{'node_1': {'foo': 'hi! foo'}}
{'subgraph_node_1': {'bar': 'bar'}}
{'subgraph_node_2': {'foo': 'hi! foobar'}}
{'node_2': {'foo': 'hi! foobar'}}
我们现在可以通过调用graph.get_state()并使用与调用图相同的配置来查看父图的状态。

graph.get_state(config).values

{'foo': 'hi! foobar'}

要查看子图的状态，我们需要做两件事：

  1. 找到子图最新的配置值
  2. 使用 graph.get_state() 获取该子图最新配置的状态值。

为了找到正确的配置，我们可以从父图的状态历史中查找在从 node_2（带有子图的节点）返回结果之前的状态快照：

state_with_subgraph = [
    s for s in graph.get_state_history(config) if s.next == ("node_2",)
][0]

状态快照将包括下一个要执行的任务列表tasks。在使用子图时，tasks将包含我们用来检索子图状态的配置：

subgraph_config = state_with_subgraph.tasks[0].state
subgraph_config

{'configurable': {'thread_id': '1',
  'checkpoint_ns': 'node_2:6ef111a6-f290-7376-0dfc-a4152307bc5b'}}

graph.get_state(subgraph_config).values

{'foo': 'hi! foobar', 'bar': 'bar'}

若想了解更多关于如何修改人机交互工作流中的子图状态的信息，请参阅此操作指南。

Comments