將 Model Armor 記錄與 Gemini Enterprise 記錄相互關聯

本文說明如何在 Cloud Logging 中，將 Model Armor 淨化記錄與 Gemini Enterprise 平台記錄和追蹤範圍相互關聯。本文說明用戶端到代理程式 (輸入) 和代理程式到任何位置 (輸出) 流量的關聯機制，列出產生追蹤記錄的必要條件，並提供逐步說明和程式碼範例，協助您在記錄處理管道中加入這些記錄項目。

調查 Model Armor 追蹤範圍或記錄項目時，您可能需要在 Cloud Logging 中找出對應的記錄，才能取得要求的完整脈絡。例如：

• 如果從追蹤記錄時距開始，您可能需要判斷使用者身分或檢查詳細的清除發現項目。
• 如果從 Model Armor 淨化記錄項目開始，您可能需要將其與使用者身分或追蹤資訊建立關聯。

記錄關聯性的運作方式

在 Gemini Enterprise 中，Model Armor 可在下列通訊點篩選提示詞和回覆：

• 用戶端到 Agent (傳入) 流量：使用者向 Gemini Enterprise 助理提交提示時，Gemini Enterprise 會直接呼叫 Model Armor API。產生的 Model Armor 平台記錄 (SanitizeOperation) 不會直接包含 OpenTelemetry trace 或 spanId 欄位。如要將這些記錄與使用者身分和追蹤範圍建立關聯，請在安全資訊與事件管理 (SIEM) 或記錄處理管道中，使用工作階段符記執行記錄聯結。

• 代理到任何位置 (輸出) 的流量：當代理呼叫外部工具、Model Context Protocol (MCP) 伺服器或外部大型語言模型 (LLM) 時，流量會透過 Agent Gateway 和 Secure Web Proxy 傳輸。如果是輸出呼叫，啟用 OpenTelemetry 檢測功能後，Model Armor SanitizeOperation 記錄會直接包含 trace 和 spanId 欄位。您可以直接篩選記錄，並在 Cloud Trace 或 Agent Registry 中查看追蹤記錄範圍。

關聯機制摘要

Flow	路徑和轉送	Model Armor 記錄中的追蹤記錄	關聯方法
用戶端至代理 (輸入)	Gemini Enterprise 直接呼叫 Model Armor API	系統不會填入 trace 和 spanId。	使用 client_correlation_id 和 assistToken 的工作階段符記記錄加入情形
代理至任意目的地 (輸出)	透過 Agent Gateway 和 Secure Web Proxy 轉送	系統會填入 trace 和 spanId。	直接比對 trace ID 和檢查追蹤記錄範圍

事前準備

開始將 Model Armor 記錄與 Gemini Enterprise 記錄相互關聯前，請先按照下列步驟操作：

1. 在 Gemini Enterprise 中啟用 Model Armor。
2. 如要在 Gemini Enterprise 和 Model Armor 記錄中產生追蹤記錄內容並查看追蹤記錄詳細資料，請在可觀測性設定中開啟「啟用 OpenTelemetry 追蹤記錄和記錄的檢測功能」，並視需要開啟「啟用提示詞輸入內容和回覆輸出內容的記錄功能」。如需操作說明，請參閱「開啟觀測能力設定」。
3. 如要篩選輸出流量，請在 Agent Gateway 上設定 Model Armor。

必要的角色

如要取得查看及關聯記錄和追蹤範圍所需的權限，請要求管理員在專案中授予您下列 IAM 角色：

• Cloud Trace 使用者 (roles/cloudtrace.user)
• 記錄檢視器 (roles/logging.viewer)

如要進一步瞭解如何授予角色，請參閱「管理專案、資料夾和組織的存取權」。

您或許也能透過自訂角色或其他預先定義的角色，取得必要權限。

如要瞭解您可能需要的其他角色，請參閱「Trace 存取控管」和「Cloud Logging 存取控管」。

關聯用戶端至代理 (輸入) 記錄

單一 StreamAssist 互動會在 Cloud Logging 中產生三種不同的記錄項目：

• Model Armor 淨化記錄 (SanitizeOperation)：
  • 受監控的資源： modelarmor.googleapis.com/SanitizeOperation
  • 屬性：包含詳細的清除判定和安全發現 (例如個人識別資訊的編輯、負責任的 AI 技術篩選器比對，或提示注入偵測)，但不包含追蹤內容或使用者身分。
  • 關聯鍵： labels."modelarmor.googleapis.com/client_correlation_id"
• Gemini Enterprise StreamAssist 記錄 (consumed_api)：
  • 受監控的資源： consumed_api
  • 屬性：包含使用者身分 (userIamPrincipal)、追蹤詳細資料 (trace 和 spanId)，以及工作階段符記 (response.assistToken)。
  • 關聯鍵： jsonPayload.response.assistToken
• Gemini Enterprise ModelArmorAudit Log (Agent)：
  • 受控資源： discoveryengine.googleapis.com/Agent，其中 jsonPayload.logMetadata.methodName 為 ModelArmorAudit。
  • 屬性：反映高層級的清除結果，並包含追蹤內容 (trace 和 spanId)，但不包含詳細的調查結果或關聯 ID。
  • 關聯鍵： trace

關聯性彙整索引鍵

Model Armor 淨化記錄包含以管道分隔結構的 client_correlation_id 標籤。這個標籤的第三個區段是 base64url 編碼的工作階段符記，與 consumed_api 記錄中為 StreamAssist 記錄的 assistToken 欄位相符。

client_correlation_id 標籤的格式如下：

AS|ASSISTANT_RESOURCE|SESSION_TOKEN

關聯 ID 包含下列值：

• ASSISTANT_RESOURCE：Gemini Enterprise Assistant 資源的完整資源名稱，格式如下：

  projects/PROJECT/locations/LOCATION/collections/COLLECTION/engines/ENGINE/assistants/ASSISTANT

• SESSION_TOKEN：不重複的工作階段符記，與 consumed_api 記錄中的 assistToken 相符，且已將 base64url 填補內容正規化。

比對邏輯

如要將 Model Armor 清理記錄項目與 Gemini Enterprise StreamAssist 記錄建立關聯，請在記錄處理管道中實作下列比對邏輯：

1. 從 Model Armor 項目中擷取工作階段符記：

   1. 在 Model Armor 項目中找到 labels 物件。
   2. 擷取 modelarmor.googleapis.com/client_correlation_id 標籤的值。
   3. 使用直立線字元 (|) 分隔這個標籤的值。
   4. 擷取第三個區段，代表以 base64url 編碼的工作階段權杖。

2. 從 StreamAssist 項目中擷取 assistToken 值：針對每個候選 StreamAssist consumed_api 記錄項目，請按照下列步驟操作：

   1. 找出 jsonPayload 物件。
   2. 從 response.assistToken 欄位擷取權杖值。

3. 正規化及比較權杖：如要比較權杖，請正規化兩個權杖字串：

   1. 將所有連字號 (-) 替換成加號 (+)。
   2. 將所有底線 (_) 換成正斜線 (/)。
   3. 移除結尾的等號 (=)。
   4. 如果正規化權杖相符，請將記錄項目建立關聯。

4. 擷取相關資料：如果找到相符項目，請從相符項目中擷取下列欄位：

   • 使用者 IAM 身分：StreamAssist 項目中的 userIamPrincipal 欄位
   • 追蹤記錄 ID：StreamAssist 項目中的 trace 欄位
   • 範圍 ID：StreamAssist 項目中的 spanId 欄位
   • 清理結果：Model Armor 項目中 sanitizationVerdict 欄位下方的 jsonPayload.sanitizationResult

Python 關聯範例

下列 Python 指令碼示範如何查詢 Cloud Logging 中的 Model Armor 和 Gemini Enterprise 記錄，執行權杖正規化和比對，並輸出相關記錄：

#!/usr/bin/env python3
from datetime import datetime, timedelta, timezone
from google.cloud import logging

# Google Cloud project ID
PROJECT_ID = "YOUR_PROJECT_ID"


def correlate_logs(ma_entry, de_consumed_entries):
  """Correlates a Model Armor log entry with StreamAssist logs."""
  # 1. Extract client_correlation_id from Model Armor log labels
  labels = ma_entry.get("labels", {})
  client_corr_id = labels.get(
      "modelarmor.googleapis.com/client_correlation_id", ""
  )
  if not client_corr_id:
    return None

  # 2. Extract session token (3rd pipe-delimited segment)
  parts = client_corr_id.split("|")
  if len(parts) < 3:
    return None
  ma_token = parts[2]

  # 3. Normalize base64url padding for comparison
  ma_token_normalized = ma_token.replace("-", "+").replace("_", "/").rstrip("=")

  # 4. Search for matching assistToken in StreamAssist logs
  for de in de_consumed_entries:
    payload = de.get("jsonPayload", {})
    de_token = payload.get("response", {}).get("assistToken", "")
    de_token_normalized = (
        de_token.replace("-", "+").replace("_", "/").rstrip("=")
    )

    if ma_token_normalized == de_token_normalized:
      return {
          "user": payload.get("userIamPrincipal"),
          "trace": de.get("trace"),
          "span_id": de.get("spanId"),
          "verdict": (
              ma_entry.get("jsonPayload", {})
              .get("sanitizationResult", {})
              .get("sanitizationVerdict")
          ),
      }
  return None


def main():
  # Initialize Google Cloud Logging Client
  print(f"Connecting to Google Cloud Logging (Project: {PROJECT_ID})...")
  client = logging.Client(project=PROJECT_ID)

  # Calculate ISO timestamp for 1 hour ago
  one_hour_ago = (
      datetime.now(timezone.utc) - timedelta(hours=1)
  ).strftime("%Y-%m-%dT%H:%M:%SZ")
  print(f"Filtering logs starting from: {one_hour_ago}")

  # Build log query filters
  ma_filter = f"""
    resource.type="modelarmor.googleapis.com/SanitizeOperation"
    AND timestamp >= "{one_hour_ago}"
    """

  de_filter = f"""
    resource.type="consumed_api"
    AND jsonPayload.response.assistToken:*
    AND timestamp >= "{one_hour_ago}"
    """

  # Fetch Model Armor log entries
  print("Fetching Model Armor log entries...")
  ma_entries = [
      entry.to_api_repr()
      for entry in client.list_entries(filter_=ma_filter, max_results=100)
  ]
  print(f"Found {len(ma_entries)} Model Armor entries.")

  # Fetch Gemini Enterprise log entries
  print("Fetching Gemini Enterprise StreamAssist log entries...")
  de_entries = [
      entry.to_api_repr()
      for entry in client.list_entries(filter_=de_filter, max_results=500)
  ]
  print(f"Found {len(de_entries)} Gemini Enterprise entries.")

  # Perform Correlation
  print("\n================ Correlating Logs ================")
  correlated_results = []
  for ma in ma_entries:
    match = correlate_logs(ma, de_entries)
    if match:
      correlated_results.append(match)
      print(f"  User IAM Principal  : {match['user']}")
      print(f"  Sanitization Verdict: {match['verdict']}")
      print(f"  Trace ID            : {match['trace']}")
      print(f"  Span ID             : {match['span_id']}")
      print("-" * 50)

  print(f"\nDone. Total Correlated Records: {len(correlated_results)}")


if __name__ == "__main__":
  main()

將代理至任意目的地 (輸出) 記錄與追蹤記錄範圍建立關聯

當代理執行受 Agent Gateway 和 Model Armor 保護的工具呼叫 (例如與 MCP 伺服器或外部 API 互動) 時，該要求會屬於「代理對任何位置」流量。

在應用程式上啟用 OpenTelemetry 檢測功能後，產生的 SanitizeOperation 記錄檔項目會自動包含 trace 和 spanId 欄位。

在 Cloud Logging 中篩選輸出記錄

如要找出與 Cloud Logging 中特定追蹤記錄相關聯的所有 Model Armor 消毒記錄，請使用下列查詢篩選器：

resource.type="modelarmor.googleapis.com/SanitizeOperation"
trace="TRACE_ID"

將 TRACE_ID 替換為代理程式互動中的追蹤 ID。

詳情請參閱「查看及分析記錄項目」。

查看追蹤記錄時距

在「追蹤」或「Agent Registry」中，您可以查看代理程式互動的執行圖表和時間軸。Model Armor 會產生下列範圍：

• 父項範圍：apply_guardrail "Google Cloud Model Armor"
• 子跨度：Request Path 和 Response Path

每個時距都包含政策 ID、安全性決策和相符的篩選器違規事項等屬性。詳情請參閱「查看 Model Armor 追蹤跨度」。

後續步驟

• 在 Gemini Enterprise 中啟用 Model Armor
• 查看 Model Armor 追蹤跨度
• 在 Model Armor 中設定記錄
• 在 Gemini Enterprise 中管理可觀測性設定