如何在节点之间传递私有状态¶

先决条件

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

• 多个模式

在某些情况下，您可能希望节点交换对中间逻辑至关重要的信息，但这些信息不需要成为图的主要模式的一部分。此私有数据与图的总体输入/输出无关，并且应该只在某些节点之间共享。

在本操作指南中，我们将创建一个由三个节点（node_1、node_2 和 node_3）组成的示例顺序图，其中私有数据在第一步和第二步（node_1 和 node_2）之间传递，而第三步（node_3）只访问公共总体状态。

设置¶

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

在 [1]

%%capture --no-stderr
%pip install -U langgraph

%%capture --no-stderr %pip install -U langgraph

为 LangGraph 开发设置 LangSmith

注册 LangSmith 以快速发现问题并提高 LangGraph 项目的性能。LangSmith 允许您使用跟踪数据来调试、测试和监控使用 LangGraph 构建的 LLM 应用程序 - 了解有关如何开始使用的更多信息 此处。

定义和使用图¶

在 [1]

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


# The overall state of the graph (this is the public state shared across nodes)
class OverallState(TypedDict):
    a: str


# Output from node_1 contains private data that is not part of the overall state
class Node1Output(TypedDict):
    private_data: str


# The private data is only shared between node_1 and node_2
def node_1(state: OverallState) -> Node1Output:
    output = {"private_data": "set by node_1"}
    print(f"Entered node `node_1`:\n\tInput: {state}.\n\tReturned: {output}")
    return output


# Node 2 input only requests the private data available after node_1
class Node2Input(TypedDict):
    private_data: str


def node_2(state: Node2Input) -> OverallState:
    output = {"a": "set by node_2"}
    print(f"Entered node `node_2`:\n\tInput: {state}.\n\tReturned: {output}")
    return output


# Node 3 only has access to the overall state (no access to private data from node_1)
def node_3(state: OverallState) -> OverallState:
    output = {"a": "set by node_3"}
    print(f"Entered node `node_3`:\n\tInput: {state}.\n\tReturned: {output}")
    return output


# Build the state graph
builder = StateGraph(OverallState)
builder.add_node(node_1)  # node_1 is the first node
builder.add_node(
    node_2
)  # node_2 is the second node and accepts private data from node_1
builder.add_node(node_3)  # node_3 is the third node and does not see the private data
builder.add_edge(START, "node_1")  # Start the graph with node_1
builder.add_edge("node_1", "node_2")  # Pass from node_1 to node_2
builder.add_edge(
    "node_2", "node_3"
)  # Pass from node_2 to node_3 (only overall state is shared)
builder.add_edge("node_3", END)  # End the graph after node_3
graph = builder.compile()

# Invoke the graph with the initial state
response = graph.invoke(
    {
        "a": "set at start",
    }
)

print()
print(f"Output of graph invocation: {response}")

from langgraph.graph import StateGraph, START, END from typing_extensions import TypedDict # 图的总体状态（这是跨节点共享的公共状态） class OverallState(TypedDict): a: str # node_1 的输出包含不属于总体状态的私有数据 class Node1Output(TypedDict): private_data: str # 私有数据仅在 node_1 和 node_2 之间共享 def node_1(state: OverallState) -> Node1Output: output = {"private_data": "由 node_1 设置"} print(f"进入节点 `node_1`:\n\t输入: {state}.\n\t返回: {output}") return output # node 2 输入仅请求 node_1 之后可用的私有数据 class Node2Input(TypedDict): private_data: str def node_2(state: Node2Input) -> OverallState: output = {"a": "由 node_2 设置"} print(f"进入节点 `node_2`:\n\t输入: {state}.\n\t返回: {output}") return output # node 3 只能访问总体状态（无法访问 node_1 的私有数据） def node_3(state: OverallState) -> OverallState: output = {"a": "由 node_3 设置"} print(f"进入节点 `node_3`:\n\t输入: {state}.\n\t返回: {output}") return output # 构建状态图 builder = StateGraph(OverallState) builder.add_node(node_1) # node_1 是第一个节点 builder.add_node( node_2 ) # node_2 是第二个节点，并接受来自 node_1 的私有数据 builder.add_node(node_3) # node_3 是第三个节点，不会看到私有数据 builder.add_edge(START, "node_1") # 使用 node_1 启动图 builder.add_edge("node_1", "node_2") # 从 node_1 传递到 node_2 builder.add_edge( "node_2", "node_3" ) # 从 node_2 传递到 node_3（仅共享总体状态） builder.add_edge("node_3", END) # 在 node_3 之后结束图 graph = builder.compile() # 使用初始状态调用图 response = graph.invoke( { "a": "在开始时设置", } ) print() print(f"图调用输出: {response}")

Entered node `node_1`:
	Input: {'a': 'set at start'}.
	Returned: {'private_data': 'set by node_1'}
Entered node `node_2`:
	Input: {'private_data': 'set by node_1'}.
	Returned: {'a': 'set by node_2'}
Entered node `node_3`:
	Input: {'a': 'set by node_2'}.
	Returned: {'a': 'set by node_3'}

Output of graph invocation: {'a': 'set by node_3'}

注释