Analisar registros de conversas

Neste documento, mostramos como consultar, avaliar e resolver problemas de registros de conversas de análise de conversação no BigQuery usando receitas do GoogleSQL, um notebook interativo do Colab e o SDK do Python de análise de agentes do BigQuery. Para mais informações sobre as colunas da tabela de destino e a estrutura JSON, consulte Esquema de registro de conversas.

Usar o notebook do Colab e o SDK para Python

É possível executar as receitas SQL neste documento com o SDK do Python do bigquery-agent-analytics (v0.5.4 ou mais recente) usando o notebook inicial de análise de clientes, FinOps e avaliação (abrir no Colaboratory | visualizar no GitHub).

Para usar o SDK do Python com a tabela de registro de conversas, primeiro crie a visualização de compatibilidade v_bqca_sdk_events. Nas linhas AGENT_RESPONSE, a análise de conversação armazena um objeto JSON em content.response, enquanto os avaliadores LLMAsJudge do SDK esperam que JSON_VALUE(content, '$.response') retorne uma string escalar. Essa visualização define $.text_summary e $.response como a resposta em texto simples, mantendo o objeto original em $.raw_response.

Nos snippets de SQL e Python a seguir, substitua estes marcadores de posição:

  • PROJECT_ID: o ID do projeto que contém o agente de dados e o conjunto de dados de registro.

  • LOG_DATASET: o ID do conjunto de dados do BigQuery que armazena a tabela de registros de conversas brutos.

  • LOG_TABLE: o ID da tabela de registro de conversas, como agent_events.

  • ANALYTICS_DATASET: o ID do conjunto de dados do BigQuery no mesmo local em que você cria visualizações de análise e funções de tabela.

  • LOCATION: o local do BigQuery do conjunto de dados de análise, como US.

Para criar visualizações e funções de tabela no ANALYTICS_DATASET, é necessário ter acesso de gravação no ANALYTICS_DATASET, como o papel de Editor de dados do BigQuery (roles/bigquery.dataEditor), e o papel de Leitor de dados do BigQuery (roles/bigquery.dataViewer) no LOG_DATASET. Os analistas que apenas consultam as visualizações de dados precisam do papel de Leitor de dados do BigQuery em ANALYTICS_DATASET e do papel de Usuário de jobs do BigQuery (roles/bigquery.jobUser) no projeto em que executam as consultas.

A consulta a seguir cria a visualização v_bqca_sdk_events, que é usada na próxima etapa para validar o esquema da tabela e a 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;

Depois de criar a visualização de compatibilidade, inicialize o SDK Client. Se quiser, execute client.doctor() para validar o esquema da tabela e a 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)

A função client.doctor() retorna um dicionário que valida o esquema da tabela e compara a cobertura de eventos com os tipos de evento padrão do Kit de Desenvolvimento de Agente (ADK). Como a visualização de compatibilidade v_bqca_sdk_events expõe apenas tipos de eventos de análise de conversação na lista de permissões, são esperados avisos sobre eventos AGENT_COMPLETED, AGENT_STARTING, LLM_REQUEST, TOOL_COMPLETED ou TOOL_STARTING ausentes. Um erro ou um aviso de coluna ausente significa que os parâmetros Client não correspondem ao local em que a visualização foi criada ou que você não tem acesso de leitura ao conjunto de dados, enquanto um resultado event_coverage vazio significa que a visualização não tem conversas dos últimos sete dias.

Para mais informações sobre como usar o SDK para avaliar a qualidade do agente, reconstruir traces e monitorar desvios, consulte Usar o SDK de análise de agentes do BigQuery e o notebook interativo do Colab vinculado nesta seção.

Receitas SQL

Nas consultas do GoogleSQL a seguir, use os mesmos marcadores de posição PROJECT_ID, LOG_DATASET, LOG_TABLE e ANALYTICS_DATASET descritos em Usar o notebook do Colab e o SDK do Python.

Comece com Criar a visualização de eventos entre agentes (v_ca_events) e Criar a função de tabela de turnos (ca_turns). A maioria das consultas a seguir usa a função de tabela ca_turns ou a visualização v_ca_events. A consulta de resumo de erros AI.GENERATE e a consulta de eliminação de duplicação leem a tabela de registros diretamente.

Criar a visualização de eventos entre agentes (v_ca_events)

Essa visualização retorna uma linha por evento, com ID do agente, ID da conversa, pergunta, texto da resposta, insights, tokens e latência como colunas simples. Os filtros em timestamp são transmitidos para a tabela de registros, de modo que a remoção de partições ainda funcione. A visualização não remove eventos duplicados. Novas tentativas de gravação são raras, mas se você quiser inspecionar ou filtrar eventos duplicados por event_id, consulte Remover linhas 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`;

Crie a função de tabela de turnos (ca_turns): uma linha por pergunta e resposta

Essa função de tabela usa uma janela de tempo e retorna uma linha por turno. Como ele aplica o filtro de tempo à tabela de registros, você paga apenas pelos dias que solicitar:

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 as perguntas e respostas de um usuário (últimos 30 dias)

Defina target_user como o endereço de e-mail do usuário ou como a chave interna para autores de chamadas sem um endereço de e-mail:

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;

Exemplos de perguntas e respostas de todos os agentes para revisão

Essa consulta retorna cerca de 10% das interações das últimas 24 horas. Mude 0.10 para definir a taxa de revisão:

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;

Adoção por dia

O campo internal_user_key volta para attributes.session_metadata.user_id quando user_id é NULL (por exemplo, para algumas identidades federadas), de modo que as contagens diárias de usuários permaneçam consistentes por pessoa:

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;

Adoção por usuário: agentes, conversas e perguntas por pessoa

Para contar agentes, conversas e perguntas distintas por usuário nos últimos 30 dias, execute a seguinte 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;

Adoção e integridade por agente

Para comparar usuários, conversas, volume de perguntas, latência média, taxa de erros e total de tokens em todos os agentes de dados do seu projeto, execute a seguinte 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;

Resumos de conversas

Para agrupar conversas de vários turnos nos últimos sete dias, execute a seguinte 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 e por agente (últimos 7 dias)

Para detalhar as contagens de chamadas de modelo e o uso de tokens (incluindo tokens de raciocínio) por data, data_agent_id e model_version nos últimos 7 dias, consulte v_ca_events para linhas 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;

Erros por agente, com resumos de causa raiz assistidos por IA

Para verificar as contagens de erros por agente e dia nos últimos sete dias, execute a seguinte 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 a causa raiz de erros recentes, use a função AI.GENERATE.

A CTE limita as linhas antes que o modelo seja chamado, de modo que o custo permaneça previsível:

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;

Remover linhas duplicadas

Esta etapa é opcional. Uma gravação repetida pode, ocasionalmente, produzir o mesmo evento duas vezes. A consulta a seguir demonstra como selecionar linhas exclusivas por event_id nas ú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;

Resolver problemas de registro de conversas

Se a tabela de registro de conversas estiver ausente, vazia ou sem turnos de usuários específicos, ou se você encontrar erros de validação de destino, consulte Resolver problemas de registro de conversas.

A seguir