@entrypoint / @task:函数式 API

源码版本1.2.9

职责

@entrypoint 和 @task 是 LangGraph 函数式 API 的两块基石,在 libs/langgraph/langgraph/func/__init__.py 里定义。@entrypoint 把一个普通 Python 函数(同步或异步)包成一个 Pregel 实例——返回值就是个编译好的图,可以直接 .invoke / .astream(entrypoint.__call__:516)。@task 把一个函数包成 _TaskFunction(_TaskFunction:59),调用时返回 SyncAsyncFuture,只能在 @entrypoint 或 StateGraph 节点内部调用。

它的位置在 StateGraph 之上,是另一条上层入口——不需要先建 StateGraph、add_node / add_edge / compile,直接写函数就行。底层仍是 Pregel:entrypoint.__call__ 最后 return Pregel(...),把 entrypoint 函数包成单节点图(func.__name__ 是唯一节点名,触发器是 START),节点内执行 entrypoint 函数体,函数返回时把返回值写进 END 通道(Pregel construction:572)。

设计动机

少一层样板:StateGraph 要求显式声明状态 schema、节点函数、边、条件边;@entrypoint 让你只写一个函数,函数体里的 task 调用序列就是图结构。简单工作流(几步调用、串几个工具)用函数式 API 写出来更轻。
保留 Pregel 全部能力:函数式 API 不是阉割版,底层就是 Pregel,所以 checkpointer、store、cache、retry_policy、context_schema、interrupt()、Command(resume=...) 都能用——你在 entrypoint 函数里 interrupt(),跟在 StateGraph 节点里 interrupt() 走的是同一套机制。
entrypoint.final 解耦返回值与持久化值:很多对话型工作流需要「本次返回给调用方 X,但记到 checkpoint 里给下次用的是 Y」(previous 参数读上次 save)。entrypoint.final(value=X, save=Y)(entrypoint.final:477)显式把这两件事拆开,而不是用 dict 约定。
previous 参数模型:entrypoint 函数签名可以加 *, previous: Any = None,下次同一 thread_id 调用时拿到的就是上次 save 的值,等价于一个「跨调用记忆变量」,不需要自己声明 state schema 和 reducer。
任务调用即图边:在 entrypoint 函数体里 task_a(...) 返回 future,.result() / await 拿结果——这个调用关系天然就是一条边,Pregel 把它当成 PUSH 任务塞进 TASKS 通道,下一步 prepare_next_tasks 自然会调度。不需要 add_edge。

关键文件

_TaskFunction:59 — @task 装饰产物,持有 func / retry_policy / cache_policy / timeout。
_TaskFunction.__call__:86 — 调用 task 时走 _call_with_options,从 config 拿 CONFIG_KEY_CALL 回调,把任务塞给当前 runner。
task decorator:110 — @task 装饰器入口,支持 @task / @task(retry_policy=...) 两种形式。
entrypoint class:262 — @entrypoint 装饰器类,final 作为内嵌 dataclass 挂在类上。
entrypoint.__init__:437 — 接收 checkpointer / store / cache / retry_policy / timeout / context_schema,存到 self。
entrypoint.final:477 — final(value=R, save=S) 数据类,return 给调用方 value,checkpoint 存 save。
entrypoint.__call__:516 — 把函数变成 Pregel 实例,真正的转换逻辑在这。
Pregel construction:572 — 构造 Pregel(nodes={func.__name__: PregelNode(bound=bound, triggers=[START], ...)}),channels 是 START / END / PREVIOUS 三个 LastValue。
get_runnable_for_entrypoint:175 — 把 entrypoint 函数包成 RunnableCallable,同步版走 run_in_executor。
get_runnable_for_task:200 — 把 task 函数包成 RunnableSeq(run, ChannelWrite([RETURN])),task 返回值写到 RETURN 通道。

数据流

@entrypoint 把函数变 Pregel 的核心是构造一个只有三个通道的极简图:

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 是 EphemeralValue(每步开始重置),END 和 PREVIOUS 都是 LastValue(只保留最新)。entrypoint 函数的返回值被两个 mapper 拆开:_pluck_return_value 把 entrypoint.final.value(或裸值)写到 END 给调用方拿;_pluck_save_value 把 entrypoint.final.save(或裸值)写到 PREVIOUS 通道,作为下次调用的 previous(pluck mappers:547)。stream_mode="updates" 是函数式 API 的默认值(不是 values),因为 entrypoint 内部可能有多个 task,updates 更能反映中间进度。

task 调用链:task_fn(arg) → _TaskFunction.__call__ → _call_with_options(_call_with_options:276) → 从 config[CONF][CONFIG_KEY_CALL] 取 impl 回调 → impl(func, args, kwargs, retry_policy=..., callbacks=..., timeout=...) 返回 SyncAsyncFuture。这个 impl 是 PregelRunner.tick / atick 在调度任务时通过 partial(...) 注入的(CONFIG_KEY_CALL inject:211)——所以 task 不自己跑,而是把函数和参数交给当前 runner,runner 决定何时并发执行,完成后通过 future 把结果回传给调用方。

边界与失败

不支持生成器:entrypoint 函数不能是 def gen / async def gen(带 yield),__call__ 开头显式 raise NotImplementedError(no generators:525)。要流式输出用 StreamWriter(stream_mode="custom")而不是 yield。
task 不能在外部调用:从 @entrypoint / StateGraph 节点之外调 task_fn(...) 会失败,因为 get_config()[CONF][CONFIG_KEY_CALL] 不存在。task 的存在依赖当前运行期的 runner 上下文。
同步 task 不能配 timeout:task(timeout=...) 只对异步函数有效,同步函数会 sync_timeout_unsupported(sync timeout unsupported:237)——同步 task 跑在 executor 线程里,Python 没法在进程内安全取消。
previous 默认 None:首次调用时 previous 是 None,需要 entrypoint 函数自己处理「没有上次值」的分支;如果 entrypoint.final 没用,返回值会同时被写到 END 和 PREVIOUS,下次 previous 拿到的就是上次返回值本身。
final 必须正确参数化:返回类型注解 -> entrypoint.final[int, str] 必须两个类型参数都给,只给一个或不给会在 __call__ 里报 TypeError(final param check:562)。
依赖 checkpointer:entrypoint(checkpointer=...) 没配 checkpointer 时,interrupt() 和 previous 跨调用记忆都不能用——前者会抛 RuntimeError,后者每次都是 None。

小结

函数式 API 是 StateGraph 的语法糖,底层仍是 Pregel 单节点图 + TASKS 通道 + runner 注入的 CONFIG_KEY_CALL。entrypoint 函数体里的 task(...) 调用序列就是隐式的图结构。并发细节看 /func/concurrency,中断恢复看 /interrupt/interrupt。对照官方资料:LangGraph 文档 · README。

@entrypoint / @task:函数式 API ​

职责 ​

设计动机 ​

关键文件 ​

数据流 ​

边界与失败 ​

小结 ​

@entrypoint / @task:函数式 API

职责

设计动机

关键文件

数据流

边界与失败

小结