コンテンツにスキップ

LangGraph Functional API 写経メモ — @task / @entrypoint / interrupt

LangGraph 0.3 で stable 化した Functional API（@task / @entrypoint / previous / interrupt / entrypoint.final）を、StateGraph を一切書かずに「普通の Python 関数にデコレータを被せるだけ」でワークフローを組む練習。 サンプルの並びは README.md、連載との対応は第21回（../../software-design/21/STUDY_NOTES.md）／第22回（../../software-design/22/STUDY_NOTES.md）。 兄弟レクチャーの Graph API 版は ../langgraph_basics/。

0. このフォルダで何が分かるか（30秒サマリ）

Graph API（StateGraph → add_node → add_edge → compile）が「グラフを宣言してからランタイムに渡す」スタイルなのに対し、Functional API は「普通の関数を書いて呼ぶだけ」。制御フロー（if / for / while）を Python の素の構文でそのまま書けるのが最大の違い。にもかかわらず、checkpointer による state 永続化・並列実行・interrupt による中断再開という LangGraph の旨味は全部使える。

やりたいこと Graph API Functional API
ノードを作る add_node("name", fn) @task を関数に被せる
全体を組む StateGraph(State).compile() @entrypoint() を関数に被せる
state を持ち越す State の reducer previous 引数 + checkpointer
並列実行 Send API Future をためてから .result()
人間に承認を取る ノード内 interrupt() @entrypoint 内 interrupt()
分岐・ループ 条件付き edge / Command(goto=) 素の if / for / while

1. 全体像 — サンプル間の関係と学ぶ順序

# ファイル 中心 API 一言で LLM
01 ex01_minimal_task.py @task / @entrypoint / .result() task は Future を返す。.result() で初めて走る ❌
02 ex02_previous_state.py previous + checkpointer return 値が次回の previous に入る（Redux 同型） ❌
03 ex03_parallel_tasks.py Future の fan-out Future を先に集める→後でまとめて .result()＝並列 ❌
04 ex04_task_with_llm.py @task 内 llm.invoke task の中で LLM を叩く。previous で履歴蓄積 ✅
05 ex05_interrupt_hitl.py interrupt / Command(resume=) 停止→人間の入力で再開。承認ループの基本形 ❌
06 ex06_final_separate.py entrypoint.final(value=, save=) 「外向きの戻り値」と「checkpoint に保存する state」を分離 ❌

補足: フォルダには ex01.py / ex02.py という短い同名ファイルもある。これは ex01_minimal_task.py / ex02_previous_state.py の冒頭 docstring を省いた写経の途中成果物（手で打ち直したコピー）で、ロジックは本体と同一。学習対象は番号付きの長い方（ex0N_*.py）。

1-1. @task / @entrypoint の実行モデル（ex01）

@task を被せた関数は呼んでも即実行されない。Future（「あとで結果が手に入る引換券」）が返るだけで、.result() を呼んだ瞬間に LangGraph ランタイムが実体を走らせて値を返す。

flowchart TD
    A["workflow.invoke(5)"] --> B["@entrypoint 関数本体に入る"]
    B --> C["fut1 = double(5)<br/>Future が返る・まだ走らない"]
    C --> D["fut1.result()<br/>★ここで double が実行 → 10"]
    D --> E["add_ten(10).result()<br/>★ここで add_ten が実行 → 20"]
    E --> F["return 20 → invoke 呼び出し側へ"]

1-2. previous による state 持ち越し（ex02・Redux 同型）

@entrypoint(checkpointer=...) を付けると、return した値が thread_id ごとに保存され、次の invoke で previous 引数に注入される。関数は毎回頭から再実行されるが、前回の state が previous で渡ってくるので累積を作れる。

sequenceDiagram
    participant U as 呼び出し側
    participant W as counter(@entrypoint)
    participant C as MemorySaver<br/>(thread=session-A)

    U->>W: invoke(5, thread=session-A)
    Note over W: previous=None → 初期値{count:0}
    W->>C: return {count:5} を保存
    W-->>U: {count:5, history:[5]}

    U->>W: invoke(3, thread=session-A)
    C->>W: previous={count:5} を注入
    Note over W: 5+3=8
    W->>C: return {count:8} を保存
    W-->>U: {count:8, history:[5,8]}

new_state = f(previous_state, input) という形は Redux/Elm の reducer (state, action) => state と完全に同型。戻り値が次の入力になる点を掴むのがこのフォルダの肝。

1-3. interrupt による中断・再開（ex05）

flowchart TD
    A["invoke({seed:7})"] --> B["while not approved"]
    B --> C["propose_value → proposed を作る"]
    C --> D["interrupt(payload)<br/>★ここで停止・payload を invoke 戻り値として返す"]
    D -. プロセスは生きたまま待つ .-> E{"次の invoke で<br/>Command(resume=X)"}
    E -->|resume=retry| B
    E -->|resume=approve| F["approved=True → while を抜ける"]
    F --> G["return state（最終 state を返す）"]

2. サンプル別の要点

ex01 — @task と @entrypoint の最小構成

StateGraph も add_node も add_edge も書かない。デコレータ2種だけでワークフローが動く、という体験が目的。

@task                      # ① 「Future を返すノード」に変換
def double(x: int) -> int:
    return x * 2

@entrypoint()              # ② ワークフロー全体の入口（checkpointer 省略可）
def workflow(initial: int) -> int:
    fut1 = double(initial) # ③ Future が返る・まだ走らない
    doubled = fut1.result()# ④ .result() で初めて実体が走る
    final = add_ten(doubled).result()  # ⑤ 取得→次へ渡す定番イディオム
    return final           # ⑥ invoke の戻り値として呼び出し側へ
行 やってること なぜそうする
① 関数を「遅延実行のノード」に変える task 化すると LangGraph が実行を管理（並列化・retry・checkpoint）できる
③ task を呼ぶ＝Future を作る この時点では計算しない。スケジューリングの余地を残すため
④ .result() で同期 wait ここで初めて double が走る。値が要る所まで遅延される
② @entrypoint() は括弧必須 デコレータファクトリ。checkpointer= などの設定をここに渡す

内部メカニズム: @task の戻り値の Future は、concurrent.futures.Future に似た「結果の引換券」。LangGraph はこの Future の依存関係を見て、独立なものを並列に、checkpointer があれば各 task の結果を保存しながら実行する。ex01 は checkpointer なしなので、ただの遅延評価される関数呼び出しに近い。

ex02 — previous で state を持ち越す

@entrypoint(checkpointer=MemorySaver())     # ① 保存先を渡すと previous が有効化
def counter(increment: int, *, previous: Optional[CounterState] = None) -> CounterState:
    state = previous or {"count": 0, "history": []}   # ② 初回 vs 継続を1行で
    new_count = state["count"] + increment
    new_history = state["history"] + [new_count]      # ③ 新オブジェクトを作る（mutate しない）
    return {"count": new_count, "history": new_history}# ④ return が次回の previous
行 やってること なぜそうする
① checkpointer=MemorySaver() を渡す これがないと previous は常に None。永続化の有無がここで決まる
② previous or {初期値} 初回は previous=None（falsy）→初期値、2回目以降は前回 state。Python の or の短絡を使う定番
③ + [new_count] で新リストを作る 元の history を破壊せず新オブジェクトを返す（immutable パターン。reducer の作法と同じ）
④ return が次の previous 戻り値＝次回入力という reducer 構造

*, 以降は keyword-only 引数で、previous は LangGraph が注入する予約引数。ユーザーが counter.invoke(5, previous=...) のように明示的に渡してはいけない（注入と衝突する）。thread_id（config={"configurable":{"thread_id":...}}）を変えると state は別系統になり、session-A と session-B が独立して進む。

ex03 — Future の fan-out で並列化

@entrypoint()
def serial_workflow(nums):
    results = []
    for n in nums:
        results.append(slow_double(n).result())  # ② 即 .result()＝1個ずつ待つ＝直列
    return results

@entrypoint()
def parallel_workflow(nums):
    futures = [slow_double(n) for n in nums]      # ③ Future を全部ためる（まだ走らない）
    return [f.result() for f in futures]          # ④ まとめて wait＝並列実行

ここが落とし穴: 並列にしたいなら「呼び出し（Future 生成）」と「.result()（wait）」を分ける。ループ内で即 .result() すると、1個投げて待つ→次を投げて待つ…で直列になる（slow_double は1秒 sleep するので 4個で約4秒）。Future を先に全部ためてから一気に .result() すれば、依存のない4つを LangGraph が並列に走らせるので約1秒で済む。Graph API の Send API に相当する fan-out を、素の Python のリスト内包表記で書けるのが Functional API の読みやすさ。

ex04 — @task の中で LLM を呼ぶ

@task
def generate_haiku(theme, feedback_history):
    # feedback_history をプロンプトに織り込む
    response = llm.invoke(prompt)          # task の中で素直に LLM を叩く
    return str(response.content).strip()

@entrypoint(checkpointer=MemorySaver())
def workflow(inputs, *, previous=None):
    state = previous or {..., "feedback_history": []}
    if "feedback" in inputs and inputs["feedback"]:
        state["feedback_history"] = state["feedback_history"] + [inputs["feedback"]]  # 履歴に追記
    haiku = generate_haiku(state["theme"], state["feedback_history"]).result()
    critique = critique_haiku(haiku).result()
    state["haiku"], state["critique"] = haiku, critique
    return state

学ぶポイントは「LLM を呼ぶ＝特別なことは何もない。@task を被せた関数の中で llm.invoke() を呼ぶだけ」。previous でフィードバック履歴を貯め、2回目以降の invoke で渡した feedback を履歴に積み上げて次の生成プロンプトに織り込む。連載第21回の generate_content（フィードバック反復で文章を改善するエージェント）と同じ発想を、最小コードで再現している。

ex05 — interrupt + Command(resume=) で Human-in-the-Loop

while not state["approved"]:
    state["attempts"] += 1
    state["proposed"] = propose_value(state["seed"], state["attempts"]).result()
    user_action = interrupt({                     # ① ここで停止。payload を呼び出し側へ返す
        "question": "この値を採用しますか？", "proposed": state["proposed"], ...
    })
    if user_action == "approve":                  # ③ resume の値が interrupt の戻り値になる
        state["approved"] = True
    elif user_action == "retry":
        continue
行 やってること なぜそうする
① interrupt(payload) を呼ぶ 実行をその場で凍結し、payload を invoke の戻り値として返す。state は checkpointer に保存される
— 1回目の invoke({seed:7}) の戻り 最終 state ではなく interrupt の payload。ここを勘違いしやすい
③ Command(resume="approve") を渡す resume の値が interrupt() の戻り値として注入され、停止点の続きから再開する

内部メカニズム: interrupt() は内部的に特殊な例外（GraphInterrupt）を投げてランタイムを停止させ、現在の state を checkpointer に書き出す。プロセスが終わっても thread_id さえ分かれば後から再開できる（Web アプリで「ユーザーの返信待ち」を実装できる）。while ループの中に置いているので、retry を返す限り「提案→お伺い」を何度でも繰り返し、approve で初めてループを抜けて最終 state を返す。checkpointer は 必須（中断点を保存する先がないと再開できない）。連載第22回の領収書 OCR エージェント（OCR 結果を人間が承認/修正するループ）と同じパターン。

ex06 — entrypoint.final(value=, save=) で戻り値と保存 state を分離

@entrypoint(checkpointer=MemorySaver())
def workflow(inputs, *, previous=None) -> entrypoint.final:
    state = previous or {"full_history": [], "cumulative_score": 0.0}
    state["full_history"].append({"input": inputs, "score": score})  # 巨大履歴を貯める
    response = {"latest_score": score, "turn_count": len(state["full_history"])}  # 軽い返り値
    return entrypoint.final(value=response, save=state)  # ④ value→呼び出し側 / save→次回 previous

通常の @entrypoint は「return した値」が呼び出し側の戻り値と次回の previous を兼ねる。だが「クライアントには軽いレスポンスだけ返したい、checkpoint には巨大な履歴を保存したい」ときは両者を分けたい。entrypoint.final(value=X, save=Y) を return すると、value=X が invoke の戻り値に、save=Y が次回 previous に入る。workflow.get_state(config).values で覗くと、保存されているのは save 側（巨大履歴つき internal state）であって、value 側の軽量レスポンスではないことが確認できる。API エンドポイント化するときの定番パターン。

3. 用語集（最重要）— 写経で混乱したところ

3-1. @task と @entrypoint（一番の基本対比）

用語 何を指すか 具体例 作り方 単独で invoke できる？
@task ワークフロー内の1ノード（部品）。呼ぶと Future を返す double / slow_double / generate_haiku @task を関数に被せる ❌（entrypoint の中から呼ぶ）
@entrypoint ワークフロー全体の入口（指揮者）。.invoke() の対象 workflow / counter @entrypoint() を関数に被せる ✅（workflow.invoke(...)）

判定基準: 「外から呼ぶ入口」が @entrypoint、「その中で使う部品」が @task。Graph API でいえば @entrypoint ＝ compile 済みグラフ、@task ＝ ノード。

よくある誤解: 「@task も .invoke() できる」→ できない。task は Future を返す部品で、必ず @entrypoint の内側から呼んで .result() する。checkpointer・interrupt・previous といった機能は全部 @entrypoint 側に属する。

3-2. task(x)（Future）と task(x).result()（値）

書き方 戻り値 いつ実行されるか 使いどころ
fut = double(5) Future（引換券） まだ実行されない 並列のために複数ためる（ex03）
double(5).result() int（実体） .result() の瞬間 すぐ値が欲しいとき（ex01・ex04）

判定基準: その値が今すぐ要るなら .result() を即呼ぶ。複数 task を並列で走らせたいなら Future を先にためて、後でまとめて .result()。

3-3. previous 引数 と 通常の位置引数

引数 誰が渡すか 役割 注意
第1引数（increment / inputs） ユーザーが invoke(...) で渡す 今回の入力 普通の引数
previous（keyword-only） LangGraph が自動注入 前回の return 値（前回 state） ユーザーは渡さない・*, の後ろに置く・初回は None

判定基準: invoke() の引数に書いていいのは第1引数だけ。previous は「フレームワークが裏で差し込む読み取り専用の前回 state」。

よくある誤解: 「previous は引数だから自分で渡せる」→ ダメ。*, 以降の keyword-only に置くのは「ユーザーが誤って渡せないようにする」ための作法。checkpointer がないと previous は常に None のままになる点も要注意（永続化先がないので前回値を保持できない）。

3-4. interrupt() と通常の return

動作 interrupt(payload) return value
ワークフローの状態 一時停止（凍結。checkpointer に保存） 完了（終了）
invoke の戻り値 payload（お伺いの中身） value（最終結果）
続きがあるか ある（Command(resume=) で再開） ない
戻り値の受け取り先 interrupt() の戻り値＝次回 resume の値 呼び出し側のみ

判定基準: 「人間/外部の入力を待って続きをやりたい」＝ interrupt。「もう終わり」＝ return。

よくある誤解（README でも警告）: 「interrupt 後の最初の invoke の戻り値＝最終 state」→ 違う。返るのは interrupt の payload。最終 state が返るのは approve でループを抜けて return state に到達したとき。

3-5. return state と entrypoint.final(value=, save=)

方式 呼び出し側に返る値 次回 previous に入る値
return state state（同じもの） state（同じもの。戻り値と保存が一体）
entrypoint.final(value=R, save=S) R（軽量レスポンス） S（内部 state）

判定基準: 「外に見せる返り値」と「内部に貯める state」が同じでいいなら return state。分けたい（クライアントには軽く、checkpoint には重い履歴）なら entrypoint.final。

3-6. checkpointer / thread_id / MemorySaver

用語 役割 具体例
checkpointer state の保存先（永続化の有無を決める）。@entrypoint(checkpointer=...) ex02・04・05・06 で使用
MemorySaver checkpointer の一実装。プロセスメモリ上に保存（再起動で消える） 全サンプル。本番なら SqliteSaver / PostgresSaver に差し替え
thread_id 会話/セッションの識別子。config={"configurable":{"thread_id": "..."}} session-A と session-B で別 state（ex02）

判定基準: previous・interrupt を使いたい→ checkpointer 必須。state を貯めず1回限りの計算→ checkpointer 不要（ex01・ex03）。同じ thread_id ＝ 同じ会話の続き、別 thread_id ＝ 別の会話。

3-7. クイック早見表（困りごと → 使うもの）

困りごと 使うもの
関数をワークフローのノードにしたい @task を被せる
ワークフロー全体の入口を作りたい @entrypoint() を被せる
前回の結果を引き継ぎたい（累積カウンタ・会話履歴） @entrypoint(checkpointer=...) + previous 引数
初回と継続を分けたい state = previous or {初期値}
複数の task を並列で走らせたい Future を先にためる → 後でまとめて .result()
1個ずつ確実に順番に処理したい ループ内で即 .result()（直列）
処理の途中で人間に承認を取りたい interrupt(payload) → Command(resume=...)
interrupt の戻り値が最終 state じゃない それは payload。最終 state は return 到達時
クライアントには軽く返し内部には重い履歴を貯めたい entrypoint.final(value=軽量, save=重い state)
会話/セッションを分けたい thread_id を変える
checkpoint の中身を確認したい workflow.get_state(config).values

4. 学んだこと（要点）

  • Functional API ＝「普通の Python 関数 + デコレータ2種」。StateGraph のグラフ宣言が不要で、分岐・ループ・並列を素の if / for / while / リスト内包表記で書ける。可読性が高い反面、グラフ構造が一目で見える Graph API の利点は失われる（トレードオフ）。
  • @task は遅延評価。呼んだだけでは走らず Future が返り、.result() で初めて実行される。この「呼び出しと wait の分離」が並列化の鍵（ex03）。
  • previous は reducer 構造。new_state = f(previous, input)、戻り値が次回入力。checkpointer がないと previous は常に None。
  • interrupt は state を凍結して再開できる。プロセスをまたいでも thread_id があれば続きをやれる＝Web の HITL が組める。1回目 invoke の戻りは payload であって最終 state ではない、が最頻出の落とし穴。
  • entrypoint.final は「見せる値」と「貯める state」の分離装置。API 化のとき効く。
  • MemorySaver は揮発性。プロセス再起動で消えるので、本番は SqliteSaver / PostgresSaver（または LangGraph Platform）に差し替える。

5. 拡張アイデア（最低3案）

  1. ex03 の真の並列を検証する — slow_double の中で print(time.time()) を出し、4つの開始時刻がほぼ同時刻に並ぶことを確認する。さらに nums を 8 / 16 個に増やして、所要時間が個数に比例せず頭打ちになる（並列度の上限）かを観察する。
  2. ex05 を実 LLM 承認ループに拡張 — propose_value を @task + llm.invoke（例: メール文面のドラフト生成）に差し替え、interrupt で「この文面で送る？ / 直して」を人間に聞く。retry 時はユーザーの修正指示を Command(resume={"action":"retry","note":"もっと丁寧に"}) のように dict で渡し、次の生成プロンプトに織り込む（連載第22回の OCR 承認ループの一般化）。
  3. MemorySaver を SqliteSaver に差し替えて永続化を体感 — from langgraph.checkpoint.sqlite import SqliteSaver に変え、ex02 を「スクリプトを2回別々に起動しても count が引き継がれる」状態にする。プロセスをまたいだ state 復元（＝interrupt 中断からの後日再開）の本質を確認する。
  4. ex06 の get_state で checkpoint を観察 — value（軽量レスポンス）と save（内部 state）が別物であることを、workflow.get_state(config).values の出力と invoke の戻り値を並べて比較し、「外向きの API レスポンス」と「内部に貯まる履歴」が確かに分離されていることを目視する。
  5. Graph API 版（../langgraph_basics/ex01）と行数・可読性を比較 — 同じ「+1 する」処理を StateGraph 版と @task 版で書き比べ、ボイラープレートの差と「グラフ構造の見やすさ vs 制御フローの書きやすさ」のトレードオフを言語化する。

6. 現代版に移植するなら / 注意点

  • API は現行（langgraph>=0.6）準拠。pyproject.toml が langgraph>=0.6.0 で、from langgraph.func import entrypoint, task / from langgraph.types import Command, interrupt はいずれも現行の正規 import。古い書き方の混入はなく、そのまま動く想定。
  • entrypoint.final の型注釈 — def workflow(...) -> entrypoint.final: は型注釈としてはやや変則的（実体はジェネリック）。実行には影響しないが、厳密に書くなら entrypoint.final[PublicResponse, InternalState] のようにパラメータ化できる（任意）。
  • state["full_history"].append(...)（ex06）と state["feedback_history"] = ... + [...]（ex04）の流儀差 — ex06 は previous の dict を直接 mutate しており、ex04 は新リストを作って再代入している。MemorySaver は同一プロセス内では参照を共有しうるので、厳密な immutable を貫くなら ex06 も「新 dict を作って返す」方が安全（このフォルダでは挙動に実害はない、というレベルの注意点）。
  • 本番化 — MemorySaver はメモリ揮発なので、永続化が要るなら SqliteSaver / PostgresSaver、デプロイまで含めるなら LangGraph Platform（連載第16回）に載せ替える。

7. 記事参照

作成: 2026-06-12 ／ 最終更新: 2026-06-12