Se usó la API de Cloud Translation para traducir esta página.

Compila un agente de datos con HTTP y Python

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

En el código de ejemplo de Python que se incluye en esta página, se muestra cómo completar las siguientes tareas:

Configura los parámetros iniciales y la autenticación
Conéctate a una fuente de datos de Looker, BigQuery o Looker Studio
Crea un agente de datos
Cómo crear una conversación
Administra agentes de datos y conversaciones
Usa la API para hacer preguntas
Crea una conversación sin estado de varios turnos

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.

Configura los parámetros iniciales y la autenticación

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

Importa las bibliotecas de Python necesarias
Obtiene un token de acceso para la autenticación HTTP con Google Cloud CLI
Define variables para el 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'

Reemplaza los valores de muestra de la siguiente manera:

YOUR-BILLING-PROJECT: Es el ID del proyecto de facturación en el que habilitaste las APIs requeridas.
YOUR-SYSTEM-INSTRUCTIONS: Son 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 comerciales, controlar la longitud de la respuesta o configurar el formato de los datos. Lo ideal es definir las instrucciones del sistema con el formato YAML recomendado en Cómo escribir instrucciones del sistema eficaces para proporcionar orientación detallada y estructurada.

Autenticación en Looker

Si planeas conectarte a una fuente de datos de Looker, deberás autenticarte en la instancia de Looker.

Cómo usar claves de API

En el siguiente muestra de código de Python, se muestra cómo autenticar tu agente en una instancia de Looker con claves de API.

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

Reemplaza los valores de muestra de la siguiente manera:

YOUR-LOOKER-CLIENT-ID: Es el ID de cliente de la clave de la API de Looker que generaste.
YOUR-LOOKER-CLIENT-SECRET: Es el secreto del cliente de la clave de API de Looker que generaste.

Cómo usar tokens de acceso

En el siguiente muestra de código de Python, se muestra cómo autenticar tu agente en una instancia de Looker con tokens de acceso.

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

Reemplaza los valores de muestra de la siguiente manera:

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

Conectándose a una fuente de datos

En los siguientes ejemplos de código de Python, se muestra cómo definir la fuente de datos de Looker, BigQuery o Looker Studio para que la use tu agente.

Conéctate a los datos de Looker

En el siguiente código de muestra, se define una conexión a un Explorar de Looker. Para establecer una conexión con una instancia de Looker, verifica que hayas generado claves de la API de Looker, como se describe en Cómo autenticarse y conectarse a una fuente de datos con la API de Conversational Analytics. Con la API de 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
    }
}

Reemplaza los valores de muestra de la siguiente manera:

https://your_company.looker.com: Es la URL completa de tu instancia de Looker.
your_model: Es el nombre del modelo de LookML que incluye la exploración a la que deseas conectarte.
your_explore: Es el nombre de la exploración de Looker que deseas que consulte el agente de datos.

Conéctate a los datos de BigQuery

Con la API de Conversational Analytics, puedes conectarte a un máximo de 10 tablas de BigQuery y realizar consultas en ellas al mismo tiempo.

En el siguiente código de ejemplo, se define una conexión a una tabla 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": "amy_dataset_id_2",
                "tableId": "my_table_id_2"
            },
            # Up to 10 total tables may be included
        ]
    }
}

Reemplaza los valores de muestra de la siguiente manera:

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

Conéctate a los datos de Looker Studio

En el siguiente código de muestra, se 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"
            }
        ]
    }
}

Reemplaza studio_datasource_id por el ID de la fuente de datos.

Crea un agente de datos

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

Es el nombre completo del recurso del agente. Este valor incluye el ID del proyecto, la ubicación y un identificador único para el agente.
Es la descripción del agente de datos.
El contexto del agente de datos, incluida la descripción del sistema (definida en Configura los parámetros iniciales y la autenticación) y la fuente de datos que usa el agente (definida en Conéctate a una fuente de datos).

También puedes habilitar de forma opcional el análisis avanzado con Python si incluyes 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)

Reemplaza los valores de muestra de la siguiente manera:

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

Cómo 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)

Reemplaza los valores de muestra de la siguiente manera:

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

Administra agentes de datos y conversaciones

En los siguientes ejemplos de código, se muestra cómo administrar tus agentes de datos y conversaciones con la API de Conversational Analytics. Puedes realizar las siguientes tareas:

Obtén un agente de datos
Enumera los agentes de datos
Actualiza un agente de datos
Cómo establecer la política de IAM para un agente de datos
Obtén la política de IAM para un agente de datos
Cómo borrar un agente de datos
Obtener una conversación
Enumera conversaciones
Cómo enumerar los mensajes de una conversación

Obtén un agente de datos

En el siguiente código de muestra, se muestra cómo recuperar un agente de datos existente 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, reemplaza data_agent_1 por el ID del agente de datos que deseas recuperar.

Enumera los 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 GET HTTP al extremo dataAgents.

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

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

Actualiza 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 valores nuevos para los campos que deseas cambiar, y los parámetros de la solicitud incluyen un parámetro updateMask, que especifica los campos que se actualizarán.

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)

Reemplaza los valores de muestra de la siguiente manera:

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

Configura la política de IAM para un agente de datos

Para compartir un agente, puedes usar el método setIamPolicy para asignar roles de IAM a los usuarios en un agente específico. En el siguiente código de ejemplo, se muestra cómo realizar una llamada a POST a la URL del agente de datos con una carga útil que incluye vinculaciones. La vinculación 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)

Reemplaza los valores de muestra de la siguiente manera:

YOUR-BILLING-PROJECT: Es el ID de tu proyecto de facturación.
data_agent_1: Es el ID del agente de datos para el que deseas establecer la política de IAM.
222larabrown@gmail.com, cloudysanfrancisco@gmail.com: Es una lista separada por comas de las direcciones de correo electrónico de los usuarios a los que deseas otorgar el rol especificado.

Obtén la política de IAM para un agente de datos

En el siguiente código de muestra, se muestra cómo recuperar 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)

Reemplaza los valores de muestra de la siguiente manera:

YOUR-BILLING-PROJECT: Es el ID de tu proyecto de facturación.
data_agent_1: Es el ID del agente de datos para el que deseas obtener la política de IAM.

Borra un agente de datos

En el siguiente código de ejemplo, se muestra cómo borrar de forma no definitiva un agente de datos enviando una solicitud HTTP DELETE a la URL del recurso del agente de datos. La eliminación temporal significa que el agente se borra, 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)

Reemplaza los valores de muestra de la siguiente manera:

YOUR-BILLING-PROJECT: Es el ID de tu proyecto de facturación.
data_agent_1: Es el ID del agente de datos que deseas borrar.

Obtén una conversación

En el siguiente código de muestra, se muestra cómo recuperar una conversación existente 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)

Reemplaza los valores de muestra de la siguiente manera:

YOUR-BILLING-PROJECT: Es el ID de tu proyecto de facturación.
conversation_1: ID de la conversación que deseas recuperar.

Mostrar lista de conversaciones

En el siguiente código de ejemplo, se muestra cómo enumerar conversaciones para un proyecto determinado enviando una solicitud HTTP GET al extremo conversations.

De forma predeterminada, este método devuelve las conversaciones que creaste. Los administradores (usuarios con el rol de IAM de 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)

Reemplaza YOUR-BILLING-PROJECT por el ID del proyecto de facturación en el que habilitaste las APIs requeridas.

Cómo enumerar los mensajes de una conversación

En el siguiente código de muestra, se muestra cómo enumerar todos los mensajes de una conversación enviando una solicitud HTTP GET al extremo messages de la conversación.

Para enumerar 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)

Reemplaza los valores de muestra de la siguiente manera:

YOUR-BILLING-PROJECT: Es el ID de tu proyecto de facturación.
conversation_1: Es el ID de la conversación para la que deseas enumerar los mensajes.

Usa la API para hacer preguntas

Después de crear un agente de datos y una conversación, puedes hacer preguntas sobre tus datos.

La API de 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 administrar el historial de conversaciones:

Chat con estado: Google Cloud Almacena y administra el historial de conversaciones. El chat con estado es inherentemente de varios turnos, ya que la API retiene el contexto de los mensajes anteriores. Solo debes enviar el mensaje actual en cada turno.
Chat sin estado: Tu aplicación administra el historial de conversaciones. Debes incluir los mensajes anteriores pertinentes con cada mensaje nuevo. Para obtener ejemplos detallados de cómo administrar conversaciones de varios turnos en modo sin estado, consulta Crea una conversación de varios turnos sin estado.

Chat con estado

Envía una solicitud de chat con estado y una referencia de conversación

En el siguiente código de muestra, se demuestra cómo hacerle preguntas a la API usando la conversación que definiste en los pasos anteriores. En este ejemplo, se usa una función de ayuda 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)

Reemplaza los valores de muestra de la siguiente manera:

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

Chat sin estado

Envía una solicitud de chat sin estado con una referencia de agente de datos

En el siguiente ejemplo de código, se muestra cómo hacerle a la API una pregunta sin estado con el agente de datos que definiste en los pasos anteriores. En este ejemplo, se usa una función de ayuda 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)

Reemplaza los valores de muestra de la siguiente manera:

data_agent_1: Es el ID del agente de datos, como se define en el bloque de código de ejemplo en Crea un agente de datos.
Se usó Make a bar graph for the top 5 states by the total number of airports como ejemplo de instrucción.

Envía una solicitud de chat sin estado con contexto intercalado

En el siguiente ejemplo de código, se muestra cómo hacerle a la API una pregunta sin estado usando el contexto intercalado. En este ejemplo, se usa una función auxiliar get_stream para transmitir la respuesta y se usa una fuente de datos de BigQuery como ejemplo.

También puedes habilitar de forma opcional el análisis avanzado con Python si incluyes 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)

Crea una conversación sin estado de varios turnos

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

Envía solicitudes de varios turnos
Procesa respuestas

Envía solicitudes de varios turnos

La siguiente función auxiliar multi_turn_Conversation administra el contexto de la conversación almacenando los mensajes en una lista. Esto te permite enviar preguntas de seguimiento que se basan 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 con el contexto intercalado.

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, reemplaza data_agent_1 por el ID del agente de datos, como se define en el bloque de código de muestra en Crea un agente de datos.

Puedes llamar a la función auxiliar multi_turn_Conversation para cada turno de la conversación. En el siguiente código de ejemplo, se muestra cómo enviar una solicitud inicial y, luego, 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, reemplaza los valores de muestra de la siguiente manera:

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

Procesa respuestas

La siguiente función get_stream_multi_turn procesa la respuesta de la API de transmisión. 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 = ''

Muestra de código de extremo a extremo

El siguiente muestra de código expandible contiene todas las tareas que se abordan en esta guía.

Crea 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 muestra de código expandible 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 = ''

Compila un agente de datos con HTTP y Python Organiza tus páginas con colecciones Guarda y categoriza el contenido según tus preferencias.

Configura los parámetros iniciales y la autenticación

Autenticación en Looker

Cómo usar claves de API

Cómo usar tokens de acceso

Conectándose a una fuente de datos

Conéctate a los datos de Looker

Conéctate a los datos de BigQuery

Conéctate a los datos de Looker Studio

Crea un agente de datos

Cómo crear una conversación

Administra agentes de datos y conversaciones

Obtén un agente de datos

Enumera los agentes de datos

Actualiza un agente de datos

Configura la política de IAM para un agente de datos

Obtén la política de IAM para un agente de datos

Borra un agente de datos

Obtén una conversación

Mostrar lista de conversaciones

Cómo enumerar los mensajes de una conversación

Usa la API para hacer preguntas

Chat con estado

Envía una solicitud de chat con estado y una referencia de conversación

Chat sin estado

Envía una solicitud de chat sin estado con una referencia de agente de datos

Envía una solicitud de chat sin estado con contexto intercalado

Crea una conversación sin estado de varios turnos

Envía solicitudes de varios turnos

Procesa respuestas

Muestra de código de extremo a extremo

Crea un agente de datos con HTTP y Python

Funciones auxiliares de Python para transmitir respuestas de chat

Compila un agente de datos con HTTP y Python