会話ログを分析する

このドキュメントでは、GoogleSQL レシピ、インタラクティブな Colab ノートブック、BigQuery エージェント分析 Python SDK を使用して、BigQuery で会話分析の会話ログをクエリ、評価、トラブルシューティングする方法について説明します。宛先テーブルの列と JSON 構造の詳細については、会話ログのスキーマをご覧ください。

Colab ノートブックと Python SDK を使用する

このドキュメントの SQL レシピは、Customer Analytics、FinOps & Evaluation Starter Notebook（Colaboratory で開く | GitHub で表示）を使用して、bigquery-agent-analytics Python SDK（v0.5.4 以降）とともに実行できます。

会話ログテーブルで Python SDK を使用するには、まず v_bqca_sdk_events 互換性ビューを作成します。AGENT_RESPONSE 行では、会話分析は JSON オブジェクトを content.response に保存しますが、SDK の LLMAsJudge 評価ツールは JSON_VALUE(content, '$.response') がスカラー文字列を返すことを想定しています。このビューでは、$.text_summary と $.response を書式なしテキストの回答に設定し、元のオブジェクトを $.raw_response に保持します。

次の SQL と Python のスニペットで、次のプレースホルダを置き換えます。

  • PROJECT_ID: データ エージェントとログ データセットを含むプロジェクトの ID。

  • LOG_DATASET: 未加工の会話ログテーブルを保存する BigQuery データセットの ID。

  • LOG_TABLE: 会話ログテーブルの ID（agent_events など）。

  • ANALYTICS_DATASET: アナリティクス ビューとテーブル関数を作成する場所と同じ場所にある BigQuery データセットの ID。

  • LOCATION: 分析データセットの BigQuery ロケーション（US など）。

ANALYTICS_DATASET でビューとテーブル関数を作成するには、ANALYTICS_DATASET に対する書き込みアクセス権（BigQuery データ編集者（roles/bigquery.dataEditor）ロールなど）と、LOG_DATASET に対する BigQuery データ閲覧者（roles/bigquery.dataViewer）ロールが必要です。ビューに対してクエリのみを実行するアナリストには、ANALYTICS_DATASET に対する BigQuery データ閲覧者のロールと、クエリを実行するプロジェクトに対する BigQuery ジョブユーザー（roles/bigquery.jobUser）のロールが必要です。

次のクエリは、v_bqca_sdk_events ビューを作成します。このビューは、次のステップでテーブル スキーマとイベント カバレッジを検証するために使用されます。

CREATE OR REPLACE VIEW `PROJECT_ID.ANALYTICS_DATASET.v_bqca_sdk_events` AS
WITH allowlisted AS (
  SELECT
    *,
    IF(
      event_type = 'AGENT_RESPONSE' AND content IS NOT NULL,
      COALESCE(
        NULLIF(JSON_VALUE(content, '$.text_summary'), ''),
        NULLIF(ARRAY_TO_STRING(ARRAY(
          SELECT JSON_VALUE(part, '$.markdown')
          FROM UNNEST(JSON_QUERY_ARRAY(content, '$.response.parts')) AS part WITH OFFSET AS part_offset
          WHERE JSON_VALUE(part, '$.markdown') IS NOT NULL
          ORDER BY part_offset
        ), '\n\n'), ''),
        NULLIF(JSON_VALUE(content, '$.response.clarifying_question'), ''),
        JSON_VALUE(content, '$.response')
      ),
      NULL
    ) AS answer_text
  FROM `PROJECT_ID.LOG_DATASET.LOG_TABLE`
  WHERE event_type IN (
    'INVOCATION_STARTING', 'USER_MESSAGE_RECEIVED', 'AGENT_RESPONSE',
    'INVOCATION_COMPLETED', 'LLM_RESPONSE', 'EMBEDDING_SUGGESTION',
    'INVOCATION_ERROR', 'AGENT_ERROR', 'LLM_ERROR'
  )
)
SELECT
  * EXCEPT(answer_text)
  REPLACE (
    CASE
      WHEN event_type = 'AGENT_RESPONSE' AND content IS NOT NULL THEN
        JSON_SET(
          content,
          '$.raw_response', JSON_QUERY(content, '$.response'),
          '$.text_summary', answer_text,
          '$.response', answer_text
        )
      ELSE content
    END AS content,
    JSON_SET(
      COALESCE(attributes, JSON '{}'),
      '$.root_agent_name',
      JSON_VALUE(attributes, '$.adk.app_name')
    ) AS attributes
  )
FROM allowlisted;

互換性ビューを作成したら、SDK Client を初期化します。必要に応じて、client.doctor() を実行してテーブル スキーマとイベント カバレッジを検証できます。

from datetime import datetime, timedelta, timezone
from bigquery_agent_analytics import Client, TraceFilter

client = Client(
    project_id="PROJECT_ID",
    dataset_id="ANALYTICS_DATASET",
    table_id="v_bqca_sdk_events",
    location="LOCATION",
)

# Optionally run diagnostic checks over recent partitions:
report = client.doctor(
    filters=TraceFilter(start_time=datetime.now(timezone.utc) - timedelta(days=7))
)
print(report)

client.doctor() 関数は、テーブル スキーマを検証し、イベント カバレッジを標準の Agent Development Kit（ADK）イベントタイプと比較する辞書を返します。v_bqca_sdk_events 互換性ビューでは、許可リストに登録された会話分析イベントタイプのみが公開されるため、AGENT_COMPLETED、AGENT_STARTING、LLM_REQUEST、TOOL_COMPLETED、TOOL_STARTING イベントの欠落に関する警告が表示されることが想定されます。エラーまたは列の欠落の警告は、Client パラメータがビューの作成場所と一致しないか、データセットへの読み取りアクセス権がないことを意味します。event_coverage の結果が空の場合は、ビューに過去 7 日間の会話がないことを意味します。

SDK を使用してエージェントの品質を評価し、トレースを再構築して、ドリフトをモニタリングする方法については、BigQuery エージェント分析 SDK を使用すると、このセクションにリンクされているインタラクティブな Colab ノートブックをご覧ください。

SQL レシピ

次の GoogleSQL クエリでは、Colab ノートブックと Python SDK を使用するで説明されている PROJECT_ID、LOG_DATASET、LOG_TABLE、ANALYTICS_DATASET のプレースホルダを使用します。

エージェント間のイベントビュー（v_ca_events）を作成するとターンテーブル関数（ca_turns）を作成するから始めます。次のクエリのほとんどは、ca_turns テーブル関数または v_ca_events ビューを使用します。AI.GENERATE エラーサマリー クエリと重複除去クエリは、ログテーブルを直接読み取ります。

クロスエージェント イベント ビュー（v_ca_events）を作成する

このビューは、イベントごとに 1 行を返します。エージェント ID、会話 ID、質問、回答テキスト、分析情報、トークン、レイテンシがプレーン列として返されます。timestamp のフィルタはログテーブルに渡されるため、パーティションのプルーニングは引き続き機能します。このビューでは、重複するイベントは削除されません。書き込みの再試行はまれですが、重複するイベントを event_id で検査またはフィルタする場合は、重複する行を削除するをご覧ください。

CREATE OR REPLACE VIEW `PROJECT_ID.ANALYTICS_DATASET.v_ca_events` AS
SELECT
  timestamp,
  DATE(timestamp) AS event_date,
  event_id,
  event_type,
  status,
  error_message,
  user_id,
  COALESCE(NULLIF(user_id, ''), JSON_VALUE(attributes, '$.session_metadata.user_id')) AS internal_user_key,
  JSON_VALUE(attributes, '$.session_metadata.state."data-agent-id"') AS data_agent_id,
  JSON_VALUE(attributes, '$.adk.app_name') AS data_agent_resource_name,
  JSON_VALUE(attributes, '$.session_metadata.state."conversation-id"') AS conversation_id,
  session_id AS turn_id,
  invocation_id,
  IF(event_type = 'USER_MESSAGE_RECEIVED', JSON_VALUE(content, '$.text_summary'), NULL) AS user_question,
  IF(event_type = 'AGENT_RESPONSE',
    COALESCE(
      NULLIF(JSON_VALUE(content, '$.text_summary'), ''),
      NULLIF(ARRAY_TO_STRING(ARRAY(
        SELECT JSON_VALUE(p, '$.markdown')
        FROM UNNEST(JSON_QUERY_ARRAY(content, '$.response.parts')) AS p WITH OFFSET AS o
        WHERE JSON_VALUE(p, '$.markdown') IS NOT NULL
        ORDER BY o), '\n\n'), ''),
      NULLIF(JSON_VALUE(content, '$.response.clarifying_question'), ''),
      JSON_VALUE(content, '$.response')),
    NULL) AS agent_response,
  IF(event_type = 'AGENT_RESPONSE',
    NULLIF(ARRAY_TO_STRING(ARRAY(
      SELECT JSON_VALUE(p, '$.insight')
      FROM UNNEST(JSON_QUERY_ARRAY(content, '$.response.parts')) AS p WITH OFFSET AS o
      WHERE JSON_VALUE(p, '$.insight') IS NOT NULL
      ORDER BY o), '\n'), ''),
    NULL) AS agent_insights,
  IF(event_type = 'AGENT_RESPONSE',
    NULLIF(ARRAY_TO_STRING(ARRAY(
      SELECT JSON_VALUE(p, '$.assumption')
      FROM UNNEST(JSON_QUERY_ARRAY(content, '$.response.parts')) AS p WITH OFFSET AS o
      WHERE JSON_VALUE(p, '$.assumption') IS NOT NULL
      ORDER BY o), '\n'), ''),
    NULL) AS agent_assumptions,
  IF(event_type = 'AGENT_RESPONSE',
    NULLIF(ARRAY_TO_STRING(ARRAY(
      SELECT JSON_VALUE(q)
      FROM UNNEST(JSON_QUERY_ARRAY(content, '$.response.followup_questions')) AS q WITH OFFSET AS o
      WHERE JSON_VALUE(q) IS NOT NULL
      ORDER BY o), '\n'), ''),
    NULL) AS followup_questions,
  IF(event_type = 'AGENT_RESPONSE',
    NULLIF(ARRAY_TO_STRING(ARRAY(
      SELECT artifact_name
      FROM UNNEST(JSON_QUERY_ARRAY(content, '$.response.parts')) AS p WITH OFFSET AS o,
        UNNEST([JSON_VALUE(p, '$.data_result_artifact_name'), JSON_VALUE(p, '$.chart_artifact_name')]) AS artifact_name WITH OFFSET AS a_o
      WHERE artifact_name IS NOT NULL
      ORDER BY o, a_o), ', '), ''),
    NULL) AS result_artifact_names,
  IF(event_type = 'LLM_RESPONSE', JSON_VALUE(attributes, '$.model_version'), NULL) AS model_version,
  IF(event_type = 'LLM_RESPONSE', SAFE_CAST(JSON_VALUE(content, '$.usage.prompt') AS INT64), NULL) AS prompt_tokens,
  IF(event_type = 'LLM_RESPONSE', SAFE_CAST(JSON_VALUE(content, '$.usage.completion') AS INT64), NULL) AS completion_tokens,
  IF(event_type = 'LLM_RESPONSE', SAFE_CAST(JSON_VALUE(content, '$.usage.total') AS INT64), NULL) AS total_tokens,
  IF(event_type = 'LLM_RESPONSE', SAFE_CAST(JSON_VALUE(attributes, '$.usage_metadata.thoughts_token_count') AS INT64), NULL) AS thoughts_tokens,
  IF(event_type = 'INVOCATION_COMPLETED', SAFE_CAST(JSON_VALUE(latency_ms, '$.total_ms') AS INT64), NULL) AS turn_latency_ms,
  is_truncated,
  trace_id,
  span_id,
  parent_span_id
FROM `PROJECT_ID.LOG_DATASET.LOG_TABLE`;

ターン テーブル関数（ca_turns）を作成する: 質問と回答ごとに 1 行

このテーブル関数は、時間枠を受け取り、ターンごとに 1 行を返します。時間フィルタがログテーブルに適用されるため、リクエストした日数分のみが課金されます。

CREATE OR REPLACE TABLE FUNCTION `PROJECT_ID.ANALYTICS_DATASET.ca_turns`(start_ts TIMESTAMP, end_ts TIMESTAMP) AS
SELECT
  turn_id,
  MIN(timestamp) AS turn_start,
  MAX(timestamp) AS turn_end,
  DATE(MIN(timestamp)) AS turn_date,
  NULLIF(MAX(user_id), '') AS user_id,
  MAX(internal_user_key) AS internal_user_key,
  MAX(data_agent_id) AS data_agent_id,
  MAX(conversation_id) AS conversation_id,
  ARRAY_AGG(user_question IGNORE NULLS ORDER BY timestamp, event_id LIMIT 1)[SAFE_OFFSET(0)] AS user_question,
  ARRAY_AGG(agent_response IGNORE NULLS ORDER BY timestamp DESC, event_id DESC LIMIT 1)[SAFE_OFFSET(0)] AS agent_response,
  ARRAY_AGG(agent_insights IGNORE NULLS ORDER BY timestamp DESC, event_id DESC LIMIT 1)[SAFE_OFFSET(0)] AS agent_insights,
  ARRAY_AGG(agent_assumptions IGNORE NULLS ORDER BY timestamp DESC, event_id DESC LIMIT 1)[SAFE_OFFSET(0)] AS agent_assumptions,
  ARRAY_AGG(followup_questions IGNORE NULLS ORDER BY timestamp DESC, event_id DESC LIMIT 1)[SAFE_OFFSET(0)] AS followup_questions,
  ARRAY_AGG(result_artifact_names IGNORE NULLS ORDER BY timestamp DESC, event_id DESC LIMIT 1)[SAFE_OFFSET(0)] AS result_artifact_names,
  MAX(turn_latency_ms) AS turn_latency_ms,
  SUM(total_tokens) AS total_tokens,
  COUNTIF(status = 'ERROR') AS error_events,
  LOGICAL_OR(IFNULL(is_truncated, FALSE)) AS any_truncated
FROM `PROJECT_ID.ANALYTICS_DATASET.v_ca_events`
WHERE timestamp >= start_ts
  AND timestamp < end_ts
  AND turn_id IS NOT NULL
GROUP BY turn_id;

1 人のユーザーのすべての質問と回答（過去 30 日間）

target_user をユーザーのメールアドレスに設定するか、メールアドレスのない呼び出し元の内部キーに設定します。

DECLARE target_user STRING DEFAULT 'analyst@example.com';

SELECT turn_start, data_agent_id, conversation_id, user_question, agent_response, agent_insights
FROM `PROJECT_ID.ANALYTICS_DATASET.ca_turns`(
  TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY), CURRENT_TIMESTAMP())
WHERE user_id = target_user
   OR internal_user_key = target_user
ORDER BY turn_start;

すべてのエージェントの質問と回答のサンプルを確認

このクエリは、過去 24 時間のターンの約 10% を返します。0.10 を変更して、レビュー率を設定します。

SELECT data_agent_id, turn_start, user_question, agent_response, agent_assumptions
FROM `PROJECT_ID.ANALYTICS_DATASET.ca_turns`(
  TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY), CURRENT_TIMESTAMP())
WHERE user_question IS NOT NULL
  AND RAND() < 0.10
ORDER BY data_agent_id, turn_start;

曜日別の導入状況

user_id が NULL の場合（一部のフェデレーション ID など）、internal_user_key フィールドは attributes.session_metadata.user_id にフォールバックされるため、1 日あたりのユーザー数はユーザーごとに一貫しています。

SELECT
  turn_date,
  COUNT(DISTINCT internal_user_key) AS active_users,
  COUNT(DISTINCT data_agent_id) AS active_agents,
  COUNT(DISTINCT conversation_id) AS conversations,
  COUNT(*) AS questions
FROM `PROJECT_ID.ANALYTICS_DATASET.ca_turns`(
  TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY), CURRENT_TIMESTAMP())
GROUP BY turn_date
ORDER BY turn_date DESC;

ユーザー別の導入状況: エージェント、会話、1 人あたりの質問数

過去 30 日間のユーザーごとの個別のエージェント、会話、質問をカウントするには、次のクエリを実行します。

SELECT
  t.internal_user_key,
  COUNT(DISTINCT t.data_agent_id) AS agents_used,
  COUNT(DISTINCT t.conversation_id) AS conversations,
  COUNT(*) AS questions,
  MIN(t.turn_start) AS first_seen,
  MAX(t.turn_start) AS last_seen
FROM `PROJECT_ID.ANALYTICS_DATASET.ca_turns`(
  TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY), CURRENT_TIMESTAMP()) AS t
GROUP BY internal_user_key
ORDER BY questions DESC;

エージェント別の導入状況と健全性

プロジェクト内のすべてのデータ エージェントのユーザー、会話、質問数、平均レイテンシ、エラー率、合計トークン数を比較するには、次のクエリを実行します。

SELECT
  data_agent_id,
  COUNT(DISTINCT internal_user_key) AS users,
  COUNT(DISTINCT conversation_id) AS conversations,
  COUNT(*) AS questions,
  ROUND(AVG(turn_latency_ms)) AS avg_turn_latency_ms,
  ROUND(100 * SAFE_DIVIDE(COUNTIF(error_events > 0), COUNT(*)), 2) AS pct_turns_with_errors,
  SUM(total_tokens) AS total_tokens
FROM `PROJECT_ID.ANALYTICS_DATASET.ca_turns`(
  TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY), CURRENT_TIMESTAMP())
GROUP BY data_agent_id
ORDER BY questions DESC;

会話の要約

過去 7 日間のマルチターン会話をロールアップするには、次のクエリを実行します。

SELECT
  conversation_id,
  MAX(data_agent_id) AS data_agent_id,
  MAX(internal_user_key) AS internal_user_key,
  COUNT(*) AS turns,
  MIN(turn_start) AS started_at,
  TIMESTAMP_DIFF(MAX(turn_end), MIN(turn_start), SECOND) AS duration_seconds,
  ARRAY_AGG(user_question IGNORE NULLS ORDER BY turn_start LIMIT 1)[SAFE_OFFSET(0)] AS first_question
FROM `PROJECT_ID.ANALYTICS_DATASET.ca_turns`(
  TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY), CURRENT_TIMESTAMP())
WHERE conversation_id IS NOT NULL
GROUP BY conversation_id
ORDER BY started_at DESC
LIMIT 100;

モデル別およびエージェント別のトークン使用量（過去 7 日間）

過去 7 日間のモデル呼び出し数とトークン使用量（思考トークンを含む）を日付、data_agent_id、model_version で分類するには、v_ca_events で LLM_RESPONSE 行をクエリします。

SELECT
  event_date,
  data_agent_id,
  model_version,
  COUNT(*) AS model_calls,
  SUM(prompt_tokens) AS prompt_tokens,
  SUM(completion_tokens) AS completion_tokens,
  SUM(thoughts_tokens) AS thoughts_tokens,  -- thinking tokens, already included in total_tokens
  SUM(total_tokens) AS total_tokens
FROM `PROJECT_ID.ANALYTICS_DATASET.v_ca_events`
WHERE event_type = 'LLM_RESPONSE'
  AND timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)
GROUP BY event_date, data_agent_id, model_version
ORDER BY event_date DESC, total_tokens DESC;

エージェント別のエラー（AI を活用した根本原因の概要付き）

過去 7 日間のエージェント別および日別のエラー数を確認するには、次のクエリを実行します。

SELECT
  event_date,
  data_agent_id,
  COUNT(*) AS error_events,
  COUNT(DISTINCT turn_id) AS turns_with_errors,
  ANY_VALUE(error_message) AS sample_error
FROM `PROJECT_ID.ANALYTICS_DATASET.v_ca_events`
WHERE status = 'ERROR'
  AND timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)
GROUP BY event_date, data_agent_id
ORDER BY event_date DESC, error_events DESC;

最近のエラーの根本原因を要約するには、AI.GENERATE 関数を使用します。

CTE はモデルが呼び出される前に行数を制限するため、費用を予測可能な状態に保つことができます。

WITH recent_errors AS (
  SELECT
    timestamp,
    JSON_VALUE(attributes, '$.session_metadata.state."data-agent-id"') AS data_agent_id,
    session_id AS turn_id,
    error_message
  FROM `PROJECT_ID.LOG_DATASET.LOG_TABLE`
  WHERE status = 'ERROR'
    AND timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY)
  ORDER BY timestamp DESC
  LIMIT 10
)
SELECT
  timestamp,
  data_agent_id,
  turn_id,
  error_message,
  AI.GENERATE(
    CONCAT(
      'Analyze this BigQuery Conversational Analytics error log and summarize the likely root cause in two sentences. Error: ',
      IFNULL(error_message, '(none)'))
  ).result AS ai_root_cause_summary
FROM recent_errors;

重複する行を削除する

この手順は省略可能です。再試行された書き込みによって、同じイベントが 2 回生成されることがあります。次のクエリは、過去 24 時間の event_id ごとに一意の行を選択する方法を示しています。

SELECT * EXCEPT (row_num)
FROM (
  SELECT
    *,
    ROW_NUMBER() OVER (
      PARTITION BY event_id
      ORDER BY timestamp) AS row_num
  FROM `PROJECT_ID.LOG_DATASET.LOG_TABLE`
  WHERE timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY)
)
WHERE row_num = 1 OR event_id IS NULL;

会話のロギングのトラブルシューティング

会話ログテーブルが見つからない、空である、特定のユーザーのターンが見つからない、または宛先の検証エラーが発生した場合は、会話ロギングのトラブルシューティングをご覧ください。

次のステップ