Crie um agente de dados usando o Python SDK

Esta página mostra como usar o SDK Python para fazer solicitações à API de Análise Conversacional . O código Python de exemplo demonstra como concluir as seguintes tarefas:

Autentique e configure seu ambiente

Para usar o SDK Python para a API de Análise Conversacional, siga as instruções no bloco de notas do Colaboratory sobre o SDK da API de Análise Conversacional para baixar e instalar o SDK. Observe que o método de download e o conteúdo do SDK Colab estão sujeitos a alterações.

Depois de concluir as instruções de configuração no notebook, você pode usar o seguinte código para importar as bibliotecas do SDK necessárias, autenticar sua Conta do Google em um ambiente Colaboratory e inicializar um cliente para fazer solicitações de API:

from google.colab import auth
auth.authenticate_user()

from google.cloud import geminidataanalytics

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

Especifique o projeto de faturamento e as instruções do sistema

O código Python de exemplo a seguir define o projeto de cobrança e as instruções do sistema que são usadas em todo o seu script:

# Billing project
billing_project = "my_project_name"

# System description
system_description = "Help the user analyze their data."

Substitua os valores de amostra da seguinte forma:

  • my_project_name : O ID do seu projeto de faturamento que tem as APIs necessárias habilitadas .
  • Help the user analyze their data. Instruções do sistema para orientar o comportamento do agente e personalizá-lo de acordo com suas necessidades. Por exemplo, você pode usar instruções do sistema para definir termos comerciais (como o que constitui um "cliente fiel"), controlar o tamanho da resposta ("resumir em menos de 20 palavras") ou definir a formatação dos dados ("atender aos padrões da empresa").

Conectar a uma fonte de dados

Os exemplos de código Python a seguir mostram como definir os detalhes de conexão para a fonte de dados do Looker , BigQuery ou Looker Studio que seu agente consultará para responder a perguntas.

Conectar aos dados do Looker

Os exemplos de código a seguir mostram como definir os detalhes de uma conexão com um Looker Explore com chaves de API ou um token de acesso.

Chaves de API

Você pode estabelecer uma conexão com uma instância do Looker com chaves de API do Looker geradas, conforme descrito em Autenticar e conectar-se a uma fonte de dados com a API do Conversational Analytics .

looker_client_id = "my_looker_client_id"
looker_client_secret = "my_looker_client_secret"
looker_instance_uri = "https://my_company.looker.com"
lookml_model = "my_model"
explore = "my_explore"

looker_explore_reference = geminidataanalytics.LookerExploreReference()
looker_explore_reference.looker_instance_uri = looker_instance_uri
looker_explore_reference.lookml_model = lookml_model
looker_explore_reference.explore = explore

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

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

Substitua os valores de amostra da seguinte forma:

  • my_looker_client_id : O ID do cliente da sua chave de API do Looker gerada.
  • my_looker_client_secret : O segredo do cliente da sua chave de API do Looker gerada.
  • https://my_company.looker.com : O URL completo da sua instância do Looker.
  • my_model : O nome do modelo LookML que inclui o Explore ao qual você deseja se conectar.
  • my_explore : O nome do Looker Explore que você deseja que o agente de dados consulte.

Token de acesso

Você pode estabelecer uma conexão com uma instância do Looker usando um token de acesso, conforme descrito em Autenticar e conectar-se a uma fonte de dados com a API do Conversational Analytics .

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

looker_explore_reference = geminidataanalytics.LookerExploreReference()
looker_explore_reference.looker_instance_uri = looker_instance_uri
looker_explore_reference.lookml_model = lookml_model
looker_explore_reference.explore = explore

credentials = geminidataanalytics.Credentials()
credentials.oauth.token.access_token = looker_access_token

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

Substitua os valores de amostra da seguinte forma:

  • my_access_token : O valor access_token que você gera para autenticar no Looker.
  • https://my_company.looker.com : O URL completo da sua instância do Looker.
  • my_model : O nome do modelo LookML que inclui o Explore ao qual você deseja se conectar.
  • my_explore : O nome do Looker Explore que você deseja que o agente de dados consulte.

Conectar aos dados do BigQuery

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

bq_project_id = "my_project_id"
bq_dataset_id = "my_dataset_id"
bq_table_id = "my_table_id"

bigquery_table_reference = geminidataanalytics.BigQueryTableReference()
bigquery_table_reference.project_id = bq_project_id
bigquery_table_reference.dataset_id = bq_dataset_id
bigquery_table_reference.table_id = bq_table_id

# Connect to your data source
datasource_references = geminidataanalytics.DatasourceReferences()
datasource_references.bq.table_references = [bigquery_table_reference]

Substitua os valores de amostra 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 você deseja se conectar. Para se conectar a um conjunto de dados público , especifique bigquery-public-data .
  • my_dataset_id : O ID do conjunto de dados do BigQuery. Por exemplo, san_francisco .
  • my_table_id : O ID da tabela do BigQuery. Por exemplo, street_trees .

Conecte-se aos dados do Looker Studio

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

studio_datasource_id = "my_datasource_id"

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

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

No exemplo anterior, substitua my_datasource_id pelo ID da fonte de dados.

Configurar contexto para bate-papo com ou sem estado

O código Python de exemplo a seguir demonstra como configurar o contexto para um bate-papo com ou sem estado :

  • Bate-papo com estado : Google Cloud armazena e gerencia o histórico de conversas. Você precisa enviar apenas a mensagem atual de cada vez.

  • Bate-papo sem estado : você deve enviar o histórico completo da conversa com cada mensagem.

Bate-papo com estado

O exemplo de código a seguir configura o contexto para um bate-papo com estado, onde Google Cloud armazena e gerencia o histórico de conversas. Você também pode habilitar a análise avançada com Python, incluindo a linha published_context.options.analysis.python.enabled = True no código de exemplo a seguir.

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

Bate-papo sem estado

O código de exemplo a seguir configura o contexto para um chat sem estado, onde você deve enviar todo o histórico da conversa com cada mensagem. Você também pode habilitar a análise avançada com Python, incluindo a linha inline_context.options.analysis.python.enabled = True no código de exemplo a seguir.

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

Criar um agente de dados

O código Python de exemplo a seguir faz uma solicitação de API para criar um agente de dados, que você pode usar para conversar sobre seus dados. O agente de dados é configurado com a fonte de dados, as instruções do sistema e o contexto especificados.

data_agent_id = "data_agent_1"

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

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

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

No exemplo anterior, substitua o valor data_agent_1 por um identificador exclusivo para o agente de dados.

Recuperar um agente de dados

O código Python de exemplo a seguir demonstra como fazer uma solicitação de API para recuperar um agente de dados que você criou anteriormente. 

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

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

# Handle the response
print(response)

No exemplo anterior, substitua o valor data_agent_1 pelo identificador exclusivo do agente de dados que você deseja recuperar.

Criar uma conversa

O código Python de exemplo a seguir faz uma solicitação de API para criar uma conversa. 

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

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

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

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

# Handle the response
print(response)

Substitua os valores de amostra 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 .
  • conversation_1 : Um identificador exclusivo para a conversa.

Use a API para fazer perguntas

Após criar um agente de dados e uma conversa , o código Python de exemplo a seguir envia uma consulta ao agente. O código usa o contexto que você configurou para o chat com ou sem estado . A API retorna um fluxo de mensagens que representam as etapas que o agente executa para responder à consulta.

Bate-papo com estado

Enviar uma solicitação de bate-papo com estado com uma referência Conversation

Você pode enviar uma solicitação de bate-papo com estado ao agente de dados referenciando um recurso Conversation que você criou anteriormente.

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

data_agent_id = "data_agent_1"
conversation_id = "conversation_1"

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

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

# Make the request
stream = data_chat_client.chat(request=request)

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

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.
  • data_agent_1 : O identificador exclusivo para o agente de dados, conforme definido em Criar um agente de dados .
  • conversation_1 : O identificador exclusivo da conversa, conforme definido em Criar uma conversa .

Bate-papo sem estado

Os exemplos de código a seguir demonstram como enviar uma consulta ao agente de dados quando você configura o contexto para um chat sem estado. Você pode enviar consultas sem estado referenciando um recurso DataAgent definido anteriormente ou usando o contexto embutido na solicitação.

Enviar uma solicitação de bate-papo sem estado com uma referência DataAgent

Você pode enviar uma consulta ao agente de dados referenciando um recurso DataAgent criado anteriormente.

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

data_agent_id = "data_agent_1"

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

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

# Make the request
stream = data_chat_client.chat(request=request)

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

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.
  • data_agent_1 : O identificador exclusivo para o agente de dados, conforme definido em Criar um agente de dados .

Enviar uma solicitação de bate-papo sem estado com contexto em linha

O código de exemplo a seguir demonstra como usar o parâmetro inline_context para fornecer contexto diretamente na sua solicitação de bate-papo sem estado. 

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

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

# Make the request
stream = data_chat_client.chat(request=request)

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

No exemplo anterior, substitua Which species of tree is most prevalent? por uma pergunta em linguagem natural para enviar ao agente de dados.

Fazer solicitações multi-turn

Você pode criar uma conversa multi-turno enviando perguntas de acompanhamento ao agente de dados. O código de exemplo a seguir demonstra como fazer solicitações multi-turno, aproveitando respostas anteriores para refinar a conversa. Para mais informações, consulte Criar uma conversa multi-turno . 

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

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

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

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

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

  # Make the request
  stream = data_chat_client.chat(request=request)

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

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

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

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 desenvolve ou refina a pergunta anterior.

Definir funções auxiliares

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

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

import proto
from google.protobuf.json_format import MessageToDict, MessageToJson

def handle_text_response(resp):
  parts = getattr(resp, 'parts')
  print(''.join(parts))

def display_schema(data):
  fields = getattr(data, 'fields')
  df = pd.DataFrame({
    "Column": map(lambda field: getattr(field, 'name'), fields),
    "Type": map(lambda field: getattr(field, 'type'), fields),
    "Description": map(lambda field: getattr(field, 'description', '-'), fields),
    "Mode": map(lambda field: getattr(field, 'mode'), fields)
  })
  display(df)

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

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

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

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

  print(source_name)
  display_schema(datasource.schema)

def handle_schema_response(resp):
  if 'query' in resp:
    print(resp.query.question)
  elif 'result' in resp:
    display_section_title('Schema resolved')
    print('Data sources:')
    for datasource in resp.result.datasources:
      display_datasource(datasource)

def handle_data_response(resp):
  if 'query' in resp:
    query = resp.query
    display_section_title('Retrieval query')
    print('Query name: {}'.format(query.name))
    print('Question: {}'.format(query.question))
    print('Data sources:')
    for datasource in query.datasources:
      display_datasource(datasource)
  elif 'generated_sql' in resp:
    display_section_title('SQL generated')
    print(resp.generated_sql)
  elif 'result' in resp:
    display_section_title('Data retrieved')

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

    display(pd.DataFrame(d))

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

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

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

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