【Agent的革命之路——LangGraph】工作流中的 Command 模式 ...

这篇文章我们将 LangGraph中的控制流（边）和状态更新（节点）联合起来使用。比如，我们渴望同时执行状态更新并决定下一步要转到哪个节点，且这些操作在同一个节点中完成。而正好 LangGraph 提供了一种方法，可以通过从节点函数返回一个 Command 对象来实现这一点。

1. def my_node(state: State) -> Command[Literal["my_other_node"]]:
2.     return Command(
3.         # state update
4.         update={"foo": "bar"},
5.         # control flow
6.         goto="my_other_node"
7.     )

复制代码

假如我们使用的是 subgraphs（子图） 我们还可以通过 Command 中指定 graph=Command.PARENT这种方式，实现节点从一个子图导航到其他子图。

1. def my_node(state: State) -> Command[Literal["my_other_node"]]:
2.     return Command(
3.         update={"foo": "bar"},
4.         goto="other_subgraph",  # where `other_subgraph` is a node in the parent graph
5.         graph=Command.PARENT
6.     )

复制代码

当我们从子图节点向父图节点发送更新的时候，且更新的键同时存在于父图和子图的状态模式中时，我们必须为父图状态中正在更新的键界说一个 reducer。
下面我们开始我们的例子。
界说节点和State

界说 graph State：

1. class State(TypedDict):
2.     foo: str

复制代码

界说节点A，

1. def node_a(state: State) -> Command[Literal["node_b", "node_c"]]:
2.     print("Called A")
3.     value = random.choice(["a", "b"])
4.     # 判断应该跳转到哪个节点
5.     if value == "a":
6.         goto = "node_b"
7.     else:
8.         goto = "node_c"
10.     # Command 允许我们同时更新图状态并路由到下一个节点。
11.     return Command(
12.         # 更新图中的状态
13.         update={"foo": value},
14.         # 替换edge
15.         goto=goto,
16.     )

复制代码

界说节点b大概c：

1. def node_b(state: State):
2.     print("Called B")
3.     return {"foo": state["foo"] + "b"}
6. def node_c(state: State):
7.     print("Called C")
8.     return {"foo": state["foo"] + "c"}

复制代码

界说好了上面节点之后，现在，我们可以使用上述节点创建 StateGraph。必要注意的是，该图没有界说条件的路由edge！这是由于控制流是用 Command 内部 node_a 界说的。

1. builder = StateGraph(State)
2. builder.add_edge(START, "node_a")
3. builder.add_node(node_a)
4. builder.add_node(node_b)
5. builder.add_node(node_c)
7. graph = builder.compile()

复制代码

不知道大家注意到没有，我们的 node_a 使用了 Command 作为返回类型注解，比方 Command[Literal[“node_b”, “node_c”]]。这一步对于图的渲染是非常必要的，它会告诉 LangGraph：node_a 可以导航到 node_b 和 node_c。
现在我们看看图的·1展示效果：

1. from IPython.display import display, Image
3. display(Image(graph.get_graph().draw_mermaid_png()))

复制代码

我们来试试多次调用graph：

1. graph.invoke({"foo": ""})

复制代码

将会看到，graph会根据节点A中的随机选择，采用不同的路径（A-> B或A-> C）。
导航到 parent graph 中的节点

现在，让我们演示怎样从 subgraph 内部导航到 parent graph 中的不同节点。我们将通过将上述示例中的 node_a 转换为一个单节点图来实现这一点，然后将其作为子图添加到 parent graph 中。

1. import operator
2. from typing_extensions import Annotated
5. class State(TypedDict):
6.     # N定义一个reducer 自动附加消息
7.     foo: Annotated[str, operator.add]
10. def node_a(state: State):
11.     print("Called A")
12.     value = random.choice(["a", "b"])
13.     if value == "a":
14.         goto = "node_b"
15.     else:
16.         goto = "node_c"
18.     # Command 允许我们同时更新图状态并路由到下一个节点。
19.     return Command(
20.         update={"foo": value},
21.         goto=goto,
22.         # 这会告诉 LangGraph 导航到父图中的 node_b 或 node_c 这将导航到相对于子图的最近的父图
23.         graph=Command.PARENT,
24.     )
26. subgraph = StateGraph(State).add_node(node_a).add_edge(START, "node_a").compile()
28. def node_b(state: State):
29.     print("Called B")
30.     # 由于我们已经定义了一个 reducer，因此不需要手动将新字符附加到现有的 foo 值上。相反，reducer 会自动（通过 operator.add）将这些值附加到现有值中。
31.     return {"foo": "b"}
34. def node_c(state: State):
35.     print("Called C")
36.     return {"foo": "c"}

复制代码

然后我们来构造graph：

1. builder = StateGraph(State)
2. builder.add_edge(START, "subgraph")
3. builder.add_node("subgraph", subgraph)
4. builder.add_node(node_b)
5. builder.add_node(node_c)
7. graph = builder.compile()

复制代码

检察效果：

1. Called A
2. Called C
3. {'foo': 'bc'}

复制代码

到这里我们的代码就竣事了，我们通过这种方式展示了怎样使用 LangGraph 的 Command 对象在子图中更新状态并导航到父图中的不同节点。并通过 reducer 简化状态管理，就是自动保存消息链。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。