Google 会使用 AI 技术将内容翻译成您偏好的语言。AI 翻译可能包含错误。

使用 A2A 编排数据代理

Google Cloud 中的 Conversational Analytics API 实现了开放的 Agent-to-Agent (A2A) 协议，该协议可让多智能体工作流中的智能体发现功能、委托分析查询，以及以流式传输结构化响应，例如可执行的 SQL 查询和图表可视化。

您可以通过在 API 请求中传递数据集上下文，查询内置于 Conversational Analytics API for BigQuery 和 Looker 中的数据代理，也可以查询根据贵组织的业务逻辑配置的自定义数据代理。

如需直接构建和查询数据代理，请参阅使用 Python SDK 构建数据代理或使用 HTTP 构建数据代理。

了解 Gemini for Google Cloud 如何以及何时使用您的数据。

数据智能体编排的工作原理

将 Conversational Analytics API 数据智能体集成到应用或多智能体系统中时，编排工作流会遵循以下操作：

在委派分析查询之前，编排器代理会通过检查数据代理的代理卡片来发现数据代理的功能和技能。
编排器代理会发送消息，通过在请求中指定数据源来查询内置数据代理（agents/bigquery-ca 或 agents/looker-ca），或者查询配置了网域业务逻辑的自定义数据代理 (dataAgents/DATA_AGENT_ID)。
数据代理会处理请求、执行所需的查询，并返回结果，可以是完整的回答，也可以通过流式传输推理进度和结构化制品（例如可执行的 SQL 和 Vega-Lite 图表规范）来返回结果。

准备工作

在开始之前，请完成以下前提条件：

在您的 Google Cloud 项目中启用 Conversational Analytics API、BigQuery API 和 Looker API。
验证您是否拥有所需的 IAM 角色和权限。
针对 Conversational Analytics API 进行身份验证，并安装客户端库或获取授权令牌。

所需的角色

如需获得通过 A2A 发现和查询数据代理所需的权限，请让管理员为您授予项目的以下 IAM 角色：

Gemini Data Analytics Data Agent User (roles/geminidataanalytics.dataAgentUser)
对于无状态查询：Gemini Data Analytics Data Agent Stateless User (roles/geminidataanalytics.dataAgentStatelessUser)

如需详细了解如何授予角色，请参阅管理对项目、文件夹和组织的访问权限。

您也可以通过自定义角色或其他预定义角色来获取所需的权限。

如需查询底层数据源，您还必须拥有对目标 BigQuery 数据集（例如 roles/bigquery.dataViewer）或 Looker 探索的读取权限。

探索智能体功能

在将查询委托给数据代理之前，编排代理或客户端应用可以检查代理卡，以查看其功能和配置，例如其说明、可用技能和受支持的扩展程序。您可以使用 getCard 方法检索内置数据代理（agents/bigquery-ca 和 agents/looker-ca）和自定义数据代理 (dataAgents/DATA_AGENT_ID) 的代理卡片。

检索智能体卡片

以下代码示例展示了如何检索代理卡片。这些示例以内置 BigQuery 数据代理 (agents/bigquery-ca) 为例，但您可以通过更改请求中的代理资源名称来检索内置 Looker 代理 (agents/looker-ca) 或自定义数据代理 (dataAgents/DATA_AGENT_ID) 的卡片：

Python SDK

from google.cloud import geminidataanalytics_v1

client = geminidataanalytics_v1.DataA2AServiceClient()

agent_name = "projects/PROJECT_ID/locations/LOCATION/agents/bigquery-ca"

request = geminidataanalytics_v1.GetAgentCardRequest(tenant=agent_name)
card = client.get_agent_card(request=request)

print(card)

在上面的示例中，按如下所示替换值：

PROJECT_ID：您的 Google Cloud 项目的 ID
LOCATION：代理资源的位置（例如 us、us-east4、eu 或 global）

HTTP

curl -X GET \
  -H "Authorization: Bearer $(gcloud auth application-default print-access-token)" \
  "https://geminidataanalytics.googleapis.com/v1/a2a/projects/PROJECT_ID/locations/LOCATION/agents/bigquery-ca/v1/card"

在上面的示例中，按如下所示替换值：

PROJECT_ID：您的 Google Cloud 项目的 ID
LOCATION：代理资源的位置（例如 us、us-east4、eu 或 global）

了解代理卡片结构

如果请求成功，则会返回一个包含元数据、支持的技能和扩展程序的智能体卡片对象：

{
  "name": "BigQuery Conversational Analytics Agent",
  "description": "This agent can answer questions about your data using BigQuery.",
  "protocolVersion": "1.0",
  "skills": [
    {
      "id": "data-analysis",
      "name": "Data Analysis",
      "description": "Provides data analysis assistance",
      "examples": [
        "What is the total sales for the last 3 months?"
      ],
      "inputModes": [
        "text/plain"
      ],
      "outputModes": [
        "text/plain",
        "application/json"
      ]
    }
  ],
  "capabilities": {
    "streaming": true,
    "extensions": [
      {
        "uri": "https://docs.cloud.google.com/gemini/docs/conversational-analytics-api/reference/a2a/extensions/bigquery_context/v1",
        "description": "Google Data Analytics BigQuery Context extension"
      }
    ]
  },
  "defaultInputModes": [
    "text/plain"
  ],
  "defaultOutputModes": [
    "text/plain",
    "application/json"
  ]
}

代理卡片包含以下字段：

name：数据代理的显示名称
description：数据智能体分析功能的摘要
protocolVersion：端点支持的 A2A protocol 版本（例如 1.0）
skills：数据代理可以执行的任务，包括示例提示 (examples) 和支持的数据格式（inputModes 和 outputModes，例如 text/plain 或 application/json）
capabilities.streaming：一个布尔值，用于指示数据代理是否支持通过 stream 方法进行实时流式传输
capabilities.extensions：数据代理支持的 A2A 扩展服务，例如 bigquery_context/v1、stateless/v1 和 kms/v1
defaultInputModes 和 defaultOutputModes：请求和响应载荷的默认数据格式（例如 text/plain 或 application/json）

向数据智能体发送消息

如需向数据代理发送消息，请使用 send 方法。数据智能体处理请求，针对您的数据生成并运行所需的 SQL 查询，并返回自然语言答案以及生成的数据制品。

发送消息

发送消息时，请在资源路径中指定目标数据智能体：

对于内置数据代理（agents/bigquery-ca 或 agents/looker-ca），请使用 bigquery_context/v1 或 looker_context/v1 扩展程序在 metadata 字段中传递目标表或探索引用。
对于自定义数据代理 (dataAgents/DATA_AGENT_ID)，请省略 metadata 字段，因为上下文、架构和指令直接在代理资源上配置。

如需处理查询而不将对话记录存储在 Google Cloud中，请在请求中添加 stateless/v1 扩展程序。如需使用客户管理的加密密钥加密存储的对话数据和元数据，请传递包含 Cloud KMS 密钥名称的 kms/v1 扩展程序。如需了解详情，请参阅客户管理的加密密钥 (CMEK)。

以下代码示例展示了如何向内置的 BigQuery 数据代理发送消息：

Python SDK

from google.cloud import geminidataanalytics_v1

client = geminidataanalytics_v1.DataA2AServiceClient()

agent_name = "projects/PROJECT_ID/locations/LOCATION/agents/bigquery-ca"

request = geminidataanalytics_v1.SendMessageRequest(
    tenant=agent_name,
    message=geminidataanalytics_v1.Message(
        role="ROLE_USER",
        # Optional: Pass context_id to continue an existing conversation
        # context_id="projects/PROJECT_ID/locations/LOCATION/conversations/CONVERSATION_ID",
        parts=[
            geminidataanalytics_v1.Part(
                text="What are the top 5 countries where our users are located?"
            )
        ],
    ),
    configuration=geminidataanalytics_v1.SendMessageConfiguration(
        return_immediately=False
    ),
    metadata={
        "https://docs.cloud.google.com/gemini/docs/conversational-analytics-api/reference/a2a/extensions/bigquery_context/v1": {
            "datasource_references": {
                "bq": {
                    "tableReferences": [
                        {
                            "projectId": "DATASET_PROJECT_ID",
                            "datasetId": "DATASET_ID",
                            "tableId": "TABLE_ID",
                        }
                    ]
                }
            }
        },
        # Optional: Process queries without storing conversation history in Google Cloud
        # "https://docs.cloud.google.com/gemini/docs/conversational-analytics-api/reference/a2a/extensions/stateless/v1": {},
        # Optional: Encrypt conversation history and metadata with a customer-managed encryption key (CMEK)
        # "https://docs.cloud.google.com/gemini/docs/conversational-analytics-api/reference/a2a/extensions/kms/v1": {
        #     "kmsKey": "projects/PROJECT_ID/locations/LOCATION/keyRings/KEY_RING/cryptoKeys/KEY_NAME"
        # },
    },
)

response = client.send_message(request=request)

print(response)

在上面的示例中，按如下所示替换值：

PROJECT_ID：您的 Google Cloud 项目的 ID
LOCATION：代理资源的位置（例如 us、us-east4、eu 或 global）
CONVERSATION_ID：（可选）要继续的现有对话会话的 ID
What are the top 5 countries where our users are located?：向数据智能体提出的自然语言问题
DATASET_PROJECT_ID：包含 BigQuery 数据集的 Google Cloud 项目的 ID（例如 bigquery-public-data）
DATASET_ID：BigQuery 数据集的 ID（例如 thelook_ecommerce）
TABLE_ID：BigQuery 表的 ID（例如 users）
KEY_RING：（可选）使用 CMEK 时 Cloud KMS 密钥环的名称
KEY_NAME：（可选）使用 CMEK 时 Cloud KMS 加密密钥的名称

HTTP

curl -X POST \
  -H "Authorization: Bearer $(gcloud auth application-default print-access-token)" \
  -H "Content-Type: application/json; charset=utf-8" \
  -H "A2A-Extensions: https://docs.cloud.google.com/gemini/docs/conversational-analytics-api/reference/a2a/extensions/bigquery_context/v1" \
  -d '{
    "message": {
      "role": "ROLE_USER",
      "parts": [
        {
          "text": "What are the top 5 countries where our users are located?"
        }
      ]
    },
    "configuration": {
      "return_immediately": false
    },
    "metadata": {
      "https://docs.cloud.google.com/gemini/docs/conversational-analytics-api/reference/a2a/extensions/bigquery_context/v1": {
        "datasource_references": {
          "bq": {
            "tableReferences": [
              {
                "projectId": "DATASET_PROJECT_ID",
                "datasetId": "DATASET_ID",
                "tableId": "TABLE_ID"
              }
            ]
          }
        }
      }
    }
  }' \
  "https://geminidataanalytics.googleapis.com/v1/a2a/projects/PROJECT_ID/locations/LOCATION/agents/bigquery-ca/v1/message:send"

在上面的示例中，按如下所示替换值：

PROJECT_ID：您的 Google Cloud 项目的 ID
LOCATION：代理资源的位置（例如 us、us-east4、eu 或 global）
What are the top 5 countries where our users are located?：向数据智能体提出的自然语言问题
DATASET_PROJECT_ID：包含 BigQuery 数据集的 Google Cloud 项目的 ID（例如 bigquery-public-data）
DATASET_ID：BigQuery 数据集的 ID（例如 thelook_ecommerce）
TABLE_ID：BigQuery 表的 ID（例如 users）

了解响应结构

如果请求成功，则会返回一个 task 对象，其中包含最终状态、对话标识符和生成的制品：

{
  "task": {
    "id": "ab12d1f2-e170-4c4f-aff4-be466c6beaaa",
    "contextId": "projects/my-project/locations/us/conversations/conv-67890",
    "status": {
      "state": "TASK_STATE_COMPLETED"
    },
    "artifacts": [
      {
        "artifactId": "synthetic-8a368e8e-378a-4b80-9f07-0b1a397c221d",
        "name": "Final response",
        "description": "Final response from the agent.",
        "parts": [
          {
            "text": "The top 5 countries where our users are located are China (33,783), the United States (22,701), Brasil (14,620), South Korea (5,302), and France (4,645)."
          }
        ]
      },
      {
        "artifactId": "synthetic-50483808-b231-4b30-a859-2c30d0355a8d",
        "name": "Generated SQL",
        "description": "Generated SQL from the agent.",
        "parts": [
          {
            "text": "SELECT country, COUNT(DISTINCT id) AS user_count FROM `bigquery-public-data.thelook_ecommerce.users` GROUP BY country ORDER BY user_count DESC LIMIT 5",
            "mediaType": "text/x-sql"
          }
        ]
      }
    ]
  }
}

响应包括以下关键字段：

task.id：执行任务的唯一标识符
task.contextId：对话的资源路径，您可以在后续请求的 message.contextId 字段中传递该路径，以继续会话
task.status.state：任务的执行状态（例如 TASK_STATE_COMPLETED）
task.artifacts[]：由数据智能体生成的结构化资产，例如自然语言答案 (Final response)、可执行的 SQL 查询 (Generated SQL) 和表格形式的结果行 (Data result)

从数据代理流式传输响应

如需在数据代理推理查询时接收实时更新，请使用 stream 方法。响应会通过中间想法和增量制品 (artifact_update)（例如生成的 SQL 查询和 Vega-Lite 图表规范）来流式传输状态更新 (status_update)。

流式请求还支持继续对话 (context_id)、无状态处理 (stateless/v1) 和客户管理的加密密钥 (kms/v1)。

发送流式消息

以下代码示例展示了如何从内置的 BigQuery 数据代理中流式传输事件。如需从自定义数据代理 (dataAgents/DATA_AGENT_ID) 进行流式传输，请以自定义代理资源路径为目标，并省略 metadata 字段：

Python SDK

from google.cloud import geminidataanalytics_v1

client = geminidataanalytics_v1.DataA2AServiceClient()

agent_name = "projects/PROJECT_ID/locations/LOCATION/agents/bigquery-ca"

request = geminidataanalytics_v1.SendMessageRequest(
    tenant=agent_name,
    message=geminidataanalytics_v1.Message(
        role="ROLE_USER",
        parts=[
            geminidataanalytics_v1.Part(
                text="What are the top 5 countries where our users are located? Please show a pie chart."
            )
        ],
    ),
    metadata={
        "https://docs.cloud.google.com/gemini/docs/conversational-analytics-api/reference/a2a/extensions/bigquery_context/v1": {
            "datasource_references": {
                "bq": {
                    "tableReferences": [
                        {
                            "projectId": "DATASET_PROJECT_ID",
                            "datasetId": "DATASET_ID",
                            "tableId": "TABLE_ID",
                        }
                    ]
                }
            }
        },
        # Optional: Process queries without storing conversation history in Google Cloud
        # "https://docs.cloud.google.com/gemini/docs/conversational-analytics-api/reference/a2a/extensions/stateless/v1": {},
        # Optional: Encrypt conversation history and metadata with a customer-managed encryption key (CMEK)
        # "https://docs.cloud.google.com/gemini/docs/conversational-analytics-api/reference/a2a/extensions/kms/v1": {
        #     "kmsKey": "projects/PROJECT_ID/locations/LOCATION/keyRings/KEY_RING/cryptoKeys/KEY_NAME"
        # },
    },
)

# Stream response events
stream = client.send_streaming_message(request=request)

for chunk in stream:
  print(chunk)

在上面的示例中，按如下所示替换值：

PROJECT_ID：您的 Google Cloud 项目的 ID
LOCATION：代理资源的位置（例如 us、us-east4、eu 或 global）
What are the top 5 countries where our users are located? Please show a pie chart.：向数据智能体提出的自然语言问题
DATASET_PROJECT_ID：包含 BigQuery 数据集的 Google Cloud 项目的 ID（例如 bigquery-public-data）
DATASET_ID：BigQuery 数据集的 ID（例如 thelook_ecommerce）
TABLE_ID：BigQuery 表的 ID（例如 users）
KEY_RING：（可选）使用 CMEK 时 Cloud KMS 密钥环的名称
KEY_NAME：（可选）使用 CMEK 时 Cloud KMS 加密密钥的名称

HTTP

curl -X POST \
  -N \
  -H "Authorization: Bearer $(gcloud auth application-default print-access-token)" \
  -H "Content-Type: application/json; charset=utf-8" \
  -H "Accept: text/event-stream, application/json" \
  -H "A2A-Extensions: https://docs.cloud.google.com/gemini/docs/conversational-analytics-api/reference/a2a/extensions/bigquery_context/v1" \
  -d '{
    "message": {
      "role": "ROLE_USER",
      "parts": [
        {
          "text": "What are the top 5 countries where our users are located? Please show a pie chart."
        }
      ]
    },
    "metadata": {
      "https://docs.cloud.google.com/gemini/docs/conversational-analytics-api/reference/a2a/extensions/bigquery_context/v1": {
        "datasource_references": {
          "bq": {
            "tableReferences": [
              {
                "projectId": "DATASET_PROJECT_ID",
                "datasetId": "DATASET_ID",
                "tableId": "TABLE_ID"
              }
            ]
          }
        }
      }
    }
  }' \
  "https://geminidataanalytics.googleapis.com/v1/a2a/projects/PROJECT_ID/locations/LOCATION/agents/bigquery-ca/v1/message:stream"

在上面的示例中，按如下所示替换值：

PROJECT_ID：您的 Google Cloud 项目的 ID
LOCATION：代理资源的位置（例如 us、us-east4、eu 或 global）
What are the top 5 countries where our users are located? Please show a pie chart.：向数据智能体提出的自然语言问题
DATASET_PROJECT_ID：包含 BigQuery 数据集的 Google Cloud 项目的 ID（例如 bigquery-public-data）
DATASET_ID：BigQuery 数据集的 ID（例如 thelook_ecommerce）
TABLE_ID：BigQuery 表的 ID（例如 users）

了解流式传输响应结构

当您发送流式请求时，服务器会返回一个事件对象 (StreamResponse) 流。每个事件都包含状态更新或制品更新。

状态更新事件会在数据代理根据您的查询进行推理时，提供中间进度通知和思考消息：

{
  "statusUpdate": {
    "taskId": "f41cd8e3-e665-460c-aceb-7b337f1848ef",
    "status": {
      "state": "TASK_STATE_WORKING",
      "message": {
        "role": "ROLE_AGENT",
        "parts": [
          {
            "text": "Analyzing context"
          },
          {
            "text": "Retrieved context for 1 table."
          }
        ]
      }
    }
  }
}

制品更新事件会提供结构化输出对象，例如可执行的 SQL 查询、自然语言回答或图表规范：

{
  "artifactUpdate": {
    "taskId": "f41cd8e3-e665-460c-aceb-7b337f1848ef",
    "artifact": {
      "artifactId": "synthetic-7ca98286-0a15-4ca0-a8bc-f14dc231b3ba",
      "name": "Chart result",
      "description": "Chart visualization generated by the data agent.",
      "parts": [
        {
          "data": {
            "title": "Top 5 Countries by User Population",
            "mark": "arc",
            "encoding": {
              "color": {
                "field": "country",
                "type": "nominal"
              },
              "theta": {
                "field": "user_count",
                "type": "quantitative"
              }
            },
            "data": {
              "values": [
                {
                  "country": "China",
                  "user_count": 33783
                },
                {
                  "country": "United States",
                  "user_count": 22701
                },
                {
                  "country": "Brasil",
                  "user_count": 14620
                },
                {
                  "country": "South Korea",
                  "user_count": 5302
                },
                {
                  "country": "France",
                  "user_count": 4645
                }
              ]
            }
          }
        }
      }
    },
    "lastChunk": true
  }
}

流式响应包含以下关键字段：

statusUpdate.status.state：任务的中间状态或最终状态（例如 TASK_STATE_WORKING 或 TASK_STATE_COMPLETED）
statusUpdate.status.message.parts[]：执行期间发出的想法或进度说明
artifactUpdate.artifact：由数据代理生成的结构化资产，例如 Vega-Lite 图表规范 (data) 或 SQL 查询 (text)
artifactUpdate.lastChunk：一个布尔值标志，用于指示制品流是否已完成

如需在 Python 或前端应用中呈现返回的 Vega 或 Vega-Lite 规范，请参阅将代理响应呈现为可视化图表。

后续步骤

了解如何使用 Vega-Lite 和 Altair 将代理回答呈现为可视化图表。
了解如何使用编写的上下文来引导代理行为，以配置业务规则和经过验证的查询。
探索多智能体系统的架构集成模式。
查看 Gemini Data Analytics API REST 参考文档。

使用 A2A 编排数据代理 使用集合让一切井井有条 根据您的偏好保存内容并对其进行分类。

数据智能体编排的工作原理

准备工作

所需的角色

探索智能体功能

检索智能体卡片

Python SDK

HTTP

了解代理卡片结构

向数据智能体发送消息

发送消息

Python SDK

HTTP

了解响应结构

从数据代理流式传输响应

发送流式消息

Python SDK

HTTP

了解流式传输响应结构

后续步骤

使用 A2A 编排数据代理