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 增加了自动模式升级（安全地向现有表添加新列）、工具来源 (Tool Provenance) 追踪（LOCAL、MCP、SUB_AGENT、A2A、TRANSFER_AGENT）以及用于人机交流的 HITL 事件追踪。版本 1.27.0 增加了自动视图创建（生成扁平且易于查询的事件视图）。

BigQuery 智能体分析插件处于预览发布阶段。如果你想了解更多信息，请参见发布阶段描述。

BigQuery 存储写入 API

此功能使用 BigQuery 存储写入 API，这是一个付费服务。有关成本信息，请参见 BigQuery 文档。

使用场景¶

智能体工作流调试与分析：将广泛的插件生命周期事件（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 权限。

在智能体中使用¶

你可以通过配置并将 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 Vertex AI 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") # Optional

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-2.5-flash")

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;

Deploy to Agent Engine with the plugin¶

You can deploy an agent with the BigQuery Agent Analytics plugin to Vertex AI Agent Engine. This section walks through the steps to deploy using the ADK CLI, and alternatively using the Vertex AI SDK programmatically.

Version Requirement

Use ADK Python version 1.24.0 or higher to deploy with this plugin to Agent Engine. Earlier versions had an issue where the plugin's asynchronous log writer could be terminated by the serverless runtime before flushing pending events. Starting from 1.24.0, the plugin performs a synchronous flush at the end of each invocation to ensure all events are written.

Prerequisites¶

Before deploying, ensure you have completed the general Agent Engine setup, including:

A Google Cloud project with the Vertex AI API and Cloud Resource Manager API enabled.
A BigQuery dataset in the target project (or a cross-project dataset with the correct permissions).
A Cloud Storage staging bucket for deployment artifacts.
The deploying service account has the IAM roles listed in IAM permissions.
Your coding environment is authenticated with gcloud auth login and gcloud auth application-default login.

Step 1: Define the agent and plugin¶

Create your agent project folder with an App object that includes the plugin. The App object is required for Agent Engine deployments with plugins.

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 Vertex AI
# 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)
)

# --- Agent ---
root_agent = Agent(
    model=Gemini(model="gemini-2.5-flash"),
    name="my_bq_agent",
    instruction="You are a helpful assistant with access to BigQuery tools.",
    tools=[bigquery_toolset],
)

# --- App (required for Agent Engine 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

Step 2: Deploy using ADK CLI¶

Use the adk deploy agent_engine command to deploy the agent. The --adk_app flag tells the CLI which App object to use:

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 flag

The --adk_app flag specifies the module path and variable name of the App object (in the format module.variable). In this example, agent.app refers to the app variable in agent.py. This ensures the deployment correctly picks up the plugin configuration.

Once successfully deployed, you should see output like:

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

Note the Resource name for the next step.

Step 3: Test the deployed agent¶

After deployment, you can query the agent using the Vertex AI SDK:

test_deployed_agent.py

import uuid
import vertexai

PROJECT_ID = "your-gcp-project-id"
LOCATION = "us-central1"
AGENT_ID = "751619551677906944"  # from deployment output

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="List datasets in my project", user_id=user_id
):
    print(chunk, end="", flush=True)

Step 4: Verify events in BigQuery¶

After sending a few queries to the deployed agent, verify that events are being logged by querying your BigQuery table:

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

You should see events such as INVOCATION_STARTING, LLM_REQUEST, LLM_RESPONSE, TOOL_STARTING, TOOL_COMPLETED, and INVOCATION_COMPLETED.

Alternative: Deploy using the Vertex AI SDK¶

You can also deploy programmatically using the Vertex AI SDK directly. This is useful for CI/CD pipelines or custom deployment workflows:

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 Engine）服务账号需要对目标表拥有 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 以排除缓冲问题。

Tracing and Observability¶

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

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

配置选项¶

你可以使用 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)：BigQuery 连接 ID，用于授权访问存储在 GCS 中的内容（如在 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、initial_delay、multiplier、max_delay）。
log_session_metadata (bool, 默认: True)：如果为 True，则将会话信息记录到 attributes 列中，包括 session_id、app_name、user_id 和会话 state 字典（例如，自定义状态如 gchat 线程 ID、customer_id）。
custom_tags (Dict[str, Any], 默认: {})：包要包含在每个事件的 attributes 列中的静态标签字典（例如 {"env": "prod", "version": "1.0"}）。
auto_schema_upgrade (bool, 默认: True)：启用后，当插件模式演进时，插件会自动向现有表添加新列。仅进行增量更改（永远不会删除或更改列）。表上的版本标签（adk_schema_version）确保差异检查在每个模式版本中最多运行一次。可以安全地保持启用状态。
create_views (bool, 默认: True)：在 1.27.0 中添加。启用后，自动生成按事件类型分类的 BigQuery 视图，将结构化 JSON 数据（如 content 或 attributes）展开为扁平的类型化列，从而显著简化 SQL 查询。

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

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)

模式和生产设置¶

模式参考¶

事件表（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-2.5-flash", "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`	多模态段数组（文本、图像、二进制对象）。当内容无法序列化为简单 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 函数的需求。

例如，视图 v_llm_request 包含以下模式：

字段名称	类型	描述
(通用列)	`VARIES`	包括标准元数据：`timestamp`、`event_type`、`agent`、`session_id`、`invocation_id`、`user_id`、`trace_id`、`span_id`、`parent_span_id`、`status`、`error_message`、`is_truncated`。
model	`STRING`	用于请求的 LLM 模型名称。
request_content	`JSON`	原始 LLM 请求负载。
llm_config	`JSON`	传递给 LLM 的配置参数（temperature、top_p 等）。
tools	`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-2.5-flash",
    "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-2.5-flash-001",
    "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` 实例
`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

追踪智能体内部状态的更改（例如，令牌缓存更新）。

{
  "event_type": "STATE_DELTA",
  "attributes": {
    "state_delta": {
      "bigquery_token_cache": "{\"token\": \"ya29...\", \"expiry\": \"...\"}"
    }
  }
}

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

事件类型

内容（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 请求事件是通过 on_event_callback 中的 function_call 部分检测到的。HITL 完成事件是通过 on_event_callback 和 on_user_message_callback 中的 function_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;

令牌使用分析¶

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 和工具）¶

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,
  event_type,
  agent,
  error_message,
  JSON_VALUE(content, '$.tool') as tool_name,
  CAST(JSON_VALUE(latency_ms, '$.total_ms') AS INT64) as latency_ms
FROM `your-gcp-project-id.your-dataset-id.agent_events`
WHERE event_type IN ('LLM_ERROR', 'TOOL_ERROR')
ORDER BY timestamp DESC
LIMIT 20;

工具来源分析¶

SELECT
  JSON_VALUE(content, '$.tool_origin') as tool_origin,
  JSON_VALUE(content, '$.tool') as tool_name,
  COUNT(*) as call_count,
  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 = '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-2.5-flash'
    ).result AS root_cause_explanation
FROM SessionContext;

BigQuery 中的对话分析¶

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

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

Looker Studio 仪表板¶

你可以使用我们预构建的 Looker Studio 仪表板模板可视化智能体的性能。

要将此仪表板连接到你自己的 BigQuery 表，请使用以下链接格式，将占位符替换为你特定的项目、数据集和表 ID：

https://lookerstudio.google.com/reporting/create?c.reportId=f1c5b513-3095-44f8-90a2-54953d41b125&ds.ds3.connector=bigQuery&ds.ds3.type=TABLE&ds.ds3.projectId=<your-project-id>&ds.ds3.datasetId=<your-dataset-id>&ds.ds3.tableId=<your-table-id>

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

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

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

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

为了防止敏感凭据被持久化到 BigQuery 中，请采用以下一种或多种方法：

使用 `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 限制能够读取已记录事件数据的人员。
定期审计日志，确保没有捕捉到意外的敏感数据。

反馈¶

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