Analizza i log delle conversazioni

Questo documento mostra come eseguire query, valutare e risolvere i problemi dei log delle conversazioni di analisi conversazionale in BigQuery utilizzando ricette GoogleSQL, un notebook Colab interattivo e l'SDK Python di BigQuery Agent Analytics. Per saperne di più sulle colonne della tabella di destinazione e sulla struttura JSON, consulta Schema dei log delle conversazioni.

Utilizza il notebook di Colab e l'SDK Python

Puoi eseguire le formule SQL di questo documento insieme all'SDK Python bigquery-agent-analytics (v0.5.4 o successive) utilizzando il blocco note iniziale per l'analisi dei clienti, FinOps e valutazione (apri in Colaboratory | visualizza su GitHub).

Per utilizzare l'SDK Python con la tabella dei log delle conversazioni, crea prima la vista di compatibilità v_bqca_sdk_events. Nelle righe AGENT_RESPONSE, l'analisi conversazionale archivia un oggetto JSON in content.response, mentre i valutatori LLMAsJudge dell'SDK si aspettano che JSON_VALUE(content, '$.response') restituisca una stringa scalare. Questa visualizzazione imposta $.text_summary e $.response sulla risposta in testo normale, mantenendo l'oggetto originale in $.raw_response.

Negli snippet SQL e Python seguenti, sostituisci questi segnaposto:

• PROJECT_ID: l'ID del progetto che contiene l'agente di dati e il set di dati dei log.

• LOG_DATASET: l'ID del set di dati BigQuery che archivia la tabella dei log delle conversazioni non elaborati.

• LOG_TABLE: l'ID della tabella del log delle conversazioni, ad esempio agent_events.

• ANALYTICS_DATASET: l'ID del set di dati BigQuery nella stessa località in cui crei le viste di analisi e le funzioni di tabella.

• LOCATION: la località BigQuery del set di dati di analisi, ad esempio US.

La creazione di viste e funzioni di tabella in ANALYTICS_DATASET richiede l'accesso in scrittura a ANALYTICS_DATASET, ad esempio il ruolo BigQuery Data Editor (roles/bigquery.dataEditor) e il ruolo BigQuery Data Viewer (roles/bigquery.dataViewer) su LOG_DATASET. Gli analisti che eseguono query solo sulle viste devono avere il ruolo BigQuery Data Viewer su ANALYTICS_DATASET e il ruolo BigQuery Job User (roles/bigquery.jobUser) sul progetto in cui eseguono le query.

La seguente query crea la vista v_bqca_sdk_events, che viene utilizzata nel passaggio successivo per convalidare lo schema della tabella e la copertura degli eventi:

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;

Dopo aver creato la visualizzazione di compatibilità, inizializza l'SDK Client. Se vuoi, puoi eseguire client.doctor() per convalidare lo schema della tabella e la copertura degli eventi:

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 funzione client.doctor() restituisce un dizionario che convalida lo schema della tabella e confronta la copertura degli eventi con i tipi di eventi standard dell'Agent Development Kit (ADK). Poiché la vista di compatibilità v_bqca_sdk_events espone solo i tipi di eventi di analisi conversazionale inclusi nella lista consentita, sono previsti avvisi relativi a eventi AGENT_COMPLETED, AGENT_STARTING, LLM_REQUEST, TOOL_COMPLETED o TOOL_STARTING mancanti. Un errore o un avviso di colonna mancante indica che i parametri Client non corrispondono al punto in cui è stata creata la visualizzazione o che non disponi dell'accesso in lettura al set di dati, mentre un risultato event_coverage vuoto indica che la visualizzazione non contiene conversazioni degli ultimi 7 giorni.

Per saperne di più sull'utilizzo dell'SDK per valutare la qualità degli agenti, ricostruire le tracce e monitorare la deviazione, consulta Utilizza l'SDK BigQuery Agent Analytics e il notebook di Colab interattivo collegato in questa sezione.

Ricette SQL

Nelle seguenti query GoogleSQL, utilizza gli stessi segnaposto PROJECT_ID, LOG_DATASET, LOG_TABLE e ANALYTICS_DATASET descritti in Utilizza il notebook di Colab e l'SDK Python.

Inizia con Crea la vista degli eventi tra agenti (v_ca_events) e Crea la funzione della tabella dei turni (ca_turns). La maggior parte delle query seguenti utilizza la funzione della tabella ca_turns o la vista v_ca_events; la query di riepilogo degli errori AI.GENERATE e la query di deduplicazione leggono direttamente la tabella di log.

Crea la visualizzazione degli eventi tra agenti (v_ca_events)

Questa vista restituisce una riga per evento, con ID agente, ID conversazione, domanda, testo della risposta, insight, token e latenza come colonne semplici. I filtri su timestamp passano alla tabella di log, quindi la riduzione delle partizioni funziona ancora. La vista non rimuove gli eventi duplicati; i nuovi tentativi di scrittura sono rari, ma se vuoi ispezionare o filtrare gli eventi duplicati in base a event_id, consulta Rimuovi le righe duplicate:

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 funzione di tabella dei turni (ca_turns): una riga per ogni domanda e risposta

Questa funzione di tabella accetta una finestra temporale e restituisce una riga per turno. Poiché applica il filtro temporale alla tabella dei log, paghi solo per i giorni richiesti:

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;

Ogni domanda e risposta per un utente (ultimi 30 giorni)

Imposta target_user sull'indirizzo email dell'utente o sulla chiave interna per i chiamanti senza un indirizzo email:

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;

Domande e risposte di esempio per tutti gli agenti da esaminare

Questa query restituisce circa il 10% dei turni delle ultime 24 ore. Modifica 0.10 per impostare la frequenza di revisione:

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;

Adozione per giorno

Il campo internal_user_key viene sostituito da attributes.session_metadata.user_id quando user_id è NULL (ad esempio, per alcune identità federate), in modo che i conteggi degli utenti giornalieri rimangano coerenti per 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;

Adozione per utente: agenti, conversazioni e domande per persona

Per conteggiare agenti, conversazioni e domande distinti per utente negli ultimi 30 giorni, esegui questa query:

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;

Adozione e integrità per agente

Per confrontare utenti, conversazioni, volume delle domande, latenza media, tasso di errore e token totali in tutti gli agenti di dati del tuo progetto, esegui la seguente query:

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;

Riepiloghi delle conversazioni

Per raggruppare le conversazioni multi-turn degli ultimi 7 giorni, esegui la seguente query:

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;

Utilizzo dei token per modello e per agente (ultimi 7 giorni)

Per suddividere i conteggi delle chiamate ai modelli e l'utilizzo dei token (inclusi i token di ragionamento) per data, data_agent_id e model_version negli ultimi 7 giorni, esegui una query su v_ca_events per le righe 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;

Errori per agente, con riepiloghi delle cause principali assistiti dall'AI

Per controllare i conteggi degli errori per agente e giorno negli ultimi 7 giorni, esegui la seguente query:

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;

Per riassumere la causa principale degli errori recenti, utilizza la funzione AI.GENERATE.

Attenzione: questa query invia messaggi di errore dai log a un modello Gemini su Gemini Enterprise Agent Platform nel tuo progetto. Controlla prima le norme sulla governance dei dati. La persona che esegue la query deve avere il ruolo Utente della piattaforma agente (roles/aiplatform.user) oppure puoi utilizzare una connessione BigQuery che lo abbia.

La CTE limita le righe prima che venga chiamato il modello, in modo che il costo rimanga prevedibile:

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;

Rimuovi le righe duplicate

Questo passaggio è facoltativo. Una scrittura ritentata può occasionalmente produrre lo stesso evento due volte. La seguente query mostra come selezionare righe univoche per event_id per le ultime 24 ore:

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;

Risolvi i problemi relativi al logging delle conversazioni

Se la tabella dei log delle conversazioni è mancante, vuota o non contiene turni di utenti specifici o se si verificano errori di convalida della destinazione, consulta Risolvi i problemi relativi al logging delle conversazioni.

Passaggi successivi

• Esamina lo schema del log delle conversazioni.

• Leggi la panoramica del logging delle conversazioni.

• Crea e configura gli agenti di dati.