ADK 的 BigQuery 智能体分析插件¶

Supported in ADKPython v1.21.0

版本要求

使用 ADK 的 最新版本（版本 1.21.0 或更高版本）以充分利用本规范中描述的功能。

BigQuery 智能体分析插件为智能体开发工具包 (ADK) 提供了健壮的智能体行为分析解决方案。利用 ADK 插件架构和 BigQuery 存储写入 API (Storage Write API)，它能直接将关键操作事件捕获并记录到 Google BigQuery 表中，为你提供用于调试、实时监控和全面离线性能评估的高级功能。

请使用 ADK Python 1.26.0 或更高版本，以充分利用本文档中描述的功能，包括自动模式升级 (auto-schema-upgrade)、工具来源追踪 (tool provenance tracking) 和人机回环 (HITL) 事件追踪。

1.26.0 版本新增了自动架构升级（安全地向现有表添加新列）、工具来源追踪（LOCAL、MCP、SUB_AGENT、A2A、TRANSFER_AGENT、TRANSFER_A2A）以及用于人机回环交互的 HITL 事件追踪。1.27.0 版本新增了自动创建视图（生成扁平化、便于查询的事件视图）。

BigQuery Storage Write API

智能体工作流调试与分析：将广泛的插件生命周期事件（LLM 调用、工具使用）和智能体产生的事件（用户输入、模型响应）捕获到定义良好的模式中。
高容量分析与调试：使用 Storage Write API 异步执行日志记录操作，实现高吞吐量和低延迟。
多模态分析：记录并分析文本、图像和其他模态数据。大文件会被卸载到 GCS，使其可以通过对象表 (Object Tables) 被 BigQuery ML 访问。
分布式追踪：内置支持 OpenTelemetry 风格的追踪（trace_id、span_id），以可视化智能体执行流。
工具来源：追踪每个工具调用的来源（本地函数、MCP 服务、子智能体、A2A 远程智能体或移交智能体）。
人机回环 (HITL) 追踪：为凭证请求、确认提示和用户输入请求提供专用事件类型。
可查询的事件视图：自动创建扁平化的、按事件类型分类的 BigQuery 视图（例如 v_llm_request、v_tool_completed），通过展开 JSON 负载数据来简化下游分析。

记录的智能体事件数据根据 ADK 事件类型而有所不同。如果你想了解更多信息，请参见事件类型和负载。

先决条件¶

Google Cloud 项目：已启用 BigQuery API。
BigQuery 数据集：在使用插件前创建一个数据集来存储日志表。如果表不存在，插件会在数据集中自动创建必要的事件表。
Google Cloud Storage 存储桶（可选）：如果你计划记录多模态内容（图像、音频等），建议创建一个 GCS 存储桶来卸载大文件。
身份验证：
- 本地：运行 gcloud auth application-default login。
- 云端：确保你的服务账号具有所需的权限。

IAM 权限¶

为了让智能体正常工作，运行智能体的主体（例如，服务账号、用户账号）需要以下 Google Cloud 角色： - roles/bigquery.jobUser：在项目级别运行 BigQuery 查询。 - roles/bigquery.dataEditor：在表级别写入日志/事件数据。 - 如果使用 GCS 卸载：对目标存储桶具有 roles/storage.objectCreator 和 roles/storage.objectViewer 权限。

注意：Gemini 模型选择器 gemini-flash-latest

ADK 文档中的大多数代码示例使用 gemini-flash-latest 来选择最新可用的 Gemini Flash 版本。但是，如果你从区域端点（如 us-central1）访问 Gemini，此选择字符串可能不起作用。在这种情况下，请使用 Gemini 模型页面或 Google Cloud Gemini 模型列表中的特定模型版本字符串。

IAM 权限¶

你可以通过配置并将 BigQuery 智能体分析插件注册到你的 ADK 智能体应用程序对象来使用它。以下示例展示了使用此插件的智能体实现，包括 GCS 卸载：

# my_bq_agent/agent.py
import os
import google.auth
from google.adk.apps import App
from google.adk.plugins.bigquery_agent_analytics_plugin import BigQueryAgentAnalyticsPlugin, BigQueryLoggerConfig
from google.adk.agents import Agent
from google.adk.models.google_llm import Gemini
from google.adk.tools.bigquery import BigQueryToolset, BigQueryCredentialsConfig

# --- OpenTelemetry TracerProvider 设置 (可选) ---
# ADK 将 OpenTelemetry 作为核心依赖项包含在内。
# 配置 TracerProvider 可实现完整的分布式追踪
# (使用标准的 OTel 标识符填充 trace_id、span_id)。
# 如果未配置 TracerProvider，插件将回退到内部 UUID 
# 进行跨度关联，同时仍保留父子层次结构。
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
trace.set_tracer_provider(TracerProvider())

# --- 配置 ---
PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT", "your-gcp-project-id")
DATASET_ID = os.environ.get("BIG_QUERY_DATASET_ID", "your-big-query-dataset-id")
# GOOGLE_CLOUD_LOCATION must be a valid Agent Platform region (e.g., "us-central1").
# BQ_LOCATION is the BigQuery dataset location, which can be a multi-region
# like "US" or "EU", or a single region like "us-central1".
VERTEX_LOCATION = os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1")
BQ_LOCATION = os.environ.get("BQ_LOCATION", "US")
GCS_BUCKET = os.environ.get("GCS_BUCKET_NAME", "your-gcs-bucket-name") # 可选

if PROJECT_ID == "your-gcp-project-id":
    raise ValueError("请设置 GOOGLE_CLOUD_PROJECT 环境变量或更新代码中的项目 ID。")

# --- 关键：在 Gemini 实例化之前设置环境变量 ---
os.environ['GOOGLE_CLOUD_PROJECT'] = PROJECT_ID
os.environ['GOOGLE_CLOUD_LOCATION'] = VERTEX_LOCATION
os.environ['GOOGLE_GENAI_USE_VERTEXAI'] = 'True'

# --- 使用配置初始化插件 ---
bq_config = BigQueryLoggerConfig(
    enabled=True,
    gcs_bucket_name=GCS_BUCKET, # 为多模态内容启用 GCS 卸载
    log_multi_modal_content=True,
    max_content_length=500 * 1024, # 500 KB 限制用于内联文本
    batch_size=1, # 默认为 1 以实现低延迟，增加此值可实现高吞吐量
    shutdown_timeout=10.0
)

bq_logging_plugin = BigQueryAgentAnalyticsPlugin(
    project_id=PROJECT_ID,
    dataset_id=DATASET_ID,
    table_id="agent_events", # 默认表名为 agent_events
    config=bq_config,
    location=BQ_LOCATION
)

# --- 初始化工具和模型 ---
credentials, _ = google.auth.default(scopes=["https://www.googleapis.com/auth/cloud-platform"])
bigquery_toolset = BigQueryToolset(
    credentials_config=BigQueryCredentialsConfig(credentials=credentials)
)

llm = Gemini(model="gemini-flash-latest")

root_agent = Agent(
    model=llm,
    name='my_bq_agent',
    instruction="你是一个可以访问 BigQuery 工具的得力助手。",
    tools=[bigquery_toolset]
)

# --- 创建 App 对象 ---
app = App(
    name="my_bq_agent",
    root_agent=root_agent,
    plugins=[bq_logging_plugin],
)

运行并测试智能体¶

通过运行智能体并在聊天界面发出一些请求来测试插件，例如“告诉我你能做什么”或“在我的云项目中列出数据集”。这些操作会产生事件并记录在你的 Google Cloud 项目 BigQuery 实例中。一旦这些事件被处理，你就可以在 BigQuery 控制台使用以下查询查看数据：

SELECT timestamp, event_type, content
FROM `your-gcp-project-id.your-big-query-dataset-id.agent_events`
ORDER BY timestamp DESC
LIMIT 20;

使用插件部署到 Agent Runtime¶

你可以将带有 BigQuery 智能体分析插件的智能体部署到 Agent Runtime。本节将介绍使用 ADK CLI 进行部署的步骤，以及使用 Agent Platform SDK 以编程方式部署的替代方法。

版本要求

使用此插件部署到 Agent Runtime 需要 ADK Python 版本 1.24.0 或更高。早期版本存在一个问题，即在刷新待处理事件之前，插件异步日志写入器可能被无服务器运行时终止。从 1.24.0 开始，该插件在每次调用结束时执行同步刷新，以确保所有事件都被写入。

前置条件¶

在部署之前，确保你已经完成了常规的 Agent Runtime 设置，包括：

一个启用了 Agent Platform API 和 Cloud Resource Manager API 的 Google Cloud 项目。
目标项目中的 BigQuery 数据集（或具有正确权限的跨项目数据集）。
用于部署工件的 Cloud Storage 分阶段存储桶。
部署服务账号具有 IAM 权限中列出的 IAM 角色。
你的编码环境已通过 gcloud auth login 和 gcloud auth application-default login 完成身份验证。

步骤 1：定义智能体和插件¶

创建一个包含插件的 App 对象的智能体项目文件夹。对于带有插件的 Agent Runtime 部署，App 对象是必需的。

my_bq_agent/
├── __init__.py
├── agent.py
└── requirements.txt

my_bq_agent/__init__.py

from . import agent

my_bq_agent/agent.py

import os
import google.auth
from google.adk.agents import Agent
from google.adk.apps import App
from google.adk.models.google_llm import Gemini
from google.adk.plugins.bigquery_agent_analytics_plugin import (
    BigQueryAgentAnalyticsPlugin,
    BigQueryLoggerConfig,
)
from google.adk.tools.bigquery import BigQueryToolset, BigQueryCredentialsConfig

# --- Configuration ---
PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT", "your-gcp-project-id")
DATASET_ID = os.environ.get("BQ_DATASET", "agent_analytics")
# BQ_LOCATION is the BigQuery dataset location (multi-region "US"/"EU" or
# a single region like "us-central1"). This is separate from the Agent Platform
# region used by GOOGLE_CLOUD_LOCATION.
BQ_LOCATION = os.environ.get("BQ_LOCATION", "US")

os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "True"

# --- Plugin ---
bq_analytics_plugin = BigQueryAgentAnalyticsPlugin(
    project_id=PROJECT_ID,
    dataset_id=DATASET_ID,
    location=BQ_LOCATION,
    config=BigQueryLoggerConfig(
        batch_size=1,
        batch_flush_interval=0.5,
        log_session_metadata=True,
    ),
)

# --- Tools ---
credentials, _ = google.auth.default(
    scopes=["https://www.googleapis.com/auth/cloud-platform"]
)
bigquery_toolset = BigQueryToolset(
    credentials_config=BigQueryCredentialsConfig(credentials=credentials)
)

# --- 智能体 ---
root_agent = Agent(
    model=Gemini(model="gemini-flash-latest"),
    name="my_bq_agent",
    instruction="你是一个可以访问 BigQuery 工具的得力助手。",
    tools=[bigquery_toolset],
)

# --- App (required for Agent Runtime with plugins) ---
app = App(
    name="my_bq_agent",
    root_agent=root_agent,
    plugins=[bq_analytics_plugin],
)

my_bq_agent/requirements.txt

google-adk[bigquery]
google-cloud-bigquery-storage
pyarrow
opentelemetry-api
opentelemetry-sdk

步骤 2：使用 ADK CLI 部署¶

使用 adk deploy agent_engine 命令部署智能体。--adk_app 标志告诉 CLI 使用哪个 App 对象：

PROJECT_ID=your-gcp-project-id
LOCATION=us-central1

adk deploy agent_engine \
    --project=$PROJECT_ID \
    --region=$LOCATION \
    --staging_bucket=gs://your-staging-bucket \
    --display_name="My BQ Analytics Agent" \
    --adk_app=agent.app \
    my_bq_agent

--adk_app 标志

--adk_app 标志指定 App 对象的模块路径和变量名（格式为 module.variable）。在此示例中，agent.app 指的是 agent.py 中的 app 变量。这确保部署正确提取了插件配置。

成功部署后，你应该会看到类似如下的输出：

AgentEngine created. Resource name: projects/123456789/locations/us-central1/reasoningEngines/751619551677906944

请记下 Resource name，以便进行下一步操作。

步骤 3：测试已部署的智能体¶

部署后，你可以使用 Agent Platform SDK 查询智能体：

test_deployed_agent.py

import uuid
import vertexai

PROJECT_ID = "your-gcp-project-id"
LOCATION = "us-central1"
AGENT_ID = "751619551677906944"  # 来自部署输出

vertexai.init(project=PROJECT_ID, location=LOCATION)
client = vertexai.Client(project=PROJECT_ID, location=LOCATION)

agent = client.agent_engines.get(
    name=f"projects/{PROJECT_ID}/locations/{LOCATION}/reasoningEngines/{AGENT_ID}"
)

user_id = f"test_user_{uuid.uuid4().hex[:8]}"
for chunk in agent.stream_query(
    message="列出我的项目中的数据集", user_id=user_id
):
    print(chunk, end="", flush=True)

步骤 4：在 BigQuery 中验证事件¶

向已部署的智能体发送几次查询后，通过查询 BigQuery 表来验证事件是否正在被记录：

SELECT timestamp, event_type, agent, content
FROM `your-gcp-project-id.agent_analytics.agent_events`
ORDER BY timestamp DESC
LIMIT 20;

你应该会看到诸如 INVOCATION_STARTING、LLM_REQUEST、LLM_RESPONSE、TOOL_STARTING、TOOL_COMPLETED 和 INVOCATION_COMPLETED 等事件。

替代方案：使用 Agent Platform SDK 部署¶

你也可以直接使用 Agent Platform SDK 以编程方式部署。这对于 CI/CD 流水线或自定义部署工作流非常有用：

deploy.py

import vertexai
from vertexai import agent_engines
from my_bq_agent.agent import app

PROJECT_ID = "your-gcp-project-id"
LOCATION = "us-central1"
STAGING_BUCKET = "gs://your-staging-bucket"

vertexai.init(
    project=PROJECT_ID, location=LOCATION, staging_bucket=STAGING_BUCKET
)
client = vertexai.Client(project=PROJECT_ID, location=LOCATION)

remote_app = client.agent_engines.create(
    agent=app,
    config={
        "display_name": "My BQ Analytics Agent",
        "staging_bucket": STAGING_BUCKET,
        "requirements": [
            "google-adk[bigquery]",
            "google-cloud-aiplatform[agent_engines]",
            "google-cloud-bigquery-storage",
            "pyarrow",
            "opentelemetry-api",
            "opentelemetry-sdk",
        ],
    },
)
print(f"Deployed agent: {remote_app.api_resource.name}")

故障排查¶

如果部署后事件未出现在你的 BigQuery 表中：

检查 ADK 版本：确保在你的 requirements 中包含 google-adk>=1.24.0。早期版本在服务器端运行时挂起进程之前不会刷新待处理的事件。

启用调试日志：在你的 agent.py 顶部添加以下内容以暴露任何隐式错误：

import logging
logging.basicConfig(level=logging.INFO)
logging.getLogger("google_adk").setLevel(logging.DEBUG)

检查 IAM 权限：Agent Runtime 服务账号需要在目标表上拥有 roles/bigquery.dataEditor 权限，在项目上拥有 roles/bigquery.jobUser 权限。对于跨项目日志记录，还要确保源项目中启用了 BigQuery API，并且服务账号在目标表上具有 bigquery.tables.updateData 权限。
验证插件初始化：在 Cloud Logging 中，通过 resource.type="reasoning_engine" 进行过滤，查找插件启动消息或错误日志。
调试时使用立即刷新：在 BigQueryLoggerConfig 中设置 batch_size=1 和 batch_flush_interval=0.1 以排除缓冲问题。

追踪与可观测性¶

该插件支持 OpenTelemetry 用于分布式追踪。OpenTelemetry 已作为 ADK 的核心依赖项包含，且始终可用。

自动跨度管理：插件自动为智能体执行、LLM 调用和工具执行生成跨度 (Spans)。
OpenTelemetry 集成：如果配置了 TracerProvider（如上例所示），插件将使用有效的 OTel 跨度，并使用标准的 OTel 标识符填充 trace_id、span_id 和 parent_span_id。这允许你将智能体日志与分布式系统中的其他服务进行关联。
回退机制：如果未配置 TracerProvider（即仅激活了默认的无操作提供程序），插件会自动回退到为跨度生成内部 UUID，并使用 invocation_id 作为跟踪 ID。这确保了父子层次结构（智能体 -> 跨度 -> 工具/LLM）在 BigQuery 日志中始终得到保留。

配置选项¶

BigQueryAgentAnalyticsPlugin 构造函数接受以下参数：

project_id (str): Google Cloud 项目 ID。
dataset_id (str): BigQuery 数据集 ID。
table_id (Optional[str], 默认值: None): BigQuery 表 ID。如果同时设置，将覆盖 BigQueryLoggerConfig 中的 table_id。
config (Optional[BigQueryLoggerConfig], 默认值: None): 用于详细配置的 BigQueryLoggerConfig 实例。
location (str, 默认值: "US"): BigQuery 数据集位置（例如 "US"、"EU"、"us-central1"）。
credentials (Optional[google.auth.credentials.Credentials], 默认值: None): 插件 BigQuery 和 GCS 客户端的显式 Google 凭据。如果为 None，插件使用应用默认凭据。适用于服务帐号、模拟身份或跨项目设置。

构造函数也接受 **kwargs，它们会直接转交给 BigQueryLoggerConfig。这使你无需创建单独的配置对象即可传递配置字段（如 batch_size 或 enabled）：

plugin = BigQueryAgentAnalyticsPlugin(
    project_id="my-project",
    dataset_id="my_dataset",
    batch_size=10,           # forwarded to BigQueryLoggerConfig
    shutdown_timeout=5.0,    # forwarded to BigQueryLoggerConfig
)

enabled (bool, 默认值: True): 要禁用该插件记录智能体数据到 BigQuery 表，请将此参数设置为 False。
table_id (str, 默认值: "agent_events"): 数据集中的 BigQuery 表 ID。也可以通过 BigQueryAgentAnalyticsPlugin 构造函数上的 table_id 参数覆盖，后者具有更高优先级。
clustering_fields (List[str], 默认值: ["event_type", "agent", "user_id"]): 自动创建表时用于聚簇 BigQuery 表的字段。
gcs_bucket_name (Optional[str], 默认值: None): 用于卸载大内容（图像、blob、大文本）的 GCS 存储桶名称。如果未提供，大内容可能会被截断或替换为占位符。
connection_id (Optional[str], 默认值: None): 用作 ObjectRef 列授权者的 BigQuery 连接 ID（例如 us.my-connection）。使用 BigQuery ML 的 ObjectRef 时需要。
max_content_length (int, 默认值: 500 * 1024): 在卸载到 GCS（如果配置了）或截断之前，在 BigQuery 中内联存储的文本内容的最大长度（字符数）。默认值为 500 KB。
batch_size (int, 默认值: 1): 在写入 BigQuery 之前批量缓存的事件数。
batch_flush_interval (float, 默认值: 1.0): 刷新部分批次前等待的最长时间（秒）。
shutdown_timeout (float, 默认值: 10.0): 关闭时等待日志刷新的最长时间（秒）。
event_allowlist (Optional[List[str]], 默认值: None): 要记录的事件类型列表。如果为 None，则记录除 event_denylist 中指定类型之外的所有事件。有关支持的事件类型的完整列表，请参阅事件类型和负载部分。
event_denylist (Optional[List[str]], 默认值: None): 要跳过的日志记录事件类型列表。有关支持的事件类型的完整列表，请参阅事件类型和负载部分。
content_formatter (Optional[Callable[[Any, str], Any]], 默认值: None): 一个可选的函数，用于在记录之前格式化事件内容。该函数接收两个参数：原始内容和事件类型字符串（例如 "LLM_REQUEST"）。
log_multi_modal_content (bool, 默认值: True): 是否记录详细的内容部分（包括 GCS 引用）。
queue_max_size (int, 默认值: 10000): 在丢弃新事件之前，内存队列中可容纳的最大事件数。
retry_config (RetryConfig, 默认值: RetryConfig()): 重试失败 BigQuery 写入的配置。子字段：max_retries（默认值 3）、initial_delay（默认值 1.0 秒）、multiplier（默认值 2.0）、max_delay（默认值 10.0 秒）。
log_session_metadata (bool, 默认值: True): 如果为 True，将会话信息记录到 attributes 列中，包括 session_id、app_name、user_id 以及会话 state 字典（例如自定义状态，如 gchat thread-id、customer_id）。前缀为 temp: 或 secret: 的状态键会自动脱敏。请参阅内置脱敏。
custom_tags (Dict[str, Any], 默认值: {}): 一个静态标签字典（例如 {"env": "prod", "version": "1.0"}），将包含在每个事件的 attributes 列中。
auto_schema_upgrade (bool, 默认值: True): 启用后，当插件架构发生变化时，插件会自动向现有表添加新列。仅进行追加更改（从不删除或修改列）。表上的版本标签（adk_schema_version）确保每次架构版本最多运行一次差异。建议保留启用状态。
create_views (bool, 默认值: True): 在 1.27.0 中新增。启用后，会自动生成按事件类型分类的 BigQuery 视图，将结构化的 JSON 数据（如 content 或 attributes）展开为扁平的、带类型的列，显著简化 SQL 查询。
view_prefix (str, 默认值: "v"): 自动创建的视图名称前缀。默认值 "v" 会生成如 v_llm_request 的视图名称。当多个插件实例共享同一数据集时，为每个实例设置不同的前缀以避免视图名称冲突（例如 "v_staging" 会生成 v_staging_llm_request）。必须非空。

示例：自定义格式化器以脱敏美元金额：

import json
import re
from typing import Any
from google.adk.plugins.bigquery_agent_analytics_plugin import BigQueryLoggerConfig

def redact_dollar_amounts(event_content: Any, event_type: str) -> str:
    """
    用于脱敏美元金额（例如：$600，$12.50）的自定义格式化器，
    并确保如果输入是字典则输出 JSON。

    参数：
        event_content: 事件的原始内容。
        event_type: 事件类型字符串（例如，"LLM_REQUEST", "LLM_RESPONSE"）。
    """
    text_content = ""
    if isinstance(event_content, dict):
        text_content = json.dumps(event_content)
    else:
        text_content = str(event_content)

    # 正则表达式查找美元金额：$ 后跟数字，可选择带有逗号或小数。
    # 示例：$600，$1,200.50，$0.99
    redacted_content = re.sub(r'\$\d+(?:,\d{3})*(?:\.\d+)?', 'xxx', text_content)
    return redacted_content

config = BigQueryLoggerConfig(
    enabled=True,
    event_allowlist=["LLM_REQUEST", "LLM_RESPONSE"], # 仅记录这些事件
    # event_denylist=["TOOL_STARTING"], # 跳过这些事件
    shutdown_timeout=10.0, # 退出时等待日志刷新的最长时间为 10 秒
    max_content_length=500, # 将内容截断为 500 个字符
    content_formatter=redact_dollar_amounts, # 脱敏日志内容中的美元金额
    queue_max_size=10000, # 内存中容纳的最大事件数
    auto_schema_upgrade=True, # 自动向现有表添加新列
    create_views=True, # 自动创建按事件类型分类的视图
    # retry_config=RetryConfig(max_retries=3), # 可选：配置重试
)

plugin = BigQueryAgentAnalyticsPlugin(..., config=config)

公共方法¶

该插件公开了几个用于生命周期管理的公共方法：

await plugin.flush()：将所有待处理事件刷新到 BigQuery。在关闭之前调用以避免数据丢失。
await plugin.shutdown(timeout=None)：优雅地关闭插件，刷新待处理事件并释放资源。可选的 timeout 参数会覆盖配置中的 shutdown_timeout。
await plugin.create_analytics_views()：手动（重新）创建所有按事件类型分类的分析视图。在架构升级或需要刷新视图时使用。

异步上下文管理器：该插件支持 async with 用于自动启动和关闭：

async with BigQueryAgentAnalyticsPlugin(
    project_id=PROJECT_ID, dataset_id=DATASET_ID
) as plugin:
    # plugin is initialized and ready to use
    ...
# plugin.shutdown() is called automatically on exit

模式和生产设置¶

模式参考¶

事件表（agent_events）使用灵活的模式。下表提供了带有示例值的全面参考。

字段名称	类型	模式	描述	示例值
timestamp	`TIMESTAMP`	`REQUIRED`	事件创建的 UTC 时间戳。用作主要排序键和每日分区键。精度为微秒。	`2026-02-03 20:52:17 UTC`
event_type	`STRING`	`NULLABLE`	规范的事件类别。标准值包括 `LLM_REQUEST`、`LLM_RESPONSE`、`LLM_ERROR`、`TOOL_STARTING`、`TOOL_COMPLETED`、`TOOL_ERROR`、`AGENT_STARTING`、`AGENT_COMPLETED`、`STATE_DELTA`、`INVOCATION_STARTING`、`INVOCATION_COMPLETED`、`USER_MESSAGE_RECEIVED` 以及 HITL 事件（请参见 HITL 事件）。用于高级过滤。	`LLM_REQUEST`
agent	`STRING`	`NULLABLE`	负责此事件的智能体的名称。在智能体初始化期间或通过 `root_agent_name` 上下文定义。	`my_bq_agent`
session_id	`STRING`	`NULLABLE`	整个对话线程的持久标识符。在多次对话轮次和子智能体调用中保持不变。	`04275a01-1649-4a30-b6a7-5b443c69a7bc`
invocation_id	`STRING`	`NULLABLE`	单次执行轮次或请求周期的唯一标识符。在许多上下文中对应于 `trace_id`。	`e-b55b2000-68c6-4e8b-b3b3-ffb454a92e40`
user_id	`STRING`	`NULLABLE`	发起会话的用户（人或系统）的标识符。从 `User` 对象或元数据中提取。	`test_user`
trace_id	`STRING`	`NULLABLE`	OpenTelemetry 跟踪 ID（32 位十六进制）。连接单个分布式请求生命周期内的所有操作。	`e-b55b2000-68c6-4e8b-b3b3-ffb454a92e40`
span_id	`STRING`	`NULLABLE`	OpenTelemetry 跨度 ID（16 位十六进制）。唯一标识此特定原子操作。	`69867a836cd94798be2759d8e0d70215`
parent_span_id	`STRING`	`NULLABLE`	直接调用方的跨度 ID。用于重建父子执行树 (DAG)。	`ef5843fe40764b4b8afec44e78044205`
content	`JSON`	`NULLABLE`	主要事件负载。结构基于 `event_type` 多态。	`{"system_prompt": "You are...", "prompt": [{"role": "user", "content": "hello"}], "response": "Hi", "usage": {"total": 15}}`
attributes	`JSON`	`NULLABLE`	元数据/增强信息（使用统计、模型信息、工具来源、自定义标签）。	`{"model": "gemini-flash-latest", "usage_metadata": {"total_token_count": 15}, "session_metadata": {"session_id": "...", "app_name": "...", "user_id": "...", "state": {}}, "custom_tags": {"env": "prod"}}`
latency_ms	`JSON`	`NULLABLE`	性能指标。标准键为 `total_ms`（墙钟时间）和 `time_to_first_token_ms`（流式延迟）。	`{"total_ms": 1250, "time_to_first_token_ms": 450}`
status	`STRING`	`NULLABLE`	高级别结果。值：`OK`（成功）或 `ERROR`（失败）。	`OK`
error_message	`STRING`	`NULLABLE`	人类可读的异常消息或堆栈跟踪片段。仅在 `status` 为 `ERROR` 时填充。	`Error 404: Dataset not found`
is_truncated	`BOOLEAN`	`NULLABLE`	如果 `content` 或 `attributes` 超过 BigQuery 单元大小限制（默认 10MB）并被部分丢弃，则为 `true`。	`false`
content_parts	`RECORD`	`REPEATED`	多模态片段数组（文本、图像、Blob）。当内容无法序列化为简单 JSON 时使用（例如大二进制文件或 GCS 引用）。	`[{"mime_type": "text/plain", "text": "hello"}]`

如果表不存在，插件会自动创建表。但是，对于生产环境，我们建议使用以下 DDL 手动创建表，该 DDL 利用 JSON 类型实现灵活性，并利用 REPEATED RECORD 实现多模态内容。

推荐的 DDL：

CREATE TABLE `your-gcp-project-id.adk_agent_logs.agent_events`
(
  timestamp TIMESTAMP NOT NULL OPTIONS(description="记录事件的 UTC 时间。"),
  event_type STRING OPTIONS(description="指示正在记录的事件类型（例如，'LLM_REQUEST'，'TOOL_COMPLETED'）。"),
  agent STRING OPTIONS(description="与事件关联的 ADK 智能体或作者的名称。"),
  session_id STRING OPTIONS(description="用于在单个对话或用户会话内对事件进行分组的唯一标识符。"),
  invocation_id STRING OPTIONS(description="会话中每次单独的智能体执行或回合的唯一标识符。"),
  user_id STRING OPTIONS(description="与当前会话关联的用户标识符。"),
  trace_id STRING OPTIONS(description="用于分布式跟踪的 OpenTelemetry 跟踪 ID。"),
  span_id STRING OPTIONS(description="此特定操作的 OpenTelemetry 跨度 ID。"),
  parent_span_id STRING OPTIONS(description="OpenTelemetry 父跨度 ID 以重建层次结构。"),
  content JSON OPTIONS(description="作为 JSON 存储的事件特定数据（负载）。"),
  content_parts ARRAY<STRUCT<
    mime_type STRING,
    uri STRING,
    object_ref STRUCT<
      uri STRING,
      version STRING,
      authorizer STRING,
      details JSON
    >,
    text STRING,
    part_index INT64,
    part_attributes STRING,
    storage_mode STRING
  >> OPTIONS(description="多模态数据的详细内容部分。"),
  attributes JSON OPTIONS(description="用于额外元数据的任意键值对（例如，'root_agent_name'、'model_version'、'usage_metadata'、'session_metadata'、'custom_tags'）。"),
  latency_ms JSON OPTIONS(description="延迟测量值（例如，total_ms）。"),
  status STRING OPTIONS(description="事件的结果，通常为 'OK' 或 'ERROR'。"),
  error_message STRING OPTIONS(description="如果发生错误则填充。"),
  is_truncated BOOLEAN OPTIONS(description="标识内容是否被截断。")
)
PARTITION BY DATE(timestamp)
CLUSTER BY event_type, agent, user_id;

自动创建的视图¶

当 create_views=True 时（1.27.0 及更高版本的默认设置），插件会自动为每个事件类型生成视图，将常见的 JSON 结构展开为扁平的类型化列。这显著简化了 SQL，消除了显式编写复杂的 JSON_VALUE 或 JSON_QUERY 函数的需求。

视图名称遵循 {view_prefix}_{event_type_lowercase} 的约定（例如，使用默认前缀 "v"，LLM_REQUEST 变为 v_llm_request）。当多个插件实例写入同一数据集中的不同表时，在 BigQueryLoggerConfig 中将 view_prefix 设置为不同的值，以防止视图名称冲突：

# Two plugins in the same dataset with distinct view prefixes
plugin_prod = BigQueryAgentAnalyticsPlugin(
    project_id=PROJECT_ID, dataset_id=DATASET_ID,
    table_id="agent_events_prod",
    config=BigQueryLoggerConfig(view_prefix="v_prod"),
)
# Creates views: v_prod_llm_request, v_prod_tool_completed, ...

plugin_staging = BigQueryAgentAnalyticsPlugin(
    project_id=PROJECT_ID, dataset_id=DATASET_ID,
    table_id="agent_events_staging",
    config=BigQueryLoggerConfig(view_prefix="v_staging"),
)
# Creates views: v_staging_llm_request, v_staging_tool_completed, ...

你也可以调用公共异步方法 await plugin.create_analytics_views() 来手动刷新视图，例如在架构升级之后。

每个视图都包含以下公共列：timestamp、event_type、agent、session_id、invocation_id、user_id、trace_id、span_id、parent_span_id、status、error_message、is_truncated。

以下表格列出了所有 16 个自动创建的视图及其事件特定列：

视图名称	事件特定列
`v_user_message_received`	（仅公共列）
`v_llm_request`	`model` (STRING)、`request_content` (JSON)、`llm_config` (JSON)、`tools` (JSON)
`v_llm_response`	`response` (JSON)、`usage_prompt_tokens` (INT64)、`usage_completion_tokens` (INT64)、`usage_total_tokens` (INT64)、`usage_cached_tokens` (INT64)、`total_ms` (INT64)、`ttft_ms` (INT64)、`model_version` (STRING)、`usage_metadata` (JSON)、`cache_metadata` (JSON)、`context_cache_hit_rate` (FLOAT64)
`v_llm_error`	`total_ms` (INT64)
`v_tool_starting`	`tool_name` (STRING)、`tool_args` (JSON)、`tool_origin` (STRING)
`v_tool_completed`	`tool_name` (STRING)、`tool_result` (JSON)、`tool_origin` (STRING)、`total_ms` (INT64)
`v_tool_error`	`tool_name` (STRING)、`tool_args` (JSON)、`tool_origin` (STRING)、`total_ms` (INT64)
`v_agent_starting`	`agent_instruction` (STRING)
`v_agent_completed`	`total_ms` (INT64)
`v_invocation_starting`	（仅公共列）
`v_invocation_completed`	（仅公共列）
`v_state_delta`	`state_delta` (JSON)
`v_hitl_credential_request`	`tool_name` (STRING)、`tool_args` (JSON)
`v_hitl_confirmation_request`	`tool_name` (STRING)、`tool_args` (JSON)
`v_hitl_input_request`	`tool_name` (STRING)、`tool_args` (JSON)
`v_a2a_interaction`	`response_content` (JSON)、`a2a_task_id` (STRING)、`a2a_context_id` (STRING)、`a2a_request` (JSON)、`a2a_response` (JSON)

事件类型和负载¶

content 列现在包含特定于 event_type 的 JSON 对象。content_parts 列提供了内容的结构化视图，对于图像或卸载的数据特别有用。

内容截断

可变内容字段被截断为 max_content_length（在 BigQueryLoggerConfig 中配置，默认 500KB）。
如果配置了 gcs_bucket_name，大内容将被卸载到 GCS 而不是被截断，并且引用存储在 content_parts.object_ref 中。

LLM 交互 (插件生命周期)¶

LLM_REQUEST：记录发送给模型的详细提示词和配置。
LLM_RESPONSE：记录模型的输出内容及令牌消耗。
LLM_ERROR：捕获调用失败时的异常信息。

这些事件追踪发送给 LLM 的原始请求以及从 LLM 接收到的响应。

1. LLM_REQUEST

捕获发送给模型的提示语，包括对话历史和系统指令。

{
  "event_type": "LLM_REQUEST",
  "content": {
    "system_prompt": "You are a helpful assistant...",
    "prompt": [
      {
        "role": "user",
        "content": "hello how are you today"
      }
    ]
  },
  "attributes": {
    "root_agent_name": "my_bq_agent",
    "model": "gemini-flash-latest",
    "tools": ["list_dataset_ids", "execute_sql"],
    "llm_config": {
      "temperature": 0.5,
      "top_p": 0.9
    }
  }
}

2. LLM_RESPONSE

捕获模型的输出和令牌（token）使用统计信息。

{
  "event_type": "LLM_RESPONSE",
  "content": {
    "response": "text: 'Hello! I'm doing well...'",
    "usage": {
      "completion": 19,
      "prompt": 10129,
      "total": 10148
    }
  },
  "attributes": {
    "root_agent_name": "my_bq_agent",
    "model_version": "gemini-flash-latest",
    "usage_metadata": {
      "prompt_token_count": 10129,
      "candidates_token_count": 19,
      "total_token_count": 10148
    }
  },
  "latency_ms": {
    "time_to_first_token_ms": 2579,
    "total_ms": 2579
  }
}

3. LLM_ERROR

当 LLM 调用失败并抛出异常时记录。捕获错误消息并关闭当前跨度（span）。

{
  "event_type": "LLM_ERROR",
  "content": null,
  "attributes": {
    "root_agent_name": "my_bq_agent"
  },
  "error_message": "Error 429: Resource exhausted",
  "latency_ms": {
    "total_ms": 350
  }
}

工具使用 (插件生命周期)¶

追踪工具执行情况，tool_origin 字段标识来源：LOCAL (本地脚本), MCP (MCP 服务器), SUB_AGENT (子智能体) 等。 - TOOL_STARTING：工具开始运行。 - TOOL_COMPLETED：工具运行完成并返回结果。 - TOOL_ERROR：工具运行出错。

这些事件追踪智能体执行工具的情况。每个工具事件都包含一个 tool_origin 字段，用于对工具的来源进行分类：

工具来源	描述
`LOCAL`	`FunctionTool` 实例（本地 Python 函数）
`MCP`	模型上下文协议工具（`McpTool` 实例）
`SUB_AGENT`	`AgentTool` 实例（子智能体）
`A2A`	远程智能体到智能体实例（`RemoteA2aAgent`）
`TRANSFER_AGENT`	`TransferToAgentTool` 实例（通用智能体移交）
`TRANSFER_A2A`	移交给 `RemoteA2aAgent` 的 `TransferToAgentTool` 实例（在调用级别分类）
`UNKNOWN`	未分类的工具

4. TOOL_STARTING

当智能体开始执行工具时记录。

{
  "event_type": "TOOL_STARTING",
  "content": {
    "tool": "list_dataset_ids",
    "args": {
      "project_id": "bigquery-public-data"
    },
    "tool_origin": "LOCAL"
  }
}

5. TOOL_COMPLETED

当工具执行结束时记录。

{
  "event_type": "TOOL_COMPLETED",
  "content": {
    "tool": "list_dataset_ids",
    "result": [
      "austin_311",
      "austin_bikeshare"
    ],
    "tool_origin": "LOCAL"
  },
  "latency_ms": {
    "total_ms": 467
  }
}

6. TOOL_ERROR

当工具执行失败并抛出异常时记录。捕获工具名称、参数、工具来源和错误消息。

{
  "event_type": "TOOL_ERROR",
  "content": {
    "tool": "list_dataset_ids",
    "args": {
      "project_id": "nonexistent-project"
    },
    "tool_origin": "LOCAL"
  },
  "error_message": "Error 404: Dataset not found",
  "latency_ms": {
    "total_ms": 150
  }
}

状态管理¶

这些事件追踪智能体状态的更改，通常由工具触发。

7. STATE_DELTA

追踪智能体内部状态的更改（例如，由工具更新的自定义应用程序状态）。

Built-in redaction

State keys prefixed with temp: or secret: are automatically redacted to [REDACTED] in the logged state_delta. See Built-in redaction for details.

{
  "event_type": "STATE_DELTA",
  "attributes": {
    "state_delta": {
      "customer_tier": "enterprise",
      "last_query_dataset": "bigquery-public-data.samples"
    }
  }
}

智能体生命周期和通用事件¶

事件类型

内容（JSON）结构

INVOCATION_STARTING

{}

INVOCATION_COMPLETED

{}

AGENT_STARTING

"你是一个有用的智能体..."

AGENT_COMPLETED

{}

USER_MESSAGE_RECEIVED

{"text_summary": "帮我预订航班。"}

人机回环 (HITL) 事件¶

自动检测并记录专用 HITL 工具调用： - HITL_CREDENTIAL_REQUEST：请求凭证。 - HITL_CONFIRMATION_REQUEST：请求确认。 - HITL_INPUT_REQUEST：请求用户输入。

插件会自动检测对 ADK 合成人机回环（HITL）工具的调用，并为它们发出专用事件类型。这些事件会额外于正常的 TOOL_STARTING / TOOL_COMPLETED 事件进行记录。

系统可识别以下 HITL 工具名称：

adk_request_credential — 请求用户凭据（例如 OAuth 令牌）
adk_request_confirmation — 在继续操作前请求用户确认
adk_request_input — 请求用户自由形式的输入

事件类型	触发器	内容 (JSON) 结构
`HITL_CREDENTIAL_REQUEST`	智能体调用 `adk_request_credential`	`{"tool": "adk_request_credential", "args": {...}}`
`HITL_CREDENTIAL_REQUEST_COMPLETED`	用户提供凭据响应	`{"tool": "adk_request_credential", "result": {...}}`
`HITL_CONFIRMATION_REQUEST`	智能体调用 `adk_request_confirmation`	`{"tool": "adk_request_confirmation", "args": {...}}`
`HITL_CONFIRMATION_REQUEST_COMPLETED`	用户提供确认响应	`{"tool": "adk_request_confirmation", "result": {...}}`
`HITL_INPUT_REQUEST`	智能体调用 `adk_request_input`	`{"tool": "adk_request_input", "args": {...}}`
`HITL_INPUT_REQUEST_COMPLETED`	用户提供输入响应	`{"tool": "adk_request_input", "result": {...}}`

HITL 事件的视图

自动创建的视图仅适用于三种请求事件类型(v_hitl_credential_request、v_hitl_confirmation_request、v_hitl_input_request)。三种 *_COMPLETED 事件类型会记录到基础表中，但没有专用的视图。请通过 agent_events 表使用 WHERE event_type LIKE 'HITL_%_COMPLETED' 直接查询。

A2A 交互事件¶

当你的智能体通过智能体到智能体（A2A）协议与远程智能体通信时，插件会记录 A2A_INTERACTION 事件，捕获请求和响应详情。

A2A_INTERACTION

当 A2A 远程智能体调用完成时记录。

{
  "event_type": "A2A_INTERACTION",
  "content": {
    "response_content": "The remote agent's response...",
    "a2a_task_id": "task-abc123",
    "a2a_context_id": "ctx-def456",
    "a2a_request": { ... },
    "a2a_response": { ... }
  }
}

GCS 卸载示例（多模态与大文本）¶

当配置了 gcs_bucket_name 时，大文本和多模态内容（图像、音频等）会自动卸载到 GCS。content 列将包含摘要或占位符，而 content_parts 包含指向 GCS URI 的 object_ref。

卸载文本示例

{
  "event_type": "LLM_REQUEST",
  "content_parts": [
    {
      "part_index": 1,
      "mime_type": "text/plain",
      "storage_mode": "GCS_REFERENCE",
      "text": "AAAA... [OFFLOADED]",
      "object_ref": {
        "uri": "gs://haiyuan-adk-debug-verification-1765319132/2025-12-10/e-f9545d6d/ae5235e6_p1.txt",
        "authorizer": "us.bqml_connection",
        "details": {"gcs_metadata": {"content_type": "text/plain"}}
      }
    }
  ]
}

卸载图像示例

{
  "event_type": "LLM_REQUEST",
  "content_parts": [
    {
      "part_index": 2,
      "mime_type": "image/png",
      "storage_mode": "GCS_REFERENCE",
      "text": "[MEDIA OFFLOADED]",
      "object_ref": {
        "uri": "gs://haiyuan-adk-debug-verification-1765319132/2025-12-10/e-f9545d6d/ae5235e6_p2.png",
        "authorizer": "us.bqml_connection",
        "details": {"gcs_metadata": {"content_type": "image/png"}}
      }
    }
  ]
}

查询卸载内容（获取签名 URL）

SELECT
  timestamp,
  event_type,
  part.mime_type,
  part.storage_mode,
  part.object_ref.uri AS gcs_uri,
  -- 生成签名 URL 以直接读取内容（需要 connection_id 配置）
  STRING(OBJ.GET_ACCESS_URL(part.object_ref, 'r').access_urls.read_url) AS signed_url
FROM `your-gcp-project-id.your-dataset-id.agent_events`,
UNNEST(content_parts) AS part
WHERE part.storage_mode = 'GCS_REFERENCE'
ORDER BY timestamp DESC
LIMIT 10;

高级分析查询¶

追踪特定对话轮次¶

SELECT timestamp, event_type, agent, JSON_VALUE(content, '$.response') as summary
FROM `your-project.your-dataset.agent_events`
WHERE trace_id = 'your-trace-id'
ORDER BY timestamp ASC;

令牌使用分析¶

使用 v_llm_response 视图（推荐）：

SELECT
  AVG(usage_total_tokens) as avg_tokens,
  AVG(usage_prompt_tokens) as avg_prompt_tokens,
  AVG(usage_completion_tokens) as avg_completion_tokens
FROM `your-gcp-project-id.your-dataset-id.v_llm_response`;

或者使用基础表配合 JSON 提取：

SELECT
  AVG(CAST(JSON_VALUE(content, '$.usage.total') AS INT64)) as avg_tokens
FROM `your-project.your-dataset.agent_events`
WHERE event_type = 'LLM_RESPONSE';

查询多模态内容（使用 content_parts 和 ObjectRef）¶

SELECT
  timestamp,
  part.mime_type,
  part.object_ref.uri as gcs_uri
FROM `your-gcp-project-id.your-dataset-id.agent_events`,
UNNEST(content_parts) as part
WHERE part.mime_type LIKE 'image/%'
ORDER BY timestamp DESC;

使用 BigQuery 远程模型（Gemini）分析多模态内容¶

SELECT
  logs.session_id,
  -- 获取图像的签名 URL
  STRING(OBJ.GET_ACCESS_URL(parts.object_ref, "r").access_urls.read_url) as signed_url,
  -- 使用远程模型（例如，gemini-pro-vision）分析图像
  AI.GENERATE(
    ('简要描述此图像。什么公司标志？', parts.object_ref)
  ) AS generated_result
FROM
  `your-gcp-project-id.your-dataset-id.agent_events` logs,
  UNNEST(logs.content_parts) AS parts
WHERE
  parts.mime_type LIKE 'image/%'
ORDER BY logs.timestamp DESC
LIMIT 1;

延迟分析（LLM 和工具）¶

使用视图（推荐）：

-- LLM 延迟
SELECT AVG(total_ms) as avg_llm_ms, AVG(ttft_ms) as avg_ttft_ms
FROM `your-gcp-project-id.your-dataset-id.v_llm_response`;

-- 按工具名称统计工具延迟
SELECT tool_name, tool_origin, AVG(total_ms) as avg_tool_ms
FROM `your-gcp-project-id.your-dataset-id.v_tool_completed`
GROUP BY tool_name, tool_origin
ORDER BY avg_tool_ms DESC;

或者使用基础表：

SELECT
  event_type,
  AVG(CAST(JSON_VALUE(latency_ms, '$.total_ms') AS INT64)) as avg_latency_ms
FROM `your-gcp-project-id.your-dataset-id.agent_events`
WHERE event_type IN ('LLM_RESPONSE', 'TOOL_COMPLETED')
GROUP BY event_type;

跨度层次结构和时长分析¶

SELECT
  span_id,
  parent_span_id,
  event_type,
  timestamp,
  -- 从 latency_ms 中提取已完成操作的持续时间
  CAST(JSON_VALUE(latency_ms, '$.total_ms') AS INT64) as duration_ms,
  -- 识别特定工具或操作
  COALESCE(
    JSON_VALUE(content, '$.tool'),
    'LLM_CALL'
  ) as operation
FROM `your-gcp-project-id.your-dataset-id.agent_events`
WHERE trace_id = 'your-trace-id'
  AND event_type IN ('LLM_RESPONSE', 'TOOL_COMPLETED')
ORDER BY timestamp ASC;

错误分析（LLM 和工具错误）¶

使用视图（推荐）：

-- 带来源的工具错误
SELECT timestamp, agent, tool_name, tool_origin, error_message, total_ms
FROM `your-gcp-project-id.your-dataset-id.v_tool_error`
ORDER BY timestamp DESC
LIMIT 20;

-- LLM 错误
SELECT timestamp, agent, error_message, total_ms
FROM `your-gcp-project-id.your-dataset-id.v_llm_error`
ORDER BY timestamp DESC
LIMIT 20;

工具来源分析¶

使用 v_tool_completed 视图（推荐）：

SELECT
  tool_origin,
  tool_name,
  COUNT(*) as call_count,
  AVG(total_ms) as avg_latency_ms
FROM `your-gcp-project-id.your-dataset-id.v_tool_completed`
GROUP BY tool_origin, tool_name
ORDER BY call_count DESC;

人机回环 (HITL) 交互分析¶

SELECT
  timestamp,
  event_type,
  session_id,
  JSON_VALUE(content, '$.tool') as hitl_tool,
  content
FROM `your-gcp-project-id.your-dataset-id.agent_events`
WHERE event_type LIKE 'HITL_%'
ORDER BY timestamp DESC
LIMIT 20;

AI 驱动的根本原因分析 (Agent Ops)¶

自动分析失败的会话，使用 BigQuery ML 和 Gemini 确定错误的根本原因。

DECLARE failed_session_id STRING;
-- 查找最近失败的会话
SET failed_session_id = (
    SELECT session_id
    FROM `your-gcp-project-id.your-dataset-id.agent_events`
    WHERE error_message IS NOT NULL
    ORDER BY timestamp DESC
    LIMIT 1
);

-- 重建完整的对话上下文
WITH SessionContext AS (
    SELECT
        session_id,
        STRING_AGG(CONCAT(event_type, ': ', COALESCE(TO_JSON_STRING(content), '')), '\n' ORDER BY timestamp) as full_history
    FROM `your-project.your-dataset.agent_events`
    WHERE session_id = 'failed-session-id'
    GROUP BY session_id
)
SELECT
    session_id,
    AI.GENERATE(
        ('分析此对话日志并解释失败的根本原因。日志：', full_history),
        endpoint => 'gemini-flash-latest'
    ).result AS root_cause_explanation
FROM SessionContext;

BigQuery 中的对话分析¶

你还可以使用 BigQuery 会话分析 (Conversational Analytics) 使用自然语言分析你的智能体日志。使用此工具来回答诸如以下问题：

"显示随时间变化的错误率"
"最常见的工具调用是什么？"
"识别高令牌使用的会话"

使用 BigQuery 智能体分析 SDK 消费记录数据¶

BigQuery 智能体分析 SDK 提供了一种便捷的方式来消费和分析 BigQuery 智能体分析插件记录的数据。该 SDK 提供了预构建的工具，用于直接从 BigQuery 查询、聚合和可视化智能体的操作数据。

仪表板¶

BigQuery 智能体分析 SDK 包含一个示例 Jupyter notebook，演示了如何查询和可视化智能体的性能数据。将其作为起点，构建适合你 BigQuery 智能体分析数据集的自定义仪表板。

安全性：避免记录敏感凭据¶

请勿记录 OAuth 令牌、API 密钥或客户端密钥

BigQuery 智能体分析插件会捕捉详细的事件负载，包括工具参数、LLM 提示词以及身份验证相关事件（如 HITL 凭据请求）。如果你的智能体使用了受验证的工具（例如使用 OAuth2 的 AuthenticatedFunctionTool），插件可能会将 client_secret、access_token 或 API 密钥等敏感值记录到 BigQuery 表的 content 列中。

这是一个已知问题（google/adk-python#3845），并可能导致你的分析数据中发生凭据泄露。

该插件包含内置脱敏功能，可自动保护常见密钥。如需额外控制，你可以在其上层叠自定义脱敏。

内置脱敏¶

该插件会自动脱敏以下已知键名（不区分大小写）在 content 或 attributes JSON 中出现的任何值：

client_secret、access_token、refresh_token、id_token、api_key、 password

此外，任何以 temp: 或 secret: 为前缀的状态键在记录的 state_delta 中都会自动替换为 [REDACTED]。这意味着存储在 secret: 作用域下的 ADK 会话状态（例如凭据服务缓存的 OAuth 令牌）永远不会持久化到 BigQuery 日志中，因为 secret:* 密钥会自动脱敏。

内置脱敏始终对结构化属性和状态日志记录生效，并递归应用于嵌套字典和属性值中的 JSON 编码字符串。自定义 content_formatter 在原始内容上首先运行，因此使用它来为可能出现在自由格式负载中的密钥添加掩码。

使用 `content_formatter` 脱敏额外的密钥¶

在 BigQueryLoggerConfig 中提供一个自定义的 content_formatter 函数，以便在写入之前剥离或遮蔽敏感字段：

import json
import re
from typing import Any

SENSITIVE_KEYS = {"client_secret", "access_token", "refresh_token", "api_key", "secret"}

def redact_credentials(event_content: Any, event_type: str) -> str:
    """从记录的内容中脱敏 OAuth 密钥和令牌。"""
    if isinstance(event_content, dict):
        text = json.dumps(event_content)
    else:
        text = str(event_content)

    for key in SENSITIVE_KEYS:
        # 在类 JSON 字符串中脱敏值："client_secret": "GOCSPX-xxx"
        text = re.sub(
            rf'("{key}"\s*:\s*)"[^"]*"',
            rf'\1"[REDACTED]"',
            text,
            flags=re.IGNORECASE,
        )
    return text

config = BigQueryLoggerConfig(
    content_formatter=redact_credentials,
    # ... 其他选项
)

使用 `event_denylist` 跳过凭据事件¶

如果你不需要记录身份验证相关的事件，可以将其完全排除：

config = BigQueryLoggerConfig(
    event_denylist=[
        "HITL_CREDENTIAL_REQUEST",
        "HITL_CREDENTIAL_REQUEST_COMPLETED",
    ],
    # ... 其他选项
)

通用最佳实践¶

切勿在智能体源代码中硬编码密钥。请使用环境变量或密钥管理服务（如 Google Cloud Secret Manager）来管理 OAuth 客户端密钥和 API 密钥。
限制 BigQuery 表访问权限：使用 IAM 限制能够读取已记录事件数据的人员。
定期审计日志，确保没有捕捉到意外的敏感数据。

Multiprocessing and fork safety¶

The plugin is fork-aware: it sets GRPC_ENABLE_FORK_SUPPORT=1 before loading the gRPC C-core library and registers an os.register_at_fork handler that resets inherited runtime state (gRPC channels, write streams, event loops) in child processes. This means the plugin can survive os.fork() without leaking file descriptors or sending data on a parent's connection.

However, spawn is the recommended multiprocessing start method for production deployments. fork copies the parent's address space — including any in-flight gRPC state — and the post-fork reset adds latency to the first write in each child. With spawn, each worker initializes the plugin cleanly.

For Gunicorn deployments specifically:

Prefer --preload combined with lazy plugin initialization (the plugin defers setup until the first event is logged), or
Initialize the plugin inside a post_fork hook so each worker gets its own client.

Note

The fork-safety mechanism resets runtime state only. It does not replay events that were queued but not yet flushed in the parent process at the time of fork. Call await plugin.flush() before forking if you need to guarantee delivery.

Feedback¶

We welcome your feedback on BigQuery Agent Analytics. If you have questions, suggestions, or encounter any issues, please reach out to the team at bqaa-feedback@google.com.

我们欢迎你对 BigQuery 智能体分析插件提供反馈。如果你有任何疑问、建议或遇到任何问题，请通过 bqaa-feedback@google.com 联系团队。

ADK 的 BigQuery 智能体分析插件¶

先决条件¶

IAM 权限¶

IAM 权限¶

运行并测试智能体¶

使用插件部署到 Agent Runtime¶

前置条件¶

步骤 1：定义智能体和插件¶

步骤 2：使用 ADK CLI 部署¶

步骤 3：测试已部署的智能体¶

步骤 4：在 BigQuery 中验证事件¶

替代方案：使用 Agent Platform SDK 部署¶

故障排查¶

追踪与可观测性¶

配置选项¶

公共方法¶

模式和生产设置¶

模式参考¶

自动创建的视图¶

事件类型和负载¶

LLM 交互 (插件生命周期)¶

工具使用 (插件生命周期)¶

状态管理¶

智能体生命周期和通用事件¶

人机回环 (HITL) 事件¶

A2A 交互事件¶

GCS 卸载示例（多模态与大文本）¶

高级分析查询¶

追踪特定对话轮次¶

令牌使用分析¶

查询多模态内容（使用 content_parts 和 ObjectRef）¶

使用 BigQuery 远程模型（Gemini）分析多模态内容¶

延迟分析（LLM 和工具）¶

跨度层次结构和时长分析¶

错误分析（LLM 和工具错误）¶

工具来源分析¶

人机回环 (HITL) 交互分析¶

AI 驱动的根本原因分析 (Agent Ops)¶

BigQuery 中的对话分析¶

使用 BigQuery 智能体分析 SDK 消费记录数据¶

仪表板¶

安全性：避免记录敏感凭据¶

内置脱敏¶

使用 content_formatter 脱敏额外的密钥¶

使用 event_denylist 跳过凭据事件¶

通用最佳实践¶

Multiprocessing and fork safety¶

Feedback¶

其他资源¶

使用 `content_formatter` 脱敏额外的密钥¶

使用 `event_denylist` 跳过凭据事件¶