企业官网建设流程全解析

用 ADK Data Agent 工具集成 Gemini Conversational Analytics：自然语言数据查询、Agent 生命周期管理与图表生成实战

【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python

导读

本文基于 ADK 官方示例 contributing/samples/integrations/data_agent 及其配套源码，讲解如何在 ADK 智能体中接入 Google Cloud 的Data Agent（数据代理）能力：通过google.adk.tools.data_agent模块提供的第一方工具，让 Agent 能够用自然语言列出、查看、对话乃至创建、修改、删除 Data Agent，并借助有状态会话实现多轮追问。读完本文，你将掌握 DataAgentToolset 的完整配置方法、六种工具的调用契约、三种凭据接入方式，以及如何结合图表生成工具构建一个开箱即用的数据问答 Agent。

一、Data Agent 工具能做什么

Data Agent 是 Google Cloud 提供的对话式数据分析能力（Conversational Analytics）：它指向你的 BigQuery 表或其他数据源，接受自然语言问题并返回 SQL、检索结果与最终回答。ADK 的google.adk.tools.data_agent模块把这些能力封装成标准工具，让 ADK Agent 可以直接调度它们：

• 列出你有权限访问的 Data Agent；
• 查看某个 Data Agent 的详细信息（数据源引用、系统指令等）；
• 对话：用自然语言向指定 Data Agent 提问；
• 创建 / 删除 / 更新Data Agent 资源（实验性，需显式开启）。

一个关键特性是有状态会话：在同一个 session 内可以连续追问（如"上季度我的前 3 名客户是谁？"→"那再前一个季度呢？"），Agent 会维持上下文，无需反复交代背景。

模块的公开 API 入口见 src/google/adk/tools/data_agent/init.py，导出了三个核心类：DataAgentCredentialsConfig、DataAgentToolConfig、DataAgentToolset。

二、架构与调用链：从 Agent 到 Gemini Data Analytics API

从源码结构看，Data Agent 工具的调用链分为四层：

1. Agent 层：示例 Agent 定义 将DataAgentToolset实例直接放入tools列表，同时混入generate_chart和load_artifacts等自定义工具；
2. Toolset 层：data_agent_toolset.py 的DataAgentToolset.get_tools()按配置装配工具——默认装配 3 个只读工具，仅当enable_data_agent_modification=True时才追加 3 个写工具，每个工具都被包装为GoogleTool（google_tool.py），从而复用 ADK 的 Google API 凭据机制；
3. 工具函数层：data_agent_tool.py 定义 6 个工具函数，负责参数校验、构造请求、调用 Gemini Data Analytics REST API（端点默认geminidataanalytics.googleapis.com/v1）；
4. 流式处理层：ask_data_agent通过_gda_stream_util.get_stream以流式方式消费会话回复，并用DataAgentToolConfig.max_query_result_rows限制返回行数。

每个工具函数返回统一的字典结构：{"status": "SUCCESS"|"ERROR", "response": ...}或{"status": "ERROR", "error_details": ...}，方便 Agent 在指令中约定"根据 status 判断成败"。

三、前置条件

运行该示例前需要准备：

1. 一个已启用的 Google Cloud 项目：需要开启 BigQuery 和 Gemini API（官方指引中同时要求启用 Conversational Analytics 相关 API 并按文档配置 IAM 权限与数据源认证，本文不再展开外部文档细节）。
2. 配置 Application Default Credentials（ADC）：

   gcloud auth application-default login

3. 至少一个已创建的 Data Agent：可以通过 Conversational Analytics API、其 Python SDK，或直接在 BigQuery Studio 中创建。这些 Agent 在 Google Cloud 控制台配置，指向你的 BigQuery 表或其他数据源。
4. 按官方 Setup 指南完成 API 启用与 IAM 权限配置，确保数据源可被 Data Agent 访问。

四、六种工具详解

原文档列出的 6 个工具在 data_agent_tool.py 中实现，签名与要点如下：

4.1 资源命名与参数校验

所有 Agent 均使用资源全名定位，格式为：

projects/{project}/locations/{location}/dataAgents/{agent}

例如示例提示词中的projects/my-project/locations/global/dataAgents/sales-agent-123。工具内部用正则_DATA_AGENT_NAME_RE校验该格式，project、location、agent 三个路径段只允许字母数字以及-、_、.（见 data_agent_tool.py 中的_validate_path_segment）。格式错误会直接返回{"status": "ERROR", "error_details": ...}。

4.2 查询返回结构

ask_data_agent的返回是一个步骤列表（steps），每个步骤是包含不同键的字典，典型流程如下（源码 docstring 示例）：

{ "status": "SUCCESS", "response": [ {"text": {"parts": ["Analyzing context", "Retrieved context for 1 table."], "textType": "THOUGHT"}}, {"data": {"generatedSql": "SELECT AVG(SAFE_CAST(street_trees.dbh AS FLOAT64)) AS average_height FROM bigquery-public-data.san_francisco.street_trees AS street_trees;"}}, {"Data Retrieved": {"headers": ["average_height"], "rows": [[10.073475670972512]], "summary": "Showing all 1 rows."}}, {"text": {"parts": ["### Summary\nBased on the street tree data for San Francisco, the average height ... is approximately 10.07."], "textType": "FINAL_RESPONSE"}} ] }

可见返回中既包含模型的思考（textType: THOUGHT）、生成的 SQL（generatedSql）、检索到的数据表格，也有最终回答（textType: FINAL_RESPONSE）。有状态会话的底层实现是：ask_data_agent先调用_get_data_agent_info拿到 Agent 信息，再向{resource_parent}:chat端点发起带clientIdEnum=GOOGLE_ADK的流式请求（见 data_agent_tool.py），同一 ADK session 内的连续提问即构成上下文延续。

4.3 写操作与长任务轮询

create/update/delete三个写工具都会发起一个长期运行操作（LRO），并在data_agent_modification_timeout_seconds（默认 60 秒）内以data_agent_modification_poll_interval_seconds（默认 2 秒）为间隔轮询GET /operations/{name}，直到done为止；可重试的 HTTP 状态码（429/500/502/503/504）会自动重试。注意：轮询超时并不代表操作失败——操作可能仍在后台执行，超时响应中会附带operation_name供后续查询，源码注释明确提示"不要重试该操作"。

update_data_agent有一个防误删保护：update_mask中列出的每个字段必须同时出现在agent_config中，否则返回错误，避免因遗漏字段导致 API 清空未提及的属性（见_mask_field_present逻辑）。

五、DataAgentToolConfig 配置详解

工具行为通过DataAgentToolConfig（config.py）控制，这是一个 pydantic 模型，extra="forbid"表示不认识的字段会直接报错：

六、凭据接入：三种认证方式

DataAgentCredentialsConfig（credentials.py）封装凭据配置，默认 OAuth scope 为https://www.googleapis.com/auth/bigquery，并使用data_agent_token_cache作为令牌缓存键。示例 agent.py 通过CREDENTIALS_TYPE变量演示了三种接入方式：

CREDENTIALS_TYPE = None # 默认使用 ADC if CREDENTIALS_TYPE == AuthCredentialTypes.OAUTH2: # 交互式 OAuth2，需设置 OAUTH_CLIENT_ID / OAUTH_CLIENT_SECRET 环境变量 credentials_config = DataAgentCredentialsConfig( client_id=os.getenv("OAUTH_CLIENT_ID"), client_secret=os.getenv("OAUTH_CLIENT_SECRET"), ) elif CREDENTIALS_TYPE == AuthCredentialTypes.SERVICE_ACCOUNT: # 服务账号密钥文件，需替换为你的 key 文件路径 creds, _ = google.auth.load_credentials_from_file( "service_account_key.json", scopes=["https://www.googleapis.com/auth/cloud-platform"], ) creds.refresh(google.auth.transport.requests.Request()) credentials_config = DataAgentCredentialsConfig(credentials=creds) else: # Application Default Credentials（推荐本地开发） application_default_credentials, _ = google.auth.default() if not application_default_credentials.valid: application_default_credentials.refresh( google.auth.transport.requests.Request() ) credentials_config = DataAgentCredentialsConfig( credentials=application_default_credentials )

• ADC（默认）：本地开发最省事，配合gcloud auth application-default login即可；
• OAuth2：适合需要交互式授权页面的场景；
• Service Account：适合 CI/服务器等无交互环境。

七、组合 Toolset 与图表生成：示例 Agent 全解

示例 Agent 将 Data Agent 工具集与自定义工具组合，形成一个完整的"查询 + 可视化"智能体（完整代码见 agent.py）：

tool_config = DataAgentToolConfig( max_query_result_rows=100, # 每查询最多返回 100 行 enable_data_agent_modification=True, # 允许创建/更新/删除 ) da_toolset = DataAgentToolset( credentials_config=credentials_config, data_agent_tool_config=tool_config, tool_filter=[ "list_accessible_data_agents", "get_data_agent_info", "ask_data_agent", "create_data_agent", "delete_data_agent", "update_data_agent", ], ) root_agent = Agent( name="data_agent", description="Agent to answer user questions using Data Agents and generate charts.", instruction=( "## Persona\nYou are a helpful assistant that uses Data Agents" " to answer user questions about their data.\n\n" "## Tools\n- You can list available data agents using `list_accessible_data_agents`.\n" "- You can get information about a specific data agent using `get_data_agent_info`.\n" "- You can chat with a specific data agent using `ask_data_agent`.\n" "- You can create/delete/update data agents using the corresponding tools.\n" "- `generate_chart` renders professional charts from a `chart_spec` (Vega-Lite JSON).\n" "- You can load artifacts using `load_artifacts`.\n" ), tools=[da_toolset, generate_chart, load_artifacts], )

几点值得注意的实现细节：

• tool_filter语义：与基类BaseToolset不同，DataAgentToolset在tool_filter为空列表时不会装配任何工具（见 data_agent_toolset.py）；且即使过滤列表里写了写工具，只要enable_data_agent_modification=False，这些工具依然不会被创建（见get_tools的装配逻辑，以及 test_data_agent_toolset.py 中test_data_agent_toolset_tools_selective_modification_disabled的验证）。
• 图表工具：generate_chart接收 Vega-Lite JSON 规格，通过 Altair +vl-convert渲染成 PNG 并调用tool_context.save_artifact("chart.png", ...)保存为 artifact。需要额外安装：pip install altair vl-convert-python。Agent 指令要求"需要可视化时使用它，不要向用户展示原始 JSON"。
• 工具装配的默认行为：get_tools()默认只创建 3 个只读工具；开启修改开关后共 6 个工具——这一点被测试test_data_agent_toolset_tools_default（3 个）与test_data_agent_toolset_tools_with_mutation_enabled（6 个）精确断言。

八、运行方式与示例提示词

1. 进入 ADK 仓库根目录；
2. 使用 ADK CLI 运行示例：

   adk run contributing/samples/integrations/data_agent

3. CLI 进入交互模式后即可提问。

原文档给出的四组示例提示词，覆盖了"查列表 → 查详情 → 有状态追问 → 创建资源"的完整链路：

• "List accessible data agents."—— 列出可访问的 Data Agent；
• "Using agent projects/my-project/locations/global/dataAgents/sales-agent-123, who were my top 3 customers last quarter?"—— 指定 Agent 做具体分析；
• "How does that compare to the quarter before?"——无需重复指定 Agent，直接追问上一季度对比（有状态会话的体现）；
• "Create a new data agent named my-new-agent."—— 触发create_data_agent写操作（需已开启修改开关）。

九、测试验证：行为契约有据可依

仓库在 tests/unittests/tools/data_agent/ 提供了两层测试：

• Toolset 层（test_data_agent_toolset.py）：断言默认装配 3 个只读工具、开启修改后装配 6 个、tool_filter白名单过滤、未知工具名被忽略、修改未开启时写工具即使列入 filter 也不出现；
• 工具函数层（test_data_agent_tool.py）：mockget_gda_session/get_gda_endpoint验证list_accessible_data_agents的请求 URL（/v1/projects/{project}/locations/{location}/dataAgents:listAccessible）、请求头（X-Goog-API-Client: GOOGLE_ADK）以及异常路径返回ERROR字典等行为。

这些测试同时是学习工具契约（参数、返回结构、错误格式）的绝佳参考。

十、小结

• google.adk.tools.data_agent将 Gemini Conversational Analytics 的 Data Agent 能力封装为 6 个第一方工具，通过DataAgentToolset统一装配并复用 ADK 的 Google 凭据体系；
• 默认只读安全：只有显式设置enable_data_agent_modification=True才会暴露创建/更新/删除能力；
• ask_data_agent天然支持有状态多轮追问，返回结构包含思考、SQL、数据与最终回答，配合max_query_result_rows控制数据量；
• 结合自定义generate_chart（Altair + vl-convert）可将查询结果直接渲染为图表 artifact，形成"自然语言提问 → 数据分析 → 可视化"的完整闭环；
• 上述所有行为均有源码与测试佐证，可放心在此基础上扩展你的数据问答 Agent。

【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python