@entrypoint / @task: functional API

源码版本1.2.9

Responsibilities

@entrypoint and @task are the two cornerstones of LangGraph's functional API, defined in libs/langgraph/langgraph/func/__init__.py. @entrypoint wraps an ordinary Python function (sync or async) as a Pregel instance — the return value is a compiled graph that can be .invoked / .astreamed directly (entrypoint.__call__:516). @task wraps a function as a _TaskFunction (_TaskFunction:59); calling it returns a SyncAsyncFuture, and it can only be called inside an @entrypoint or a StateGraph node.

It sits above StateGraph as an alternative upper-layer entrypoint — no need to build a StateGraph, call add_node / add_edge / compile; just write a function. The underlying engine is still Pregel: entrypoint.__call__ ultimately does return Pregel(...), wrapping the entrypoint function as a single-node graph (func.__name__ is the only node name, the trigger is START); the node executes the entrypoint function body, and when the function returns it writes the return value into the END channel (Pregel construction:572).

Design motivation

• Less boilerplate: StateGraph requires explicitly declaring a state schema, node functions, edges, and conditional edges. @entrypoint lets you write a single function; the sequence of task calls inside the body is the graph structure. For simple workflows (a few calls, chaining a couple of tools) the functional API is lighter.
• Preserves all of Pregel: the functional API is not a stripped-down version — the underlying engine is Pregel, so checkpointer, store, cache, retry_policy, context_schema, interrupt(), Command(resume=...) all work. Calling interrupt() inside an entrypoint function goes through the same mechanism as calling it inside a StateGraph node.
• entrypoint.final decouples return value from persisted value: many conversational workflows need "return X to the caller this turn, but persist Y into the checkpoint for next time" (the previous parameter reads the last save). entrypoint.final(value=X, save=Y) (entrypoint.final:477) explicitly splits these two concerns, instead of relying on dict conventions.
• previous parameter model: the entrypoint function signature may include *, previous: Any = None; the next call with the same thread_id receives the value saved last time, equivalent to a "cross-call memory variable" without declaring a state schema and reducer.
• Task calls are graph edges: inside an entrypoint body task_a(...) returns a future; .result() / await retrieves the result — this call relationship is naturally an edge. Pregel treats it as a PUSH task pushed into the TASKS channel; the next prepare_next_tasks schedules it. No add_edge needed.

Key files

• _TaskFunction:59 — product of the @task decorator; holds func / retry_policy / cache_policy / timeout.
• _TaskFunction.__call__:86 — when a task is called, routes through _call_with_options, which pulls the CONFIG_KEY_CALL callback from config and hands the task to the current runner.
• task decorator:110 — entry point of the @task decorator; supports both @task and @task(retry_policy=...) forms.
• entrypoint class:262 — the @entrypoint decorator class; final is attached as a nested dataclass on the class.
• entrypoint.__init__:437 — accepts checkpointer / store / cache / retry_policy / timeout / context_schema and stores them on self.
• entrypoint.final:477 — final(value=R, save=S) dataclass; returns value to the caller, persists save to the checkpoint.
• entrypoint.__call__:516 — turns the function into a Pregel instance; the actual conversion logic lives here.
• Pregel construction:572 — constructs Pregel(nodes={func.__name__: PregelNode(bound=bound, triggers=[START], ...)}); the channels are START / END / PREVIOUS, all LastValue.
• get_runnable_for_entrypoint:175 — wraps the entrypoint function as a RunnableCallable; the sync version goes through run_in_executor.
• get_runnable_for_task:200 — wraps the task function as RunnableSeq(run, ChannelWrite([RETURN])); the task return value is written to the RETURN channel.

Data flow

The core of @entrypoint turning a function into Pregel is constructing a minimal graph with only three channels:

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 is an EphemeralValue (reset at the start of each step); END and PREVIOUS are both LastValue (keep only the latest). The entrypoint function's return value is split by two mappers: _pluck_return_value writes entrypoint.final.value (or the bare value) to END for the caller to read; _pluck_save_value writes entrypoint.final.save (or the bare value) to PREVIOUS as the next call's previous (pluck mappers:547). stream_mode="updates" is the default for the functional API (not values), because an entrypoint may contain multiple tasks and updates better reflects intermediate progress.

Task call chain: task_fn(arg) → _TaskFunction.__call__ → _call_with_options (_call_with_options:276) → reads impl from config[CONF][CONFIG_KEY_CALL] → impl(func, args, kwargs, retry_policy=..., callbacks=..., timeout=...) returns a SyncAsyncFuture. This impl is injected by PregelRunner.tick / atick via partial(...) while scheduling tasks (CONFIG_KEY_CALL inject:211) — so a task does not run itself; it hands its function and arguments to the current runner, which decides when to run it concurrently and returns the result via a future.

Boundaries and failure modes

• Generators not supported: the entrypoint function cannot be def gen / async def gen (with yield); __call__ explicitly raise NotImplementedError at the top (no generators:525). For streaming output use StreamWriter (stream_mode="custom") instead of yield.
• Tasks cannot be called externally: calling task_fn(...) from outside an @entrypoint / StateGraph node fails, because get_config()[CONF][CONFIG_KEY_CALL] does not exist. A task's existence depends on the current runtime runner context.
• Sync tasks cannot take a timeout: task(timeout=...) only works for async functions; a sync function raises sync_timeout_unsupported (sync timeout unsupported:237) — a sync task runs in an executor thread, and Python has no safe way to cancel it in-process.
• previous defaults to None: on the first call previous is None; the entrypoint function must handle the "no previous value" branch itself. If entrypoint.final is not used, the return value is written to both END and PREVIOUS, so the next call's previous receives the previous return value itself.
• final must be properly parameterized: the return type annotation -> entrypoint.final[int, str] must provide both type parameters; providing only one or none raises TypeError in __call__ (final param check:562).
• Depends on a checkpointer: when entrypoint(checkpointer=...) is not configured with a checkpointer, neither interrupt() nor previous cross-call memory works — the former raises RuntimeError, the latter is None every time.

Summary

The functional API is syntactic sugar over StateGraph; underneath it is still a Pregel single-node graph + the TASKS channel + a runner-injected CONFIG_KEY_CALL. The sequence of task(...) calls inside the entrypoint body is the implicit graph structure. For concurrency details see /func/concurrency; for interrupt/resume see /interrupt/interrupt. See official docs: LangGraph · README.