はじめに
AI エージェントを開発・運用する中で、「ユーザーはどの機能をよく使っているのだろう？」「処理に時間がかかっている箇所はどこだろう？」「エラーの発生率は？」といった疑問を持つことは少なくありません。エージェントの品質を継続的に向上させるためには、これらの利用状況をデータに基づいて分析することが不可欠です。しかし、そのための分析基盤を自前で構築するのは手間がかかります。
そんな課題を解決するのが、Google が提供する Agent Development Kit (ADK) 向けの公式プラグイン 「BigQuery Agent Analytics」 です。2025 年 11 月に Preview で公開しました。
この記事では「BigQuery Agent Analytics」の概要から、実際に ADK アプリケーションに導入して利用状況を分析するまでの手順を解説します。
「BigQuery Agent Analytics」とは？
「BigQuery Agent Analytics」は、ADK で開発されたエージェントのイベントデータを、簡単に BigQuery へストリーミングできる公式プラグインです。エージェントの分析基盤をクイックに構築できます。
主な特徴は以下の通りです。
- 簡単なセットアップ: わずか数行のコードを追記するだけで、既存の ADK アプリケーションに導入できます。
- 高性能なストリーミング: 内部で BigQuery Storage Write API を利用しており、大量のデータを低コストかつリアルタイムに BigQuery へ書き込むことができます。
- 最適化されたスキーマ: エージェントの行動分析（トレース、ツール利用、実行時間など）に最適化されたテーブルスキーマが予め定義されています。
- 柔軟なカスタマイズ: ログに記録するイベントを選択したり、個人情報（PII）などを送信前に秘匿化したりといった、高度なカスタマイズも可能です。
アーキテクチャ
このプラグインは、ADK のイベント処理の仕組みにフックする形で動作します。エージェントのライフサイクルで発生したイベント (対話の開始、ツールの呼び出し、終了など) を検知し、裏側で BigQuery Storage Write API を呼び出して、指定された BigQuery テーブルにデータを直接送信します。
開発者はこの一連の流れで、どのようなコードが実行されているか意識する必要はなく導入できます。
実装チュートリアル
それでは、実際に ADK アプリケーションにプラグインを導入してみましょう。
Step 1: 事前準備
まず、Google Cloud プロジェクトでいくつかの準備を行います。
-
API の有効化: BigQuery API を有効にしておきます。
gcloud services enable bigquery.googleapis.com -
IAM ロールの設定: プラグインが利用するサービスアカウント（Cloud Run であれば、その実行サービスアカウント）に、以下の IAM ロールを付与します。
-
BigQuery ユーザー(roles/bigquery.user) -
BigQuery データ編集者(roles/bigquery.dataEditor)
-
-
BigQuery データセットの作成: 分析データを格納するための BigQuery データセットを作成します。
bq mk --dataset --location=asia-northeast1 agent_events -
BigQuery テーブルの作成: 下記のスキーマで BigQuery テーブルを作成します。
CREATE TABLE `your-gcp-project-id.adk_agent_logs.agent_events` ( timestamp TIMESTAMP NOT NULL OPTIONS(description="The UTC time at which the event was logged."), event_type STRING OPTIONS(description="Indicates the type of event being logged (e.g., 'LLM_REQUEST', 'TOOL_COMPLETED')."), agent STRING OPTIONS(description="The name of the ADK agent or author associated with the event."), session_id STRING OPTIONS(description="A unique identifier to group events within a single conversation or user session."), invocation_id STRING OPTIONS(description="A unique identifier for each individual agent execution or turn within a session."), user_id STRING OPTIONS(description="The identifier of the user associated with the current session."), content STRING OPTIONS(description="The event-specific data (payload). Format varies by event_type."), error_message STRING OPTIONS(description="Populated if an error occurs during the processing of the event."), is_truncated BOOLEAN OPTIONS(description="Boolean flag indicates if the content field was truncated due to size limits.") ) PARTITION BY DATE(timestamp) CLUSTER BY event_type, agent, user_id;
event_type には LLM_REQUEST, LLM_RESPONSE, TOOL_CALL, USER_MESSAGE_RECEIVED といったイベントが格納されます。
Step 2: サンプルエージェントの作成
公式ドキュメントの クイックスタート のサンプルの天気を調べるエージェントを使います。ここでは Cloud Shell を使用します。
# 環境準備
python -m venv .venv
pip install --upgrade google-adk
# サンプルエージェントの作成
mkdir my-agents
adk create multi_tool_agent
echo "from . import agent" > multi_tool_agent/__init__.py
touch multi_tool_agent/agent.py
adk create multi_tool_agent のウィザードでは、下記を選択します。
- モデルは
gemini-2.5-flashを選択 - バックエンドは
Vertex AIを選択 - 利用したい Google Cloud Project を選択
- 利用したいリージョン (
asia-northeast1など) を選択
Step 3: プラグインの導入
プラグインを使用するコードを追加します。まずは導入するためのコードの全体を俯瞰してみましょう。アプリケーションのエントリーポイント (agent.py や main.py など) で行うのが一般的です。
from adk.api import app
# プラグインのインポート
from adk.plugins import bigquery_analytics
# プラグインの初期化
bq_logging_plugin = BigQueryAgentAnalyticsPlugin(
project_id=PROJECT_ID,
dataset_id=DATASET_ID,
table_id="agent_events"
)
# モデルの初期化
llm = Gemini(
model="gemini-2.5-flash",
)
# ルート エージェントの初期化
root_agent = Agent(
model=llm,
name='my_adk_agent',
instruction="You are a helpful assistant"
)
# アプリの作成
app = App(
name="my_adk_agent",
root_agent=root_agent,
# ここでプラグインを登録
plugins=[bq_logging_plugin],
)
Step 2 で作成したサンプルエージェントに組み込む場合は、下記のようになります。
import os
import datetime
from zoneinfo import ZoneInfo
from google.adk.apps import App
from google.adk.agents import Agent
from google.adk.plugins.bigquery_agent_analytics_plugin import BigQueryAgentAnalyticsPlugin
PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT", "default-project")
DATASET_ID = os.environ.get("BIG_QUERY_DATASET_ID", "default-detaset")
# BigQueryAgentAnalyticsPlugin の使用
bq_logging_plugin = BigQueryAgentAnalyticsPlugin(
project_id=PROJECT_ID,
dataset_id=DATASET_ID,
table_id="agent_events",
)
def get_weather(city: str) -> dict:
"""Retrieves the current weather report for a specified city.
Args:
city (str): The name of the city for which to retrieve the weather report.
Returns:
dict: status and result or error msg.
"""
if city.lower() == "new york":
return {
"status": "success",
"report": (
"The weather in New York is sunny with a temperature of 25 degrees"
" Celsius (77 degrees Fahrenheit)."
),
}
else:
return {
"status": "error",
"error_message": f"Weather information for '{city}' is not available.",
}
def get_current_time(city: str) -> dict:
"""Returns the current time in a specified city.
Args:
city (str): The name of the city for which to retrieve the current time.
Returns:
dict: status and result or error msg.
"""
if city.lower() == "new york":
tz_identifier = "America/New_York"
else:
return {
"status": "error",
"error_message": (
f"Sorry, I don't have timezone information for {city}."
),
}
tz = ZoneInfo(tz_identifier)
now = datetime.datetime.now(tz)
report = (
f'The current time in {city} is {now.strftime("%Y-%m-%d %H:%M:%S %Z%z")}'
)
return {"status": "success", "report": report}
root_agent = Agent(
name="weather_time_agent",
model="gemini-2.5-flash",
description=(
"Agent to answer questions about the time and weather in a city."
),
instruction=(
"You are a helpful agent who can answer user questions about the time and weather in a city."
),
tools=[get_weather, get_current_time],
)
app = App(
name="multi_tool_agent",
root_agent=root_agent,
# ここでプラグインを登録
plugins=[bq_logging_plugin],
)
project_id, dataset_id, table_id は、ご自身の環境に合わせて変更してください。たったこれだけで、プラグインの導入は完了です。
Step 4: Cloud Run へのデプロイ
ローカル (adk web コマンド) で確認するだけでも良いですが、ここでは Cloud Run にデプロイします。ここでは adk deploy コマンドを使用します。
adk deploy cloud_run \
--project=$GOOGLE_CLOUD_PROJECT \
--region=asia-northeast1 \
--service_name=multi-tool-agent \
--app_name=multi-tool-agent \
--with_ui \
multi_tool_agent
デプロイ後、Cloud Run のエンドポイントにアクセスすると ADK Web コンソールで動作確認が行えます。ログを蓄積するため、適当な質問をしておきます。
Step 5: BigQuery での分析
ここまでで問題なく設定ができていれば、Step 1 で作成しておいた BigQuery テーブルにイベントが蓄積されているはずです。
BigQuery コンソール にアクセスし、簡単なクエリを実行してみましょう。
SELECT timestamp, event_type, content
FROM `<Project Name>.<Dataset ID>.<Table ID>`
ORDER BY timestamp DESC
LIMIT 20;
次のように結果が表示されれば成功です。
Step 6: 高度な設定 (任意)
プラグインは、ログに記録する内容をカスタマイズする機能も提供しています。
イベントのフィルタリング
例えば、ツールの実行に関するログ（on_tool_start, on_tool_end）のみを記録したい場合は、以下のようにフィルタ関数を渡すことができます。
def should_log(event_type, payload):
return event_type.startswith("on_tool")
app.use(bigquery_analytics,
# ... (他の設定) ...
filter=should_log)
内容の秘匿化
ログを送信する前に、個人情報などをマスキングすることも可能です。
def redact_payload(event_type, payload):
if "user_message" in payload:
payload["user_message"] = "[REDACTED]"
return payload
app.use(bigquery_analytics,
# ... (他の設定) ...
preprocessor=redact_payload)
まとめ
この記事では、ADK の公式プラグイン「BigQuery Agent Analytics」を使って、AI エージェントの行動分析基盤を簡単に構築する方法を紹介しました。このプラグインを活用することで、開発者は分析基盤の構築・運用に煩わされることなく、データに基づいたエージェントの品質向上に集中できます。
ここからさらに、Looker Studio を使って分析結果をダッシュボードとして可視化したり、BigQuery ML を使って感情分析したりと、様々な発展が考えられます。
ぜひご自身の ADK アプリケーションに導入して、データ駆動なエージェント開発を始めてみてください。
Codelab も公開されていますので、こちらもぜひお試しください。
Discussion