STUDY NOTES

第18回: データ分析エージェント — Claude × Python REPL × リフレクションで「調査→可視化→自己検証」を回す¶

ユーザーが「2024年の日経平均推移をCSVとチャートにして」と頼むだけで、Web検索→CSV作成→matplotlibでチャート描画→自己レビュー→必要なら手戻り までを 3 体のエージェントが協調して完了させる回。第17回までの「単一 ReAct エージェント」「Plan→Execute型」から一歩進んで、役割分担 × 動的ルーティング × 自己検証ループ を組み合わせた「マルチエージェント協調」の最小実装になっている。

LangGraph 0.2 系で導入された Command(goto=...) による動的ハンドオフ を使い、add_conditional_edges を 1 個も書かずに 3 ノード間を自由に行き来する設計が一番の見どころ。

全体像¶

sd_18/agent.py
├── ResearchAgent     ← Tavily 検索ツールでデータ収集
├── CodeGenerator     ← Python REPL で pandas + matplotlib 実行
├── ReflectionAgent   ← file_reader + Tavily で成果物を検証、次ノードを決める
└── WorkflowGraph     ← 3エージェントを LangGraph の StateGraph に組み込む

prompts/
├── researcher.prompt       ← データ収集の基準と最低データ量
├── code_generator.prompt   ← CSV/チャートの保存規約と日本語フォント設定
└── reflection.prompt       ← チェックリスト形式の検証フォーマット

データの流れ（動的ルーティング込み）:

flowchart TD
    Start([START]) --> Researcher["researcher<br/>(Tavily で情報収集)"]
    Researcher --> Reflection["reflection<br/>(成果物を検証して次ノードを決定)"]
    Reflection -->|"判定 = researcher<br/>（情報不足）"| Researcher
    Reflection -->|"判定 = code_generator<br/>（情報OK、可視化へ）"| CodeGen["code_generator<br/>(Python REPL で CSV + matplotlib)"]
    Reflection -->|"判定 = end<br/>（最終承認）"| End([END])
    CodeGen --> Reflection

「reflection が中心ハブ」になっていて、研究も生成も毎回ここに戻ってきて品質チェックを受ける構造。古典的な「Plan → Execute → Critic」三役モデルの Critic をルータ兼用にした形。

実際の State 遷移を時系列で見ると:

sequenceDiagram
    participant User
    participant R as researcher
    participant Ref as reflection
    participant C as code_generator

    User->>R: "2024年の日経平均推移を…"
    R->>R: Tavily 検索（複数キーワード）
    R-->>Ref: HumanMessage(name="researcher", 収集結果)
    Note over Ref: file_reader で生成済みファイル確認<br/>+ Tavily で裏取り
    Ref->>Ref: regex で "次のノード: ..." を抽出
    alt 情報不足
        Ref-->>R: Command(goto="researcher")
    else 情報OK
        Ref-->>C: Command(goto="code_generator")
        C->>C: Python REPL で pandas + matplotlib<br/>→ CSV + PNG を出力
        C-->>Ref: HumanMessage(name="code_generator", 実行結果)
        Note over Ref: file_reader で実ファイル検証
        alt 成果物OK
            Ref-->>User: Command(goto=END)<br/>"FINAL ANSWER ..."
        else 成果物NG
            Ref-->>C: Command(goto="code_generator")
        end
    end

ポイント: メッセージは MessagesState（全エージェント共有）に積まれていくので、reflection は researcher が集めた検索結果も、code_generator が出したコード実行ログも、全部参照しながら判定できる。LLM への「文脈」がそのまま会話履歴として渡る LangChain 流。

使用ライブラリ・原理¶

1. `langgraph.types.Command` による動的ハンドオフ¶

LangGraph 0.2 系で追加された機能。ノードの戻り値として Command(update=..., goto=...) を返すと、「state をこう更新したうえで、次は goto のノードへ飛べ」 という意味になる。

from langgraph.types import Command
return Command(update={"messages": [...]}, goto="code_generator")

何が嬉しいか:

add_conditional_edges を書かなくていい（条件ルーティング関数とエッジ定義を分離せず、ノード関数の中で next を決められる）
ノードの戻り値の型ヒントで遷移先を表現できる: Command[Literal["researcher", "code_generator", END]] と書けば、Studio や mypy が「このノードから出るエッジは3本」と理解する
エージェント自身が次のホップを動的に決めるマルチエージェント設計に直接マッチする

第17回までの add_conditional_edges("node", router_fn, {True: "a", False: "b"}) パターンとの違いは、判断ロジックがノードと一体化していること。マルチエージェントの supervisor / hand-off パターンを LangGraph 公式が後押しするための仕組み。

2. `MessagesState` — メッセージリスト 1 本で構成する State¶

LangGraph が用意した「messages: list[BaseMessage] だけを持つビルトイン State」。

from langgraph.graph import MessagesState
workflow = StateGraph(MessagesState)

これは内部で messages: Annotated[list[BaseMessage], add_messages] を持つ TypedDict と等価。add_messages リデューサは メッセージ ID を見て賢くマージ（重複は上書き、新規は追加）してくれる。各ノードは {"messages": [新しいメッセージ群]} を返すだけで、State 全体が自然に積み上がる。

第17回の AgentState（Input / Private / Output を分割）と対照的に、マルチエージェントが共通の会話履歴を参照するユースケースでは「全部 messages に詰める」のがシンプル。

3. `create_react_agent` の `state_modifier`¶

self.agent = create_react_agent(
    self.llm,
    tools=self.config.tools,
    state_modifier=self._make_system_prompt(),  # ← 各エージェント独自の役割定義
)

state_modifier は エージェントへの入力 state を LLM に渡す直前に書き換えるフック。文字列を渡すと system プロンプトとして冒頭に注入してくれる。同じ create_react_agent を使いながら、エージェントごとに役割（researcher / code_generator / reflection）を切り替えるのに使う。

注: 現代版 langgraph では state_modifier は非推奨で、prompt= 引数に置き換わっている。prompt="..." で system プロンプトを渡せる。

4. `PythonREPL` — `langchain_experimental` のサンドボックス Python 実行ツール¶

from langchain_experimental.utilities import PythonREPL
repl = PythonREPL()
result = repl.run("import pandas as pd; df = pd.read_csv('...'); print(df.head())")

LLM が生成した Python コードをその場で実行し、stdout を文字列で返すユーティリティ。langchain_experimental パッケージにあるのは 「LLM が生成したコードを実行 = 任意コード実行リスクあり」 という危険度ゆえ。本番では RestrictedPython や Docker サンドボックス等で隔離する前提。

このサンプルでは、LLM が「df.to_csv('output/.../data/xxx.csv') で保存する Python コード」を書いて、PythonREPL が実際にローカルディスクにファイルを書き出す。LLM がツールを介してファイルシステムを操作する典型例。

5. `japanize_matplotlib` — matplotlib の日本語文字化け対策¶

import japanize_matplotlib  # import するだけで font 設定が完了

matplotlib のデフォルトフォントは日本語非対応で「□□□」になりがち（いわゆる豆腐）。japanize_matplotlib は import するだけで IPAGothic 相当の日本語フォントを rcParams['font.family'] にセットする副作用ライブラリ。

code_generator.py の _setup_visualization_env では、japanize_matplotlib が無い環境向けに IPAGothic / Noto Sans CJK JP を直接指定するフォールバックも書かれていて、Linux サーバ運用も想定した堅さがある。

6. `file_reader` ツール — Reflection エージェントの「目」¶

@tool
def file_reader(file_path: str) -> str:
    # CSV: pd.read_csv で 先頭10行プレビュー
    # PNG: PIL.Image で サイズ・フォーマット情報

reflection エージェントは、code_generator が「保存しました」と報告しただけでは信用せず、実ファイルを読みに行って中身を確認する。これが「リフレクション = LLM の自己評価」を実在ファイル基準で裏付ける仕組み。

「LLM の発言が嘘かもしれない」を前提にツール経由で事実確認する設計は、ReAct の中でも「Verifier」「Checker」パターンと呼ばれる。

7. `recursion_limit` — エージェントの暴走防止柵¶

events = workflow.workflow.stream(
    {"messages": [("user", message)]},
    {"recursion_limit": 150},  # ← 重要
)

Command(goto=...) で動的にループするので、reflection が永遠に「要改善」を出し続けるとエージェントが無限ループに陥る。recursion_limit は ノード実行回数の上限で、150 を超えたら GraphRecursionError を投げる。

第17回の閉じたグラフと違い、本サンプルは reflection が条件を満たすまで researcher / code_generator を呼び続ける開いたループなので、この柵は実質必須。

ファイル別の役割¶

ファイル	役割
`sd_18/agent.py`	3エージェント定義 + WorkflowGraph 組み立て + `__main__` での実行エントリ。これ1ファイルで完結
`prompts/researcher.prompt`	「時系列は12ポイント以上」「カテゴリは5項目以上」などデータ量の最低基準を明示。reflection の判定基準と対応する
`prompts/code_generator.prompt`	保存先パス、ファイル命名規則、CSV 形式、日本語フォント設定など、コード生成のフォーマット規約
`prompts/reflection.prompt`	チェックリストと判定フォーマットを厳密に固定。reflection の出力は regex で機械的にパースする前提なので、形式逸脱を防ぐためにテンプレ化されている
`pyproject.toml`	`langchain-experimental`（PythonREPL）、`japanize-matplotlib`、`pandas`、`matplotlib` が新顔。第17回までと違い OpenAI ではなく `langchain-anthropic` のみ
`repomix.config.json`	repomix（コードベースを1ファイルに集約するツール）の設定。LLM に渡すコード bundle を生成するためのもの。サンプル本体には不要
`output/{TIMESTAMP}/`	生成物の出力先（gitignore対象）。毎回タイムスタンプ付きディレクトリを切るので過去の成果物が残る

行レベルの工夫¶

`agent.py:175` — `claude-3-5-sonnet-latest` を「全エージェント共通」で使う¶

self.llm = ChatAnthropic(model="claude-3-5-sonnet-latest")

3 エージェントとも同じ Claude 3.5 Sonnet を使い回す。役割分担は LLM の種類ではなくシステムプロンプトとツール構成で実現する設計。コスト最適化したいなら researcher は claude-3-5-haiku-latest、reflection は claude-3-opus のように分けることもできる。

`agent.py:195-198` — エージェント出力を `HumanMessage(name=...)` で偽装する¶

result["messages"][-1] = HumanMessage(
    content=result["messages"][-1].content, name=self.config.name
)
return Command(update={"messages": result["messages"]}, goto=next_node)

ReAct エージェントの最終発話は本来 AIMessage だが、次のエージェントから見ると「同僚の人間が報告してきたメッセージ」として扱いたい。なので HumanMessage(name="researcher") に変換する。name フィールドが「誰の発言か」を保持するので、後段の LLM は「researcher さんからの報告だな」と認識できる。

これは LangGraph 公式の「Multi-agent supervisor」サンプルでも使われる定石パターン。

`agent.py:307-321` — Reflection が regex で次ノードを抽出¶

next_node_match = re.search(
    r"次のノード:\s*(researcher|code_generator|end)", content
)
if next_node_match:
    goto = next_node_match.group(1)
    if goto == "end":
        goto = END
else:
    goto = next_node  # フォールバック
    content += "\n\n次のノードが不明確です。正しい形式で次のノードを指定してください。"

LLM の自由記述から 「次のノード: code_generator」という固定パターンを正規表現で抜き出す愚直な実装。with_structured_output(Pydantic) を使えばもっと型安全だが、ここではチェックリスト形式の出力に自然に埋め込めることを優先している。

抽出失敗時はメッセージに自己改善要求を追記して同じノードに戻す → 次イテレーションで正しい形式を出してもらう。プロンプトと regex の二重契約。

`agent.py:361-372` — `add_edge` を一切書かない設計¶

workflow = StateGraph(MessagesState)
workflow.add_node("researcher", self._research_node)
workflow.add_node("reflection", self._reflection_node)
workflow.add_node("code_generator", self._code_node)
workflow.set_entry_point("researcher")
return workflow.compile()

add_edge も add_conditional_edges も一切登場しない。ノード遷移は全部 Command(goto=...) の戻り値で決まる。グラフ定義はノードを登録するだけ。動的ハンドオフの利点を最大限活かした書き方。

エッジを書かないので、Studio で見るとノードが「島」のように浮いて見えるが、Command[Literal["researcher", "code_generator", END]] の型ヒントから Studio は遷移可能性を推論して描画してくれる。

`agent.py:333-334` — file_reader を Reflection の `init` で受け取る¶

self.file_reader = FileReader(self.timestamp)
self.reflection_agent = ReflectionAgent(self.file_reader)

FileReader は timestamp をクロージャに閉じ込めたツールを返す。これにより reflection の file_reader("data/xxx.csv") は 常に output/{timestamp}/data/xxx.csv を見にいくようになる（相対パス → 自動で timestamp 付きパスに解決）。

実行時タイムスタンプを Tool 内部状態として埋め込む、シンプルだが効くテクニック。

`agent.py:357-358` — `os.makedirs(..., exist_ok=True)` で起動時にディレクトリ確保¶

os.makedirs(f"output/{self.timestamp}/charts", exist_ok=True)
os.makedirs(f"output/{self.timestamp}/data", exist_ok=True)

LLM がコードを書く時点ですでに保存先が存在することを保証。LLM 側で os.makedirs を書き忘れる事故を防ぐ。「LLM が失敗しやすいセルフサービス処理を、決定論的なコードで前倒し」 する考え方。

学んだこと（要点）¶

Command(goto=...) は LangGraph のマルチエージェント設計に必須の道具。add_conditional_edges を書かずに、各ノードが「次は誰に渡す」を能動的に決められる。Supervisor pattern と Hand-off pattern の両方が表現できる
MessagesState は「全員共有のチャットルーム」。エージェントが個別 state を持つのではなく、共通の会話履歴に発言を積み上げて互いの仕事を参照する。LangGraph 公式の Multi-agent パターンの基本
エージェント出力を HumanMessage(name=...) に変換するのが定石。次のエージェントから見て「同僚の発言」として扱えるようにするため
リフレクションは「ツールで実在を確認」してこそ意味がある。LLM の発言だけで判定すると、code_generator の「保存しました」という嘘も通ってしまう。file_reader で実物を読みに行く設計が重要
regex でルーティング情報を抽出するなら、プロンプト側で出力フォーマットを厳密固定する。reflection.prompt がチェックリストとして書かれているのは LLM の自由度を縛るため
recursion_limit は動的ループ型エージェントの安全装置。デフォルトの 25 では足りないので明示的に 150 等に上げる
japanize_matplotlib は import するだけで日本語化。本番 Linux 環境では fonts-ipafont-gothic / fonts-noto-cjk の OS インストールが前提
PythonREPL は便利だが本番では危険。langchain_experimental にあるのは「実験用」のサイン。実運用ではコンテナ隔離や RestrictedPython が必要

拡張アイデア¶

エージェント追加で 4 ノード化 — presentation_agent（最終レポートを Markdown でまとめる）を追加し、reflection が end の代わりに presentation に goto するようにする。プレゼン用画像をチャートから引用させる
recursion_limit を消費した時点で再計画 — GraphRecursionError をキャッチし、ResearchAgent を呼び直して「これまでの失敗履歴」を踏まえて方針を変えさせる。MemorySaver + try/except で実装
複数チャート種類の比較 — code_generator が同じデータから複数チャート（折れ線・棒・散布図）を生成し、reflection がどれが最も視認性が高いか LLM 自己評価で選ぶ
データ取得を Tavily から JQuants API に置換 — 日経平均など金融データは Tavily で取るより専用 API のほうが精度が高い。ResearchAgent のツールを差し替えて挙動を比較
生成された CSV と過去実行の CSV を diff — output/ の過去実行ディレクトリを find で列挙して、データの差分や統計の変化を要約する comparator_agent を追加。時系列で「データの鮮度・変化」を可視化
with_structured_output 化 — reflection の判定を Pydantic モデルで返すようにし、regex 抽出を廃止。型安全に。プロンプトもチェックリスト形式から構造化出力に移行

現代版に移植するなら¶

state_modifier → prompt — create_react_agent(llm, tools, state_modifier=...) は langgraph 0.3+ で prompt= に置き換わった。新しい書き方は create_react_agent(llm, tools=tools, prompt="あなたは…")。state_modifier は当面互換維持されているが Deprecation Warning が出る
claude-3-5-sonnet-latest → claude-haiku-4-5 ✅ 移植済み — 連載当時のモデル ID claude-3-5-sonnet-latest は新規アカウントから 404 になる（Anthropic 側で旧モデル/旧エイリアスのアクセス制御が変わった）。本リポジトリでは agent.py:175 を ChatAnthropic(model="claude-haiku-4-5") に更新済み。最初 claude-sonnet-4-6 を試したが usage tier 1 の 30,000 input tokens/min レートリミットに掛かるため、Haiku 4.5 に下げた（このサンプルは reflection ループで同じ会話履歴を何度も投げるので input トークンが膨らみやすい）。本番品質を求めるなら Sonnet 4.6 / Opus 4.7 + usage tier 2 以上が安全。プロンプトキャッシュ (cache_control) を併用すればトークン消費を大幅に減らせる
TavilySearchResults → TavilySearch — langchain_community.tools.tavily_search は非推奨。現代版は langchain-tavily パッケージの from langchain_tavily import TavilySearch
LANGSMITH_TRACING_V2 → LANGSMITH_TRACING — 環境変数名が変わっている（第17回の注意点と同じ）
PythonREPL を Docker で隔離 — 本番でこのコードをそのまま使うなら、code_generator の REPL を Docker コンテナ + リソース制限で隔離するべき。langchain_experimental.tools.PythonAstREPLTool は AST チェックがあるぶん少しは安全
タイムスタンプを ULID/UUID に — 同一秒に複数実行されると衝突する。uuid.uuid4().hex[:8] を suffix に付けると安全
@dataclass(frozen=True) — AgentConfig を frozen にすると不変性が保証され、エージェント設定の取り違え事故を防げる
構造化ロギング — 現状 print(f"[{node_name}] {content}") だが、structlog か loguru で JSON ログにすれば LangSmith とは別の観測パスができる

既知の不具合・注意点¶

Agent.run の戻り値の型ヒントが緩い — Command を返すが Command[Literal[...]] 型パラメータが付いていない。Studio が遷移先候補を完全には推論できない（実害は小さい）
reflection の自己改善メッセージが goto に乗り損ねる — goto = next_node のフォールバック時、append したメッセージで「次のノード: …」を求めても、結局フォールバック先（デフォルトは researcher）に飛んでしまうので、最大 recursion_limit まで空回りする可能性がある。実運用では再試行回数のカウンタが欲しい
PythonREPL の標準出力に依存 — repl.run(code) は print 出力を文字列化する。LLM が print() を書き忘れると「実行結果: None」になり、reflection が「失敗」と誤判定する場合がある。プロンプトで「結果を必ず print してください」を強調するか、評価式を別途取れるラッパーが望ましい
output/{timestamp}/ の cleanup なし — 実行ごとにディレクトリが増え続ける。長期運用なら最古のディレクトリを自動 prune する仕組みが必要
Tavily API の rate limit — 短時間に多数検索すると 429。reflection ループが回ると Tavily が複数エージェントから叩かれるので、現実的には max_results を絞るか sleep を挟む
macOS の brew install font-ipa は cask 名が変わっている可能性 — 現代版は brew install --cask font-ipafont-mincho のような表記。japanize_matplotlib を入れていれば不要
reflection の regex が markdown 強調をパースできない（実測で発覚） — agent.py:309 の re.search(r"次のノード:\s*(researcher|code_generator|end)", content) は LLM が 次のノード: **end** のように markdown 強調付きで返すとマッチ失敗する。Haiku 4.5 で実測したところ毎回のように ** 付きで返ってきてフォールバック発動。修正案: r"次のノード:\s*\**(researcher|code_generator|end)\**" または with_structured_output への移行
reflection が code_generator を呼ばずに END へ飛ぶ事故（実測で発覚） — researcher が表形式や箇条書きで「データらしきもの」を返すと、reflection が「もう成果物揃ってる」と勘違いし、CSV/PNG 未生成のまま 次のノード: end を出す。メモ本文に書いた「LLM の発言だけで判定すると嘘も通る」がそのまま起きる。修正案: reflection.prompt で「CSV/PNG ファイルが実在しないうちは絶対に end に飛ばすな、必ず code_generator を経由せよ」を強調、または file_reader 必須呼び出しをハードコード
reflection.prompt を強化したら recursion_limit ループに転じる（実測で発覚） — 上記 2 件を「regex を \** 対応にする + reflection.prompt に厳守事項を追加」で修正したところ、今度は reflection が「file_reader を必ず呼べ」を律儀に守りすぎ、researcher → reflection → researcher → reflection ... と無限に行き来して recursion_limit=12 に到達。Haiku 4.5 ではプロンプト指示と「次に何をするか」の判断がうまく結びつかない。本格的に動かすには (1) reflection 側で「リサーチが N 回続いたら強制的に code_generator」のような決定論的フォールバック、(2) with_structured_output で goto を Pydantic 必須フィールドにする、(3) Sonnet 4.6 以上のモデル、のいずれかが必要

記事参照¶

Software Design 2025年2月号連載第18回「データ分析エージェント」
関連サンプル: 第17回（マルチエージェントの基礎、サブグラフ）、第20回（MCP × create_react_agent で SQLite 操作）、第21回（Human-in-the-Loop による改善ループ）。本回の「reflection が中心ハブ」設計は第10回 CRAG の「retriever→grader→generator」ループの一般化と読める
公式リファレンス: LangGraph Multi-agent supervisor tutorial、Command primitive docs

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

STUDY NOTES

第18回: データ分析エージェント — Claude × Python REPL × リフレクションで「調査→可視化→自己検証」を回す¶

全体像¶

使用ライブラリ・原理¶

1. langgraph.types.Command による動的ハンドオフ¶

2. MessagesState — メッセージリスト 1 本で構成する State¶

3. create_react_agent の state_modifier¶

4. PythonREPL — langchain_experimental のサンドボックス Python 実行ツール¶

5. japanize_matplotlib — matplotlib の日本語文字化け対策¶

6. file_reader ツール — Reflection エージェントの「目」¶

7. recursion_limit — エージェントの暴走防止柵¶