Command : primitif trois-en-un resume / goto / update

源码版本1.2.9

Responsabilités

Command est le « primitif de contrôle trans-graphe » de LangGraph, défini dans libs/langgraph/langgraph/types.py(Command class:759). Une instance de Command peut porter simultanément trois sémantiques — resume (reprendre l'exécution depuis une interruption), goto (indiquer vers quel nœud sauter ensuite), update (écrire une valeur dans les canaux d'état du graphe courant). Une fonction de nœud peut renvoyer Command(...), ou l'appelant peut le fournir comme entrée de graph.invoke(Command(...)) ; Pregel le décompose en pending writes écrits dans les canaux.

Sa position se situe entre le nœud et la boucle principale Pregel : le nœud produit un Command, Pregel utilise map_command(map_command:56) pour décomposer ses champs en triplets (task_id, channel, value) écrits dans les pending writes ; au tour suivant, prepare_next_tasks les voit naturellement et ordonnance en conséquence — goto devient un Send écrit dans le canal TASKS, update écrit directement dans le canal correspondant, resume passe par le canal RESUME pour être traité par interrupt().

Motivation de conception

• Expression unifiée des « effets de bord » : en plus d'écrire dans l'état, un nœud peut vouloir « sauter vers un nœud suivant non par défaut », « transmettre un message au graphe père » ou « reprendre une interruption » — auparavant dispersées dans plusieurs API, Command les regroupe en un type de retour unique ; la signature -> Command[Literal["node_a", "node_b"]] permet aussi à la compilation de valider les cibles de saut autorisées.
• Command.PARENT franchit la hiérarchie de graphes : un nœud dans un sous-graphe peut return Command(graph=Command.PARENT, update=...) pour écrire dans l'état du graphe père — map_command lève raise InvalidUpdateError("There is no parent graph") si graph == "__parent__"(PARENT guard:58) ; l'écriture effective dans le père est traitée dans le _first` du graphe père à la réception des writes remontés par le sous-graphe.
• goto accepte Send : Command(goto=Send("node", arg)) équivaut à déclencher un fan-out (dispersion) via une tâche PUSH ; goto="node" équivaut à goto=Send("node", START) (déclenche le PULL de ce nœud). Ainsi, Command incorpore la sémantique des conditional edges dans la valeur de retour du nœud, sans avoir à appeler add_conditional_edges séparément.
• resume à entrée unique : Command(resume=...) est le seul chemin de reprise pour interrupt() — il accepte aussi bien une valeur unique Command(resume="answer") qui se route automatiquement vers le seul interrupt en attente, qu'un dict Command(resume={interrupt_id: value, ...}) pour reprendre plusieurs interrupts simultanément.
• update s'appuie sur le reducer : Command(update={"messages": [msg]}) passe par le reducer du canal — si le canal messages est configuré avec operator.add, cette update est un append et non un écrasement. Le wrapper Overwrite permet de court-circuiter le reducer(Overwrite:938), via le chemin d'écriture directe du canal BinaryOperatorAggregate.

Fichiers clés

• Command class:759 — @dataclass class Command(graph, update, resume, goto), le générique N représente le type du nom de nœud goto.
• Command fields:780 — définitions et docstrings des champs graph / update / resume / goto.
• _update_as_tuples:791 — décompose update en [(channel, value), ...] ; supporte dict, list[tuple] et annotated dataclass.
• map_command:56 — fonction centrale qui décompose Command en une séquence de pending writes.
• PARENT guard:58 — erreur « pas de graphe père » quand cmd.graph == Command.PARENT.
• goto as sends:62 — cmd.goto décomposé en liste de Send ; les Send sont écrits dans le canal TASKS, les str dans branch:to:<node> pour déclencher START.
• resume write:72 — cmd.resume écrit dans le canal RESUME, lu par interrupt() à la ré-entrée.
• update writes:74 — cmd.update décomposé en (channel, value) écrit dans le canal correspondant, via le reducer.
• Command resume mapping:904 — entrée où le graphe père, après avoir reçu Command(resume=...), route le resume vers le pending interrupt correspondant.
• Overwrite:938 — wrapper Overwrite(value=...), utilisé avec un canal BinaryOperatorAggregate pour écrire en direct en contournant le reducer.

Flux de données

map_command est le cœur du décodage de 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)

Le task_id est toujours NULL_TASK_ID — ces writes n'appartiennent à aucune tâche spécifique, ce sont des « sorties » au niveau du graphe : le Send du goto va dans le canal TASKS (prepare_next_tasks le verra au tour suivant et l'ordonnancera comme une tâche PUSH) ; le str du goto passe par le canal branch:to:<node> (canal dédié aux conditional edges, écrire START signifie « déclencher l'entrée de ce nœud ») ; resume va dans le canal RESUME, consommé par interrupt() à la ré-entrée du nœud ; chaque (k, v) de update va dans le canal d'état correspondant, fusionné par le reducer.

Quand Command est l'entrée de graph.stream(...), PregelLoop._first(Command resume mapping:904)le convertit d'abord en pending writes, puis les regroupe par task_id pour écrire danscheckpoint_pending_writes, et prepare_next_tasksles traite naturellement. QuandCommandest la valeur de retour d'un nœud,attach_nodeaccroche un mapper_get_updatesauxwritersdu nœud(<SrcLink path="libs/langgraph/langgraph/graph/state.py" lines="1431" label="attach_node"/>), qui extrait l'update depuisCommand._update_as_tuples()pour l'écrire dans les canaux d'état, tandis que goto et resume sont routés via le mapper_control_branchvers le canalbranch:to:etRESUME`.

Limites et échecs

• Command.PARENT échoue sur le graphe racine : map_command lève directement raise InvalidUpdateError("There is no parent graph")(PARENT guard:58) — Command.PARENT n'a de sens que dans un nœud de sous-graphe ; un appel depuis le graphe racine échoue.
• Restriction de type pour goto : goto accepte uniquement Send / str ou une séquence de ceux-ci ; tout autre type lève raise TypeError(goto type:69) — les entiers, dicts, etc. ne sont pas autorisés.
• resume exige un seul pending si l'id n'est pas spécifié : Command(resume="single") lève RuntimeError s'il y a plusieurs pending interrupts(multi interrupt error:917) ; il faut alors Command(resume={id: value, ...}).
• update passe par le reducer, pas d'écriture directe : le canal BinaryOperatorAggregate (par ex. Annotated[list, operator.add]) passe par défaut par le reducer, et plusieurs updates sont fusionnés ; pour contourner le reducer, utiliser Overwrite(value=...)(Overwrite:938), mais plusieurs Overwrite sur un même canal au cours d'un même superpas lèvent une InvalidUpdateError.
• Command mutuellement exclusif avec entrypoint.final : une fonction @entrypoint renvoie entrypoint.final(value=..., save=...), qui passe en interne par les canaux END / PREVIOUS ; renvoyer Command passe par les canaux de contrôle — les deux ne peuvent pas être utilisés simultanément, la fonction entrypoint doit renvoyer soit Command soit final, pas mélanger.
• Command vide échoue : quand les trois champs de Command() sont tous None / (), map_command ne yield rien, et _first détectant des writes vides lève raise EmptyInputError("Received empty Command input")(empty command:928).

Résumé

Command est le primitif trois-en-un resume / goto / update, décomposé en writes de canaux via map_command, puis naturellement traité au tour suivant par prepare_next_tasks. Il unifie conditional edges, interrupt resume et mise à jour d'état en une seule valeur de retour. Le interrupt() associé se voit dans /interrupt/interrupt, et la sémantique de fan-out de Send dans /subgraph/send-command. Voir la documentation officielle : LangGraph 文档 · README