用于 ADK 的 Google Cloud Data Agents 工具¶

Supported in ADKPython v1.23.0

这是一组旨在与 Conversational Analytics API 驱动的数据智能体集成的工具。数据智能体是由 AI 驱动的智能体，可以帮助你使用自然语言分析数据。在配置数据智能体时，你可以从受支持的数据源中选择，包括 BigQuery、Looker 和 Looker Studio。 DataAgentToolset 默认包含以下只读工具：

list_accessible_data_agents：列出你在指定 Google Cloud 项目中有权访问的数据智能体。支持可选的 location 覆盖，以及自动或手动分页（page_size 和 page_token）。
get_data_agent_info：根据完整资源名称（projects/{project}/locations/{location}/dataAgents/{agent}）获取特定数据智能体的详情和已发布的上下文。
ask_data_agent：向特定数据智能体发送自然语言问题并返回其回答。

如果你在 DataAgentToolConfig 中设置了 enable_data_agent_modification=True，该工具集还会包含以下工具：

create_data_agent：在 Google Cloud 项目中，使用给定的 data_agent_id 从遵循 DataAgent 资源 schema 的 JSON agent_config 创建新的数据智能体。location 参数是可选的。
update_data_agent：根据 JSON agent_config 和以逗号分隔的 camelCase 字段名列表 update_mask（例如 displayName,description）更新现有数据智能体。update_mask 中列出的每个字段也必须出现在 agent_config 中。
delete_data_agent：根据完整资源名称删除现有数据智能体。

这些修改工具会等待底层的长时间运行操作完成，最长等待 data_agent_modification_timeout_seconds。

前提条件¶

在使用这些工具之前，请在 Google Cloud 中完成以下步骤：

在你的 Google Cloud 项目中启用 Gemini Data Analytics API（geminidataanalytics.googleapis.com）。
确保工具集使用的凭据具备数据智能体及其底层数据源所需的 IAM 权限。有关将你的智能体连接到 Google Cloud 的更多信息，请参阅连接到 Google Cloud 和 Agent Platform 指南。
get_data_agent_info 和 ask_data_agent 工具需要一个已存在的数据智能体。你可以使用 create_data_agent（当 enable_data_agent_modification=True 时）创建一个，或按照以下任一指南操作：

身份验证¶

DataAgentToolset 需要一个 DataAgentCredentialsConfig，并支持多种身份验证机制。你必须提供 credentials、external_access_token_key，或一对 client_id 和 client_secret。默认情况下，DataAgentCredentialsConfig 使用 https://www.googleapis.com/auth/bigquery OAuth 作用域，你可以在配置 OAuth 客户端凭据时通过 scopes 覆盖它。

Experimental

DataAgentCredentialsConfig 类继承自 BaseGoogleCredentialsConfig，后者是实验性的，不应用于生产项目。

应用程序默认凭据¶

对于本地开发以及在 Google Cloud 服务（如 Cloud Run 和 GKE）上运行，你应该使用这种方式。

import google.auth
from google.adk.tools.data_agent import DataAgentToolset, DataAgentCredentialsConfig

# 加载应用程序默认凭据
credentials, project_id = google.auth.default()

# 配置工具集
credentials_config = DataAgentCredentialsConfig(credentials=credentials)
data_agent_toolset = DataAgentToolset(credentials_config=credentials_config)

服务账号¶

你可以显式地提供服务账号文件或信息。

from google.oauth2 import service_account
from google.adk.tools.data_agent import DataAgentToolset, DataAgentCredentialsConfig

# 加载服务账号凭据
credentials = service_account.Credentials.from_service_account_file('path/to/key.json')

# 配置工具集
credentials_config = DataAgentCredentialsConfig(credentials=credentials)
data_agent_toolset = DataAgentToolset(credentials_config=credentials_config)

外部访问令牌¶

对于需要代表最终用户执行操作的应用，你可以直接使用访问令牌（例如来自 OAuth2 流程或外部 IDP）实例化用户凭据并传入。

from google.oauth2.credentials import Credentials
from google.adk.tools.data_agent import DataAgentToolset, DataAgentCredentialsConfig

# 假设 'user_token' 是通过外部 OAuth 流程获得的
credentials = Credentials(token=user_token)

# 配置工具集
credentials_config = DataAgentCredentialsConfig(credentials=credentials)
data_agent_toolset = DataAgentToolset(credentials_config=credentials_config)

外部身份验证提供方¶

如果你要与令牌由平台管理的外部身份验证提供方（例如 Gemini Enterprise）集成，请使用 external_access_token_key。

from google.adk.tools.data_agent import DataAgentToolset, DataAgentCredentialsConfig

# 用于在会话状态中查找访问令牌的键
credentials_config = DataAgentCredentialsConfig(
    external_access_token_key="YOUR_AUTH_ID"
)
data_agent_toolset = DataAgentToolset(credentials_config=credentials_config)

交互式身份验证（ADK Web）¶

在使用 adk web 界面进行交互式会话时，你可以提供 OAuth 2.0 客户端凭据来触发登录流程。这种方式既适用于本地开发，也适用于将你的 ADK 智能体部署到 Cloud Run 等环境的情况。

from google.adk.tools.data_agent import DataAgentToolset, DataAgentCredentialsConfig

# 提供 OAuth 2.0 客户端 ID 和密钥
credentials_config = DataAgentCredentialsConfig(
    client_id="YOUR_CLIENT_ID",
    client_secret="YOUR_CLIENT_SECRET"
)
data_agent_toolset = DataAgentToolset(credentials_config=credentials_config)

配置¶

你可以使用 DataAgentToolConfig 自定义工具行为：

max_query_result_rows（int，默认值：50）：ask_data_agent 对每个数据结果返回的最大行数。
location（str | None，默认值：None）：默认的 Google Cloud 位置（例如 global、us 或 eu），用于选择 API 端点。接受 location 参数的工具在未设置该参数时使用此值，否则回退到 global。其他工具使用数据智能体资源名称中的位置，但 ask_data_agent 除外——若设置了此值，它会使用此值。
api_endpoint（str | None，默认值：None）：Conversational Analytics API 请求的可选自定义 API 端点。如果提供，它将覆盖默认的或由位置推导出的 API 端点。
enable_data_agent_modification（bool，默认值：False）：为 True 时，工具集还会包含 create_data_agent、update_data_agent 和 delete_data_agent。为 False 时，工具集为只读。
data_agent_modification_timeout_seconds（int，默认值：60）：轮询长时间运行的创建、更新或删除操作时的总超时时间（秒）。必须大于 0。
data_agent_modification_poll_interval_seconds（int，默认值：2）：等待创建、更新或删除操作完成时的轮询间隔（秒）。必须大于 0。

请谨慎使用

设置 enable_data_agent_modification=True 会让智能体能够在你的 Google Cloud 项目中创建、更新和删除数据智能体。请确保工具集使用的凭据被限制在已授权的项目内，并且只具备最低限度的必要 IAM 权限。你也可以向 DataAgentToolset 传入 tool_filter，只暴露特定工具（例如排除 delete_data_agent）。

import google.auth
from google.adk.tools.data_agent import DataAgentToolset, DataAgentCredentialsConfig
from google.adk.tools.data_agent.config import DataAgentToolConfig

credentials, _ = google.auth.default()
credentials_config = DataAgentCredentialsConfig(credentials=credentials)

tool_config = DataAgentToolConfig(
    max_query_result_rows=100,
    enable_data_agent_modification=True,
    data_agent_modification_timeout_seconds=120,
)
data_agent_toolset = DataAgentToolset(
    credentials_config=credentials_config,
    data_agent_tool_config=tool_config,
)

示例代码¶

以下示例代码演示了如何在 ADK 智能体中使用应用程序默认凭据（ADC）来使用 DataAgentToolset。

import asyncio

from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools.data_agent import DataAgentCredentialsConfig
from google.adk.tools.data_agent import DataAgentToolset
from google.adk.tools.data_agent.config import DataAgentToolConfig
from google.genai import types
import google.auth

# Define constants for this example agent
AGENT_NAME = "data_agent_example"
APP_NAME = "data_agent_app"
USER_ID = "user1234"
SESSION_ID = "1234"
GEMINI_MODEL = "gemini-flash-latest"

# Define tool configuration
# Set `enable_data_agent_modification=True` to also expose `create_data_agent`,
# `update_data_agent`, and `delete_data_agent`.
tool_config = DataAgentToolConfig(
    max_query_result_rows=100,
    enable_data_agent_modification=False,
)

# Use Application Default Credentials (ADC)
# https://cloud.google.com/docs/authentication/provide-credentials-adc
application_default_credentials, _ = google.auth.default()
credentials_config = DataAgentCredentialsConfig(
    credentials=application_default_credentials
)

# Instantiate a Data Agent toolset
da_toolset = DataAgentToolset(
    credentials_config=credentials_config,
    data_agent_tool_config=tool_config,
)

# Agent Definition
data_agent = Agent(
    name=AGENT_NAME,
    model=GEMINI_MODEL,
    description="Agent to answer user questions using data agents.",
    instruction=(
        "## Persona\nYou are a helpful assistant that uses data agents"
        " to answer user questions about their data.\n\n"
    ),
    tools=[da_toolset],
)


# Session and Runner
async def setup_session_and_runner():
    session_service = InMemorySessionService()
    await session_service.create_session(
        app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID
    )
    return Runner(
        agent=data_agent, app_name=APP_NAME, session_service=session_service
    )


# Agent Interaction
async def call_agent_async(runner, query):
    """
    Helper function to call the agent with a query.
    """
    content = types.Content(role="user", parts=[types.Part(text=query)])
    events = runner.run_async(
        user_id=USER_ID, session_id=SESSION_ID, new_message=content
    )

    print("USER:", query)
    async for event in events:
        if event.is_final_response():
            final_response = event.content.parts[0].text
            print("AGENT:", final_response)


async def main():
    runner = await setup_session_and_runner()
    # Replace `<PROJECT_ID>` with your Google Cloud project ID, and replace
    # `<DATA_AGENT_NAME>` with a full resource name in the format:
    # `projects/{project}/locations/{location}/dataAgents/{agent}`
    await call_agent_async(
        runner, "List accessible data agents in project <PROJECT_ID>."
    )
    await call_agent_async(runner, "Get information about <DATA_AGENT_NAME>.")
    # The data agent in this example is configured with the BigQuery table:
    # `bigquery-public-data.san_francisco.street_trees`
    await call_agent_async(
        runner, "Ask <DATA_AGENT_NAME> to count the rows in the table."
    )
    await call_agent_async(runner, "What are the columns in the table?")
    await call_agent_async(runner, "What are the top 5 tree species?")
    await call_agent_async(
        runner, "For those species, what is the distribution of legal status?"
    )

注意：如果你想把 BigQuery 表和数据集直接作为工具来查询，请参阅 BigQuery tool for ADK。