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

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

アーキテクチャ

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

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

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

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

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

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

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

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

  • オーケストレーション フレームワーク プラグイン: エージェント オーケストレーション ツールキットで提供される標準のロギング プラグインを使用します。たとえば、Agent Development Kit（ADK）の BigQueryAgentAnalyticsPlugin はエージェント ランナーにフックして、イベントを自動的にインターセプト、シリアル化、ストリーミングします。
  • フレームワーク コールバック ハンドラ: 一般的な エージェント環境で標準のコールバックを統合します。たとえば、LangGraph と LangChain の組み込み BigQuery ハンドラを使用して、トレースをインターセプトして転送できます。
  • 直接 API 取り込み: カスタム フレームワークまたは独自仕様のフレームワークの場合は、 Google Cloud クライアント ライブラリを使用して、Storage Write API を使用して構造化イベントを イベント テーブルに直接ストリーミングします。

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

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

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

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

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

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

  • データをクエリして エージェント フローごとに費用を分類し、絞り込みエージェントなどの特定のエージェントが、最終的なレスポンスへの貢献度と比較して 不均衡な量のトークンを消費しているかどうかを判断します。
  • BigQuery 会話分析エージェント を使用して、the AI.GENERATE 関数でクエリを実行し、AI を活用した根本原因分析を行います。 例: 「この会話ログを分析して、失敗の根本原因を説明してください。」

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

ビジネスに関する分析情報とコンテキスト化

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

その他の例については、高度な分析クエリをご覧ください。

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

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

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

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

SDK でできること

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

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

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

通常、SDK をエージェント ワークフローに統合するには、次の手順を行います。

  1. インタラクションをログに記録する: エージェント オーケストレーション フレームワークにロガー プラグイン（ADK の BigQueryAgentAnalyticsPluginなど）またはコールバック ハンドラをアタッチします。ユーザーがエージェントとやり取りすると、高スループットの Storage Write API を使用してログが BigQuery に非同期でストリーミングされます。

  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）。

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

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

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

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

  4. 自動評価を実行する: セッション トラジェクトリをプログラムでスコアリングするか ゴールデン テストスイートに対して回帰を確認します。

    from bigquery_agent_analytics.evaluators import CodeEvaluator, LLMAsJudge
from bigquery_agent_analytics.grader_pipeline import GraderPipeline

# Create a grader pipeline with deterministic and semantic metrics
evaluator = GraderPipeline(
    graders=[
        CodeEvaluator.latency(threshold_ms=5000),
        LLMAsJudge.correctness(),
    ]
)
report = client.evaluate(evaluator, session_ids=["session_1", "session_2"])
print(f"Evaluation Pass Rate: {report.pass_rate:.2%}")

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

    • evaluator: 構造化された GraderPipeline コンパイル済みロジック。異種ルールベースの評価指標とセマンティック評価指標で構成されます。
    • session_ids: BigQuery でバッチ評価を実行するセッション ID 文字列のリスト。
    • report: 生のセッション グレード、指標の概要、トライアル統計、自動評価ツールからのフィードバックを含む、結果の EvaluationReport オブジェクト。

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

BigQuery エージェント分析をワークフローに統合するには、フレームワークのドキュメントをご覧ください。

次のステップ