将控制流（边）和状态更新（节点）结合起来会很有用。例如，可能希望在同一个节点中同时执行状态更新和决定下一步转到哪个节点。LangGraph 提供了一种通过从Command节点函数返回对象来实现此目的的方法：def my_node(state: State) -> Command[Literal["my_other_node"]]: return Command( # state update update={"foo": "bar"}, # control flow goto="my_other_node" )
让我们创建一个包含 3 个节点的简单图：A、B 和 C。我们将首先执行节点 A，然后根据节点 A 的输出决定下一步是否转到节点 B 还是节点 C。import randomfrom typing_extensions import TypedDict, Literalfrom langgraph.graph import StateGraph, STARTfrom langgraph.types import Commandclass State(TypedDict): foo: str# Define the nodesdef node_a(state: State) -> Command[Literal["node_b", "node_c"]]: print("Called A") value = random.choice(["a", "b"]) # 这是一种 conditional edge fn 的另一种写法 if value == "a": goto = "node_b" else: goto = "node_c" # 这里展示了Command如何允许你同时更新图状态和路由到下一个节点 return Command( # state update update={"foo": value}, # 路由到一个 edge goto=goto, )# Nodes B and C are unchangeddef 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)# 中文：注意：这里没有 edge 连接节点A、B和C！graph = builder.compile()print(graph.invoke({"foo": ""}))
Called ACalled C{'foo': 'bc'}
