Command:resume / goto / update の 3 in 1 プリミティブ

源码版本1.2.9

役割

Command は LangGraph の「グラフ横断制御プリミティブ」で、libs/langgraph/langgraph/types.py で定義されています(Command class:759)。1 つの Command インスタンスは 3 つの意味を同時に運べます——resume (中断からの実行再開)、goto (次ステップでどのノードに飛ぶか)、update (現在のグラフの状態チャネルへの値書き込み)。ノード関数が Command(...) を返すか、呼び出し側が graph.invoke(Command(...)) の入力として使い、Pregel がこれを pending writes に分解してチャネルに書き込みます。

位置としてはノードと Pregel のメインループの間にあります。ノードが Command を産出し、Pregel は map_command(map_command:56) でそのフィールドを (task_id, channel, value) の 3 要組に分解して pending writes に書き込みます。次ラウンドの prepare_next_tasks がこれらの writes を自然に見て、スケジュールします——goto は Send になって TASKS チャネルに書き込まれ、update は対応チャネルに直接書き込まれ、resume は RESUME チャネル経由で interrupt() に渡されます。

設計動機

• 「副作用」の統一表現:ノードは状態に値を書くだけでなく、「非デフォルトの次ノードに飛ぶ」「親グラフにメッセージを伝える」「中断を回復する」こともできます。以前はこれらが分散した API でしたが、Command が単一の戻り値型にパッケージします。ノード関数シグネチャ -> Command[Literal["node_a", "node_b"]] でコンパイル時に有効な遷移先を検証できます。
• Command.PARENT でグラフ階層を横断:サブグラフ (subgraph) のノードが 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)) はファンアウト (fan-out) の 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 の 3 形式をサポートします。
• 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_updates mapper を取り付け (attach_node:1431)、Command._update_as_tuples() から update を抽出して状態チャネルに書き込み、goto と resume は _control_branch mapper で 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() の 3 フィールドがすべて None / () の場合、map_command は何も yield せず、_first が writes が空であることを検出して raise EmptyInputError("Received empty Command input") します(empty command:928)。

まとめ

Command は resume / goto / update の 3 in 1 プリミティブで、map_command でチャネル writes に分解され、次ラウンドの prepare_next_tasks が自然に処理します。conditional edges、interrupt resume、状態更新を 1 つの戻り値に統合します。対応する interrupt() は /interrupt/interrupt、Send のファンアウトセマンティクスは /subgraph/send-command を参照してください。公式資料:LangGraph ドキュメント · README