BigQuery エージェント分析を使用する

BigQuery エージェント分析は、リクエスト、レスポンス、ツール呼び出し、エラーなどの未加工のエージェント インタラクションを BigQuery に直接ストリーミングすることで、マルチモーダル エージェント インタラクション データを大規模にキャプチャ、分析、可視化できるオープンソース ソリューションです。このソリューションを使用すると、AI による評価の実施、エージェント プロンプトの最適化、長期記憶の抽出を行い、今後のやり取りを強化できます。BigQuery エージェント分析は、Agent Development Kit（ADK）と LangGraph（プレビュー）でサポートされています。

アーキテクチャ

BigQuery エージェント分析は、BigQuery Storage Write API（gRPC）を使用して、エージェント アクティビティ データを BigQuery にストリーミングします。これにより、エージェントの実行をブロックすることなく、高スループット、低レイテンシのログ ストリーミングが実現します。

データフローは次のステージで構成されます。

  1. キャプチャ。ADK のロギング プラグインまたは LangGraph のコールバック ハンドラを使用して、エージェントのインタラクション イベントをインターセプトしてキャプチャします。
  2. ストリーム。インタラクション イベントは、Storage Write API（gRPC）を介して BigQuery に送信されます。標準化されたスキーマが存在しない場合、エージェントは自動的にスキーマを作成します。
  3. 消費。ログに記録されたエージェント データを分析して評価します。SQL を使用して元データをクエリしたり、カスタム ダッシュボードで指標を追跡したり、 BigQuery エージェント分析 SDK を使用して複雑なマルチターン エージェント実行トレースを再構築して評価したりできます。

オーケストレーション フレームワークから BigQuery にエージェント アクティビティ データが流れて分析される様子を示す図。

エージェント分析のメリット

  • 1 行のコードで包括的なロギングを有効にし、スキーマ管理を自動化します。
  • オブジェクト テーブルを使用して、テキスト、画像、動画、音声などのマルチモーダル データをロギングして分析する。
  • トークン消費量やレイテンシなどの運用指標を、堅牢な事前定義スキーマ内で追跡します。
  • BigQuery の生成 AI 関数とベクトル検索を使用して、最適化の機会を特定する。
  • きめ細かいアクセス制御、データ マスキング、暗号化により、エージェント ログを保護します。

ワークフローの各ステージに適したライブラリを選択する

BigQuery エージェント分析には、2 つのロギング統合と 1 つのアナリティクス SDK があります。ロギング統合とアナリティクス SDK は、ワークフローの異なるステージに対応します。

  • エージェント イベントをログに記録する: エージェント フレームワークのロギング統合（ADK プラグイン、LangGraph、LangChain コールバック ハンドラ）を使用します。必要なロギング統合は 1 つだけで、イベントをログに記録するためにアナリティクス SDK は必要ありません。他のフレームワークの場合、または会話分析データ エージェントの組み込み会話ロギングを有効にする場合は、エージェント ログデータをキャプチャする方法をご覧ください。
  • エージェント イベントを分析する: BigQuery エージェント分析 SDK を使用して、ログに記録されたイベントをクエリ、再構築、評価します。SQL クエリとダッシュボードを使用してイベントを分析することもできます。

次の図は、各ワークフロー ステージで使用するライブラリを示しています。

ADK プラグインまたは LangGraph と LangChain のコールバック ハンドラは、エージェント イベントを BigQuery イベント テーブルにストリーミングします。このテーブルは、BigQuery エージェント分析アナリティクス SDK、SQL クエリ、ダッシュボードを使用して分析します。

次の表に、各ロギング統合とアナリティクス SDK の説明を示します。

ワークフローのステージ 統合または SDK パッケージとエントリ ポイント 使用する場面
エージェント イベントをログに記録する ADK BigQuery エージェント分析プラグイン

google-adk[bigquery-analytics] Python パッケージ

BigQueryAgentAnalyticsPlugin クラス

Java などの他の言語については、 ADK プラグイン ガイドをご覧ください。

 ADK エージェントで使用して、LLM リクエスト、ツール呼び出し、ワークフロー イベントを BigQuery にストリーミングします。
エージェント イベントをログに記録する LangGraph と LangChain の BigQuery コールバック ハンドラ（プレビュー）

langchain-google-community[bigquery] Python パッケージ

BigQueryCallbackHandler クラス

 LangGraph または LangChain アプリケーションで使用して、エージェント実行イベントを BigQuery にストリーミングします。
エージェント イベントを分析する BigQuery エージェント分析 SDK

bigquery-agent-analytics Python パッケージ

Client クラスと GraderPipeline クラス

 評価パイプライン、ノートブック、監査サービスで使用して、ロギング統合が書き込むイベント テーブルを読み取り、マルチターン セッションを再構築して、自動評価を実行します。

エージェント ログデータをキャプチャする方法

エージェントのインタラクション テレメトリー（リクエスト、レスポンス、ツール呼び出し、エラーログ）を BigQuery にネイティブにキャプチャするには、いくつかの方法でイベントデータをログに記録します。

  • オーケストレーション フレームワーク プラグイン: エージェント オーケストレーション ツールキットに用意されている標準のロギング プラグインを使用します。たとえば、ADK の BigQueryAgentAnalyticsPlugin はエージェント ランナーにフックして、イベントを自動的にインターセプト、シリアル化、ストリーミングします。
  • フレームワーク コールバック ハンドラ: 一般的なエージェント環境に標準のコールバックを統合します。たとえば、LangGraph と LangChain の langchain-google-community パッケージの BigQueryCallbackHandler を使用して、トレースをインターセプトして BigQuery に転送できます。
  • API による直接取り込み: カスタム フレームワークまたは独自のフレームワークの場合は、Google Cloud クライアント ライブラリを使用して、Storage Write API（gRPC）で構造化イベントをイベント テーブルに直接ストリーミングします。
  • 会話分析データ エージェント: BigQuery Studio または Conversational Analytics API を介して BigQuery データ エージェントの設定を構成するときに、組み込みの会話ロギングを有効にして、会話イベントを宛先の BigQuery テーブルに自動的にストリーミングし、会話ログを分析します。

どの方法でも、すべてのロギング オプションで、低レイテンシ、高スループットの BigQuery Storage Write API（gRPC）が使用されます。この API は、行を commit する前にメモリ内で非同期にバッファリングしてシリアル化（PyArrow エンジンを使用）する堅牢なストリーミング エンドポイントを提供します。これにより、オブザーバビリティ パイプライン タスクがユーザー向けエージェントの実行ターンをブロックしないようにします。

エージェント ログデータを分析する方法

エージェントのパフォーマンスを把握して最適化するには、次の方法でインタラクション ログを分析して評価します。

  • 直接 SQL クエリ: BigQuery でカスタムクエリを実行して、トークン消費量や実行レイテンシなどの指標を計算します。AI.GENERATE を使用してエラーの根本原因を自動的に分析したり、ビジネス テーブルと結合してビジネスへの影響を測定することもできます。
  • インタラクティブなダッシュボード: データポータルなどの可視化ツールを、事前構築済みまたはカスタムの BigQuery ビューに接続して、エージェントの健全性、エラー率、使用状況の傾向を時系列で追跡します。Looker のお客様は、Looker Marketplace の事前構築済みの BigQuery Agent Analytics ダッシュボード テンプレート（Looker Block）を使用することもできます。
  • Jupyter ノートブック: インタラクティブな環境で Python ライブラリ、pandas、BigFrames を使用して、ログデータを探索してテストします。
  • Python SDK: アプリケーション コードまたは自動評価パイプラインで直接、エージェントの実行トレースをプログラムでクエリ、再構築、監査します。

エージェント ログデータの操作例

以下に、BigQuery でエージェント ログデータを操作する一般的なユースケースと例を示します。

オブザーバビリティと運用指標

  • データをクエリして、エージェント フロー別に費用を分類し、特定のエージェント（リファインメント エージェントなど）が最終的な回答への貢献度と比較して不釣り合いな量のトークンを消費しているかどうかを判断します。
  • BigQuery 会話分析エージェントを使用し、AI.GENERATE 関数でクエリを実行して、AI を活用した根本原因分析を行います。たとえば、「この会話ログを分析して、失敗の根本原因を説明してください」などです。

エージェントの評価と品質分析

  • AI.SCORE 関数を使用して、会話をランク付けし、エージェントのランクを時系列で測定します。
  • ベクトル検索で SQL クエリを使用して、エージェントがユーザーをサポートできなかった会話クラスタを特定し、ユーザーの元のインテントと比較します。これにより、エージェントのツールやナレッジベースのギャップを特定できます。

ビジネス分析情報とコンテキスト化

エージェント データをコンテキスト化するには、agent_events テーブルを他のビジネス テーブルと結合します。たとえば、AI エージェントを操作した顧客と検索バーを使用した顧客の平均注文額（AOV）を表示します。

その他の例については、クエリレシピをご覧ください。

Jupyter ノートブックを使用してエージェント ログを操作する

このサンプルの Colab Jupyter ノートブックを使用して、エージェント ログをインタラクティブにクエリ、可視化、評価します。

BigQuery エージェント分析 SDK を使用する

BigQuery エージェント分析 SDK は、長期的なエージェントのオブザーバビリティのための消費レイヤと評価レイヤを提供するオープンソースの Python ライブラリです。ダッシュボードとノートブックはアドホックな探索に最適です。SDK を使用すると、エージェントの動作を大規模に体系的に分析して監査できます。

SDK でできること

BigQuery エージェント アナリティクス SDK を使用すると、次のタスクを実行できます。ログ分析の詳細な例については、SDK の GitHub リポジトリをご覧ください。

  • トレースの再構築: 多態性イベントログをイベントの因果関係チェーンに再構築し、ネストされたツールや LLM 呼び出しなど、複数のターンにわたるセッションをデバッグします。
  • 決定的評価とセマンティック評価: レイテンシ、ターン数、エラー率などのルールベースの基準と、正確性、感情、ハルシネーションなどのセマンティック基準に照らして、エージェントの品質を評価します。
  • 軌跡の一致: 実際のエージェントの実行パスを、期待されるゴールデン軌跡と比較して、ステップの効率とツールが正しい順序で使用されたかどうかを検証します。
  • 行動モニタリングとドリフト検出: 非決定的なエージェント出力に対して統計分析を実行し、ユーザー リクエストの分布をモニタリングして、本番環境の回帰やセマンティック ドリフトを検出します。
  • 長期的なエージェント メモリ: エージェントにセッション間のコンテキスト、ユーザー プロファイルのセマンティック検索、トークン予算を考慮したエピソード メモリを提供します。これらは BigQuery にネイティブに保存されます。

エージェントのアクティビティをログに記録して分析する

SDK は、ロギング統合が書き込むイベントを読み取って評価します。一般的なワークフローは、次の手順で構成されます。

  1. インタラクションをログに記録する: エージェント オーケストレーション フレームワークで、ADK の BigQueryAgentAnalyticsPlugin や LangGraph の BigQueryCallbackHandler などのロギング統合をアタッチします。ユーザーがエージェントを操作すると、統合は高スループットの Storage Write API（gRPC）を使用して、イベントを BigQuery に非同期でストリーミングします。このステップでは SDK は必要ありません。

  2. クライアントを初期化する: Python SDK からログ データセットに接続します。

    from google.cloud import bigquery
from bigquery_agent_analytics import Client

client = Client(
    project_id="YOUR_PROJECT_ID",
    dataset_id="YOUR_DATASET_ID",
    table_id="agent_events",
)

    このスニペットでは、次のコンポーネントを使用しています。

    • client: クエリをプログラムでルーティングし、アクティブなデータベース接続を管理する親 Client インスタンス。
    • project_id: ターゲット データセットを格納する Google Cloud プロジェクト ID。
    • dataset_id: ログを保存する BigQuery データセットの名前。
    • table_id: ログに記録されたイベントを保存するテーブル（デフォルトでは agent_events）。この値は、ロギング統合が書き込むテーブルと一致する必要があります。たとえば、langchain-google-community バージョン 4.0.0 以前の BigQueryCallbackHandler は、デフォルトで agent_events_v2 に書き込みます。

  3. セッション トレースを再構築する: 特定の会話セッションを取得して、イベントの正確なシーケンスを可視化して確認します。

    trace = client.get_session_trace(
    session_id="YOUR_SESSION_ID"
)
trace.render()

    このスニペットでは、次のコンポーネントを使用しています。

    • trace: すべてのユーザーとエージェントのアクションの再構築された因果関係のある階層 DAG スパンツリーを格納する、ハイドレートされた Trace オブジェクト。
    • YOUR_SESSION_ID: 検査するセッションの一意の ID。

  4. 自動評価を実行する: セッションの軌跡をプログラムでスコア付けするか、ゴールデン テストスイートに対する回帰をチェックします。

    from bigquery_agent_analytics import (
    LLMAsJudge,
    TraceFilter,
)

# Run semantic evaluations across target sessions
evaluator = LLMAsJudge.correctness()
filters = TraceFilter(session_ids=["session_1", "session_2"])

report = client.evaluate(evaluator, filters=filters)
print(f"Evaluation Pass Rate: {report.pass_rate:.2%}")

    このスニペットでは、次のコンポーネントを使用しています。

    • evaluator: セッションのスコア付けに使用されるセマンティック（LLMAsJudge）評価ツール（ルールベースの SystemEvaluator チェックを実行したり、評価ツールを重み付けされた GraderPipeline に組み合わせたりすることもできます）。
    • filters: BigQuery でバッチ評価を実行する session_ids リスト（または時間枠とエージェント フィルタ）を指定する TraceFilter インスタンス。
    • report: 未加工のセッション グレード、指標の概要、自動評価フィードバックを含む、結果の EvaluationReport オブジェクト。

BigQuery エージェント分析をワークフローに統合する

BigQuery エージェント分析をワークフローに統合する手順は次のとおりです。

  1. エージェント フレームワークのロギング統合を設定します。

  2. BigQuery エージェント分析アナリティクス SDK、Colab ノートブックの例、SQL クエリレシピを使用して、ログに記録されたイベントを分析します。

次のステップ