Command:resume / goto / update 三合一原语

源码版本1.2.9

职责

Command 是 LangGraph 的「跨图控制原语」,定义在 libs/langgraph/langgraph/types.py(Command class:759)。一个 Command 实例可以同时承载三种语义——resume(从中断处恢复执行)、goto(指定下一步跳到哪个节点)、update(往当前图的状态通道写值)。节点函数返回 Command(...),或者调用方把它作为 graph.invoke(Command(...)) 的输入,Pregel 把它拆成 pending writes 写进通道。

它的位置在节点和 Pregel 主循环之间:节点产出 Command,Pregel 用 map_command(map_command:56)把它的字段拆成 (task_id, channel, value) 三元组写进 pending writes,下一轮 prepare_next_tasks 自然会看到这些 writes 并据此调度——goto 变成 Send 写进 TASKS 通道、update 直接写对应通道、resume 走 RESUME 通道交给 interrupt() 处理。

设计动机

• 统一表达「副作用」:节点除了往状态写值,可能还想「跳到非默认下一节点」「向父图传消息」「恢复中断」——以前这些是分散的 API,Command 把它们打包成单一返回类型,节点函数签名 -> Command[Literal["node_a", "node_b"]] 还能让编译期校验合法跳转目标。
• Command.PARENT 跨图层级:子图里的节点可以 return Command(graph=Command.PARENT, update=...),把更新写到父图状态——map_command 看到 graph == "__parent__"(PARENT guard:58)raise InvalidUpdateError("There is no parent graph"),真正落到父图是在父图的 _first 里处理子图上报的 writes。
• goto 接受 Send:Command(goto=Send("node", arg)) 等价于扇出一个 PUSH 任务;goto="node" 等价于 goto=Send("node", START)(触发那个节点的 PULL)。所以 Command 把 conditional edges 的能力内嵌到节点返回值里,不需要单独写 add_conditional_edges。
• resume 单点入口:Command(resume=...) 是 interrupt() 的唯一恢复路径——既支持单值 Command(resume="answer") 自动路由到唯一 pending interrupt,也支持 dict Command(resume={interrupt_id: value, ...}) 多 interrupt 并发恢复。
• update 配合 reducer:Command(update={"messages": [msg]}) 走的还是通道的 reducer——messages 通道配 operator.add,这条 update 就是 append 而非覆盖。Overwrite 包装可以绕过 reducer(Overwrite:938),走 BinaryOperatorAggregate 通道的直写路径。

关键文件

• Command class:759 — @dataclass class Command(graph, update, resume, goto),泛型 N 表示 goto 节点名类型。
• Command fields:780 — graph / update / resume / goto 字段定义及 docstring。
• _update_as_tuples:791 — 把 update 拆成 [(channel, value), ...],支持 dict、list[tuple]、annotated dataclass 三种形式。
• map_command:56 — 把 Command 拆成 pending writes 序列的核心函数。
• PARENT guard:58 — cmd.graph == Command.PARENT 时报「没有父图」错误。
• goto as sends:62 — cmd.goto 拆成 Send 列表,Send 写进 TASKS 通道,str 写进 branch:to:<node> 触发 START。
• resume write:72 — cmd.resume 写进 RESUME 通道,由 interrupt() 在重入时读取。
• update writes:74 — cmd.update 拆成 (channel, value) 写进对应通道,走 reducer。
• Command resume mapping:904 — 父图收到 Command(resume=...) 后,把 resume 路由到对应 pending interrupt 的入口。
• Overwrite:938 — Overwrite(value=...) 包装,配 BinaryOperatorAggregate 通道时绕过 reducer 直写。

数据流

map_command 是 Command 解析的核心(map_command:56):

def map_command(cmd: Command) -> Iterator[tuple[str, str, Any]]:
    """Map input chunk to a sequence of pending writes in the form (channel, value)."""
    if cmd.graph == Command.PARENT:
        raise InvalidUpdateError("There is no parent graph")
    if cmd.goto:
        if isinstance(cmd.goto, (tuple, list)):
            sends = cmd.goto
        else:
            sends = [cmd.goto]
        for send in sends:
            if isinstance(send, Send):
                yield (NULL_TASK_ID, TASKS, send)
            elif isinstance(send, str):
                yield (NULL_TASK_ID, f"branch:to:{send}", START)
            else:
                raise TypeError(
                    f"In Command.goto, expected Send/str, got {type(send).__name__}"
                )
    if cmd.resume is not None:
        yield (NULL_TASK_ID, RESUME, cmd.resume)
    if cmd.update:
        for k, v in cmd._update_as_tuples():
            yield (NULL_TASK_ID, k, v)

task_id 全是 NULL_TASK_ID——这些 writes 不属于任何具体任务,而是图级别的「外发」:goto 的 Send 进 TASKS 通道(下一轮 prepare_next_tasks 看到后会作为 PUSH 任务调度),goto 的 str 走 branch:to:<node> 通道(条件边专用通道,写 START 表示「触发那个节点的入口」);resume 进 RESUME 通道,在节点重入时由 interrupt() 消费;update 的每个 (k, v) 进对应状态通道,由 reducer 合并。

Command 作为 graph.stream(...) 的输入时,PregelLoop._first(Command resume mapping:904)先把它转成 pending writes,再按 task_id 分组写到 checkpoint_pending_writes,然后 prepare_next_tasks 自然会把它们处理掉。Command 作为节点返回值时,attach_node在节点writers里挂了_get_updatesmapper(<SrcLink path="libs/langgraph/langgraph/graph/state.py" lines="1431" label="attach_node"/>),从Command._update_as_tuples()抽出 update 写进状态通道,goto 和 resume 通过_control_branchmapper 路由到branch:to:通道和RESUME`。

边界与失败

• Command.PARENT 在根图报错:map_command 直接 raise InvalidUpdateError("There is no parent graph")(PARENT guard:58)——Command.PARENT 只在子图节点里有意义,根图调用直接报错。
• goto 类型限制:goto 只接受 Send / str 或它们的序列,其他类型 raise TypeError(goto type:69)——不允许直接传整数、dict 之类。
• resume 不指定 id 时强制单 pending:Command(resume="single") 在多个 pending interrupt 时 raise RuntimeError(multi interrupt error:917),必须改 Command(resume={id: value, ...})。
• update 走 reducer 不能直写:BinaryOperatorAggregate 通道(如 Annotated[list, operator.add])默认走 reducer,多个 update 合并;想绕过 reducer 用 Overwrite(value=...)(Overwrite:938),但同一超步对同一通道多次 Overwrite 会 InvalidUpdateError。
• Command 与 entrypoint.final 互斥:@entrypoint 函数返回 entrypoint.final(value=..., save=...),内部走 END / PREVIOUS 通道;返回 Command 走控制通道——两者不能同时用,entrypoint 函数应该返回 Command 或 final,不混。
• 空 Command 报错:Command() 三个字段都是 None / () 时,map_command 什么都不 yield,_first 检测到 writes 为空会 raise EmptyInputError("Received empty Command input")(empty command:928)。

小结

Command 是 resume / goto / update 的三合一原语,通过 map_command 拆成通道 writes,下一轮 prepare_next_tasks 自然处理。它把 conditional edges、interrupt resume、状态更新统一到一个返回值里。配套的 interrupt() 看 /interrupt/interrupt,Send 扇出语义看 /subgraph/send-command。对照官方资料:LangGraph 文档 · README。