跳到正文

如何使用 Command 结合控制流和状态更新

先决条件

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

结合控制流（边）和状态更新（节点）会很有用。例如，您可能希望在同一个节点中既执行状态更新，又决定接下来要进入哪个节点。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"
    )

如果您正在使用子图，您可能希望从子图中的节点导航到不同的子图（即父图中的不同节点）。为此，您可以在 Command 中指定 graph=Command.PARENT。

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

使用 Command.PARENT 进行状态更新

当您从子图节点向父图节点发送更新，并且更新的键被父图和子图的状态模式共享时，您必须为在父图状态中更新的键定义一个reducer。请参阅下面的示例。

本指南将展示如何使用 Command 在 LangGraph 应用中添加动态控制流。

设置

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

pip install -U langgraph

设置 LangSmith 以进行 LangGraph 开发

注册 LangSmith，快速发现问题并提高 LangGraph 项目的性能。LangSmith 让您可以使用跟踪数据来调试、测试和监控使用 LangGraph 构建的 LLM 应用——在此处阅读更多关于如何入门的信息。

让我们创建一个包含 3 个节点的简单图：A、B 和 C。我们将首先执行节点 A，然后根据节点 A 的输出决定接下来是进入节点 B 还是节点 C。

基本用法

API 参考：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()

重要提示

您可能已经注意到，我们使用了 Command 作为返回类型注解，例如 Command[Literal["node_b", "node_c"]]。这对于图渲染是必要的，并告诉 LangGraph node_a 可以导航到 node_b 和 node_c。

from IPython.display import display, Image

display(Image(graph.get_graph().draw_mermaid_png()))

如果我们多次运行该图，会看到它根据节点 A 中的随机选择走不同的路径（A -> B 或 A -> C）。

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

Called A
Called C


{'foo': 'bc'}

现在，让我们演示如何从子图内部导航到父图中的不同节点。我们将通过将上述示例中的 node_a 更改为一个单节点图，并将其添加为父图的子图来实现。

使用 Command.PARENT 进行状态更新

当您从子图节点向父图节点发送更新，并且更新的键被父图和子图的状态模式共享时，您必须为在父图状态中更新的键定义一个reducer。

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()


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

Called A
Called C


{'foo': 'bc'}

评论