社内 BigQuery にアクセスして Deep Research してくれるエージェントを作る

こんにちは！コミューン株式会社で機械学習エンジニアをしている深澤です。本記事はCommune Developers Advent Calendar 2025 の 25 日目の記事です。

コミューンの社内分析用 AI エージェント; Community Sage

所属するコミューン株式会社には google-adk を用いた社内分析用の AI エージェントがいます。

雑な聞き方をしてもいい感じに数字を引っ張ってきてくれる、とてもいいやつです。作ってよかったなと思っています。

さて、このエージェントは google-adk の BigQuery に接続するパターンの example をもとに作りました。

https://google.github.io/adk-docs/tools/built-in-tools/#vertex-ai-search

プロンプトはもちろん変えたり、subagent as tool で渡しているものがあったりなどしてはいるものの、ほとんどサンプル通りに作ってすっと動いています。ユーザインタフェースを提供するところには労力がかかりますが、簡単なインタフェースを作るだけなら streamlit か fastapi + htmx ですぐできるので、皆さんもぜひ試してみてください。

BigQuery にアクセスしてもらう Deep Research Agent が欲しい

サンプルをトレースするだけで分析エージェントが作れるのであれば、Deep Research っぽいものもすぐに作れるのでは？と思ってサンプルを覗いてみると、web search を行うものはいくつか見受けられます。こちらなど → https://github.com/google/adk-samples/tree/main/python/agents/deep-search

一方で、BigQuery にアクセスするパターンのサンプルはなさそうでした。ですが、そこまで苦労せずにできるのでは？と思い、作ってみました。

全体構成

google-adk には SequentialAgent と LoopAgent という、複数のエージェントを組み合わせるためのユーティリティがあります。今回はこれらを使い、以下のようなフローで Deep Research を実現しました。

Plan → [Get Data → Reflection] x N → Deep Insight Report

擬似コードで書くとこんな感じです:

from google.adk.agents import LlmAgent, LoopAgent, SequentialAgent
from google.adk.tools.bigquery import BigQueryToolset, BigQueryToolConfig

MAXIMUM_BYTES_BILLED = 1 * 1024 * 1024 * 1024  # 1GBに制限
tool_cfg = BigQueryToolConfig(
    write_mode=WriteMode.BLOCKED,
    maximum_bytes_billed=MAXIMUM_BYTES_BILLED,
)
bq_toolset = BigQueryToolset(bigquery_tool_config=tool_cfg)

# Phase 1: 分析計画を立てる
planner_agent = LlmAgent(
    name="PlannerAgent",
    instruction="ユーザの質問に答えるための分析計画を立ててください...",
    output_key="analysis_plan",
)

# Phase 2-1: データを取得する
data_collector_agent = LlmAgent(
    name="DataCollectorAgent",
    instruction="計画に従ってBigQueryからデータを取得してください...",
    tools=[bq_toolset],
    output_key="current_data",
)

# Phase 2-2: 振り返りを行い、次のアクションを決める
reflection_agent = LlmAgent(
    name="ReflectionAgent",
    instruction="取得したデータを解釈し、十分な洞察が得られたか判断してください...",
    tools=[record_finding, mark_analysis_complete, exit_loop],
    output_key="reflection",
)

# Phase 2: データ取得と振り返りをループ
analysis_loop = LoopAgent(
    name="AnalysisLoop",
    sub_agents=[data_collector_agent, reflection_agent],
    max_iterations=10,
)

# Phase 3: 最終レポートを作成
insight_report_agent = LlmAgent(
    name="InsightReportAgent",
    instruction="これまでの発見をまとめ、深い洞察レポートを作成してください...",
    output_key="insight_report",
)

# すべてを直列につなげる
root_agent = SequentialAgent(
    name="deep_community_sage_agent",
    sub_agents=[
        planner_agent,       # Phase 1
        analysis_loop,       # Phase 2 (ループ)
        insight_report_agent # Phase 3
    ],
)

ポイント: LoopAgent と exit_loop

Deep Research の肝は「十分な情報が集まるまで繰り返しデータを取得する」という部分です。google-adk の LoopAgent は、サブエージェントを最大 N 回まで繰り返し実行してくれます。

ただし、毎回 N 回ループするのは無駄なので、「もう十分」と判断したら早期に抜ける仕組みが必要です。これには exit_loop ツールを使います。

from google.adk.tools.exit_loop_tool import exit_loop

reflection_agent = LlmAgent(
    name="ReflectionAgent",
    instruction="""
    取得したデータを評価してください。
    - 十分な洞察が得られた場合: exit_loop ツールを呼び出してループを終了
    - まだ不十分な場合: 次に調べるべきことを提案
    """,
    tools=[exit_loop],  # これでループを抜けられる
)

ポイント: state を使ったエージェント間の情報共有

各エージェントは output_key で指定したキーに出力を保存します。後続のエージェントはこれを参照できるため、エージェント間で情報をリレーできます。

# Planner が "analysis_plan" に計画を保存
planner_agent = LlmAgent(..., output_key="analysis_plan")

# DataCollector のプロンプトで参照
data_collector_instruction = """
以下の分析計画に従ってデータを取得してください:
{analysis_plan}
"""

また、ツール内でも tool_context.state を使って状態を読み書きできます。これを使って「発見リスト」を蓄積していきます:

def record_finding(finding: str, tool_context: ToolContext) -> str:
    """重要な発見を記録する"""
    current = tool_context.state.get("findings", "")
    new_findings = f"{current}\n\n{finding}".strip()
    tool_context.state["findings"] = new_findings
    return "発見を記録しました"