Unterhaltungslogs analysieren

In diesem Dokument wird beschrieben, wie Sie Konversationslogs für die konversationelle Analyse in BigQuery mithilfe von GoogleSQL-Rezepten, einem interaktiven Colab-Notebook und dem Python SDK für BigQuery Agent Analytics abfragen, bewerten und Fehler beheben. Weitere Informationen zu den Spalten der Zieltabelle und zur JSON-Struktur finden Sie unter Schema für Unterhaltungslogs.

Colab-Notebook und Python SDK verwenden

Sie können die SQL-Rezepte in diesem Dokument zusammen mit dem bigquery-agent-analytics Python SDK (v0.5.4 oder höher) ausführen, indem Sie das Starter-Notebook für Kundenanalysen, FinOps und Bewertung verwenden (in Colaboratory öffnen | auf GitHub ansehen).

Wenn Sie das Python SDK mit Ihrer Tabelle für das Unterhaltungslog verwenden möchten, erstellen Sie zuerst die Kompatibilitätsansicht v_bqca_sdk_events. In AGENT_RESPONSE-Zeilen speichert die konversationelle Analyse ein JSON-Objekt in content.response, während die LLMAsJudge-Evaluatoren des SDK erwarten, dass JSON_VALUE(content, '$.response') einen skalaren String zurückgibt. In dieser Ansicht werden $.text_summary und $.response auf die Nur-Text-Antwort gesetzt, während das ursprüngliche Objekt in $.raw_response beibehalten wird.

Ersetzen Sie in den folgenden SQL- und Python-Snippets diese Platzhalter:

  • PROJECT_ID: die ID des Projekts, das Ihren KI-Datenagenten und Ihr Log-Dataset enthält.

  • LOG_DATASET: die ID des BigQuery-Datasets, in dem die Tabelle mit den Rohdaten des Unterhaltungsprotokolls gespeichert ist.

  • LOG_TABLE: die ID der Tabelle mit dem Unterhaltungsprotokoll, z. B. agent_events.

  • ANALYTICS_DATASET: die ID des BigQuery-Datasets am selben Speicherort, an dem Sie Analyseansichten und Tabellenfunktionen erstellen.

  • LOCATION: der BigQuery-Standort des Analyse-Datasets, z. B. US.

Zum Erstellen von Ansichten und Tabellenfunktionen in ANALYTICS_DATASET ist Schreibzugriff auf ANALYTICS_DATASET erforderlich, z. B. die Rolle BigQuery-Dateneditor (roles/bigquery.dataEditor) und die Rolle BigQuery-Datenbetrachter (roles/bigquery.dataViewer) für LOG_DATASET. Analysten, die nur die Ansichten abfragen, benötigen die Rolle „BigQuery-Datenbetrachter“ für ANALYTICS_DATASET und die Rolle BigQuery-Jobnutzer (roles/bigquery.jobUser) für das Projekt, in dem sie Abfragen ausführen.

Mit der folgenden Abfrage wird die Ansicht v_bqca_sdk_events erstellt, die im nächsten Schritt verwendet wird, um das Tabellenschema und die Ereignisabdeckung zu validieren:

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;

Initialisieren Sie nach dem Erstellen der Kompatibilitätsansicht das SDK Client. Sie können optional client.doctor() ausführen, um das Tabellenschema und die Ereignisabdeckung zu validieren:

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)

Die Funktion client.doctor() gibt ein Dictionary zurück, das das Tabellenschema validiert und die Ereignisabdeckung mit den Standardereignistypen des Agent Development Kit (ADK) vergleicht. Da in der Kompatibilitätsansicht v_bqca_sdk_events nur auf der Zulassungsliste stehende Ereignistypen für die konversationelle Analyse angezeigt werden, sind Warnungen zu fehlenden AGENT_COMPLETED-, AGENT_STARTING-, LLM_REQUEST-, TOOL_COMPLETED- oder TOOL_STARTING-Ereignissen zu erwarten. Ein Fehler oder eine Warnung zu einer fehlenden Spalte bedeutet, dass die Client-Parameter nicht mit dem Ort übereinstimmen, an dem die Ansicht erstellt wurde, oder dass Sie keinen Lesezugriff auf das Dataset haben. Ein leeres event_coverage-Ergebnis bedeutet, dass die Ansicht keine Unterhaltungen aus den letzten 7 Tagen enthält.

Weitere Informationen zur Verwendung des SDK zum Bewerten der Agent-Qualität, zum Rekonstruieren von Traces und zum Überwachen von Abweichungen finden Sie unter BigQuery Agent Analytics SDK verwenden und im interaktiven Colab-Notebook, das in diesem Abschnitt verlinkt ist.

SQL-Rezepte

Verwenden Sie in den folgenden GoogleSQL-Abfragen die Platzhalter PROJECT_ID, LOG_DATASET, LOG_TABLE und ANALYTICS_DATASET, die unter Colab-Notebook und Python SDK verwenden beschrieben sind.

Beginnen Sie mit Ansicht für agentübergreifende Ereignisse erstellen (v_ca_events) und Tabellenfunktion für Turns erstellen (ca_turns). Die meisten der folgenden Abfragen verwenden die Tabellenfunktion ca_turns oder die Ansicht v_ca_events. Die Abfrage AI.GENERATE zur Fehlerzusammenfassung und die Abfrage zur Deduplizierung lesen die Protokolltabelle direkt.

Agentenübergreifende Ereignisansicht (v_ca_events) erstellen

Diese Ansicht gibt eine Zeile pro Ereignis zurück, mit Agent-ID, Konversations-ID, Frage, Antworttext, Insights, Tokens und Latenz als einfache Spalten. Filter für timestamp werden an die Logtabelle weitergegeben, sodass die Partitionsbereinigung weiterhin funktioniert. In der Ansicht werden doppelte Ereignisse nicht entfernt. Wiederholungen von Schreibvorgängen sind selten. Wenn Sie jedoch doppelte Ereignisse nach event_id prüfen oder filtern möchten, finden Sie weitere Informationen unter Doppelte Zeilen entfernen:

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`;

Erstellen Sie die Tabellenfunktion für die Turns (ca_turns): eine Zeile pro Frage und Antwort.

Diese Tabellenfunktion verwendet ein Zeitfenster und gibt eine Zeile pro Runde zurück. Da der Zeitfilter auf die Logtabelle angewendet wird, zahlen Sie nur für die Tage, die Sie anfordern:

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;

Alle Fragen und Antworten für einen Nutzer (letzte 30 Tage)

Legen Sie für target_user die E-Mail-Adresse des Nutzers oder den internen Schlüssel für Aufrufer ohne E-Mail-Adresse fest:

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;

Beispielfragen und -antworten aller Agenten zur Überprüfung

Diese Abfrage gibt etwa 10 % der Turns der letzten 24 Stunden zurück. Ändern Sie 0.10, um die Überprüfungsrate festzulegen:

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;

Akzeptanz nach Tag

Das Feld internal_user_key wird auf attributes.session_metadata.user_id zurückgesetzt, wenn user_id den Wert NULL hat (z. B. bei einigen föderierten Identitäten). So bleibt die tägliche Nutzerzahl pro Person konsistent:

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;

Annahme durch Nutzer: KI-Agenten, Unterhaltungen und Fragen pro Person

Führen Sie die folgende Abfrage aus, um die Anzahl der unterschiedlichen Agenten, Unterhaltungen und Fragen pro Nutzer in den letzten 30 Tagen zu zählen:

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;

Akzeptanz und Zustand nach KI-Agenten

Führen Sie die folgende Abfrage aus, um Nutzer, Unterhaltungen, Fragevolumen, durchschnittliche Latenz, Fehlerrate und Gesamtzahl der Tokens für alle KI-Datenagenten in Ihrem Projekt zu vergleichen:

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;

Zusammenfassungen von Unterhaltungen

Führen Sie die folgende Abfrage aus, um Multi-Turn-Unterhaltungen der letzten 7 Tage zusammenzufassen:

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;

Tokennutzung nach Modell und KI-Agent (letzte 7 Tage)

Wenn Sie die Anzahl der Modellaufrufe und die Tokennutzung (einschließlich Thinking-Tokens) nach Datum, data_agent_id und model_version für die letzten 7 Tage aufschlüsseln möchten, fragen Sie v_ca_events nach LLM_RESPONSE-Zeilen ab:

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;

Fehler nach KI-Agent, mit KI-gestützten Zusammenfassungen der Ursachen

Führen Sie die folgende Abfrage aus, um die Fehleranzahl nach Agent und Tag in den letzten 7 Tagen zu prüfen:

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;

Verwenden Sie die Funktion AI.GENERATE, um die Ursache der letzten Fehler zusammenzufassen.

Die CTE begrenzt die Zeilen, bevor das Modell aufgerufen wird, sodass die Kosten vorhersehbar bleiben:

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;

Doppelte Zeilen entfernen

Dieser Schritt ist optional. Ein wiederholter Schreibvorgang kann gelegentlich dasselbe Ereignis zweimal erzeugen. Die folgende Abfrage zeigt, wie Sie eindeutige Zeilen pro event_id für die letzten 24 Stunden auswählen:

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;

Fehlerbehebung beim Logging von Unterhaltungen

Wenn die Tabelle mit dem Unterhaltungsprotokoll fehlt, leer ist oder Beiträge von bestimmten Nutzern fehlen oder wenn Fehler bei der Zielvalidierung auftreten, finden Sie weitere Informationen unter Fehlerbehebung beim Unterhaltungsprotokoll.

Nächste Schritte