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:

• Configurar as definições iniciais e a autenticação.
• Conectar-se a uma fonte de dados do Looker, do BigQuery, do Data Studio, do AlloyDB para PostgreSQL, do Cloud SQL ou do Spanner.
• Criar um agente de dados.
• Criar uma conversa.
• Gerenciar agentes de dados e conversas.
• Usar a API para fazer perguntas.
• Criar uma conversa multiturno sem estado.
• Defina funções auxiliares.

Dica: você pode executar os exemplos de código nesta página no bloco interativo do Colaboratory HTTP da API Conversational Analytics.

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 Google Cloud CLI.
• Define variáveis para o projeto de faturamento, o local e as instruções do sistema.
• Define o endpoint de API com base na variável location.

import json
import json as json_lib
import textwrap

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

from google.colab import auth
auth.authenticate_user()

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

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

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

Substitua os valores de exemplo da seguinte forma:

• BILLING_PROJECT: o ID do projeto do Google Cloud em que você ativou as APIs necessárias.

• LOCATION: o local em que os recursos são armazenados. Para usar o endpoint global, especifique global. Para atender aos requisitos de residência de dados, especifique uma região única (como us-east4) ou multirregião (como eu ou us) compatível. Para mais informações, consulte Residência de dados.

  Importante: o local no caminho da solicitação precisa corresponder ao local do endpoint que você está chamando. Por exemplo, se você chamar o endpoint geminidataanalytics.us.rep.googleapis.com, especifique us no caminho da solicitação (como /locations/us).

• 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": "CLIENT_ID",
            "client_secret": "CLIENT_SECRET",
        }
    }
}

Substitua os valores de exemplo da seguinte forma:

• CLIENT_ID: o ID do cliente da chave de API do Looker gerada.
• 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": "ACCESS_TOKEN",
        }
    }
}

Substitua os valores de exemplo da seguinte forma:

• ACCESS_TOKEN: o valor access_token gerado para autenticar no Looker.

Dica: armazene o token de acesso do Looker (access_token) como uma variável de ambiente para melhorar a segurança.

Autenticar no AlloyDB para PostgreSQL

Se você planeja se conectar a uma fonte de dados do AlloyDB, use a CLI gcloud para se autenticar 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. Conceder papéis do IAM. O principal (usuário ou conta de serviço) que chama a API Conversational Analytics precisa ter as permissões apropriadas do IAM no projeto doGoogle Cloud . Isso geralmente inclui papéis 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 de 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, precisa se autenticar 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 instância do Cloud SQL para MySQL. É possível adicionar isso ao criar uma instância ou ao aplicar um patch em uma instância atual.

   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 do IAM cloudsql.instances.login e cloudsql.instances.connect no projeto. Para conceder esses papéis e permissões, siga um destes procedimentos:

   1. Conceda o papel roles/cloudsql.instanceUser. Isso inclui permissões para fazer login na instância (cloudsql.instances.login) e conectar-se usando o proxy (cloudsql.instances.connect).
   2. Como alternativa, conceda um papel personalizado 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. Conceder 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 a esse novo usuário do banco de dados permissões 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
   "[e-mail]";
   

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

   Os clientes geralmente se conectam pelo proxy de autenticação do Cloud SQL ou por uma biblioteca de conectores, 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, é usado um token OAuth 2.0, geralmente processado automaticamente pelo proxy de autenticação do Cloud SQL ou pela Google Cloud CLI.

   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, será necessário fazer a autenticação na instância do Spanner.

1. Conceder papéis do IAM. O principal (usuário ou conta de serviço) que chama a API Conversational Analytics precisa das permissões apropriadas do IAM no seu projeto doGoogle Cloud . Isso geralmente inclui papéis 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 as 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 refinado.

Conectar a uma fonte de dados

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

• Looker
• Data Studio
• BigQuery
• Dados do AlloyDB
• Dados do Cloud SQL para MySQL
• Dados do Cloud SQL para PostgreSQL
• Dados do Spanner

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 Looker, conforme descrito em Autenticar e conectar a uma fonte de dados. É possível se conectar a até cinco Análises do Looker por vez com a API Conversational Analytics.

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 realizar consultas em várias Análises ao mesmo tempo.
• Um agente pode consultar várias Análises na mesma conversa.

• Um agente pode consultar várias Análises em uma conversa que inclui perguntas com várias partes ou em conversas que incluem perguntas complementares.

  Por exemplo: um usuário conecta duas Análises, uma chamada cat-explore e outra chamada dog-explore. O usuário insere a pergunta "O que é maior: o número de gatos ou o número 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 o número das duas consultas depois de concluí-las.

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

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

Substitua os valores de exemplo da seguinte forma:

• INSTANCE_URI: o URL completo da sua instância do Looker (por exemplo, https://mycompany.looker.com).
• MODEL: o nome do modelo do LookML que inclui a Análise de destino.
• EXPLORE: o nome da Análise do Looker que o agente de dados deve consultar.
• MODEL_2: o nome do segundo modelo do LookML que inclui a Análise de destino. É possível repetir essa variável para outros modelos em até cinco análises detalhadas.
• EXPLORE_2: o nome da Análise do Looker adicional que o agente de dados deve consultar. Você pode repetir essa variável para incluir até cinco análises.

Conectar-se aos dados do BigQuery

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

Observação: consultas que exigem junções complexas em várias tabelas podem resultar em respostas menos precisas.

As seções a seguir mostram como se conectar aos dados do BigQuery usando referências de tabela, de gráfico ou ambas.

Conectar-se a tabelas do BigQuery

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 o desempenho do agente, você pode fornecer contexto estruturado para as tabelas do BigQuery, como descrições de tabelas e colunas, sinônimos, tags, consultas de exemplo e funções definidas pelo usuário. Para mais informações, consulte Definir o contexto do agente de dados para fontes de dados do BigQuery.

Importante: verifique se você tem as permissões necessárias do Identity and Access Management (IAM) para consultar as tabelas do BigQuery a que está se conectando.

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

Substitua os valores de exemplo da seguinte forma:

• 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.
• DATASET_ID: ID do conjunto de dados do BigQuery.
• TABLE_ID: ID da tabela do BigQuery.
• TABLE_DESCRIPTION: uma descrição opcional do conteúdo e da finalidade da tabela.
• COLUMN_NAME: o nome de uma coluna na tabela para a qual você está fornecendo uma descrição opcional.
• COLUMN_DESCRIPTION: uma descrição opcional do conteúdo e da finalidade da coluna.

Conectar-se ao BigQuery Graph

Também é possível se conectar a gráficos usando o BigQuery Graph. Forneça uma ou mais referências de gráfico no campo propertyGraphReferences.

Para informações sobre os requisitos da edição do BigQuery e as limitações de consulta de gráfico, consulte Conectar ao BigQuery Graph.

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

Substitua os valores de exemplo da seguinte forma:

• GRAPH_PROJECT_ID: o ID do projeto do Google Cloud que contém o gráfico do BigQuery.
• GRAPH_DATASET_ID: o ID do conjunto de dados do BigQuery que contém o gráfico.
• GRAPH_ID: o ID do gráfico do BigQuery.

Para consultar tabelas e gráficos com o mesmo agente de dados, inclua tableReferences e propertyGraphReferences no objeto bq.

Conectar aos dados do Data Studio

O exemplo de código a seguir define uma conexão com uma fonte de dados do Data 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 Conversational Analytics 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 do Google Cloud que contém o cluster do AlloyDB.
• REGION: a região em que o cluster do AlloyDB está localizado, 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 deve 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 o 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 Conversational Analytics usa um campo database_reference no campo sql para se conectar à 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 do Google Cloud que contém a instância do Cloud SQL.
• REGION: a região em que a instância do Cloud SQL reside, 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 deve 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 o 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 Spanner

A API Conversational Analytics usa um campo database_reference no campo spanner para se conectar à 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 do 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 deve 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 o 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:

• O nome completo do recurso do agente. Esse valor inclui o ID do projeto, o local e um identificador exclusivo do agente.
• A descrição do agente de dados.
• O contexto do agente de dados, incluindo a descrição do sistema (definida em Configurar as definições iniciais e a autenticação) e a fonte de dados usada pelo agente (definida em Conectar a uma fonte de dados).

Se quiser, você pode proteger o agente de dados com chaves de criptografia gerenciadas pelo cliente (CMEK) incluindo o campo kms_key no payload da solicitação ao criar o agente de dados. 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 o 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.

Aviso: o código Python retornado pelos agentes de dados foi projetado para ser executado em um sandbox seguro gerenciado pelo Google. Executar esse código em um ambiente local ou não verificado ignora as proteções de segurança do sandbox e pode expor o sistema a riscos de segurança, como a execução de código mal-intencionado. Para saber como inspecionar e processar com segurança o código Python gerado pelo agente, consulte Como posso inspecionar e processar com segurança o código Python gerado pelo agente?

O exemplo a seguir usa a variável bigquery_data_sources criada na seção sobre conexão com o BigQuery para configurar o agente de dados e se conectar a uma tabela do BigQuery.

Para usar dados do Looker, substitua bigquery_data_sources por looker_data_source.

Para usar os dados do Data Studio, substitua bigquery_data_sources por looker_studio_data_source.

Para usar dados do AlloyDB, substitua bigquery_data_sources por alloydb_data_source.

Para usar dados do Cloud SQL, substitua bigquery_data_sources por sql_data_source.

Para usar dados do Spanner, substitua bigquery_data_sources por spanner_data_sources.

Síncrona

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

data_agent_id = "DATA_AGENT_ID"

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

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

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

params = {"data_agent_id": data_agent_id} # Optional

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

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

Assíncrono

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

data_agent_id = "DATA_AGENT_ID"

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

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

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

params = {"data_agent_id": data_agent_id} # Optional

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

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

Substitua os valores de exemplo da seguinte forma:

• DATA_AGENT_ID: 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 CMEK, o ID do projeto em que a chave está hospedada. Se não for fornecido, o padrão será o projeto de faturamento.
• KEY_RING_NAME: se você estiver usando CMEK, o nome do keyring do Cloud KMS.
• KEY_NAME: se você estiver usando CMEK, o nome da chave do Cloud KMS.
• KEY_LOCATION: se você estiver usando CMEK, o local do keyring. Para a multirregião eu, use europe. Para todos os outros locais, use o mesmo local do agente de dados. Se não for fornecido, o padrão será o local do agente de dados, que funciona para todos os locais, exceto a multirregião eu. Para mais informações, consulte Requisitos para locais da chave.
• This is the description of the data agent.: 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.

Se quiser, proteja a conversa com CMEK incluindo o campo kms_key no payload da solicitação ao criar a conversa. Para mais informações, consulte Chaves de criptografia gerenciadas pelo cliente (CMEK).

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

data_agent_id = "DATA_AGENT_ID"
conversation_id = "CONVERSATION_ID"

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

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

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

params = {
    "conversation_id": conversation_id
}

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

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

Substitua os valores de exemplo da seguinte forma:

• DATA_AGENT_ID: o ID do agente de dados, conforme definido no bloco de exemplo de código em Criar agente de dados.
• CONVERSATION_ID: um identificador exclusivo da conversa.
• KMS_PROJECT_ID: se você estiver usando CMEK, o ID do projeto em que a chave está hospedada. Se não for fornecido, o padrão será o projeto de faturamento.
• KEY_RING_NAME: se você estiver usando CMEK, o nome do keyring do Cloud KMS.
• KEY_NAME: se você estiver usando CMEK, o nome da chave do Cloud KMS.
• KEY_LOCATION: se você estiver usando CMEK, o local do keyring. Para a multirregião us, use us-central1. Para a multirregião eu, use europe-west1. Para uma única região, use a mesma região da conversa. Se não for fornecido, o padrão será o local da conversa, que funciona apenas para regiões únicas. Para mais informações, consulte Requisitos para locais da chave.

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:

• Recuperar um agente de dados
• Listar agentes de dados
• Listar agentes de dados acessíveis
• Atualizar um agente de dados
• Acessar a política do IAM de um agente de dados
• Definir a política do IAM de um agente de dados
• Excluir um agente de dados
• Acessar uma conversa
• Listar conversas
• Listar mensagens em uma conversa

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_ID"
data_agent_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"

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

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

No exemplo anterior, substitua DATA_AGENT_ID 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 = "BILLING_PROJECT"
data_agent_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/dataAgents"

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

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

Substitua 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 = "BILLING_PROJECT"
creator_filter = "CREATOR_FILTER"
data_agent_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/dataAgents:listAccessible"

params = {
    "creator_filter": creator_filter
}

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

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

Substitua os valores de exemplo da seguinte forma:

• BILLING_PROJECT: o ID do seu projeto de faturamento.
• 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_ID"
billing_project = "BILLING_PROJECT"

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

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

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

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

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

Assíncrono

data_agent_id = "DATA_AGENT_ID"
billing_project = "BILLING_PROJECT"

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

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

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

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

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

Substitua os valores de exemplo da seguinte forma:

• DATA_AGENT_ID: o ID do agente de dados que você quer atualizar.
• 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 = "BILLING_PROJECT"
data_agent_id = "DATA_AGENT_ID"

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

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

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

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

Substitua os valores de exemplo da seguinte forma:

• BILLING_PROJECT: o ID do seu projeto de faturamento.
• DATA_AGENT_ID: 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.

Importante: a chamada de API setIamPolicy substitui as permissões atuais do recurso. Para preservar as políticas atuais, chame Acessar política do IAM na API para buscar a política atual e transmita-a com outras mudanças na chamada para SetIamPolicy.

billing_project = "BILLING_PROJECT"
data_agent_id = "DATA_AGENT_ID"
role = "roles/geminidataanalytics.dataAgentEditor"
users = "[e-mail], [e-mail]"

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

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

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

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

Substitua os valores de exemplo da seguinte forma:

• BILLING_PROJECT: o ID do seu projeto de faturamento.
• DATA_AGENT_ID: o ID do agente de dados para o qual você quer definir a política do IAM.
• [e-mail], [e-mail]: 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 excluir um agente de dados de forma síncrona ou assíncrona enviando uma solicitação HTTP DELETE para o URL do recurso do agente de dados. Quando você exclui um agente de dados, ele entra em um período de exclusão reversível de 30 dias, em que não pode ser recuperado. Para mais informações, consulte Retenção e exclusão de dados.

Síncrona

billing_project = "BILLING_PROJECT"
data_agent_id = "DATA_AGENT_ID"

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

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

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

Assíncrono

billing_project = "BILLING_PROJECT"
data_agent_id = "DATA_AGENT_ID"

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

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

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

Substitua os valores de exemplo da seguinte forma:

• BILLING_PROJECT: o ID do seu projeto de faturamento.
• DATA_AGENT_ID: 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 = "BILLING_PROJECT"
conversation_id = "CONVERSATION_ID"

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

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

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

Substitua os valores de exemplo da seguinte forma:

• BILLING_PROJECT: o ID do seu projeto de faturamento.
• CONVERSATION_ID: 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 = "BILLING_PROJECT"
conversation_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/conversations"

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

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

Substitua 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.

Observação: essa operação de lista também é compatível com paginação.

billing_project = "BILLING_PROJECT"

conversation_id = "CONVERSATION_ID"

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

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

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

Substitua os valores de exemplo da seguinte forma:

• BILLING_PROJECT: o ID do seu projeto de faturamento.
• CONVERSATION_ID: 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 cloudaicompanion.topicAdmin do 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 = "BILLING_PROJECT"
conversation_id = "CONVERSATION_ID"

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

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

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

Substitua os valores de exemplo da seguinte forma:

• BILLING_PROJECT: o ID do seu projeto de faturamento.
• CONVERSATION_ID: o ID da conversa que você quer excluir.

Usar a API para fazer perguntas

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

Quando você envia uma consulta, a API retorna um stream 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 dele 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 processo de raciocínio e tomada de decisão do agente e contêm duas partes: parts[0] é o resumo do pensamento, que resume brevemente o texto completo do pensamento, e parts[1] é o texto completo do pensamento.
• PROGRESS: informa o progresso do agente em uma ação, como a 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 exibir 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 chat com e sem estado

É possível 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 conversas com e sem estado fazendo uma solicitação POST ao endpoint :chat.

Atenção: os exemplos de código nesta seção usam a biblioteca padrão requests do Python para demonstrar o streaming HTTP. Os clientes HTTP ou REST padrão não podem ler trailers HTTP/2. Se ocorrer um erro fatal (como um tempo limite) durante o streaming, a API poderá retornar uma mensagem de erro genérica, como 503 Connection reset by peer ou 500 Internal Server Error. Para aplicativos de produção, recomendamos o uso do SDK do Python, que usa o gRPC para analisar corretamente esses erros finais.

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.

Observação: se o agente usar uma fonte de dados do Looker, remova a marca de comentário da linha "credentials": looker_credentials.

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

data_agent_id = "DATA_AGENT_ID"
conversation_id = "CONVERSATION_ID"

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

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

Substitua os valores de exemplo da seguinte forma:

• DATA_AGENT_ID: o ID do agente de dados, conforme definido no bloco de exemplo de código em Criar agente de dados.
• CONVERSATION_ID: 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 um 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.

Observação: se o agente usar uma fonte de dados do Looker, remova a marca de comentário da linha "credentials": looker_credentials.

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

data_agent_id = "DATA_AGENT_ID"

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

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

Substitua os valores de exemplo da seguinte forma:

• DATA_AGENT_ID: o ID do agente de dados, conforme definido no bloco de exemplo de código em Criar 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.

Observação: se o agente usar uma fonte de dados do Looker, remova a marca de comentário da linha "credentials": looker_credentials nos exemplos de código a seguir.

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.

Aviso: o código Python retornado pelos agentes de dados foi projetado para ser executado em um sandbox seguro gerenciado pelo Google. Executar esse código em um ambiente local ou não verificado ignora as proteções de segurança do sandbox e pode expor o sistema a riscos de segurança, como a execução de código mal-intencionado. Para saber como inspecionar e processar com segurança o código Python gerado pelo agente, consulte Como posso inspecionar e processar com segurança o código Python gerado pelo agente?

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

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

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

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/v1/projects/{billing_project}/locations/{location}:chat"

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

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

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

get_stream(chat_url, chat_payload)

Criar uma conversa multiturno sem estado

Para o usuário fazer perguntas complementares em uma conversa sem estado, o aplicativo precisa enviar todo o histórico de mensagens em cada nova solicitação. As conversas multiturno permitem que você envie perguntas de acompanhamento com base em interações anteriores.

O exemplo a seguir define e chama uma função multi_turn_conversation para gerenciar o contexto da conversa armazenando mensagens em uma lista. A função multi_turn_conversation chama a função get_stream_multi_turn para analisar a resposta da API de streaming e atualizar a lista conversation_messages. Esse processo preserva o contexto para as interações seguintes. Para ver o código de get_stream_multi_turn e outras funções auxiliares necessárias, consulte Funções auxiliares de chat por streaming.

O exemplo de código a seguir usa uma referência de agente de dados. Para usar o contexto inline, remova a marca de comentário do bloco inline_context e adicione um comentário ao bloco data_agent_context. A amostra de contexto inline usa uma fonte de dados do BigQuery.

Se o agente usar uma fonte de dados do Looker, remova a marca de comentário da linha "credentials": looker_credentials.

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

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

data_agent_id = "DATA_AGENT_ID"

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

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

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

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

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

No exemplo anterior, substitua DATA_AGENT_ID 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, baseada na pergunta anterior.

Definir funções auxiliares

A seção a seguir contém as definições das funções auxiliares usadas nos exemplos de código anteriores para analisar as respostas da API e mostrar os resultados.

Funções auxiliares de chat de streaming

As funções a seguir analisam e mostram a resposta de streaming da API para solicitações de chat padrão (turno único).

Mostrar funções auxiliares 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 = ''