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 ou do Looker Studio
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

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

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

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

Importa as bibliotecas necessárias do Python.
Recebe um token de acesso para autenticação HTTP usando a CLI do Google Cloud.
Define variáveis para o projeto de faturamento e 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",
}

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

Autenticar no Looker

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

Como usar chaves de API

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

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

Substitua os valores de exemplo da seguinte forma:

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

Como usar tokens de acesso

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

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

Substitua os valores de exemplo da seguinte forma:

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

Conectar a uma fonte de dados

Os exemplos de código Python a seguir mostram como definir a fonte de dados do Looker, do BigQuery ou do Looker Studio para uso do seu agente.

Conectar aos dados do Looker

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

looker_data_source = {
    "looker": {
        "explore_references": {
            "looker_instance_uri": "https://your_company.looker.com",
            "lookml_model": "your_model",
            "explore": "your_explore",
       },
       # Do not include the following line during agent creation
       "credentials": looker_credentials
    }
}

Substitua os valores de exemplo da seguinte forma:

https://your_company.looker.com: o URL completo da sua instância do Looker.
your_model: o nome do modelo do LookML que inclui a Análise de destino.
your_explore: o nome da Análise do Looker que o agente de dados deve consultar.

Conectar-se aos dados do BigQuery

Com a API Conversational Analytics, é possível acessar e consultar até 10 tabelas do BigQuery por vez.

O exemplo de código a seguir define uma conexão com uma tabela do BigQuery.

bigquery_data_sources = {
    "bq": {
        "tableReferences": [
            {
                "projectId": "my_project_id",
                "datasetId": "my_dataset_id",
                "tableId": "my_table_id"
            },
            {
                "projectId": "my_project_id_2",
                "datasetId": "amy_dataset_id_2",
                "tableId": "my_table_id_2"
            },
            # Up to 10 total tables may be included
        ]
    }
}

Substitua os valores de exemplo da seguinte forma:

my_project_id: o ID do projeto do Google Cloud que contém o conjunto de dados e a tabela do BigQuery a que você quer se conectar. Para se conectar a um conjunto de dados público, especifique bigquery-public-data.
my_dataset_id: ID do conjunto de dados do BigQuery.
my_table_id: ID da tabela do BigQuery.

Conectar aos dados do Looker Studio

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

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

Substitua studio_datasource_id pelo ID da fonte de dados.

Criar um agente de dados

O exemplo de código a seguir demonstra como criar o agente de dados 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).

Você também pode ativar a análise avançada com Python incluindo o parâmetro options no payload da solicitação.

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 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.
This is the description of data_agent_1.: uma descrição para o agente de dados.

Criar uma conversa

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

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 exemplo de código em Criar agente de dados.
conversation_1: um identificador exclusivo da conversa.

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
Atualizar um agente de dados
Definir a política do IAM de um agente de dados
Acessar 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_1"
data_agent_url = f"{base_url}/v1beta/projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"

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

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

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

Listar agentes de dados

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

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

billing_project = "YOUR-BILLING-PROJECT"
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 Agent Listed successfully!")
    print(json.dumps(data_agent_response.json(), indent=2))
else:
    print(f"Error Listing Data Agent: {data_agent_response.status_code}")

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

Atualizar um agente de dados

O exemplo de código a seguir demonstra como atualizar um agente de dados 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.

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 você quer atualizar.
YOUR-BILLING-PROJECT: o ID do seu projeto de faturamento.
Updated description of the data agent.: uma nova descrição para o agente de dados.

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

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

billing_project = "YOUR-BILLING-PROJECT"
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 faturamento.
data_agent_1: o ID do agente de dados para o qual você quer definir a política do IAM.
222larabrown@gmail.com, cloudysanfrancisco@gmail.com: uma lista separada por vírgulas de e-mails de usuários a quem você quer conceder o papel especificado.

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

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

billing_project = "YOUR-BILLING-PROJECT"
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 faturamento.
data_agent_1: o ID do agente de dados cuja política do IAM você quer buscar.

Excluir um agente de dados

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

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 faturamento.
data_agent_1: o ID do agente de dados que você quer excluir.

Acessar uma conversa

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

billing_project = "YOUR-BILLING-PROJECT"
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 faturamento.
conversation_1: o ID da conversa que você quer buscar.

Listar conversas

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

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

billing_project = "YOUR-BILLING-PROJECT"
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 faturamento em que você ativou as APIs necessárias.

Listar mensagens em uma conversa

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

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

billing_project = "YOUR-BILLING-PROJECT"
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 faturamento.
conversation_1: o ID da conversa cujas mensagens você quer listar.

Usar a API para fazer perguntas

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

A API Conversational Analytics é compatível com conversas multiturno, que permitem aos usuários fazer perguntas de acompanhamento com base no contexto anterior. A API fornece os seguintes métodos para gerenciar o histórico de conversas:

Chat com estado: 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.

Chat com estado

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

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

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

data_agent_id = "data_agent_1"
conversation_id = "conversation_1"

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

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

Substitua os valores de exemplo da seguinte forma:

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

Chat sem estado

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

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

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

data_agent_id = "data_agent_1"

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

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

Substitua os valores de exemplo da seguinte forma:

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

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

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

Você também pode ativar a análise avançada com Python incluindo o parâmetro options no payload da solicitação.

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)

Criar uma conversa multiturno sem estado

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

Enviar solicitações multiturno
Processar respostas

Enviar solicitações multiturno

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

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

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

data_agent_id = "data_agent_1"

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

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

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

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

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

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

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

# Send first-turn request
multi_turn_Conversation("Which species of tree is most prevalent?")

# Send follow-up-turn request
multi_turn_Conversation("Can you show me the results as a bar chart?")

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

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

Processar respostas

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

def get_stream_multi_turn(url, json, conversation_messages):
    s = requests.Session()

    acc = ''

    with s.post(url, json=json, headers=headers, stream=True) as resp:
        for line in resp.iter_lines():
            if not line:
                continue

            decoded_line = str(line, encoding='utf-8')

            if decoded_line == '[{':
                acc = '{'
            elif decoded_line == '}]':
                acc += '}'
            elif decoded_line == ',':
                continue
            else:
                acc += decoded_line

            if not is_json(acc):
                continue

            data_json = json_lib.loads(acc)
            # Store the response that will be used in the next iteration
            conversation_messages.append(data_json)

            if not 'systemMessage' in data_json:
                if 'error' in data_json:
                    handle_error(data_json['error'])
                continue

            if 'text' in data_json['systemMessage']:
                handle_text_response(data_json['systemMessage']['text'])
            elif 'schema' in data_json['systemMessage']:
                handle_schema_response(data_json['systemMessage']['schema'])
            elif 'data' in data_json['systemMessage']:
                handle_data_response(data_json['systemMessage']['data'])
            elif 'chart' in data_json['systemMessage']:
                handle_chart_response(data_json['systemMessage']['chart'])
            else:
                colored_json = highlight(acc, lexers.JsonLexer(), formatters.TerminalFormatter())
                print(colored_json)
            print('\n')
            acc = ''

Exemplo de código completo

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

Criar um agente de dados usando 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",
    }

    ################### 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 exemplo de código expansível a seguir contém as funções auxiliares do Python usadas para transmitir respostas de chat.

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

    def is_json(str):
      try:
          json_object = json_lib.loads(str)
      except ValueError as e:
          return False
      return True

    def handle_text_response(resp):
      parts = resp['parts']
      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 = ''

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

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

Autenticar no Looker

Como usar chaves de API

Como usar tokens de acesso

Conectar a uma fonte de dados

Conectar aos dados do Looker

Conectar-se aos dados do BigQuery

Conectar aos dados do Looker Studio

Criar um agente de dados

Criar uma conversa

Gerenciar agentes de dados e conversas

Receber um agente de dados

Listar agentes de dados

Atualizar um agente de dados

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

Acessar 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

Usar a API para fazer perguntas

Chat com estado

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

Chat sem estado

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

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

Criar uma conversa multiturno sem estado

Enviar solicitações multiturno

Processar respostas

Exemplo de código completo

Criar um agente de dados usando HTTP e Python

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

Criar um agente de dados usando HTTP e Python