Construya un agente de datos usando el SDK de Python

Esta página muestra cómo usar el SDK de Python para realizar solicitudes a la API de Análisis Conversacional . El código de ejemplo de Python muestra cómo realizar las siguientes tareas:

Autenticar y configurar su entorno

Para usar el SDK de Python para la API de Análisis Conversacional, siga las instrucciones del cuaderno de Colaboratory sobre el SDK de la API de Análisis Conversacional para descargarlo e instalarlo. Tenga en cuenta que el método de descarga y el contenido del SDK de Colab están sujetos a cambios.

Una vez que haya completado las instrucciones de configuración en el cuaderno, puede usar el siguiente código para importar las bibliotecas SDK necesarias, autenticar su cuenta de Google dentro de un entorno de Colaboratory e inicializar un cliente para realizar solicitudes 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 el proyecto de facturación y las instrucciones del sistema.

El siguiente código Python de muestra define el proyecto de facturación y las instrucciones del sistema que se utilizan en todo el script:

# Billing project
billing_project = "my_project_name"

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

Reemplace los valores de muestra de la siguiente manera:

  • my_project_name : el ID de su proyecto de facturación que tiene las API requeridas habilitadas .
  • Help the user analyze their data. Instrucciones del sistema para guiar el comportamiento del agente y personalizarlo según sus necesidades. Por ejemplo, puede usar las instrucciones del sistema para definir términos comerciales (como qué constituye un "cliente fiel"), controlar la longitud de la respuesta ("resumir en menos de 20 palabras") o configurar el formato de los datos ("adaptarse a los estándares de la empresa").

Conectarse a una fuente de datos

Los siguientes ejemplos de código de Python muestran cómo definir los detalles de conexión para la fuente de datos Looker , BigQuery o Looker Studio que su agente consultará para responder preguntas.

Conectarse a los datos de Looker

Los siguientes ejemplos de código muestran cómo definir los detalles para una conexión a Looker Explore con claves API o un token de acceso.

Claves API

Puede establecer una conexión con una instancia de Looker con claves de API de Looker generadas, como se describe en Autenticar y conectarse a una fuente de datos con la API de análisis conversacional .

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]

Reemplace los valores de muestra de la siguiente manera:

  • my_looker_client_id : el ID de cliente de su clave API de Looker generada.
  • my_looker_client_secret : el secreto de cliente de su clave API de Looker generada.
  • https://my_company.looker.com : la URL completa de su instancia de Looker.
  • my_model : el nombre del modelo LookML que incluye el Explore al que desea conectarse.
  • my_explore : el nombre del Looker Explore que desea que el agente de datos consulte.

Token de acceso

Puede establecer una conexión con una instancia de Looker mediante un token de acceso, como se describe en Autenticar y conectarse a una fuente de datos con la API de 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]

Reemplace los valores de muestra de la siguiente manera:

  • my_access_token : el valor de access_token que genera para autenticarse en Looker.
  • https://my_company.looker.com : la URL completa de su instancia de Looker.
  • my_model : el nombre del modelo LookML que incluye el Explore al que desea conectarse.
  • my_explore : el nombre del Looker Explore que desea que el agente de datos consulte.

Conectarse a los datos de BigQuery

El siguiente código de muestra define una conexión a una sola tabla de 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]

Reemplace los valores de muestra de la siguiente manera:

  • my_project_id : El ID del proyecto Google Cloud Proyecto que contiene el conjunto de datos y la tabla de BigQuery a los que desea conectarse. Para conectarse a un conjunto de datos público , especifique bigquery-public-data .
  • my_dataset_id : El ID del conjunto de datos de BigQuery. Por ejemplo, san_francisco .
  • my_table_id : El ID de la tabla de BigQuery. Por ejemplo, street_trees .

Conectarse a los datos de Looker Studio

El siguiente código de muestra define una conexión a una fuente de datos de 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]

En el ejemplo anterior, reemplace my_datasource_id con el ID de la fuente de datos.

Configurar el contexto para el chat con estado o sin estado

El siguiente código de Python de muestra demuestra cómo configurar el contexto para un chat con estado o sin estado :

  • Chat con estado : Google Cloud Almacena y gestiona el historial de conversaciones. Solo necesitas enviar el mensaje actual en cada turno.

  • Chat sin estado : debes enviar el historial completo de la conversación con cada mensaje.

Chat con estado

El siguiente ejemplo de código configura el contexto para el chat con estado, donde Google Cloud Almacena y gestiona el historial de conversaciones. También puede habilitar el análisis avanzado con Python incluyendo la línea published_context.options.analysis.python.enabled = True en el siguiente código de ejemplo.

# 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

Chat sin estado

El siguiente código de ejemplo configura el contexto para un chat sin estado, donde se debe enviar el historial completo de la conversación con cada mensaje. También puede habilitar el análisis avanzado con Python incluyendo la línea inline_context.options.analysis.python.enabled = True en el siguiente código de ejemplo.

# 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

Crear un agente de datos

El siguiente código de ejemplo de Python realiza una solicitud a la API para crear un agente de datos, que puede usar para conversar sobre sus datos. El agente de datos se configura con la fuente de datos, las instrucciones del sistema y el 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}")

En el ejemplo anterior, reemplace el valor data_agent_1 con un identificador único para el agente de datos.

Recuperar un agente de datos

El siguiente código Python de muestra demuestra cómo realizar una solicitud de API para recuperar un agente de datos que creó previamente. 

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

En el ejemplo anterior, reemplace el valor data_agent_1 con el identificador único del agente de datos que desea recuperar.

Crear una conversación

El siguiente código Python de ejemplo realiza una solicitud de API para crear una conversación. 

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

Reemplace los valores de muestra de la siguiente manera:

  • data_agent_1 : el ID del agente de datos, tal como se define en el bloque de código de muestra en Crear un agente de datos .
  • conversation_1 : Un identificador único para la conversación.

Utilice la API para hacer preguntas

Tras crear un agente de datos y una conversación , el siguiente código de Python de ejemplo envía una consulta al agente. El código utiliza el contexto configurado para chat con o sin estado . La API devuelve un flujo de mensajes que representa los pasos que el agente realiza para responder a la consulta.

Chat con estado

Enviar una solicitud de chat con estado con una referencia Conversation

Puede enviar una solicitud de chat con estado al agente de datos haciendo referencia a un recurso Conversation que haya creado previamente.

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

Reemplace los valores de muestra de la siguiente manera:

  • Which species of tree is most prevalent? Una pregunta en lenguaje natural para enviar al agente de datos.
  • data_agent_1 : el identificador único del agente de datos, como se define en Crear un agente de datos .
  • conversation_1 : El identificador único de la conversación, como se define en Crear una conversación .

Chat sin estado

Los siguientes ejemplos de código muestran cómo enviar una consulta al agente de datos tras configurar el contexto para el chat sin estado. Puede enviar consultas sin estado haciendo referencia a un recurso DataAgent previamente definido o utilizando contexto en línea en la solicitud.

Envíe una solicitud de chat sin estado con una referencia DataAgent

Puede enviar una consulta al agente de datos haciendo referencia a un recurso DataAgent que haya creado previamente.

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

Reemplace los valores de muestra de la siguiente manera:

  • Which species of tree is most prevalent? Una pregunta en lenguaje natural para enviar al agente de datos.
  • data_agent_1 : el identificador único del agente de datos, como se define en Crear un agente de datos .

Enviar una solicitud de chat sin estado con contexto en línea

El siguiente código de muestra demuestra cómo utilizar el parámetro inline_context para proporcionar contexto directamente dentro de su solicitud de chat sin 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)

En el ejemplo anterior, reemplace Which species of tree is most prevalent? con una pregunta en lenguaje natural para enviar al agente de datos.

Realizar solicitudes multiturno

Puede crear una conversación multiturno enviando preguntas de seguimiento al agente de datos. El siguiente código de ejemplo muestra cómo realizar solicitudes multiturno, basándose en respuestas anteriores para refinar la conversación. Para obtener más información, consulte Crear una conversación multiturno . 

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

En el ejemplo anterior, reemplace los valores de muestra de la siguiente manera:

  • Which species of tree is most prevalent? Una pregunta en lenguaje natural para enviar al agente de datos.
  • Can you show me the results as a bar chart? : Una pregunta de seguimiento que amplía o perfecciona la pregunta anterior.

Definir funciones auxiliares

El siguiente código de ejemplo contiene definiciones de funciones auxiliares utilizadas en los ejemplos de código anteriores. Estas funciones ayudan a analizar la respuesta de la API y a mostrar los 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')