🚴‍♂️
DS-STAR を手元環境で再現する: Google ADK で組むマルチエージェント DS パイプライン

2025/11/24に公開
 1. 導入：なぜ DS-STAR のアルゴリズム実装なのかhttps://research.google/blog/ds-star-a-state-of-the-art-versatile-data-science-agent/
現場のデータサイエンスは、「ざっとデータを見る → 仮説を立てる → コードを書く → 失敗して直す」を人間が何度も回すプロセスになりがちです。Google の DS-STAR は、これを マルチエージェントで自動化するアーキテクチャ として提案されたものです。
本記事では、その DS-STAR のアルゴリズム（タスク分解と改善ループ）を、自分の環境で動く コード実装 として再現したプロトタイプを紹介します。読み手が「自分のデータセットでも DS-STAR 風のパイプラインを試せる」よう、アルゴリズム → アーキテクチャ → 実装・コード例 → 実行例、という順でまとめます。

この記事で扱うこと
DS-STAR のコアとなるアルゴリズム構造（役割分担とループ設計）
それを Google ADK 上でどうコードとして表現したか
実際のデータセットを使った実行例と、やってみて分かった限界・トレードオフ


 2. DS-STAR の概要とアルゴリズム（公式ブログ準拠の要約）
公式ブログで示されている DS-STAR の要点は、ざっくり言うと「多様なデータを扱える、検証付きの逐次計画エージェント」です。本記事では、実装に必要な部分だけをコンパクトに押さえます。
2-1. 3つの中核アイデア

データファイル解析モジュール

多様な形式のデータファイルから、自動でコンテキスト（構造・統計情報など）を抽出する。

LLM ベースの検証ステップ

各ステップの計画や結果に対して、「これで十分か？」を判定する LLM 判定器を差し込む。

逐次的な計画メカニズム

シンプルな実行可能な計画から始め、検証のフィードバックをもとに計画とコードを反復的に洗練する。
2-2. アルゴリズムの流れ（高レベル）

ステップ 1: データファイル解析

データディレクトリを走査し、対象ファイルの構造・カラム・統計量などのコンテキストを抽出。

ステップ 2: 初期計画の作成

コンテキストとユーザ質問から、Analyzer / Planner が「前処理 → 集計 → 可視化 → モデル → 解釈」のようなステップ列を組む。

ステップ 3: 計画に基づくコード生成

Coder が計画を実行可能な Python コードに変換し、データ読み込み〜分析ロジックを 1 本のスクリプトにする。

ステップ 4: 検証とフィードバック

Verifier がコードを実行／チェックし、「この計画とコードで質問に答えられているか？」を LLM ベースで評価する。

ステップ 5: 改善ループ

Router がフィードバックを見て、「Planner に戻す（計画を変える）」か「Coder に戻す（実装だけ直す）」かを選び、必要に応じてループ。

ステップ 6: 最終レポート生成

検証を通過した結果を、Finalyzer が自然言語レポートにまとめる（結論・根拠・限界を含む）。
この抽象アルゴリズムを、このリポジトリではどのようなアーキテクチャとコードで表現しているかを、次章以降で見ていきます。

 3. このリポジトリでのアーキテクチャと実装
 3-1. アーキテクチャ全体像https://github.com/sunyeul/DS-STAR
この節では、2章のアルゴリズムの流れを、このリポジトリ内の どのファイルとエージェントが担っているか に対応づけます。
どのファイルがどのステップを担当しているか


アルゴリズムのステップ
担当エージェント / ファイル


1. データファイル解析

Analyzer / sub_agent/analyzer_agent/agent.py


2. 計画の立案（追加・修正）

Planner / sub_agent/planner_agent/agent.py


3. 計画に基づくコード生成

Coder / sub_agent/coder_agent/agent.py


4. 検証（完了判定）

Verifier / sub_agent/verifier_agent/agent.py


5. 改善ループ（計画修正の判断）

Router / sub_agent/router_agent/agent.py


6. 最終回答コード生成

Finalyzer / sub_agent/finalyzer_agent/agent.py


全体のワークフロー制御（ループ含む）

DSStarAgent / ds_star/agent.py


プロジェクト構成（最小限）

config/config.py: パスなどの基本設定

data/: 分析対象データ（例: yulu_bike_sharing_dataset.csv）

ds_star/agent.py: ルートエージェントとループ構造

ds_star/sub_agent/: 各専門エージェントの実装
ワークフローのイメージ
ユーザ質問 + データ
        │
        ▼
Analyzer  ── データ構造の把握・読み込みコード生成
        │
        ▼
┌─> Planner   ── 「次の1ステップ」を計画に追加、または既存計画を修正
│       │
│       ▼
│   Coder     ── 現在の全計画に基づくコード生成
│       │
│       ▼
│   Verifier  ── 実行＆判定 (OKなら Finalyzer へ)
│       │ NO
│       ▼
└── Router    ── 「計画修正(Modify)」か「新規追加(Add)」かを選択

        ▼ YES (Verifier OK)

Finalyzer ── 最終回答を出力するコードの生成

 3-2. 実装方針と各エージェントのコード構造まずは、全体の実装方針と実行環境について簡単に触れ、そのうえで各エージェントの「役割・入出力・実装パターン」を見ていきます。

 3-2-1. Analyzer Agent：データをどう理解させるか役割

data/ 配下の CSV などを読み込み、データ構造の要約＋読み込みコード を出す
カラム名・型・欠損・簡単な統計量を見て、「どんなデータか」をざっくり説明する
入出力のイメージ
入力: filename と、そのファイルのサンプル行・カラム一覧・簡単な統計情報
出力: 自然言語の要約テキスト＋pandas ベースの読み込みコード
実装上のポイント
LLM には「まず概要 → その後コード」というフォーマットを強制
大きなデータはサンプリングし、統計情報も必要最低限だけ渡す
型推論や欠損の扱いは完璧を狙わず、Verifier 側での補正を前提にする
analyzer_agent のコード例（抜粋）
from google.adk.agents import LlmAgent
from .models import ExecutableCode
from .callback import after_analyzer_agent_callback

# エージェント定義
analyzer_agent = LlmAgent(
    name="analyzer_agent",
    model="gemini-2.5-flash",
    instruction="""
あなたは専門のデータ分析者です。
{filename}の内容を読み込み、説明するPythonコードを生成してください。

# 要件
- ファイルは構造化データまたは非構造化データのいずれでも可能です。
- 構造化データが多すぎる場合は、数例のみを出力してください。
- 重要な情報を出力してください。例えば、すべての列名を出力してください。
- Pythonコードは{filename}の内容を出力する必要があります。
- コードは単一ファイルのPythonプログラムで、自己完結型であり、そのまま実行可能である必要があります。
- 応答には単一のコードブロックのみを含めてください。
- 重要: エラーが発生した場合にデバッグするため、ダミーの内容を含めないでください。
- エラーを防ぐためにtry:とexcept:を使用しないでください。後でデバッグします。
    """,
    description="ファイルの内容を分析し、内容を説明するPythonコードを生成します。",
    output_key="executable_code",
    output_schema=ExecutableCode,
    disallow_transfer_to_parent=True,
    disallow_transfer_to_peers=True,
    after_agent_callback=after_analyzer_agent_callback,
)

 3-2-2. Planner Agent：逐次的な計画立案役割
これまでの計画と実行結果を踏まえ、「次に行うべき1ステップ」 だけを提案します。
一度に全工程を計画するのではなく、ループごとに計画リスト（state["plans"]）に手順を積み上げていく方式です。
入出力のイメージ
入力: ユーザ質問、これまでの計画リスト、直前のコード実行結果
出力: 自然言語による次のステップの説明（JSONではなくテキスト）
実装上のポイント
初回は initial_planner_impl、2回目以降は planner_impl が動作し、文脈に応じた計画を立てます。
コールバック関数 append_plan で、生成されたテキストをセッション状態のリストに追加しています。
planner_agent のコード例（抜粋）
class PlannerAgent(BaseAgent):
    """
    初期計画と後続計画の策定を動的に処理する統合プランナーエージェント。
    """

    initial_agent: Agent
    next_step_agent: Agent

    def __init__(self, name: str = "planner_agent"):
        initial_agent = Agent(
            model="gemini-2.5-flash",
            name="initial_planner_impl",  # 内部識別用の名前
            description="最初のステップを計画する初期プランナーエージェント。",
            instruction="""
            あなたは専門のデータ分析者です。
            与えられたデータに基づいて事実質問に答えるためには、まず効果的に計画を立てる必要があります。
            """,
            output_key="current_plan",
            before_model_callback=initial_planner_before_model_callback,
            after_agent_callback=append_plan,
        )

        next_step_agent = Agent(
            model="gemini-2.5-flash",
            name="planner_impl",  # 内部識別用の名前
            description="次のステップを計画するプランナーエージェント。",
            instruction="""
            あなたは専門のデータ分析者です。
            与えられたデータに基づいて事実質問に答えるためには、まず効果的に計画を立てる必要があります。
            あなたのタスクは、質問に答えるために次に行うべき計画を提案することです。
            """,
            output_key="current_plan",
            before_model_callback=planner_before_model_callback,
            after_agent_callback=append_plan,
        )
        super().__init__(
            name=name, initial_agent=initial_agent, next_step_agent=next_step_agent
        )

    async def _run_async_impl(
        self, ctx: InvocationContext
    ) -> AsyncGenerator[Event, None]:
        """
        現在のセッション状態（計画の存在有無）を確認し、適切なエージェントに実行を委譲します。
        """
        plans = ctx.session.state.get("plans", [])

        if len(plans) == 0:
            selected_agent = self.initial_agent
        else:
            selected_agent = self.next_step_agent

        async for event in selected_agent.run_async(ctx):
            yield event
# callbacks.py より抜粋
def append_plan(callback_context: CallbackContext) -> Optional[types.Content]:
    state = callback_context.session.state
    current_plan = state.get("current_plan")  # LLMが生成した自然言語のステップ
    if "plans" not in state:
        state["plans"] = [current_plan]
    else:
        state["plans"].append(current_plan)  # リストに追記
    return None

 3-2-3. Coder Agent：累積的なコード生成
役割
現在の計画リスト（state["plans"]）の すべて を満たす Python コードを毎回生成（または更新）します。


入出力のイメージ
入力: 全ての計画ステップ (plans)、Analyzer が生成した読み込みコード
出力: base_code（そのまま実行できる Python コード）


実装上のポイント
ステップごとの差分生成ではなく、「その時点での完全なスクリプト」 を出力させることで、依存関係の断絶を防ぎます。
前回のコードがある場合は next_step_agent、ない場合は initial_agent が担当します。


 3-2-4. Verifier & Router：判定と軌道修正役割

Verifier: 生成コードを実行し、その結果が「質問に答えるのに十分か」を True/False で判定します。

Router: Verifier が NG を出した場合、「既存の計画を修正する (Modify Plan)」か「新しい計画を追加する (Add Plan)」かを決定します。
実装上のポイント
Verifier は詳細なエラーレポートを出力せず、実行ログ（result）をコンテキストとして共有します。

Router から Coder への直接バックトラックはありません。実装ミスであっても、「計画を見直す（例: ステップを具体化する）」という形で Planner を経由して修正を行います。
Verifier / Router のエージェント定義（抜粋）
verifier_agent = Agent(
    model="gemini-2.5-flash",
    name="verifier_agent",
    description="現在の計画とそのコード実装が質問に答えるのに十分かどうかを検証する検証エージェント。",
    instruction="""
あなたは専門のデータ分析者です。
あなたのタスクは、現在の計画とそのコード実装が質問に答えるのに十分かどうかを確認することです。
""",
    output_key="is_enough",
    output_schema=IsEnough,
    disallow_transfer_to_parent=True,
    disallow_transfer_to_peers=True,
    before_model_callback=verifier_before_model_callback,
    after_agent_callback=after_verifier_agent_callback,
)

router_agent = Agent(
    model="gemini-2.5-flash",
    name="router_agent",
    description="質問を適切なエージェントにルーティングするルーターエージェント。",
    instruction="""
あなたは専門のデータ分析者です。
現在の計画では質問に答えるのに不十分なため、あなたのタスクは質問に答えるために計画をどのように改善するかを決定することです。
""",
    output_key="router_output",
    output_schema=RouterOutput,
    disallow_transfer_to_parent=True,
    disallow_transfer_to_peers=True,
    before_model_callback=router_before_model_callback,
)
Verifier / Router の出力モデル
# verifier_agent/models.py
class IsEnough(BaseModel):
    result: bool = Field(
        ...,
        description="Whether the current plan and its code implementation is enough to answer the question.",
    )

# router_agent/models.py
class RouterOutput(BaseModel):
    choice: Literal["Add Plan", "Modify Plan"] = Field(...)
    plan_number: Optional[int] = Field(..., description="The number of the plan to modify.")

 3-2-5. Finalyzer：最終回答コードの生成役割
自然言語のレポートを書くのではなく、「質問への最終回答を出力する Python コード」 を生成します。
これまでの試行錯誤で得られた base_code と実行結果を参考に、クリーンな回答用スクリプトを作成・実行します。
finalyzer_agent のコード例（抜粋）
finalyzer_agent = Agent(
    model="gemini-2.5-flash",
    name="finalyzer_agent",
    description="質問への回答を最終化するファイナライザーエージェント。",
    instruction="""
あなたは専門のデータ分析者です。
以下にリストされたファイル/ドキュメントを読み込み、参照して事実質問に答えます。
また、参照コードも持っています。
あなたのタスクは、与えられたガイドラインに従って質問の答えを出力するソリューションコードを作成することです。
""",
    output_key="final_answer",
    before_model_callback=finalyzer_before_model_callback,
)

 4. 実際に動かしてみる：Yulu Bike Sharing データセットでの実行例https://www.kaggle.com/datasets/ranitsarkar01/yulu-bike-sharing-data
題材として使用したのは、都市型シェアサイクルの利用状況を記録した data/yulu_bike_sharing_dataset.csv です。日時や天候、気温、各時間帯のレンタル台数といったカラムを持つ、典型的な時系列＋環境情報のデータセットです。

 4-1. 実行手順（ざっくり）Python 3.13 環境を用意し、uv sync で依存関係をインストール

uv run adk web でADKのWeb UIを起動
「state」にquestionを設定し、エージェントを起動

 4-2. 試した質問例
 4-2-0. 質問「Business Case のデータセットを使って、天候によって自転車のレンタル数に差があるかどうか仮説検定してもらえますか？」

 4-2-1. データファイル分析結果データの先頭5行（head）
全12列のカラム名
各列のデータ型と Non-Null 件数（info 出力）
数値列の記述統計（describe）
季節・天気・休日・勤務日などのカテゴリ列の value_counts

 4-2-2. 1回目計画&コード生成&判定weather=4 のデータが 1件しかない

→ 平均比較や統計検定の前提（分布・分散）が成立しない
よって 単純平均を出しただけでは仮説検定の根拠にならない
「これだけでは “天候で差がある” と判断できる材料が不足」という判断

 4-2-3. 2回目計画&コード生成&判定weather=4（データ1件）を除外
weather=1,2,3 の 3 群で 一元配置分散分析（ANOVA） を実施
ANOVA 結果 → F=98.2836, p=0.0000
有意水準 0.05 より十分小さい

→ 帰無仮説（“天候で平均レンタル数は同じ”）を棄却
天候カテゴリ（1,2,3）の間でレンタル数に統計的に有意な差がある

→ この結論はあなたの質問に対して必要十分。

 4-2-5. 最終回答F-statistic: 98.2836
P-value: 0.0000
P値 (0.0000) は有意水準 (0.05) より小さいため、帰無仮説を棄却します。
少なくともいずれかの天候カテゴリ間で自転車レンタル数の平均に統計的に有意な差があります。
import pandas as pd
from scipy import stats

# データセットの読み込み
df = pd.read_csv('data/yulu_bike_sharing_dataset.csv')

# weatherカテゴリが4のデータを除外
# 与えられた説明によると、weatherカテゴリ4は"Heavy Rain + Ice Pallets + Thunderstorm + Mist, Snow + Fog"を意味し、データが1件しかないため分析から除外するのが適切です。
df_filtered = df[df['weather'] != 4]

# weatherカテゴリごとに'count'のデータを抽出
weather_group_1 = df_filtered[df_filtered['weather'] == 1]['count'] # Clear, Few clouds, Partly cloudy, Partly cloudy
weather_group_2 = df_filtered[df_filtered['weather'] == 2]['count'] # Mist + Cloudy, Mist + Broken clouds, Mist + Few clouds, Mist
weather_group_3 = df_filtered[df_filtered['weather'] == 3]['count'] # Light Snow, Light Rain + Thunderstorm + Scattered clouds, Light Rain + Scattered clouds

# 一元配置分散分析（ANOVA）を実施
# 帰無仮説 (H0): 全ての天候カテゴリ間で自転車レンタル数の平均に差はない。
# 対立仮説 (H1): 少なくともいずれかの天候カテゴリ間で自転車レンタル数の平均に統計的に有意な差がある。
f_statistic, p_value = stats.f_oneway(weather_group_1, weather_group_2, weather_group_3)

# P値の解釈
alpha = 0.05
print(f"F-statistic: {f_statistic:.4f}")
print(f"P-value: {p_value:.4f}")

if p_value < alpha:
    print(f"P値 ({p_value:.4f}) は有意水準 ({alpha}) より小さいため、帰無仮説を棄却します。\n少なくともいずれかの天候カテゴリ間で自転車レンタル数の平均に統計的に有意な差があります。")
else:
    print(f"P値 ({p_value:.4f}) は有意水準 ({alpha}) より大きいため、帰無仮説を棄却できません。\n天候カテゴリ間で自転車レンタル数の平均に統計的に有意な差があるとは言えません。")

 5. 課題とトレードオフ
良かった点
マルチエージェント構造にしたことで、プロンプトと責務の分離 がかなりやりやすくなった
「どの役割が失敗しているか」が切り分けやすく、デバッグもしやすい


難しかった点
品質の鍵は Analyzer / Router にあり、ここが弱いと全体も脆くなる
「それっぽいが間違ったコード」をどこまで自動検知・修正できるかがボトルネック

→ 元論文では「debugging agent」があるので、活用できるかも知らない
UIを含めたワークフロー導線の設計が大変だった

→ 本記事では「Google ADK Web UI」をメインに使っているが、専用のUIやCLIとして実装した方がより実践的に使えると思う


コストと現実解
ループを増やすほど API コールや実行コストが増えるため、最大 5 回で打ち切る設計にした
元論文では gemini-2.5-proとgpt-5を使用しているが、コスト面でgemini-2.5-flashを採用
論文どおりの完全再現ではなく、「自分の環境で現実的に回せるライン」を優先したシンプル構成にしている


 今後の発展アイデア
データファイル解析ステップの分離
データフォルダーのファイルを事前に解析して解析ステップとメインループを分離する


機能拡張の方向性
可視化専用エージェントやモデル選択専用エージェントなど、さらに役割を細分化する
評価指標の自動設計や、複数モデル比較の自動化を組み込む


運用・プロダクション視点
Web UI／ダッシュボード化して、「質問を投げるだけ」でレポートが返る形にする
エージェントを API 化し、既存のデータパイプラインから呼び出せるようにする


研究的な方向性
Human-in-the-loop で「どこまで人が介入すべきか」を設計する
判断や生成物を記録し、後から監査・説明できるようにする（説明可能性・安全性）
アルゴリズムのステップ	担当エージェント / ファイル
1. データファイル解析	`Analyzer` / `sub_agent/analyzer_agent/agent.py`
2. 計画の立案（追加・修正）	`Planner` / `sub_agent/planner_agent/agent.py`
3. 計画に基づくコード生成	`Coder` / `sub_agent/coder_agent/agent.py`
4. 検証（完了判定）	`Verifier` / `sub_agent/verifier_agent/agent.py`
5. 改善ループ（計画修正の判断）	`Router` / `sub_agent/router_agent/agent.py`
6. 最終回答コード生成	`Finalyzer` / `sub_agent/finalyzer_agent/agent.py`
全体のワークフロー制御（ループ含む）	`DSStarAgent` / `ds_star/agent.py`
1. 導入：なぜ DS-STAR のアルゴリズム実装なのか

2. DS-STAR の概要とアルゴリズム（公式ブログ準拠の要約）

3. このリポジトリでのアーキテクチャと実装

3-1. アーキテクチャ全体像

3-2. 実装方針と各エージェントのコード構造

3-2-1. Analyzer Agent：データをどう理解させるか

3-2-2. Planner Agent：逐次的な計画立案

3-2-3. Coder Agent：累積的なコード生成

3-2-4. Verifier & Router：判定と軌道修正

3-2-5. Finalyzer：最終回答コードの生成

4. 実際に動かしてみる：Yulu Bike Sharing データセットでの実行例

4-1. 実行手順（ざっくり）

4-2. 試した質問例

4-2-0. 質問

4-2-1. データファイル分析結果

4-2-2. 1回目計画&コード生成&判定

4-2-3. 2回目計画&コード生成&判定

4-2-5. 最終回答

5. 課題とトレードオフ

今後の発展アイデア

Discussion
