如何结合控制流与状态更新使用Command¶
结合控制流（边）和状态更新（节点）是有用的。例如，您可能希望在一个节点中同时执行状态更新并决定下一步跳转到哪个节点。LangGraph提供了一种方法，通过从节点函数返回一个Command对象来实现这一点：
def my_node(state: State) -> Command[Literal["my_other_node"]]:
return Command(
# 状态更新
update={"foo": "bar"},
# 控制流
goto="my_other_node"
)
如果您使用了子图，您可能希望从一个子图中的节点导航到另一个子图（即父图中的不同节点）。为此，您可以在Command中指定graph=Command.PARENT：
def my_node(state: State) -> Command[Literal["my_other_node"]]:
return Command(
update={"foo": "bar"},
goto="other_subgraph", # 其中 `other_subgraph` 是父图中的一个节点
graph=Command.PARENT
)
本指南展示了如何在您的LangGraph应用程序中使用Command添加动态控制流。
设置环境¶
首先，让我们安装所需的包
为LangGraph开发设置LangSmith
注册LangSmith可以快速发现并解决您的LangGraph项目中的问题，提高性能。LangSmith允许您使用跟踪数据来调试、测试和监控使用LangGraph构建的LLM应用程序——更多关于如何开始的信息，请参阅这里。
让我们创建一个简单的图，包含3个节点：A、B和C。我们将首先执行节点A，然后根据节点A的输出结果决定接下来是执行节点B还是节点C。
基本用法¶
API Reference: StateGraph | START | Command
import random
from typing_extensions import TypedDict, Literal
from langgraph.graph import StateGraph, START
from langgraph.types import Command
# Define graph state
class State(TypedDict):
foo: str
# Define the nodes
def node_a(state: State) -> Command[Literal["node_b", "node_c"]]:
print("Called A")
value = random.choice(["a", "b"])
# this is a replacement for a conditional edge function
if value == "a":
goto = "node_b"
else:
goto = "node_c"
# note how Command allows you to BOTH update the graph state AND route to the next node
return Command(
# this is the state update
update={"foo": value},
# this is a replacement for an edge
goto=goto,
)
def node_b(state: State):
print("Called B")
return {"foo": state["foo"] + "b"}
def node_c(state: State):
print("Called C")
return {"foo": state["foo"] + "c"}
我们现在可以使用上述节点创建StateGraph了。请注意，该图没有用于路由的条件边! 这是因为控制流是在node_a内的Command中定义的。
builder = StateGraph(State)
builder.add_edge(START, "node_a")
builder.add_node(node_a)
builder.add_node(node_b)
builder.add_node(node_c)
# NOTE: there are no edges between nodes A, B and C!
graph = builder.compile()
Important
您可能已经注意到我们使用了Command作为返回类型注解，例如Command[Literal["node_b", "node_c"]]。这是图形渲染所必需的，并告诉LangGraph程序，node_a可以导航到node_b和node_c。
如果我们多次运行该图，会看到它根据节点A中的随机选择采取不同的路径（A -> B 或 A -> C）。
在父图中导航到一个节点¶
现在让我们演示如何可以从子图内部导航到父图中的不同节点。我们将通过将上面示例中的node_a改为一个单节点图，并将其作为子图添加到我们的父图中来实现这一点。
使用 Command.PARENT 更新状态
当您从子图节点向父图节点发送更新，并且这些更新涉及由父图和子图共享的状态模式中的同一个键时，您**必须**在父图状态中为要更新的键定义一个归约器。
import operator
from typing_extensions import Annotated
class State(TypedDict):
# NOTE: we define a reducer here
foo: Annotated[str, operator.add]
def node_a(state: State):
print("Called A")
value = random.choice(["a", "b"])
# this is a replacement for a conditional edge function
if value == "a":
goto = "node_b"
else:
goto = "node_c"
# note how Command allows you to BOTH update the graph state AND route to the next node
return Command(
update={"foo": value},
goto=goto,
# this tells LangGraph to navigate to node_b or node_c in the parent graph
# NOTE: this will navigate to the closest parent graph relative to the subgraph
graph=Command.PARENT,
)
subgraph = StateGraph(State).add_node(node_a).add_edge(START, "node_a").compile()
def node_b(state: State):
print("Called B")
# NOTE: since we've defined a reducer, we don't need to manually append
# new characters to existing 'foo' value. instead, reducer will append these
# automatically (via operator.add)
return {"foo": "b"}
def node_c(state: State):
print("Called C")
return {"foo": "c"}
builder = StateGraph(State)
builder.add_edge(START, "subgraph")
builder.add_node("subgraph", subgraph)
builder.add_node(node_b)
builder.add_node(node_c)
graph = builder.compile()