@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 要求顯式宣告 state 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

職責

設計動機

關鍵檔案

資料流

邊界與失敗

小結