在某些情况下,您可能希望节点交换对中间逻辑至关重要但不需要成为图的主要架构一部分的信息。这些私有数据与图的整体输入/输出无关,仅应在某些节点之间共享。在本操作指南中,我们将创建一个由三个节点(node_1、node_2 和 node_3)组成的示例顺序图,其中私有数据在前两个步骤(node_1 和 node_2)之间传递,而第三步(node_3)只能访问公共整体状态。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}")
核心思路是当你当前节点设置了数据之后,如果希望下个节点使用,你就把这份数据return,他会被携带到下一个节点。当你不希望被传递到下一个节点,你就只返回公共的部分即可。