このページは Cloud Translation API によって翻訳されました。

BigQuery エージェント分析プラグインを使用して ADK エージェントの実行を分析する

1. はじめに

この Codelab では、Agent Development Kit（ADK）を使用してマルチエージェントシステムを構築し、BigQuery Agent Analytics Plugin を使用してエージェントのオブザーバビリティを有効にします。エージェントに一連の質問をしてから、BigQuery を使用して会話のトレースとエージェントツールの使用状況を分析します。

演習内容

ADK を使用してマルチエージェント小売アシスタントを構築する
BigQuery エージェント分析プラグインを初期化して、このエージェントの実行に関するトレースデータをキャプチャし、BigQuery に保存します。
BigQuery でエージェントのログデータを分析する

必要なもの

ウェブブラウザ（Chrome など）
課金が有効になっている Google Cloud プロジェクト
Gmail アカウント。次のセクションでは、この Codelab で使用できる 5 ドルの無料クレジットを利用して、新しいプロジェクトを設定する方法について説明します。

この Codelab は、初心者を含むあらゆるレベルのデベロッパーを対象としています。ADK 開発には、Google Cloud Shell のコマンドラインインターフェースと Python コードを使用します。Python の専門家である必要はありませんが、コードの読み取り方法に関する基本的な知識があると、コンセプトを理解するのに役立ちます。

2. 始める前に

Google Cloud プロジェクトの作成

Google Cloud コンソールのプロジェクト選択ページで、Google Cloud プロジェクトを選択または作成します。

Cloud プロジェクトに対して課金が有効になっていることを確認します。詳しくは、プロジェクトで課金が有効になっているかどうかを確認する方法をご覧ください。

Cloud Shell の起動

Cloud Shell は、必要なツールがプリロードされた Google Cloud で動作するコマンドライン環境です。

Google Cloud コンソールの上部にある [Cloud Shell をアクティブにする] をクリックします。

Cloud Shell に接続したら、次のコマンドを実行して Cloud Shell で認証を確認します。

gcloud auth list

次のコマンドを実行して、プロジェクトが gcloud で使用するように構成されていることを確認します。

gcloud config get project

プロジェクトが想定どおりに構成されていない場合は、次のコマンドを使用してプロジェクトを設定します。

export PROJECT_ID=<YOUR_PROJECT_ID>
gcloud config set project $PROJECT_ID

API を有効にする

次のコマンドを実行して、必要なすべての API とサービスを有効にします。

gcloud services enable bigquery.googleapis.com \
cloudresourcemanager.googleapis.com \
aiplatform.googleapis.com

コマンドが正常に実行されると、次のようなメッセージが表示されます。

Operation "operations/..." finished successfully.

3. インストールとセットアップ

Cloud Shell に戻り、ホームディレクトリにいることを確認します。

Cloud Shell で次のコマンドを実行して、BigQuery に adk_logs という新しいデータセットを作成します。

bq mk --dataset --location=US adk_logs

次に、仮想 Python 環境を作成し、必要なパッケージをインストールします。

Cloud Shell で新しいターミナルタブを開き、次のコマンドを実行して adk-agent-observability という名前のフォルダを作成して移動します。

mkdir adk-agent-observability
cd adk-agent-observability

仮想 Python 環境を作成します。

python -m venv .venv

仮想環境をアクティブにします。

source .venv/bin/activate

ADK をインストールする

pip install --upgrade google-adk

4. ADK アプリケーションを作成する

それでは、小売アシスタントエージェントを作成しましょう。このエージェントは、...

adk create ユーティリティコマンドを実行して、必要なフォルダとファイルを含む新しいエージェントアプリケーションをスキャフォールディングします。

adk create retail_assistant_app

画面上の指示に沿って操作します。

モデルに gemini-2.5-flash を選択します。
バックエンドに Vertex AI を選択します。
デフォルトの Google Cloud プロジェクト ID とリージョンを確認します。

以下にやり取りの例を示します。

Cloud Shell で [エディタを開く] ボタンをクリックして Cloud Shell エディタを開き、新しく作成したフォルダとファイルを表示します。

生成されたファイルを確認します。

retail_assistant_app/
├── .venv/
└── retail_assistant_app/
    ├── __init__.py
    ├── agent.py
    └── .env

init.py: フォルダを Python モジュールとしてマークします。
agent.py: 初期エージェント定義が含まれています。
.env: このファイルを表示するには、[View] > [Toggle Hidden Files] をクリックする必要がある場合があります。

.env ファイルにはプロジェクトの環境変数が含まれています。プロンプトで正しく設定されなかった変数を更新します。

GOOGLE_GENAI_USE_VERTEXAI=1
GOOGLE_CLOUD_PROJECT=<YOUR_GOOGLE_PROJECT_ID>
GOOGLE_CLOUD_LOCATION=<YOUR_GOOGLE_CLOUD_REGION>

5. エージェントを定義する

次に、階層型マルチエージェントシステムを定義します。

リアルタイムトレンドエージェント: Google 検索を使用して、現在のファッショントレンドを検索します。
在庫データエージェント: BigQuery ツールセットを使用して、一般公開の thelook_ecommerce データセットで利用可能な商品をクエリします。
小売アシスタント（ルート）エージェント: トレンドエージェントにアドバイスを求め、在庫エージェントに一致する商品を求めることで、ワークフローをオーケストレートします。

retail_assistant_app/agent.py の内容全体を次のコードに置き換えます。

import os
import uuid
import asyncio
import google.auth
import dotenv
from google.genai import types
from google.adk.agents import Agent
from google.adk.apps import App
from google.adk.runners import InMemoryRunner
from google.adk.tools import AgentTool, google_search
from google.adk.tools.bigquery import BigQueryCredentialsConfig, BigQueryToolset
from google.adk.plugins.bigquery_agent_analytics_plugin import BigQueryAgentAnalyticsPlugin

dotenv.load_dotenv()

# --- Configuration ---

PROJECT_ID = os.getenv('GOOGLE_CLOUD_PROJECT', 'project_not_set')
DATASET_ID = "adk_logs"
TABLE_ID = "retail_assistant_agent_logs"
APP_NAME = "retail_assistant_agent"
USER_ID = "test_user"

# --- Toolsets ---

credentials, _ = google.auth.default()
credentials_config = BigQueryCredentialsConfig(credentials=credentials)
bigquery_toolset = BigQueryToolset(
 credentials_config=credentials_config
)

# --- Agents ---

# 1. Trend Spotter
real_time_agent = Agent(
   name="real_time_agent",
   model="gemini-2.5-flash",
    description="Researches external factors like weather, local events, and current fashion trends.",
   instruction="""
       You are a real-time research agent.
       Use Google Search to find real-time information relevant to the user's request,
       such as the current weather in their location or trending styles.
   """,
   tools=[google_search]
)

# 2. Inventory Manager
inventory_data_agent = Agent(
   name="inventory_data_agent",
   model="gemini-2.5-flash",
   description="Oversees product inventory in the BigQuery `thelook_ecommerce` dataset to find available items and prices.",
   instruction=f"""
   You manage the inventory. You have access to the `bigquery-public-data.thelook_ecommerce` dataset via the BigQuery toolset.
   Run all BigQuery queries the project id of: '{PROJECT_ID}'
  
   Your workflow:
   1. Look at the products table.
   2. Find items that match the requirements, factor in the results from the trend_setter agent if there are any.
   3. Return with a user friendly response, including the list of specific products and prices.
   """,
   tools=[bigquery_toolset]
)

# 3. Root Orchestrator
root_agent = Agent(
   name="retail_assistant",
   model="gemini-2.5-flash",   
   description="The primary orchestrator, responsible for handling user input, delegating to sub-agents, and synthesizing the final product recommendation.",
   instruction="""
   You are a Retail Assistant.   
   You can ask the 'real_time_agent' agent for any realtime information needed, or style advice, include any information provided by the user.
   You should ask the 'inventory_data_agent' agent to find a maximum of 3 available items matching that style.
   Combine the results into a recommendation.
   """,
   tools=[AgentTool(agent=real_time_agent)],
   sub_agents=[inventory_data_agent]
)

6. BigQuery エージェント分析プラグインでログを生成する

次に、実行データをキャプチャするように BigQuery エージェント分析プラグインを構成します。

これを行うには、App クラスのインスタンスを作成します。このクラスは、エージェントのランタイムコンテナとして機能します。会話ループを管理し、ユーザーの状態を処理し、接続されたプラグイン（エージェント分析ロガーなど）をオーケストレートします。

以下のコード:

ロギングプラグインを初期化する: 必要な接続の詳細を含む BigQueryAgentAnalyticsPlugin を作成します。
プラグインを統合する: 初期化された BigQuery プラグインを App コンストラクタに渡し、エージェント実行イベントが自動的にキャプチャされてログに記録されるようにします。
Runs and Logs Agent Execution: runner.run_async を介して会話フローを実行します。プラグインは、リソースを閉じる前に、イベントのシーケンス全体を同時に収集して BigQuery に送信します。

次のコードをコピーして、agent.py ファイルのエージェント定義の下に貼り付けます。

async def main(prompt: str):
    """Runs a conversation with the BigQuery agent using the ADK Runner."""
    bq_logger_plugin = BigQueryAgentAnalyticsPlugin(
        project_id=PROJECT_ID, dataset_id=DATASET_ID, table_id=TABLE_ID
    )
    app = App(name=APP_NAME, root_agent=root_agent, plugins=[bq_logger_plugin])
    runner = InMemoryRunner(app=app)

    try:
        session_id = f"{USER_ID}_{uuid.uuid4().hex[:8]}"

        my_session = await runner.session_service.create_session(
            app_name=APP_NAME, user_id=USER_ID, session_id=session_id
        )

        async for event in runner.run_async(
            user_id=USER_ID,
            new_message=types.Content(
                role="user", parts=[types.Part.from_text(text=prompt)]
            ),
            session_id=my_session.id,
        ):
            if event.content.parts and event.content.parts[0].text:
                print(f"** {event.author}: {event.content.parts[0].text}")

    except Exception as e:
        print(f"Error in main: {e}")
    finally:
        print("Closing BQ Plugin...")
        await bq_logger_plugin.close()
        print("BQ Plugin closed.")

if __name__ == "__main__":
   prompts = [
       "what outfits do you have available that are suitable for the weather in london this week?",      
       "You are such a cool agent! I need a gift idea for my friend who likes yoga.",       
       "I'd like to complain - the products sold here are not very good quality!"
   ]
   for prompt, prompt in enumerate(prompts):
       asyncio.run(main(prompt))

計測が完了したので、エージェントの動作を確認しましょう。スクリプトを実行して会話ワークフローをトリガーします。

python retail_assistant_app/agent.py

小売アシスタントがワークフローを調整していることがわかります。

リアルタイムトレンドエージェント（real_time_agent）に、ロンドンの天気を特定し、適切なファッショントレンドを検索するよう依頼します。
次に、インベントリデータエージェント（inventory_data_agent）に委任して、これらの傾向に一致する特定の商品の thelook_ecommerce BigQuery データセットをクエリします。
最後に、ルートオーケストレーターが結果を統合して最終的な推奨事項を作成します。

その間、プラグインはエージェントの実行トレースを BigQuery にストリーミングしています。

7. エージェントログを分析する

ツールの使用

これで、エージェントがバックグラウンドで何を行っていたかを確認できるようになりました。データが BigQuery にストリーミングされ、分析の準備が整いました。

Google Cloud コンソールで [BigQuery] を検索します。
[エクスプローラ] ペインでプロジェクトを探します。
adk_logs データセットを開きます。
retail_assitant_agent_logs テーブルを開き、[クエリ] をクリックします。

エージェントが行ったツール呼び出しを確認し、ツールエラーをキャプチャするには、BigQuery エディタで次のクエリを実行します。

SELECT
   -- Extract text between "Tool Name: " and the next comma (or end of line)
   REGEXP_EXTRACT(content, r'Tool Name: ([^,]+)') AS tool_name,
   -- Count every time a tool finished (successfully or with an error)
   COUNT(*) AS total_finished_runs,
   -- Count it as a failure if it's an explicit system error OR contains "error" in the text
   COUNTIF(event_type = 'TOOL_ERROR' OR REGEXP_CONTAINS(content, r'(?i)\berror\b')) AS failure_count
FROM
   `.adk_logs.retail_assistant_agent_logs`
WHERE
   event_type IN ('TOOL_COMPLETED', 'TOOL_ERROR')
GROUP BY
   1

[ビジュアリゼーション] をクリックして、グラフとして表示します。

トークンの使用量

エージェントの費用を推測するには、個別のエージェントごとに使用されたプロンプトトークンと候補トークンを集計します。

SELECT
     t.agent,
     SUM(CAST(REGEXP_EXTRACT(t.content, r'prompt:\s*(\d+)') AS INT64)) AS prompt_tokens,
     SUM(CAST(REGEXP_EXTRACT(t.content, r'candidates:\s*(\d+)') AS INT64)) AS candidate_tokens
   FROM
     `adk_logs.retail_assistant_agent_logs` AS t
   WHERE
     t.event_type = 'LLM_RESPONSE'
     AND t.content LIKE '%Token Usage: %'
   GROUP BY 1

[ビジュアリゼーション] をクリックして、グラフとして表示します。

8. [ボーナス] ユーザーの感情を分析する

次に、エージェントに提供されたユーザーの入力の感情を分析します。

クラウドリソース接続を作成して、BigQuery が Vertex AI サービスとやり取りできるようにします。

bq mk --connection --location=us \
    --connection_type=CLOUD_RESOURCE test_connection

次のようなレスポンスが表示されます。

接続 517325854360.us.test_connection が正常に作成されました

クラウドリソース接続を作成します。

export SERVICE_ACCOUNT_EMAIL=$(bq show --format=prettyjson --connection us.test_connection | grep "serviceAccountId" | cut -d '"' -f 4)

次のコマンドを実行して、サービスアカウントが正常に作成されたことを確認します。

echo $SERVICE_ACCOUNT_EMAIL

サービスアカウントが表示されます。

Vertex AI とのやり取りに必要なプロジェクトレベルの権限をリソース接続サービスアカウントに付与します。

gcloud projects add-iam-policy-binding $(gcloud config get-value project) \
    --member="serviceAccount:$SERVICE_ACCOUNT_EMAIL" \
    --role='roles/bigquery.connectionUser' \
gcloud projects add-iam-policy-binding $(gcloud config get-value project) \
    --member="serviceAccount:$SERVICE_ACCOUNT_EMAIL" \
    --role='roles/aiplatform.user'

数分待ってから、BigQuery の AI.SCORE 関数を実行してユーザーの感情を分析します。

SELECT
 timestamp,
 user_id,
 content,
 AI.SCORE((
   'What is the sentiment of the user in this text:', content,
   'Use a scale from 1 to 5.'),
 connection_id => 'us.test_connection') AS user_sentiment
FROM
 `adk_logs.retail_assistant_agent_logs`
WHERE
  event_type = 'USER_MESSAGE_RECEIVED'
ORDER BY
 user_sentiment DESC;

AI.SCORE 関数は、ユーザー入力ごとに 1 ～ 5 の感情値を割り当てます。次のような結果が表示されます。

9. クリーンアップ

Google Cloud アカウントに継続的に課金されないようにするには、このワークショップで作成したリソースを削除します。

スクリプトで作成したロギングデータセットを削除します。

bq rm -r -f -d $PROJECT_ID:adk_logs

bigquery-adk-codelab ディレクトリとその内容を削除するには:

cd .. 
rm -rf adk-agent-observability

10. 完了

おめでとうございます！Agent Development Kit（ADK）を使用してマルチエージェントシステムを構築し、BigQuery エージェント分析プラグインを統合して、エージェントの動作を追跡して監査しました。