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

使用 HTTP 和 Python 构建数据代理

本页面将引导您使用 Python 向 Conversational Analytics API（通过 geminidataanalytics.googleapis.com 访问）发出 HTTP 请求。

本页面上的示例 Python 代码展示了如何完成以下任务：

配置初始设置和身份验证。
连接到 Looker、BigQuery、数据洞察、AlloyDB for PostgreSQL、Cloud SQL 或 Spanner 数据源。
创建数据智能体。
创建对话。
管理数据智能体及对话。
使用 API 提问。
创建无状态多轮对话。
定义辅助函数。

配置初始设置和身份验证

以下示例 Python 代码会执行以下任务：

导入所需的 Python 库。
使用 Google Cloud CLI 获取用于 HTTP 身份验证的访问令牌。
定义结算项目、位置和系统指令的变量。
根据 location 变量设置 API 端点。

import json
import json as json_lib
import textwrap

import altair as alt
import google.auth
from google.auth.transport.requests import Request
import IPython
from IPython.display import display, HTML
import pandas as pd
from pygments import highlight, lexers, formatters
import requests

from google.colab import auth
auth.authenticate_user()

access_token = !gcloud auth application-default print-access-token
headers = {
        "Authorization": f"Bearer {access_token[0]}",
        "Content-Type": "application/json",
        "x-server-timeout": "300", # Custom timeout up to 600s
}

billing_project = "BILLING_PROJECT"
location = "LOCATION"
system_instruction = "SYSTEM_INSTRUCTIONS"

# Set the base URL based on location.
if not location or location == "global":
  base_url = "https://geminidataanalytics.googleapis.com"
elif "-" in location:
  # Regional endpoints
  base_url = f"https://geminidataanalytics-{location}.googleapis.com"
else:
  # Multi-regional endpoints
  base_url = f"https://geminidataanalytics.{location}.rep.googleapis.com"

按如下所示替换示例值：

BILLING_PROJECT：启用了所需 API 的 Google Cloud 项目的 ID。
LOCATION：资源的存储位置。如需使用全球端点，请指定 global。为满足数据驻留要求，请指定受支持的单区域（例如 us-east4）或多区域（例如 eu 或 us）。如需了解详情，请参阅数据驻留。

重要提示：请求路径中的位置必须与您调用的端点的位置一致。例如，如果您调用 geminidataanalytics.us.rep.googleapis.com 端点，请在请求路径中指定 us（例如 /locations/us）。
SYSTEM_INSTRUCTIONS：用于指导智能体行为并根据您的数据需求自定义智能体的系统指令。例如，您可以使用系统指令来定义业务术语、控制回答长度或设置数据格式。最好使用撰写有效的系统指令部分推荐的 YAML 格式定义系统指令，以提供详尽的结构化指导。

向 Looker 进行身份验证

如果您打算连接到 Looker 数据源，则需要对 Looker 实例进行身份验证。

使用 API 密钥

以下 Python 代码示例演示了如何使用 API 密钥向 Looker 实例进行代理身份验证。

looker_credentials = {
    "oauth": {
        "secret": {
            "client_id": "CLIENT_ID",
            "client_secret": "CLIENT_SECRET",
        }
    }
}

按如下所示替换示例值：

CLIENT_ID：您生成的 Looker API 密钥的客户端 ID。
CLIENT_SECRET：您生成的 Looker API 密钥的客户端密钥。

使用访问令牌

以下 Python 代码示例演示了如何使用访问令牌向 Looker 实例进行代理身份验证。

looker_credentials = {
    "oauth": {
        "token": {
            "access_token": "ACCESS_TOKEN",
        }
    }
}

按如下所示替换示例值：

ACCESS_TOKEN：您生成的用于向 Looker 进行身份验证的 access_token 值。

向 AlloyDB for PostgreSQL 进行身份验证

如果您计划连接到 AlloyDB 数据源，请使用 gcloud CLI 向 AlloyDB 实例进行身份验证。

通过在 AlloyDB 实例上启用 alloydb.iam_authentication 标志，在 AlloyDB 上启用 Identity and Access Management (IAM) 身份验证。
授予 IAM 角色。调用 Conversational Analytics API 的主账号（用户或服务账号）需要在您的Google Cloud 项目中拥有适当的 IAM 权限。这通常包括 alloydb.databaseUser 等角色。
创建相应的数据库用户。创建一个镜像 IAM 主账号的 AlloyDB 用户，并使用用户或服务账号电子邮件地址作为用户名。

如需了解详情，请参阅 IAM 数据库身份验证。

向 Cloud SQL for MySQL 和 Cloud SQL for PostgreSQL 进行身份验证

如果您打算连接到 Cloud SQL for MySQL 或 Cloud SQL for PostgreSQL 数据源，则必须向 Cloud SQL 实例进行身份验证。

启用 IAM 数据库身份验证。确保在 Cloud SQL for MySQL 实例中将 cloudsql.iam_authentication 标志设置为 on。您可以在创建实例或修补现有实例时添加此设置。
```
gcloud sql instances patch INSTANCE_NAME --database-flags
cloudsql.iam_authentication=on
```
授予所需的 IAM 角色。

尝试连接的主账号（用户或服务账号）需要对项目拥有 cloudsql.instances.login 和 cloudsql.instances.connect IAM 权限。如需授予这些角色和权限，请执行以下操作之一：
1. 授予 roles/cloudsql.instanceUser 角色。这包括登录实例 (cloudsql.instances.login) 和使用代理 (cloudsql.instances.connect) 进行连接的权限。
2. 或者，授予至少包含 cloudsql.instances.login 和 cloudsql.instances.connect 的自定义角色。
将 IAM 主账号添加为数据库用户。在 Cloud SQL 实例中创建与 IAM 主体的电子邮件地址对应的用户。

对于服务账号，请运行以下命令：
```
gcloud sql users create SERVICE_ACCOUNT_EMAIL --instance=INSTANCE_NAME
--type=cloud_iam_service_account
```
对于用户账号，请运行以下命令：
```
gcloud sql users create USER_EMAIL --instance=INSTANCE_NAME
--type=cloud_iam_user
```
授予数据库权限。使用 IAM 进行连接可向数据库验证用户身份，但不会自动授予数据库中的权限。您需要使用标准 PostgreSQL 命令（例如 GRANT）向此新数据库用户授予对特定对象（例如表和架构）的权限。

以超级用户（例如默认 postgres 用户）身份连接，然后运行以下命令：

-- Example: Connect using psql as the 'postgres' user

GRANT SELECT ON ALL TABLES IN SCHEMA public TO
"user-or-service-account@example.com";

使用 IAM 身份验证进行连接。

客户端通常通过 Cloud SQL Auth 代理或连接器库进行连接，后者负责处理令牌交换。数据库用户名是 IAM 主账号的邮箱。不使用密码。而是使用 OAuth 2.0 令牌，通常由 Cloud SQL Auth 代理或 Google Cloud CLI 自动处理。

以下示例使用 psql 和 gcloud CLI 生成令牌：
```
PGPASSWORD=$(gcloud sql generate-login-token) psql --host=HOST_IP \
--username=IAM_PRINCIPAL_EMAIL --dbname=DATABASE_NAME\
```
如需了解详情，请参阅 IAM 身份验证。

按如下所示替换示例值：
- INSTANCE_NAME：您要修补以启用 IAM 数据库身份验证的 Cloud SQL 实例的 ID。
- SERVICE_ACCOUNT_EMAIL：服务账号的电子邮件地址。
- USER_EMAIL：用户账号的邮箱。
- HOST_IP：Cloud SQL 实例的 IP 地址。
- IAM_PRINCIPAL_EMAIL：IAM 主体（例如服务账号或用户）的电子邮件地址。
- DATABASE_NAME：要连接的数据库名称。

向 Spanner 进行身份验证

如果您计划连接到 Spanner 数据源，则需要向 Spanner 实例进行身份验证。

授予 IAM 角色。调用 Conversational Analytics API 的主账号（用户或服务账号）需要在您的Google Cloud 项目中拥有适当的 IAM 权限。这通常包括 databaseUser 等角色。
使用数据定义语言 (DDL) 在 Spanner 中创建相应的抽象数据库角色。
使用 IAM 政策让 IAM 主账号代入这些数据库角色。

如需了解详情，请参阅精细访问权限控制权限。

连接到数据源

以下部分介绍了如何为代理的数据源定义连接详细信息。代理可以通过以下方式连接到数据：

Looker
数据洞察
BigQuery
AlloyDB 数据
Cloud SQL for MySQL 数据
Cloud SQL for PostgreSQL 数据
Spanner 数据

连接到 Looker 数据

以下示例代码定义了与 Looker 探索的连接。如需与 Looker 实例建立连接，请验证您是否已生成 Looker API 密钥，如对数据源进行身份验证并连接到数据源中所述。使用 Conversational Analytics API 时，您最多可以同时连接 5 个 Looker 探索。

looker_instance_uri = "INSTANCE_URI"
looker_data_source = {
    "looker": {
        "explore_references": [
            {
                "looker_instance_uri": looker_instance_uri,
                "lookml_model": "MODEL",
                "explore": "EXPLORE",
            },
            {
                "looker_instance_uri": looker_instance_uri,
                "lookml_model": "MODEL_2",
                "explore": "EXPLORE_2",
            },
            # Add up to 5 total Explore references
        ],
    }
}

按如下所示替换示例值：

INSTANCE_URI：Looker 实例的完整网址（例如 https://mycompany.looker.com）。
MODEL：包含您要连接的探索的 LookML 模型的名称。
EXPLORE：您希望数据智能体查询的 Looker 探索的名称。
MODEL_2：包含您要连接的探索的第二个 LookML 模型的名称。您可以为其他模型重复使用此变量，最多可用于五个探索。
EXPLORE_2：您希望数据智能体查询的其他 Looker 探索的名称。您可以重复使用此变量，最多可包含五个探索。

如需详细了解如何连接到 Looker 探索（包括所需角色），请参阅连接到 Looker。

连接到 BigQuery 数据

以下部分介绍了如何使用表引用、图引用或同时使用这两者来连接到 BigQuery 数据。如需详细了解如何连接到表和图，包括所需角色，请参阅连接到 BigQuery。

连接到 BigQuery 表

以下示例代码定义了与多个 BigQuery 表的连接，并包含可选的结构化上下文字段的示例。

bigquery_data_sources = {
    "bq": {
        "tableReferences": [
            {
                "projectId": "PROJECT_ID",
                "datasetId": "DATASET_ID",
                "tableId": "TABLE_ID",
                "schema": {
                    "description": "TABLE_DESCRIPTION",
                    "fields": [{
                        "name": "COLUMN_NAME",
                        "description": "COLUMN_DESCRIPTION"
                    }]
                }
            },
            {
                "projectId": "PROJECT_ID_2",
                "datasetId": "DATASET_ID_2",
                "tableId": "TABLE_ID_2"
            },
            {
                "projectId": "PROJECT_ID_3",
                "datasetId": "DATASET_ID_3",
                "tableId": "TABLE_ID_3"
            },
        ]
    }
}

按如下所示替换示例值：

PROJECT_ID：包含您要连接到的 BigQuery 数据集和表的 Google Cloud 项目的 ID。如需连接到公共数据集，请指定 bigquery-public-data。
DATASET_ID：BigQuery 数据集的 ID。
TABLE_ID：BigQuery 表的 ID。
TABLE_DESCRIPTION：对表内容和用途的可选说明。
COLUMN_NAME：表中要为其提供可选说明的列的名称。
COLUMN_DESCRIPTION：对列内容和用途的可选说明。

连接到 BigQuery Graph

您还可以使用 BigQuery Graph 连接到图。在 propertyGraphReferences 字段中提供一个或多个图表引用。

如需了解 BigQuery 版本要求和图查询限制，请参阅连接到 BigQuery Graph。

bigquery_data_sources = {
    "bq": {
        "propertyGraphReferences": [
          {
            "projectId": "GRAPH_PROJECT_ID",
            "datasetId": "GRAPH_DATASET_ID",
            "propertyGraphId": "GRAPH_ID"
          }
        ]
    }
}

按如下所示替换示例值：

GRAPH_PROJECT_ID：包含 BigQuery 图的 Google Cloud 项目的 ID。
GRAPH_DATASET_ID：包含图的 BigQuery 数据集的 ID。
GRAPH_ID：BigQuery 图的 ID。

如需使用同一数据智能体查询表和图表，请在 bq 对象中同时添加 tableReferences 和 propertyGraphReferences。

连接到数据洞察数据

以下示例代码定义了与数据洞察数据源的连接。

looker_studio_data_source = {
    "studio":{
        "studio_references": [
            {
              "studio_datasource_id": "studio_datasource_id"
            }
        ]
    }
}

将 studio_datasource_id 替换为数据源 ID。

连接到 AlloyDB 数据

Conversational Analytics API 使用 alloydb 字段中的 database_reference 字段连接到 AlloyDB 集群。

以下示例定义了与 AlloyDB 数据库的连接。

alloydb_data_source = {
    "alloydb": {
        "database_reference":
          {
            "project_id":"PROJECT_ID",
            "region":"REGION",
            "cluster_id":"CLUSTER_ID",
            "instance_id":"INSTANCE_ID",
            "database_id":"DATABASE",
            "table_ids":["TABLE_1_ID", "TABLE_2_ID"]
          }
          # Optional: Include this if you have pre-authored context for the agent
          # "agent_context_reference": {
          #     "context_set_id": f"projects/billing_project/locations/location/contextSets/your_context_set_id"
          # }

    }
}

按如下所示替换示例值：

PROJECT_ID：包含 AlloyDB 集群的 Google Cloud 项目的 ID。
REGION：AlloyDB 集群所在的区域，例如 us-central1。
CLUSTER_ID：AlloyDB 集群的 ID。
INSTANCE_ID：AlloyDB 实例的 ID。
DATABASE：目标数据库的名称，例如 financial。
TABLE_1_ID、TABLE_2_ID：可选。建议数据智能体使用的数据库中的表列表，例如 "loan"、"client"、"disp"。
billing_project：启用了所需 API 的结算项目的 ID。
location：上下文集所在的位置，例如 global。
your_context_set_id：如果您为代理创建了高级上下文，则为上下文集的 ID。

连接到 Cloud SQL for MySQL 和 Cloud SQL for PostgreSQL 数据

Conversational Analytics API 使用 sql 字段中的 database_reference 字段连接到您的 Cloud SQL 实例。

以下示例定义了与 Cloud SQL for MySQL 或 Cloud SQL for PostgreSQL 数据库的连接。

sql_data_source = {
    "cloud_sql_reference": {
        "database_reference":
          {
            "project_id":"PROJECT_ID",
            "region":"REGION",
            "cluster_id":"CLUSTER_ID",
            "instance_id":"INSTANCE_ID",
            "database_id":"DATABASE_ID",
            "table_ids":["TABLE_1_ID", "TABLE_2_ID"]
          }
          # Optional: Include this if you have pre-authored context for the agent
          # "agent_context_reference": {
          #     "context_set_id": f"projects/billing_project/locations/location/contextSets/your_context_set_id"
          # }
    }
}

按如下所示替换示例值：

PROJECT_ID：包含 Cloud SQL 实例的 Google Cloud 项目的 ID。
REGION：Cloud SQL 实例所在的区域，例如 us-central1。
CLUSTER_ID：Cloud SQL 集群的 ID。
INSTANCE_ID：Cloud SQL 实例的 ID。
DATABASE_ID：目标数据库的名称，例如 financial。
TABLE_1_ID、TABLE_2_ID：可选。建议数据智能体使用的数据库中的表列表，例如 "loan"、"client"、"disp"。
billing_project：启用了所需 API 的结算项目的 ID。
location：上下文集所在的位置，例如 global。
your_context_set_id：如果您为代理创建了高级上下文，则为上下文集的 ID。

连接到 Spanner 数据

Conversational Analytics API 使用 spanner 字段中的 database_reference 字段连接到 Spanner 实例。

以下示例定义了与 Spanner 数据库的连接。

spanner_data_sources = {
    "spanner_reference": {
        "database_reference": {
            "project_id": "PROJECT_ID",
            "region": location,
            "engine": GOOGLE_SQL,
            "instance_id": "INSTANCE_ID",
            "database_id": "DATABASE",
            "table_ids":["TABLE_1_ID", "TABLE_2_ID"]
        },
        # Optional: Include this if you have pre-authored context for the agent
        # "agent_context_reference": {
        #     "context_set_id": f"projects/billing_project/locations/location/contextSets/your_context_set_id"
        # }
    }
}

按如下所示替换示例值：

PROJECT_ID：包含 Spanner 实例的 Google Cloud 项目的 ID。
INSTANCE_ID：Spanner 实例的 ID。
DATABASE：目标数据库的名称，例如 financial。
TABLE_1_ID、TABLE_2_ID：建议数据智能体使用的数据库中的表列表，例如 "loan"、"client"、"disp"。
billing_project：启用了所需 API 的结算项目的 ID。
location：上下文集所在的位置，例如 global。
your_context_set_id：如果您为代理创建了进阶版内容相关定位，则为上下文集的 ID。

创建数据智能体

以下示例代码演示了如何通过向数据智能体创建端点发送 HTTP POST 请求来同步或异步创建数据智能体。请求载荷包含以下详细信息：

代理的完整资源名称。此值包括项目 ID、位置和代理的唯一标识符。
数据智能体的说明。
数据智能体的上下文，包括系统说明（在配置初始设置和身份验证中定义）和智能体使用的数据源（在连接到数据源中定义）。

您可以选择使用客户管理的加密密钥 (CMEK) 来保护数据智能体，方法是在创建数据智能体时在请求载荷中添加 kms_key 字段。如需了解详情，请参阅客户管理的加密密钥 (CMEK)。

您还可以选择在请求载荷中添加 options 参数，以启用使用 Python 进行高级分析。如需详细了解 options 参数以及您可以为对话配置的选项，请参阅 REST 资源：projects.locations.dataAgents。

同步

data_agent_url = f"https://geminidataanalytics.googleapis.com/v1beta/projects/{billing_project}/locations/{location}/dataAgents:createSync"

data_agent_id = "DATA_AGENT_ID"

# Optional: If using CMEK, replace the empty strings with your KMS key details.
kms_project_id = ""  # KMS_PROJECT_ID (Defaults to billing_project if empty)
key_ring_name = ""   # KEY_RING_NAME
key_name = ""        # KEY_NAME
key_location = ""    # KEY_LOCATION (Defaults to location if empty; except eu)

data_agent_payload = {
      "name": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}", # Optional
      "description": "This is the description of the data agent.", # Optional
      "data_analytics_agent": {
          "published_context": {
              "datasource_references": bigquery_data_sources,
              "system_instruction": system_instruction,
              "options": {
                  # Optional: To enable advanced analysis with Python
                  "analysis": {
                      "python": {
                          "enabled": True
                      }
                  },
                  # Optional: To limit query costs to 100MiB
                  "datasource": {
                      "bigQueryMaxBilledBytes": "104857600"
                  }
              }
          }
      }
  }

# If key details are provided, construct and add the kms_key field.
if key_ring_name and key_name:
  if not kms_project_id:
    kms_project_id = billing_project
  if not key_location:
    key_location = location
  data_agent_payload["kms_key"] = f"projects/{kms_project_id}/locations/{key_location}/keyRings/{key_ring_name}/cryptoKeys/{key_name}"

params = {"data_agent_id": data_agent_id} # Optional

data_agent_response = requests.post(
    data_agent_url, params=params, json=data_agent_payload, headers=headers
)

if data_agent_response.status_code == 200:
    print("Data Agent created successfully!")
    print(json.dumps(data_agent_response.json(), indent=2))
else:
    print(f"Error creating Data Agent: {data_agent_response.status_code}")
    print(data_agent_response.text)

异步

data_agent_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/dataAgents"

data_agent_id = "DATA_AGENT_ID"

# Optional: If using CMEK, replace the empty strings with your KMS key details.
kms_project_id = ""  # KMS_PROJECT_ID (Defaults to billing_project if empty)
key_ring_name = ""   # KEY_RING_NAME
key_name = ""        # KEY_NAME
key_location = ""    # KEY_LOCATION (Defaults to location if empty; except eu)

data_agent_payload = {
      "name": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}", # Optional
      "description": "This is the description of the data agent.", # Optional
      "data_analytics_agent": {
          "published_context": {
              "datasource_references": bigquery_data_sources,
              "system_instruction": system_instruction,
              "options": {
                  # Optional: To enable advanced analysis with Python
                  "analysis": {
                      "python": {
                          "enabled": True
                      }
                  },
                  # Optional: To limit query costs to 100MiB
                  "datasource": {
                      "bigQueryMaxBilledBytes": "104857600"
                  }
              }
          }
      }
  }

# If key details are provided, construct and add the kms_key field.
if key_ring_name and key_name:
  if not kms_project_id:
    kms_project_id = billing_project
  if not key_location:
    key_location = location
  data_agent_payload["kms_key"] = f"projects/{kms_project_id}/locations/{key_location}/keyRings/{key_ring_name}/cryptoKeys/{key_name}"

params = {"data_agent_id": data_agent_id} # Optional

data_agent_response = requests.post(
    data_agent_url, params=params, json=data_agent_payload, headers=headers
)

if data_agent_response.status_code == 200:
    print("Data Agent created successfully!")
    print(json.dumps(data_agent_response.json(), indent=2))
else:
    print(f"Error creating Data Agent: {data_agent_response.status_code}")
    print(data_agent_response.text)

按如下所示替换示例值：

DATA_AGENT_ID：数据智能体的唯一标识符。此值用于代理的资源名称，并用作 data_agent_id 网址查询参数。
KMS_PROJECT_ID：如果使用 CMEK，则为密钥托管所在的项目 ID。如果未提供，则默认为您的结算项目。
KEY_RING_NAME：如果使用 CMEK，则为 Cloud KMS 密钥环的名称。
KEY_NAME：如果使用 CMEK，则为 Cloud KMS 密钥的名称。
KEY_LOCATION：如果使用 CMEK，则为密钥环的位置。对于 eu 多区域位置，请使用 europe。对于所有其他位置，请使用与数据智能体相同的位置。如果未提供，则默认为数据智能体的位置，这适用于除 eu 多区域之外的所有位置。如需了解详情，请参阅密钥位置要求。
This is the description of the data agent.：数据智能体的说明。

创建对话

以下示例代码演示了如何创建与数据智能体的对话。

您可以选择在创建对话时在请求载荷中添加 kms_key 字段，从而使用 CMEK 保护对话。如需了解详情，请参阅客户管理的加密密钥 (CMEK)。

conversation_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/conversations"

data_agent_id = "DATA_AGENT_ID"
conversation_id = "CONVERSATION_ID"

# Optional: If using CMEK, replace the empty strings with your KMS key details.
kms_project_id = ""  # KMS_PROJECT_ID (Defaults to billing_project if empty)
key_ring_name = ""   # KEY_RING_NAME
key_name = ""        # KEY_NAME
key_location = ""    # KEY_LOCATION (Defaults to location if empty; single regions only)

conversation_payload = {
    "agents": [
        f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"
    ],
    "name": f"projects/{billing_project}/locations/{location}/conversations/{conversation_id}",
}

# If key details are provided, construct and add the kms_key field.
if key_ring_name and key_name:
  if not kms_project_id:
    kms_project_id = billing_project
  if not key_location:
    key_location = location
  conversation_payload["kms_key"] = f"projects/{kms_project_id}/locations/{key_location}/keyRings/{key_ring_name}/cryptoKeys/{key_name}"

params = {
    "conversation_id": conversation_id
}

conversation_response = requests.post(conversation_url, headers=headers, params=params, json=conversation_payload)

if conversation_response.status_code == 200:
    print("Conversation created successfully!")
    print(json.dumps(conversation_response.json(), indent=2))
else:
    print(f"Error creating Conversation: {conversation_response.status_code}")
    print(conversation_response.text)

按如下所示替换示例值：

DATA_AGENT_ID：数据智能体的 ID，如创建数据智能体部分的示例代码块中所定义。
CONVERSATION_ID：对话的唯一标识符。
KMS_PROJECT_ID：如果使用 CMEK，则为密钥托管所在的项目 ID。如果未提供，则默认为您的结算项目。
KEY_RING_NAME：如果使用 CMEK，则为 Cloud KMS 密钥环的名称。
KEY_NAME：如果使用 CMEK，则为 Cloud KMS 密钥的名称。
KEY_LOCATION：如果使用 CMEK，则为密钥环的位置。对于 us 多区域位置，请使用 us-central1。对于 eu 多区域位置，请使用 europe-west1。对于单个区域，请使用与对话相同的区域。如果未提供，则默认为对话的位置，这仅适用于单个区域。如需了解详情，请参阅密钥位置要求。

管理数据智能体及对话

以下代码示例展示了如何使用 Conversational Analytics API 管理数据智能体及对话。您可以执行以下任务：

获取数据智能体
列出数据智能体
列出可访问的数据智能体
更新数据智能体
获取数据智能体的 IAM 政策
为数据智能体设置 IAM 政策
删除数据智能体
获取对话
列出对话
列出对话中的消息

获取数据智能体

以下示例代码演示了如何通过向数据智能体资源网址发送 HTTP GET 请求来获取现有数据智能体。

data_agent_id = "DATA_AGENT_ID"
data_agent_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"

data_agent_response = requests.get(
    data_agent_url, headers=headers
)

if data_agent_response.status_code == 200:
    print("Fetched Data Agent successfully!")
    print(json.dumps(data_agent_response.json(), indent=2))
else:
    print(f"Error: {data_agent_response.status_code}")
    print(data_agent_response.text)

在上面的示例中，将 DATA_AGENT_ID 替换为要获取的数据智能体的 ID。

列出数据智能体

以下代码演示了如何通过向 dataAgents 端点发送 HTTP GET 请求来列出给定项目的所有数据智能体。

如需列出所有智能体，您必须拥有项目的 geminidataanalytics.dataAgents.list 权限。如需详细了解有哪些 IAM 角色包含此权限，请参阅预定义角色列表。

billing_project = "BILLING_PROJECT"
data_agent_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/dataAgents"

data_agent_response = requests.get(
    data_agent_url, headers=headers
)

if data_agent_response.status_code == 200:
    print("Data Agents Listed successfully!")
    print(json.dumps(data_agent_response.json(), indent=2))
else:
    print(f"Error Listing Data Agents: {data_agent_response.status_code}")

将 BILLING_PROJECT 替换为您的结算项目的 ID。

列出可访问的数据智能体

以下代码演示了如何通过向 dataAgents:listAccessible 端点发送 HTTP GET 请求来列出给定项目的所有可访问的数据智能体。

billing_project = "BILLING_PROJECT"
creator_filter = "CREATOR_FILTER"
data_agent_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/dataAgents:listAccessible"

params = {
    "creator_filter": creator_filter
}

data_agent_response = requests.get(
    data_agent_url, headers=headers, params=params
)

if data_agent_response.status_code == 200:
    print("Accessible Data Agents Listed successfully!")
    print(json.dumps(data_agent_response.json(), indent=2))
else:
    print(f"Error Listing Accessible Data Agents: {data_agent_response.status_code}")

按如下所示替换示例值：

BILLING_PROJECT：结算项目的 ID。
CREATOR_FILTER：要根据数据智能体的创建者应用的过滤条件。可能的值包括 NONE（默认）、CREATOR_ONLY 和 NOT_CREATOR_ONLY。

更新数据智能体

以下示例代码演示了如何通过向数据智能体资源网址发送 HTTP PATCH 请求来同步或异步更新数据智能体。请求载荷包含您要更改的字段的新值；请求参数包含 updateMask 参数，用于指定要更新的字段。

同步

data_agent_id = "DATA_AGENT_ID"
billing_project = "BILLING_PROJECT"

data_agent_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}:updateSync"

payload = {
    "description": "Updated description of the data agent.",
    "data_analytics_agent": {
        "published_context": {
            "datasource_references": bigquery_data_sources,
            "system_instruction": system_instruction
        }
    },
}

fields = ["description", "data_analytics_agent"]
params = {
    "updateMask": ",".join(fields)
}

data_agent_response = requests.patch(
    data_agent_url, headers=headers, params=params, json=payload
)

if data_agent_response.status_code == 200:
    print("Data Agent updated successfully!")
    print(json.dumps(data_agent_response.json(), indent=2))
else:
    print(f"Error Updating Data Agent: {data_agent_response.status_code}")
    print(data_agent_response.text)

异步

data_agent_id = "DATA_AGENT_ID"
billing_project = "BILLING_PROJECT"

data_agent_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"

payload = {
    "description": "Updated description of the data agent.",
    "data_analytics_agent": {
        "published_context": {
            "datasource_references": bigquery_data_sources,
            "system_instruction": system_instruction
        }
    },
}

fields = ["description", "data_analytics_agent"]
params = {
    "updateMask": ",".join(fields)
}

data_agent_response = requests.patch(
    data_agent_url, headers=headers, params=params, json=payload
)

if data_agent_response.status_code == 200:
    print("Data Agent updated successfully!")
    print(json.dumps(data_agent_response.json(), indent=2))
else:
    print(f"Error Updating Data Agent: {data_agent_response.status_code}")
    print(data_agent_response.text)

按如下所示替换示例值：

DATA_AGENT_ID：要更新的数据智能体的 ID。
BILLING_PROJECT：结算项目的 ID。
Updated description of the data agent.：数据智能体的新说明。

获取数据智能体的 IAM 政策

以下示例代码演示了如何通过向数据智能体网址发送 HTTP POST 请求来获取数据智能体的 IAM 政策。请求载荷包含数据智能体路径。

billing_project = "BILLING_PROJECT"
data_agent_id = "DATA_AGENT_ID"

data_agent_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}:getIamPolicy"

# Request body
payload = {
    "resource": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"
}

data_agent_response = requests.post(
    data_agent_url, headers=headers, json=payload
)

if data_agent_response.status_code == 200:
    print("IAM Policy fetched successfully!")
    print(json.dumps(data_agent_response.json(), indent=2))
else:
    print(f"Error fetching IAM policy: {data_agent_response.status_code}")
    print(data_agent_response.text)

按如下所示替换示例值：

BILLING_PROJECT：结算项目的 ID。
DATA_AGENT_ID：要获取其 IAM 政策的数据智能体的 ID。

为数据智能体设置 IAM 政策

如需共享智能体，您可以使用 setIamPolicy 方法向用户分配特定智能体的 IAM 角色。以下示例代码演示了如何使用包含绑定的载荷对数据智能体网址进行 POST 调用。绑定指定了应向哪些用户分配哪些角色。

billing_project = "BILLING_PROJECT"
data_agent_id = "DATA_AGENT_ID"
role = "roles/geminidataanalytics.dataAgentEditor"
users = "222larabrown@gmail.com, cloudysanfrancisco@gmail.com"

data_agent_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}:setIamPolicy"

# Request body
payload = {
    "policy": {
        "bindings": [
            {
                "role": role,
                "members": [
                    f"user:{i.strip()}" for i in users.split(",")
                ]
            }
        ]
    }
}

data_agent_response = requests.post(
    data_agent_url, headers=headers, json=payload
)

if data_agent_response.status_code == 200:
    print("IAM Policy set successfully!")
    print(json.dumps(data_agent_response.json(), indent=2))
else:
    print(f"Error setting IAM policy: {data_agent_response.status_code}")
    print(data_agent_response.text)

按如下所示替换示例值：

BILLING_PROJECT：结算项目的 ID。
DATA_AGENT_ID：要为其设置 IAM 政策的数据智能体的 ID。
222larabrown@gmail.com, cloudysanfrancisco@gmail.com：要向其授予指定角色的用户的邮箱列表（以英文逗号分隔）。

删除数据智能体

以下示例代码演示了如何通过向数据智能体资源网址发送 HTTP DELETE 请求来同步或异步删除数据智能体。删除数据智能体后，该智能体会进入为期 30 天的软删除期，在此期间无法恢复。如需了解详情，请参阅数据保留和删除。

同步

billing_project = "BILLING_PROJECT"
data_agent_id = "DATA_AGENT_ID"

data_agent_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}:deleteSync"

data_agent_response = requests.delete(
    data_agent_url, headers=headers
)

if data_agent_response.status_code == 200:
    print("Data Agent deleted successfully!")
    print(json.dumps(data_agent_response.json(), indent=2))
else:
    print(f"Error Deleting Data Agent: {data_agent_response.status_code}")
    print(data_agent_response.text)

异步

billing_project = "BILLING_PROJECT"
data_agent_id = "DATA_AGENT_ID"

data_agent_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"

data_agent_response = requests.delete(
    data_agent_url, headers=headers
)

if data_agent_response.status_code == 200:
    print("Data Agent deleted successfully!")
    print(json.dumps(data_agent_response.json(), indent=2))
else:
    print(f"Error Deleting Data Agent: {data_agent_response.status_code}")
    print(data_agent_response.text)

按如下所示替换示例值：

BILLING_PROJECT：结算项目的 ID。
DATA_AGENT_ID：要删除的数据智能体的 ID。

获取对话

以下示例代码演示了如何通过向对话资源网址发送 HTTP GET 请求来提取现有对话。

billing_project = "BILLING_PROJECT"
conversation_id = "CONVERSATION_ID"

conversation_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/conversations/{conversation_id}"

conversation_response = requests.get(conversation_url, headers=headers)

# Handle the response
if conversation_response.status_code == 200:
    print("Conversation fetched successfully!")
    print(json.dumps(conversation_response.json(), indent=2))
else:
    print(f"Error while fetching conversation: {conversation_response.status_code}")
    print(conversation_response.text)

按如下所示替换示例值：

BILLING_PROJECT：结算项目的 ID。
CONVERSATION_ID：要提取的对话的 ID。

列出对话

以下示例代码演示了如何通过向 conversations 端点发送 HTTP GET 请求来列出给定项目的对话。

默认情况下，此方法会返回您创建的对话。管理员（具有 cloudaicompanion.topicAdmin IAM 角色的用户）可以查看项目中的所有对话。

billing_project = "BILLING_PROJECT"
conversation_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/conversations"

conversation_response = requests.get(conversation_url, headers=headers)

# Handle the response
if conversation_response.status_code == 200:
    print("Conversation fetched successfully!")
    print(json.dumps(conversation_response.json(), indent=2))
else:
    print(f"Error while fetching conversation: {conversation_response.status_code}")
    print(conversation_response.text)

将 BILLING_PROJECT 替换为启用了所需 API 的结算项目的 ID。

列出对话中的消息

以下示例代码演示了如何通过向对话的 messages 端点发送 HTTP GET 请求来列出对话中的所有消息。

如需列出消息，您必须拥有对话的 cloudaicompanion.topics.get 权限。

billing_project = "BILLING_PROJECT"

conversation_id = "CONVERSATION_ID"

conversation_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/conversations/{conversation_id}/messages"

conversation_response = requests.get(conversation_url, headers=headers)

# Handle the response
if conversation_response.status_code == 200:
    print("Conversation fetched successfully!")
    print(json.dumps(conversation_response.json(), indent=2))
else:
    print(f"Error while fetching conversation: {conversation_response.status_code}")
    print(conversation_response.text)

按如下所示替换示例值：

BILLING_PROJECT：结算项目的 ID。
CONVERSATION_ID：要列出其消息的对话的 ID。

删除对话

以下示例代码演示了如何通过向对话资源网址发送 HTTP DELETE 请求来删除对话。管理员（具有 cloudaicompanion.topicAdmin Identity and Access Management 角色的用户）或具有 cloudaicompanion.topics.delete Identity and Access Management 权限的用户可以删除项目中的对话。

billing_project = "BILLING_PROJECT"
conversation_id = "CONVERSATION_ID"

conversation_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/conversations/{conversation_id}"

conversation_response = requests.delete(conversation_url, headers=headers)

# Handle the response
if conversation_response.status_code == 200:
    print("Conversation deleted successfully!")
    print(json.dumps(conversation_response.json(), indent=2))
else:
    print(f"Error while deleting conversation: {conversation_response.status_code}")
    print(conversation_response.text)

按如下所示替换示例值：

BILLING_PROJECT：结算项目的 ID。
CONVERSATION_ID：要删除的对话的 ID。

使用 API 提问

创建数据智能体和对话（对于有状态聊天）后，您可以向智能体发送查询。

当您发送查询时，API 会返回 Message 对象的数据流。此数据流可以包含不同类型的消息，包括文本、数据表和图表。文本消息可以提供对智能体推理过程的洞见、报告其进度或给出最终答案。每条短信的用途由其 TextType 值指示：

THOUGHT：显示智能体在规划如何回答您的查询时的内部思考过程。THOUGHT 消息可让您逐步了解代理在回答查询时所用的推理和决策流程，这些消息包含两部分：parts[0] 是思考摘要，用于简要总结完整的思考文本；parts[1] 是完整的思考文本。
PROGRESS：报告智能体在某项操作（例如数据检索或正在调用的工具）上的进度。此值仅针对 Looker 数据源返回，包含两部分：parts[0] 是摘要，parts[1] 是完整的进度文本。
FINAL_RESPONSE：提供查询的最终答案。

以下部分中的示例使用用于流式传输聊天回答的辅助 Python 函数中定义的辅助函数，来处理和显示 API 流式回答中的每条消息。如需了解在使用 Looker 数据源时如何在界面中呈现这些消息，请参阅呈现 Looker 数据源的代理回答。

通过有状态和无状态对话提问

您可以通过以下主要模式与 API 进行交互：

有状态聊天： Google Cloud 会存储和管理对话历史记录。有状态聊天本质上是多轮对话，因为 API 会保留之前消息的上下文。您只需在每轮中发送当前消息。
无状态聊天：您的应用会管理对话历史记录。您必须在每条新消息中添加相关的先前消息。如需查看有关如何在无状态模式下管理多轮对话的详细示例，请参阅创建无状态多轮对话。

以下示例演示了如何通过向 :chat 端点发出 POST 请求，使用 API 进行有状态和无状态聊天。

注意：本部分中的代码示例使用标准 Python requests 库来演示 HTTP 流式传输。标准 HTTP 或 REST 客户端无法读取 HTTP/2 尾部。如果在流式传输期间发生严重错误（例如超时），API 可能会返回一般出错提示，例如 503 Connection reset by peer 或 500 Internal Server Error。对于生产应用，我们强烈建议您使用 Python SDK，该 SDK 使用 gRPC 正确解析这些尾随错误。

有状态聊天

发送带有对话引用的有状态对话请求

以下示例代码演示了如何使用您在之前的步骤中定义的对话向 API 提问。此示例使用 get_stream 辅助函数来流式传输回答。

chat_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}:chat"

data_agent_id = "DATA_AGENT_ID"
conversation_id = "CONVERSATION_ID"

# Construct the payload
chat_payload = {
    "parent": f"projects/{billing_project}/locations/{location}",
    # "credentials": looker_credentials,
    "messages": [
        {
            "userMessage": {
                "text": "Make a bar graph for the top 5 states by the total number of airports"
            }
        }
    ],
    "conversation_reference": {
        "conversation": f"projects/{billing_project}/locations/{location}/conversations/{conversation_id}",
        "data_agent_context": {
            "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
        }
    }
}

# Call the get_stream function to stream the response
get_stream(chat_url, chat_payload)

按如下所示替换示例值：

DATA_AGENT_ID：数据智能体的 ID，如创建数据智能体部分的示例代码块中所定义。
CONVERSATION_ID：对话的唯一标识符。
Make a bar graph for the top 5 states by the total number of airports用作示例提示。

无状态聊天

以下示例展示了如何使用数据智能体引用或内嵌上下文发送无状态请求。

发送包含数据智能体引用的无状态对话请求

以下示例代码演示了如何使用您在之前的步骤中定义的数据智能体向 API 提出无状态问题。此示例使用 get_stream 辅助函数来流式传输回答。

chat_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}:chat"

data_agent_id = "DATA_AGENT_ID"

# Construct the payload
chat_payload = {
    "parent": f"projects/{billing_project}/locations/{location}",
    # "credentials": looker_credentials,
    "messages": [
        {
            "userMessage": {
                "text": "Make a bar graph for the top 5 states by the total number of airports"
            }
        }
    ],
    "data_agent_context": {
        "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
    }
}

# Call the get_stream function to stream the response
get_stream(chat_url, chat_payload)

按如下所示替换示例值：

DATA_AGENT_ID：数据智能体的 ID，如创建数据智能体部分的示例代码块中所定义。
Make a bar graph for the top 5 states by the total number of airports用作示例提示。

发送包含内嵌上下文的无状态对话请求

以下示例代码演示了如何使用内嵌上下文向 API 提出无状态问题。此示例使用 get_stream 辅助函数来流式传输回答，并使用 BigQuery 数据源作为示例。

您还可以选择在请求载荷中添加 options 参数，以启用使用 Python 进行高级分析。如需详细了解 options 参数以及您可以为对话配置的选项，请参阅 REST 资源：projects.locations.dataAgents 页面。

chat_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}:chat"

# Construct the payload
chat_payload = {
    "parent": f"projects/{billing_project}/locations/{location}",
    # "credentials": looker_credentials,
    "messages": [
        {
            "userMessage": {
                "text": "Make a bar graph for the top 5 states by the total number of airports"
            }
        }
    ],
    "inline_context": {
        "datasource_references": bigquery_data_sources,
          # Optional: To enable advanced analysis with Python, include the following options block:
          "options": {
              "analysis": {
                  "python": {
                      "enabled": True
                  }
              }
          }
    }
}

# Call the get_stream function to stream the response
get_stream(chat_url, chat_payload)

以下示例代码演示了如何使用内嵌上下文和 AlloyDB 数据源向 API 提出无状态问题。

chat_url = f"https://geminidataanalytics.googleapis.com/v1/projects/{billing_project}/locations/{location}:chat"

# The natural language question to ask the data agent
user_prompt = "What is the average loan amount in the US?"  # Replace with your question

# Construct the payload
chat_payload = {
    "parent": f"projects/{billing_project}/locations/{location}",
    # "credentials": looker_credentials,
    "messages": [
        {
            "userMessage": {
                "text": user_prompt
            }
        }
    ],
    "data_agent_context": {
          "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
    }
}

print(f"Sending prompt to :chat: '{user_prompt}'")
print(f"Endpoint: {chat_url}")
print(f"Payload: {json.dumps(chat_payload, indent=2)}")

get_stream(chat_url, chat_payload)

创建无状态多轮对话

如需在无状态对话中提出后续问题，您的应用必须管理对话上下文，方法是在每次新请求中发送完整的消息记录。通过多轮对话，您可以发送基于之前对话轮次的后续问题。

以下示例定义并调用了 multi_turn_conversation 函数，通过将消息存储在列表中来管理对话上下文。multi_turn_conversation 函数会调用 get_stream_multi_turn 函数来解析流式 API 响应并更新 conversation_messages 列表。此过程会保留后续对话轮次的上下文。如需查看 get_stream_multi_turn 和其他必需辅助函数的代码，请参阅流式对话辅助函数。

chat_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}:chat"

# List that is used to track previous turns and is reused across requests
conversation_messages = []

data_agent_id = "DATA_AGENT_ID"

# Helper function for calling the API
def multi_turn_conversation(msg):

  userMessage = {
      "userMessage": {
          "text": msg
      }
  }

  # Send a multi-turn request by including previous turns and the new message
  conversation_messages.append(userMessage)

  # Construct the payload
  chat_payload = {
      "parent": f"projects/{billing_project}/locations/{location}",
      # "credentials": looker_credentials,
      "messages": conversation_messages,
      # Use a data agent reference
      "data_agent_context": {
          "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
      },
      # Use inline context
      # "inline_context": {
      #     "datasource_references": bigquery_data_sources,
      # }
  }

  # Call the get_stream_multi_turn helper function to stream the response
  get_stream_multi_turn(chat_url, chat_payload, conversation_messages)

在上述示例中，将 DATA_AGENT_ID 替换为数据智能体的 ID，如创建数据智能体部分的示例代码块中所定义。

您可以在每轮对话中调用 multi_turn_conversation 辅助函数。以下示例代码展示了如何发送初始请求，然后发送基于上一个回答的后续请求：

# Send first-turn request
multi_turn_conversation("Which species of tree is most prevalent?")

# Send follow-up-turn request
multi_turn_conversation("Can you show me the results as a bar chart?")

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

Which species of tree is most prevalent?：要发送给数据智能体的自然语言问题。
Can you show me the results as a bar chart?：基于或优化上一个问题的后续问题。

定义辅助函数

以下部分包含辅助函数定义，这些定义在前面的代码示例中用于解析来自 API 的回答并显示结果。

流式聊天辅助函数

以下函数用于解析和显示来自 API 的流式响应，以处理标准（单轮）对话请求。

显示用于流式传输聊天回答的辅助函数

    def is_json(str):
      try:
          json_object = json_lib.loads(str)
      except ValueError as e:
          return False
      return True

    def handle_text_response(resp):
      parts = resp['parts']
      full_text = "".join(parts)
      if "\n" not in full_text and len(full_text) > 80:
        wrapped_text = textwrap.fill(full_text, width=80)
        print(wrapped_text)
      else:
        print(full_text)

    def get_property(data, field_name, default = ''):
      return data[field_name] if field_name in data else default

    def display_schema(data):
      fields = data['fields']
      df = pd.DataFrame({
        "Column": map(lambda field: get_property(field, 'name'), fields),
        "Type": map(lambda field: get_property(field, 'type'), fields),
        "Description": map(lambda field: get_property(field, 'description', '-'), fields),
        "Mode": map(lambda field: get_property(field, 'mode'), fields)
      })
      display(df)

    def display_section_title(text):
      display(HTML('<h2>{}</h2>'.format(text)))

    def format_bq_table_ref(table_ref):
      return '{}.{}.{}'.format(table_ref['projectId'], table_ref['datasetId'], table_ref['tableId'])

    def format_looker_table_ref(table_ref):
      return 'lookmlModel: {}, explore: {}, lookerInstanceUri: {}'.format(table_ref['lookmlModel'], table_ref['explore'], table_ref['lookerInstanceUri'])

    def display_datasource(datasource):
      source_name = ''

      if 'studioDatasourceId' in datasource:
        source_name = datasource['studioDatasourceId']
      elif 'lookerExploreReference' in datasource:
        source_name = format_looker_table_ref(datasource['lookerExploreReference'])
      else:
        source_name = format_bq_table_ref(datasource['bigqueryTableReference'])

      print(source_name)
      display_schema(datasource['schema'])

    def handle_schema_response(resp):
      if 'query' in resp:
        print(resp['query']['question'])
      elif 'result' in resp:
        display_section_title('Schema resolved')
        print('Data sources:')
        for datasource in resp['result']['datasources']:
          display_datasource(datasource)

    def handle_data_response(resp):
      if 'query' in resp:
        query = resp['query']
        display_section_title('Retrieval query')
        print('Query name: {}'.format(query['name']))
        if 'question' in query:
          print('Question: {}'.format(query['question']))
        if 'datasources' in query:
          print('Data sources:')
          for datasource in query['datasources']:
            display_datasource(datasource)
      elif 'generatedSql' in resp:
        display_section_title('SQL generated')
        print(resp['generatedSql'])
      elif 'result' in resp:
        display_section_title('Data retrieved')

        fields = map(lambda field: get_property(field, 'name'), resp['result']['schema']['fields'])
        dict = {}

        for field in fields:
          dict[field] = map(lambda el: get_property(el, field), resp['result']['data'])

        display(pd.DataFrame(dict))

    def handle_chart_response(resp):
      if 'query' in resp:
        print(resp['query']['instructions'])
      elif 'result' in resp:
        vegaConfig = resp['result']['vegaConfig']
        alt.Chart.from_json(json_lib.dumps(vegaConfig)).display();

    def handle_error(resp):
      display_section_title('Error')
      print('Code: {}'.format(resp['code']))
      print('Message: {}'.format(resp['message']))

    def get_stream(url, json):
      s = requests.Session()

      acc = ''

      with s.post(url, json=json, headers=headers, stream=True) as resp:
        for line in resp.iter_lines():
          if not line:
            continue

          decoded_line = str(line, encoding='utf-8')

          if decoded_line == '[{':
            acc = '{'
          elif decoded_line == '}]':
            acc += '}'
          elif decoded_line == ',':
            continue
          else:
            acc += decoded_line

          if not is_json(acc):
            continue

          data_json = json_lib.loads(acc)

          if not 'systemMessage' in data_json:
            if 'error' in data_json:
                handle_error(data_json['error'])
            continue

          if 'text' in data_json['systemMessage']:
            handle_text_response(data_json['systemMessage']['text'])
          elif 'schema' in data_json['systemMessage']:
            handle_schema_response(data_json['systemMessage']['schema'])
          elif 'data' in data_json['systemMessage']:
            handle_data_response(data_json['systemMessage']['data'])
          elif 'chart' in data_json['systemMessage']:
            handle_chart_response(data_json['systemMessage']['chart'])
          else:
            colored_json = highlight(acc, lexers.JsonLexer(), formatters.TerminalFormatter())
            print(colored_json)
            print('\n')
            acc = ''

    def get_stream_multi_turn(url, json, conversation_messages):
        s = requests.Session()

        acc = ''

        with s.post(url, json=json, headers=headers, stream=True) as resp:
            for line in resp.iter_lines():
                if not line:
                    continue

                decoded_line = str(line, encoding='utf-8')

                if decoded_line == '[{':
                    acc = '{'
                elif decoded_line == '}]':
                    acc += '}'
                elif decoded_line == ',':
                    continue
                else:
                    acc += decoded_line

                if not is_json(acc):
                    continue

                data_json = json_lib.loads(acc)
                # Store the response that will be used in the next iteration
                conversation_messages.append(data_json)

                if not 'systemMessage' in data_json:
                    if 'error' in data_json:
                        handle_error(data_json['error'])
                    continue

                if 'text' in data_json['systemMessage']:
                    handle_text_response(data_json['systemMessage']['text'])
                elif 'schema' in data_json['systemMessage']:
                    handle_schema_response(data_json['systemMessage']['schema'])
                elif 'data' in data_json['systemMessage']:
                    handle_data_response(data_json['systemMessage']['data'])
                elif 'chart' in data_json['systemMessage']:
                    handle_chart_response(data_json['systemMessage']['chart'])
                else:
                    colored_json = highlight(acc, lexers.JsonLexer(), formatters.TerminalFormatter())
                    print(colored_json)
                print('\n')
                acc = ''

使用 HTTP 和 Python 构建数据代理 使用集合让一切井井有条 根据您的偏好保存内容并对其进行分类。

配置初始设置和身份验证

向 Looker 进行身份验证

使用 API 密钥

使用访问令牌

向 AlloyDB for PostgreSQL 进行身份验证

向 Cloud SQL for MySQL 和 Cloud SQL for PostgreSQL 进行身份验证

向 Spanner 进行身份验证

连接到数据源

连接到 Looker 数据

连接到 BigQuery 数据

连接到 BigQuery 表

连接到 BigQuery Graph

连接到数据洞察数据

连接到 AlloyDB 数据

连接到 Cloud SQL for MySQL 和 Cloud SQL for PostgreSQL 数据

连接到 Spanner 数据

创建数据智能体

同步

异步

创建对话

管理数据智能体及对话

获取数据智能体

列出数据智能体

列出可访问的数据智能体

更新数据智能体

同步

异步

获取数据智能体的 IAM 政策

为数据智能体设置 IAM 政策

删除数据智能体

同步

异步

获取对话

列出对话

列出对话中的消息

删除对话

使用 API 提问

通过有状态和无状态对话提问

有状态聊天

发送带有对话引用的有状态对话请求

无状态聊天

发送包含数据智能体引用的无状态对话请求

发送包含内嵌上下文的无状态对话请求

创建无状态多轮对话

定义辅助函数

流式聊天辅助函数

显示用于流式传输聊天回答的辅助函数

使用 HTTP 和 Python 构建数据代理