Google 會運用 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 資料庫驗證」。

驗證 MySQL 適用的 Cloud SQL 和 PostgreSQL 適用的 Cloud SQL

如要連線至 MySQL 適用的 Cloud SQL 或 PostgreSQL 適用的 Cloud SQL 資料來源，必須驗證 Cloud SQL 執行個體。

啟用 IAM 資料庫驗證功能。請確認 MySQL 適用的 Cloud SQL 執行個體中，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) 和使用 Proxy 連線的權限 (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 Proxy 或連接器程式庫連線，由這些工具處理權杖交換作業。資料庫使用者名稱是 IAM 主體的電子郵件地址。不使用密碼。而是使用 OAuth 2.0 權杖，通常由 Cloud SQL Auth Proxy 或 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 資料
MySQL 適用的 Cloud SQL 資料
PostgreSQL 適用的 Cloud SQL 資料
Spanner 資料

連結 Looker 資料

下列程式碼範例定義了與 Looker 探索的連線。如要與 Looker 執行個體建立連線，請確認您已產生 Looker API 金鑰，詳情請參閱「驗證並連線至資料來源」。使用 Conversational Analytics API 時，一次最多可連結五個 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 Explore 名稱。您可以重複使用這個變數，最多可加入五個探索。

如要進一步瞭解如何連結至 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： Google Cloud 專案的 ID，其中包含您要連結的 BigQuery 資料集和資料表。如要連結公開資料集，請指定 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。

連線至 MySQL 適用的 Cloud SQL 和 PostgreSQL 適用的 Cloud SQL 資料

Conversational Analytics API 會使用 sql 欄位中的 database_reference 欄位，連線至 Cloud SQL 執行個體。

以下範例定義了與 MySQL 適用的 Cloud SQL 或 PostgreSQL 適用的 Cloud SQL 資料庫的連線。

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、位置和代理程式的專屬 ID。
資料代理的說明。
資料代理的背景資訊，包括系統說明 (在「調整初始設定和驗證」中定義) 和代理使用的資料來源 (在「連結至資料來源」中定義)。

建立資料代理時，您可以在要求酬載中加入 kms_key 欄位，選擇使用客戶自行管理的加密金鑰 (CMEK) 保護資料代理。詳情請參閱「客戶自行管理的加密金鑰 (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：資料代理的專屬 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：對話的專屬 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。

列出資料代理

下列程式碼示範如何傳送 HTTP GET 要求至 dataAgents 端點，列出特定專案的所有資料代理。

如要列出所有代理，您必須具備專案的 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。

列出對話

下列程式碼範例示範如何將 HTTP GET 要求傳送至 conversations 端點，列出特定專案的對話。

根據預設，這個方法會傳回您建立的對話。管理員 (具備 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。

列出對話中的訊息

下列程式碼範例示範如何將 HTTP GET 要求傳送至對話的 messages 端點，列出對話中的所有訊息。

如要列出訊息，您必須具備對話的 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：對話的專屬 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 進行驗證

驗證 MySQL 適用的 Cloud SQL 和 PostgreSQL 適用的 Cloud SQL

向 Spanner 驗證

連結至資料來源

連結 Looker 資料

連結 BigQuery 資料

連結至 BigQuery 資料表

連結至 BigQuery Graph

連結至數據分析資料

連線至 AlloyDB 資料

連線至 MySQL 適用的 Cloud SQL 和 PostgreSQL 適用的 Cloud SQL 資料

連結至 Spanner 資料

建立資料代理

同步

非同步

建立對話

管理資料代理和對話

取得資料代理

列出資料代理

列出可存取的資料代理

更新資料代理

同步

非同步

取得資料代理的 IAM 政策

設定資料代理的 IAM 政策

刪除資料代理

同步

非同步

取得對話

列出對話

列出對話中的訊息

刪除對話

使用 API 提問

透過有狀態和無狀態對話提問

有狀態的對話

傳送有狀態的對話要求，並附上對話參照

無狀態對話

傳送無狀態對話要求，並提供資料代理參照

傳送無狀態對話要求，並提供內嵌背景資訊

建立無狀態多輪對話

定義輔助函式

串流對話輔助函式

顯示串流對話回覆的輔助函式

使用 HTTP 和 Python 建構資料代理程式