第10章 Command 与高级控制流

10.1 引言

在前面的章节中，我们看到了 LangGraph 如何通过 Channel 和边（Edge）来定义数据的流转路径，通过 interrupt() 实现暂停与恢复。然而，在真实的 Agent 应用中，控制流往往不能在编译时完全确定——节点可能需要根据运行时的结果动态决定下一步去哪里，可能需要在更新状态的同时跳转到特定节点，甚至可能需要跨越图的边界向父图发送指令。

Command 类型正是 LangGraph 对这些需求的统一回答。它是一个多功能的控制流原语，将状态更新（update）、路由跳转（goto）、中断恢复（resume）和跨图通信（graph）四种能力融合在一个简洁的接口中。从设计哲学上看，Command 实现了一种"从节点内部控制图"的模式——节点不再是被动的数据处理器，而是可以主动驾驭图的执行引擎。

在传统的图计算框架中，控制流（节点之间的跳转）和数据流（状态的更新）是分离的——你在编译时通过边和条件边定义控制流，在运行时通过节点返回值更新数据。这种分离在简单场景下工作得很好，但在 Agent 系统中会带来摩擦：Agent 的决策往往是"在同一个推理步骤中既产生数据又决定下一步去哪里"。Command 打破了这个人为的分离，让节点可以在一次返回中同时表达数据更新和控制流意图。这种统一大幅简化了动态路由、分支合并、错误恢复等复杂场景的实现。

本章将从源码层面深入分析 Command 的实现：从数据结构定义、到 Pregel 循环中的处理逻辑、再到跨图通信的实现机制。我们还将对比 Command 与 Send 的区别，并通过实际用例展示其在复杂工作流中的应用。通过本章的学习，读者将能够在自己的 Agent 系统中自如地运用 Command 来构建灵活的动态控制流，包括条件路由、多目标分发、人机交互后的分支选择以及跨图的协调通信。

本章要点

1. Command 类定义：理解 update/resume/goto/graph 四个字段的语义与交互
2. 从节点内控制流程：掌握 Command 作为节点返回值时的处理机制
3. map_command 映射：深入 Command 到 pending writes 的转换逻辑
4. Command.PARENT 跨图通信：理解 ParentCommand 异常和父子图之间的控制流传递
5. 与 Send 的区别：明确两者在语义、作用域和使用场景上的差异
6. 实际用例：通过完整示例展示动态路由、多目标跳转和跨图控制的应用

10.2 Command 类定义

10.2.1 数据结构

Command 定义在 langgraph/types.py 中，是一个不可变的泛型数据类：

# langgraph/types.py

@dataclass(**_DC_KWARGS)  # kw_only=True, slots=True, frozen=True
class Command(Generic[N], ToolOutputMixin):
    """One or more commands to update the graph's state and send messages to nodes."""

    graph: str | None = None
    update: Any | None = None
    resume: dict[str, Any] | Any | None = None
    goto: Send | Sequence[Send | N] | N = ()

    PARENT: ClassVar[Literal["__parent__"]] = "__parent__"

四个字段各司其职，它们的设计体现了"正交组合"的原则——每个字段独立控制一个维度的行为，可以任意组合使用：

update：要应用到图状态的更新。可以是字典、元组列表或 Pydantic 模型——与节点直接返回字典时的效果相同。当 update 为 None 时，不对状态做任何修改。这允许你创建纯控制流的 Command（只有 goto 没有数据更新），或纯恢复的 Command（只有 resume 没有状态变更）。update 的类型灵活性使得它可以适配不同的 Schema 定义方式，无论你使用 TypedDict、Pydantic 模型还是简单的字典。

goto：指定下一步要执行的节点。可以是单个节点名字符串、节点名列表、Send 对象或 Send 对象列表。这是 Command 最强大的能力——从节点内部直接控制路由，无需在编译时定义条件边。当 goto 为空序列时（默认值），不产生任何路由指令，图按照正常的边定义继续执行。字符串形式的 goto 和 Send 形式的 goto 在底层使用不同的 Channel 机制——字符串触发的是 PULL 类型的任务（从全局状态读取输入），Send 触发的是 PUSH 类型的任务（使用自定义的输入参数）。

resume：中断恢复值。可以是单个值或中断 ID 到值的映射。在上一章中我们已经详细讨论过。值得补充的是，resume 可以与 update 和 goto 同时使用——这在"恢复后立即跳转到特定节点"的场景中非常有用，例如人类审批后根据审批结果路由到不同的处理流程。

graph：指定命令的目标图。None 表示当前图，Command.PARENT（即 "__parent__"）表示父图。PARENT 被定义为 ClassVar（类变量），这意味着它是一个常量，不参与实例化和序列化。它的值 "__parent__" 是一个内部约定的哨兵字符串，在 _control_branch 和 map_command 中被专门检测和处理。

10.2.2 _update_as_tuples 方法

Command 的 update 字段需要转换为 (channel, value) 元组列表才能被 Pregel 循环处理。_update_as_tuples 方法负责这个转换：

def _update_as_tuples(self) -> Sequence[tuple[str, Any]]:
    if isinstance(self.update, dict):
        return list(self.update.items())
    elif isinstance(self.update, (list, tuple)) and all(
        isinstance(t, tuple) and len(t) == 2 and isinstance(t[0], str)
        for t in self.update
    ):
        return self.update
    elif keys := get_cached_annotated_keys(type(self.update)):
        # Pydantic 模型或 dataclass
        return get_update_as_tuples(self.update, keys)
    elif self.update is not None:
        # 标量值映射到 __root__ channel
        return [("__root__", self.update)]
    else:
        return []

这个方法支持四种输入格式，从简单到复杂逐级处理：