Analiza los registros de conversaciones

En este documento, se muestra cómo consultar, evaluar y solucionar problemas de los registros de conversaciones de Conversational Analytics en BigQuery con recetas de GoogleSQL, un notebook de Colab interactivo y el SDK de Python de análisis de agentes de BigQuery. Para obtener más información sobre las columnas de la tabla de destino y la estructura de JSON, consulta Esquema de registro de conversaciones.

Usa el notebook de Colab y el SDK de Python

Puedes ejecutar las recetas de SQL de este documento junto con el SDK de Python de bigquery-agent-analytics (v0.5.4 o posterior) con el notebook de inicio de evaluación, FinOps y estadísticas de clientes (abrir en Colaboratory | ver en GitHub).

Para usar el SDK de Python con tu tabla de registro de conversaciones, primero crea la vista de compatibilidad de v_bqca_sdk_events. En las filas AGENT_RESPONSE, las estadísticas de conversación almacenan un objeto JSON en content.response, mientras que los evaluadores LLMAsJudge del SDK esperan que JSON_VALUE(content, '$.response') devuelva una cadena escalar. Esta vista establece $.text_summary y $.response en la respuesta de texto sin formato, mientras que mantiene el objeto original en $.raw_response.

En los siguientes fragmentos de SQL y Python, reemplaza estos marcadores de posición:

  • PROJECT_ID: Es el ID del proyecto que contiene tu agente de datos y el conjunto de datos de registros.

  • LOG_DATASET: Es el ID del conjunto de datos de BigQuery que almacena la tabla de registro de conversaciones sin procesar.

  • LOG_TABLE: Es el ID de la tabla de registro de conversaciones, como agent_events.

  • ANALYTICS_DATASET: Es el ID del conjunto de datos de BigQuery en la misma ubicación en la que creas vistas de análisis y funciones de tabla.

  • LOCATION: Es la ubicación de BigQuery del conjunto de datos de estadísticas, como US.

Para crear vistas y funciones de tabla en ANALYTICS_DATASET, se requiere acceso de escritura en ANALYTICS_DATASET, como el rol de editor de datos de BigQuery (roles/bigquery.dataEditor) y el rol de visualizador de datos de BigQuery (roles/bigquery.dataViewer) en LOG_DATASET. Los analistas que solo consultan las vistas necesitan el rol de visualizador de datos de BigQuery en ANALYTICS_DATASET y el rol de usuario de trabajo de BigQuery (roles/bigquery.jobUser) en el proyecto en el que ejecutan las consultas.

La siguiente consulta crea la vista v_bqca_sdk_events, que se usa en el siguiente paso para validar el esquema de la tabla y la cobertura de eventos:

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;

Después de crear la vista de compatibilidad, inicializa el SDK Client. De manera opcional, puedes ejecutar client.doctor() para validar el esquema de la tabla y la cobertura de eventos:

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 función client.doctor() devuelve un diccionario que valida el esquema de la tabla y compara la cobertura de eventos con los tipos de eventos estándar del Kit de desarrollo de agentes (ADK). Debido a que la vista de compatibilidad v_bqca_sdk_events solo expone los tipos de eventos de estadísticas de conversación incluidos en la lista de entidades permitidas, se esperan advertencias sobre eventos AGENT_COMPLETED, AGENT_STARTING, LLM_REQUEST, TOOL_COMPLETED o TOOL_STARTING faltantes. Un error o una advertencia de columna faltante significa que los parámetros Client no coinciden con el lugar donde se creó la vista o que no tienes acceso de lectura al conjunto de datos, mientras que un resultado event_coverage vacío significa que la vista no tiene conversaciones de los últimos 7 días.

Para obtener más información sobre el uso del SDK para evaluar la calidad de los agentes, reconstruir los seguimientos y supervisar la desviación, consulta Usa el SDK de análisis de agentes de BigQuery y el notebook de Colab interactivo vinculado en esta sección.

Recetas de SQL

En las siguientes consultas de GoogleSQL, usa los mismos marcadores de posición PROJECT_ID, LOG_DATASET, LOG_TABLE y ANALYTICS_DATASET que se describen en Usa el notebook de Colab y el SDK de Python.

Comienza con Crea la vista de eventos entre agentes (v_ca_events) y Crea la función de tabla de turnos (ca_turns). La mayoría de las siguientes consultas usan la función de tabla ca_turns o la vista v_ca_events; la consulta de resumen de errores AI.GENERATE y la consulta de anulación de duplicación leen la tabla de registro directamente.

Crea la vista de eventos entre agentes (v_ca_events)

Esta vista devuelve una fila por evento, con el ID del agente, el ID de la conversación, la pregunta, el texto de la respuesta, las estadísticas, los tokens y la latencia como columnas sin formato. Los filtros en timestamp pasan a la tabla de registro, por lo que la reducción de particiones sigue funcionando. La vista no quita los eventos duplicados. Los reintentos de escritura son poco frecuentes, pero si quieres inspeccionar o filtrar los eventos duplicados por event_id, consulta Quita las filas duplicadas:

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

Crea la función de tabla de turnos (ca_turns): una fila por pregunta y respuesta

Esta función de tabla toma un período y devuelve una fila por turno. Debido a que aplica el filtro de tiempo a la tabla de registros, solo pagas por los días que solicitas:

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;

Todas las preguntas y respuestas de un usuario (últimos 30 días)

Establece target_user en la dirección de correo electrónico del usuario o en la clave interna para los emisores sin una dirección de correo electrónico:

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;

Preguntas y respuestas de muestra de todos los agentes para su revisión

Esta consulta devuelve alrededor del 10% de los turnos de las últimas 24 horas. Cambia 0.10 para establecer tu tasa de revisión:

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;

Adopción por día

El campo internal_user_key recurre a attributes.session_metadata.user_id cuando user_id es NULL (por ejemplo, para algunas identidades federadas), por lo que los recuentos diarios de usuarios se mantienen coherentes por persona:

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;

Adopción por usuario: agentes, conversaciones y preguntas por persona

Para contar los agentes, las conversaciones y las preguntas distintas por usuario en los últimos 30 días, ejecuta la siguiente consulta:

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;

Adopción y estado por agente

Para comparar usuarios, conversaciones, volumen de preguntas, latencia promedio, tasa de errores y tokens totales en todos los agentes de datos de tu proyecto, ejecuta la siguiente consulta:

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;

Resúmenes de conversaciones

Para resumir las conversaciones de varios turnos de los últimos 7 días, ejecuta la siguiente consulta:

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;

Uso de tokens por modelo y por agente (últimos 7 días)

Para desglosar los recuentos de llamadas al modelo y el uso de tokens (incluidos los tokens de razonamiento) por fecha, data_agent_id y model_version durante los últimos 7 días, consulta v_ca_events para las filas 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;

Errores por agente, con resúmenes de causas raíz asistidos por IA

Para verificar los recuentos de errores por agente y día durante los últimos 7 días, ejecuta la siguiente consulta:

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;

Para resumir la causa raíz de los errores recientes, usa la función AI.GENERATE.

La CTE limita las filas antes de que se llame al modelo, por lo que el costo se mantiene predecible:

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;

Quitar filas duplicadas

Este paso es opcional. En ocasiones, una escritura reintentada puede producir el mismo evento dos veces. En la siguiente consulta, se muestra cómo seleccionar filas únicas por event_id durante las últimas 24 horas:

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;

Soluciona problemas del registro de conversaciones

Si falta la tabla de registro de conversaciones, está vacía o faltan turnos de usuarios específicos, o si encuentras errores de validación de destino, consulta Soluciona problemas de registro de conversaciones.

¿Qué sigue?