대화 로그 분석

이 문서에서는 GoogleSQL 레시피, 대화형 Colab 노트북, BigQuery 에이전트 분석 Python SDK를 사용하여 BigQuery에서 대화형 분석 대화 로그를 쿼리, 평가, 문제 해결하는 방법을 보여줍니다. 대상 테이블 열 및 JSON 구조에 대한 자세한 내용은 대화 로그 스키마를 참조하세요.

Colab 노트북 및 Python SDK 사용

고객 분석, FinOps, 평가 시작 노트북(Colaboratory에서 열기 | GitHub에서 보기)을 사용하여 bigquery-agent-analytics Python SDK(v0.5.4 이상)와 함께 이 문서의 SQL 레시피를 실행할 수 있습니다.

대화 로그 테이블과 함께 Python SDK를 사용하려면 먼저 v_bqca_sdk_events 호환성 뷰를 만듭니다. AGENT_RESPONSE 행에서 대화 분석은 JSON 객체를 content.response에 저장하는 반면 SDK의 LLMAsJudge 평가자는 JSON_VALUE(content, '$.response')가 스칼라 문자열을 반환할 것으로 예상합니다. 이 뷰는 $.text_summary 및 $.response를 일반 텍스트 답변으로 설정하고 원래 객체는 $.raw_response에 유지합니다.

다음 SQL 및 Python 스니펫에서 다음 자리표시자를 바꿉니다.

• PROJECT_ID: 데이터 에이전트 및 로그 데이터 세트가 포함된 프로젝트의 ID입니다.

• LOG_DATASET: 원시 대화 로그 테이블을 저장하는 BigQuery 데이터 세트의 ID입니다.

• LOG_TABLE: 대화 로그 테이블의 ID입니다(예: agent_events).

• ANALYTICS_DATASET: 분석 뷰 및 테이블 함수를 만드는 동일한 위치에 있는 BigQuery 데이터 세트의 ID입니다.

• LOCATION: 분석 데이터 세트의 BigQuery 위치입니다(예: US).

ANALYTICS_DATASET에서 뷰 및 테이블 함수를 만들려면 ANALYTICS_DATASET에 대한 쓰기 액세스 권한(예: BigQuery 데이터 편집자(roles/bigquery.dataEditor) 역할)과 LOG_DATASET에 대한 BigQuery 데이터 뷰어(roles/bigquery.dataViewer) 역할이 필요합니다. 뷰만 쿼리하는 분석가에게는 ANALYTICS_DATASET에 대한 BigQuery 데이터 뷰어 역할과 쿼리를 실행하는 프로젝트에 대한 BigQuery 작업 사용자 (roles/bigquery.jobUser) 역할이 필요합니다.

다음 쿼리는 다음 단계에서 테이블 스키마 및 이벤트 범위를 검증하는 데 사용되는 v_bqca_sdk_events 뷰를 만듭니다.

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;

호환성 뷰를 만든 후 SDK Client를 초기화합니다. 필요한 경우 client.doctor()를 실행하여 테이블 스키마와 이벤트 범위를 검증할 수 있습니다.

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)

client.doctor() 함수는 테이블 스키마를 검증하고 이벤트 범위를 표준 에이전트 개발 키트 (ADK) 이벤트 유형과 비교하는 딕셔너리를 반환합니다. v_bqca_sdk_events 호환성 뷰는 허용 목록에 추가된 대화 분석 이벤트 유형만 노출하므로 AGENT_COMPLETED, AGENT_STARTING, LLM_REQUEST, TOOL_COMPLETED 또는 TOOL_STARTING 이벤트 누락에 대한 경고가 표시될 수 있습니다. 오류 또는 누락된 열 경고는 Client 매개변수가 뷰가 생성된 위치와 일치하지 않거나 데이터 세트에 대한 읽기 액세스 권한이 없음을 의미하며, 빈 event_coverage 결과는 뷰에 지난 7일 동안의 대화가 없음을 의미합니다.

SDK를 사용하여 에이전트 품질을 평가하고, trace를 재구성하고, 드리프트를 모니터링하는 방법에 대한 자세한 내용은 BigQuery 에이전트 분석 SDK 사용 및 이 섹션에 연결된 대화형 Colab 노트북을 참조하세요.

SQL 레시피

다음 GoogleSQL 쿼리에서 Colab 노트북 및 Python SDK 사용에 설명된 것과 동일한 PROJECT_ID, LOG_DATASET, LOG_TABLE, ANALYTICS_DATASET 자리표시자를 사용합니다.

에이전트 간 이벤트 뷰 만들기(v_ca_events) 및 대화 차례 테이블 함수 만들기 (ca_turns)로 시작합니다. 다음 쿼리 대부분은 ca_turns 테이블 함수 또는 v_ca_events 뷰를 사용합니다. AI.GENERATE 오류 요약 쿼리와 중복 삭제 쿼리는 로그 테이블을 직접 읽습니다.

교차 에이전트 이벤트 뷰 만들기 (v_ca_events)

이 뷰는 이벤트당 하나의 행을 반환하며, 에이전트 ID, 대화 ID, 질문, 답변 텍스트, 인사이트, 토큰, 지연 시간이 일반 열로 표시됩니다. timestamp의 필터는 로그 테이블로 전달되므로 파티션 프루닝이 계속 작동합니다. 뷰는 중복 이벤트를 삭제하지 않습니다. 쓰기 재시도는 드물지만 event_id로 중복 이벤트를 검사하거나 필터링하려면 중복 행 삭제를 참조하세요.

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

대화 차례 테이블 함수 (ca_turns)를 만듭니다. 질문 및 답변당 한 행입니다.

이 테이블 함수는 기간을 사용하고 차례당 하나의 행을 반환합니다. 로그 테이블에 시간 필터를 적용하므로 요청한 일수에 대해서만 비용을 지불합니다.

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;

한 사용자의 모든 질문 및 답변 (지난 30일)

target_user를 사용자의 이메일 주소 또는 이메일 주소가 없는 호출자의 내부 키로 설정합니다.

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;

검토를 위해 모든 상담사의 샘플 질문 및 답변

이 쿼리는 지난 24시간 동안의 대화 중 약 10% 를 반환합니다. 0.10를 변경하여 검토 비율을 설정합니다.

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;

요일별 채택

user_id가 NULL인 경우(예: 일부 제휴 ID의 경우) internal_user_key 필드가 attributes.session_metadata.user_id로 대체되므로 일일 사용자 수가 사용자당 일관되게 유지됩니다.

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;

사용자별 채택: 사용자당 에이전트, 대화, 질문

지난 30일 동안 사용자별 고유 에이전트, 대화, 질문을 집계하려면 다음 쿼리를 실행합니다.

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;

에이전트별 도입 및 상태

프로젝트의 모든 데이터 에이전트에서 사용자, 대화, 질문 수, 평균 지연 시간, 오류율, 총 토큰을 비교하려면 다음 쿼리를 실행합니다.

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;

대화 요약

지난 7일 동안의 멀티턴 대화를 롤업하려면 다음 쿼리를 실행합니다.

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;

모델 및 에이전트별 토큰 사용량 (지난 7일)

지난 7일 동안의 모델 호출 수와 토큰 사용량 (사고 토큰 포함)을 날짜, data_agent_id, model_version별로 분류하려면 v_ca_events에서 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;

AI 지원 근본 원인 요약이 포함된 에이전트별 오류

지난 7일 동안의 에이전트 및 일별 오류 수를 확인하려면 다음 쿼리를 실행합니다.

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;

최근 오류의 근본 원인을 요약하려면 AI.GENERATE 함수를 사용합니다.

주의: 이 쿼리는 로그의 오류 메시지를 프로젝트의 Gemini Enterprise Agent Platform에 있는 Gemini 모델로 전송합니다. 먼저 데이터 거버넌스 정책을 확인하세요. 쿼리를 실행하는 사용자에게 에이전트 플랫폼 사용자 (roles/aiplatform.user) 역할이 필요하거나 이 역할이 있는 BigQuery 연결을 사용할 수 있습니다.

CTE는 모델이 호출되기 전에 행을 제한하므로 비용을 예측할 수 있습니다.

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;

중복 행 삭제

이 단계는 선택사항입니다. 재시도된 쓰기는 동일한 이벤트를 두 번 생성할 수 있습니다. 다음 쿼리는 지난 24시간 동안 event_id당 고유 행을 선택하는 방법을 보여줍니다.

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;

대화 로깅 문제 해결

대화 로그 테이블이 누락되었거나 비어 있거나 특정 사용자의 차례가 누락되었거나 대상 검증 오류가 발생하면 대화 로깅 문제 해결을 참조하세요.

다음 단계

• 대화 로그 스키마를 검토합니다.

• 대화 로깅 개요 읽기

• 데이터 에이전트 만들기 및 구성