STUDY NOTES

第21回: LangGraph Functional API — `@task` / `@entrypoint` で StateGraph を書かずに済ませる¶

第7回〜第20回まで、LangGraph といえば「StateGraph を作って add_node / add_edge でグラフを宣言する」Graph API が主役だった。第21回は 2025 年に LangGraph 0.3 で stable 化した Functional API（@task / @entrypoint）を初めて使う回。

比喩: Graph API は「DAG を組み立てる宣言的 DSL」、Functional API は「普通の Python 関数として書く命令的 API」。React で言えば Class Component と関数 Hooks の関係に近い。同じランタイム（同じ checkpointer、同じ persistence）を使うが、書き味が違う。

Functional API の3つの新概念:

キーワード	役割	比喩
`@task`	関数を「将来 (Future) を返す並列実行可能なノード」にする	`asyncio.create_task` の同期版
`@entrypoint(checkpointer=...)`	ワークフロー全体のラッパー。`previous` 引数で前回 return を受け取る	reducer のない React の `useState`、または「再実行可能な main 関数」
`task(...).result()`	future を await	`concurrent.futures.Future.result()`

サンプルアプリは X（旧 Twitter）ポスト生成エージェント: テーマを入力 → LLM がポスト生成 → 改善フィードバックの選択肢 3 つを併せて生成 → ユーザが選択 or 自由入力 → 新フィードバックを反映して再生成、を Streamlit UI で繰り返す。

全体像¶

21/
├── run.py                                  ← uv run run.py のエントリ（streamlit cli を起動）
├── src/
│   ├── main.py                             ← uv run -m src.main 経由のもう1つのエントリ
│   └── content_creator/
│       ├── agent.py                        ← ★Functional API の本体（@task × 2 + @entrypoint × 1）
│       ├── app.py                          ← Streamlit UI ロジック + workflow.stream() 駆動
│       └── ui_components.py                ← Streamlit ウィジェット（チャット欄/コンテンツ欄/フィードバックボタン）

ランタイムフロー（テーマ入力 → ポスト生成 → フィードバック → 再生成のループ）:

sequenceDiagram
    participant User
    participant ST as Streamlit (app.py)
    participant WF as @entrypoint workflow
    participant T1 as @task generate_content
    participant T2 as @task generate_feedback_options
    participant CP as MemorySaver (checkpointer)

    User->>ST: テーマ「LangGraph について」を入力
    ST->>WF: workflow.stream({user_input: "LangGraph について"}, thread_id=UUID)
    Note over WF,CP: 初回 → previous=None
    WF->>WF: state = {theme:"", content:"", options:[], messages:[]}
    WF->>WF: state["theme"] = inputs["user_input"]
    WF->>T1: generate_content(theme, messages) → Future
    T1-->>WF: .result() で sync wait
    WF->>T2: generate_feedback_options(content) → Future
    T2-->>WF: .result() で sync wait
    WF->>CP: state を thread_id ごとに保存
    WF-->>ST: state（content + options 含む）
    ST->>User: 右カラムにポスト、フィードバック3択+自由入力欄を表示

    User->>ST: 「もっと短く」を選択
    ST->>WF: workflow.stream({user_input: "もっと短く"}, thread_id=同じ)
    Note over WF,CP: 2回目 → previous=前回の state
    WF->>WF: state = previous  (theme/content/messages を継承)
    WF->>WF: messages.append({"role":"user","content":"もっと短く"})
    WF->>T1: generate_content(theme, messages) ← feedback 履歴を context に注入
    T1-->>WF: 改訂版ポスト
    WF->>T2: generate_feedback_options(改訂版)
    T2-->>WF: 新しい3択
    WF->>CP: state を更新保存
    WF-->>ST: state
    ST->>User: 改訂版ポスト + 新3択

Functional API の心臓部: @entrypoint でデコレートされた関数は、stream() / invoke() を呼ぶたびに 頭から再実行される。前回の戻り値は previous 引数で渡ってくる。Graph API のように「ノード単位で部分実行」はしない。関数全体を毎回回す代わりに、@task 単位でキャッシュが効く（後述）。

使用ライブラリ・原理¶

`langgraph.func.task` — 関数を「Future を返すノード」化する¶

from langgraph.func import task

@task
def generate_content(theme, messages) -> str:
    ...  # 普通の関数
    return str(response.content)

ポイント:

generate_content(...) を呼ぶと すぐには実行されず SyncRunnableFuture 的なオブジェクトが返る
.result() で同期 wait して実際の戻り値を取り出す
LangGraph ランタイムは @task 単位で checkpoint を取るので、@entrypoint が再実行されても前回成功した @task の戻り値はキャッシュから返る（idempotent な部分実行）
asyncio.gather 的に「複数 task を投げて並列実行 → まとめて wait」も書ける

Graph API との対比: StateGraph.add_node("generate_content", generate_content) + add_edge(...) で結線するのが Graph API。Functional API はその「ノード = task」「エッジ = Python の制御フロー」と読み替える書き方。

`langgraph.func.entrypoint` — ワークフロー全体の入口¶

@entrypoint(checkpointer=MemorySaver())
def workflow(inputs, *, previous=None):
    state = previous or {"theme": "", ...}
    ...
    return state

ポイント:

checkpointer= で指定した永続層に return 値が保存される
次回 workflow.invoke(...) 時、previous キーワード引数に 前回 return 値がそのまま注入される（同じ thread_id の場合）
inputs は今回の新しい入力。previous は前回の蓄積。Redux の (state, action) => state と完全に同型
* 以降は keyword-only。previous は LangGraph フレームワークが注入するのでユーザコードから明示的に渡してはいけない

`ChatPromptTemplate.from_messages` + `llm.with_structured_output(Schema)`¶

第14回や第27回でも見たパターン。Pydantic スキーマを渡すと LLM が JSON で返すよう Anthropic / OpenAI の tool calling が自動で構成される。

class FeedbackList(BaseModel):
    values: List[str]

chain = prompt | llm.with_structured_output(FeedbackList)
response: FeedbackList = chain.invoke({"content": content})

response.values は必ず List[str]。プロンプトで「3つ生成して」と書きつつ、構造化出力でフォーマット崩れを防ぐ二段構え。

`MemorySaver` + Streamlit `session_state` で thread_id を持つ¶

app.py の st.session_state.thread_id = str(uuid.uuid4()) は ブラウザタブごとに 1 つの会話スレッドを作る仕組み。Streamlit は再実行のたびに Python スクリプトを頭から流すが、session_state だけは保持されるので、UUID が安定する → MemorySaver が前回の state を引っ張れる。

ファイル別の役割¶

ファイル	役割
`run.py`	プロジェクト直下から `uv run run.py` で起動。`.env` を読み込み、`ANTHROPIC_API_KEY` 必須チェック後、`streamlit.web.cli` を呼ぶ
`src/main.py`	`pyproject.toml` の `[project.scripts]` に登録された CLI エントリ。`content-creator` コマンド経由。中身は run.py とほぼ同じ
`src/content_creator/agent.py`	本回の中核。Functional API の `@task` 2 個と `@entrypoint` 1 個でワークフローを定義
`src/content_creator/app.py`	Streamlit のメイン UI。2 カラム（左:チャット / 右:コンテンツ）、`process_workflow` で `workflow.stream` を駆動
`src/content_creator/ui_components.py`	Streamlit ウィジェットの分離。CSS パネル定義、サイドバー、メッセージ表示、フィードバック 3 択ボタン + 自由入力フォーム

行レベルの工夫（中核ロジックの抜粋）¶

① `@entrypoint` — 状態を `previous` で復元する Reducer 的書き方 (`agent.py:89-123`)¶

@entrypoint(checkpointer=MemorySaver())                                       # ①
def workflow(
    inputs: Dict[str, Any],
    *,
    previous: Optional[Dict[str, Any]] = None,                                # ②
) -> Dict[str, Any]:
    state = previous or {                                                     # ③
        "theme": "",
        "content": "",
        "options": [],
        "messages": [],
    }

    if state["theme"] == "":                                                  # ④
        state["theme"] = inputs["user_input"]

    state["messages"].append({"role": "user", "content": inputs["user_input"]})

    content = generate_content(state["theme"], state["messages"]).result()    # ⑤
    state["content"] = content

    feedback_options = generate_feedback_options(content).result()
    state["options"] = feedback_options

    message = "コンテンツを生成しました。フィードバックをお願いします。"
    state["messages"].append({"role": "assistant", "content": message})

    return state                                                              # ⑥

行	やってること	なぜそうする
①	`@entrypoint(checkpointer=MemorySaver())` でワークフロー定義	`checkpointer` に return 値が保存され、次回呼び出しで `previous` に流れ込む。Graph API の `StateGraph(...).compile(checkpointer=...)` に相当
②	`previous` を keyword-only で受ける	フレームワークが注入する特別引数。`previous` という名前は固定（変更不可）
③	`previous or {初期 dict}` で「初回 vs 継続」を 1 行で扱う	初回は `previous=None` なので `or` で初期 state にフォールバック。これだけで「永続化された state の復元」が完結
④	テーマは「最初の入力のみ」で確定	ユーザの 2 回目以降の入力はフィードバック扱いにする UX 設計。`if state["theme"] == ""` で初回判定
⑤	`.result()` で task の Future を sync wait	`generate_content(...)` は `@task` なので Future が返る。`.result()` を呼んで初めて LLM が走る
⑥	`return state` で戻り値が次回の `previous` になる	Redux/useReducer と同じ「不変更新して返す」発想

Graph API との対比: 同じ機能を Graph API で書くと、State を TypedDict で宣言 → 各ノード関数を state -> dict で書く → add_edge で結線、と最低 30 行は必要。Functional API ならこの 1 関数で完結する。

② `@task generate_content` — フィードバック履歴を毎回 LLM に注入 (`agent.py:19-61`)¶

@task
def generate_content(theme: str, messages: Optional[List[BaseMessage]] = None) -> str:
    prompt = ChatPromptTemplate.from_messages([
        ("system", "あなたはX（旧：Twitter）のポストを作成するエージェントです..."),
        ("user", "次のテーマにしたがってポストを作成してください：{theme}{feedback_context}..."),
    ])

    feedback_context = ""                                                     # ①
    if messages and len(messages) > 0:
        feedback_context = "\n\n以下はこれまでに受け取ったフィードバックです..."
        for i, message in enumerate(messages, 1):
            if isinstance(message, HumanMessage):                             # ②
                feedback_context += f"{i}. {message.content}\n"
            elif isinstance(message, Dict) and "role" in message and message["role"] == "user":
                feedback_context += f"{i}. {message['content']}\n"

    chain = prompt | llm                                                      # ③
    response = chain.invoke({"theme": theme, "feedback_context": feedback_context})
    return str(response.content)

行	やってること	なぜそうする
①	プロンプト末尾の `{feedback_context}` 差し替え文字列を組み立て	「これまで何を直してきたか」を毎回プロンプトに丸ごと注入することで、LLM がフィードバックを忘れないようにする
②	`HumanMessage` と `Dict` 両方を受ける	本サンプルでは実際には `Dict` だけが入る（`workflow` 内で `{"role": "user", "content": ...}` を append している）。`HumanMessage` 分岐は将来の互換用と思われるが、現状デッドコードでもある
③	LCEL の `prompt \| llm` パイプ	第3回以降ずっと出てくる定型。`.invoke()` で同期実行

@task の旨味はキャッシュ: もし generate_content の中で例外が出て @entrypoint が途中で落ち、再開（リトライ）した場合、前回成功した @task の戻り値は MemorySaver から復元される。冪等な部分実行が無料で得られる。

③ `process_workflow` — Streamlit から workflow.stream を駆動 (`app.py:90-120`)¶

def process_workflow(user_input: str):
    input_data = {"user_input": user_input}
    config = {"configurable": {"thread_id": st.session_state.thread_id}}      # ①

    for chunk in workflow.stream(                                             # ②
        input=input_data,
        config=config,
    ):
        if "workflow" in chunk:                                               # ③
            workflow_data = chunk["workflow"]
            st.session_state.debug_info["interrupt_data"] = workflow_data
            st.session_state.workflow_state = "feedback"
            st.session_state.current_data = workflow_data

    st.rerun()                                                                # ④

行	やってること	なぜそうする
①	`thread_id` を `session_state` から取る	ブラウザタブ単位で UUID を固定 → `MemorySaver` が同じスレッドの履歴を引っ張れる
②	`workflow.stream()` で chunk ごとに受ける	Functional API でも Graph API と同じく `stream` が使える。chunk のキーは `@entrypoint` の関数名（=`"workflow"`）または `@task` の関数名
③	`"workflow"` キーの chunk のみ拾う	task 単位の chunk（`"generate_content"` 等）は無視し、entrypoint 完了時の最終 state だけ UI に反映
④	`st.rerun()` で Streamlit を頭から再実行	Streamlit のお作法。`session_state` に保存した `current_data` をもとに UI が再描画される

学んだこと（要点）¶

Functional API は Graph API の「もう一つの書き方」であって、ランタイム（checkpointer / state persistence / streaming）は同じ
@entrypoint は Redux reducer 同型: (inputs, previous) => next_state。previous を or で初期 state にフォールバックするイディオムだけ覚えれば書ける
@task は Future を返す。.result() で sync wait。複数 task を並列に投げて .result() を後でまとめて呼べば並列実行になる（本サンプルでは直列）
@task 単位で checkpoint が取られるので、@entrypoint がリトライされても重複 LLM 呼び出しが起きない（冪等性が無料）
Graph API の「ノード単位の宣言と接続」より「普通の Python のフロー制御」で書けるので、条件分岐やループが多いワークフローは Functional API のほうが読みやすい
逆に「分岐が複雑で可視化したい」「Studio で見やすくしたい」場合は Graph API のほうがメリット大（Functional API は Studio での可視化が弱い）
Streamlit + LangGraph の組み合わせは、thread_id を session_state に持つだけで会話継続が成立する。MemorySaver の thread_id を session に紐付ける がポイント
構造化出力 (with_structured_output(Schema)) は LLM の JSON 崩れを防ぐ古典かつ強力なテク。フィードバック 3 択のような「個数固定の配列」には特に有効

拡張アイデア¶

@task の並列実行 — generate_content と並行して「ハッシュタグ候補生成」「画像プロンプト生成」を独立 @task で走らせる。fut1 = task1(...); fut2 = task2(...); fut1.result(); fut2.result() で並列化 → API 呼び出しレイテンシを 1/3 に
SqliteSaver に置き換えてマルチセッション永続化 — 現状の MemorySaver はプロセス再起動で消える。from langgraph.checkpoint.sqlite import SqliteSaver に差し替え、Streamlit にもサイドバーで「過去スレッド一覧」を追加
Human-in-the-loop の interrupt — from langgraph.types import interrupt を @entrypoint 内で使うと「ユーザ入力を待つ」明示ポイントが作れる。現状は「3 択ボタン or 自由入力」を UI 側で出しているが、interrupt() でワークフロー側に止め点を持たせる方が安全
A/B 生成 + ユーザ選択 — 1 つのテーマから 2 案を並列生成（@task 2 並列）し、ユーザに「どっちを採用？」を聞く。reducer 的に state にどちらを採用したか記録
langgraph dev で Studio から可視化 — 第17回で学んだ LangGraph Studio から Functional API のグラフを開いてみる。Studio 側の可視化が Graph API ほどリッチでないことを確認するのも勉強
generate_content のリトライ + バックオフ — Anthropic API が rate limit で 429 を返すケースに備え、tenacity で @task をラップ。@task キャッシュが効くので部分リトライしても重複呼び出しにならない

現代版に移植するなら¶

1. デッドコード `isinstance(message, HumanMessage)` 分岐を削る¶

workflow の中で messages.append({"role": "user", "content": ...}) と dict 形式しか入れていないため、generate_content の HumanMessage 分岐は永久に通らない。読み手の認知負荷になるので削除推奨。

2. `claude-3-7-sonnet-20250219` → 最新モデル¶

2026 年現在は claude-sonnet-4-6 等が利用可能。model_name を差し替えるだけ。max_tokens も拡張余地あり。

3. `MemorySaver` → `SqliteSaver`¶

from langgraph.checkpoint.sqlite import SqliteSaver

checkpointer = SqliteSaver.from_conn_string(".cache/workflow.sqlite")

@entrypoint(checkpointer=checkpointer)
def workflow(...): ...

ブラウザ再起動 → ローカル DB から会話復元、が成立する。

4. `entrypoint.final` で外向き戻り値と内部 state を分離¶

LangGraph 0.3.x 以降、from langgraph.func import entrypoint; entrypoint.final(value=..., save=...) を return することで、呼び出し側に返す値とprevious に保存する値を別にできる。本サンプルは UI でも state 全体を欲しがっているので不要だが、APIエンドポイント化するなら有用。

5. `unsafe_allow_html=True` を控える¶

ui_components.py の render_content_area は st.markdown(f'<div class="content-box">{content}</div>', unsafe_allow_html=True) で LLM 出力を生 HTML として埋め込んでいる。LLM が悪意ある HTML や <script> を吐く可能性は低いとはいえ、st.markdown(content) か st.write(content) で十分。XSS リスクを減らすほうが筋が良い。

既知の不具合・注意点¶

render_content_area の XSS リスク: 上記の通り unsafe_allow_html=True で LLM 出力を生 HTML 注入している。テーマに「<script>alert(1)</script> をテーマにポスト作って」と入れると LLM がそれをポストに含む可能性があり、Streamlit がそのまま実行する
render_feedback_options のフォーム送信判定: st.form 内の text_input は Enter キーで送信される。一方ボタンクリックでも返るので、ユーザが「3択ボタンを押した後、自由入力欄に何か残っていた」場合に挙動が読みにくい
process_workflow 内の chunk フィルタ: if "workflow" in chunk: の判定はキー名（=関数名）に依存している。@entrypoint の関数名を workflow 以外に変えると壊れる
Streamlit session_state のリセット手段がない: 「新規スレッド作成」ボタンが UI にないので、現在のセッションを最初からやり直したい時はブラウザタブを開き直すしかない。サイドバーに「リセット」ボタンを足すべき
run.py と src/main.py の重複: ほぼ同じことをしている 2 つのエントリポイントが共存していて、どちらを使うべきか README には uv run run.py しか書かれていない。pyproject.toml の content-creator スクリプトを使うなら uv run content-creator と書くべき

記事参照¶

Software Design 2025 年 6 月号（推定）連載第21回「LangGraph Functional API」
関連: 第14回 STUDY_NOTES — 同じく Streamlit ベース。Graph API との比較に良い
関連: 第16回 STUDY_NOTES — LangGraph Platform（同じく LangGraph 0.3 で安定化した周辺機能）
公式 Functional API ガイド: https://langchain-ai.github.io/langgraph/concepts/functional_api/
公式 @task リファレンス: https://langchain-ai.github.io/langgraph/reference/func/#langgraph.func.task

作成: 2026-05-25 ／最終更新: 2026-06-10

STUDY NOTES

第21回: LangGraph Functional API — @task / @entrypoint で StateGraph を書かずに済ませる¶

全体像¶

使用ライブラリ・原理¶

langgraph.func.task — 関数を「Future を返すノード」化する¶

langgraph.func.entrypoint — ワークフロー全体の入口¶

ChatPromptTemplate.from_messages + llm.with_structured_output(Schema)¶

MemorySaver + Streamlit session_state で thread_id を持つ¶