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

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

アーキテクチャ

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

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

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

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

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

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

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

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

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

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

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

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

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

エージェント ログデータの使用例

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

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

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

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

  • 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 など）またはコールバック ハンドラをアタッチします。ユーザーがエージェントとやり取りすると、ログは高スループットの 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: ターゲット データセットをホストする Cloud de Confiance by S3NS プロジェクト 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 エージェント分析をワークフローに統合するには、フレームワークのドキュメントをご覧ください。

次のステップ