使用 BigQuery 代理程式數據分析

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

架構

BigQuery 代理程式分析會使用 BigQuery Storage Write API，將代理程式活動記錄串流至 BigQuery。這項 API 提供高處理量、低延遲的記錄串流功能，不會阻礙代理程式執行作業。

資料流程包含下列階段：

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

這張圖表顯示代理程式活動記錄從協調架構流入 BigQuery，以供分析。

代理程式數據分析的優點

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

擷取代理程式記錄資料的方法

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

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

無論使用哪種方法，所有記錄選項都會使用低延遲、高處理量的 BigQuery Storage Write API。這個 API 提供強大的串流端點，可使用 PyArrow 引擎在記憶體中非同步緩衝及序列化資料列，再提交資料列，確保可觀測性管道工作不會封鎖面向使用者的代理執行回合。

分析代理程式記錄檔資料的方式

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

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

使用代理程式記錄資料的範例

以下是常見用途，以及在 BigQuery 中處理代理程式記錄資料的範例。

可觀測性和運作指標

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

代理程式評估和品質分析

  • 使用 AI.SCORE 函式，為對話排序並評估服務專員的排名變化。
  • 使用 Vector Search 執行 SQL 查詢，找出代理程式無法協助使用者的對話叢集，然後與使用者的原始意圖進行比較。這有助於找出客服人員工具或知識庫的不足之處。

業務洞察和情境化

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

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

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

使用這個 Colab Jupyter 筆記本範例，以互動方式查詢、顯示及評估虛擬服務專員記錄。

使用 BigQuery 代理程式分析 SDK

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

SDK 的用途

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

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

記錄及分析代理程式活動

將 SDK 整合至代理程式工作流程通常需要執行下列步驟：

  1. 記錄互動：在代理程式協調架構中附加記錄器外掛程式 (例如 ADK 中的 BigQueryAgentAnalyticsPlugin) 或回呼處理常式。使用者與代理程式互動時，系統會使用高處理量的 Storage Write API，以非同步方式將記錄串流至 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_trace(
    session_id="YOUR_SESSION_ID"
)
trace.render()

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

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

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

    from bigquery_agent_analytics.evaluators import CodeEvaluator, LLMAsJudge
from bigquery_agent_analytics.grader_pipeline import GraderPipeline

# Create a grader pipeline with deterministic and semantic metrics
evaluator = GraderPipeline(
    graders=[
        CodeEvaluator.latency(threshold_ms=5000),
        LLMAsJudge.correctness(),
    ]
)
report = client.evaluate(evaluator, session_ids=["session_1", "session_2"])
print(f"Evaluation Pass Rate: {report.pass_rate:.2%}")

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

    • evaluator：編譯的結構化 GraderPipeline 邏輯，由異質規則和語意評估指標組成。
    • session_ids：要在 BigQuery 中執行批次評估的工作階段 ID 字串清單。
    • report：產生的 EvaluationReport 物件，內含原始課程成績、指標摘要、測驗統計資料和自動評分員意見回饋。

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

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

後續步驟