Command: primitiva triple resume / goto / update

源码版本1.2.9

Responsabilidades

Command es la «primitiva de control entre grafos» de LangGraph, definida en libs/langgraph/langgraph/types.py(Command class:759). Una instancia de Command puede portar simultáneamente tres semánticas — resume (reanudar la ejecución desde una interrupción), goto (indicar a qué nodo saltar a continuación) y update (escribir valores en los canales de estado del grafo actual). La función nodo devuelve Command(...), o bien el llamador lo pasa como entrada de graph.invoke(Command(...)), y Pregel lo descompone en pending writes que se escriben en los canales.

Su posición está entre el nodo y el bucle principal de Pregel: el nodo produce Command, y Pregel usa map_command(map_command:56) para descomponer sus campos en tuplas (task_id, channel, value) que se escriben como pending writes; la siguiente ronda de prepare_next_tasks los ve de forma natural y programa según ellos — goto se convierte en un Send escrito en el canal TASKS, update escribe directamente el canal correspondiente, y resume va por el canal RESUME para que lo consuma interrupt().

Motivación de diseño

• Expresión unificada de «efectos»: además de escribir valores en el estado, un nodo puede querer «saltar a un nodo no predeterminado», «enviar un mensaje al grafo padre» o «reanudar una interrupción» — antes estas operaciones eran APIs dispersas; Command las empaqueta en un único tipo de retorno, y la firma -> Command[Literal["node_a", "node_b"]] permite incluso validar en compilación los destinos de salto válidos.
• Command.PARENT entre niveles de grafo: un nodo dentro de un subgrafo puede hacer return Command(graph=Command.PARENT, update=...) para escribir actualizaciones en el estado del grafo padre — map_command, al ver graph == "__parent__"(PARENT guard:58)raise InvalidUpdateError("There is no parent graph"); el verdadero alcance al padre ocurre en el _first del grafo padre, que procesa los writes reportados por el subgrafo.
• goto acepta Send: Command(goto=Send("node", arg)) equivale a lanzar un task PUSH de tipo fan-out; goto="node" equivale a goto=Send("node", START) (dispara el PULL de ese nodo). Por tanto Command integra la capacidad de las conditional edges dentro del valor de retorno del nodo, sin necesidad de escribir add_conditional_edges por separado.
• resume como única entrada: Command(resume=...) es la única vía de reanudación de interrupt() — soporta tanto un valor único Command(resume="answer") que se enruta automáticamente al único pending interrupt, como un dict Command(resume={interrupt_id: value, ...}) para reanudar varios interrupts concurrentes.
• update con reducer: Command(update={"messages": [msg]}) pasa por el reducer del canal — el canal messages configurado con operator.add hace que este update sea un append, no una sobrescritura. El envoltorio Overwrite permite saltarse el reducer(Overwrite:938), yendo por la ruta de escritura directa del canal BinaryOperatorAggregate.

Archivos clave

• Command class:759 — @dataclass class Command(graph, update, resume, goto), el genérico N representa el tipo del nombre de nodo de goto.
• Command fields:780 — definición y docstring de los campos graph / update / resume / goto.
• _update_as_tuples:791 — descompone update en [(channel, value), ...], soportando dict, list[tuple] y annotated dataclass.
• map_command:56 — función central que descompone Command en una secuencia de pending writes.
• PARENT guard:58 — cuando cmd.graph == Command.PARENT se lanza el error «no hay grafo padre».
• goto as sends:62 — descompone cmd.goto en una lista de Send; Send se escribe en el canal TASKS, str se escribe en branch:to:<node> para disparar START.
• resume write:72 — cmd.resume se escribe en el canal RESUME, para que lo lea interrupt() al reentrar.
• update writes:74 — cmd.update se descompone en (channel, value) escrito en el canal correspondiente, pasando por el reducer.
• Command resume mapping:904 — entrada donde el grafo padre, al recibir Command(resume=...), enruta resume al pending interrupt correspondiente.
• Overwrite:938 — envoltorio Overwrite(value=...), que al combinarse con un canal BinaryOperatorAggregate ignora el reducer y escribe directamente.

Flujo de datos

map_command es el núcleo del parsing 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)

El task_id es siempre NULL_TASK_ID — estos writes no pertenecen a ningún task concreto, sino que son «envíos» a nivel de grafo: el Send de goto entra en el canal TASKS (la siguiente ronda de prepare_next_tasks lo ve y lo programa como task PUSH), el str de goto va al canal branch:to:<node> (canal específico de conditional edge, escribir START significa «disparar la entrada de ese nodo»); resume entra en el canal RESUME, para que lo consuma interrupt() al reentrar al nodo; cada (k, v) de update entra en el canal de estado correspondiente y se fusiona mediante el reducer.

Cuando Command se pasa como entrada de graph.stream(...), PregelLoop._first(Command resume mapping:904)primero lo convierte en pending writes, los agrupa por task_id y los escribe en checkpoint_pending_writes; después prepare_next_taskslos procesa de forma natural. CuandoCommandes el valor de retorno de un nodo,attach_nodecoloca un mapper_get_updatesen loswritersdel nodo(<SrcLink path="libs/langgraph/langgraph/graph/state.py" lines="1431" label="attach_node"/>), que extrae el update deCommand._update_as_tuples()y lo escribe en los canales de estado, mientras que goto y resume se enrutan mediante el mapper_control_branchal canalbranch:to:y aRESUME`.

Límites y fallos

• Command.PARENT falla en el grafo raíz: map_command lanza directamente raise InvalidUpdateError("There is no parent graph")(PARENT guard:58) — Command.PARENT sólo tiene sentido dentro de un nodo de subgrafo; invocarlo en el grafo raíz falla al instante.
• Restricciones de tipo en goto: goto sólo acepta Send / str o secuencias de ellos; cualquier otro tipo lanza raise TypeError(goto type:69) — no se permite pasar enteros, dicts, etc.
• resume sin id fuerza un único pending: Command(resume="single") con varios pending interrupts lanza raise RuntimeError(multi interrupt error:917); debe cambiarse a Command(resume={id: value, ...}).
• update pasa por reducer, no escribe directamente: los canales BinaryOperatorAggregate (por ejemplo Annotated[list, operator.add]) pasan por defecto por el reducer y fusionan múltiples updates; para saltar el reducer se usa Overwrite(value=...)(Overwrite:938), pero varias llamadas Overwrite al mismo canal en el mismo superstep lanzan InvalidUpdateError.
• Command y entrypoint.final son mutuamente excluyentes: la función @entrypoint devuelve entrypoint.final(value=..., save=...), que internamente va por los canales END / PREVIOUS; devolver Command va por canales de control — no se pueden usar a la vez; la función entrypoint debería devolver Command o final, sin mezclar.
• Command vacío lanza error: cuando los tres campos de Command() son None / (), map_command no produce nada, y _first al ver writes vacíos lanza raise EmptyInputError("Received empty Command input")(empty command:928).

Resumen

Command es la primitiva triple resume / goto / update; a través de map_command se descompone en writes a canales y la siguiente ronda de prepare_next_tasks los procesa de forma natural. Unifica conditional edges, interrupt resume y actualización de estado en un único valor de retorno. La primitiva interrupt() complementaria se trata en /interrupt/interrupt; la semántica de扇出 con Send en /subgraph/send-command. Véase la documentación oficial: Documentación de LangGraph · README