BigQuery 智能体分析插件¶

Supported in ADKPython v1.19.0Preview

BigQuery 智能体分析插件通过提供强大的深度智能体行为分析解决方案，显著增强了智能体开发工具包 (ADK)。它使用 ADK 插件架构和 BigQuery Storage Write API，将关键的操作事件直接捕获并记录到 Google BigQuery 表中，为你提供调试、实时监控和全面离线性能评估的高级能力。

预览版本

BigQuery 智能体分析插件处于预览版本。有关更多信息,请参阅发布阶段说明。

BigQuery Storage Write API

此功能使用 BigQuery Storage Write API,这是一项付费服务。有关费用信息,请参阅 BigQuery 文档。

用例¶

智能体工作流调试和分析： 将各种插件生命周期事件(LLM 调用、工具使用) 和智能体产生的事件(用户输入、模型响应) 捕获到定义良好的模式中。
大容量分析和调试： 日志操作在单独的线程中异步执行，以避免阻塞主智能体执行。该插件旨在处理大量事件，通过时间戳保留事件顺序。

记录的智能体事件数据根据 ADK 事件类型而有所不同。有关更多信息，请参阅事件类型和负载。

先决条件¶

Google Cloud 项目,已启用 BigQuery API。
BigQuery 数据集： 在使用插件之前创建一个数据集来存储日志表。如果表不存在，插件会自动在数据集中创建必要的事件表。默认情况下，此表名为 agent_events，你可以在插件配置中使用 table_id 参数自定义此名称。
身份验证：
- 本地： 运行 gcloud auth application-default login。
- 云端： 确保你的服务账号具有所需的权限。

IAM 权限¶

为了使智能体正常工作，运行智能体的主体 (例如服务账号、用户账号) 需要以下 Google Cloud 角色： * 项目级别的 roles/bigquery.jobUser,用于在你的项目中运行 BigQuery 查询。此角色本身不授予对任何数据的访问权限。 * 表级别的 roles/bigquery.dataEditor,用于将日志/事件数据写入你选择的 BigQuery 表。如果你需要智能体创建此表，则需要在要创建表的 BigQuery 数据集上授予 roles/bigquery.dataEditor 角色。

与智能体一起使用¶

你可以通过配置 BigQuery 分析插件并将其注册到 ADK 智能体的 App 对象来使用它。以下示例展示了启用此插件和 BigQuery 工具的智能体实现：

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
from google.adk.agents import Agent
from google.adk.models.google_llm import Gemini
from google.adk.tools.bigquery import BigQueryToolset, BigQueryCredentialsConfig

# --- 配置 ---
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", "your-gcp-project-location") # 使用你的 Google Cloud 项目的位置

if PROJECT_ID == "your-gcp-project-id":
    raise ValueError("Please set GOOGLE_CLOUD_PROJECT or update the code.")
if DATASET_ID == "your-big-query-dataset-id":
    raise ValueError("Please set BIG_QUERY_DATASET_ID or update the code.")
if LOCATION == "your-gcp-project-location":
    raise ValueError("Please set GOOGLE_CLOUD_LOCATION or update the code.")

# --- 关键：在 Gemini 实例化之前设置环境变量 ---
os.environ['GOOGLE_CLOUD_PROJECT'] = PROJECT_ID
os.environ['GOOGLE_CLOUD_LOCATION'] = LOCATION
os.environ['GOOGLE_GENAI_USE_VERTEXAI'] = 'True' # 确保你已启用 Vertex AI API

# --- 初始化插件 ---
bq_logging_plugin = BigQueryAgentAnalyticsPlugin(
    project_id=PROJECT_ID, # project_id 是用户的必需输入
    dataset_id=DATASET_ID, # dataset_id 是用户的必需输入
    table_id="agent_events" # 可选：默认为 "agent_events"。如果表不存在，插件会自动创建此表。
)

# --- 初始化工具和模型 ---
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="You are a helpful assistant with access to BigQuery tools.",
    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;

配置选项¶

你可以使用 BigQueryLoggerConfig 自定义插件。

enabled (bool,默认值：True): 要禁用插件将智能体数据记录到 BigQuery 表，请将此参数设置为 False。
event_allowlist (Optional[List[str]],默认值：None): 要记录的事件类型列表。如果为 None,则记录所有事件，除了 event_denylist 中的事件。有关支持的事件类型的完整列表，请参阅事件类型和负载部分。
event_denylist (Optional[List[str]],默认值：None): 要跳过记录的事件类型列表。有关支持的事件类型的完整列表，请参阅事件类型和负载部分。
content_formatter (Optional[Callable[[Any], str]],默认值：None): 用于在记录之前格式化事件内容的可选函数。以下代码说明了如何实现内容格式化程序。
shutdown_timeout (float,默认值：5.0): 关闭期间等待日志刷新的秒数。
client_close_timeout (float,默认值：2.0): 等待 BigQuery 客户端关闭的秒数。
max_content_length (int,默认值：500): 截断前内容部分的最大长度。

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

import json
import re

from google.adk.plugins.bigquery_agent_analytics_plugin import BigQueryLoggerConfig

def redact_dollar_amounts(event_content: Any) -> str:
    """
    自定义格式化程序，用于隐藏美元金额 (例如 $600、$12.50)
    并确保如果输入是字典则输出 JSON。
    """
    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 秒以刷新日志
    client_close_timeout=2.0, # 等待最多 2 秒以关闭 BQ 客户端
    max_content_length=500, # 将内容截断为 500 个字符 (默认)
    content_formatter=redact_dollar_amounts, # 隐藏日志内容中的美元金额

)

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

Schema 和生产环境设置¶

如果表不存在，插件会自动创建表。但是，对于生产环境，我们建议手动创建带有分区和聚类的表，以优化性能和成本。

推荐的 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="与当前会话关联的用户标识符。"),
  content STRING OPTIONS(description="事件特定数据(负载)。格式因 event_type 而异。"),
  error_message STRING OPTIONS(description="如果在处理事件期间发生错误则填充。"),
  is_truncated BOOLEAN OPTIONS(description="布尔标志,指示内容字段是否因大小限制而被截断。")
)
PARTITION BY DATE(timestamp)
CLUSTER BY event_type, agent, user_id;

事件类型和负载¶

content 列包含特定于 event_type 的格式化字符串。下表描述了这些事件和相应的内容。

Note

所有可变内容字段 (例如用户输入、模型响应、工具参数、系统提示)
都会被截断为 max_content_length 个字符
(在 BigQueryLoggerConfig 中配置，默认为 500) 以管理日志大小。

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

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

事件类型 触发条件 内容格式逻辑 示例内容

LLM_REQUEST

before_model_callback

Model: {model} | Prompt: {prompt} | System Prompt: Model: {model} | Prompt: {formatted_contents} | System Prompt: {system_prompt} | Params: {params} | Available Tools: {tool_names}

Model: gemini-2.5-flash | Prompt: user: Model: gemini-flash-2.5| Prompt: user: text: 'Hello'| System Prompt: You are a helpful assistant. | Params: {temperature=1.0} | Available Tools: ['bigquery_tool']

LLM_RESPONSE

after_model_callback

如果是工具调用：

Tool Name: {func_names} | Token
Usage: {usage}

如果是文本：

Tool Name: text_response, text: '{text}' | Token Usage:
{usage}

Tool Name: text_response, text: 'Here is the data.' | Token Usage: {prompt: 10, candidates: 5, total: 15}

LLM_ERROR

on_model_error_callback

None(错误详情在 error_message column)

None

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

这些事件追踪智能体执行的工具。

事件类型 触发条件 内容格式逻辑 示例内容

TOOL_STARTING

before_tool_callback

Tool Name: {name}, Description: {desc}, Arguments: {args}

Tool Name: list_datasets, Description: Lists datasets..., Arguments: {'project_id': 'my-project'}

TOOL_COMPLETED

after_tool_callback

Tool Name: {name}, Result: {result}

Tool Name: list_datasets, Result: ['dataset_1', 'dataset_2']

TOOL_ERROR

on_tool_error_callback

Tool Name: {name}, Arguments: {args}(错误详情在 error_message)

Tool Name: list_datasets, Arguments: {}

智能体生命周期 (插件生命周期)¶

这些事件追踪智能体执行的开始和结束，包括子智能体。

事件类型

触发条件

内容格式逻辑

示例内容

INVOCATION_STARTING

before_run_callback

None

None

INVOCATION_COMPLETED

after_run_callback

None

None

AGENT_STARTING

before_agent_callback

Agent Name: {agent_name}

Agent Name: sub_agent_researcher

AGENT_COMPLETED

after_agent_callback

Agent Name: {agent_name}

Agent Name: sub_agent_researcher

用户和通用事件 (事件流)¶

这些事件源自智能体或运行器产生的 Event 对象。

事件类型 触发条件 内容格式逻辑 示例内容

USER_MESSAGE_RECEIVED

on_user_message_callback

User Content: {formatted_message}

User Content: text: 'Show me the sales data.'

TOOL_CALL

event.get_function_calls() is true

call: {func_name}

call: list_datasets

TOOL_RESULT

event.get_function_responses() is true

resp: {func_name}

resp: list_datasets

MODEL_RESPONSE

event.content has parts

text: '{text}'

text: 'I found 2 datasets.'

高级分析查询¶

以下示例查询演示了如何从 BigQuery 中记录的 ADK 智能体事件分析数据中提取信息。你可以使用 BigQuery 控制台运行这些查询。

在执行这些查询之前，请确保更新提供的 SQL 中的 GCP 项目 ID、BigQuery 数据集 ID 和表 ID(如果未指定，则默认为"agent_events")。

追踪特定对话轮次

SELECT timestamp, event_type, agent, content
FROM `your-gcp-project-id.your-dataset-id.agent_events`
WHERE invocation_id = 'your-invocation-id'
ORDER BY timestamp ASC;

每日调用量

SELECT DATE(timestamp) as log_date, COUNT(DISTINCT invocation_id) as count
FROM `your-gcp-project-id.your-dataset-id.agent_events`
WHERE event_type = 'INVOCATION_STARTING'
GROUP BY log_date ORDER BY log_date DESC;

Token 使用分析

SELECT
  AVG(CAST(REGEXP_EXTRACT(content, r"Token Usage:.*total: ([0-9]+)") AS INT64)) as avg_tokens
FROM `your-gcp-project-id.your-dataset-id.agent_events`
WHERE event_type = 'LLM_RESPONSE';

错误监控

SELECT timestamp, event_type, error_message
FROM `your-gcp-project-id.your-dataset-id.agent_events`
WHERE error_message IS NOT NULL
ORDER BY timestamp DESC LIMIT 50;

BigQuery 智能体分析插件¶

用例¶

先决条件¶

IAM 权限¶

与智能体一起使用¶

运行和测试智能体¶

配置选项¶

Schema 和生产环境设置¶

事件类型和负载¶

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

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

智能体生命周期 (插件生命周期)¶

用户和通用事件 (事件流)¶

高级分析查询¶

其他资源¶