当前位置：首页 > article >正文

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

article 2025/2/23 7:05:10

这篇文章我们将 LangGraph中的控制流（边）和状态更新（节点）结合起来使用。比如，我们希望同时执行状态更新并决定下一步要转到哪个节点，且这些操作在同一个节点中完成。而正好 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"
    )

如果我们使用的是 subgraphs（子图）我们还可以通过 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
    )

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

定义节点和State

定义 graph State：

class State(TypedDict):
    foo: str

定义节点A，

def node_a(state: State) -> Command[Literal["node_b", "node_c"]]:
    print("Called A")
    value = random.choice(["a", "b"])
    # 判断应该跳转到哪个节点
    if value == "a":
        goto = "node_b"
    else:
        goto = "node_c"

    # Command 允许我们同时更新图状态并路由到下一个节点。
    return Command(
        # 更新图中的状态
        update={"foo": value},
        # 替换edge
        goto=goto,
    )

定义节点b或者c：

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。需要注意的是，该图没有定义条件的路由edge！这是因为控制流是用 Command 内部 node_a 定义的。

builder = StateGraph(State)
builder.add_edge(START, "node_a")
builder.add_node(node_a)
builder.add_node(node_b)
builder.add_node(node_c)

graph = builder.compile()

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

from IPython.display import display, Image

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

在这里插入图片描述
我们来试试多次调用graph：

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

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

导航到 parent graph 中的节点

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

import operator
from typing_extensions import Annotated


class State(TypedDict):
    # N定义一个reducer 自动附加消息
    foo: Annotated[str, operator.add]


def node_a(state: State):
    print("Called A")
    value = random.choice(["a", "b"])
    if value == "a":
        goto = "node_b"
    else:
        goto = "node_c"

    # Command 允许我们同时更新图状态并路由到下一个节点。
    return Command(
        update={"foo": value},
        goto=goto,
        # 这会告诉 LangGraph 导航到父图中的 node_b 或 node_c 这将导航到相对于子图的最近的父图
        graph=Command.PARENT,
    )

subgraph = StateGraph(State).add_node(node_a).add_edge(START, "node_a").compile()

def node_b(state: State):
    print("Called B")
    # 由于我们已经定义了一个 reducer，因此不需要手动将新字符附加到现有的 foo 值上。相反，reducer 会自动（通过 operator.add）将这些值附加到现有值中。
    return {"foo": "b"}


def node_c(state: State):
    print("Called C")
    return {"foo": "c"}

然后我们来构造graph：

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