Esta página foi traduzida pela API Cloud Translation.

Crie um agente de dados através de HTTP e Python

Esta página explica como usar o Python para fazer pedidos HTTP à API Conversational Analytics (acedida através de geminidataanalytics.googleapis.com).

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

Configure as definições iniciais e a autenticação
Estabeleça ligação a uma origem de dados do Looker, do BigQuery ou do Looker Studio
Crie um agente de dados
Crie uma conversa
Faça a gestão dos agentes de dados e das conversas
Use a API para fazer perguntas
Crie uma conversa sem estado com várias interações

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

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

O seguinte exemplo de código Python realiza estas tarefas:

Importa as bibliotecas Python necessárias
Obtém um token de acesso para autenticação HTTP através da CLI do Google Cloud
Define variáveis para o projeto de faturação e as instruções do sistema

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

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'
system_instruction = 'YOUR-SYSTEM-INSTRUCTIONS'

Substitua os valores de exemplo da seguinte forma:

YOUR-BILLING-PROJECT: o ID do projeto de faturação onde ativou as APIs necessárias.
YOUR-SYSTEM-INSTRUCTIONS: instruções do sistema para orientar o comportamento do agente e personalizá-lo de acordo com as suas necessidades de dados. Por exemplo, pode usar instruções do sistema para definir termos empresariais, controlar o comprimento das respostas ou definir a formatação de dados. Idealmente, defina instruções do sistema usando o formato YAML recomendado em Escreva instruções do sistema eficazes para fornecer orientações detalhadas e estruturadas.

Autentique no Looker

Se planeia estabelecer ligação a uma origem de dados do Looker, tem de se autenticar na instância do Looker.

Usar chaves da API

O exemplo de código Python seguinte demonstra como autenticar o seu agente numa instância do Looker através de chaves da 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 de cliente da chave da API Looker gerada.
YOUR-LOOKER-CLIENT-SECRET: o segredo do cliente da chave da API Looker gerada.

Usar tokens de acesso

O seguinte exemplo de código Python demonstra como autenticar o seu agente numa instância do Looker através de 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 que gera para autenticar no Looker.

Faça a associação a uma origem de dados

As secções seguintes mostram como definir os detalhes da ligação para as origens de dados do seu agente. O seu agente pode estabelecer ligação a dados no Looker, no BigQuery ou no Looker Studio.

Associe-se a dados do Looker

O seguinte exemplo de código define uma ligação a um conteúdo de Explorar do Looker. Para estabelecer uma ligação com uma instância do Looker, verifique se gerou chaves da API Looker, conforme descrito no artigo Autentique e ligue-se a uma origem de dados com a API Conversational Analytics. Pode estabelecer ligação a um máximo de cinco opções Explorar do Looker em simultâneo com a API Conversational Analytics.

Quando se liga a uma origem de dados do Looker, tenha em atenção o seguinte:

Pode consultar qualquer Explorar incluído numa conversa.
Um agente só pode consultar uma análise detalhada de cada vez. Não é possível executar consultas em vários Explorers em simultâneo.
Um agente pode consultar várias explorações na mesma conversa.
Um agente pode consultar várias análises detalhadas numa conversa que inclua perguntas com várias partes ou em conversas que incluam perguntas de seguimento.

Por exemplo: um utilizador associa duas explorações, uma denominada cat-explore e outra denominada dog-explore. O utilizador introduz a pergunta "O que é maior: o número de gatos ou o número de cães?" Isto criaria duas consultas: uma para contar o número de gatos em cat-explore e outra para contar o número de cães em dog-explore. O agente compara o número de ambas as consultas depois de as concluir.

looker_instance_uri = "https://my_company.looker.com"
looker_data_source = {
    "looker": {
        "explore_references": {
            "looker_instance_uri": "https://your_company.looker.com"
            "lookml_model": "your_model",
            "explore": "your_explore",
       },
       {
            "looker_instance_uri": looker_instance_uri,
            "lookml_model": "your_model_2",
            "explore": "your_explore_2",
       },
       # 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 detalhada à qual quer estabelecer ligação.
your_explore: o nome da análise detalhada do Looker que quer que o agente de dados consulte.
my_model_2: o nome do segundo modelo do LookML que inclui a análise detalhada à qual quer estabelecer ligação. Pode repetir esta variável para modelos adicionais até um máximo de cinco explorações.
my_explore_2: o nome da análise detalhada do Looker adicional que quer que o agente de dados consulte. Pode repetir esta variável para incluir até cinco explorações.

Estabeleça ligação aos dados do BigQuery

Com a API Conversational Analytics, não existem limites rígidos quanto ao número de tabelas do BigQuery às quais se pode ligar. No entanto, a ligação a um grande número de tabelas pode reduzir a precisão ou fazer com que exceda o limite de tokens de entrada do modelo.

O exemplo de código seguinte define uma ligação a várias tabelas do BigQuery e inclui exemplos de campos de contexto estruturados opcionais. Para melhorar o desempenho do agente, pode fornecer opcionalmente contexto estruturado para as suas tabelas do BigQuery, como descrições de tabelas e colunas, sinónimos, etiquetas e consultas de exemplo. Para mais informações, consulte o artigo Defina o contexto do agente de dados para origens 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 Google Cloud projeto que contém o conjunto de dados e a tabela do BigQuery aos quais quer estabelecer ligação. Para estabelecer ligação a um conjunto de dados público, especifique bigquery-public-data.
my_dataset_id: o ID do conjunto de dados do BigQuery.
my_table_id: o 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 está a fornecer uma descrição opcional.
my_column_description: uma descrição opcional do conteúdo e da finalidade da coluna.

Estabeleça ligação aos dados do Looker Studio

O seguinte código de exemplo define uma ligação a uma origem 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 origem de dados.

Crie um agente de dados

O código de exemplo seguinte demonstra como criar o agente de dados enviando um pedido HTTP POST para o ponto final de criação do agente de dados. O payload do pedido inclui os seguintes detalhes:

O nome do recurso completo do agente. Este valor inclui o ID do projeto, a localização e um identificador exclusivo para o agente.
A descrição do agente de dados.
O contexto do agente de dados, incluindo a descrição do sistema (definida em Configure as definições iniciais e a autenticação) e a origem de dados que o agente usa (definida em Estabeleça ligação a uma origem de dados).

Também pode ativar opcionalmente a análise avançada com Python, incluindo o parâmetro options no payload do pedido. Consulte o recurso REST: projects.locations.dataAgents para obter mais informações sobre o parâmetro options e as opções que pode configurar para a conversa.

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_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
                      }
                  }
              }
          }
      }
  }

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 do agente de dados. Este valor é usado no nome do recurso do agente e como o parâmetro de consulta do URL data_agent_id.
This is the description of data_agent_1.: uma descrição do agente de dados.

Crie uma conversa

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

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)

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 Crie um agente de dados.
conversation_1: um identificador exclusivo da conversa.

Faça a gestão dos agentes de dados e das conversas

Os seguintes exemplos de código mostram como gerir os seus agentes de dados e conversas através da API Conversational Analytics. Pode realizar as seguintes tarefas:

Obtenha um agente de dados
Crie listas de agentes de dados
Apresente agentes de dados acessíveis
Atualize um agente de dados
Obtenha a Política IAM para um agente de dados
Defina a Política IAM para um agente de dados
Elimine um agente de dados
Obtenha uma conversa
Liste conversas
Liste mensagens numa conversa

Obtenha um agente de dados

O seguinte código de exemplo demonstra como obter um agente de dados existente através do envio de um pedido 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 quer obter.

Apresente agentes de dados

O código seguinte demonstra como listar todos os agentes de dados de um determinado projeto enviando um pedido HTTP GET para o ponto final dataAgents.

Para listar todos os agentes, tem de ter a autorização geminidataanalytics.dataAgents.list no projeto. Para mais informações sobre as funções de IAM que incluem esta autorização, consulte a lista de funções predefinidas.

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
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 faturação.

Apresente os agentes de dados acessíveis

O código seguinte demonstra como listar todos os agentes de dados acessíveis para um determinado projeto através do envio de um pedido HTTP GET para o ponto final dataAgents:listAccessible.

billing_project = "YOUR-BILLING-PROJECT"
creator_filter = "YOUR-CREATOR-FILTER"
location = "global"
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 faturação.
YOUR-CREATOR-FILTER: o filtro a aplicar com base no criador do agente de dados. Os valores possíveis incluem NONE (predefinição), CREATOR_ONLY e NOT_CREATOR_ONLY.

Atualize um agente de dados

O seguinte exemplo de código demonstra como atualizar um agente de dados através do envio de um pedido HTTP PATCH para o URL do recurso do agente de dados. A carga útil do pedido inclui os novos valores dos campos que quer alterar, e os parâmetros do pedido incluem um parâmetro updateMask, que especifica os campos a atualizar.

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

data_agent_url = f"{base_url}/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 quer atualizar.
YOUR-BILLING-PROJECT: o ID do seu projeto de faturação.
Updated description of the data agent.: uma nova descrição para o agente de dados.

Obtenha a Política IAM para um agente de dados

O seguinte código de exemplo demonstra como obter a política de IAM para um agente de dados através do envio de um pedido HTTP POST para o URL do agente de dados. O payload do pedido inclui o caminho do agente de dados.

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
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 faturação.
data_agent_1: o ID do agente de dados para o qual quer obter a Política IAM.

Defina a Política IAM para um agente de dados

Para partilhar um agente, pode usar o método setIamPolicy para atribuir funções da IAM a utilizadores num agente específico. O seguinte código de exemplo demonstra como fazer uma chamada POST para o URL do agente de dados com uma carga útil que inclua associações. A associação especifica as funções que devem ser atribuídas a que utilizadores.

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
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 faturação.
data_agent_1: o ID do agente de dados para o qual quer definir a política IAM.
222larabrown@gmail.com, cloudysanfrancisco@gmail.com: uma lista de emails de utilizadores separados por vírgulas aos quais quer conceder a função especificada.

Elimine um agente de dados

O seguinte exemplo de código demonstra como eliminar temporariamente um agente de dados enviando um pedido HTTP DELETE para o URL do recurso do agente de dados. A eliminação temporária significa que o agente é eliminado, mas ainda pode ser obtido no prazo de 30 dias.

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
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.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 faturação.
data_agent_1: o ID do agente de dados que quer eliminar.

Receba uma conversa

O exemplo de código seguinte demonstra como obter uma conversa existente enviando um pedido HTTP GET para o URL do recurso de conversa.

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
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 faturação.
conversation_1: o ID da conversa que quer obter.

Listar conversas

O seguinte código de exemplo demonstra como listar conversas para um determinado projeto enviando um pedido HTTP GET para o ponto final conversations.

Por predefinição, este método devolve as conversas que criou. Os administradores (utilizadores com a cloudaicompanion.topicAdminfunção do IAM) podem ver todas as conversas no projeto.

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
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 faturação no qual ativou as APIs necessárias.

Liste mensagens numa conversa

O exemplo de código seguinte demonstra como listar todas as mensagens numa conversa enviando um pedido HTTP GET para o ponto final messages da conversa.

Para listar mensagens, tem de ter a autorização cloudaicompanion.topics.get na conversa.

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

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 faturação.
conversation_1: o ID da conversa para a qual quer listar mensagens.

Use a API para fazer perguntas

Depois de criar um agente de dados e uma conversa, pode fazer perguntas sobre os seus dados.

A API Conversational Analytics suporta conversas com várias interações, que permitem aos utilizadores fazer perguntas de seguimento baseadas no contexto anterior. A API oferece os seguintes métodos para gerir o histórico de conversas:

Chat com estado: Google Cloud armazena e gere o histórico de conversas. O chat com estado é inerentemente de várias interações, uma vez que a API retém o contexto das mensagens anteriores. Só tem de enviar a mensagem atual para cada turno.
Chat sem estado: a sua aplicação gere o histórico de conversas. Tem de incluir as mensagens anteriores relevantes em cada nova mensagem. Para ver exemplos detalhados de como gerir conversas com várias interações no modo sem estado, consulte o artigo Crie uma conversa com várias interações sem estado.

Chat com estado

Envie um pedido de chat com estado com uma referência de conversa

O exemplo de código seguinte demonstra como fazer perguntas à API usando a conversa que definiu nos passos anteriores. Este exemplo usa uma get_streamfunção auxiliar 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 código de exemplo em Crie um 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 comando de exemplo.

Chat sem estado

Envie um pedido de chat sem estado com uma referência de agente de dados

O exemplo de código seguinte demonstra como fazer uma pergunta sem estado à API usando o agente de dados que definiu nos passos anteriores. Este exemplo usa uma get_streamfunção auxiliar 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 Crie um agente de dados.
Make a bar graph for the top 5 states by the total number of airports foi usado como comando de exemplo.

Envie um pedido de chat sem estado com contexto inline

O seguinte código de exemplo demonstra como fazer uma pergunta sem estado à API através do contexto inline. Este exemplo usa uma get_streamfunção auxiliar para transmitir a resposta e usa uma origem de dados do BigQuery como exemplo.

Também pode ativar opcionalmente a análise avançada com Python, incluindo o parâmetro options no payload do pedido. Consulte a página REST Resource: projects.locations.dataAgents para ver mais informações sobre o parâmetro options e as opções que pode configurar 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)

Crie uma conversa sem estado com várias interações

Para fazer perguntas de seguimento numa conversa sem estado, a sua aplicação tem de gerir o contexto da conversa enviando todo o histórico de mensagens com cada novo pedido. As secções seguintes mostram como definir e chamar funções auxiliares para criar uma conversa com várias interações:

Envie pedidos de várias interações
Processar respostas

Envie pedidos de várias interações

A seguinte função auxiliar multi_turn_Conversation gere o contexto da conversa armazenando mensagens numa lista. Isto permite-lhe enviar perguntas de seguimento que se baseiam em interações anteriores. No payload da função, pode fazer referência a um agente de dados ou fornecer a origem de dados diretamente através do contexto em linha.

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 código de exemplo em Crie um agente de dados.

Pode chamar a função auxiliar multi_turn_Conversation para cada turno da conversa. O seguinte exemplo de código mostra como enviar um pedido inicial e, em seguida, um pedido de seguimento 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 de seguimento que se baseia na pergunta anterior ou a refina.

Processar respostas

A seguinte função get_stream_multi_turn processa a resposta da API de streaming. Esta função é semelhante à função auxiliar get_stream, mas armazena a resposta na lista conversation_messages para guardar o contexto da conversa para a interação seguinte.

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

Exemplo de código ponto a ponto

O exemplo de código expansível seguinte contém todas as tarefas abordadas neste guia.

Crie um agente de dados através de HTTP e Python

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

    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": "https://my_company.looker.com",
            "lookml_model": "my_model",
            "explore": "my_explore",
        },
        # Do not include the following line during agent creation
        # "credentials": looker_credentials
    }

    # Looker Studio data source
    looker_studio_data_source = {
        "studio":{
            "studio_references":
            [
                {
                "datasource_id": "your_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?")

O seguinte exemplo de código expansível contém as funções auxiliares do Python usadas para fazer stream de respostas de chat.

Funções Python auxiliares para transmitir respostas do 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']
      print(''.join(parts))

    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']))
        print('Question: {}'.format(query['question']))
        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 = ''

Crie um agente de dados através de HTTP e Python Mantenha tudo organizado com as coleções Salve e categorize o conteúdo com base nas suas preferências.

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

Autentique no Looker

Usar chaves da API

Usar tokens de acesso

Faça a associação a uma origem de dados

Associe-se a dados do Looker

Estabeleça ligação aos dados do BigQuery

Estabeleça ligação aos dados do Looker Studio

Crie um agente de dados

Crie uma conversa

Faça a gestão dos agentes de dados e das conversas

Obtenha um agente de dados

Apresente agentes de dados

Apresente os agentes de dados acessíveis

Atualize um agente de dados

Obtenha a Política IAM para um agente de dados

Defina a Política IAM para um agente de dados

Elimine um agente de dados

Receba uma conversa

Listar conversas

Liste mensagens numa conversa

Use a API para fazer perguntas

Chat com estado

Envie um pedido de chat com estado com uma referência de conversa

Chat sem estado

Envie um pedido de chat sem estado com uma referência de agente de dados

Envie um pedido de chat sem estado com contexto inline

Crie uma conversa sem estado com várias interações

Envie pedidos de várias interações

Processar respostas

Exemplo de código ponto a ponto

Crie um agente de dados através de HTTP e Python

Funções Python auxiliares para transmitir respostas do chat

Crie um agente de dados através de HTTP e Python