使用 Python SDK 建構資料代理

本頁說明如何使用 Python SDK 向對話式數據分析 API 提出要求。範例 Python 程式碼會示範如何完成下列工作：

驗證及設定環境

如要使用對話式數據分析 API 的 Python SDK，請按照對話式數據分析 API SDK Colaboratory 筆記本中的操作說明下載及安裝 SDK。請注意，SDK Colab 的下載方法和內容可能會有所變更。

完成筆記本中的設定說明後，您可以使用下列程式碼匯入必要的 SDK 程式庫、在 Colaboratory 環境中驗證 Google 帳戶，以及初始化用戶端以發出 API 要求：

from google.colab import auth
auth.authenticate_user()

from google.cloud import geminidataanalytics

data_agent_client = geminidataanalytics.DataAgentServiceClient()
data_chat_client = geminidataanalytics.DataChatServiceClient()

指定帳單專案和系統指示

下列 Python 程式碼範例會定義整個指令碼中使用的帳單專案、位置和系統指令：

# Billing project
billing_project = "my_project_name"
location = "global"

# System instructions
system_instruction = "Help the user analyze their data."

請依下列方式替換範例值：

  • my_project_name：已啟用必要 API 的帳單專案 ID。
  • Help the user analyze their data.：系統指令，可引導代理程式的行為，並根據資料需求進行自訂。舉例來說，您可以使用系統指令定義業務用語、控制回覆長度或設定資料格式。建議使用「撰寫有效的系統指令」中建議的 YAML 格式定義系統指令，提供詳細且結構化的指引。

連結至資料來源

以下各節說明如何定義代理程式資料來源的連線詳細資料。虛擬服務專員可透過下列方式連結至資料：

連結至 Looker 資料

下列程式碼範例說明如何使用 API 金鑰或存取權杖，定義 Looker 探索的連線詳細資料。您一次最多可透過 Conversational Analytics API 連線至五個 Looker 探索。

連結至 Looker 資料來源時，請注意下列事項：

  • 您可以在對話中查詢任何隨附的探索。
  • 代理程式一次只能查詢一個 Explore。您無法同時對多個探索執行查詢。
  • 在同一場對話中，代理程式可以查詢多個探索。

  • 在包含多個問題的對話中，或在包含後續問題的對話中，代理程式可以查詢多個探索。

    舉例來說，使用者連結了兩項探索，分別是 cat-explore 和 dog-explore。使用者輸入「貓的數量比較多還是狗的數量比較多？」這個問題會建立兩項查詢：一項是計算 cat-explore 中的貓數量，另一項是計算 dog-explore 中的狗數量。代理程式完成這兩項查詢後，會比較兩項查詢的結果。

下列程式碼範例會定義與多個 Looker 探索的連線。如要提升代理程式效能，您可以選擇提供黃金查詢，做為探索的結構化背景資訊。詳情請參閱「為 Looker 資料來源定義資料代理程式環境」。

API 金鑰

如要與 Looker 執行個體建立連結，請使用產生的 Looker API 金鑰，詳情請參閱「使用 Conversational Analytics API 驗證及連結資料來源」。

looker_client_id = "my_looker_client_id"
looker_client_secret = "my_looker_client_secret"
looker_instance_uri = "https://my_company.looker.com"
lookml_model_1 = "my_model"
explore_1 = "my_explore"
lookml_model_2 = "my_model_2"
explore_2 = "my_explore_2"

looker_explore_reference = geminidataanalytics.LookerExploreReference()
looker_explore_reference.looker_instance_uri = looker_instance_uri
looker_explore_reference.lookml_model = "my_model"
looker_explore_reference.explore = "my_explore"

looker_explore_reference2 = geminidataanalytics.LookerExploreReference()
looker_explore_reference2.looker_instance_uri = looker_instance_uri
looker_explore_reference2.lookml_model = "my_model_2"
looker_explore_reference2.explore = "my_explore_2"

credentials = geminidataanalytics.Credentials()
credentials.oauth.secret.client_id = looker_client_id
credentials.oauth.secret.client_secret = looker_client_secret

datasource_references = geminidataanalytics.DatasourceReferences()
datasource_references.looker.explore_references = [looker_explore_reference, looker_explore_reference2]

# Do not include the following line during agent creation
datasource_references.credentials = credentials

請依下列方式替換範例值：

  • my_looker_client_id：您產生的 Looker API 金鑰用戶端 ID。
  • my_looker_client_secret：您產生的 Looker API 金鑰用戶端密鑰。
  • https://my_company.looker.com：Looker 執行個體的完整網址。
  • my_model：包含要連結之「探索」的 LookML 模型名稱。
  • my_explore：您希望資料代理程式查詢的 Looker 探索名稱。
  • my_model_2：第二個 LookML 模型的名稱，其中包含要連結的「探索」。最多可重複使用這個變數五次，為其他模型新增探索。
  • my_explore_2：您希望資料代理程式查詢的其他 Looker 探索名稱。您可以重複使用這個變數，最多納入五個探索。

存取權杖

如要與 Looker 執行個體建立連結，請使用存取權杖，詳情請參閱「使用 Conversational Analytics API 驗證及連結資料來源」。

looker_access_token = "my_access_token"
looker_instance_uri = "https://my_company.looker.com"
lookml_model = "my_model"
explore = "my_explore"
lookml_model_2 = "my_model_2"
explore_2 = "my_explore_2"

looker_explore_reference = geminidataanalytics.LookerExploreReference()
looker_explore_reference.looker_instance_uri = looker_instance_uri
looker_explore_reference.lookml_model = "my_model"
looker_explore_reference.explore = "my_explore"

looker_explore_reference2 = geminidataanalytics.LookerExploreReference()
looker_explore_reference2.looker_instance_uri = looker_instance_uri
looker_explore_reference2.lookml_model = "my_model_2"
looker_explore_reference2.explore = "my_explore_2"

credentials = geminidataanalytics.Credentials()
credentials.oauth.secret.client_id = looker_client_id
credentials.oauth.secret.client_secret = looker_client_secret

datasource_references = geminidataanalytics.DatasourceReferences()
datasource_references.looker.explore_references = [looker_explore_reference, looker_explore_reference2]

# Do not include the following line during agent creation
datasource_references.credentials = credentials

請依下列方式替換範例值：

  • my_access_token：您產生的 access_token 值，用於向 Looker 驗證。
  • https://my_company.looker.com：Looker 執行個體的完整網址。
  • my_model：包含要連結之「探索」的 LookML 模型名稱。
  • my_explore：您希望資料代理程式查詢的 Looker 探索名稱。
  • my_model_2：第二個 LookML 模型的名稱，其中包含要連結的「探索」。最多可重複使用這個變數五次，為其他模型新增探索。
  • my_explore_2：您希望資料代理程式查詢的其他 Looker 探索名稱。您可以重複使用這個變數，最多納入五個探索。

連結至 BigQuery 資料

使用對話式數據分析 API 時，可連結的 BigQuery 資料表數量沒有硬性限制。不過，連結大量資料表可能會降低準確率，或導致您超出模型的輸入權杖限制。

下列程式碼範例定義了與多個 BigQuery 資料表的連線，並包含選用結構化內容欄位的範例。如要提升代理程式效能，您可以選擇提供 BigQuery 資料表的結構化內容，例如資料表和資料欄說明、同義詞、標記和範例查詢。詳情請參閱「定義 BigQuery 資料來源的資料代理程式內容」。

bigquery_table_reference_1 = geminidataanalytics.BigQueryTableReference()
bigquery_table_reference_1.project_id = "my_project_id_1"
bigquery_table_reference_1.dataset_id = "my_dataset_id_1"
bigquery_table_reference_1.table_id = "my_table_id_1"
bigquery_table_reference_1.schema = geminidataanalytics.Schema()
bigquery_table_reference_1.schema.description = "my_table_description"
bigquery_table_reference_1.schema.fields = [
  geminidataanalytics.Field(
    name="my_column_name",
    description="my_column_description"
  )
]

bigquery_table_reference_2 = geminidataanalytics.BigQueryTableReference()
bigquery_table_reference_2.project_id = "my_project_id_2"
bigquery_table_reference_2.dataset_id = "my_dataset_id_2"
bigquery_table_reference_2.table_id = "my_table_id_2"

bigquery_table_reference_3 = geminidataanalytics.BigQueryTableReference()
bigquery_table_reference_3.project_id = "my_project_id_3"
bigquery_table_reference_3.dataset_id = "my_dataset_id_3"
bigquery_table_reference_3.table_id = "my_table_id_3"

# Connect to your data source
datasource_references = geminidataanalytics.DatasourceReferences()
datasource_references.bq.table_references = [bigquery_table_reference_1, bigquery_table_reference_2, bigquery_table_reference_3]

請依下列方式替換範例值：

  • my_project_id_1：包含要連結的 BigQuery 資料集和資料表的 Google Cloud 專案 ID。如要連線至公開資料集，請指定 bigquery-public-data。
  • my_dataset_id_1：BigQuery 資料集的 ID。例如：san_francisco。
  • my_table_id_1：BigQuery 資料表的 ID。例如：street_trees。
  • my_table_description：資料表內容和用途的選用說明。
  • my_column_name：表格中資料欄的名稱，您可以為該資料欄提供選填說明。
  • my_column_description：資料欄內容和用途的選用說明。

連結至數據分析資料

下列程式碼範例定義了與數據分析資料來源的連線。

studio_datasource_id = "my_datasource_id"

studio_references = geminidataanalytics.StudioDatasourceReference()
studio_references.datasource_id = studio_datasource_id

## Connect to your data source
datasource_references.studio.studio_references = [studio_references]

在先前的範例中，請將 my_datasource_id 替換為資料來源 ID。

連線至 AlloyDB 資料

下列程式碼範例使用 Python SDK 定義與 AlloyDB 資料庫的連線。

# Define the AlloyDB database reference
alloydb_database_reference = geminidataanalytics.AlloyDbDatabaseReference()
alloydb_database_reference.project_id = "PROJECT_ID"
alloydb_database_reference.region = "REGION"
alloydb_database_reference.cluster_id = "CLUSTER_ID"
alloydb_database_reference.instance_id = "INSTANCE_ID"
alloydb_database_reference.database_id = "DATABASE"
alloydb_database_reference.table_ids = ["TABLE_1_ID", "TABLE_2_ID"]

# Optional: Include this if you have created advanced context for the agent
agent_context_reference = geminidataanalytics.AgentContextReference()
agent_context_reference.context_set_id = f"projects/{billing_project}/locations/{location}/contextSets/your_context_set_id"

# Add the AlloyDB reference to the DatasourceReferences object
datasource_references = geminidataanalytics.DatasourceReferences()
datasource_references.alloydb.database_reference = alloydb_database_reference
# Optional:
# datasource_references.alloydb.agent_context_reference = agent_context_reference

請依下列方式替換範例值：

  • 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"。
  • your_context_set_id：如果您為代理程式建立了進階情境，請提供情境集的 ID。

連線至 MySQL 適用的 Cloud SQL 資料

如要連線至 MySQL 適用的 Cloud SQL 執行個體，請使用 geminidataanalytics SDK 中的 CloudSqlReference類別。這包括指定專案、執行個體、資料庫，以及 (選用) 代理程式要考量的特定資料表。

連線至 MySQL 適用的 Cloud SQL 資料前，請務必使用 Identity and Access Management (IAM) 驗證 MySQL 適用的 Cloud SQL。詳情請參閱「向 MySQL 適用的 Cloud SQL 和 PostgreSQL 適用的 Cloud SQL 進行驗證」。

以下範例使用 Python SDK 定義與 MySQL 適用的 Cloud SQL 資料庫的連線。

# Import the necessary library
from google.cloud import geminidataanalytics

# Define the Cloud SQL database reference
cloud_sql_database_reference = geminidataanalytics.CloudSqlDatabaseReference(
    project_id="PROJECT_ID",
    instance_id="INSTANCE_ID",
    database_id="DATABASE_ID",
    # Optional: Specify table IDs to guide the agent
    table_ids=["TABLE_1_ID", "TABLE_2_ID"]
)

# Create a CloudSqlReference object
cloud_sql_reference = geminidataanalytics.CloudSqlReference(
    database_reference=cloud_sql_database_reference
    # Optional: Include this if you have created advanced context for the agent
    # agent_context_reference=geminidataanalytics.AgentContextReference(
    # context_set_id=f"projects/{billing_project}/locations/{location}/contextSets/your_context_set_id"
    # )
)

# Add the Cloud SQL reference to the DatasourceReferences object
datasource_references = geminidataanalytics.DatasourceReferences()
datasource_references.cloud_sql = cloud_sql_reference

請依下列方式替換範例值：

  • PROJECT_ID：包含 MySQL 適用的 Cloud SQL 執行個體的專案 ID。 Google Cloud
  • INSTANCE_ID：MySQL 適用的 Cloud SQL 執行個體 ID。
  • DATABASE_ID：執行個體中的目標資料庫名稱，例如 financial。
  • TABLE_1_ID、TABLE_2_ID：(選用) 資料庫中的特定資料表名稱清單，建議資料代理程式使用，例如 "orders" 和 "customers"。
  • your_context_set_id：如果您為代理程式建立了進階情境，請提供情境集的 ID。

建立或與資料代理互動時，可以使用 datasource_references 物件。

連線至 PostgreSQL 適用的 Cloud SQL 資料

如要連線至 PostgreSQL 適用的 Cloud SQL 執行個體，請使用 geminidataanalytics SDK 中的 CloudSqlReference類別。這包括指定專案、執行個體、資料庫，以及 (選用) 代理程式要考量的特定資料表。

連線至 PostgreSQL 適用的 Cloud SQL 資料前，請務必使用 IAM 向 PostgreSQL 適用的 Cloud SQL 進行驗證。詳情請參閱「向 MySQL 適用的 Cloud SQL 和 PostgreSQL 適用的 Cloud SQL 進行驗證」。

下列範例使用 Python SDK 定義與 PostgreSQL 適用的 Cloud SQL 資料庫的連線。

# Import the necessary library
from google.cloud import geminidataanalytics

# Define the Cloud SQL database reference
cloud_sql_database_reference = geminidataanalytics.CloudSqlDatabaseReference(
    project_id="PROJECT_ID",
    instance_id="INSTANCE_ID",
    database_id="DATABASE_ID",
    # Optional: Specify table IDs to guide the agent
    table_ids=["TABLE_1_ID", "TABLE_2_ID"]
)

# Create a CloudSqlReference object
cloud_sql_reference = geminidataanalytics.CloudSqlReference(
    database_reference=cloud_sql_database_reference
    # Optional: Include this if you have created advanced context for the agent
    # agent_context_reference=geminidataanalytics.AgentContextReference(
    # context_set_id=f"projects/{billing_project}/locations/{location}/contextSets/your_context_set_id"
    # )
)

# Add the Cloud SQL reference to the DatasourceReferences object
datasource_references = geminidataanalytics.DatasourceReferences()
datasource_references.cloud_sql = cloud_sql_reference

請依下列方式替換範例值：

  • PROJECT_ID：包含 PostgreSQL 適用的 Cloud SQL 執行個體的 Google Cloud 專案 ID。
  • INSTANCE_ID：PostgreSQL 適用的 Cloud SQL 執行個體 ID。
  • DATABASE_ID：執行個體中的目標資料庫名稱，例如 postgres。
  • TABLE_1_ID、TABLE_2_ID：(選用) 資料庫中建議資料代理程式使用的特定資料表名稱清單，例如 "users" 和 "products"。
  • your_context_set_id：如果您為代理程式建立了進階情境，請提供情境集的 ID。

建立或與資料代理互動時，可以使用 datasource_references 物件。

連結至 Spanner 資料

如要連線至 Spanner 執行個體，請使用 geminidataanalytics SDK 的 SpannerReference 類別。這包括指定專案、執行個體、資料庫，以及 (選用) 代理程式要考量的特定資料表。

連線至 Spanner 資料前，請確認使用者或服務帳戶具有 spanner.databaseReader IAM 角色。詳情請參閱「套用 IAM 角色」。

下列 Python 程式碼範例會使用 Python SDK 定義與 Spanner 資料庫的連線。

# Import the necessary library
from google.cloud import geminidataanalytics

# Define the Spanner database reference
spanner_database_reference = geminidataanalytics.SpannerDatabaseReference(
    project_id="PROJECT_ID",
    instance_id="INSTANCE_ID",
    database_id="DATABASE_ID",
    # Optional: Specify table IDs to guide the agent
    table_ids=["TABLE_1_ID", "TABLE_2_ID"]
)

# Create a SpannerReference object
spanner_reference = geminidataanalytics.SpannerReference(
    database_reference=spanner_database_reference
    # Optional: Include this if you have created advanced context for the agent
    # agent_context_reference=geminidataanalytics.AgentContextReference(
    # context_set_id=f"projects/{billing_project}/locations/{location}/contextSets/your_context_set_id"
    # )
)

# Add the Spanner reference to the DatasourceReferences object
datasource_references = geminidataanalytics.DatasourceReferences()
datasource_references.spanner = spanner_reference

請依下列方式替換範例值：

  • PROJECT_ID：包含 Spanner 執行個體的 Google Cloud 專案 ID。
  • INSTANCE_ID：Spanner 執行個體的 ID。
  • DATABASE_ID：執行個體中目標資料庫的名稱。
  • TABLE_1_ID、TABLE_2_ID：(選用) 資料庫中建議資料代理程式使用的特定資料表名稱清單，例如 "Singers" 和 "Albums"。
  • your_context_set_id：如果您為代理程式建立了進階情境，請提供情境集的 ID。

建立或與資料代理互動時，可以使用 datasource_references 物件。

設定有狀態或無狀態即時通訊的脈絡

對話式數據分析 API 支援多輪對話，可讓使用者根據先前的脈絡提出後續問題。下列 Python 程式碼範例示範如何為有狀態或無狀態對話設定環境：

  • 具狀態的即時通訊： Google Cloud 儲存及管理對話記錄。有狀態的對話本質上是多輪對話，因為 API 會保留先前訊息的脈絡。您只需要在每個回合傳送當下的訊息。
  • 無狀態對話：應用程式會管理對話記錄。每則新訊息都必須包含完整的對話記錄。如需在無狀態模式下管理多輪對話的詳細範例，請參閱「建立無狀態多輪對話」。

具狀態的對話

下列程式碼範例會設定有狀態的對話內容，其中 Google Cloud 會儲存及管理對話記錄。您也可以在下列程式碼範例中加入 published_context.options.analysis.python.enabled = True 行，選擇啟用 Python 進階分析功能。

# Set up context for stateful chat
published_context = geminidataanalytics.Context()
published_context.system_instruction = system_instruction
published_context.datasource_references = datasource_references
# Optional: To enable advanced analysis with Python, include the following line:
published_context.options.analysis.python.enabled = True

無狀態對話

下列程式碼範例會設定無狀態對話的環境，您必須在每則訊息中傳送完整的對話記錄。您也可以在下列程式碼範例中加入 inline_context.options.analysis.python.enabled = True 行，選擇透過 Python 啟用進階分析。

# Set up context for stateless chat
# datasource_references.looker.credentials = credentials
inline_context = geminidataanalytics.Context()
inline_context.system_instruction = system_instruction
inline_context.datasource_references = datasource_references
# Optional: To enable advanced analysis with Python, include the following line:
inline_context.options.analysis.python.enabled = True

建立資料代理

下列 Python 程式碼範例會發出 API 要求來建立資料代理程式，然後您就可以使用該代理程式，與資料進行對話。您可以同步或非同步建立代理程式。資料代理程式會根據指定的資料來源、系統指令和內容進行設定。

您可以選擇在建立資料代理程式時提供 kms_key，藉此使用 CMEK 保護資料代理程式。CMEK 僅支援 Looker 資料來源。詳情請參閱「客戶自行管理的加密金鑰 (CMEK)」一文。

同步

  data_agent_id = "data_agent_1"

  # Optional: If using CMEK, replace the empty strings with your KMS key details.
  key_ring = ""    # KEY_RING_NAME
  key_name = ""    # KEY_NAME
  key_project = "" # KMS_PROJECT_ID (Defaults to billing_project if empty)

  data_agent = geminidataanalytics.DataAgent()
  data_agent.data_analytics_agent.published_context = published_context
  data_agent.name = f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}" # Optional

  # Optional: Add CMEK key if details are provided
  if key_ring and key_name:
    if not key_project:
      key_project = billing_project
    kms_key_data_agent = f"projects/{key_project}/locations/{location}/keyRings/{key_ring}/cryptoKeys/{key_name}"
    data_agent.kms_key = kms_key_data_agent # Add for CMEK

  request = geminidataanalytics.CreateDataAgentRequest(
    parent=f"projects/{billing_project}/locations/{location}",
    data_agent_id=data_agent_id, # Optional
    data_agent=data_agent,
  )

  try:
    response = data_agent_client.create_data_agent_sync(request=request)
    print("Data Agent created")
    print(response)
  except Exception as e:
    print(f"Error creating Data Agent: {e}")

非同步

  data_agent_id = "data_agent_1"

  # Optional: If using CMEK, replace the empty strings with your KMS key details.
  key_ring = ""    # KEY_RING_NAME
  key_name = ""    # KEY_NAME
  key_project = "" # KMS_PROJECT_ID (Defaults to billing_project if empty)

  data_agent = geminidataanalytics.DataAgent()
  data_agent.data_analytics_agent.published_context = published_context
  data_agent.name = f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}" # Optional

  # Optional: Add CMEK key if details are provided
  if key_ring and key_name:
    if not key_project:
      key_project = billing_project
    kms_key_data_agent = f"projects/{key_project}/locations/{location}/keyRings/{key_ring}/cryptoKeys/{key_name}"
    data_agent.kms_key = kms_key_data_agent # Add for CMEK

  request = geminidataanalytics.CreateDataAgentRequest(
    parent=f"projects/{billing_project}/locations/{location}",
    data_agent_id=data_agent_id, # Optional
    data_agent=data_agent,
  )

  try:
    data_agent_client.create_data_agent(request=request)
    print("Data Agent created")
  except Exception as e:
    print(f"Error creating Data Agent: {e}")

在先前的範例中，請依下列方式替換值：

  • data_agent_1：資料代理程式的專屬 ID。
  • KEY_RING_NAME：如果使用 CMEK，這是 Cloud KMS 金鑰環的名稱。
  • KEY_NAME：如果使用 CMEK，請輸入 Cloud KMS 金鑰的名稱。
  • KMS_PROJECT_ID：如要使用 CMEK，請提供金鑰所在的專案 ID。

建立對話

下列 Python 程式碼範例會發出 API 要求，建立對話。

您可以在建立對話時提供 kms_key，選擇使用 CMEK 保護對話。CMEK 僅支援 Looker 資料來源。詳情請參閱「客戶自行管理的加密金鑰 (CMEK)」。

# Initialize request arguments
data_agent_id = "data_agent_1"
conversation_id = "conversation_1"


# Optional: If using CMEK, replace the empty strings with your KMS key details.
key_ring = ""    # KEY_RING_NAME
key_name = ""    # KEY_NAME
key_project = "" # KMS_PROJECT_ID (Defaults to billing_project if empty)

if key_project == "":
  key_project = billing_project
kms_key_conversation = f"projects/{key_project}/locations/{location}/keyRings/{key_ring}/cryptoKeys/{key_name}"

conversation = geminidataanalytics.Conversation()
conversation.agents = [f'projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}']
conversation.name = f"projects/{billing_project}/locations/{location}/conversations/{conversation_id}"

# Optional: Add CMEK key if details are provided
if key_ring and key_name:
  if not key_project:
    key_project = billing_project
  kms_key_conversation = f"projects/{key_project}/locations/{location}/keyRings/{key_ring}/cryptoKeys/{key_name}"
  conversation.kms_key = kms_key_conversation # Add for CMEK

request = geminidataanalytics.CreateConversationRequest(
  parent=f"projects/{billing_project}/locations/{location}",
  conversation_id=conversation_id,
  conversation=conversation,
)

# Make the request
response = data_chat_client.create_conversation(request=request)

# Handle the response
print(response)

請依下列方式替換範例值：

  • data_agent_1：資料代理程式的 ID，如「建立資料代理程式」中的程式碼範例區塊所定義。
  • conversation_1：對話的專屬 ID。
  • KEY_RING_NAME：如果使用 CMEK，這是 Cloud KMS 金鑰環的名稱。
  • KEY_NAME：如果使用 CMEK，請輸入 Cloud KMS 金鑰的名稱。
  • KMS_PROJECT_ID：如要使用 CMEK，請提供金鑰所在的專案 ID。

管理資料代理程式和對話

下列程式碼範例說明如何使用 Conversational Analytics API 管理資料代理程式和對話。您可以執行下列工作：

取得資料代理

以下 Python 程式碼範例示範如何發出 API 要求，擷取先前建立的資料代理程式。

# Initialize request arguments
data_agent_id = "data_agent_1"
request = geminidataanalytics.GetDataAgentRequest(
  name=f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
)

# Make the request
response = data_agent_client.get_data_agent(request=request)

# Handle the response
print(response)

在先前的範例中，請將 data_agent_1 值替換為要擷取的資料代理程式專屬 ID。

列出資料代理

下列程式碼示範如何呼叫 list_data_agents 方法，列出特定專案的所有資料代理程式。如要列出所有代理程式，您必須擁有專案的 geminidataanalytics.dataAgents.list 權限。如要進一步瞭解哪些 IAM 角色包含這項權限，請參閱預先定義的角色清單。

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
request = geminidataanalytics.ListDataAgentsRequest(
  parent=f"projects/{billing_project}/locations/{location}",
)

# Make the request
page_result = data_agent_client.list_data_agents(request=request)

# Handle the response
for response in page_result:
  print(response)

將 YOUR-BILLING-PROJECT 替換為帳單專案的 ID。

列出可存取的資料代理

下列程式碼示範如何呼叫 list_accessible_data_agents 方法，列出指定專案的所有可存取資料代理程式。

billing_project = "YOUR-BILLING-PROJECT"
creator_filter = "YOUR-CREATOR-FILTER"
location = "global"
request = geminidataanalytics.ListAccessibleDataAgentsRequest(
  parent=f"projects/{billing_project}/locations/{location}",
  creator_filter=creator_filter
)

# Make the request
page_result = data_agent_client.list_accessible_data_agents(request=request)

# Handle the response
for response in page_result:
  print(response)

請依下列方式替換範例值：

  • YOUR-BILLING-PROJECT：帳單專案的 ID。
  • YOUR-CREATOR-FILTER：根據資料代理程式建立者套用的篩選條件。可能的值包括 NONE (預設值)、CREATOR_ONLY 和 NOT_CREATOR_ONLY。

更新資料代理程式

下列程式碼範例說明如何呼叫資料代理程式資源的 update_data_agent_sync 或 update_data_agent 方法，同步或非同步更新資料代理程式。這項要求需要 DataAgent 物件，其中包含要變更的欄位新值，以及 update_mask 參數 (採用 FieldMask 物件)，用於指定要更新的欄位。

如要更新資料代理程式，您必須擁有該代理程式的 geminidataanalytics.dataAgents.update IAM 權限。如要進一步瞭解哪些 IAM 角色包含這項權限，請參閱預先定義的角色清單。

同步

data_agent_id = "data_agent_1"
billing_project = "YOUR-BILLING-PROJECT"
data_agent = geminidataanalytics.DataAgent()
data_agent.data_analytics_agent.published_context = published_context
data_agent.name = f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"
data_agent.description = "Updated description of the data agent."

update_mask = field_mask_pb2.FieldMask(paths=['description', 'data_analytics_agent.published_context'])

request = geminidataanalytics.UpdateDataAgentRequest(
  data_agent=data_agent,
  update_mask=update_mask,
)

try:
  # Make the request
  response = data_agent_client.update_data_agent_sync(request=request)
  print("Data Agent Updated")
  print(response)
except Exception as e:
  print(f"Error updating Data Agent: {e}")

非同步

data_agent_id = "data_agent_1"
billing_project = "YOUR-BILLING-PROJECT"
data_agent = geminidataanalytics.DataAgent()
data_agent.data_analytics_agent.published_context = published_context
data_agent.name = f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"
data_agent.description = "Updated description of the data agent."

update_mask = field_mask_pb2.FieldMask(paths=['description', 'data_analytics_agent.published_context'])

request = geminidataanalytics.UpdateDataAgentRequest(
  data_agent=data_agent,
  update_mask=update_mask,
)

try:
  # Make the request
  data_agent_client.update_data_agent(request=request)
  print("Data Agent Updated")
except Exception as e:
  print(f"Error updating Data Agent: {e}")

請依下列方式替換範例值：

  • data_agent_1：要更新的資料代理程式 ID。
  • YOUR-BILLING-PROJECT：帳單專案的 ID。
  • Updated description of the data agent.：更新後資料代理程式的說明。

取得資料代理程式的身分與存取權管理政策

以下程式碼範例示範如何使用 get_iam_policy 方法擷取資料代理程式的 IAM 政策。要求會指定資料代理程式資源路徑。

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
data_agent_id = "data_agent_1"

resource = f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"
request = iam_policy_pb2.GetIamPolicyRequest(
      resource=resource,
    )
try:
  response = data_agent_client.get_iam_policy(request=request)
  print("IAM Policy fetched successfully!")
  print(f"Response: {response}")
except Exception as e:
  print(f"Error setting IAM policy: {e}")

請依下列方式替換範例值：

  • YOUR-BILLING-PROJECT：帳單專案的 ID。
  • data_agent_1：要取得 IAM 政策的資料代理程式 ID。

設定資料代理程式的 IAM 政策

如要共用代理程式，可以使用 set_iam_policy 方法，將 IAM 角色指派給特定代理程式的使用者。要求包含繫結，可指定要將哪些角色指派給哪些使用者。

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
data_agent_id = "data_agent_1"
role = "roles/geminidataanalytics.dataAgentEditor"
users = "EMAIL_ADDRESS"

resource = f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"

# Construct the IAM policy
binding = policy_pb2.Binding(
  role=role,
  members= [f"user:{i.strip()}" for i in users.split(",")]
)

policy = policy_pb2.Policy(bindings=[binding])

# Create the request
request = iam_policy_pb2.SetIamPolicyRequest(
  resource=resource,
  policy=policy
)

# Send the request
try:
  response = data_agent_client.set_iam_policy(request=request)
  print("IAM Policy set successfully!")
  print(f"Response: {response}")
except Exception as e:
  print(f"Error setting IAM policy: {e}")

請依下列方式替換範例值：

  • YOUR-BILLING-PROJECT：帳單專案的 ID。
  • data_agent_1：要設定 IAM 政策的資料代理程式 ID。
  • EMAIL_ADDRESSES：以半形逗號分隔的使用者電子郵件地址清單，您要將指定角色授予這些使用者。

刪除資料代理程式

以下程式碼範例說明如何使用 delete_data_agent_sync 或 delete_data_agent 方法，同步或非同步虛刪除資料代理程式。虛刪除代理程式時，該代理程式會遭到刪除，但仍可在 30 天內擷取。要求會指定資料代理程式資源網址。

同步

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
data_agent_id = "data_agent_1"

request = geminidataanalytics.DeleteDataAgentRequest(
  name=f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
)

try:
  # Make the request
  data_agent_client.delete_data_agent_sync(request=request)
  print("Data Agent Deleted")
except Exception as e:
  print(f"Error deleting Data Agent: {e}")

非同步

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
data_agent_id = "data_agent_1"

request = geminidataanalytics.DeleteDataAgentRequest(
  name=f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
)

try:
  # Make the request
  data_agent_client.delete_data_agent(request=request)
  print("Data Agent Deleted")
except Exception as e:
  print(f"Error deleting Data Agent: {e}")

請依下列方式替換範例值：

  • YOUR-BILLING-PROJECT：帳單專案的 ID。
  • data_agent_1：要刪除的資料代理程式 ID。

取得對話

下列程式碼範例示範如何使用 get_conversation 方法，擷取現有對話的相關資訊。要求會指定對話資源路徑。

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
conversation_id = "conversation_1"

request = geminidataanalytics.GetConversationRequest(
  name = f"projects/{billing_project}/locations/{location}/conversations/{conversation_id}"
)

# Make the request
response = data_chat_client.get_conversation(request=request)

# Handle the response
print(response)

請依下列方式替換範例值：

  • YOUR-BILLING-PROJECT：帳單專案的 ID。
  • conversation_1：要擷取的對話 ID。

列出對話

下列程式碼範例示範如何呼叫 list_conversations 方法，列出特定專案的對話。要求會指定父項資源網址，也就是專案和位置 (例如 projects/my-project/locations/global)。

根據預設，這個方法會傳回您建立的對話。管理員 (具備 cloudaicompanion.topicAdmin IAM 角色的使用者) 可以查看專案中的所有對話。

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
request = geminidataanalytics.ListConversationsRequest(
  parent=f"projects/{billing_project}/locations/{location}",
)

# Make the request
response = data_chat_client.list_conversations(request=request)

# Handle the response
print(response)

將 YOUR-BILLING-PROJECT 替換為已啟用必要 API 的帳單專案 ID。

列出對話中的訊息

下列程式碼範例說明如何使用 list_messages 方法擷取對話中的所有訊息。要求會指定對話資源路徑。

如要列出訊息，您必須具備對話的 cloudaicompanion.topics.get 權限。

billing_project = "YOUR-BILLING-PROJECT"
location = "global"

conversation_id = "conversation_1"

request = geminidataanalytics.ListMessagesRequest(
  parent=f"projects/{billing_project}/locations/{location}/conversations/{conversation_id}",
)

# Make the request
response = data_chat_client.list_messages(request=request)

# Handle the response
print(response)

請依下列方式替換範例值：

  • YOUR-BILLING-PROJECT：帳單專案的 ID。
  • conversation_1：要列出訊息的對話 ID。

刪除對話

下列程式碼範例說明如何刪除對話。管理員 (具備 cloudaicompanion.topicAdmin Identity and Access Management 角色的使用者) 或具備 cloudaicompanion.topics.delete Identity and Access Management 權限的使用者，可以刪除專案中的對話。

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
conversation_id = "conversation_1"

request = geminidataanalytics.DeleteConversationRequest(
  name = f"projects/{billing_project}/locations/{location}/conversations/{conversation_id}"
)

# Make the request
response = data_chat_client.delete_conversation(request=request)

# Handle the response
print(response)

請依下列方式替換範例值：

  • YOUR-BILLING-PROJECT：帳單專案的 ID。
  • conversation_1：要列出訊息的對話 ID。

使用 API 提問

建立資料代理程式，並在有狀態的對話中建立對話後，即可將查詢傳送至代理程式。下列程式碼範例說明如何呼叫 data_chat_client.chat 方法。這些範例會使用您在「設定有狀態或無狀態對話的環境」中設定的環境變數 (例如資料來源和系統指令)。

傳送查詢時，API 會傳回 Message 物件的串流。這個串流可以包含不同類型的訊息，包括文字、資料表和圖表。簡訊可提供代理的推論過程、進度報告或最終答案。每則簡訊的用途會以 TextType 值表示：

  • THOUGHT：顯示代理程式的內部思考過程，瞭解代理程式如何規劃查詢回覆。THOUGHT 訊息會逐步深入瞭解代理程式的推論和決策過程，並包含兩部分：parts[0] 是思考摘要，簡要總結完整的思考文字，而 parts[1] 則是完整的思考文字。
  • PROGRESS：回報代理程式的動作進度，例如資料檢索或正在叫用的工具。這個值只會針對 Looker 資料來源傳回，並包含兩部分：parts[0] 是摘要，parts[1] 則是完整進度文字。
  • FINAL_RESPONSE：提供查詢的最終答案。

下列程式碼範例會使用 定義輔助函式中定義的 show_message 輔助函式，處理及顯示串流中的每則訊息。如要瞭解如何在使用 Looker 資料來源時，在使用者介面中顯示這些訊息，請參閱「顯示 Looker 資料來源的代理程式回覆」。

具狀態的對話

傳送含有 Conversation 參照的具狀態對話要求

您可以參照先前建立的 Conversation 資源，將有狀態的即時通訊要求傳送給資料代理程式。

# Create a request that contains a single user message (your question)
question = "Which species of tree is most prevalent?"
messages = [geminidataanalytics.Message()]
messages[0].user_message.text = question

data_agent_id = "data_agent_1"
conversation_id = "conversation_1"

# Create a conversation_reference
conversation_reference = geminidataanalytics.ConversationReference()
conversation_reference.conversation = f"projects/{billing_project}/locations/{location}/conversations/{conversation_id}"
conversation_reference.data_agent_context.data_agent = f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"
# conversation_reference.data_agent_context.credentials = credentials

# Form the request
request = geminidataanalytics.ChatRequest(
    parent = f"projects/{billing_project}/locations/{location}",
    messages = messages,
    conversation_reference = conversation_reference
)

# Make the request
stream = data_chat_client.chat(request=request, timeout=300 #custom timeout up to 600s)

# Handle the response
for response in stream:
    show_message(response)

請依下列方式替換範例值：

  • Which species of tree is most prevalent?：要傳送給資料代理程式的自然語言問題。
  • data_agent_1：資料代理的專屬 ID，如「建立資料代理」一文所述。
  • conversation_1：對話的專屬 ID，如「建立對話」中所定義。

無狀態對話

下列程式碼範例說明在設定無狀態即時通訊的環境後，如何將查詢傳送至資料代理程式。您可以參照先前定義的 DataAgent 資源，或在要求中使用內嵌環境，傳送無狀態查詢。

傳送含有 DataAgent 參照的無狀態即時通訊要求

您可以參照先前建立的 DataAgent 資源，將查詢傳送至資料代理程式。

# Create a request that contains a single user message (your question)
question = "Which species of tree is most prevalent?"
messages = [geminidataanalytics.Message()]
messages[0].user_message.text = question

data_agent_id = "data_agent_1"

data_agent_context = geminidataanalytics.DataAgentContext()
data_agent_context.data_agent = f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"
# data_agent_context.credentials = credentials

# Form the request
request = geminidataanalytics.ChatRequest(
    parent=f"projects/{billing_project}/locations/{location}",
    messages=messages,
    data_agent_context = data_agent_context
)

# Make the request
stream = data_chat_client.chat(request=request, timeout=300 #custom timeout up to 600s)

# Handle the response
for response in stream:
    show_message(response)

請依下列方式替換範例值：

  • Which species of tree is most prevalent?：要傳送給資料代理程式的自然語言問題。
  • data_agent_1：資料代理的專屬 ID，如「建立資料代理」一文所述。

傳送含有內嵌背景資訊的無狀態對話要求

以下程式碼範例說明如何使用 inline_context 參數，在無狀態的即時通訊要求中直接提供背景資訊。

# Create a request that contains a single user message (your question)
question = "Which species of tree is most prevalent?"
messages = [geminidataanalytics.Message()]
messages[0].user_message.text = question

request = geminidataanalytics.ChatRequest(
    inline_context=inline_context,
    parent=f"projects/{billing_project}/locations/{location}",
    messages=messages,
)

# Make the request
stream = data_chat_client.chat(request=request, timeout=300 #custom timeout up to 600s)

# Handle the response
for response in stream:
    show_message(response)

在上一個範例中，請將 Which species of tree is most prevalent? 換成自然語言問題，傳送給資料代理程式。

建立無狀態多輪對話

如要在無狀態對話中詢問後續問題，應用程式必須管理對話內容，在每次提出新要求時傳送完整訊息記錄。以下範例說明如何參照資料代理程式，或使用內嵌脈絡直接提供資料來源，建立多輪對話。

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

data_agent_id = "data_agent_1"

# Use data agent context
data_agent_context = geminidataanalytics.DataAgentContext()
data_agent_context.data_agent = f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"
# data_agent_context.credentials = credentials

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

  message = geminidataanalytics.Message()
  message.user_message.text = msg

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

  request = geminidataanalytics.ChatRequest(
    parent=f"projects/{billing_project}/locations/{location}",
    messages=conversation_messages,
    # Use data agent context
    data_agent_context=data_agent_context,
    # Use inline context
    # inline_context=inline_context,
  )

  # Make the request
  stream = data_chat_client.chat(request=request, timeout=300 #custom timeout up to 600s)

  # Handle the response
  for response in stream:
    show_message(response)
    conversation_messages.append(response)

# Send the 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?")

在先前的範例中，請依下列方式替換範例值：

  • data_agent_1：資料代理程式的專屬 ID，如「建立資料代理程式」中的範例程式碼區塊所定義。
  • Which species of tree is most prevalent?：要傳送給資料代理程式的自然語言問題。
  • Can you show me the results as a bar chart?：根據上一個問題延伸或修正的後續問題。

定義輔助函式

下列程式碼範例包含先前程式碼範例中使用的輔助函式定義。這些函式有助於剖析 API 的回應並顯示結果。

import json as json_lib
import textwrap
import time

import altair as alt
import IPython
import pandas as pd
import proto
import requests

from google.protobuf.json_format import MessageToDict, MessageToJson
from IPython.display import display, HTML
from pygments import highlight, lexers, formatters
from google.protobuf import field_mask_pb2
from google.iam.v1 import policy_pb2
from google.iam.v1 import iam_policy_pb2

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 display_schema(data):
  fields = getattr(data, "fields")
  df = pd.DataFrame({
    "Column": map(lambda field: getattr(field, 'name'), fields),
    "Type": map(lambda field: getattr(field, 'type'), fields),
    "Description": map(lambda field: getattr(field, 'description', '-'), fields),
    "Mode": map(lambda field: getattr(field, 'mode'), fields)
  })
  display(df)

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

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

def format_bq_table_ref(table_ref):
  return '{}.{}.{}'.format(table_ref.project_id, table_ref.dataset_id, table_ref.table_id)

def display_datasource(datasource):
  source_name = ''
  if 'studio_datasource_id' in datasource:
    source_name = getattr(datasource, 'studio_datasource_id')
  elif 'looker_explore_reference' in datasource:
    source_name = format_looker_table_ref(getattr(datasource, 'looker_explore_reference'))
  else:
    source_name = format_bq_table_ref(getattr(datasource, 'bigquery_table_reference'))

  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(f"Query name: {query.name}")
    if "question" in query:
      print(f"Question: {query.question}")
    if "datasources" in query:
      print("Data sources:")
      for datasource in query.datasources:
        display_datasource(datasource)
  elif "generated_sql" in resp:
    display_section_title("SQL generated")
    print(resp.generated_sql)
  elif "result" in resp:
    display_section_title("Data retrieved")

    fields = [field.name for field in resp.result.schema.fields]
    d = {}
    for el in resp.result.data:
      for field in fields:
        if field in d:
          d[field].append(el[field])
        else:
          d[field] = [el[field]]

    display(pd.DataFrame(d))

def handle_chart_response(resp):
  def _value_to_dict(v):
    if isinstance(v, proto.marshal.collections.maps.MapComposite):
      return _map_to_dict(v)
    elif isinstance(v, proto.marshal.collections.RepeatedComposite):
      return [_value_to_dict(el) for el in v]
    elif isinstance(v, (int, float, str, bool)):
      return v
    else:
      return MessageToDict(v)

  def _map_to_dict(d):
    out = {}
    for k in d:
      if isinstance(d[k], proto.marshal.collections.maps.MapComposite):
        out[k] = _map_to_dict(d[k])
      else:
        out[k] = _value_to_dict(d[k])
    return out

  if 'query' in resp:
    print(resp.query.instructions)
  elif 'result' in resp:
    vegaConfig = resp.result.vega_config
    vegaConfig_dict = _map_to_dict(vegaConfig)
    alt.Chart.from_json(json_lib.dumps(vegaConfig_dict)).display();

def show_message(msg):
  m = msg.system_message
  if 'text' in m:
    handle_text_response(getattr(m, 'text'))
  elif 'schema' in m:
    handle_schema_response(getattr(m, 'schema'))
  elif 'data' in m:
    handle_data_response(getattr(m, 'data'))
  elif 'chart' in m:
    handle_chart_response(getattr(m, 'chart'))
  print('\n')