使用 BigQuery 代理數據分析

BigQuery 代理分析是一項開放原始碼解決方案，可將原始代理互動 (例如要求、回覆、工具呼叫和錯誤) 直接串流至 BigQuery，大規模擷取、分析及視覺化多模態代理互動資料。這項解決方案可讓您執行 AI 輔助評估、最佳化代理提示，以及擷取長期記憶，提升日後的互動品質。 Agent Development Kit (ADK) 和 LangGraph (預先發布版) 支援 BigQuery 代理分析。

架構

BigQuery 代理數據分析會使用 BigQuery Storage Write API (gRPC)，將代理活動資料串流至 BigQuery，提供高處理量、低延遲的記錄串流，不會阻礙代理執行。

資料流程包含下列階段：

  1. 擷取。使用 Agent Development Kit (ADK) 中的外掛程式，或 LangGraph 中的回呼攔截互動事件，擷取代理事件。
  2. 串流：互動事件會透過 Storage Write API (gRPC) 傳送至 BigQuery。如果沒有標準化結構定義，代理程式會自動建立。
  3. Consume。分析及評估記錄的代理資料。您可以使用 SQL 查詢原始資料、在自訂資訊主頁追蹤指標，或使用 BigQuery 代理分析 SDK 重建及評估複雜的多輪代理執行追蹤記錄。

圖表：代理活動資料從自動化調度管理架構流向 BigQuery 進行分析。

代理數據分析的優點

  • 只要一行程式碼，就能啟用全方位的記錄功能，並自動管理結構定義。
  • 使用物件資料表記錄及分析多模態資料，例如文字、圖片、影片和音訊。
  • 在預先定義的健全結構定義中，追蹤詞元消耗量和延遲時間等作業指標。
  • 使用 BigQuery 生成式 AI 函式和向量搜尋，找出最佳化商機。
  • 透過精細的存取控管、資料遮蓋和加密功能，確保代理記錄安全無虞。

擷取代理記錄資料的方式

如要將代理的互動遙測資料 (要求、回覆、工具呼叫和錯誤記錄) 原生擷取至 BigQuery，可以透過多種方式記錄事件資料：

  • 自動化調度管理架構外掛程式：使用代理自動化調度管理工具包提供的標準記錄外掛程式。舉例來說，Agent Development Kit (ADK) 中的 BigQueryAgentAnalyticsPlugin 會連結至代理執行工具，自動攔截、序列化及串流事件。
  • 架構回呼處理常式：在熱門代理環境中整合標準回呼。例如，您可以使用 LangGraph 和 LangChain 內建的 BigQuery 處理常式，攔截及轉送追蹤記錄。
  • 直接 API 擷取：如果是自訂或專屬架構，請使用Cloud de Confiance by S3NS 用戶端程式庫，透過 Storage Write API (gRPC) 將結構化事件直接串流至事件資料表。
  • 對話式數據分析資料代理：在 BigQuery Studio 或透過 Conversational Analytics API 設定 BigQuery 資料代理時，啟用內建的對話記錄功能，即可自動將對話事件串流至目的地 BigQuery 資料表，並分析對話記錄。

無論採用何種方法，所有記錄選項都會使用低延遲、高處理量的 BigQuery Storage Write API (gRPC)。這個 API 提供強大的串流端點，可在提交資料列前，先在記憶體中以非同步方式緩衝及序列化資料列 (使用 PyArrow 引擎)，確保觀測能力管道任務不會阻礙面向使用者的代理執行。

分析代理記錄資料的方法

如要瞭解及最佳化代理效能，請透過下列方式分析及評估互動記錄：

  • 直接 SQL 查詢：在 BigQuery 執行自訂查詢，計算權杖用量和執行延遲時間等指標。您也可以使用 AI.GENERATE 自動分析錯誤的根本原因，或與業務資料表進行彙整，評估業務影響。
  • 互動式資訊主頁：將數據分析等視覺化工具連結至預先建構或自訂的 BigQuery 檢視畫面，追蹤代理健康狀態、錯誤率和使用趨勢。Looker 客戶也可以使用 Looker Marketplace 中預先建構的 BigQuery 代理分析資訊主頁範本 (Looker 模塊)。
  • Jupyter 筆記本：在互動式環境中，使用 Python 程式庫、pandas 或 BigFrames 探索及實驗記錄資料。
  • Python SDK：直接在應用程式程式碼或自動評估管道中，透過程式輔助查詢、重建及稽核代理執行追蹤記錄。

使用代理記錄資料的範例

以下是在 BigQuery 中使用代理記錄資料的常見用途和範例。

觀測能力和營運指標

  • 查詢資料，依代理流程細分費用，判斷特定代理 (例如修正代理) 耗用的詞元數量，是否與其對最終回覆的貢獻不成比例。
  • 使用 BigQuery 對話式數據分析代理，透過 AI.GENERATE 函式執行查詢，進行 AI 輔助根本原因分析。例如：「分析這份對話記錄，並說明失敗的根本原因。」

代理評估和品質分析

  • 使用 AI.SCORE 函式，為對話評分，並評估服務專員在一段時間內的評分。
  • 使用 SQL 查詢和向量搜尋，找出代理無法協助使用者的對話叢集，然後與使用者的原始意圖進行比較。這有助於找出代理工具或知識庫的不足之處。

業務洞察和情境化

如要為代理資料提供背景資訊，請將 agent_events 資料表與其他業務資料表彙整。舉例來說，您可以比較與 AI 代理互動的顧客，以及使用搜尋列的顧客，兩者的平均訂單價值 (AOV)。

如需更多範例，請參閱「進階分析查詢」。

使用 Jupyter 筆記本處理代理記錄

使用這個範例 Colab Jupyter 筆記本，以互動方式查詢、視覺化呈現及評估代理記錄。

使用 BigQuery 代理 Analytics SDK

BigQuery 代理數據分析 SDK 是開放原始碼 Python 程式庫，提供長期代理觀測的消耗和評估層。資訊主頁和筆記本非常適合臨時探索，而 SDK 則提供系統化、大規模分析及稽核代理行為的方法。

SDK 的用途

您可以使用 BigQuery Agent Analytics SDK 執行下列工作。如需記錄分析的詳細範例，請參閱 SDK 的 GitHub 存放區。

  • 追蹤重建：將多型事件記錄重建為因果事件鏈，以便在多輪對話中偵錯工作階段，包括巢狀工具和 LLM 呼叫。
  • 確定性和語意評估：根據規則式條件 (例如延遲時間、對話輪數和錯誤率) 以及語意條件 (例如正確性、情緒和幻覺)，評估代理品質。
  • 軌跡比對：比較實際的代理執行路徑與預期的黃金軌跡，驗證步驟效率和工具使用順序是否正確。
  • 行為監控和漂移偵測：對非確定性代理輸出內容進行統計分析、監控使用者要求分布情形，以及偵測正式環境迴歸或語意漂移。
  • 長程代理記憶體：為代理提供跨工作階段脈絡、使用者個人資料語意檢索，以及原生儲存於 BigQuery 的權杖預算感知情節記憶。

記錄及分析代理活動

將 SDK 整合至代理工作流程，通常包含下列步驟：

  1. 記錄互動：在代理協調架構中，附加記錄器外掛程式 (例如 ADK 中的 BigQueryAgentAnalyticsPlugin) 或回呼處理常式。使用者與代理互動時，系統會使用高處理量的 Storage Write API (gRPC)，將記錄非同步串流至 BigQuery。

  2. 初始化用戶端：透過 Python SDK 連線至記錄資料集：

    from google.cloud import bigquery
from bigquery_agent_analytics import Client

client = Client(
    project_id="YOUR_PROJECT_ID",
    dataset_id="YOUR_DATASET_ID",
    table_id="agent_events",
)

    程式碼片段使用下列元件：

    • client：父項 Client 執行個體，會以程式輔助方式轉送查詢，並管理有效的資料庫連線。
    • project_id：目標資料集所在的 Cloud de Confiance by S3NS 專案 ID。
    • dataset_id：儲存記錄的 BigQuery 資料集名稱。
    • table_id：儲存遙測事件的特定資料表 (預設為 agent_events)。

  3. 重建工作階段追蹤記錄：擷取特定對話工作階段，以視覺化方式呈現及查看確切的事件序列：

    trace = client.get_session_trace(
    session_id="YOUR_SESSION_ID"
)
trace.render()

    程式碼片段使用下列元件：

    • trace：經補充的 Trace 物件，內含所有使用者和代理動作的重建因果連結階層式 DAG 跨距樹狀結構。
    • YOUR_SESSION_ID：要檢查的工作階段專屬 ID。

  4. 執行自動評估：透過程式輔助方式為工作階段軌跡評分，或根據黃金測試套件檢查迴歸：

    from bigquery_agent_analytics import (
    LLMAsJudge,
    TraceFilter,
)

# Run semantic evaluations across target sessions
evaluator = LLMAsJudge.correctness()
filters = TraceFilter(session_ids=["session_1", "session_2"])

report = client.evaluate(evaluator, filters=filters)
print(f"Evaluation Pass Rate: {report.pass_rate:.2%}")

    程式碼片段使用下列元件：

    • evaluator：用於評估工作階段的語意 (LLMAsJudge) 評估器 (您也可以執行規則式 SystemEvaluator 檢查，或將評估器合併為加權 GraderPipeline)。
    • filters：TraceFilter 執行個體，指定要在 BigQuery 中執行批次評估的 session_ids 清單 (或時間範圍和代理篩選器)。
    • report：產生的 EvaluationReport 物件，包含原始工作階段成績、指標摘要和自動評分者意見回饋。

將 BigQuery 代理數據分析功能整合至工作流程

如要將 BigQuery 代理數據分析整合至工作流程，請參閱所用架構的說明文件：

後續步驟