STUDY NOTES

第17回: LangGraph Studio — サブグラフによるマルチエージェント構成の可視化・デバッグ¶

第16回が「LangGraph Platform に乗せて API として配信する」回だったのに対し、第17回はその開発体験を支える LangGraph Studio（ローカル GUI / クラウド GUI でグラフの実行・ステート可視化・タイムトラベルができる開発ツール）にフォーカスする回。サンプル本体は 3つのグラフを langgraph.json で並列に Studio に登録し、サブグラフ構造のまま可視化する ことを狙ったコードになっている。

全体像¶

langgraph.json で3つのグラフを別個に Studio へ登録しているのがこの回最大の仕掛け。

{
  "graphs": {
    "agent": "./my_agent/agent.py:graph",              // 親グラフ（全体オーケストレーション）
    "task_planner": "./my_agent/task_planner_agent.py:graph",   // サブグラフ単体
    "task_executor": "./my_agent/task_executor_agent.py:graph"  // サブグラフ単体
  }
}

これにより Studio 上で「親グラフを実行してエンドツーエンドで挙動を見る」「task_planner だけを切り出して入力を投げ込み、改善ループを単体デバッグする」のどちらもできる。サブグラフをライブラリのように使いつつ単独でも回せる設計。

親グラフ agent.py のオーケストレーションは「タスク計画 → 並列タスク実行 → レポート」という古典的な研究エージェントパターン（第9回 Research Agent の発展形）。

flowchart TD
    Start([START]) --> Planner["task_planner<br/>（サブグラフ）<br/>タスク分解 + 自己レビュー"]
    Planner --> Executor["task_executor<br/>（サブグラフ）<br/>各タスクを並列 ReAct 実行"]
    Executor --> Reporter["reporter<br/>結果を整理してレポート化"]
    Reporter --> End([END])

    subgraph Planner_Internal["task_planner の内部"]
        P_Start([START]) --> Decompose["decompose_query<br/>LLM で 3〜5 タスクに分解"]
        Decompose --> Check["check_approval<br/>LLM で自己レビュー"]
        Check -->|is_approved=True| P_End([END])
        Check -->|is_approved=False| Decompose
    end

    subgraph Executor_Internal["task_executor の内部"]
        E_Start([START]) -->|Send×N| Exec["execute_task<br/>create_react_agent + Tavily"]
        Exec --> E_End([END])
    end

    Planner -.-> Planner_Internal
    Executor -.-> Executor_Internal

データの流れ（State の受け渡し）:

sequenceDiagram
    participant User
    participant Parent as agent.py (AgentState)
    participant Planner as task_planner subgraph
    participant Executor as task_executor subgraph
    participant Reporter

    User->>Parent: messages=[HumanMessage("...")]
    Parent->>Planner: AgentInputState{messages}
    Note over Planner: 自己レビューループ<br/>（reason をフィードバック）
    Planner-->>Parent: {tasks: ["...", "...", ...]}
    Parent->>Executor: {tasks}
    Note over Executor: Send で N 並列起動<br/>各タスク = 独立 ReAct エージェント
    Executor-->>Parent: {results: [...]} ※operator.add で集約
    Parent->>Reporter: {results}
    Reporter-->>User: {final_output: "..."}

ポイントは 親 State と子 State でキー名を合わせると、子サブグラフは親の State から該当キーだけを自動で受け取り、自分の出力キーを親に書き戻す こと。子サブグラフ独自の private state（is_approved, reason など）は親には漏れない。AgentState を Input / Private / Output の 3 つに分けて多重継承する設計は、まさにこの「外と内を分離する」ためのテクニック。

使用ライブラリ・原理¶

1. LangGraph Studio とは¶

LangChain 社が提供する LangGraph 専用の開発 GUI。VS Code の Jupyter のような立ち位置で、グラフを書き換えるたびにブラウザで以下が確認できる:

グラフ構造の可視化（ノード・エッジを Mermaid 風にレンダリング、サブグラフは折りたたみ展開可能）
State の時系列ビュー（各ノード実行後の State スナップショットがタイムラインで並ぶ）
タイムトラベルデバッグ（途中の State を編集して、そこから再実行できる）
Threads（会話履歴）（MemorySaver などの checkpointer で持っている対話履歴を一覧）
入力フォーム自動生成（Input 用 TypedDict から JSON エディタを生成）

起動は uv run langgraph dev。これは内部で langgraph-cli[inmem] を使い、ローカルプロセスとしてグラフを LangGraph API サーバ相当で立ち上げ、ブラウザは https://smith.langchain.com/studio/?baseUrl=http://localhost:2024 のような URL でローカル API を覗きに行く（GUI 本体は LangSmith ホスト、データはローカル）。

第16回の langgraph up は本番デプロイ用の Docker サーバ起動、第17回の langgraph dev は開発時の in-memory サーバ起動、という対比で覚えると分かりやすい。

2. サブグラフ（Subgraph）¶

LangGraph のコンパイル済みグラフを、別のグラフのノードとしてそのまま add_node できる機能。sample/subgraph_sample.py がこの最小例:

compiled_subgraph = subgraph_builder.compile()
parent_builder.add_node("subgraph_node", compiled_subgraph)  # ← グラフをノード扱い

なぜ便利か:

独立してテストできる（サブグラフ単体に input を投げて結果を見る、Studio で別グラフとして登録できる）
State スコープを限定できる（サブグラフ内の private state は親に漏らさない）
再利用できる（同じ「タスク計画ループ」を別の親に差し込める）

parent_graph.stream(..., subgraphs=True) を指定すると、サブグラフ内部のノード単位でもストリームイベントが流れる。Studio はこれを使ってサブグラフの内部状態も逐次描画している。

3. State の 3 層分割パターン（Input / Private / Output）¶

LangGraph 0.2 系で導入された StateGraph(state_schema=..., input=..., output=...) の 3 引数を活用するパターン。

class AgentInputState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]

class AgentPrivateState(TypedDict):
    tasks: list[str]
    results: list[str]

class AgentOutputState(TypedDict):
    final_output: str

class AgentState(AgentInputState, AgentPrivateState, AgentOutputState):
    pass

graph = StateGraph(
    state_schema=AgentState,    # 内部処理用（全部入り）
    input=AgentInputState,      # 外から受け取る形（messages のみ）
    output=AgentOutputState,    # 外に返す形（final_output のみ）
)

何が嬉しいか:

Studio の入力フォームが messages だけになる（tasks や results を手で入れる必要がない）
API レスポンスが final_output だけになる（中間状態が漏れない）
内部実装を変えても外部インターフェースは固定できる

Pydantic / FastAPI で言う「Request DTO / Internal Model / Response DTO」の使い分けと同じ発想。

4. `operator.add` による並列結果の集約¶

task_executor_agent.py の出力 state:

class TaskExecutorAgentOutputState(TypedDict):
    results: Annotated[Sequence[str], operator.add]

Annotated[..., operator.add] は LangGraph に「このキーは複数ノードから書き込まれる可能性があるから、上書きではなく + で結合せよ」と指示するリデューサ宣言。リスト同士なら [a] + [b] = [a, b] で連結される。

これがないと、Send で起動した N 並列ノードが最後の1個の結果だけで上書きしてしまう。add_messages も同じ仕組みで、メッセージ ID を見て賢くマージするリデューサ。

5. `Send` による動的並列 fan-out¶

def routing_parallel_nodes(self, state) -> list[Send]:
    return [Send("execute_task", {"task": task}) for task in state.get("tasks", [])]

graph.add_conditional_edges(START, self.routing_parallel_nodes, ["execute_task"])

Send(ノード名, そのノードに渡す state) を リストで返すと、LangGraph はそのノードを タスク数だけ並列に起動する。実行時にしか個数が決まらない fan-out を実現するための仕組み。

通常の add_conditional_edges が「次の 1 ノードを選ぶ」のに対し、Send のリストは「次の N ノードを全部起動する」。MapReduce の Map に相当。各並列実行の State は ParallelState{task: str} で隔離されているので、互いに干渉しない。

6. `create_react_agent`（prebuilt）¶

from langgraph.prebuilt import create_react_agent
agent = create_react_agent(self.llm, self.tools)
result = agent.invoke(messages)

第11回でフルスクラッチで書いた ReAct ループ（Thought → Action → Observation の while ループ）を 1 行で作れる prebuilt 関数。内部では tools_condition ノードと ToolNode を組み合わせた小さな StateGraph を返す。自前で再実装したくないが ReAct パターンが必要なときに使う、LangGraph の「電池付き」エントリーポイント。

7. `with_structured_output` による型安全な LLM 出力¶

chain = prompt | self.llm.with_structured_output(DecomposedTasks)

Pydantic モデル（DecomposedTasks）を渡すと、LLM の Function Calling 機能を使って モデルの schema に沿った JSON を生成 → 自動でパース → Pydantic インスタンスを返してくれる。min_length=3, max_length=5 のような Pydantic バリデーションも効くので、「タスクは 3〜5 個」という制約をプロンプト文だけでなく型で強制できる。

ファイル別の役割¶

ファイル	役割
`langgraph.json`	Studio に登録するグラフの宣言。`graphs` に複数登録すると Studio 上でグラフ切り替え UI が出る
`my_agent/agent.py`	親グラフ本体。Planner サブグラフ → Executor サブグラフ → Reporter の直線パイプライン。`checkpointer=MemorySaver()` で会話履歴を保持
`my_agent/task_planner_agent.py`	サブグラフ①。LLM でタスク分解 → LLM 自身でレビュー → 不合格なら理由を添えてリトライする自己改善ループ
`my_agent/task_executor_agent.py`	サブグラフ②。`Send` でタスク数ぶん `create_react_agent` を並列起動。Tavily 検索ツール付き
`sample/subgraph_sample.py`	サブグラフ最小例。State の受け渡しがどう動くかを `stream(subgraphs=True)` で観察するための教材
`pyproject.toml`	`langgraph-cli[inmem]` を dev 依存に入れている点が第17回のポイント（`langgraph dev` で必要）
`.env.sample`	OPENAI / TAVILY / LANGSMITH のキー。Studio は LANGSMITH 経由で動くため LANGSMITH_API_KEY は実質必須

行レベルの工夫¶

`agent.py:31` — 多重継承で全部入り State を作る¶

class AgentState(AgentInputState, AgentPrivateState, AgentOutputState):
    pass

TypedDict の多重継承は キーの和集合 になる。コードを書き散らかさずに「内部処理用は全キー、外向きは部分集合」を表現できる Python のテクニック。

`agent.py:78` — `MemorySaver()` を親グラフだけに付ける¶

return graph.compile(checkpointer=MemorySaver())

checkpointer は親グラフだけに付ければよく、サブグラフは親の checkpointer を継承する。Studio の Threads タブで会話履歴が見えるのはこの設定があるから。サブグラフ側（task_planner / task_executor の .compile()）には付けていないことに注目。

`task_planner_agent.py:50` — タスク再生成時に `reason` をクリア¶

return {"tasks": decomposed_tasks.tasks, "reason": ""}

reason を残したまま次のレビューに進むと、「以前のダメ出し理由」と「今回のレビュー結果」が混ざる。明示的にクリアすることで、状態機械として「reason はレビュー結果を decompose に渡すための一時バッファ」という役割が明確になる。

`task_planner_agent.py:154-161` — `add_conditional_edges` の辞書ルーティング¶

graph.add_conditional_edges(
    "check_approval",
    lambda state: state["is_approved"],   # bool を返す
    {True: END, False: "decompose_query"},  # bool → 次ノード
)

条件関数の返り値で辞書を引いて次ノードを決める。True/False のような ENUM 的な値で分岐するときに、if-else を書くより意図が読み取りやすい。

`task_executor_agent.py:69-71` — `Send` の対象ノードを `["execute_task"]` で明示¶

graph.add_conditional_edges(
    START, self.routing_parallel_nodes, ["execute_task"]
)

第3引数の ["execute_task"] は Studio に「このルーティング関数は execute_task に向かう可能性がある」と教えるためのヒント。これがないと Studio はグラフ構造を静的に解析できず、START からのエッジが描画されない。実行時の挙動には影響しないが、可視化のために必須。

`task_executor_agent.py:17` — `Sequence[str]` + `operator.add`¶

results: Annotated[Sequence[str], operator.add]

list[str] でも動くが、不変的な印象を与える Sequence を使うのは LangGraph 公式サンプルの慣習。リデューサが operator.add なので、各並列ノードが {"results": [単一の結果]} を返すと、全結果が連結された 1 本のリストになる。

`agent.py:88` — `xray=2` でサブグラフを展開描画¶

png = graph.get_graph(xray=2).draw_mermaid_png()

xray=N はサブグラフを N 階層まで展開して描画する引数。xray=0 だとサブグラフは黒い箱、xray=2 だと孫サブグラフまで内部が見える。Studio の「Expand subgraphs」トグルと同じ役割。

学んだこと（要点）¶

langgraph dev（第17回）と langgraph up（第16回）は別物。前者は開発用の in-memory サーバ＋Studio GUI、後者は本番用の Docker サーバ。Studio は LangSmith にホストされた GUI がローカル API を叩く構成
サブグラフは「コンパイル済みグラフを add_node するだけ」。子の Input/Output キー名を親と合わせれば、State の受け渡しは自動
State の Input/Private/Output 分割は Studio の UX に直接効く。入力フォームが綺麗になり、API レスポンスが軽くなる
Send は MapReduce の Map。リストを返すと並列 fan-out、結果は operator.add リデューサで集約
langgraph.json に複数グラフを登録すると、サブグラフ単独デバッグが可能。これが Studio 時代のテスト戦略の根幹
xray=N でサブグラフ展開描画。デバッグ時に便利
自己レビューループ（decompose → review → 不合格ならリトライ）は、Pydantic の構造化出力 + 条件分岐 + private state（reason）の組み合わせで素直に書ける。第8回・第10回 CRAG の「自己評価でリトライ」パターンを LangGraph 流に再実装したものとして読める

拡張アイデア¶

Human-in-the-loop の追加 — task_planner の自己レビューだけでなく、interrupt() を使って人間に承認させる中断ポイントを追加し、Studio の Thread UI から resume する体験を試す
タスク実行結果の品質チェックノード — task_executor のあとに「結果が薄い場合は Tavily 検索の max_results を増やしてリトライ」する自己改善ループを足し、ネストした自己評価エージェントにする
複数の検索ツール混在 — Tavily に加えて WikipediaQueryRun や DuckDuckGoSearchResults を ReAct エージェントに渡し、タスク内容に応じてツールを使い分ける挙動を観察する
Reporter を Structured Output 化 — Reporter の最終出力を Markdown 文字列ではなく、title / sections / citations を持つ Pydantic モデルにし、API として消費しやすくする
AgentState.tasks への並列書き込み実験 — 現状は Planner が一括で書くだけだが、Planner も Send で並列化して tasks に operator.add で追記する形に変え、リデューサの挙動を体感する

現代版に移植するなら¶

langgraph>=0.2.50 指定だが、執筆時点（2026年）では langgraph>=0.6 系が安定。MemorySaver の import パスは langgraph.checkpoint.memory のままで OK だが、本番では SqliteSaver / PostgresSaver を選ぶのが現実解
TavilySearchResults は langchain_community.tools.tavily_search から import しているが、現代版では langchain-tavily パッケージ（from langchain_tavily import TavilySearch）に切り出されている。LangChain Community の非推奨化に注意
create_react_agent のシグネチャは langgraph>=0.3 で prompt 引数（system プロンプト）が追加されている。messages の冒頭に human で指示を埋め込む現コードより、prompt= で system message として分離するほうが現代的
langgraph.json の dependencies: ["."] は現行 CLI でも有効だが、依存解決を pyproject.toml 任せにするほうが運用しやすい
langchain-anthropic が dependency に入っているが、本サンプルでは未使用。実際に Claude を使うなら ChatAnthropic(model="claude-sonnet-4-6") に差し替えるだけで動く（with_structured_output も Anthropic 側でサポート済み）
OpenAI のモデル名 gpt-4o-mini は 2024 年時点の選択。現代版では gpt-4.1-mini / gpt-5-mini などコスト効率のよい新モデルへ差し替え可

既知の不具合・注意点¶

agent.py:1 で operator を import しているが本ファイルでは未使用（リファクタの取り残し）
task_planner_agent.py:98 で chain.invoke({"messages": ..., "existing_tasks": ...}) を渡しているが、プロンプトテンプレートに {existing_tasks} プレースホルダがないため、この引数は黙って捨てられる。動作はするが意図不明瞭
MemorySaver はプロセス終了で消えるため、langgraph dev を再起動すると Studio の Thread 履歴も全消去される。永続化したいときは別 checkpointer に差し替える必要あり
Studio 上で「Submit」する際、AgentInputState の messages は [{"role": "human", "content": "..."}] 形式の JSON を入力する必要がある（生の文字列ではない）
LANGSMITH_TRACING_V2=true は古い変数名。現代版は LANGSMITH_TRACING=true。両方併設しておくのが安全

記事参照¶

Software Design 2025年1月号連載第17回「LangGraph Studio で開発する」
関連サンプル: 第9回（Research Agent の素朴版）、第16回（LangGraph Platform / langgraph up）、第11回（ReAct を手書き）と読み比べると、本回が「LangGraph 開発体験の総決算」であることが分かる

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

STUDY NOTES

第17回: LangGraph Studio — サブグラフによるマルチエージェント構成の可視化・デバッグ¶

全体像¶

使用ライブラリ・原理¶

1. LangGraph Studio とは¶

2. サブグラフ（Subgraph）¶

3. State の 3 層分割パターン（Input / Private / Output）¶

4. operator.add による並列結果の集約¶

5. Send による動的並列 fan-out¶

6. create_react_agent（prebuilt）¶

7. with_structured_output による型安全な LLM 出力¶