Command 控制流

Command 是 LangGraph 中一种更精细的控制流机制，允许节点在返回 state 更新的同时，动态指定下一步要执行的节点。

注意：Command 是 LangGraph 的进阶功能，大部分场景使用 add_conditional_edges 就能满足需求。

为什么需要 Command

add_conditional_edges 的局限性：

• 路由逻辑在编译时就固定了
• 无法在运行时动态决定下一步
• 所有条件边在节点执行之前就已确定

Command 允许节点在执行时动态决定下一步：

基本用法

from langgraph.graph import Command
from typing import Literal

def dynamic_node(state: State) -> Command[Literal["next_a", "next_b"]]:
    # 执行逻辑
    result = process(state["input"])

    # 根据执行结果动态决定下一步
    if result["needs_a"]:
        return Command(
            goto="next_a",
            update={"output": result["data"]}
        )
    else:
        return Command(
            goto="next_b",
            update={"output": result["data"]}
        )

Command 包含两个部分：

• goto：下一步要执行的节点名称
• update：（可选）要合并到 state 中的更新

使用场景

1. 动态节点选择

根据 LLM 输出的具体内容选择不同的后续处理：

def llm_router(state: State) -> Command:
    response = llm.invoke(state["messages"])

    # LLM 决定下一步做什么
    if response.tool_calls:
        return Command(
            goto="execute_tool",
            update={"messages": [response]}
        )
    elif response.content.startswith("CLARIFY:"):
        return Command(
            goto="ask_clarification",
            update={"messages": [response]}
        )
    else:
        return Command(
            goto="generate_final",
            update={"messages": [response], "final": True}
        )

2. 递归处理

def recursive_processor(state: State) -> Command:
    """递归处理直到条件满足"""
    chunk = get_next_chunk(state["data"], state["position"])

    if chunk is None:
        # 没有更多数据了
        return Command(goto=END, update={"done": True})

    new_position = state["position"] + 1
    if new_position >= state["max_steps"]:
        return Command(goto=END, update={"position": new_position})

    # 继续处理下一个 chunk
    return Command(
        goto="recursive_processor",
        update={"position": new_position, "result": chunk}
    )

3. 错误恢复

def fault_tolerant_node(state: State) -> Command:
    try:
        result = risky_operation(state["input"])
        return Command(goto="next_step", update={"output": result})
    except Exception as e:
        if state.get("retries", 0) < 3:
            return Command(
                goto="fault_tolerant_node",  # 重试自己
                update={"retries": state.get("retries", 0) + 1}
            )
        else:
            return Command(
                goto="error_handler",
                update={"error": str(e)}
            )

Command vs add_conditional_edges

特性	add_conditional_edges	Command
定义时机	编译时（静态）	运行时（动态）
路由依据	节点执行完毕后的 state	节点执行过程中的任意逻辑
灵活性	固定路由表	完全动态
适用场景	大多数情况	需要运行时决策的复杂场景
可读性	好，路由逻辑集中	路由分散在节点中

最佳实践

1. 优先用 add_conditional_edges，简单直观
2. Command 适合需要运行时状态感知的动态路由
3. 不要滥用 Command，否则图结构会变得难以理解
4. Command 的 goto 目标必须已经被添加为节点

参考

• 官方文档：Graph API overview
• API 参考：Command