Crear un agente de datos con HTTP y Python

En esta página se explica cómo usar Python para enviar solicitudes HTTP a la API Conversational Analytics (a la que se accede a través de geminidataanalytics.googleapis.com).

El código de muestra de Python de esta página muestra cómo completar las siguientes tareas:

Al final de la página se incluye una versión completa del código de muestra, junto con las funciones auxiliares que se usan para transmitir la respuesta de la API.

Configurar los ajustes iniciales y la autenticación

El siguiente código de muestra de Python realiza estas tareas:

  • Importa las bibliotecas de Python necesarias.
  • Obtiene un token de acceso para la autenticación HTTP mediante la CLI de Google Cloud
  • Define las variables del proyecto de facturación y las instrucciones del 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'

Sustituye los valores de ejemplo de la siguiente manera:

  • YOUR-BILLING-PROJECT: el ID del proyecto de facturación en el que has habilitado las APIs necesarias.
  • YOUR-SYSTEM-INSTRUCTIONS: instrucciones del sistema para guiar el comportamiento del agente y personalizarlo según tus necesidades de datos. Por ejemplo, puedes usar instrucciones del sistema para definir términos empresariales, controlar la longitud de las respuestas o definir el formato de los datos. Lo ideal es definir las instrucciones del sistema con el formato YAML recomendado en Escribir instrucciones del sistema eficaces para proporcionar directrices detalladas y estructuradas.

Autenticarse en Looker

Si tienes previsto conectarte a una fuente de datos de Looker, tendrás que autenticarte en la instancia de Looker.

Utilizar claves de API

El siguiente código de ejemplo de Python muestra cómo autenticar tu agente en una instancia de Looker mediante claves de API.

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

Sustituye los valores de ejemplo de la siguiente manera:

  • YOUR-LOOKER-CLIENT-ID: el ID de cliente de la clave de API de Looker que has generado.
  • YOUR-LOOKER-CLIENT-SECRET: el secreto de cliente de la clave de API de Looker que has generado.

Usar tokens de acceso

El siguiente código de ejemplo de Python muestra cómo autenticar tu agente en una instancia de Looker mediante tokens de acceso.

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

Sustituye los valores de ejemplo de la siguiente manera:

  • YOUR-TOKEN: el valor access_token que generas para autenticarte en Looker.

Conectarse a una fuente de datos

En las siguientes secciones se muestra cómo definir los detalles de conexión de las fuentes de datos de tu agente. Tu agente puede conectarse a datos de Looker, BigQuery o Looker Studio.

Conectarse a datos de Looker

El siguiente código de ejemplo define una conexión a una Exploración de Looker. Para establecer una conexión con una instancia de Looker, comprueba que has generado claves de API de Looker, tal como se describe en el artículo Autenticar y conectar con una fuente de datos mediante la API Conversational Analytics. Con la API Conversational Analytics, solo puedes conectarte a una Exploración de Looker a la vez.

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

Sustituye los valores de ejemplo de la siguiente manera:

  • https://your_company.looker.com: URL completa de su instancia de Looker.
  • your_model: nombre del modelo de LookML que incluye la Exploración a la que quieres conectarte.
  • your_explore: el nombre de la Exploración de Looker que quieres que consulte el agente de datos.

Conectar con datos de BigQuery

Con la API Conversational Analytics, no hay límites estrictos en el número de tablas de BigQuery a las que puede conectarse. Sin embargo, si te conectas a un gran número de tablas, puede que la precisión se reduzca o que superes el límite de tokens de entrada del modelo.

El siguiente código de ejemplo define una conexión a varias tablas de BigQuery.

bigquery_data_sources = {
    "bq": {
        "tableReferences": [
            {
                "projectId": "my_project_id",
                "datasetId": "my_dataset_id",
                "tableId": "my_table_id"
            },
            {
                "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"
            },
        ]
    }
}

Sustituye los valores de ejemplo de la siguiente manera:

  • my_project_id: ID del Google Cloud proyecto que contiene el conjunto de datos y la tabla de BigQuery a los que quiere conectarse. Para conectarte a un conjunto de datos público, especifica bigquery-public-data.
  • my_dataset_id: ID del conjunto de datos de BigQuery.
  • my_table_id: el ID de la tabla de BigQuery.

Conectarse a datos de Looker Studio

El siguiente código de ejemplo define una conexión a una fuente de datos de Looker Studio.

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

Sustituye studio_datasource_id por el ID de la fuente de datos.

Crear un agente de datos

En el siguiente código de ejemplo se muestra cómo crear el agente de datos enviando una solicitud HTTP POST al endpoint de creación del agente de datos. La carga útil de la solicitud incluye los siguientes detalles:

También puedes habilitar el análisis avanzado con Python de forma opcional incluyendo el parámetro options en la carga útil de la solicitud.

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)

Sustituye los valores de ejemplo de la siguiente manera:

  • data_agent_1: identificador único del agente de datos. Este valor se usa en el nombre de recurso del agente y como parámetro de consulta de URL data_agent_id.
  • This is the description of data_agent_1.: descripción del agente de datos.

Crear una conversación

En el siguiente ejemplo de código se muestra cómo crear una conversación con tu agente de datos.

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)

Sustituye los valores de ejemplo de la siguiente manera:

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

Gestionar agentes de datos y conversaciones

Los siguientes ejemplos de código muestran cómo gestionar tus agentes de datos y conversaciones mediante la API Conversational Analytics. Puedes realizar las siguientes tareas:

Obtener un agente de datos

En el siguiente código de ejemplo se muestra cómo obtener un agente de datos enviando una solicitud HTTP GET a la URL del recurso del agente de datos.

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)

En el ejemplo anterior, sustituye data_agent_1 por el ID del agente de datos que quieras obtener.

Mostrar agentes de datos

En el siguiente código se muestra cómo enumerar todos los agentes de datos de un proyecto determinado enviando una solicitud HTTP GET al endpoint dataAgents.

Para ver todos los agentes, debes tener el permiso geminidataanalytics.dataAgents.list en el proyecto. Para obtener más información sobre qué roles de gestión de identidades y accesos incluyen este permiso, consulta la lista de roles 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 Agents Listed successfully!")
    print(json.dumps(data_agent_response.json(), indent=2))
else:
    print(f"Error Listing Data Agents: {data_agent_response.status_code}")

Sustituye YOUR-BILLING-PROJECT por el ID de tu proyecto de facturación.

Mostrar agentes de datos accesibles

En el siguiente código se muestra cómo enumerar todos los agentes de datos accesibles de un proyecto determinado enviando una solicitud HTTP GET al endpoint 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}")

Sustituye los valores de ejemplo de la siguiente manera:

  • YOUR-BILLING-PROJECT: el ID de tu proyecto de facturación.
  • YOUR-CREATOR-FILTER: el filtro que se va a aplicar en función del creador del agente de datos. Los valores posibles son NONE (predeterminado), CREATOR_ONLY y NOT_CREATOR_ONLY.

Actualizar un agente de datos

En el siguiente código de ejemplo se muestra cómo actualizar un agente de datos enviando una solicitud HTTP PATCH a la URL del recurso del agente de datos. La carga útil de la solicitud incluye los nuevos valores de los campos que quiere cambiar, y los parámetros de la solicitud incluyen un parámetro updateMask, que especifica los campos que se van a actualizar.

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)

Sustituye los valores de ejemplo de la siguiente manera:

  • data_agent_1: el ID del agente de datos que quieres actualizar.
  • YOUR-BILLING-PROJECT: el ID de tu proyecto de facturación.
  • Updated description of the data agent.: una nueva descripción del agente de datos.

Definir la política de gestión de identidades y accesos de un agente de datos

Para compartir un agente, puedes usar el método setIamPolicy para asignar roles de gestión de identidades y accesos a los usuarios de un agente específico. En el siguiente ejemplo de código se muestra cómo hacer una llamada POST a la URL del agente de datos con una carga útil que incluya vinculaciones. El enlace especifica qué roles se deben asignar a qué usuarios.

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)

Sustituye los valores de ejemplo de la siguiente manera:

  • YOUR-BILLING-PROJECT: el ID de tu proyecto de facturación.
  • data_agent_1: el ID del agente de datos para el que quieres definir la política de gestión de identidades y accesos.
  • 222larabrown@gmail.com, cloudysanfrancisco@gmail.com: lista separada por comas de las direcciones de correo de los usuarios a los que quieres asignar el rol especificado.

Obtener la política de gestión de identidades y accesos de un agente de datos

En el siguiente ejemplo de código se muestra cómo obtener la política de IAM de un agente de datos enviando una solicitud HTTP POST a la URL del agente de datos. La carga útil de la solicitud incluye la ruta del agente de datos.

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)

Sustituye los valores de ejemplo de la siguiente manera:

  • YOUR-BILLING-PROJECT: el ID de tu proyecto de facturación.
  • data_agent_1: el ID del agente de datos del que quiere obtener la política de gestión de identidades y accesos.

Eliminar un agente de datos

En el siguiente ejemplo de código se muestra cómo eliminar de forma lógica un agente de datos enviando una solicitud HTTP DELETE a la URL del recurso del agente de datos. La eliminación no definitiva significa que el agente se elimina, pero se puede recuperar en un plazo de 30 días.

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)

Sustituye los valores de ejemplo de la siguiente manera:

  • YOUR-BILLING-PROJECT: el ID de tu proyecto de facturación.
  • data_agent_1: el ID del agente de datos que quieres eliminar.

Obtener una conversación

En el siguiente código de ejemplo se muestra cómo obtener una conversación enviando una solicitud HTTP GET a la URL del recurso de conversación.

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)

Sustituye los valores de ejemplo de la siguiente manera:

  • YOUR-BILLING-PROJECT: el ID de tu proyecto de facturación.
  • conversation_1: el ID de la conversación que quieres obtener.

Mostrar conversaciones

En el siguiente código de ejemplo se muestra cómo obtener una lista de las conversaciones de un proyecto determinado enviando una solicitud HTTP GET al endpoint conversations.

De forma predeterminada, este método devuelve las conversaciones que has creado. Los administradores (usuarios con el rol de gestión de identidades y accesos cloudaicompanion.topicAdmin) pueden ver todas las conversaciones del proyecto.

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)

Sustituye YOUR-BILLING-PROJECT por el ID del proyecto de facturación en el que has habilitado las APIs necesarias.

Listar los mensajes de una conversación

El siguiente código de muestra muestra cómo enumerar todos los mensajes de una conversación enviando una solicitud HTTP GET al endpoint messages de la conversación.

Para ver los mensajes, debes tener el permiso cloudaicompanion.topics.get en la conversación.

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)

Sustituye los valores de ejemplo de la siguiente manera:

  • YOUR-BILLING-PROJECT: el ID de tu proyecto de facturación.
  • conversation_1: el ID de la conversación de la que quieres obtener una lista de mensajes.

Usar la API para hacer preguntas

Una vez que hayas creado un agente de datos y una conversación, podrás hacer preguntas sobre tus datos.

La API Conversational Analytics admite conversaciones de varios turnos, lo que permite a los usuarios hacer preguntas de seguimiento que se basan en el contexto anterior. La API proporciona los siguientes métodos para gestionar el historial de conversaciones:

  • Chat con estado: Google Cloud almacena y gestiona el historial de la conversación. El chat con estado es inherentemente de varias interacciones, ya que la API conserva el contexto de los mensajes anteriores. Solo tienes que enviar el mensaje actual en cada turno.

  • Chat sin estado: tu aplicación gestiona el historial de conversaciones. Debes incluir los mensajes anteriores pertinentes en cada nuevo mensaje. Para ver ejemplos detallados de cómo gestionar conversaciones de varios turnos en modo sin estado, consulta Crear una conversación de varios turnos sin estado.

Chat con estado

Enviar una solicitud de chat con estado con una referencia de conversación

En el siguiente ejemplo de código se muestra cómo hacer preguntas a la API usando la conversación que has definido en los pasos anteriores. En este ejemplo se usa una función auxiliar get_stream para transmitir la respuesta.

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)

Sustituye los valores de ejemplo de la siguiente manera:

  • data_agent_1: el ID del agente de datos, tal como se define en el bloque de código de ejemplo de Crear un agente de datos.
  • conversation_1: identificador único de la conversación.
  • Se ha usado Make a bar graph for the top 5 states by the total number of airports como ejemplo de petición.

Chat sin estado

Enviar una solicitud de chat sin estado con una referencia de agente de datos

En el siguiente código de ejemplo se muestra cómo hacer una pregunta sin estado a la API mediante el agente de datos que has definido en los pasos anteriores. En este ejemplo se usa una función auxiliar get_stream para transmitir la respuesta.

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)

Sustituye los valores de ejemplo de la siguiente manera:

  • data_agent_1: el ID del agente de datos, tal como se define en el bloque de código de ejemplo de Crear un agente de datos.
  • Se ha usado Make a bar graph for the top 5 states by the total number of airports como ejemplo de petición.

Enviar una solicitud de chat sin estado con contexto insertado

El siguiente código de muestra muestra cómo hacer una pregunta sin estado a la API mediante el contexto insertado. En este ejemplo se usa una función auxiliar get_stream para transmitir la respuesta y se usa una fuente de datos de BigQuery.

También puedes habilitar el análisis avanzado con Python de forma opcional incluyendo el parámetro options en la carga útil de la solicitud.

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)

Crear una conversación de varios turnos sin estado

Para hacer preguntas de seguimiento en una conversación sin estado, tu aplicación debe gestionar el contexto de la conversación enviando todo el historial de mensajes con cada nueva solicitud. En las siguientes secciones se muestra cómo definir y llamar a funciones de ayuda para crear una conversación de varios turnos:

Enviar solicitudes de varios turnos

La siguiente función auxiliar multi_turn_Conversation gestiona el contexto de la conversación almacenando los mensajes en una lista. De esta forma, puedes enviar preguntas de seguimiento que se basen en las respuestas anteriores. En la carga útil de la función, puedes hacer referencia a un agente de datos o proporcionar la fuente de datos directamente mediante el contexto insertado.

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)

En el ejemplo anterior, sustituya data_agent_1 por el ID del agente de datos, tal como se define en el bloque de código de ejemplo de Crear un agente de datos.

Puedes llamar a la función auxiliar multi_turn_Conversation en cada turno de la conversación. En el siguiente código de ejemplo se muestra cómo enviar una solicitud inicial y, después, una solicitud de seguimiento que se basa en la respuesta 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?")

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

  • Which species of tree is most prevalent?: pregunta en lenguaje natural que se envía al agente de datos.
  • Can you show me the results as a bar chart?: una pregunta de seguimiento que se basa en la pregunta anterior o la mejora.

Procesar respuestas

La función get_stream_multi_turn siguiente procesa la respuesta de la API de streaming. Esta función es similar a la función de ayuda get_stream, pero almacena la respuesta en la lista conversation_messages para guardar el contexto de la conversación para el siguiente turno.

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

Ejemplo de código completo

El siguiente ejemplo de código desplegable contiene todas las tareas que se tratan en esta guía.

Crear un agente de datos con HTTP y 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?")

El siguiente código de muestra desplegable contiene las funciones auxiliares de Python que se usan para transmitir respuestas de chat.

Funciones auxiliares de Python para transmitir respuestas 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 = ''