Analyser les journaux de conversations

Ce document explique comment interroger, évaluer et résoudre les problèmes liés aux journaux de conversation de l'analyse conversationnelle dans BigQuery à l'aide de recettes GoogleSQL, d'un notebook Colab interactif et du SDK Python BigQuery Agent Analytics. Pour en savoir plus sur les colonnes de la table de destination et la structure JSON, consultez Schéma du journal de conversation.

Utiliser le notebook Colab et le SDK Python

Vous pouvez exécuter les recettes SQL de ce document avec le SDK Python bigquery-agent-analytics (v0.5.4 ou ultérieure) à l'aide du notebook de démarrage Customer Analytics, FinOps & Evaluation (ouvrir dans Colaboratory | afficher sur GitHub).

Pour utiliser le SDK Python avec votre table de journaux de conversation, commencez par créer la vue de compatibilité v_bqca_sdk_events. Sur les lignes AGENT_RESPONSE, l'analyse conversationnelle stocke un objet JSON dans content.response, tandis que les évaluateurs LLMAsJudge du SDK s'attendent à ce que JSON_VALUE(content, '$.response') renvoie une chaîne scalaire. Cette vue définit $.text_summary et $.response sur la réponse en texte brut tout en conservant l'objet d'origine dans $.raw_response.

Dans les extraits SQL et Python suivants, remplacez ces espaces réservés :

• PROJECT_ID : ID du projet contenant votre agent de données et votre ensemble de données de journaux.

• LOG_DATASET : ID de l'ensemble de données BigQuery qui stocke votre table de journaux de conversation bruts.

• LOG_TABLE : ID de la table de journaux de conversation, par exemple agent_events.

• ANALYTICS_DATASET : ID de l'ensemble de données BigQuery au même emplacement que celui où vous créez des vues d'analyse et des fonctions de table.

• LOCATION : emplacement BigQuery de l'ensemble de données d'analyse, par exemple US.

La création de vues et de fonctions de table dans ANALYTICS_DATASET nécessite un accès en écriture sur ANALYTICS_DATASET, tel que le rôle Éditeur de données BigQuery (roles/bigquery.dataEditor), et le rôle Lecteur de données BigQuery (roles/bigquery.dataViewer) sur LOG_DATASET. Les analystes qui n'interrogent que les vues ont besoin du rôle "Lecteur de données BigQuery" sur ANALYTICS_DATASET et du rôle Utilisateur de job BigQuery (roles/bigquery.jobUser) sur le projet dans lequel ils exécutent des requêtes.

La requête suivante crée la vue v_bqca_sdk_events, qui est utilisée à l'étape suivante pour valider le schéma de votre table et la couverture des événements :

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;

Après avoir créé la vue de compatibilité, initialisez le SDK Client. Vous pouvez éventuellement exécuter client.doctor() pour valider le schéma de votre table et la couverture des événements :

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)

La fonction client.doctor() renvoie un dictionnaire qui valide le schéma de la table et compare la couverture des événements aux types d'événements standards d'Agent Development Kit (ADK). Étant donné que la vue de compatibilité v_bqca_sdk_events n'expose que les types d'événements d'analyse conversationnelle figurant sur la liste d'autorisation, des avertissements concernant des événements AGENT_COMPLETED, AGENT_STARTING, LLM_REQUEST, TOOL_COMPLETED ou TOOL_STARTING manquants sont attendus. Une erreur ou un avertissement de colonne manquante signifie que les paramètres Client ne correspondent pas à l'emplacement où la vue a été créée ou que vous ne disposez pas d'un accès en lecture à l'ensemble de données, tandis qu'un résultat event_coverage vide signifie que la vue ne contient aucune conversation des sept derniers jours.

Pour en savoir plus sur l'utilisation du SDK pour évaluer la qualité des agents, reconstruire les traces et surveiller la dérive, consultez Utiliser le SDK BigQuery Agent Analytics et le notebook Colab interactif associé dans cette section.

Recettes SQL

Dans les requêtes GoogleSQL suivantes, utilisez les mêmes espaces réservés PROJECT_ID, LOG_DATASET, LOG_TABLE et ANALYTICS_DATASET que ceux décrits dans Utiliser le notebook Colab et le SDK Python.

Commencez par Créer la vue des événements inter-agents (v_ca_events) et Créer la fonction de table des tours (ca_turns). La plupart des requêtes suivantes utilisent la fonction de table ca_turns ou la vue v_ca_events. La requête de résumé des erreurs AI.GENERATE et la requête de déduplication lisent directement la table de journaux.

Créer la vue des événements multi-agents (v_ca_events)

Cette vue renvoie une ligne par événement, avec l'ID de l'agent, l'ID de la conversation, la question, le texte de la réponse, les insights, les jetons et la latence sous forme de colonnes simples. Les filtres sur timestamp sont transmis à la table de journaux, de sorte que l'élimination des partitions fonctionne toujours. La vue ne supprime pas les événements en double. Les nouvelles tentatives d'écriture sont rares, mais si vous souhaitez inspecter ou filtrer les événements en double par event_id, consultez Supprimer les lignes en double :

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

Créez la fonction de table des tours (ca_turns) : une ligne par question et réponse.

Cette fonction de table prend une période et renvoie une ligne par tour. Comme elle applique le filtre temporel à la table de journaux, vous ne payez que pour les jours que vous demandez :

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;

Toutes les questions et réponses d'un utilisateur (30 derniers jours)

Définissez target_user sur l'adresse e-mail de l'utilisateur ou sur la clé interne pour les appelants sans adresse e-mail :

DECLARE target_user STRING DEFAULT '[e-mail]';

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;

Exemples de questions et de réponses de tous les agents pour examen

Cette requête renvoie environ 10 % des tours des dernières 24 heures. Remplacez 0.10 pour définir votre taux d'examen :

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;

Adoption par jour

Le champ internal_user_key est remplacé par attributes.session_metadata.user_id lorsque user_id est NULL (par exemple, pour certaines identités fédérées). Ainsi, le nombre d'utilisateurs quotidiens reste cohérent par personne :

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;

Adoption par utilisateur : agents, conversations et questions par personne

Pour compter les agents, les conversations et les questions distincts par utilisateur au cours des 30 derniers jours, exécutez la requête suivante :

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;

Adoption et état par agent

Pour comparer les utilisateurs, les conversations, le volume de questions, la latence moyenne, le taux d'erreur et le nombre total de jetons pour tous les agents de données de votre projet, exécutez la requête suivante :

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;

Résumés des conversations

Pour regrouper les conversations multitours des sept derniers jours, exécutez la requête suivante :

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;

Utilisation de jetons par modèle et par agent (7 derniers jours)

Pour ventiler le nombre d'appels de modèle et l'utilisation des jetons (y compris les jetons de raisonnement) par date, data_agent_id et model_version au cours des sept derniers jours, interrogez v_ca_events pour les lignes 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;

Erreurs par agent, avec des résumés des causes profondes assistés par l'IA

Pour vérifier le nombre d'erreurs par agent et par jour au cours des sept derniers jours, exécutez la requête suivante :

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;

Pour résumer la cause principale des erreurs récentes, utilisez la fonction AI.GENERATE.

Attention : Cette requête envoie des messages d'erreur de vos journaux à un modèle Gemini sur Gemini Enterprise Agent Platform dans votre projet. Vérifiez d'abord vos règles de gouvernance des données. La personne qui exécute la requête doit disposer du rôle Utilisateur de la plate-forme d'agent (roles/aiplatform.user). Vous pouvez également utiliser une connexion BigQuery qui dispose de ce rôle.

La CTE limite le nombre de lignes avant l'appel du modèle, ce qui permet de maintenir un coût prévisible :

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;

Supprimer les lignes en double

Cette étape est facultative. Une nouvelle tentative d'écriture peut parfois produire deux fois le même événement. La requête suivante montre comment sélectionner des lignes uniques par event_id pour les dernières 24 heures :

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;

Résoudre les problèmes de journalisation des conversations

Si votre table de journalisation des conversations est manquante, vide ou ne contient pas les tours de parole d'utilisateurs spécifiques, ou si vous rencontrez des erreurs de validation de la destination, consultez Résoudre les problèmes de journalisation des conversations.

Étapes suivantes

• Examinez le schéma du journal de conversation.

• Consultez la présentation de la journalisation des conversations.

• Créer et configurer des agents de données