STUDY NOTES

第24回: Supervisor + 共有メモリ + TODO リストで作る Claude Code 風文章執筆エージェント¶

第23回で学んだ langgraph-supervisor の実戦応用回。「ユーザの一言から → タスク分解 → Web 調査 → 記事執筆 → Markdown 保存」までを 3 エージェントの supervisor 構成で自走させる。

第23回との差分: 23回は「2 エージェントの計算/調査」という最小例だった。24回は 3 エージェントが「TODO リスト」と「研究データメモリ」という共有状態を介して非同期に協調する、より本格的な構成。Claude Code（このリポジトリで動いている自分自身）と同じ思想の TODO 駆動を LangGraph で再現している。

エージェント	担当	主要ツール
`task_decomposer`	ユーザ依頼を TODO に分解	`create_multiple_todos`, `search_and_save`（事前リサーチ用）
`research`	TODO に従って Web 調査・メモリへ保存	`get_research_todos`, `search_and_save`, `update_multiple_todo_status`
`writer`	調査データから Markdown 記事を生成・ファイル保存	`get_writer_todos`, `write_and_save_content`, `complete_writing_task`

特に重要な設計判断:

memory と todo_manager がモジュールレベルのシングルトンとして全 agent で共有される（LangGraph の state ではない）
create_get_my_todos_for_agent(agent_name) でクロージャを使ったエージェント固有ツールを動的生成
supervisor のプロンプトに「最終出力は file path 1 行のみ」と厳格指示することで、LLM の冗長応答を抑制

全体像¶

24/
├── main.py                                  ← `AgentRunner.run()` のラッパ
├── langgraph.json                           ← Studio に 4 グラフ公開（writing_assistant + 各サブ agent 単体）
└── src/sd_24/
    ├── main.py                              ← ★ create_supervisor で 3 agent を統合
    ├── agents/
    │   ├── task_decomposer.py               ← TODO 分解専門の create_react_agent
    │   ├── research.py                      ← Web 検索 + メモリ保存
    │   └── writer.py                        ← 記事生成 + Markdown 保存
    ├── utils/
    │   ├── memory.py                        ← Memory シングルトン（Haiku で圧縮）
    │   ├── todo_manager.py                  ← TodoManager シングルトン
    │   ├── todo_tools.py                    ← TODO 操作の @tool 群（クロージャ含む）
    │   └── search_tools.py                  ← Tavily 検索 + 結果を memory に保存
    ├── runner/
    │   └── agent_runner.py                  ← argparse + workflow.compile + UI 起動
    └── display/                             ← Claude Code 風 TUI（terminal_ui, progress_tracker, task_display）

実行フロー（「2025年AI動向をレポートして」というクエリの場合）:

sequenceDiagram
    participant User
    participant Sup as supervisor
    participant TD as task_decomposer
    participant Re as research
    participant Wr as writer
    participant Mem as memory (singleton)
    participant Todo as todo_manager (singleton)
    participant FS as ファイルシステム

    User->>Sup: 「2025年AI動向をレポートして」
    Sup->>TD: transfer_to_task_decomposer
    TD->>TD: クエリ分析
    TD->>Todo: create_multiple_todos([<br/>「AI市場規模調査」(agent=research),<br/>「主要企業動向調査」(agent=research),<br/>「総合レポート執筆」(agent=writer)])
    TD-->>Sup: 「TODOを作成しました」

    Sup->>Re: transfer_to_research
    Re->>Todo: get_research_todos() → 未完了タスク取得
    Re->>Re: search_and_save("AI市場規模", topic="market_size")
    Re->>Mem: research[market_size] = (Haiku で圧縮した検索結果)
    Re->>Re: search_and_save("主要企業動向", topic="key_players")
    Re->>Mem: research[key_players] = ...
    Re->>Todo: update_multiple_todo_status(完了 × 2)
    Re-->>Sup: 「調査完了」

    Sup->>Wr: transfer_to_writer
    Wr->>Wr: check_research_data_sufficiency() → 「執筆可能」
    Wr->>Mem: research を取得 → プロンプトに展開
    Wr->>Wr: write_and_save_content(task_id, description)<br/>内部で Claude Sonnet を呼ぶ
    Wr->>FS: output/<timestamp>/TASK-0003.md に保存
    Wr->>Mem: final_document_path = filepath
    Wr->>Todo: complete_writing_task(task_id)
    Wr-->>Sup: 「全タスク完了！ファイル: output/...」
    Sup-->>User: output/<timestamp>/TASK-0003.md

シングルトン共有メモリの是非: LangGraph 公式は「state を通して agent 間で値を渡せ」と推奨している。本サンプルはプロセスローカルなシングルトン dictで渡しているので、同時に複数セッションを走らせると state が混線する。InMemorySaver の thread_id 分離が効かない。学習用として割り切るなら問題ないが、production では state 経由に移すべき（後述「現代版に移植するなら」）。

使用ライブラリ・原理¶

`langgraph_supervisor.create_supervisor`（第23回からの再利用）¶

第23回で学んだ create_supervisor をそのまま使う。今回の特殊性は supervisor プロンプトの厳格な制約:

最終出力ルール（厳守）：
Writerから「執筆完了」の報告を受けたら：
1. 独自の内容生成は禁止
2. 「執筆結果」「説明のポイント」等の追加は禁止
3. Writerが保存したファイルパスのみを返す
4. 形式: output/[日時]/article.txt（1行のみ）
5. それ以外の文章は一切書かない

これがないと supervisor が Writer の応答を要約・改変してファイル内容と乖離した「自分の答え」を返す現象が起きる。LLM をオーケストレータにする時の古典的問題への対策。

モジュールレベルシングルトンによるエージェント間通信¶

# utils/memory.py の末尾
memory = Memory()  # シングルトン

# utils/todo_manager.py の末尾
todo_manager = TodoManager()  # シングルトン

各 agent のツールはこれらを from ..utils.memory import memory で import し、共有状態として使う。LangGraph state を介さないバイパス通信路。

比喩: LangGraph の state は「式典の進行表」、シングルトンメモリは「現場の共有ホワイトボード」。本来は進行表に書くべきものを、簡便さのためホワイトボードに殴り書きしている構造。

クロージャによるエージェント固有ツール生成¶

def create_get_my_todos_for_agent(agent_name: str) -> Runnable:
    @tool
    def get_agent_todos() -> dict:
        pending_tasks = todo_manager.get_pending_tasks(agent_name)  # ← agent_name を closure で捕捉
        return {"tasks": [...], "count": len(pending_tasks)}

    get_agent_todos.name = f"get_{agent_name}_todos"
    get_agent_todos.description = f"{agent_name}エージェントの未完了TODOを取得"
    return get_agent_todos

# research agent 側で:
tools = [create_get_my_todos_for_agent("research"), ...]

ツール名と振る舞いを agent ごとに分けるためのテク。Python の closure で agent_name を捕捉した上で .name を後付け書き換え。LLM 視点では get_research_todos と get_writer_todos という別々のツールに見える。

Claude Haiku による検索結果の圧縮¶

async def compress_research(self, topic: str, findings: str) -> str:
    prompt = f"以下はWeb検索で得られた「{topic}」に関する情報です..."
    response = await self._compressor.ainvoke(prompt)
    ...

Tavily の生検索結果（数千トークン）を Haiku で要約してから memory に保存するパターン。

なぜ Sonnet ではなく Haiku か → 圧縮は単純タスクでコスト 1/12（Haiku 4.5）
なぜ圧縮するか → 後で Writer に「全 topic の調査データ」を 1 プロンプトに詰め込むときに context 爆発を防ぐ
「LLM を 2 段（オーケストレータ Sonnet + 補助 Haiku）使い分ける」は production パターン

ファイル別の役割¶

ファイル	役割
`main.py` (ルート)	`AgentRunner().run()` を asyncio で起動
`src/sd_24/main.py`	中核。3 agent を `create_supervisor` で統合。supervisor プロンプトで「ファイル path 1 行だけ返せ」を厳命
`agents/task_decomposer.py`	TODO 作成専門。事前リサーチ用に search ツールも持つが、書き込み側は `create_multiple_todos` がメイン
`agents/research.py`	自分用 TODO をフィルタ取得 → 各 topic を `search_and_save` → 全 TODO を完了状態に
`agents/writer.py`	`write_and_save_content` の内部で Claude Sonnet を再呼び出しして記事生成。`max_tokens=8192`
`utils/memory.py`	`Memory` シングルトン。`set/get` と `compress_research` (Haiku) を持つ
`utils/todo_manager.py`	`TodoManager` シングルトン。`add_task / update_status / get_pending_tasks(agent=...)`
`utils/todo_tools.py`	`@tool` 化された TODO 操作。`create_get_my_todos_for_agent` でエージェント固有ツールを動的生成
`utils/search_tools.py`	Tavily 検索 → Haiku 圧縮 → memory.research[topic] に保存
`runner/agent_runner.py`	argparse で `--debug` 対応 + `InMemorySaver` + UI 起動
`display/*.py`	Claude Code 風 TUI（プログレスバー、タスク表示、メッセージフォーマット）

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

① Supervisor プロンプトの「ファイル path 1 行のみ」厳命 (`src/sd_24/main.py:54-60`)¶

system_prompt = f"""...
最終出力ルール（厳守）：
Writerから「執筆完了」の報告を受けたら：
1. 独自の内容生成は禁止                                                       # ①
2. 「執筆結果」「説明のポイント」等の追加は禁止
3. Writerが保存したファイルパスのみを返す                                     # ②
4. 形式: output/[日時]/article.txt（1行のみ）
5. それ以外の文章は一切書かない"""

行	やってること	なぜそうする
①	「独自の内容生成は禁止」を明記	これがないと Sonnet が「もっと役立とう」として勝手に要約・改変版を返す。LLM-as-orchestrator の頻出問題
②	「ファイルパスのみ返す」	出力契約を 1 行に固定 → 自動化スクリプトから `output/...` をパース可能

② クロージャでエージェント固有 TODO ツールを生成 (`utils/todo_tools.py:26-42`)¶

def create_get_my_todos_for_agent(agent_name: str) -> Runnable:
    @tool                                                                     # ①
    def get_agent_todos() -> dict:
        pending_tasks = todo_manager.get_pending_tasks(agent_name)            # ②
        return {
            "tasks": [task.to_dict() for task in pending_tasks],
            "count": len(pending_tasks),
        }

    get_agent_todos.name = f"get_{agent_name}_todos"                          # ③
    get_agent_todos.description = f"{agent_name}エージェントの未完了TODOを取得"
    return get_agent_todos

行	やってること	なぜそうする
①	内側に `@tool` 付きのクロージャを定義	関数自体は 1 つだけ書いて、`agent_name` 違いで複数のツール instance を作る
②	`agent_name` を closure で捕捉 → `get_pending_tasks(agent_name)` で agent 固有フィルタ	research agent が呼ぶときは `"research"` の TODO だけ、writer agent が呼ぶときは `"writer"` の TODO だけ返る
③	`.name` / `.description` を後付け書き換え	LLM 側からは別々のツール (`get_research_todos` / `get_writer_todos`) に見える → 混乱を避ける

ハマりどころ: @tool デコレータが生成する BaseTool は本来 immutable に近い扱いだが、.name / .description への属性代入は実は可能（mypy では # type: ignore が必要）。本サンプルの裏技的書き方だが、最近の LangChain では StructuredTool.from_function(..., name=..., description=...) で同じことが整然と書ける（後述）。

③ Writer の中でさらに LLM を呼ぶ (`agents/writer.py:31-110`)¶

@tool
async def write_and_save_content(task_id, task_description) -> dict:
    research_data = memory.get("research", {})                                # ①

    research_sections = "\n".join(
        [f"## {topic}\n{content}\n" for topic, content in research_data.items()])

    prompt = f"""あなたは熟練した執筆者です。以下のタスクについて...
タスク: {task_description}
利用可能な調査データ:
{research_sections}
要件:
- 調査データを分析・統合して有機的な記事として構成
- 総合的なレポートの場合は2000文字以上
..."""

    model = ChatAnthropic(                                                    # ②
        model_name="claude-sonnet-4-20250514",
        temperature=0.7,
        max_tokens=8192
    )
    response = await model.ainvoke([HumanMessage(content=prompt)])
    content = response.content

    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    output_dir = os.path.join("output", timestamp)
    os.makedirs(output_dir, exist_ok=True)
    filename = f"{task_id}.md"
    filepath = os.path.join(output_dir, filename)

    with open(filepath, "w", encoding="utf-8") as f:                          # ③
        f.write(content)
        f.flush()

    memory.set("final_document_path", filepath)
    return {"task_id": task_id, ..., "filepath": filepath, "success": True}

行	やってること	なぜそうする
①	memory から全 topic の研究データを取り出して連結	research agent がため込んだ全データを Writer が一気に消費。圧縮済みなので context が破裂しない
②	ツール内部で別の Sonnet インスタンスを起動（temperature=0.7 / max_tokens=8192）	Writer agent 本体は `temperature=0`（決定論的）だが、執筆だけは創造性が必要なのでツール内で別パラメータの LLM を呼ぶ。「Agent ≠ LLM」「Agent は LLM を内包したツール群」の良例
③	`output/<timestamp>/<task_id>.md` に保存	timestamp + task_id でファイル名衝突を防ぎつつ、複数記事を同一ディレクトリにまとめる

④ 検索 → 圧縮 → メモリ保存の 1 アクション化 (`utils/search_tools.py:8-28`)¶

@tool
async def search_and_save(query: str, topic: str) -> str:
    search = TavilySearch(max_results=3)
    response = await search.ainvoke(query)

    content = f"検索: {query}\n\n"
    if isinstance(response, dict) and "results" in response:
        for i, result in enumerate(response["results"], 1):
            content += f"{i}. {result.get('title', '')}\n"
            content += f"{result.get('content', '')}\n\n"

    compressed = await memory.compress_research(topic, content)               # ①
    research_data = memory.get("research", {})
    research_data[topic] = compressed                                         # ②
    memory.set("research", research_data)

    return f"「{topic}」の検索完了"                                            # ③

行	やってること	なぜそうする
①	検索結果を Haiku で圧縮してから memory に保存	数千トークン → 数百トークンに。Writer が全 topic を 1 プロンプトに詰める時の余地を確保
②	`memory.research[topic] = compressed` という入れ子 dict	「研究データは研究データ専用のスペース」に整理。Writer は `memory.get("research")` で全 topic にアクセス
③	LLM に返すのは「完了」だけ	検索結果の中身を LLM の context に戻さない → 圧縮メモリの存在意義を保つ

学んだこと（要点）¶

create_supervisor + プロンプト厳格化 で「LLM-as-orchestrator」の振る舞いをかなり制御できる。「最終出力は X だけを返せ」と書かないとLLM は勝手に脚色する
シングルトンメモリでエージェント間通信は書き味は最高に楽。ただしマルチセッション安全性ゼロなので、本番なら state に移すこと
クロージャでエージェント固有ツール生成は強力なテク。get_research_todos / get_writer_todos のように LLM 視点で別ツールに見える形で内部実装を共有できる
Agent ≠ LLM。Writer agent はオーケストレーション用の Sonnet（temperature=0）だが、write_and_save_content ツール内部で 別パラメータの Sonnet（temperature=0.7）を起動して執筆させている。「Agent は LLM を内包したツール群」として書くと役割が分離できる
大小 2 段の LLM 使い分け: 圧縮は Haiku、執筆は Sonnet。Haiku 4.5 はコスト 1/12 で「要約・圧縮・分類」タスクで実用十分
TODO 駆動の有効性: Claude Code 自身も TaskCreate/TaskUpdate を使っているように、LLM の作業計画と進捗管理を外部化すると LLM の「忘却」「迷い」が減る
LangGraph supervisor の recursion_limit=100 への引き上げは深い multi-agent では必要（デフォルト 25 だと「supervisor ↔ research ↔ writer」の往復で枯渇）
langgraph.json に サブ agent を単体グラフとしても公開しておくと、Studio で単独デバッグできる（writing_assistant 全体 + task_decomposer / research / writer 単体）

拡張アイデア¶

Memory をシングルトンから state ベースに移行 — state["research"] / state["todos"] を Annotated[dict, custom_reducer] で持ち、マルチセッション安全に
TODO のステータス可視化 — Streamlit / Gradio で「TODO リストの状態 + 各 agent の進捗」をリアルタイム描画。stream_mode="updates" + subgraphs=True でイベント受信
Writer の事前レビューループ — 生成した記事を 別 LLM が「品質チェック」して NG なら再生成するループを追加。第22回の interrupt() パターン応用
検索結果のキャッシュ — search_and_save で同じ query が来たら memory から返す。Haiku 圧縮は同じ生データなら同じ結果なので、API 呼び出しを削減
複数言語対応 — system_prompt + memory のキーを language で名前空間化（research[lang][topic]）して、日本語版・英語版を並行生成
画像挿入 — Writer の write_and_save_content で記事中に ![](https://...) を生成し、Tavily Image Search で実画像 URL を取得して挿入
テストの実装 — tests/__init__.py だけある空のテスト構造を埋める。todo_manager.py の単体テスト（add_task, get_pending_tasks, update_status）から始めるのが TDD として正しい

現代版に移植するなら¶

1. API キー設定は `.env.op` + `op run` に切り替える（CLAUDE.md ルール 8）¶

連載 README は cp .env.sample .env で生 API キーを書く方式だが、グローバルルールに反するため使用禁止。代わりに:

# 24/.env.op
ANTHROPIC_API_KEY=op://Personal/anthropic-api-key/credential
TAVILY_API_KEY=op://Personal/tavily-api-key/credential
LANGCHAIN_API_KEY=op://Personal/langsmith-api-key/credential
LANGCHAIN_PROJECT=sd-24
LANGCHAIN_TRACING_V2=true

起動:

op run --env-file=.env.op -- uv run python main.py "2025年AI動向をレポートして"

op run が op:// 参照を実値に展開して子プロセスに渡す。生キーがディスクに残らない。

2. シングルトン memory → LangGraph state¶

# state を拡張
class WritingAssistantState(MessagesState):
    research: Annotated[dict[str, str], merge_research_dict]
    todos: Annotated[list[TodoItem], merge_todos]
    final_document_path: str

# tool は state を受ける形に
@tool
async def search_and_save(query: str, topic: str, state: Annotated[dict, InjectedState]) -> Command:
    ...
    return Command(update={"research": {topic: compressed}})

InjectedState + Command(update=...) で state を介した正規ルートに切り替える。マルチセッション安全 + LangGraph Studio で state 履歴が見える。

3. クロージャツール → `StructuredTool.from_function`¶

def create_get_my_todos_for_agent(agent_name: str) -> StructuredTool:
    def get_agent_todos() -> dict:
        pending_tasks = todo_manager.get_pending_tasks(agent_name)
        return {"tasks": [...], "count": len(pending_tasks)}

    return StructuredTool.from_function(
        func=get_agent_todos,
        name=f"get_{agent_name}_todos",
        description=f"{agent_name}エージェントの未完了TODOを取得",
    )

.name 属性の後付け書き換えではなく、コンストラクタで指定するほうが mypy も通る。

4. `claude-3-5-haiku-20241022` → `claude-haiku-4-5-20251001`¶

memory.py の _compressor のモデル ID を最新 Haiku に。圧縮タスクで性能差は微小だが、コスト面でほぼ同じ。

5. `print()` → `logging`¶

agent_runner.py の print("🤖 エージェント実行開始...") 群は logging.info(...) に置き換えて log level 制御可能に。RichHandler を併用すれば色付き表示はそのまま。

既知の不具合・注意点¶

シングルトン共有はテスト・マルチセッション不可: memory と todo_manager がモジュールレベルにあるため、pytest で並列テストすると state が干渉する。テスト前後で memory._store.clear() / todo_manager.todos.clear() が必要
recursion_limit=100 はやや過剰: supervisor が 3 agent と何度も往復するため上げているが、深い multi-agent 構造で 100 まで使うのは agent ループ設計のミスを覆い隠す可能性。プロファイリングして真に必要な回数を測るべき
writer.py:32-34 の task_description 引数が undocumented: 関数 docstring には書かれているが、supervisor がどう値を渡すかは LLM 任せ。曖昧な task_description が渡ると記事品質が落ちる
output/ ディレクトリのお掃除なし: 毎実行で新規 timestamp dir を作るので、たくさん試すと膨らむ。output/ の自動ローテーション or .gitignore 必須
search_and_save のエラー無視: Tavily が 401/429 を返した場合の handling がなく、response["results"] で KeyError を出して落ちる
compress_research で空 findings の場合: 空文字列が渡されても Haiku は何か出力するので「ハルシネーション要約」が memory に入る可能性。if not findings: return "" のガードが必要
final_document_path の競合: 複数の writer タスクがある場合、最後に保存したものだけが final_document_path に残る。本来は list[str] で蓄積すべき（created_files はあるが supervisor がそちらを見ない）

記事参照¶

Software Design 2025 年 9 月号（推定）連載第24回「Supervisor 応用編 — Claude Code 風文章執筆システム」
関連: 第23回 STUDY_NOTES — create_supervisor の基本
関連: 第18回 STUDY_NOTES — 自前 multi-agent。state ベースの共有がどう書かれていたか比較
関連: 第20回 STUDY_NOTES — Tavily 検索の使い方
公式 create_supervisor: https://github.com/langchain-ai/langgraph-supervisor-py
LangChain InjectedState: https://python.langchain.com/docs/how_to/tool_runtime/

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