Day 15 / 共 60 天 · 阶段3 控制流

Command 对象:节点自己说"更新 + 去哪"

条件边(Day 13-14)是"在节点外面挂一个路由器"。今天的 Command 反过来——让节点在自己的返回值里同时说清"我更新哪些状态、下一步去哪个节点、甚至跳到父图去"。这是 LangGraph 1.0 里做多智能体交接(handoff)的主力武器。

📍 你在 60 天里的位置(阶段3 · 控制流 D13-18)
D13 边 D14 Branch D15 Command D16 Send D17 递归上限 D18 START/END
💡 一个类比先兜住今天(快递面单) 条件边像"分拣中心的扫码机"——包裹(节点输出)到了传送带尽头,机器扫一眼决定进哪个格口。Command 则像包裹自己贴了一张手写面单:面单上同时写着"①里面装的新东西(update)②下一站送到哪(goto)③这单归本仓还是转去总仓(graph)"。分拣中心不用再扫码猜,照着面单办就行。所以 Command = 节点亲手写的"状态更新 + 路由"合一面单。
L01

为什么已经有条件边,还要 Command?

🤔 痛点:路由依据就在节点里算出来了,为什么还要拆成两处?做多智能体时,一个节点(比如"分诊 agent")往往一边算出结论、一边就知道该交给谁。用条件边的话,你得:① 节点把"交给谁"写进 state 某字段;② 再写一个条件边函数把那个字段读出来 return。决策明明一次就有,却要在两个地方来回搬运一个中间字段——啰嗦且容易忘同步。
💡 本质:Command 把"改状态"和"选下一步"打包进节点的返回值节点不再只返回"状态更新 dict",而是可以返回一个 Command(update=..., goto=...)。引擎看到返回的是 Command,就既应用 update、又按 goto 决定下一步。路由依据不必落进 state 再被读出来——决策和跳转在同一个返回值里一气呵成。
条件边(Day 13-14)Command(今天)
路由逻辑写在节点的独立函数节点,返回值里
更新+路由分两步(写字段→读字段)一步(update 与 goto 同一对象)
跨父子图跳转不能graph=Command.PARENT
可视化好(能画确定箭头)Command[Literal[...]] 标注才画得出
大白话条件边是"门口有个保安问你去哪";Command 是"你出门时自己举个牌子写着去哪"。多智能体交接时后者顺手得多——因为"交给谁"本来就是这个 agent 自己想清楚的事。
L02

Command 的四个字段

源码在 types.py:758,本体就是个 dataclass,四个字段:

# types.py:758
@dataclass(**_DC_KWARGS)
class Command(Generic[N], ToolOutputMixin):
    graph: str | None = None                       # 这条命令发给哪张图
    update: Any | None = None                      # 要应用到状态的更新
    resume: dict[str, Any] | Any | None = None     # 恢复中断用(阶段7 human-in-loop)
    goto: Send | Sequence[Send | N] | N = ()       # 下一步去哪(节点名 / Send / 它们的列表)

    PARENT: ClassVar[Literal["__parent__"]] = "__parent__"   # types.py:808
graph命令投给哪张图。None=当前图;Command.PARENT=最近的父图(L05 详讲)。这是子图能"跳出去交接给兄弟 agent"的关键。
update要合并进状态的更新。它什么类型都行——dict、二元组列表、甚至 Pydantic/dataclass 对象。怎么规整见 L03。
resume专门配合 interrupt() 用的"恢复值"。今天先知道它是 Command 的一部分,阶段 7 人在环再展开。
goto下一步目标:单个节点名、单个 Send、或它们的列表。默认 ()(空元组)表示"不指定下一步,走图本身定义的边"。
PARENT 常量类变量,值是字符串 "__parent__"。写 Command(graph=Command.PARENT, ...) 比裸写字符串更安全、更可读。
💡 本质:Command 是"节点想对引擎下的一组指令"的声明式打包注意它什么都不执行——只是把"我想更新啥、去哪、给哪张图"这几件事装进一个不可变对象。真正执行是引擎收到它之后的事(L04)。这种"命令即数据"的设计,让节点的意图可被序列化、可被检查点保存、可被工具调用返回(ToolOutputMixin 就是让工具能直接返回 Command)。
goto 默认值用 空元组 () 而非 None,很讲究:空元组是"空的可迭代序列",后面判断 if cmd.goto: 时空元组天然为假,且遍历它零次;用 None 反而要额外判空。这是"选一个好的默认值来消除特判"的小巧思。
L03

_update_as_tuples:把千奇百怪的 update 归一化

update 允许各种类型,但引擎只吃"(通道名, 值) 的二元组序列"。转换器是 _update_as_tuplestypes.py:793

# types.py:793
def _update_as_tuples(self) -> Sequence[tuple[str, Any]]:
    if isinstance(self.update, dict):
        return list(self.update.items())                       # ① dict → 直接拆成 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                                     # ② 本就是 (str, v) 列表 → 原样
    elif keys := get_cached_annotated_keys(type(self.update)):
        return get_update_as_tuples(self.update, keys)         # ③ Pydantic/dataclass → 按字段抽
    elif self.update is not None:
        return [("__root__", self.update)]                     # ④ 其它非空 → 塞进 __root__ 通道
    else:
        return []                                              # ⑤ None → 空
dict 分支最常见:{"foo":1}[("foo",1)]。每个 key 就是一个通道名。
(str,v) 列表分支你已经手动给成二元组列表了(比如想给同一个 key 多次写入——dict 做不到重复 key,列表可以),原样放行。这是个容易被忽略但很有用的口子。
Pydantic/dataclass 分支如果 update 是个有类型注解字段的对象(Day 12 的 Pydantic State),用缓存的字段名把它拆成二元组。get_cached_annotated_keys 带缓存,避免每次反射开销。
__root__ 兜底边界:既不是 dict 也不是对象的裸值(比如一个字符串),塞进特殊通道 __root__。用于"整个状态就是单个值"的非 dict 图。
None → []没有 update 就返回空列表,下游遍历零次,无副作用。
💐 设计取舍①:为什么接受这么多种 update 类型,而不强制 dict? 因为 Command 要能服务多种状态形态:TypedDict 图用 dict、Pydantic 图用模型对象、单值图用裸值、想重复写同一通道的用二元组列表。如果强制只收 dict,Pydantic 图的用户就得每次手动 .model_dump()、单值图没法表达。代价是这个转换函数得写 5 个分支——但复杂度收敛在一个函数里,换来的是所有状态形态的用户都能自然地返回 Command。把复杂性关进一个笼子,是好的取舍。
L04

map_command:Command 真正被执行的地方

节点返回或用户传入的 Command,最终由 map_command 翻译成一串"通道写入"。pregel/_io.py:56

# pregel/_io.py:56
def map_command(cmd: Command) -> Iterator[tuple[str, str, Any]]:
    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)              # goto 是 Send → 写 TASKS 通道
            elif isinstance(send, str):
                yield (NULL_TASK_ID, f"branch:to:{send}", START)  # goto 是节点名 → 写触发通道
            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)               # resume → 写 RESUME 通道
    if cmd.update:
        for k, v in cmd._update_as_tuples():                   # update → 每个字段写一个通道
            yield (NULL_TASK_ID, k, v)
graph == PARENT 报错边界:在当前这层解析时如果要求发给父图、但根本没有父图,直接 InvalidUpdateError。(真有父图时,这个 Command 是通过 ParentCommand 异常冒泡上去的,见 L05,不走这里。)
goto 是 Send → 写 TASKS目标是 Send(带自定义状态的动态任务,Day 16)→ 写进 TASKS 通道,触发一个 PUSH 任务。
goto 是 str → 写 branch:to:X目标是节点名 → 往 branch:to:X 这个触发通道写个 START 信号。注意:这和 Day 14 条件边最终写的是同一类通道——殊途同归,都是"往触发通道写值"。
update → 每字段一个 yield调 L03 的 _update_as_tuples() 拆出二元组,每个字段作为一次通道写入。
NULL_TASK_ID这些写入都挂在"空任务 id"下——表示它们是"输入/命令级"的写入,不属于某个具体节点任务。
数据结构:一个 Command 被 map_command 摊平成多条通道写入 Command update={"foo":1} goto="bar" resume=None graph=None map_command ("__none__","foo",1) ← update 落 foo 通道 ("__none__","branch:to:bar",START) ← goto 触发 resume/graph 为空 → 不产生写入
图注:Command 是"一组意图",map_command 把它摊平成引擎唯一认识的语言——(task_id, 通道, 值) 三元组。
L05

goto 跨父图 & ParentCommand 冒泡

Command 最强的能力是跨图跳转:子图里的节点可以说"我要交接给父图里的某个节点"。它靠一个特殊异常 ParentCommand 往上冒泡。errors.py:129

# errors.py:129
class ParentCommand(GraphBubbleUp):
    args: tuple[Command]
    def __init__(self, command: Command) -> None:
        super().__init__(command)
继承 GraphBubbleUpGraphBubbleUp 是一类"专门用来向上冒泡、不算真错误"的异常基类(中断 GraphInterrupt 也继承它)。用异常做控制流,是因为要穿透多层调用栈一路把命令抛到父图的执行循环。
装着一个 Command异常体里就存着那个 Command(graph=PARENT, goto="兄弟节点")。父图的循环捕获到它,取出 Command 在父图这一层重新 map_command——这次 graph==PARENT 的检查在父图看来就不成立了(父图就是它要发的目标),于是正常落地。
📝 真实值:多智能体交接(handoff)
def travel_advisor(state) -> Command[Literal["hotel_advisor"]]:
    # 分诊 agent 决定:更新一句话,然后把控制权交给酒店 agent
    return Command(
        update={"messages": [AIMessage("帮你转接酒店顾问")]},
        goto="hotel_advisor",
        graph=Command.PARENT,     # ← 跳出当前子图,交给父图里的兄弟 agent
    )
注意返回类型标了 Command[Literal["hotel_advisor"]]——这让图可视化知道"这个节点可能去 hotel_advisor",和 Day 14 的 Literal 推断异曲同工。
💐 设计取舍②:为什么用"异常"实现跨图跳转,而不是返回值层层传递? 如果靠返回值传,子图的每一层执行循环都得写"检查返回值是不是要发给父图、是的话原样往上返回"的样板代码,侵入性极强。用 异常冒泡:子图节点直接 raise ParentCommand(...),中间各层什么都不用改,异常自动穿透到最近一个愿意捕获它的父图循环。代价是"用异常做正常控制流"看着别扭、且要小心别被通用 except Exception 吞掉——所以它专门继承 GraphBubbleUp,让框架能精确识别"这不是错误,是控制信号"。
🚫 坑:在顶层图里写 graph=Command.PARENT顶层图没有父图。L04 第一行 if cmd.graph == Command.PARENT: raise InvalidUpdateError("There is no parent graph") 就是拦这个的。多智能体交接务必确认你的 agent 是作为子图嵌进父图的,PARENT 才有意义。
L06

Command vs 条件边:到底用哪个

两者最终都翻译成"往 branch:to:X 通道写值"(对比 Day 14 L05 和今天 L04,你会发现落点一样)。区别只在"路由逻辑写在哪、能不能跨图":

👶 小白:既然落点一样,我是不是随便用哪个都行?

👨‍🏫 老师:功能上单图内确实可互换,但有选型偏好:① 路由依据是不是节点自己算的?是(多 agent 交接、工具返回时就知道去哪)→ 用 Command,省得来回搬字段;否(路由要看多个节点汇总后的状态)→ 用条件边,逻辑独立更清晰。② 要跨父子图吗?要 → 只能 Command。③ 在乎流程图好不好看?条件边天然带 path_map 好画;Command 得记得标 Command[Literal[...]]经验法则:多智能体交接首选 Command;集中式的"看全局状态分流"用条件边。

场景推荐原因
ReAct:有工具调用就去 tools条件边路由依据是"消息里有没有 tool_calls",独立函数更清晰
多 agent:分诊后交接给专家Command交给谁是分诊 agent 自己的决定,update+goto 一步到位
子图交接给父图兄弟Command(PARENT)条件边根本做不到跨图
工具函数直接决定下一步Command工具可返回 Command(ToolOutputMixin)
L07

今日小结 + 动手 + 明日预告

🧠 今天你应该能回答

  • Command 相比条件边解决了什么痛点?(更新+路由一步到位、支持跨父子图交接)
  • Command 四个字段?(graph / update / resume / goto
  • _update_as_tuples 为什么要 5 个分支?(兼容 dict / 二元组列表 / Pydantic / 裸值 / None)
  • map_command 把 goto 翻译成什么?(Send→写 TASKS;节点名→写 branch:to:X
  • 跨图跳转靠什么机制?(ParentCommand 异常冒泡,继承 GraphBubbleUp)
  • 什么时候选 Command、什么时候选条件边?(自算路由/跨图→Command;看全局分流→条件边)

✋ 10 分钟动手

# 1. 读 Command 定义与 _update_as_tuples
sed -n '758,809p' libs/langgraph/langgraph/types.py

# 2. 读 map_command(Command 落地的唯一入口)
sed -n '56,79p' libs/langgraph/langgraph/pregel/_io.py

# 3. 单图内用 Command 路由
python - <<'PY'
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from typing import TypedDict, Literal
class S(TypedDict): n: int
def a(state) -> Command[Literal["b","__end__"]]:
    return Command(update={"n": state["n"]+1},
                   goto="b" if state["n"] < 2 else END)
def b(state) -> Command[Literal["a"]]:
    return Command(update={"n": state["n"]+10}, goto="a")
g = StateGraph(S); g.add_node("a", a); g.add_node("b", b)
g.add_edge(START, "a")
print(g.compile().invoke({"n": 0}))   # 观察 a/b 互相 goto 直到 n>=2
PY
明天预告 · Day 16:今天 goto 里出现的 Send 还没细讲。Day 16 专攻 Send——它能带着自定义状态把同一个节点并行拉起 N 份(map-reduce 扇出)。我们会读 types.pySend 类和官方 bench/fanout_to_subgraph.py,看"一个主题列表如何炸成 N 个并行子任务再聚合"。
← Day 14 Branch 路由源码 Day 16 · Send 动态扇出 →