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