Criar um agente de dados usando HTTP e Python

Nesta página, orientamos você a usar o Python para fazer solicitações HTTP à API Conversational Analytics (acessada por geminidataanalytics.googleapis.com).

O exemplo de código Python nesta página mostra como concluir as seguintes tarefas:

Uma versão completa do exemplo de código está incluída no final da página, junto com as funções auxiliares usadas para transmitir a resposta da API.

Configurar as definições iniciais e a autenticação

O exemplo de código Python a seguir executa estas tarefas:

  • Importa as bibliotecas necessárias do Python.
  • Recebe um token de acesso para autenticação HTTP usando a CLI do Google Cloud.
  • Define variáveis para o projeto de faturamento, local e instruções do sistema.
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 = 'YOUR-BILLING-PROJECT'
location = "global"
system_instruction = 'YOUR-SYSTEM-INSTRUCTIONS'

Substitua os valores de exemplo da seguinte forma:

  • YOUR-BILLING-PROJECT: o ID do projeto de faturamento em que você ativou as APIs necessárias.
  • YOUR-SYSTEM-INSTRUCTIONS: instruções do sistema para orientar o comportamento do agente e personalizá-lo de acordo com suas necessidades de dados. Por exemplo, é possível usar instruções do sistema para definir termos comerciais, controlar o tamanho da resposta ou definir a formatação de dados. O ideal é definir instruções do sistema usando o formato YAML recomendado em Escrever instruções eficazes para o sistema para fornecer orientações detalhadas e estruturadas.

Autenticar no Looker

Se você planeja se conectar a uma fonte de dados do Looker, será necessário fazer a autenticação na instância do Looker.

Como usar chaves de API

O exemplo de código Python a seguir demonstra como autenticar seu agente em uma instância do Looker usando chaves de API.

looker_credentials = {
    "oauth": {
        "secret": {
            "client_id": "YOUR-LOOKER-CLIENT-ID",
            "client_secret": "YOUR-LOOKER-CLIENT-SECRET",
        }
    }
}

Substitua os valores de exemplo da seguinte forma:

  • YOUR-LOOKER-CLIENT-ID: o ID do cliente da chave de API do Looker gerada.
  • YOUR-LOOKER-CLIENT-SECRET: a chave secreta do cliente da chave de API do Looker gerada.

Como usar tokens de acesso

O exemplo de código Python a seguir demonstra como autenticar seu agente em uma instância do Looker usando tokens de acesso.

looker_credentials = {
    "oauth": {
        "token": {
            "access_token": "YOUR-TOKEN",
        }
    }
}

Substitua os valores de exemplo da seguinte forma:

  • YOUR-TOKEN: o valor access_token gerado para autenticar no Looker.

Autenticar no AlloyDB para PostgreSQL

Se você planeja se conectar a uma fonte de dados do AlloyDB, use a CLI gcloud para fazer a autenticação na instância do AlloyDB.

  1. Ative a autenticação do Identity and Access Management (IAM) no AlloyDB ativando a flag alloydb.iam_authentication na instância do AlloyDB.
  2. Conceda papéis do IAM. O principal (usuário ou conta de serviço) que chama a API Análises de conversação precisa ter as permissões adequadas do IAM no projetoGoogle Cloud . Isso geralmente inclui funções como alloydb.databaseUser.
  3. Crie um usuário de banco de dados correspondente. Crie um usuário do AlloyDB que espelhe o principal do IAM, usando o endereço de e-mail do usuário ou da conta de serviço como o nome de usuário.

Para mais informações, consulte Autenticação do banco de dados do IAM.

Autenticar no Cloud SQL para MySQL e no Cloud SQL para PostgreSQL

Se você planeja se conectar a uma fonte de dados do Cloud SQL para MySQL ou do Cloud SQL para PostgreSQL, autentique-se na instância do Cloud SQL.

  1. Ative a autenticação do banco de dados do IAM. Verifique se a flag cloudsql.iam_authentication está definida como on na sua instância do Cloud SQL para MySQL. É possível adicionar isso ao criar ou fazer patch em uma instância.

    gcloud sql instances patch INSTANCE_NAME --database-flags
cloudsql.iam_authentication=on

  2. Conceda os papéis do IAM necessários.

    O principal (usuário ou conta de serviço) que está tentando se conectar precisa das permissões cloudsql.instances.login e cloudsql.instances.connect do IAM no projeto. Para conceder essas funções e permissões, faça o seguinte:

    1. Conceda o papel roles/cloudsql.instanceUser. Isso inclui permissões para fazer login na instância (cloudsql.instances.login) e se conectar usando o proxy (cloudsql.instances.connect).
    2. Outra opção é conceder uma função personalizada que contenha pelo menos cloudsql.instances.login e cloudsql.instances.connect.

  3. Adicione o principal do IAM como um usuário do banco de dados. Crie um usuário na instância do Cloud SQL que corresponda ao endereço de e-mail do principal do IAM.

    Para uma conta de serviço, execute o seguinte comando:

    gcloud sql users create SERVICE_ACCOUNT_EMAIL --instance=INSTANCE_NAME
--type=cloud_iam_service_account

    Para uma conta de usuário, execute o seguinte comando:

    gcloud sql users create USER_EMAIL --instance=INSTANCE_NAME
--type=cloud_iam_user

  4. Conceda privilégios de banco de dados. A conexão com o IAM autentica o usuário no banco de dados, mas não concede permissões automaticamente no banco de dados. É necessário usar comandos padrão do PostgreSQL, como GRANT, para conceder permissões a esse novo usuário do banco de dados em objetos específicos, como tabelas e esquemas.

  5. Conecte-se como um superusuário, como o usuário postgres padrão, e execute os seguintes comandos:

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

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

  6. Conecte-se usando a autenticação do IAM.

    Normalmente, os clientes se conectam pelo proxy do Cloud SQL Auth ou por uma biblioteca de conector, que processa a troca de tokens. O nome de usuário do banco de dados é o e-mail do principal do IAM. As senhas não são usadas. Em vez disso, um token OAuth 2.0, geralmente processado automaticamente pelo proxy Auth do Cloud SQL ou pela Google Cloud CLI, é usado.

    O exemplo a seguir usa psql e a CLI gcloud para gerar um token:

    PGPASSWORD=$(gcloud sql generate-login-token) psql --host=HOST_IP \
--username=IAM_PRINCIPAL_EMAIL --dbname=DATABASE_NAME\

    Para mais informações, consulte Autenticação do IAM.

    Substitua os valores de exemplo da seguinte forma:

    • INSTANCE_NAME: o ID da instância do Cloud SQL que você quer corrigir para ativar a autenticação de banco de dados do IAM.
    • SERVICE_ACCOUNT_EMAIL: o endereço de e-mail da conta de serviço.
    • USER_EMAIL: o endereço de e-mail da conta de usuário.
    • HOST_IP: o endereço IP da instância do Cloud SQL.
    • IAM_PRINCIPAL_EMAIL: o endereço de e-mail do principal do IAM, por exemplo, conta de serviço ou usuário.
    • DATABASE_NAME: o nome do banco de dados a ser conectado.

Autenticar no Spanner

Se você planeja se conectar a uma fonte de dados do Spanner, é necessário fazer a autenticação na instância do Spanner.

  1. Conceda papéis do IAM. O principal (usuário ou conta de serviço) que chama a API Análises de conversação precisa ter as permissões adequadas do IAM no projetoGoogle Cloud . Isso geralmente inclui funções como databaseUser.
  2. Crie papéis de banco de dados abstratos correspondentes no Spanner usando a linguagem de definição de dados (DDL).
  3. Use políticas do IAM para permitir que os principais do IAM assumam esses papéis de banco de dados.

Para mais informações, consulte Privilégios de controle de acesso detalhado.

Conectar a uma fonte de dados

As seções a seguir mostram como definir os detalhes da conexão para as fontes de dados do seu agente. Seu agente pode se conectar aos dados das seguintes maneiras:

Conectar aos dados do Looker

O exemplo de código a seguir define uma conexão com uma análise detalhada do Looker. Para estabelecer uma conexão com uma instância do Looker, verifique se você gerou chaves de API do Looker, conforme descrito em Autenticar e conectar a uma fonte de dados com a API Conversational Analytics. É possível se conectar a até cinco Análises do Looker por vez com a API Análises de conversação.

Ao se conectar a uma fonte de dados do Looker, observe o seguinte:

  • Você pode consultar qualquer Análise incluída em uma conversa.
  • Um agente só pode consultar uma análise por vez. Não é possível fazer consultas em várias análises detalhadas ao mesmo tempo.
  • Um agente pode consultar várias análises detalhadas na mesma conversa.

  • Um agente pode consultar várias análises detalhadas em uma conversa que inclui perguntas com várias partes ou em conversas com perguntas de acompanhamento.

    Por exemplo, um usuário conecta duas análises detalhadas, uma chamada cat-explore e outra chamada dog-explore. O usuário insere a pergunta "O que é maior: a quantidade de gatos ou a quantidade de cachorros?" Isso criaria duas consultas: uma para contar o número de gatos em cat-explore e outra para contar o número de cachorros em dog-explore. O agente compara os números das duas consultas depois de concluí-las.

O exemplo de código a seguir define uma conexão com várias análises detalhadas do Looker. Para melhorar a performance do agente, você pode fornecer consultas de ouro como contexto estruturado para suas análises detalhadas. Para mais informações, consulte Definir o contexto do agente de dados para fontes de dados do Looker.

looker_instance_uri = "https://your_company.looker.com"
looker_data_source = {
    "looker": {
        "explore_references": [
            {
                "looker_instance_uri": looker_instance_uri,
                "lookml_model": "your_model",
                "explore": "your_explore",
            },
            {
                "looker_instance_uri": looker_instance_uri,
                "lookml_model": "your_model_2",
                "explore": "your_explore_2",
            },
            # Add up to 5 total Explore references
        ],
       # Do not include the following line during agent creation
       "credentials": looker_credentials
    }
}

Substitua os valores de exemplo da seguinte forma:

  • https://your_company.looker.com: o URL completo da sua instância do Looker.
  • your_model: o nome do modelo do LookML que inclui a Análise de destino.
  • your_explore: o nome da Análise do Looker que o agente de dados deve consultar.
  • your_model_2: o nome do segundo modelo do LookML que inclui a Análise de destino. Você pode repetir essa variável para outros modelos em até cinco análises.
  • your_explore_2: o nome da outra Análise do Looker que o agente de dados deve consultar. Você pode repetir essa variável para incluir até cinco análises detalhadas.

Conectar-se aos dados do BigQuery

Com a API Análises de conversação, não há limites rígidos para o número de tabelas do BigQuery a que você pode se conectar. No entanto, conectar um grande número de tabelas pode reduzir a acurácia ou fazer com que você exceda o limite de tokens de entrada do modelo.

O exemplo de código a seguir define uma conexão com várias tabelas do BigQuery e inclui exemplos de campos de contexto estruturado opcionais. Para melhorar a performance do agente, você pode fornecer um contexto estruturado para suas tabelas do BigQuery, como descrições de tabelas e colunas, sinônimos, tags e exemplos de consultas. Para mais informações, consulte Definir o contexto do agente de dados para fontes de dados do BigQuery.

bigquery_data_sources = {
    "bq": {
        "tableReferences": [
            {
                "projectId": "my_project_id",
                "datasetId": "my_dataset_id",
                "tableId": "my_table_id",
                "schema": {
                    "description": "my_table_description",
                    "fields": [{
                        "name": "my_column_name",
                        "description": "my_column_description"
                    }]
                }
            },
            {
                "projectId": "my_project_id_2",
                "datasetId": "my_dataset_id_2",
                "tableId": "my_table_id_2"
            },
            {
                "projectId": "my_project_id_3",
                "datasetId": "my_dataset_id_3",
                "tableId": "my_table_id_3"
            },
        ]
    }
}

Substitua os valores de exemplo da seguinte forma:

  • my_project_id: o ID do projeto do Google Cloud que contém o conjunto de dados e a tabela do BigQuery a que você quer se conectar. Para se conectar a um conjunto de dados público, especifique bigquery-public-data.
  • my_dataset_id: ID do conjunto de dados do BigQuery.
  • my_table_id: ID da tabela do BigQuery.
  • my_table_description: uma descrição opcional do conteúdo e da finalidade da tabela.
  • my_column_name: o nome de uma coluna na tabela para a qual você está fornecendo uma descrição opcional.
  • my_column_description: uma descrição opcional do conteúdo e da finalidade da coluna.

Conectar aos dados do Looker Studio

O exemplo de código a seguir define uma conexão com uma fonte de dados do Looker Studio.

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

Substitua studio_datasource_id pelo ID da fonte de dados.

Conectar-se aos dados do AlloyDB

A API Análises de conversação usa um campo database_reference no campo alloydb para se conectar ao cluster do AlloyDB.

O exemplo a seguir define uma conexão com um banco de dados do 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"
          # }

    }
}

Substitua os valores de exemplo da seguinte forma:

  • PROJECT_ID: o ID do projeto Google Cloud que contém o cluster do AlloyDB.
  • REGION: a região em que o cluster do AlloyDB reside, por exemplo, us-central1.
  • CLUSTER_ID: o ID do cluster do AlloyDB.
  • INSTANCE_ID: o ID da instância do AlloyDB.
  • DATABASE: o nome do banco de dados de destino, por exemplo, financial.
  • TABLE_1_ID, TABLE_2_ID: opcional. Uma lista das tabelas no banco de dados que o agente de dados é sugerido para usar. Por exemplo, "loan", "client", "disp".
  • billing_project: o ID do projeto de faturamento em que você ativou as APIs necessárias.
  • location: o local em que seu conjunto de contexto reside, por exemplo, "global".
  • your_context_set_id: se você criou um contexto avançado para o agente, o ID do conjunto de contexto.

Conectar-se aos dados do Cloud SQL para MySQL e do Cloud SQL para PostgreSQL

A API Análises de conversação usa um campo database_reference no campo sql para se conectar à sua instância do Cloud SQL.

O exemplo a seguir define uma conexão com um banco de dados do Cloud SQL para MySQL ou do Cloud SQL para 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"
          # }
    }
}

Substitua os valores de exemplo da seguinte forma:

  • PROJECT_ID: o ID do projeto Google Cloud que contém a instância do Cloud SQL.
  • REGION: a região em que a instância do Cloud SQL está localizada, por exemplo, us-central1.
  • CLUSTER_ID: o ID do cluster do Cloud SQL.
  • INSTANCE_ID: o ID da instância do Cloud SQL.
  • DATABASE_ID: o nome do banco de dados de destino, por exemplo, financial.
  • TABLE_1_ID, TABLE_2_ID: opcional. Uma lista das tabelas no banco de dados que o agente de dados é sugerido para usar. Por exemplo, "loan", "client", "disp".
  • billing_project: o ID do projeto de faturamento em que você ativou as APIs necessárias.
  • location: o local em que seu conjunto de contexto reside, por exemplo, "global".
  • your_context_set_id: se você criou um contexto avançado para o agente, o ID do conjunto de contexto.

Conectar aos dados do Spanner

A API Análises de conversação usa um campo database_reference no campo spanner para se conectar à sua instância do Spanner.

O exemplo a seguir define uma conexão com um banco de dados do 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"
        # }
    }
}

Substitua os valores de exemplo da seguinte forma:

  • PROJECT_ID: o ID do projeto Google Cloud que contém a instância do Spanner.
  • INSTANCE_ID: o ID da instância do Spanner.
  • DATABASE: o nome do banco de dados de destino, por exemplo, financial.
  • TABLE_1_ID, TABLE_2_ID: uma lista das tabelas no banco de dados que o agente de dados é sugerido para usar. Por exemplo, "loan", "client", "disp".
  • billing_project: o ID do projeto de faturamento em que você ativou as APIs necessárias.
  • location: o local em que seu conjunto de contexto reside, por exemplo, global.
  • your_context_set_id: se você criou um contexto avançado para o agente, o ID do conjunto de contexto.

Criar um agente de dados

O exemplo de código a seguir demonstra como criar o agente de dados de forma síncrona ou assíncrona enviando uma solicitação HTTP POST ao endpoint de criação do agente de dados. O payload da solicitação inclui os seguintes detalhes:

Você pode proteger o agente de dados com chaves de criptografia gerenciadas pelo cliente (CMEK) fornecendo um kms_key durante a criação. A CMEK é compatível apenas com fontes de dados do Looker. Para mais informações, consulte Chaves de criptografia gerenciadas pelo cliente (CMEK).

Você também pode ativar a análise avançada com Python incluindo o parâmetro options no payload da solicitação. Consulte Recurso REST: projects.locations.dataAgents para mais informações sobre o parâmetro options e as opções que podem ser configuradas para a conversa.

Síncrona

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

data_agent_id = "data_agent_1"

# 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

data_agent_payload = {
      "name": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}", # Optional
      "description": "This is the description of data_agent_1.", # Optional
      "data_analytics_agent": {
          "published_context": {
              "datasource_references": bigquery_data_sources,
              "system_instruction": system_instruction,
              # Optional: To enable advanced analysis with Python, include the following options block:
              "options": {
                  "analysis": {
                      "python": {
                          "enabled": True
                      }
                  }
              }
          }
      }
  }

# 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
  data_agent_payload["kms_key"] = f"projects/{kms_project_id}/locations/{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)

Assíncrono

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

data_agent_id = "data_agent_1"

# 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

data_agent_payload = {
      "name": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}", # Optional
      "description": "This is the description of data_agent_1.", # Optional
      "data_analytics_agent": {
          "published_context": {
              "datasource_references": bigquery_data_sources,
              "system_instruction": system_instruction,
              # Optional: To enable advanced analysis with Python, include the following options block:
              "options": {
                  "analysis": {
                      "python": {
                          "enabled": True
                      }
                  }
              }
          }
      }
  }

# 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
  data_agent_payload["kms_key"] = f"projects/{kms_project_id}/locations/{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)

Substitua os valores de exemplo da seguinte forma:

  • data_agent_1: um identificador exclusivo para o agente de dados. Esse valor é usado no nome do recurso do agente e como o parâmetro de consulta de URL data_agent_id.
  • KMS_PROJECT_ID: se você estiver usando a CMEK, o ID do projeto em que a chave está hospedada. Se não for fornecido, o padrão será seu projeto de faturamento.
  • KEY_RING_NAME: se você estiver usando a CMEK, o nome do keyring do Cloud KMS.
  • KEY_NAME: se você estiver usando a CMEK, o nome da chave do Cloud KMS.
  • This is the description of data_agent_1.: uma descrição para o agente de dados.

Criar uma conversa

O exemplo de código a seguir demonstra como criar uma conversa com seu agente de dados.

Você também pode proteger a conversa com a CMEK fornecendo um kms_key durante a criação. A CMEK é compatível apenas com fontes de dados do Looker. Para mais informações, consulte Chaves de criptografia gerenciadas pelo cliente (CMEK).

conversation_url = f"https://geminidataanalytics.googleapis.com/v1beta/projects/{billing_project}/locations/{location}/conversations"

data_agent_id = "data_agent_1"
conversation_id = "conversation_1"

# 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

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
  conversation_payload["kms_key"] = f"projects/{kms_project_id}/locations/{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)

Substitua os valores de exemplo da seguinte forma:

  • data_agent_1: o ID do agente de dados, conforme definido no bloco de exemplo de código em Criar agente de dados.
  • conversation_1: um identificador exclusivo da conversa.
  • KMS_PROJECT_ID: se você estiver usando a CMEK, o ID do projeto em que a chave está hospedada. Se não for fornecido, o padrão será seu projeto de faturamento.
  • KEY_RING_NAME: se você estiver usando a CMEK, o nome do keyring do Cloud KMS.
  • KEY_NAME: se você estiver usando a CMEK, o nome da chave do Cloud KMS.

Gerenciar agentes de dados e conversas

Os exemplos de código a seguir mostram como gerenciar agentes de dados e conversas usando a API Conversational Analytics. É possível fazer as seguintes tarefas:

Receber um agente de dados

O exemplo de código a seguir demonstra como buscar um agente de dados enviando uma solicitação HTTP GET para o URL do recurso do agente de dados.

data_agent_id = "data_agent_1"
data_agent_url = f"{base_url}/v1beta/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)

No exemplo anterior, substitua data_agent_1 pelo ID do agente de dados que você quer buscar.

Listar agentes de dados

O código a seguir demonstra como listar todos os agentes de dados de um determinado projeto enviando uma solicitação HTTP GET para o endpoint dataAgents.

Para listar todos os agentes, você precisa ter a permissão geminidataanalytics.dataAgents.list no projeto. Para mais informações sobre quais papéis do IAM incluem essa permissão, consulte a lista de papéis predefinidos.

billing_project = "YOUR-BILLING-PROJECT"
data_agent_url = f"{base_url}/v1beta/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}")

Substitua YOUR-BILLING-PROJECT pelo ID do seu projeto de faturamento.

Listar agentes de dados acessíveis

O código a seguir demonstra como listar todos os agentes de dados acessíveis de um determinado projeto enviando uma solicitação HTTP GET para o endpoint dataAgents:listAccessible.

billing_project = "YOUR-BILLING-PROJECT"
creator_filter = "YOUR-CREATOR-FILTER"
data_agent_url = f"{base_url}/v1beta/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}")

Substitua os valores de exemplo da seguinte forma:

  • YOUR-BILLING-PROJECT: o ID do seu projeto de faturamento.
  • YOUR-CREATOR-FILTER: o filtro a ser aplicado com base no criador do agente de dados. Os valores possíveis incluem NONE (padrão), CREATOR_ONLY e NOT_CREATOR_ONLY.

Atualizar um agente de dados

O exemplo de código a seguir demonstra como atualizar um agente de dados de forma síncrona ou assíncrona enviando uma solicitação HTTP PATCH para o URL do recurso do agente de dados. O payload da solicitação inclui os novos valores dos campos que você quer mudar, e os parâmetros da solicitação incluem um parâmetro updateMask, que especifica os campos a serem atualizados.

Síncrona

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

data_agent_url = f"https://geminidataanalytics.googleapis.com/v1beta/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)

Assíncrono

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

data_agent_url = f"https://geminidataanalytics.googleapis.com/v1beta/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)

Substitua os valores de exemplo da seguinte forma:

  • data_agent_1: o ID do agente de dados que você quer atualizar.
  • YOUR-BILLING-PROJECT: o ID do seu projeto de faturamento.
  • Updated description of the data agent.: uma nova descrição para o agente de dados.

Acessar a política de IAM de um agente de dados

O exemplo de código a seguir demonstra como buscar a política do IAM de um agente de dados enviando uma solicitação HTTP POST para o URL do agente de dados. O payload da solicitação inclui o caminho do agente de dados.

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

data_agent_url = f"{base_url}/v1beta/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)

Substitua os valores de exemplo da seguinte forma:

  • YOUR-BILLING-PROJECT: o ID do seu projeto de faturamento.
  • data_agent_1: o ID do agente de dados cuja política do IAM você quer buscar.

Definir a política do IAM para um agente de dados

Para compartilhar um agente, use o método setIamPolicy para atribuir papéis do IAM a usuários em um agente específico. O exemplo de código a seguir demonstra como fazer uma chamada POST para o URL do agente de dados com uma carga útil que inclui vinculações. A vinculação especifica quais papéis devem ser atribuídos a quais usuários.

billing_project = "YOUR-BILLING-PROJECT"
data_agent_id = "data_agent_1"
role = "roles/geminidataanalytics.dataAgentEditor"
users = "222larabrown@gmail.com, cloudysanfrancisco@gmail.com"

data_agent_url = f"{base_url}/v1beta/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)

Substitua os valores de exemplo da seguinte forma:

  • YOUR-BILLING-PROJECT: o ID do seu projeto de faturamento.
  • data_agent_1: o ID do agente de dados para o qual você quer definir a política do IAM.
  • 222larabrown@gmail.com, cloudysanfrancisco@gmail.com: uma lista separada por vírgulas de e-mails de usuários a quem você quer conceder o papel especificado.

Excluir um agente de dados

O exemplo de código a seguir demonstra como fazer a exclusão reversível de um agente de dados de maneira síncrona ou assíncrona enviando uma solicitação HTTP DELETE para o URL do recurso do agente de dados. A exclusão temporária significa que o agente é excluído, mas ainda pode ser recuperado em até 30 dias.

Síncrona

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

data_agent_url = f"https://geminidataanalytics.googleapis.com/v1beta/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)

Assíncrono

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

data_agent_url = f"https://geminidataanalytics.googleapis.com/v1beta/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)

Substitua os valores de exemplo da seguinte forma:

  • YOUR-BILLING-PROJECT: o ID do seu projeto de faturamento.
  • data_agent_1: o ID do agente de dados que você quer excluir.

Acessar uma conversa

O exemplo de código a seguir demonstra como buscar uma conversa enviando uma solicitação HTTP GET para o URL do recurso de conversa.

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

conversation_url = f"{base_url}/v1beta/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)

Substitua os valores de exemplo da seguinte forma:

  • YOUR-BILLING-PROJECT: o ID do seu projeto de faturamento.
  • conversation_1: o ID da conversa que você quer buscar.

Listar conversas

O exemplo de código a seguir demonstra como listar conversas de um determinado projeto enviando uma solicitação HTTP GET para o endpoint conversations.

Por padrão, esse método retorna as conversas que você criou. Os administradores (usuários com o papel do IAM cloudaicompanion.topicAdmin) podem acessar todas as conversas no projeto.

billing_project = "YOUR-BILLING-PROJECT"
conversation_url = f"{base_url}/v1beta/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)

Substitua YOUR-BILLING-PROJECT pelo ID do projeto de faturamento em que você ativou as APIs necessárias.

Listar mensagens em uma conversa

O exemplo de código a seguir demonstra como listar todas as mensagens em uma conversa enviando uma solicitação HTTP GET para o endpoint messages da conversa.

Para listar mensagens, é necessário ter a permissão cloudaicompanion.topics.get na conversa.

billing_project = "YOUR-BILLING-PROJECT"

conversation_id = "conversation_1"

conversation_url = f"{base_url}/v1beta/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)

Substitua os valores de exemplo da seguinte forma:

  • YOUR-BILLING-PROJECT: o ID do seu projeto de faturamento.
  • conversation_1: o ID da conversa cujas mensagens você quer listar.

Excluir uma conversa

O exemplo de código a seguir demonstra como excluir uma conversa enviando uma solicitação HTTP DELETE para o URL do recurso de conversa. Os administradores (usuários com o papel de administrador de tópicos do Cloud AI Companion no Identity and Access Management) ou usuários com a permissão cloudaicompanion.topics.delete do Identity and Access Management podem excluir conversas no projeto.

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

conversation_url = f"{base_url}/v1beta/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)

Substitua os valores de exemplo da seguinte forma:

  • YOUR-BILLING-PROJECT: o ID do seu projeto de faturamento.
  • conversation_1: o ID da conversa que você quer excluir.

Usar a API para fazer perguntas

Depois de criar um agente de dados e, para conversas com estado, uma conversa, você pode enviar consultas ao agente.

Quando você envia uma consulta, a API retorna um fluxo de objetos Message. Esse stream pode conter diferentes tipos de mensagens, incluindo texto, tabelas de dados e gráficos. As mensagens de texto podem fornecer insights sobre o raciocínio do agente, informar sobre o progresso ou dar a resposta final. A finalidade de cada mensagem de texto é indicada pelo valor TextType:

  • THOUGHT: mostra o processo de raciocínio interno do agente enquanto ele planeja como responder à sua consulta. As mensagens THOUGHT fornecem insights detalhados sobre o raciocínio e o processo de tomada de decisão do agente e contêm duas partes: parts[0] é o resumo do pensamento, que resume brevemente o texto completo, e parts[1] é o texto completo.
  • PROGRESS: informa o progresso do agente em uma ação, como recuperação de dados ou uma ferramenta que está sendo invocada. Esse valor é retornado apenas para fontes de dados do Looker e contém duas partes: parts[0] é o resumo, e parts[1] é o texto completo do progresso.
  • FINAL_RESPONSE: fornece a resposta final à sua consulta.

Os exemplos na seção a seguir usam as funções auxiliares definidas em Funções auxiliares do Python para transmitir respostas de chat para processar e mostrar cada mensagem na resposta transmitida da API. Para saber como renderizar essas mensagens em uma interface do usuário ao usar fontes de dados do Looker, consulte Renderizar respostas do agente para fontes de dados do Looker.

Fazer perguntas com conversas com e sem estado

Você pode interagir com a API nos seguintes modos principais:

  • Conversa com estado:o Google Cloud armazena e gerencia o histórico de conversas. A conversa com estado é inerentemente multiturno, já que a API retém o contexto das mensagens anteriores. Você só precisa enviar a mensagem atual a cada turno.
  • Chat sem estado: seu aplicativo gerencia o histórico de conversas. Você precisa incluir as mensagens anteriores relevantes em cada nova mensagem. Para exemplos detalhados de como gerenciar conversas multiturno no modo sem estado, consulte Criar uma conversa multiturno sem estado.

Os exemplos a seguir demonstram como usar a API para chat com e sem estado fazendo uma solicitação POST ao endpoint :chat.

Chat com estado

Enviar uma solicitação de chat com estado e uma referência de conversa

O exemplo de código a seguir demonstra como fazer perguntas à API usando a conversa definida nas etapas anteriores. Este exemplo usa uma função auxiliar get_stream para transmitir a resposta.

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

data_agent_id = "data_agent_1"
conversation_id = "conversation_1"

# Construct the payload
chat_payload = {
    "parent": f"projects/{billing_project}/locations/global",
    "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}",
            # "credentials": looker_credentials
        }
    }
}

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

Substitua os valores de exemplo da seguinte forma:

  • data_agent_1: o ID do agente de dados, conforme definido no bloco de exemplo de código em Criar agente de dados.
  • conversation_1: um identificador exclusivo da conversa.
  • Make a bar graph for the top 5 states by the total number of airports foi usado como o comando de exemplo.

Chat sem estado

Os exemplos a seguir mostram como enviar solicitações sem estado usando uma referência de agente de dados ou contexto inline.

Enviar uma solicitação de chat sem estado com uma referência de agente de dados

O exemplo de código a seguir demonstra como fazer uma pergunta sem estado à API usando o agente de dados que você definiu nas etapas anteriores. Este exemplo usa uma função auxiliar get_stream para transmitir a resposta.

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

data_agent_id = "data_agent_1"

# Construct the payload
chat_payload = {
    "parent": f"projects/{billing_project}/locations/global",
    "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}",
        # "credentials": looker_credentials
    }
}

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

Substitua os valores de exemplo da seguinte forma:

  • data_agent_1: o ID do agente de dados, conforme definido no bloco de código de exemplo em Criar um agente de dados.
  • Make a bar graph for the top 5 states by the total number of airports foi usado como o comando de exemplo.

Enviar uma solicitação de chat sem estado com contexto inline

O exemplo de código a seguir demonstra como fazer uma pergunta sem estado à API usando o contexto inline. Este exemplo usa uma função auxiliar get_stream para transmitir a resposta e usa uma fonte de dados do BigQuery como exemplo.

Você também pode ativar a análise avançada com Python incluindo o parâmetro options no payload da solicitação. Consulte a página Recurso REST: projects.locations.dataAgents para mais informações sobre o parâmetro options e as opções que podem ser configuradas para a conversa.

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

# Construct the payload
chat_payload = {
    "parent": f"projects/{billing_project}/locations/global",
    "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)

O exemplo de código a seguir demonstra como fazer uma pergunta sem estado à API usando o contexto inline e uma fonte de dados do AlloyDB como exemplo.

chat_url = f"https://geminidataanalytics.googleapis.com/v1beta/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/global",
    "messages": [
        {
            "userMessage": {
                "text": user_prompt
            }
        }
    ],
    "data_agent_context": {
          "data_agent": f"projects/{billing_project}/locations/global/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)

Criar uma conversa multiturno sem estado

Para fazer perguntas complementares em uma conversa sem estado, seu aplicativo precisa gerenciar o contexto da conversa enviando todo o histórico de mensagens com cada nova solicitação. As seções a seguir mostram como definir e chamar funções auxiliares para criar uma conversa multiturno:

Enviar solicitações multiturno

A função auxiliar multi_turn_Conversation a seguir gerencia o contexto da conversa armazenando mensagens em uma lista. Assim, você pode enviar perguntas de acompanhamento com base em interações anteriores. No payload da função, refira-se a um agente de dados ou forneça a fonte de dados diretamente usando o contexto inline.

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

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

data_agent_id = "data_agent_1"

# 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/global",
      "messages": conversation_messages,
      # Use a data agent reference
      "data_agent_context": {
          "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
          # "credentials": looker_credentials
      },
      # 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)

No exemplo anterior, substitua data_agent_1 pelo ID do agente de dados, conforme definido no bloco de exemplo de código em Criar agente de dados.

Você pode chamar a função auxiliar multi_turn_Conversation para cada turno da conversa. O exemplo de código a seguir mostra como enviar uma solicitação inicial e, em seguida, uma solicitação complementar que se baseia na resposta anterior.

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

No exemplo anterior, substitua os valores de amostra da seguinte forma:

  • Which species of tree is most prevalent?: uma pergunta em linguagem natural para enviar ao agente de dados.
  • Can you show me the results as a bar chart?: uma pergunta complementar que se baseia ou refina a pergunta anterior.

Processar respostas

A função get_stream_multi_turn a seguir processa a resposta da API Streaming. Essa função é semelhante à função auxiliar get_stream, mas armazena a resposta na lista conversation_messages para salvar o contexto da conversa para a próxima vez.

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 = ''

Exemplos de código completos

Os exemplos de código expansíveis a seguir contêm todas as tarefas abordadas neste guia.

Criar um agente de dados para o BigQuery e o Looker usando HTTP e Python

    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 formatters, highlight, lexers
    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
    }

    ################### Data source details ###################

    billing_project = "your_billing_project"
    location = "global"
    system_instruction = "Help the user in analyzing their data"


    # BigQuery data source
    bigquery_data_sources = {
        "bq": {
        "tableReferences": [
            {
            "projectId": "bigquery-public-data",
            "datasetId": "san_francisco",
            "tableId": "street_trees"
            }
        ]
        }
    }

    # Looker data source
    looker_credentials = {
        "oauth": {
            "secret": {
            "client_id": "your_looker_client_id",
            "client_secret": "your_looker_client_secret",
            }
        }
    }

    # To use access_token for authentication, uncomment the following looker_credentials code block and comment out the previous looker_credentials code block.
    # looker_credentials = {
    #     "oauth": {
    #         "token": {
    #           "access_token": "your_looker_access_token",
    #         }
    #     }
    # }

    looker_data_source = {
        "looker": {
        "explore_references": [
            {
                "looker_instance_uri": looker_instance_uri,
                "lookml_model": "your_model",
                "explore": "your_explore",
            },
            {
                "looker_instance_uri": looker_instance_uri,
                "lookml_model": "your_model_2",
                "explore": "your_explore_2",
            },
            # Add up to 5 total Explore references
        ],
        # Do not include the following line during agent creation
        # "credentials": looker_credentials
    }

    # Looker Studio data source
    looker_studio_data_source = {
        "studio":{
            "studio_references": [
                {
                  "studio_datasource_id": "studio_datasource_id"
                }
            ]
        }
    }

    ################### Create data agent ###################
    data_agent_url = f"https://geminidataanalytics.googleapis.com/v1beta/projects/{billing_project}/locations/{location}/dataAgents"

    data_agent_id = "data_agent_1"

    data_agent_payload = {
        "name": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}", # Optional
        "description": "This is the description of data_agent.", # Optional

        "data_analytics_agent": {
            "published_context": {
                "datasource_references": bigquery_data_sources,
                "system_instruction": system_instruction,
                # Optional: To enable advanced analysis with Python, include the following options block:
                "options": {
                    "analysis": {
                        "python": {
                            "enabled": True
                        }
                    }
                }
            }
        }
    }

    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)


    ################### Create conversation ###################

    conversation_url = f"https://geminidataanalytics.googleapis.com/v1beta/projects/{billing_project}/locations/{location}/conversations"

    data_agent_id = "data_agent_1"
    conversation_id = "conversation _1"

    conversation_payload = {
        "agents": [
            f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"
        ],
        "name": f"projects/{billing_project}/locations/{location}/conversations/{conversation_id}"
    }
    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)


    ################### Chat with the API by using conversation (stateful) ####################

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

    data_agent_id = "data_agent_1"
    conversation_id = "conversation _1"

    # Construct the payload
    chat_payload = {
        "parent": f"projects/{billing_project}/locations/global",
        "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}",
                # "credentials": looker_credentials
            }
        }
    }

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

    ################### Chat with the API by using dataAgents (stateless) ####################

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

    data_agent_id = "data_agent_1"

    # Construct the payload
    chat_payload = {
        "parent": f"projects/{billing_project}/locations/global",
        "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}",
            # "credentials": looker_credentials
        }
    }

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

    ################### Chat with the API by using inline context (stateless) ####################

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

    # Construct the payload
    chat_payload = {
        "parent": f"projects/{billing_project}/locations/global",
        "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 - if wanting to use advanced analysis with python
            "options": {
                "analysis": {
                    "python": {
                        "enabled": True
                    }
                }
            }
        }
    }

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

    ################### Multi-turn conversation ###################

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

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

    data_agent_id = "data_agent_1"

    # 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/global",
          "messages": conversation_messages,
          # Use a data agent reference
          "data_agent_context": {
              "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
              # "credentials": looker_credentials
          },
          # 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)

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

Definir funções auxiliares

O exemplo de código a seguir contém as definições das funções auxiliares usadas nos exemplos anteriores. Essas funções ajudam a analisar respostas da API e mostrar os resultados.

Funções auxiliares do Python para transmitir respostas de chat

    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 = ''