@entrypoint / @task: API funcional

源码版本1.2.9

Responsabilidades

@entrypoint y @task son las dos piedras angulares de la API funcional de LangGraph, definidas en libs/langgraph/langgraph/func/__init__.py. @entrypoint envuelve una función Python normal (síncrona o asíncrona) y la convierte en una instancia Pregel —el valor de retorno es un grafo compilado, listo para .invoke / .astream (entrypoint.__call__:516). @task envuelve una función en un _TaskFunction (_TaskFunction:59) y, al llamarla, devuelve un SyncAsyncFuture; sólo se puede invocar dentro de un @entrypoint o de un nodo StateGraph.

Se sitúa por encima de StateGraph: es otra entrada de nivel superior —no hace falta construir primero un StateGraph, ni add_node / add_edge / compile, basta con escribir funciones. Por debajo sigue siendo Pregel: entrypoint.__call__ termina con return Pregel(...), envolviendo la función entrypoint como un grafo de un solo nodo (func.__name__ es el único nombre de nodo, con trigger START); al ejecutarse el cuerpo del entrypoint y retornar, el valor de retorno se escribe en el canal END (Pregel construction:572).

Motivación de diseño

Una capa menos de boilerplate: StateGraph exige declarar explícitamente el schema de estado, las funciones de nodo, los edges y los edges condicionales; con @entrypoint basta una función, y la secuencia de llamadas a task en su cuerpo es la estructura del grafo. Para flujos de trabajo sencillos (unas cuantas llamadas encadenadas, un par de herramientas), la API funcional es más ligera.
Conserva todas las capacidades de Pregel: la API funcional no es una versión recortada —por debajo hay un Pregel, así que checkpointer, store, cache, retry_policy, context_schema, interrupt() y Command(resume=...) están disponibles. Hacer interrupt() dentro de un entrypoint recorre la misma maquinaria que hacerlo dentro de un nodo StateGraph.
entrypoint.final desacopla valor de retorno y valor persistido: muchos flujos conversacionales necesitan «devolver X al llamador, pero persistir Y en el checkpoint para la próxima vez» (el parámetro previous lee lo guardado). entrypoint.final(value=X, save=Y) (entrypoint.final:477) separa explícitamente estas dos cosas, sin recurrir a convenciones sobre un dict.
Modelo de parámetro previous: la firma del entrypoint puede llevar *, previous: Any = None; la próxima vez que se llame con el mismo thread_id, previous contendrá lo que se guardó en la ejecución anterior —equivale a una «variable con memoria entre llamadas», sin necesidad de declarar schema de estado ni reducer.
Una llamada a task es un edge del grafo: dentro del cuerpo del entrypoint, task_a(...) devuelve un future y .result() / await obtiene el resultado —esta relación de llamada es naturalmente un edge; Pregel la trata como una tarea PUSH y la mete en el canal TASKS, y el siguiente prepare_next_tasks la schedulea. No hace falta add_edge.

Archivos clave

_TaskFunction:59 — el producto de @task; mantiene func / retry_policy / cache_policy / timeout.
_TaskFunction.__call__:86 — al llamar a una task, delega en _call_with_options, que toma del config el callback CONFIG_KEY_CALL y le entrega la tarea al runner actual.
task decorator:110 — el decorador @task, admite @task y @task(retry_policy=...).
entrypoint class:262 — la clase del decorador @entrypoint, con final como dataclass anidada colgada de la clase.
entrypoint.__init__:437 — recibe checkpointer / store / cache / retry_policy / timeout / context_schema y los guarda en self.
entrypoint.final:477 — la data class final(value=R, save=S): devuelve value al llamador y persiste save en el checkpoint.
entrypoint.__call__:516 — convierte la función en una instancia Pregel; la lógica de transformación vive aquí.
Pregel construction:572 — construye Pregel(nodes={func.__name__: PregelNode(bound=bound, triggers=[START], ...)}); los canales son START / END / PREVIOUS, tres LastValue.
get_runnable_for_entrypoint:175 — envuelve la función entrypoint en un RunnableCallable; la versión síncrona usa run_in_executor.
get_runnable_for_task:200 — envuelve la función task en RunnableSeq(run, ChannelWrite([RETURN])), de modo que el valor de retorno de la task se escribe en el canal RETURN.

Flujo de datos

El núcleo de @entrypoint para convertir una función en Pregel es construir un grafo mínimo con tres canales:

python

graph: Pregel[Any, ContextT, Any, Any] = Pregel(
    nodes={
        func.__name__: PregelNode(
            bound=bound,
            triggers=[START],
            channels=START,
            timeout=self.timeout,
            writers=[
                ChannelWrite(
                    [
                        ChannelWriteEntry(END, mapper=_pluck_return_value),
                        ChannelWriteEntry(PREVIOUS, mapper=_pluck_save_value),
                    ]
                )
            ],
        )
    },
    channels={
        START: EphemeralValue(input_type),
        END: LastValue(output_type, END),
        PREVIOUS: LastValue(save_type, PREVIOUS),
    },
    input_channels=START,
    output_channels=END,
    stream_channels=END,
    stream_mode=stream_mode,
    stream_eager=True,
    checkpointer=self.checkpointer,
    store=self.store,
    cache=self.cache,
    # ...
)

START es un EphemeralValue (se resetea al inicio de cada paso); END y PREVIOUS son LastValue (sólo guardan el último valor). El valor de retorno del entrypoint se descompone con dos mappers: _pluck_return_value escribe en END el entrypoint.final.value (o el valor desnudo), para que el llamador lo recoja; _pluck_save_value escribe en PREVIOUS el entrypoint.final.save (o el valor desnudo), como previous para la próxima llamada (pluck mappers:547). stream_mode="updates" es el valor por defecto de la API funcional (no values), porque dentro de un entrypoint puede haber varias tasks y updates refleja mejor el progreso intermedio.

Cadena de llamada a task: task_fn(arg) → _TaskFunction.__call__ → _call_with_options (_call_with_options:276) → toma el callback impl desde config[CONF][CONFIG_KEY_CALL] → impl(func, args, kwargs, retry_policy=..., callbacks=..., timeout=...) devuelve un SyncAsyncFuture. Ese impl lo inyecta PregelRunner.tick / atick mediante partial(...) al schedulear la tarea (CONFIG_KEY_CALL inject:211) —así, la task no se ejecuta por sí misma: entrega la función y los argumentos al runner actual, que decide cuándo ejecutarla concurrentemente y, al terminar, devuelve el resultado al llamador vía future.

Límites y fallos

No soporta generadores: la función entrypoint no puede ser def gen / async def gen (con yield); __call__ lanza NotImplementedError al inicio de forma explícita (no generators:525). Para stream, usar StreamWriter (stream_mode="custom"), no yield.
Una task no se puede llamar desde fuera: llamar task_fn(...) fuera de un nodo @entrypoint / StateGraph falla, porque get_config()[CONF][CONFIG_KEY_CALL] no existe. La task depende del contexto de runner en runtime.
Las tasks síncronas no aceptan timeout: task(timeout=...) sólo funciona con funciones asíncronas; con una función síncrona lanza sync_timeout_unsupported (sync timeout unsupported:237) —las tasks síncronas corren en un hilo del executor y Python no puede cancelar de forma segura dentro del proceso.
previous por defecto es None: en la primera llamada, previous es None, y la función entrypoint debe gestionar la rama «no hay valor previo»; si no se usa entrypoint.final, el valor de retorno se escribe a la vez en END y PREVIOUS, así que la próxima vez previous será el valor retornado en la llamada anterior.
final debe estar correctamente parametrizado: la anotación de retorno -> entrypoint.final[int, str] debe llevar ambos parámetros de tipo; si se pasa sólo uno o ninguno, __call__ lanza TypeError (final param check:562).
Depende de checkpointer: si entrypoint(checkpointer=...) no configura uno, interrupt() y la memoria de previous entre llamadas no funcionan —la primera lanza RuntimeError, la segunda siempre devuelve None.

Resumen

La API funcional es azúcar sintáctico sobre StateGraph; por debajo sigue siendo un grafo Pregel de un solo nodo + canal TASKS + CONFIG_KEY_CALL inyectado por el runner. La secuencia de llamadas a task(...) dentro del cuerpo del entrypoint es la estructura implícita del grafo. Para los detalles de concurrencia, véase /func/concurrency; para interrupción y reanudación, /interrupt/interrupt. Véase la documentación oficial: documentación de LangGraph · README

@entrypoint / @task: API funcional ​

Responsabilidades ​

Motivación de diseño ​

Archivos clave ​

Flujo de datos ​

Límites y fallos ​

Resumen ​