STUDY NOTES

第22回: Functional API + Human-in-the-Loop — interrupt() と StreamWriter で領収書 OCR エージェントを作る

第21回で Functional API（@task / @entrypoint）の基本を学んだ。第22回はその実戦応用編で、2 つの新しい仕組みを組み合わせる:

仕組み	役割	比喩
interrupt()	@entrypoint の中で「ここで人間の入力を待つ」と宣言する	Jupyter の input() だが、checkpointer に状態を保存してワークフローごと冷凍する
StreamWriter	task 実行中に 任意の中間イベントを UI に push する	print() の Streamlit版／React の Server-Sent Events

サンプルアプリは 領収書 OCR エージェント: 画像アップロード → Claude Vision で OCR → Claude Sonnet が勘定科目を提案 → ユーザが「承認」or「再生成（フィードバック付き）」を選ぶ → 承認後 CSV へ追記。Human-in-the-Loop の典型例を Functional API でいかに薄く書けるかが学びどころ。

第21回との差分: 第21回は「ユーザ入力を毎回 workflow.invoke で渡し、毎回頭から再実行」する単純構造だった。第22回は ワークフロー側に「ここで待つ」マーカーを置き、Streamlit 側は Command(resume=...) で再開する。ワークフロー本体は while ループのまま動き続ける（停止しても再開できる）感覚。

---

全体像

22/
├── run.py
├── fixtures/
│   ├── receipt_meeting.png      ← テスト用領収書画像
│   └── receipt_parking.png
└── src/receipt_processor/
    ├── agent.py                 ← ★ @entrypoint + 3 つの @task + interrupt() ループ
    ├── vision.py                ← Claude Vision で OCR（base64 + structured output）
    ├── account.py               ← Claude Sonnet で勘定科目を提案（feedback 反映）
    ├── storage.py               ← AccountInfo を tmp/db.csv へ追記（バックアップ機能あり）
    ├── models.py                ← Pydantic / Enum（ReceiptOCRResult, AccountInfo, EventType, CommandType, WorkflowState）
    ├── constants.py             ← モデル ID と CSV パス
    ├── app.py                   ← Streamlit。stream_mode=["custom","values"] で双方向通信
    └── ui_components.py         ← ウィジェット部品

ランタイムフロー（OCR → 提案 → フィードバックループ → 承認 → 保存）:

sequenceDiagram
    participant User
    participant ST as Streamlit (app.py)
    participant WF as @entrypoint receipt_workflow
    participant OCR as @task process_and_ocr_image
    participant Sug as @task generate_account_suggestion
    participant Save as @task save_receipt_data
    participant CP as MemorySaver

    User->>ST: 画像アップロード + 「処理開始」
    ST->>WF: receipt_workflow.stream(image_path, stream_mode=["custom","values"])
    WF->>OCR: process_and_ocr_image(path, writer=StreamWriter)
    OCR->>OCR: Claude Vision で構造化抽出
    OCR-->>WF: ReceiptOCRResult
    OCR-->>ST: writer({event: OCR_DONE, ...})   %% stream_mode="custom" で配信
    ST->>ST: session_state.ocr_text を更新

    WF->>Sug: generate_account_suggestion(ocr, feedback_history=[])
    Sug-->>WF: AccountInfo
    Sug-->>ST: writer({event: ACCOUNT_SUGGESTED, ...})
    ST->>ST: session_state.account_info を更新

    rect rgb(255,240,200)
    Note over WF,CP: interrupt() — ワークフロー全体を冷凍してチェックポイントに保存
    WF->>WF: interrupt({ocr, account, count})
    WF-->>ST: stream に __interrupt__ が現れる (stream_mode="values")
    ST->>ST: WorkflowState.WAIT_FEEDBACK + st.rerun()
    ST-->>User: 提案を表示 + 承認/再生成ボタン
    end

    alt 「再生成」+ フィードバック入力
        User->>ST: フィードバック「会議費じゃなくて旅費交通費に」
        ST->>WF: stream(Command(resume=Feedback(REGENERATE, "...")))
        Note over WF,CP: checkpoint から復元。OCR @task はキャッシュから返る
        WF->>OCR: process_and_ocr_image  %% キャッシュHIT
        OCR-->>WF: 前回と同じ ReceiptOCRResult
        WF->>Sug: feedback_history=["..."] で再呼び出し
        Sug-->>WF: 改訂版 AccountInfo
        WF->>WF: interrupt() で再び冷凍
        WF-->>ST: 新 interrupt
    else 「承認」
        User->>ST: 「承認」クリック
        ST->>WF: stream(Command(resume=Feedback(APPROVE, "")))
        WF->>Save: save_receipt_data(account_info)
        Save->>Save: tmp/db.csv に追記 (既存ファイルは _backup.csv に)
        Save-->>ST: writer({event: SAVE_COMPLETED, ...})
        WF-->>ST: ループ break → state return
        ST->>ST: WorkflowState.WORKFLOW_COMPLETED
    end

Functional API + Human-in-the-Loop の心臓部: @entrypoint の中の while True: response = interrupt({...}) が実行を停止 → checkpoint に冷凍 → Streamlit に制御を返す。ユーザがボタンを押して Command(resume=Feedback(...)) で再開すると、interrupt(...) の戻り値が response に注入されてループの次の反復に進む。interrupt は普通の関数呼び出しに見えて、実は coroutine の yield 相当だと考えると分かりやすい。

---

使用ライブラリ・原理

langgraph.types.interrupt(value) — ワークフローを「冷凍」する

from langgraph.types import interrupt

response = interrupt({"ocr_result": ..., "account_info": ...})

挙動:

1. value を引数にして例外 GraphInterrupt(value) を内部 raise
2. LangGraph runtime がそれをキャッチし、現在の state を checkpointer に保存
3. stream() の出力に {"__interrupt__": [Interrupt(value=...)]} という特殊チャンクが流れる
4. 呼び出し側（Streamlit）は __interrupt__ を見て UI 描画 → ユーザ操作を待つ
5. workflow.stream(Command(resume=user_input)) で再開すると、interrupt(...) の戻り値が user_input になって関数の続きから実行される

Python の generator + checkpointer の合せ技: 普通の Python では関数を「途中で止めてあとで再開」は yield 文か async でしか書けないが、LangGraph は state を checkpointer に保存 → 後で同じコードを頭から再実行 + キャッシュで interrupt 直前まで早回し するイミテーションで実現している。@task の戻り値がキャッシュされるのはこのため。

langgraph.types.StreamWriter — 中間イベントを UI へ push

from langgraph.types import StreamWriter

@task
def some_task(*, writer: StreamWriter):
    ...
    writer({"event": "DONE", "data": ...})

挙動:

• writer は @task / @entrypoint の keyword-only 引数として LangGraph runtime が自動注入する
• writer(value) を呼ぶと、workflow.stream(..., stream_mode="custom") の出力に (mode="custom", payload=value) として現れる
• 用途: 長時間 task の進捗、中間結果のプレビュー、UI 状態の早めの更新

workflow.stream(input, stream_mode=["custom", "values"]) — 2 種類のストリームを同時受信

for mode, payload in receipt_workflow.stream(input_data, config=config, stream_mode=["custom", "values"]):
    if mode == "custom":
        # StreamWriter が writer() で投げたペイロード
        ...
    elif mode == "values":
        # state スナップショット（interrupt 検出はここ）
        if "__interrupt__" in payload:
            ...

stream_mode の主要バリアント:

mode	何が流れるか
"values"	state スナップショット（__interrupt__ もここに含まれる）
"updates"	直近 step での差分のみ
"messages"	LLM トークン単位（チャットボット用）
"custom"	StreamWriter で投げたペイロード
リストで複数	(mode, payload) タプルで複数モードを mux

langgraph.types.Command(resume=...) — interrupt されたワークフローを再開

from langgraph.types import Command

workflow.stream(Command(resume=user_feedback.model_dump()), config=config)

Command は LangGraph の「runtime 指示」用ラッパ。resume 以外にも goto= で別ノードへジャンプ等の用途がある。Functional API では主に Command(resume=...) で interrupt の続きから走らせるために使う。

llm.with_structured_output(Schema, mode="function_calling") — Anthropic でも tool_use で JSON 強制

第21回でも出てきたが、第22回は mode="function_calling" を明示している。Anthropic は mode="tool_use" か mode="function_calling" を内部で使い分ける（実装上はほぼ同じ）。Pydantic スキーマを「LLM が必ず守る JSON Schema」として渡す古典テク。

---

ファイル別の役割

ファイル	役割
run.py	.env 読み込み + ANTHROPIC_API_KEY 必須チェック + tmp/ 作成 + streamlit run app.py を起動
src/receipt_processor/agent.py	中核。3 つの @task と @entrypoint 1 つ。while True: interrupt() でフィードバックループ
src/receipt_processor/vision.py	PIL でグレースケール + コントラスト 2 倍 + 長辺 1000px にリサイズ → Claude Haiku Vision に base64 で投げる
src/receipt_processor/account.py	OCR 結果 + フィードバックを system+user プロンプトに組み、Claude Sonnet で AccountInfo を生成
src/receipt_processor/storage.py	csv.DictWriter で tmp/db.csv に追記。書き込み前に _backup.csv を作る
src/receipt_processor/models.py	全 Enum + Pydantic スキーマ。WorkflowState (UI 状態) と EventType (workflow イベント) を分けているのが整理されてる
src/receipt_processor/constants.py	CLAUDE_FAST_MODEL=claude-haiku-4-5-20251001, CLAUDE_SMART_MODEL=claude-sonnet-4-6, CSV_FILE_PATH="tmp/db.csv"（連載原本は claude-3-5-haiku-20241022 / claude-3-7-sonnet-20250219 だが retired のため最新世代に差し替え済み）
src/receipt_processor/app.py	Streamlit 本体。process_workflow が stream を回しながらモード別に分岐
src/receipt_processor/ui_components.py	画像アップローダ、OCR テキスト表示、AccountInfo エディタ、承認/再生成ボタン

---

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

① @entrypoint 内の while True: interrupt() ループ (agent.py:143-243)

@entrypoint(checkpointer=MemorySaver())
def receipt_workflow(image_path, *, previous=None, writer: StreamWriter) -> Dict[str, Any]:
    state = previous or {
        "image_path": image_path,
        "feedback_history": [],
        "completed": False,
    }
    if image_path != state.get("image_path", ""):                              # ①
        state["image_path"] = image_path
        state["feedback_history"] = []

    ocr_result = process_and_ocr_image(image_path, writer=writer).result()    # ②
    account_info = generate_account_suggestion(
        ocr_result, feedback_history=state.get("feedback_history", []), writer=writer
    ).result()

    while True:
        response = interrupt({                                                # ③
            "ocr_result": ocr_result.model_dump(),
            "account_info": account_info.model_dump(),
            "feedback_count": len(state.get("feedback_history", [])),
        })
        feedback: Feedback = Feedback.model_validate(response)

        if feedback.command == CommandType.APPROVE:                           # ④
            save_receipt_data(account_info, writer=writer).result()
            state.update({"ocr_result": ..., "account_info": ..., "completed": True})
            break
        elif feedback.command == CommandType.REGENERATE:                      # ⑤
            feedback_content = feedback.content
            if feedback_content:
                state["feedback_history"] = state.get("feedback_history", []) + [feedback_content]
                account_info = generate_account_suggestion(
                    ocr_result,
                    feedback_history=state["feedback_history"],
                    writer=writer,
                ).result()
        else:
            writer({"event": EventType.ERROR, "message": f"不正なコマンド..."})

    return state

行	やってること	なぜそうする
①	image_path が変わったら feedback 履歴をリセット	同じ thread_id で「別の領収書」を扱うシナリオを想定。同一画像のフィードバックだけ蓄積する
②	process_and_ocr_image(...).result() で OCR を sync wait	@task が返す Future を即座に解決。この task 戻り値は checkpointer に保存され、後の再実行ではキャッシュから返る
③	interrupt({...}) でワークフローを冷凍しユーザ入力待ち	value は Streamlit 側に渡って UI に表示される。interrupt の呼び出し直前で state は永続化されているので、プロセス再起動しても再開可能
④	承認なら save → break で while を抜ける	save_receipt_data も @task なのでキャッシュされるが、break するので 1 回しか走らない
⑤	再生成なら feedback を履歴に追加して generate_account_suggestion を再実行	この再実行で OCR は走らない（②のキャッシュが効くから）。LLM コール 1 回だけで再生成できる経済性

@task キャッシュの効き方: @entrypoint がリトライ/再開すると関数は頭から再実行されるが、各 @task 呼び出しは「この引数の組で既に成功した結果が checkpointer に残っていれば、即座にその値を返す」。OCR は引数（image_path）が同じ限りキャッシュ HIT → 課金 1 回。フィードバック反映後の generate_account_suggestion は引数（feedback_history）が変わっているので MISS → LLM 再呼び出し。これが Functional API の節約効果。

② Streamlit 側で双方向ストリーミングを処理 (app.py:91-176)

def process_workflow(input_data, spinner_message="処理中..."):
    config = {"configurable": {"thread_id": st.session_state.thread_id}}

    with st.spinner(spinner_message):
        try:
            stream_iterator = receipt_workflow.stream(
                input_data, config=config, stream_mode=["custom", "values"]   # ①
            )
            for mode, payload in stream_iterator:
                if payload is None:
                    continue

                if mode == "custom":                                          # ②
                    event = payload.get("event", "")
                    if event == EventType.OCR_DONE:
                        st.session_state.ocr_text = payload["text"]
                        st.session_state.ocr_result = ReceiptOCRResult.model_validate(payload["structured_data"])
                    elif event == EventType.ACCOUNT_SUGGESTED:
                        st.session_state.account_info = AccountInfo.model_validate(payload["account_info"])
                    elif event == EventType.SAVE_COMPLETED:
                        st.session_state.workflow_state = WorkflowState.WORKFLOW_COMPLETED
                        st.rerun()

                elif mode == "values":                                        # ③
                    if "__interrupt__" in payload:
                        interrupt = payload["__interrupt__"][0]
                        st.session_state.ocr_result = ReceiptOCRResult.model_validate(interrupt.value.get("ocr_result"))
                        st.session_state.account_info = AccountInfo.model_validate(interrupt.value.get("account_info"))
                        st.session_state.workflow_state = WorkflowState.WAIT_FEEDBACK
                        st.rerun()

行	やってること	なぜそうする
①	stream_mode=["custom", "values"] で2系統同時購読	中間イベント（custom）と「冷凍タイミング」（values の __interrupt__）を同じループで捌ける
②	custom 分岐: writer で投げた dict を type ごとに振り分け、session_state 更新	OCR が終わった時点で即 UI 反映できる（最終 state まで待たない）。UX の体感速度が変わる
③	values 分岐: __interrupt__ 検出 → interrupt.value から UI 用データ取り出し	これがあると WorkflowState.WAIT_FEEDBACK に遷移し、st.rerun() でフィードバックフォームを描画

ハマりどころ: interrupt.value.get(...) で取れるデータは interrupt({...}) に渡した dict そのまま。一方 custom のペイロードは writer({...}) のもの。同じ情報を 2 経路で送っているのは冗長だが、「interrupt を取り逃しても custom で UI 更新は済んでいる」「custom を取り逃しても interrupt 時に再度フル state が来る」という二重化の意図がある（明示的にコメントはない）。

③ Claude Vision API への構造化 OCR (vision.py:51-73, 123-167)

def build_vision_message(image_path):
    path = pathlib.Path(image_path)
    data = path.read_bytes()
    media_type = mimetypes.guess_type(path.name)[0] or "image/png"
    b64 = base64.b64encode(data).decode()
    return [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": [
            {"type": "text", "text": "この画像から全てのテキストを抽出..."},
            {"type": "image", "source": {                                     # ①
                "type": "base64", "media_type": media_type, "data": b64,
            }},
        ]},
    ]

def ocr_receipt(image_path, model_name=CLAUDE_FAST_MODEL) -> ReceiptOCRResult:
    processed_image_path = preprocess_receipt_image(image_path)               # ②
    try:
        llm = ChatAnthropic(model_name=model_name, temperature=0, ...)
        ocr_chain = llm.with_structured_output(ReceiptOCRResult)              # ③
        result = ocr_chain.invoke(build_vision_message(processed_image_path))
        return result
    finally:
        if os.path.exists(processed_image_path):
            os.unlink(processed_image_path)                                   # ④

行	やってること	なぜそうする
①	{"type": "image", "source": {"type": "base64", ...}}	Anthropic Messages API の Vision 形式。OpenAI と微妙に違うので注意。media_type を mimetypes.guess_type で取得しているのは pdf や jpeg 混在を想定
②	グレースケール + コントラスト 2 倍 + 長辺 1000px リサイズ	OCR 精度向上の前処理。Claude Vision はネイティブ高解像度を受けられるが、ローカル前処理でも有意に改善する経験則
③	with_structured_output(ReceiptOCRResult) で JSON 強制	LLM が ReceiptOCRResult の Pydantic スキーマを守った JSON を返す。プロンプトに「YYYY-MM-DD 形式で」と書いても守られない場合があるが、Pydantic validation で型エラー → リトライ自動化
④	finally で前処理画像を削除	tempfile.NamedTemporaryFile(delete=False) で作っているので明示削除しないと残る。finally で例外時もクリーンアップ

---

学んだこと（要点）

• interrupt() は Functional API 版「human-in-the-loop」の決定打。while True: response = interrupt({...}) で書ける「対話ループ + 永続化」は、Graph API の「conditional_edge + checkpointer + wait_for_user node」の自作コードを 5 行に圧縮する
• StreamWriter で中間結果をリアルタイム配信できる。OCR が終わった瞬間に Streamlit が反応する UX は、これがないと「最終 state まで何も見えない」になる
• @task の自動キャッシュは Human-in-the-Loop で特に効く。同じ OCR を 10 回再呼び出ししても課金 1 回。LangGraph runtime が「同じ task が前に成功している」を見て自動的にキャッシュ HIT
• stream_mode=["custom", "values"] のマルチモード購読が、「中間イベント + 最終 state」を 1 ループで処理する鍵。(mode, payload) のタプルで来る
• Command(resume=...) の resume 値は interrupt(...) の戻り値になる。今回は Feedback.model_dump() した dict を渡して、ワークフロー側で Feedback.model_validate() でデコード
• Claude Vision の Messages API は content が配列で {"type": "text"} と {"type": "image", "source": {"type": "base64", "media_type": ..., "data": ...}} を混ぜる形式。OpenAI とは違うので移植時要注意
• Pydantic + with_structured_output は OCR のような「日付フォーマットを揃えたい」「金額からカンマを除きたい」用途でプロンプトでお願いするより遥かに堅い
• WorkflowState (UI 状態の Enum) と EventType (workflow 内部イベントの Enum) を別の Enum として分離しているのが良い設計。混ぜると「UI 状態を変える workflow event」みたいな循環参照が生まれる

---

拡張アイデア

1. MemorySaver → SqliteSaver でセッション継続 — ブラウザを閉じても処理途中の領収書を後で再開できるようにする。thread_id 一覧をサイドバーに表示
2. 複数領収書の一括処理 — @task を for image in images: で並列起動し、各 OCR を並列実行 → 結果を全部 interrupt({"all": [...]}) で UI に渡して一括レビュー
3. PDF 領収書対応 — vision.py の media_type 判定は既に汎用なので、pdf2image で各ページを画像化 → 複数ページ送れば動く。プロンプトを「複数領収書がある場合は全部抽出」に変える
4. tool_use ベースの prompt cache — 同じシステムプロンプトを毎回送っているので、Anthropic の prompt cache（cache_control ヘッダ）を使うと OCR コール 1 回あたりの入力トークン費を 90% カット
5. CSV 編集後の再フィードバック — 現状ユーザは AccountInfo を承認/再生成しかできないが、account_info_editor で直接編集 → 編集差分を feedback として再投入、というフローを足す
6. Vision モデルの A/B 比較 — CLAUDE_FAST_MODEL (Haiku) と CLAUDE_SMART_MODEL (Sonnet) で OCR 結果を並列で取り、両方を UI に出して どちらが良かったか統計を取るループを足す
7. 承認後の Slack 通知 — save_receipt_data の中で writer({"event": SAVE_COMPLETED, ...}) の隣に Slack Webhook 投稿を加える。経費精算系で実運用すると便利

---

現代版に移植するなら

1. claude-3-7-sonnet-20250219 / claude-3-5-haiku-20241022 → 最新モデル【本リポジトリでは適用済み】

連載原本のモデル ID (claude-3-5-haiku-20241022 / claude-3-7-sonnet-20250219) は retired で 404 になるため、constants.py を claude-haiku-4-5-20251001 / claude-sonnet-4-6 に差し替えた。Vision 性能・OCR 精度が世代ごとに上がっているので 前処理（コントラスト強調）を外しても良いかもしれない。

2. interrupt の引数は dict よりも明示的な Pydantic に

現状 interrupt({"ocr_result": ..., "account_info": ..., "feedback_count": ...}) で dict を渡しているが、InterruptPayload(BaseModel) を作って interrupt(InterruptPayload(...).model_dump()) のほうが UI 側の validate も同じスキーマで揃う。

3. MemorySaver → SqliteSaver

from langgraph.checkpoint.sqlite import SqliteSaver

checkpointer = SqliteSaver.from_conn_string("tmp/checkpoints.sqlite")

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

これだけで「処理途中で Streamlit を閉じても、同じ thread_id で Command(resume=...) すれば続きから動く」が成立。

4. mode="function_calling" の明示は不要に

langchain-anthropic >= 0.3.x では with_structured_output のデフォルトが既に tool_use ベース。mode="function_calling" の明示は省略可能（互換性のため残しても害はない）。

5. print() → logging

vision.py の print("画像取得エラー...") や storage.py の print("CSV保存エラー...") は Streamlit の stdout にバラバラに出る。logging で format 統一 + log level 制御するのが本筋。

6. tempfile のリソース管理を with で

vision.py の tempfile.NamedTemporaryFile(delete=False) + finally: os.unlink は冗長。tempfile.NamedTemporaryFile(delete=True) を context manager で使えば close 時に自動削除される。PIL.save 後に tmp.flush() だけ呼べばいい。

---

既知の不具合・注意点

• OCR 提案の品目情報が失われている: account_info_editor で AccountInfo を編集できるが、AccountInfo に items: List[ReceiptItem] フィールドがない。ReceiptOCRResult から品目を読んで補助科目選定の参考にしているはずだが、編集 UI には現れない
• CommandType 不一致時のエラー処理が片手落ち: else 節で writer({"event": EventType.ERROR, ...}) を投げるだけで ループは抜けない。while 内に戻ってもう一度 interrupt() するので、UI 側でエラーバナーを出しつつ再度フィードバックを待つ動作になる。意図通りなら良いが、抜けるべきケースが混在しそう
• Streamlit の rerun タイミング: process_workflow 内で st.rerun() を複数箇所で呼んでいる。stream_iterator のループ中に st.rerun() を呼ぶとループは即終了し、Command(resume) 未送信のまま次の reruns に進む。意図通り動いているが、可読性は低い
• backup_csv の上書き: バックアップ先のパスが {stem}_backup.csv 固定なので、過去のバックアップが毎回上書きされる。複数回承認するとバックアップは最新の 1 個しか残らない
• CSV エンコーディング: utf-8 で書いているが、Excel で開くと文字化けする（Excel は utf-8-sig または cp932 期待）。encoding="utf-8-sig" にすれば BOM 付きで Excel 互換に
• 品目数が多い領収書の token 圧迫: format_prompt で全品目を for item in ocr_result.items: items_text += ... と展開している。1000 品目あると prompt が肥大化するので、件数制限が必要

---

記事参照

• Software Design 2025 年 7 月号（推定）連載第22回「Functional API + Human-in-the-Loop」
• 関連: 第21回 STUDY_NOTES — Functional API 入門。@task / @entrypoint の基本
• 関連: 第14回 STUDY_NOTES — Streamlit + LangChain の基本パターン
• 公式 Human-in-the-Loop ガイド: https://langchain-ai.github.io/langgraph/concepts/human_in_the_loop/
• 公式 interrupt リファレンス: https://langchain-ai.github.io/langgraph/reference/types/#langgraph.types.interrupt
• 公式 Streaming ガイド: https://langchain-ai.github.io/langgraph/how-tos/streaming/

---

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