ADK 的 BigQuery 智能体分析插件¶

在 ADK 中支持Python v1.21.0预览版

版本要求

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

BigQuery 智能体分析插件通过提供深入的智能体行为分析的健壮解决方案，显著增强了智能体开发工具包（ADK）。使用 ADK 插件架构和 BigQuery 存储写入 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 智能体分析插件处于预览版发布阶段。欲了解更多信息，请参见[发布阶段描述](https://cloud.google.com/products#product-launch-stages)。

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 智能体的 App 对象来使用它。以下示例显示了使用此插件的智能体实现，包括 GCS 卸载：

my_bq_agent/agent.py

# 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())

# --- Configuration ---
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")
LOCATION = os.environ.get("GOOGLE_CLOUD_LOCATION", "US") # 插件中的默认位置是 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 或更新代码。")

# --- 关键：在 Gemini 实例化之前设置环境变量 ---
os.environ['GOOGLE_CLOUD_PROJECT'] = PROJECT_ID
os.environ['GOOGLE_CLOUD_LOCATION'] = 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", # default table name is agent_events
    config=bq_config,
    location=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;

追踪与可观测性¶

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

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

配置选项¶

你可以使用 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、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 查询。

以下代码示例显示了如何为 BigQuery 智能体分析插件定义配置：

import json
import re

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)

模式和生产设置¶

模式参考 (Schema Reference)¶

事件表（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;

自动创建的视图 (1.27.0+)¶

当 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`	请求期间可用的工具数组。

Event types and payloads¶

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

内容截断

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

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

这些事件追踪发送给 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`	`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) 事件¶

插件会自动检测对 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;

高级分析查询¶

使用 trace_id 跟踪特定对话回合

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

令牌使用分析（访问 JSON 字段）

SELECT
  AVG(CAST(JSON_VALUE(content, '$.usage.total') AS INT64)) as avg_tokens
FROM `your-gcp-project-id.your-dataset-id.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 与工具错误) {: #error-analysis-llm-and-tool-errors }

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;

工具来源分析 {: #tool-provenance-analysis }

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 交互分析 {: #hitl-interaction-analysis }

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-gcp-project-id.your-dataset-id.agent_events`
    WHERE session_id = failed_session_id
    GROUP BY session_id
)
-- 询问 Gemini 诊断问题
SELECT
    session_id,
    AI.GENERATE(
        ('分析这段对话日志并说明失败的根本原因。日志内容：', full_history),
        connection_id => 'your-gcp-project-id.us.my-connection',
        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>

反馈¶

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