第14回（〜15回共通）: Streamlit × LangGraph で Human-in-the-Loop リサーチエージェント¶

Software Design 2024年12月号連載第14回（第15回も同じサンプル）。Streamlit でチャット UI を作り、その裏で LangGraph エージェントを動かす。タスク分解 → ユーザー承認 → 順次実行 という Human-in-the-Loop (HITL) フローが核。

連載これまでの「LLM が暴走する系」(11 ReAct / 12 ARAG) に対し、本回は最後の意思決定をユーザーに渡す設計。プロダクション運用で最も実用的なパターン。

全体像¶

flowchart TD
    Start([START]) --> Decompose[decompose_query<br>LLM がタスクを 3〜5 個に分解]
    Decompose --> Interrupt{{interrupt_before<br>= 'human_approval'}}
    Interrupt -.停止 → UI に通知.-> StUI[Streamlit UI<br>承認ボタン or 修正入力]

    StUI -->|APPROVE トークン| Approval[human_approval ノード<br>passthrough]
    StUI -->|フィードバック文字列| Approval

    Approval -->|入力が APPROVE| Exec[execute_task<br>ReAct で 1 タスク実行]
    Approval -->|入力がフィードバック| Decompose

    Exec --> Cond{全タスク完了?}
    Cond -->|未完了| Exec
    Cond -->|完了| End([END])

ポイント: - interrupt_before=["human_approval"] で human_approval ノードに入る直前にグラフを停止 - Streamlit 側が is_next_human_approval_node() で「次が承認ノードか」を polling - ユーザーが承認 or フィードバックを返すと、graph.update_state(..., as_node="human_approval") で状態を注入してから再開 - MemorySaver が thread_id ごとに状態を永続化するので、HTTP リクエスト跨ぎで OK

時系列で見ると:

sequenceDiagram
    autonumber
    participant U as ユーザー (Streamlit UI)
    participant ST as Streamlit (app.py)
    participant A as HumanInTheLoopAgent
    participant G as LangGraph (with checkpointer)
    participant Sub as create_react_agent (sub)

    U->>ST: "生成AI動向を調べて" (chat_input)
    ST->>A: handle_human_message(msg, thread_id)
    A->>G: update_state(human_inputs=[msg], as_node=START)
    A->>G: stream(input=None)
    G->>G: decompose_query → 5 タスクに分解
    Note over G: interrupt_before='human_approval'<br/>でここで停止
    G-->>A: タスクリスト
    A-->>ST: _notify("agent", "タスクを分解しました", ...)
    ST-->>U: タスク表示 + [承認] ボタン

    U->>ST: [承認] クリック
    ST->>A: handle_human_message("[APPROVE]", thread_id)
    A->>G: update_state(human_inputs=["[APPROVE]"], as_node="human_approval")
    A->>G: stream() 再開
    loop tasks
        G->>Sub: create_react_agent.invoke(task)
        Sub-->>G: 結果
        G-->>A: _notify("agent", task, result)
        A-->>ST: 都度表示
    end
    G-->>U: 完了

使用ライブラリ・原理¶

1. LangGraph の Human-in-the-Loop は `interrupt_before` + `checkpointer` で実現¶

LangGraph はノードベースで動くが、指定したノードの直前で実行を中断できる:

graph.compile(
    checkpointer=MemorySaver(),         # 状態を保存する場所
    interrupt_before=["human_approval"],  # ここに入る直前で停止
)

checkpointer が必須な理由: グラフ停止中の State を保存・復元するため。MemorySaver はプロセス内 dict、SqliteSaver / PostgresSaver を使えば永続化可能。

中断後の State は graph.get_state(config) で取れる。次のノードは state.next に入っており、.next == ("human_approval",) なら「次に human_approval を実行する寸前」という意味。

2. `thread_id` がセッション識別子¶

config = {"configurable": {"thread_id": "abc123"}}
graph.stream(input=None, config=config, ...)

checkpointer は thread_id 単位で状態を持つ。同一 thread_id を渡せば前回の続きから再開、新しい thread_id ならゼロから。Streamlit 側では st.session_state.thread_id = uuid4().hex でブラウザセッション単位に固定する。

3. `update_state(..., as_node=...)` で外から状態注入¶

通常 LangGraph はノードの戻り値で State を更新するが、外部から特定ノードを通過したことにして State を更新できる:

graph.update_state(
    config=self._config(thread_id),
    values={"human_inputs": [human_message]},
    as_node="human_approval",   # human_approval ノードが返したことにする
)

これにより: - 「ユーザーが承認した」事実を human_approval ノードが返したかのように扱える - 次の stream(input=None) 呼び出しで Reducer (operator.add) が走り、human_inputs リストに追記される - グラフは次のノード (_route_after_human_approval) へ進む

4. `_route_after_human_approval` で承認 / 修正を分岐¶

def _route_after_human_approval(self, state) -> Literal["decompose_query", "execute_task"]:
    is_human_approved = state.human_inputs and state.human_inputs[-1] == APPROVE_TOKEN
    return "execute_task" if is_human_approved else "decompose_query"

承認 ("[APPROVE]") → execute_task へ進んで実行開始
修正入力 (任意の文字列) → decompose_query に戻ってタスク再分解

これにより「分解 → 修正 → 再分解 → ... → 承認 → 実行」のループが組める。

5. `create_react_agent` をサブグラフとして再利用¶

TaskExecutor.run() で第11回・12回と同じ create_react_agent を呼んでいる:

agent = create_react_agent(self.llm, self.tools)  # tools = [TavilySearchResults]
result = agent.invoke(self._create_task_message(task, results))

ここでの違い: 第11回は ReAct 単体、本回は HITL グラフのノード内でサブエージェントとして実行。各タスクの実行戦略を ReAct に委ねつつ、全体フロー (タスク分解 → 承認 → ループ実行) は親グラフが制御する。

6. `with_structured_output(DecomposedTasks)` でタスク分解¶

class DecomposedTasks(BaseModel):
    tasks: list[str] = Field(default_factory=list, min_items=3, max_items=5, ...)

chain = prompt | self.llm.with_structured_output(DecomposedTasks)
return chain.invoke({...})

LLM の出力が DecomposedTasks インスタンスで返ってくる
min_items=3, max_items=5 で Pydantic がバリデーション → LLM が 2 個や 6 個返すと例外

7. Pub-Sub パターンで UI とエージェントを疎結合¶

class HumanInTheLoopAgent:
    def subscribe(self, subscriber: Callable[[str, str, str], None]) -> None:
        self.subscribers.append(subscriber)

    def _notify(self, type, title, message):
        for subscriber in self.subscribers:
            subscriber(type, title, message)

app.py 側は:

def show_message(type, title, message):
    with st.chat_message(type):
        st.markdown(f"**{title}**")
        st.markdown(message)

_agent.subscribe(show_message)

これで エージェントが Streamlit を直接 import せずに済む。CLI や別 UI に差し替え可能。

8. Streamlit の `st.session_state` でブラウザリロードを跨ぐ¶

Streamlit は 再実行ベースのフレームワーク。ユーザー操作のたびにスクリプト全体が頭から再実行される。st.session_state は ブラウザセッション単位のグローバル変数:

if "agent" not in st.session_state:
    st.session_state.agent = HumanInTheLoopAgent(...)  # 初回のみ
agent = st.session_state.agent  # 2回目以降はキャッシュから取り出し

これがないと、ボタンクリックのたびに新しいエージェントインスタンスが作られ、checkpointer のメモリも消える。

9. `st.rerun()` で UI を強制再描画¶

if approved:
    st.session_state.approval_state = "processing"
    st.rerun()

st.rerun() は スクリプトを即座に最初から再実行する。状態を変えた後で UI を更新したい時に使う。React の setState + 再レンダリングに近い。

ファイル別の役割¶

ファイル	役割
`app.py`	Streamlit UI 層。LangGraph エージェントを生成 / 購読、chat_input / 承認ボタンを表示
`agent.py`	LangGraph エージェント本体。`HumanInTheLoopAgent` クラスと 3 ノード (`decompose_query` / `human_approval` / `execute_task`)
`pyproject.toml`	Poetry 管理。`langgraph^0.2.21`, `streamlit^1.38.0`, `langchain-community^0.3.0`
`poetry.lock`	依存固定
`poetry.toml`	Poetry のローカル設定 (in-project venv 等)

学んだこと（要点）¶

1. HITL の本質は「checkpointer + interrupt + 外部 state 注入」¶

3 要素のセットで初めて成立:

要素	役割
`checkpointer`	グラフ停止中の State を保存。再開時に復元
`interrupt_before=[...]`	指定ノードの直前で停止
`graph.update_state(values, as_node=...)`	外部 (UI / API) から State を注入

これは LangGraph 公式の HITL パターン。Agentic アプリで「ユーザー承認」「外部システム待ち」「人間のフィードバックループ」を入れる定石。

2. `_human_approval` ノードは空関数でも動く¶

def _human_approval(self, state) -> dict:
    pass

pass だけ。なぜなら: - interrupt_before でこのノードに入る直前に停止する - 停止中に update_state(as_node="human_approval") で外部から State 更新 - 再開時、グラフは「human_approval が実行されて State 更新された」とみなし、次のノードへ進む

つまり _human_approval 自体は実行されない。マーカーとして配置されているだけ。これに気付くと HITL の構造が腑に落ちる。

3. Streamlit の polling パターン¶

if agent.is_next_human_approval_node(thread_id):
    # 承認ボタンを表示

毎回スクリプト再実行のたびに「次が承認ノードか」を確認 → そうならボタン表示。Streamlit の再実行モデルが LangGraph のステート保持と相性が良い。st.session_state でエージェントインスタンスをキャッシュし、thread_id で LangGraph 側の状態を引き継ぐ。

4. `_decompose_query` の差分動作¶

if len(human_inputs) > 1:
    latest_decomposed_tasks = state.tasks  # 修正ループなら既存タスクを参考にする
else:
    latest_decomposed_tasks = []  # 初回はゼロから

_latest_human_inputs() は APPROVE_TOKEN 以降の入力を返す: - 初回: ["生成AI動向を調べて"] (1個) → 初回フラグ - 修正: ["生成AI動向を調べて", "もっと細かく"] (2個) → 修正フラグ - 承認後: [..., "[APPROVE]", "別の質問"] → APPROVE 以降だけ取り出す → 1個 → 初回扱い

APPROVE トークンが「フェーズの区切り」になっている設計。

5. `state.next` で「次のノード」を知れる¶

def is_next_human_approval_node(self, thread_id):
    graph_next = self._get_state(thread_id).next
    return len(graph_next) != 0 and graph_next[0] == "human_approval"

StateSnapshot.next は 次に実行されるノード名のタプル。interrupt_before で停止中なら、("human_approval",) のように停止理由のノードが入る。len(graph_next) == 0 ならグラフ終了済み。

6. `draw_mermaid_png()` でグラフ構造を実行時に可視化¶

with st.sidebar:
    st.image(agent.mermaid_png())

LangGraph はグラフ自身を Mermaid PNG として描画できる。Streamlit のサイドバーに表示すれば「いまどんなフローで動いているか」が見える。自分が書いたグラフのデバッグ・他人への説明に最強。

拡張アイデア¶

承認 UI を「承認 / 修正 / 中止」の3択にする
中止ボタンを足し、stream を呼ばずに thread_id を破棄
agent.cancel(thread_id) メソッドを追加
タスク実行中のリアルタイム進捗表示
現状は task 完了後にまとめて _notify。実行中の途中経過を stream_mode="messages" で逐次表示すると体感速度が上がる
永続化を SqliteSaver に切り替え
MemorySaver はプロセス再起動で消える。SqliteSaver(":memory:") 経由でファイル永続化すれば、Streamlit を再起動しても thread 継続
承認待ち期限の追加
24h 以内に承認されなければ自動キャンセル
last_interrupt_at を State に持たせ、cron で掃除
複数承認ポイント
「分解後の承認」「実行前の最終確認」「実行後の品質確認」など、interrupt_before=["pre_exec", "post_exec"] で複数挿入
各承認点で異なる UI を出す
API モードを追加
app.py を FastAPI 化、Streamlit 部分を React フロントエンドに分離
subscribe の購読側を WebSocket / SSE プッシュに変える

実際にやった移植差分 (連載原典 → 本リポジトリ)¶

連載原典は LangGraph 0.2.21 + LangChain 0.3 系で既に十分新しいため、大幅な書き換えは不要。本リポジトリ方針 (1Password 経由のキー取得) に揃える + Pydantic v2 への対応のみ。

1. API キー取得を `op` 経由に (`app.py`)¶

load_dotenv(override=True) を撤去し、_ensure_key(env_name, op_ref) ヘルパで取得:

OPENAI_API_KEY → op://Personal/openAI_API/credential
TAVILY_API_KEY → op://Personal/Tavily_API_key/credential

Streamlit 特有のハマりどころ: スクリプトを毎回再実行するため、単純な module-level フラグでは「2 回目以降の _ensure_key 呼び出しが env 既存で raise」する事故が起きた。@st.cache_resource で関数の戻り値をプロセス内キャッシュし、1 回目だけ実行、2 回目以降はキャッシュ返却に変更。

@st.cache_resource
def _setup_secrets() -> bool:
    _ensure_key("OPENAI_API_KEY", "op://Personal/openAI_API/credential")
    _ensure_key("TAVILY_API_KEY", "op://Personal/Tavily_API_key/credential")
    return True

2. Pydantic v2 への対応 (`agent.py`)¶

Field(min_items=3, max_items=5) は Pydantic v1 構文。v2 では min_length / max_length にリネームされたので置き換え:

# 旧 (v1)
tasks: list[str] = Field(..., min_items=3, max_items=5, ...)

# 新 (v2)
tasks: list[str] = Field(..., min_length=3, max_length=5, ...)

3. 実行コマンドを uv で再現¶

pyproject.toml (Poetry) はそのまま残しつつ、本リポジトリ標準の uv run --no-project --with ... でも動くことを確認。10/11/12 と同じ実行方法に揃えた:

cd 14
uv run --no-project \
  --with "streamlit>=1.38,<2.0" \
  --with "langgraph>=0.2,<1.0" \
  --with "langchain-openai>=0.2,<1.0" \
  --with "langchain-community>=0.3,<1.0" \
  --with "tavily-python>=0.5" \
  --with "pydantic>=2" \
  --with "requests>=2" \
  streamlit run app.py --server.headless true --server.port 8501

4. 動作確認時に観測された Deprecation Warning¶

実行時に出る warning (実装には影響なし、いずれ追従要):

LangChainDeprecationWarning: TavilySearchResults was deprecated in LangChain 0.3.25 ... use 'from langchain_tavily import TavilySearch' — langchain-community 版から langchain-tavily パッケージに分離されたため
LangChainPendingDeprecationWarning: The default value of 'allowed_objects' will change — JsonPlusSerializer のデフォルト動作変更予告

5. 移植しなかった項目¶

interrupt_before → interrupt() (LangGraph 0.2.50+) への移行: 動的中断 API は将来的に柔軟だが、本サンプルの「停止 → UI → 再開」パターンは現状のままで十分。コードを単純に保つため見送り
モデル変更 (gpt-4o → gpt-4o-mini): タスク分解の品質を落とす可能性があるため見送り。コスト削減したい場合は app.py:73 の model="gpt-4o" を変更

既知の不具合・注意点¶

load_dotenv(override=True) がリポジトリ方針 (CLAUDE.md 項目7) に反する。移植時は撤去
Streamlit のスクリプト再実行モデルを理解していないと、エージェントインスタンスが毎回作られて checkpointer が空になる。st.session_state に必ず格納すること
MemorySaver はプロセス内 dict なので、Streamlit を再起動すると全 thread が消える。永続化が必要なら SqliteSaver
_route_after_task_execution の戻り値型 Literal["execute_task", END] で END は langgraph.graph.END (= 文字列 "__end__")。Literal に含めるには文字列リテラルが必要だが、END 定数で書けている (LangGraph 側で Literal 互換にしている)
TaskExecutor.run() 内で タスクごとに新しい ReAct agent を作っている。性能ボトルネックなら使い回す改修も可
承認 UI で「承認」以外の文字列を送ると、それがフィードバックとして再分解の入力になる。意図しない再分解を防ぐには UI を選択肢 (radio / select) に変える

記事参照¶

Software Design 2024年12月号連載第14回「Streamlit + LangGraph + Human-in-the-Loop」（第15回も同サンプル）
関連:
09回 Research Agent — タスク分解の原型 (HITL なし)
11回 ReAct Agent — create_react_agent の単体使用
12回 ARAG — ReAct サブグラフ + 静的ルーティング
LangGraph HITL 公式: https://langchain-ai.github.io/langgraph/concepts/human_in_the_loop/

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

第14回（〜15回共通）: Streamlit × LangGraph で Human-in-the-Loop リサーチエージェント¶

全体像¶

使用ライブラリ・原理¶

1. LangGraph の Human-in-the-Loop は interrupt_before + checkpointer で実現¶

2. thread_id がセッション識別子¶

3. update_state(..., as_node=...) で外から状態注入¶

4. _route_after_human_approval で承認 / 修正を分岐¶

5. create_react_agent をサブグラフとして再利用¶

6. with_structured_output(DecomposedTasks) でタスク分解¶