@entrypoint / @task:関数型 API

源码版本1.2.9

役割

@entrypoint と @task は LangGraph 関数型 API (FunctionalAPI) の2つの基礎ブロックで、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)。

設計動機

ボイラープレートを1層減らす:StateGraph は状態スキーマ、ノード関数、エッジ、条件エッジの明示的な宣言を要求します。@entrypoint なら関数を1つ書くだけで、関数本体の 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)はこの2つを明示的に分けます。dict の取り決めで代用しません。
previous パラメータモデル:entrypoint 関数のシグネチャには *, previous: Any = None を付けられます。同一 thread_id で次回呼ぶと、前回 save した値が渡されます。状態スキーマや reducer を自前で宣言せずに「呼び出し間で保持される記憶変数」相当になります。
タスク呼び出し即ちグラフのエッジ:entrypoint 関数本体内で task_a(...) が future を返し、.result() / await で結果を取る。この呼び出し関係が自然に1本のエッジになります。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=...) の2形式をサポート。
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) データクラス。呼び出し元には value を返し、checkpoint には save を保存。
entrypoint.__call__:516 — 関数を Pregel インスタンスに変換。変換ロジックの本体はここ。
Pregel construction:572 — Pregel(nodes={func.__name__: PregelNode(bound=bound, triggers=[START], ...)}) を構築。channels は START / END / PREVIOUS の3つの 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 に変換する中核は、3つのチャネルだけを持つ極小グラフの構築です:

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 関数の戻り値は2つの 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)。ストリーム出力には yield ではなく StreamWriter(stream_mode="custom")を使います。
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] は2つの型パラメータを両方与える必要があります。1つだけ、または省略すると __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

役割

設計動機

主要ファイル

データフロー

境界と失敗

まとめ