Command:resume / goto / update — drei-in-eins Primitiv

源码版本1.2.9

Verantwortung

Command ist LangGraphs «graphübergreifendes Steuerprimitiv (control primitive)», definiert in libs/langgraph/langgraph/types.py(Command class:759). Eine Command-Instanz kann gleichzeitig drei Bedeutungen tragen — resume (Ausführung aus einer Unterbrechung (interrupt) wieder aufnehmen), goto (angeben, zu welchem Knoten als Nächstes gesprungen wird) und update (Werte in die Zustandskanäle (channel) des aktuellen Graphen schreiben). Knotenfunktionen geben Command(...) zurück, oder die aufrufende Seite verwendet es als Eingabe für graph.invoke(Command(...)); Pregel zerlegt es in pending writes und schreibt sie in die Kanäle.

Seine Position liegt zwischen Knoten und dem Pregel-Hauptloop: Knoten erzeugen Command, Pregel verwendet map_command(map_command:56), um seine Felder in (task_id, channel, value)-Tripel zu zerlegen und in pending writes zu schreiben. Die nächste Runde prepare_next_tasks sieht diese writes natürlich und调度t entsprechend — goto wird zu Send und in den TASKS-Kanal geschrieben, update schreibt direkt in den entsprechenden Kanal, resume geht über den RESUME-Kanal und wird von interrupt() behandelt.

Entwurfsmotivation

• Einheitliche Darstellung von «Seiteneffekten»: Neben dem Schreiben von Werten in den Zustand möchte ein Knoten eventuell auch «zu einem nicht-Standard-Nachfolgeknoten springen», «eine Nachricht an den Elterngraphen senden» oder «eine Unterbrechung (interrupt) wiederaufnehmen» — früher waren das verteilte APIs. Command packt sie in einen einzigen Rückgabetyp; die Knotenfunktionssignatur -> Command[Literal["node_a", "node_b"]] erlaubt zudem Compile-Time-Prüfung der erlaubten Sprungziele.
• Command.PARENT für graphübergreifende Hierarchie: Ein Knoten in einem Untergraph (subgraph) kann return Command(graph=Command.PARENT, update=...) ausführen und Updates in den Zustand des Elterngraphen schreiben — map_command sieht bei graph == "__parent__"(PARENT guard:58)raise InvalidUpdateError("There is no parent graph"). Das tatsächliche Ankommen im Elterngraphen passiert im _first des Elterngraphen, wenn die vom Untergraphen gemeldeten writes verarbeitet werden.
• goto akzeptiert Send:Command(goto=Send("node", arg)) ist äquivalent dazu, eine PUSH-Aufgabe auszufächern; goto="node" ist äquivalent zu goto=Send("node", START) (löst den PULL dieses Knotens aus). So bettet Command die Fähigkeiten von conditional edges in den Rückgabewert des Knotens ein — ein separates add_conditional_edges ist nicht nötig.
• resume als einziger Einstiegspunkt:Command(resume=...) ist der einzige Wiederaufnahmepfad für interrupt() — es unterstützt sowohl einen einzelnen Wert Command(resume="answer"), der automatisch an den einzigen ausstehenden Interrupt geroutet wird, als auch ein dict Command(resume={interrupt_id: value, ...}) zur gleichzeitigen Wiederaufnahme mehrerer Interrupts.
• update zusammen mit Reducer:Command(update={"messages": [msg]}) läuft weiterhin über den Reducer des Kanals — der messages-Kanal verwendet operator.add, dieses update ist also append und kein overwrite. Das Overwrite-Wrapper umgeht den Reducer(Overwrite:938)` und geht über den Direktschreibpfad des BinaryOperatorAggregate-Kanals.

Schlüsseldateien

• Command class:759 — @dataclass class Command(graph, update, resume, goto), generisch über N (Typ der goto-Knotennamen).
• Command fields:780 — Definitionen und Docstrings der Felder graph / update / resume / goto.
• _update_as_tuples:791 — zerlegt update in [(channel, value), ...]; unterstützt dict, list[tuple] und annotated dataclass.
• map_command:56 — Kernfunktion, die Command in eine Sequenz von pending writes zerlegt.
• PARENT guard:58 — wirft «kein Elterngraph»-Fehler, wenn cmd.graph == Command.PARENT.
• goto as sends:62 — cmd.goto wird in eine Liste von Send zerlegt; Send wird in den TASKS-Kanal geschrieben, str in branch:to:<node> und löst START aus.
• resume write:72 — cmd.resume wird in den RESUME-Kanal geschrieben und beim Wiedereintritt von interrupt() gelesen.
• update writes:74 — cmd.update wird in (channel, value)-Paare zerlegt, in den entsprechenden Kanal geschrieben und über den Reducer verarbeitet.
• Command resume mapping:904 — Wenn der Elterngraph Command(resume=...) erhält, routet er resume an den Eingang des entsprechenden pending interrupt.
• Overwrite:938 — Overwrite(value=...)-Wrapper; umgeht bei Verwendung mit einem BinaryOperatorAggregate-Kanal den Reducer und schreibt direkt.

Datenfluss

map_command ist der Kern der Command-Auflösung(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 ist immer NULL_TASK_ID — diese writes gehören keiner konkreten Aufgabe an, sondern sind graph-level «Outbound»: goto's Send geht in den TASKS-Kanal (die nächste Runde prepare_next_tasks sieht es und schedult es als PUSH-Aufgabe), goto's str geht in den branch:to:<node>-Kanal (ein spezialkanal für conditional edges; START schreiben bedeutet «den Eingang dieses Knotens auslösen»); resume geht in den RESUME-Kanal und wird beim Wiedereintritt des Knotens von interrupt() konsumiert; jedes (k, v) von update geht in den entsprechenden Zustandskanal und wird vom Reducer zusammengeführt.

Wenn Command als Eingabe für graph.stream(...) dient, wandelt PregelLoop._first(Command resume mapping:904)es zuerst in pending writes um, gruppiert diese dann nach task_id und schreibt sie incheckpoint_pending_writes; anschließend verarbeitet prepare_next_taskssie auf natürliche Weise. WennCommandals Rückgabewert eines Knotens dient, hängtattach_nodeden_get_updates-Mapper in die writersdes Knotens(<SrcLink path="libs/langgraph/langgraph/graph/state.py" lines="1431" label="attach_node"/>), extrahiert das update ausCommand._update_as_tuples()und schreibt es in die Zustandskanäle; goto und resume werden über den_control_branch-Mapper in den branch:to:-Kanal bzw. nach RESUME` geroutet.

Grenzen und Fehler

• Command.PARENT fehlerhaft im Root-Graph:map_command führt direkt raise InvalidUpdateError("There is no parent graph") aus(PARENT guard:58)—Command.PARENT` ist nur in Untergraph-Knoten sinnvoll; ein Aufruf im Root-Graph fehler direkt.
• goto-Typeinschränkung:goto akzeptiert nur Send / str oder eine Sequenz davon; andere Typen führen zu raise TypeError(goto type:69)` — direkte Übergabe von int, dict usw. ist nicht erlaubt.
• resume ohne id erzwingt einzelnen pending:Command(resume="single") mit mehreren pending interrupts führt zu raise RuntimeError(multi interrupt error:917); in diesem Fall muss Command(resume={id: value, ...}) verwendet werden.
• update über Reducer nicht direkt schreibbar:BinaryOperatorAggregate-Kanäle (wie Annotated[list, operator.add]) gehen standardmäßig über den Reducer, mehrere updates werden zusammengeführt; um den Reducer zu umgehen, Overwrite(value=...)(Overwrite:938)verwenden — aber mehrfachesOverwriteauf denselben Kanal im selben Superstep (superstep) führt zuInvalidUpdateError`.
• Command und entrypoint.final sind gegenseitig exklusiv:@entrypoint-Funktionen geben entrypoint.final(value=..., save=...) zurück und verwenden intern den END- / PREVIOUS-Kanal; Command geht über den Steuerkanal — beides kann nicht gleichzeitig verwendet werden, eine entrypoint-Funktion sollte entweder Command oder final zurückgeben, nicht mischen.
• Leeres Command fehlerhaft:Wenn alle drei Felder von Command() None / () sind, liefert map_command nichts; _first erkennt leere writes und führt raise EmptyInputError("Received empty Command input") aus(empty command:928).

Zusammenfassung

Command ist das Drei-in-eins-Primitiv für resume / goto / update; über map_command wird es in Kanal-writes zerlegt, die in der nächsten Runde von prepare_next_tasks verarbeitet werden. Es vereint conditional edges, interrupt resume und Zustandsaktualisierungen in einem einzigen Rückgabewert. Das zugehörige interrupt() siehe /interrupt/interrupt, die Send-Ausfächerungs-Semantik siehe /subgraph/send-command. Siehe offizielle Dokumentation: LangGraph 文档 · README.