Google utiliza tecnología de IA para traducir contenido a tu idioma preferido. Las traducciones realizadas con IA pueden contener errores.

Crea 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 Python de ejemplo de esta página, se muestra cómo completar las siguientes tareas:

Establece la configuración y la autenticación iniciales.
Conéctate a una fuente de datos de Looker, BigQuery, Data Studio, AlloyDB para PostgreSQL, Cloud SQL o Spanner.
Crea un agente de datos.
Crea una conversación.
Administra agentes de datos y conversaciones.
Usa la API para hacer preguntas.
Crea una conversación sin estado de varios turnos.
Define funciones auxiliares.

Establece la configuración inicial y la autenticación

El siguiente código de Python de muestra 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, la ubicación y las instrucciones del sistema.
Establece el extremo de API según la variable location.

import json
import json as json_lib
import textwrap

import altair as alt
import google.auth
from google.auth.transport.requests import Request
import IPython
from IPython.display import display, HTML
import pandas as pd
from pygments import highlight, lexers, formatters
import requests

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",
        "x-server-timeout": "300", # Custom timeout up to 600s
}

billing_project = "BILLING_PROJECT"
location = "LOCATION"
system_instruction = "SYSTEM_INSTRUCTIONS"

# Set the base URL based on location.
if not location or location == "global":
  base_url = "https://geminidataanalytics.googleapis.com"
elif "-" in location:
  # Regional endpoints
  base_url = f"https://geminidataanalytics-{location}.googleapis.com"
else:
  # Multi-regional endpoints
  base_url = f"https://geminidataanalytics.{location}.rep.googleapis.com"

Reemplaza los valores de muestra de la siguiente manera:

BILLING_PROJECT: Es el ID del proyecto de Google Cloud en el que habilitaste las APIs requeridas.
LOCATION: Es la ubicación en la que se almacenan tus recursos. Para usar el extremo global, especifica global. Para cumplir con los requisitos de residencia de datos, especifica una sola región (como us-east4) o multirregión (como eu o us) compatible. Para obtener más información, consulta Residencia de datos.

Importante: La ubicación en la ruta de acceso de la solicitud debe coincidir con la ubicación del extremo al que llamas. Por ejemplo, si llamas al extremo geminidataanalytics.us.rep.googleapis.com, especifica us en la ruta de acceso de la solicitud (como /locations/us).
SYSTEM_INSTRUCTIONS: Son instrucciones del sistema para guiar el comportamiento del agente y personalizarlo según tus necesidades de datos. Por ejemplo, puedes usar las instrucciones del sistema para definir términos comerciales, controlar la longitud de las respuestas o establecer el formato de los datos. Lo ideal es definir las instrucciones del sistema con el formato YAML recomendado en Escribe instrucciones del sistema eficaces para proporcionar orientación detallada y estructurada.

Autentícate 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 la siguiente muestra de código de Python, se indica cómo autenticar tu agente en una instancia de Looker con claves de API.

looker_credentials = {
    "oauth": {
        "secret": {
            "client_id": "CLIENT_ID",
            "client_secret": "CLIENT_SECRET",
        }
    }
}

Reemplaza los valores de muestra de la siguiente manera:

CLIENT_ID: Es el ID de cliente de la clave de API de Looker que generaste.
CLIENT_SECRET: Es el secreto del cliente de la clave de API de Looker que generaste.

Usa tokens de acceso

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

looker_credentials = {
    "oauth": {
        "token": {
            "access_token": "ACCESS_TOKEN",
        }
    }
}

Reemplaza los valores de muestra de la siguiente manera:

ACCESS_TOKEN: Es el valor access_token que generas para autenticarte en Looker.

Autentícate en AlloyDB para PostgreSQL

Si planeas conectarte a una fuente de datos de AlloyDB, usa gcloud CLI para autenticarte en la instancia de AlloyDB.

Habilita la autenticación de Identity and Access Management (IAM) en AlloyDB habilitando la marca alloydb.iam_authentication en la instancia de AlloyDB.
Asigna roles de IAM. La principal (cuenta de usuario o de servicio) que llama a la API de Conversational Analytics necesita los permisos de IAM adecuados en tu proyecto deGoogle Cloud . Por lo general, esto incluye roles como alloydb.databaseUser.
Crea un usuario de base de datos correspondiente. Crea un usuario de AlloyDB que refleje la principal de IAM con la dirección de correo electrónico de la cuenta de usuario o de servicio como el nombre de usuario.

Para obtener más información, consulta Autenticación de la base de datos de IAM.

Autentícate en Cloud SQL para MySQL y Cloud SQL para PostgreSQL

Si planeas conectarte a una fuente de datos de Cloud SQL para MySQL o Cloud SQL para PostgreSQL, debes autenticarte en la instancia de Cloud SQL.

Habilita la autenticación de la base de datos de IAM. Asegúrate de que la marca cloudsql.iam_authentication esté configurada como on en tu instancia de Cloud SQL para MySQL. Puedes agregar esto cuando creas una instancia o cuando parcheas una existente.
```
gcloud sql instances patch INSTANCE_NAME --database-flags
cloudsql.iam_authentication=on
```
Otorga los roles de IAM necesarios.

La principal (cuenta de usuario o de servicio) que intenta conectarse necesita los permisos de IAM cloudsql.instances.login y cloudsql.instances.connect en el proyecto. Para otorgar estos roles y permisos, realiza una de las siguientes acciones:
1. Otorga el rol roles/cloudsql.instanceUser. Esto incluye permisos para acceder a la instancia (cloudsql.instances.login) y conectarse con el proxy (cloudsql.instances.connect).
2. Como alternativa, otorga un rol personalizado que contenga, al menos, cloudsql.instances.login y cloudsql.instances.connect.
Agrega la principal de IAM como usuario de la base de datos. Crea un usuario en la instancia de Cloud SQL que corresponda a la dirección de correo electrónico del principal de IAM.

Para una cuenta de servicio, ejecuta el siguiente comando:
```
gcloud sql users create SERVICE_ACCOUNT_EMAIL --instance=INSTANCE_NAME
--type=cloud_iam_service_account
```
Para una cuenta de usuario, ejecuta el siguiente comando:
```
gcloud sql users create USER_EMAIL --instance=INSTANCE_NAME
--type=cloud_iam_user
```
Otorgar privilegios de base de datos La conexión con IAM autentica al usuario en la base de datos, pero no otorga permisos de forma automática en la base de datos. Debes usar comandos estándar de PostgreSQL, como GRANT, para otorgar permisos a este nuevo usuario de la base de datos en objetos específicos, como tablas y esquemas.

Conéctate como un superusuario, como el usuario postgres predeterminado, y ejecuta los siguientes comandos:

-- Example: Connect using psql as the 'postgres' user

GRANT SELECT ON ALL TABLES IN SCHEMA public TO
"user-or-service-account@example.com";

Conéctate con la autenticación de IAM.

Por lo general, los clientes se conectan a través del proxy de autenticación de Cloud SQL o una biblioteca de conectores, que controla el intercambio de tokens. El nombre de usuario de la base de datos es el correo electrónico de la principal de IAM. No se usan contraseñas. En su lugar, se usa un token de OAuth 2.0, que suele controlar de forma automática el proxy de autenticación de Cloud SQL o Google Cloud CLI.

En el siguiente ejemplo, se usan psql y gcloud CLI para generar un token:
```
PGPASSWORD=$(gcloud sql generate-login-token) psql --host=HOST_IP \
--username=IAM_PRINCIPAL_EMAIL --dbname=DATABASE_NAME\
```
Para obtener más información, consulta Autenticación de IAM.

Reemplaza los valores de muestra de la siguiente manera:
- INSTANCE_NAME: Es el ID de la instancia de Cloud SQL que deseas parchar para habilitar la autenticación de la base de datos de IAM.
- SERVICE_ACCOUNT_EMAIL: La dirección de correo electrónico de la cuenta de servicio.
- USER_EMAIL: Es la dirección de correo electrónico de la cuenta de usuario.
- HOST_IP: Es la dirección IP de la instancia de Cloud SQL.
- IAM_PRINCIPAL_EMAIL: Es la dirección de correo electrónico de la principal de IAM, por ejemplo, la cuenta de servicio o el usuario.
- DATABASE_NAME: Es el nombre de la base de datos a la que te conectarás.

Autentícate en Spanner

Si planeas conectarte a una fuente de datos de Spanner, debes autenticarte en la instancia de Spanner.

Asigna roles de IAM. La principal (cuenta de usuario o de servicio) que llama a la API de Conversational Analytics necesita los permisos de IAM adecuados en tu proyecto deGoogle Cloud . Por lo general, esto incluye roles como databaseUser.
Crea roles de base de datos abstractos correspondientes en Spanner con el lenguaje de definición de datos (DDL).
Usa políticas de IAM para permitir que las principales de IAM asuman esos roles de base de datos.

Para obtener más información, consulta Privilegios de control de acceso detallado.

Conectándose 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 los datos de las siguientes maneras:

Looker
Data Studio
BigQuery
Datos de AlloyDB
Datos de Cloud SQL para MySQL
Datos de Cloud SQL para PostgreSQL
Datos de Spanner

Cómo conectarse a los datos de Looker

En el siguiente código de muestra, se define una conexión a una exploración de Looker. Para establecer una conexión con una instancia de Looker, verifica que hayas generado claves de API de Looker, como se describe en Autentica y conéctate a una fuente de datos. Puedes conectarte a un máximo de cinco exploraciones de Looker a la vez con la API de Conversational Analytics.

looker_instance_uri = "INSTANCE_URI"
looker_data_source = {
    "looker": {
        "explore_references": [
            {
                "looker_instance_uri": looker_instance_uri,
                "lookml_model": "MODEL",
                "explore": "EXPLORE",
            },
            {
                "looker_instance_uri": looker_instance_uri,
                "lookml_model": "MODEL_2",
                "explore": "EXPLORE_2",
            },
            # Add up to 5 total Explore references
        ],
    }
}

Reemplaza los valores de muestra de la siguiente manera:

INSTANCE_URI: Es la URL completa de tu instancia de Looker (por ejemplo, https://mycompany.looker.com).
MODEL: Es el nombre del modelo de LookML que incluye la exploración a la que deseas conectarte.
EXPLORE: Es el nombre de la exploración de Looker que deseas que consulte el agente de datos.
MODEL_2: Es el nombre del segundo modelo de LookML que incluye la exploración a la que deseas conectarte. Puedes repetir esta variable para modelos adicionales de hasta cinco exploraciones.
EXPLORE_2: Es el nombre de la exploración de Looker adicional que quieres que consulte el agente de datos. Puedes repetir esta variable para incluir hasta cinco exploraciones.

Para obtener más información sobre cómo conectarte a las exploraciones de Looker, incluidos los roles requeridos, consulta Cómo conectarse a Looker.

Conéctate a los datos de BigQuery

En las siguientes secciones, se muestra cómo conectarse a los datos de BigQuery con referencias de tablas, referencias de grafos o ambas. Para obtener más información sobre cómo conectarse a tablas y grafos, incluidos los roles necesarios, consulta Conéctate a BigQuery.

Conéctate a tablas de BigQuery

En el siguiente código de muestra, se define una conexión a varias tablas de BigQuery y se incluyen ejemplos de campos de contexto estructurados opcionales.

bigquery_data_sources = {
    "bq": {
        "tableReferences": [
            {
                "projectId": "PROJECT_ID",
                "datasetId": "DATASET_ID",
                "tableId": "TABLE_ID",
                "schema": {
                    "description": "TABLE_DESCRIPTION",
                    "fields": [{
                        "name": "COLUMN_NAME",
                        "description": "COLUMN_DESCRIPTION"
                    }]
                }
            },
            {
                "projectId": "PROJECT_ID_2",
                "datasetId": "DATASET_ID_2",
                "tableId": "TABLE_ID_2"
            },
            {
                "projectId": "PROJECT_ID_3",
                "datasetId": "DATASET_ID_3",
                "tableId": "TABLE_ID_3"
            },
        ]
    }
}

Reemplaza los valores de muestra de la siguiente manera:

PROJECT_ID: Es el ID del proyecto de 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.
DATASET_ID: el ID del conjunto de datos de BigQuery.
TABLE_ID: el ID de la tabla de BigQuery.
TABLE_DESCRIPTION: Es una descripción opcional del contenido y el propósito de la tabla.
COLUMN_NAME: Es el nombre de una columna de la tabla para la que proporcionas una descripción opcional.
COLUMN_DESCRIPTION: Es una descripción opcional del contenido y el propósito de la columna.

Conéctate a BigQuery Graph

También puedes conectarte a grafos con BigQuery Graph. Proporciona una o más referencias de gráficos en el campo propertyGraphReferences.

Para obtener información sobre los requisitos de la edición de BigQuery y las limitaciones de las consultas de grafos, consulta Conéctate a BigQuery Graph.

bigquery_data_sources = {
    "bq": {
        "propertyGraphReferences": [
          {
            "projectId": "GRAPH_PROJECT_ID",
            "datasetId": "GRAPH_DATASET_ID",
            "propertyGraphId": "GRAPH_ID"
          }
        ]
    }
}

Reemplaza los valores de muestra de la siguiente manera:

GRAPH_PROJECT_ID: Es el ID del proyecto de Google Cloud que contiene el grafo de BigQuery.
GRAPH_DATASET_ID: Es el ID del conjunto de datos de BigQuery que contiene el grafo.
GRAPH_ID: Es el ID del grafo de BigQuery.

Para consultar tablas y grafos con el mismo agente de datos, incluye tableReferences y propertyGraphReferences en el objeto bq.

Conéctate a los datos de Data Studio

En el siguiente código de muestra, se define una conexión a una fuente de datos de Data 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.

Conéctate a los datos de AlloyDB

La API de Conversational Analytics usa un campo database_reference en el campo alloydb para conectarse a tu clúster de AlloyDB.

En el siguiente ejemplo, se define una conexión a una base de datos de AlloyDB.

alloydb_data_source = {
    "alloydb": {
        "database_reference":
          {
            "project_id":"PROJECT_ID",
            "region":"REGION",
            "cluster_id":"CLUSTER_ID",
            "instance_id":"INSTANCE_ID",
            "database_id":"DATABASE",
            "table_ids":["TABLE_1_ID", "TABLE_2_ID"]
          }
          # Optional: Include this if you have pre-authored context for the agent
          # "agent_context_reference": {
          #     "context_set_id": f"projects/billing_project/locations/location/contextSets/your_context_set_id"
          # }

    }
}

Reemplaza los valores de muestra de la siguiente manera:

PROJECT_ID: Es el ID del proyecto de Google Cloud que contiene el clúster de AlloyDB.
REGION: Es la región en la que reside el clúster de AlloyDB, por ejemplo, us-central1.
CLUSTER_ID: Es el ID del clúster de AlloyDB.
INSTANCE_ID: Es el ID de la instancia de AlloyDB.
DATABASE: Es el nombre de la base de datos de destino, por ejemplo, financial.
TABLE_1_ID, TABLE_2_ID: Opcional. Es una lista de las tablas de la base de datos que se sugiere que use el agente de datos, por ejemplo, "loan", "client" y "disp".
billing_project: Es el ID del proyecto de facturación en el que habilitaste las APIs requeridas.
location: Es la ubicación en la que reside tu conjunto de contexto, por ejemplo, global.
your_context_set_id: Si creaste un contexto avanzado para el agente, el ID de tu conjunto de contextos.

Conéctate a los datos de Cloud SQL para MySQL y Cloud SQL para PostgreSQL

La API de Conversational Analytics usa un campo database_reference en el campo sql para conectarse a tu instancia de Cloud SQL.

En el siguiente ejemplo, se define una conexión a una base de datos de Cloud SQL para MySQL o Cloud SQL para PostgreSQL.

sql_data_source = {
    "cloud_sql_reference": {
        "database_reference":
          {
            "project_id":"PROJECT_ID",
            "region":"REGION",
            "cluster_id":"CLUSTER_ID",
            "instance_id":"INSTANCE_ID",
            "database_id":"DATABASE_ID",
            "table_ids":["TABLE_1_ID", "TABLE_2_ID"]
          }
          # Optional: Include this if you have pre-authored context for the agent
          # "agent_context_reference": {
          #     "context_set_id": f"projects/billing_project/locations/location/contextSets/your_context_set_id"
          # }
    }
}

Reemplaza los valores de muestra de la siguiente manera:

PROJECT_ID: Es el ID del proyecto de Google Cloud que contiene la instancia de Cloud SQL.
REGION: Es la región en la que reside la instancia de Cloud SQL, por ejemplo, us-central1.
CLUSTER_ID: Es el ID del clúster de Cloud SQL.
INSTANCE_ID: Es el ID de la instancia de Cloud SQL.
DATABASE_ID: Es el nombre de la base de datos de destino, por ejemplo, financial.
TABLE_1_ID, TABLE_2_ID: Opcional. Es una lista de las tablas de la base de datos que se sugiere que use el agente de datos, por ejemplo, "loan", "client" y "disp".
billing_project: Es el ID del proyecto de facturación en el que habilitaste las APIs requeridas.
location: Es la ubicación en la que reside tu conjunto de contexto, por ejemplo, global.
your_context_set_id: Si creaste un contexto avanzado para el agente, el ID de tu conjunto de contextos.

Conéctate a los datos de Spanner

La API de Conversational Analytics usa un campo database_reference en el campo spanner para conectarse a tu instancia de Spanner.

En el siguiente ejemplo, se define una conexión a una base de datos de Spanner.

spanner_data_sources = {
    "spanner_reference": {
        "database_reference": {
            "project_id": "PROJECT_ID",
            "region": location,
            "engine": GOOGLE_SQL,
            "instance_id": "INSTANCE_ID",
            "database_id": "DATABASE",
            "table_ids":["TABLE_1_ID", "TABLE_2_ID"]
        },
        # Optional: Include this if you have pre-authored context for the agent
        # "agent_context_reference": {
        #     "context_set_id": f"projects/billing_project/locations/location/contextSets/your_context_set_id"
        # }
    }
}

Reemplaza los valores de muestra de la siguiente manera:

PROJECT_ID: Es el ID del proyecto de Google Cloud que contiene la instancia de Spanner.
INSTANCE_ID: Es el ID de la instancia de Spanner.
DATABASE: Es el nombre de la base de datos de destino, por ejemplo, financial.
TABLE_1_ID, TABLE_2_ID: Es una lista de las tablas de la base de datos que se sugiere que use el agente de datos, por ejemplo, "loan", "client", "disp".
billing_project: Es el ID del proyecto de facturación en el que habilitaste las APIs requeridas.
location: Es la ubicación en la que reside tu conjunto de contextos, por ejemplo, global.
your_context_set_id: Si creaste un contexto avanzado para el agente, el ID de tu conjunto de contexto.

Crear un agente de datos

En el siguiente código de muestra, se indica cómo crear el agente de datos de forma síncrona o asíncrona enviando una solicitud HTTP POST al extremo de creación del agente de datos. La carga útil de la solicitud incluye los siguientes detalles:

Es el nombre completo del recurso para el 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 Establece la configuración inicial y la autenticación) y la fuente de datos que usa el agente (definida en Conéctate a una fuente de datos).

De manera opcional, puedes proteger el agente de datos con claves de encriptación administradas por el cliente (CMEK) incluyendo el campo kms_key en la carga útil de la solicitud cuando crees el agente de datos. Para obtener más información, consulta Claves de encriptación administradas por el cliente (CMEK).

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. Consulta el recurso de REST: projects.locations.dataAgents para obtener más información sobre el parámetro options y las opciones que puedes configurar para la conversación.

Síncrona

data_agent_url = f"https://geminidataanalytics.googleapis.com/v1beta/projects/{billing_project}/locations/{location}/dataAgents:createSync"

data_agent_id = "DATA_AGENT_ID"

# Optional: If using CMEK, replace the empty strings with your KMS key details.
kms_project_id = ""  # KMS_PROJECT_ID (Defaults to billing_project if empty)
key_ring_name = ""   # KEY_RING_NAME
key_name = ""        # KEY_NAME
key_location = ""    # KEY_LOCATION (Defaults to location if empty; except eu)

data_agent_payload = {
      "name": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}", # Optional
      "description": "This is the description of the data agent.", # Optional
      "data_analytics_agent": {
          "published_context": {
              "datasource_references": bigquery_data_sources,
              "system_instruction": system_instruction,
              "options": {
                  # Optional: To enable advanced analysis with Python
                  "analysis": {
                      "python": {
                          "enabled": True
                      }
                  },
                  # Optional: To limit query costs to 100MiB
                  "datasource": {
                      "bigQueryMaxBilledBytes": "104857600"
                  }
              }
          }
      }
  }

# If key details are provided, construct and add the kms_key field.
if key_ring_name and key_name:
  if not kms_project_id:
    kms_project_id = billing_project
  if not key_location:
    key_location = location
  data_agent_payload["kms_key"] = f"projects/{kms_project_id}/locations/{key_location}/keyRings/{key_ring_name}/cryptoKeys/{key_name}"

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)

Asíncrono

data_agent_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/dataAgents"

data_agent_id = "DATA_AGENT_ID"

# Optional: If using CMEK, replace the empty strings with your KMS key details.
kms_project_id = ""  # KMS_PROJECT_ID (Defaults to billing_project if empty)
key_ring_name = ""   # KEY_RING_NAME
key_name = ""        # KEY_NAME
key_location = ""    # KEY_LOCATION (Defaults to location if empty; except eu)

data_agent_payload = {
      "name": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}", # Optional
      "description": "This is the description of the data agent.", # Optional
      "data_analytics_agent": {
          "published_context": {
              "datasource_references": bigquery_data_sources,
              "system_instruction": system_instruction,
              "options": {
                  # Optional: To enable advanced analysis with Python
                  "analysis": {
                      "python": {
                          "enabled": True
                      }
                  },
                  # Optional: To limit query costs to 100MiB
                  "datasource": {
                      "bigQueryMaxBilledBytes": "104857600"
                  }
              }
          }
      }
  }

# If key details are provided, construct and add the kms_key field.
if key_ring_name and key_name:
  if not kms_project_id:
    kms_project_id = billing_project
  if not key_location:
    key_location = location
  data_agent_payload["kms_key"] = f"projects/{kms_project_id}/locations/{key_location}/keyRings/{key_ring_name}/cryptoKeys/{key_name}"

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_ID: 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 consulta de URL data_agent_id.
KMS_PROJECT_ID: Si usas CMEK, el ID del proyecto en el que se aloja la clave. Si no se proporciona, se usa de forma predeterminada tu proyecto de facturación.
KEY_RING_NAME: Si usas CMEK, el nombre de tu llavero de claves de Cloud KMS.
KEY_NAME: Si usas CMEK, el nombre de tu clave de Cloud KMS.
KEY_LOCATION: Si usas CMEK, es la ubicación de tu llavero de claves. Para la multirregión eu, usa europe. Para todas las demás ubicaciones, usa la misma ubicación que el agente de datos. Si no se proporciona, se establece de forma predeterminada en la ubicación del agente de datos, que funciona para todas las ubicaciones, excepto la multirregión eu. Para obtener más información, consulta Requisitos de ubicación de claves.
This is the description of the data agent.: Es una descripción del agente de datos.

Cómo crear una conversación

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

De manera opcional, puedes proteger la conversación con CMEK incluyendo el campo kms_key en la carga útil de la solicitud cuando crees la conversación. Para obtener más información, consulta Claves de encriptación administradas por el cliente (CMEK).

conversation_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/conversations"

data_agent_id = "DATA_AGENT_ID"
conversation_id = "CONVERSATION_ID"

# Optional: If using CMEK, replace the empty strings with your KMS key details.
kms_project_id = ""  # KMS_PROJECT_ID (Defaults to billing_project if empty)
key_ring_name = ""   # KEY_RING_NAME
key_name = ""        # KEY_NAME
key_location = ""    # KEY_LOCATION (Defaults to location if empty; single regions only)

conversation_payload = {
    "agents": [
        f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"
    ],
    "name": f"projects/{billing_project}/locations/{location}/conversations/{conversation_id}",
}

# If key details are provided, construct and add the kms_key field.
if key_ring_name and key_name:
  if not kms_project_id:
    kms_project_id = billing_project
  if not key_location:
    key_location = location
  conversation_payload["kms_key"] = f"projects/{kms_project_id}/locations/{key_location}/keyRings/{key_ring_name}/cryptoKeys/{key_name}"

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_ID: Es el ID del agente de datos, como se define en el bloque de código de muestra en Crea un agente de datos.
CONVERSATION_ID: Es un identificador único para la conversación.
KMS_PROJECT_ID: Si usas CMEK, el ID del proyecto en el que se aloja la clave. Si no se proporciona, se usa de forma predeterminada tu proyecto de facturación.
KEY_RING_NAME: Si usas CMEK, el nombre de tu llavero de claves de Cloud KMS.
KEY_NAME: Si usas CMEK, el nombre de tu clave de Cloud KMS.
KEY_LOCATION: Si usas CMEK, es la ubicación de tu llavero de claves. Para la multirregión us, usa us-central1. Para la multirregión eu, usa europe-west1. Para una sola región, usa la misma región que la conversación. Si no se proporciona, se establece de forma predeterminada en la ubicación de la conversación, que solo funciona para regiones individuales. Para obtener más información, consulta Requisitos de ubicación de claves.

Administra agentes de datos y conversaciones

En las siguientes muestras de código, se indica 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
Enumera los agentes de datos accesibles
Actualiza un agente de datos
Obtén la política de IAM para un agente de datos
Configura la política de IAM para un agente de datos
Borra un agente de datos
Obtén una conversación
Enumera las conversaciones
Enumera los mensajes de una conversación

Obtener un agente de datos

En el siguiente código de muestra, se indica 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_ID"
data_agent_url = f"{base_url}/v1/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_ID 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 HTTP GET 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 = "BILLING_PROJECT"
data_agent_url = f"{base_url}/v1/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}")

Reemplaza BILLING_PROJECT por el ID de tu proyecto de facturación.

Enumera los agentes de datos accesibles

En el siguiente código, se muestra cómo enumerar todos los agentes de datos accesibles para un proyecto determinado enviando una solicitud HTTP GET al extremo dataAgents:listAccessible.

billing_project = "BILLING_PROJECT"
creator_filter = "CREATOR_FILTER"
data_agent_url = f"{base_url}/v1/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}")

Reemplaza los valores de muestra de la siguiente manera:

BILLING_PROJECT: Es el ID de tu proyecto de facturación.
CREATOR_FILTER: Es el filtro que se aplicará según el creador del agente de datos. Los valores posibles incluyen NONE (predeterminado), CREATOR_ONLY y NOT_CREATOR_ONLY.

Actualiza un agente de datos

En el siguiente código de muestra, se indica cómo actualizar un agente de datos de forma síncrona o asíncrona 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.

Síncrona

data_agent_id = "DATA_AGENT_ID"
billing_project = "BILLING_PROJECT"

data_agent_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}:updateSync"

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)

Asíncrono

data_agent_id = "DATA_AGENT_ID"
billing_project = "BILLING_PROJECT"

data_agent_url = f"{base_url}/v1/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_ID: Es el ID del agente de datos que quieres actualizar.
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.

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

En el siguiente código de muestra, se indica cómo recuperar la política de IAM para 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 de acceso del agente de datos.

billing_project = "BILLING_PROJECT"
data_agent_id = "DATA_AGENT_ID"

data_agent_url = f"{base_url}/v1/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:

BILLING_PROJECT: Es el ID de tu proyecto de facturación.
DATA_AGENT_ID: Es el ID del agente de datos para el que deseas obtener la política de IAM.

Establece 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 muestra, se indica cómo realizar una llamada 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 = "BILLING_PROJECT"
data_agent_id = "DATA_AGENT_ID"
role = "roles/geminidataanalytics.dataAgentEditor"
users = "222larabrown@gmail.com, cloudysanfrancisco@gmail.com"

data_agent_url = f"{base_url}/v1/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:

BILLING_PROJECT: Es el ID de tu proyecto de facturación.
DATA_AGENT_ID: 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 los correos electrónicos de los usuarios a los que deseas otorgar el rol especificado.

Borra un agente de datos

En el siguiente código de muestra, se indica cómo borrar un agente de datos de forma síncrona o asíncrona enviando una solicitud HTTP DELETE a la URL del recurso del agente de datos. Cuando borras un agente de datos, este entra en un período de eliminación no definitiva de 30 días en el que no se puede recuperar. Para obtener más información, consulta Retención y eliminación de datos.

Síncrona

billing_project = "BILLING_PROJECT"
data_agent_id = "DATA_AGENT_ID"

data_agent_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}:deleteSync"

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)

Asíncrono

billing_project = "BILLING_PROJECT"
data_agent_id = "DATA_AGENT_ID"

data_agent_url = f"{base_url}/v1/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:

BILLING_PROJECT: Es el ID de tu proyecto de facturación.
DATA_AGENT_ID: Es el ID del agente de datos que deseas borrar.

Obtén una conversación

En el siguiente código de muestra, se indica cómo recuperar una conversación existente enviando una solicitud HTTP GET a la URL del recurso de conversación.

billing_project = "BILLING_PROJECT"
conversation_id = "CONVERSATION_ID"

conversation_url = f"{base_url}/v1/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:

BILLING_PROJECT: Es el ID de tu proyecto de facturación.
CONVERSATION_ID: Es el ID de la conversación que deseas recuperar.

Mostrar lista de conversaciones

En el siguiente código de muestra, se indica cómo enumerar las conversaciones de 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 cloudaicompanion.topicAdmin) pueden ver todas las conversaciones del proyecto.

billing_project = "BILLING_PROJECT"
conversation_url = f"{base_url}/v1/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 BILLING_PROJECT por el ID del proyecto de facturación en el que habilitaste las APIs requeridas.

Enumera los mensajes de una conversación

En el siguiente código de muestra, se indica 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 = "BILLING_PROJECT"

conversation_id = "CONVERSATION_ID"

conversation_url = f"{base_url}/v1/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:

BILLING_PROJECT: Es el ID de tu proyecto de facturación.
CONVERSATION_ID: Es el ID de la conversación para la que deseas enumerar los mensajes.

Borrar una conversación

En el siguiente código de muestra, se indica cómo borrar una conversación enviando una solicitud HTTP DELETE a la URL del recurso de conversación. Los administradores (usuarios con el rol de Identity and Access Management cloudaicompanion.topicAdmin) o los usuarios con el permiso de Identity and Access Management cloudaicompanion.topics.delete pueden borrar conversaciones dentro del proyecto.

billing_project = "BILLING_PROJECT"
conversation_id = "CONVERSATION_ID"

conversation_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}/conversations/{conversation_id}"

conversation_response = requests.delete(conversation_url, headers=headers)

# Handle the response
if conversation_response.status_code == 200:
    print("Conversation deleted successfully!")
    print(json.dumps(conversation_response.json(), indent=2))
else:
    print(f"Error while deleting conversation: {conversation_response.status_code}")
    print(conversation_response.text)

Reemplaza los valores de muestra de la siguiente manera:

BILLING_PROJECT: Es el ID de tu proyecto de facturación.
CONVERSATION_ID: Es el ID de la conversación que deseas borrar.

Usa la API para hacer preguntas

Después de crear un agente de datos y, para el chat con estado, una conversación, puedes enviar consultas al agente.

Cuando envías una consulta, la API devuelve un flujo de objetos Message. Este flujo puede contener diferentes tipos de mensajes, como texto, tablas de datos y gráficos. Los mensajes de texto pueden proporcionar información sobre el razonamiento del agente, informar sobre su progreso o dar la respuesta final. El propósito de cada mensaje de texto se indica con su valor TextType:

THOUGHT: Muestra el proceso de pensamiento interno del agente mientras planifica cómo responder tu consulta. Los mensajes de THOUGHT proporcionan información paso a paso sobre el proceso de razonamiento y toma de decisiones del agente, y contienen dos partes: parts[0] es el resumen del pensamiento, que resume brevemente el texto completo del pensamiento, y parts[1] es el texto completo del pensamiento.
PROGRESS: Informa el progreso del agente en una acción, como la recuperación de datos o una herramienta que se invoca. Este valor se devuelve solo para las fuentes de datos de Looker y contiene dos partes: parts[0] es el resumen y parts[1] es el texto completo del progreso.
FINAL_RESPONSE: Proporciona la respuesta final a tu consulta.

En los ejemplos de la siguiente sección, se usan las funciones auxiliares que se definen en Funciones auxiliares de Python para transmitir respuestas de chat para procesar y mostrar cada mensaje en la respuesta transmitida de la API. Para obtener orientación sobre cómo renderizar estos mensajes en una interfaz de usuario cuando usas fuentes de datos de Looker, consulta Renderiza respuestas de agentes para fuentes de datos de Looker.

Haz preguntas con el chat con estado y sin estado

Puedes interactuar con la API de las siguientes maneras principales:

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 conserva el contexto de los mensajes anteriores. Solo necesitas enviar el mensaje actual para cada turno.
Chat sin estado: Tu aplicación administra el historial de conversaciones. Debes incluir los mensajes anteriores pertinentes con cada mensaje nuevo. Para ver ejemplos detallados de cómo administrar conversaciones de varios turnos en modo sin estado, consulta Crea una conversación de varios turnos sin estado.

En los siguientes ejemplos, se muestra cómo usar la API para el chat con estado y sin estado realizando una solicitud POST al extremo :chat.

Precaución: Las muestras de código de esta sección usan la biblioteca estándar de Python requests para demostrar la transmisión HTTP. Los clientes HTTP o REST estándar no pueden leer los finales de HTTP/2. Si se produce un error fatal (como un tiempo de espera) durante la transmisión, la API podría devolver un mensaje de error genérico, como 503 Connection reset by peer o 500 Internal Server Error. Para las aplicaciones de producción, te recomendamos que uses el SDK de Python, que usa gRPC para analizar correctamente estos errores finales.

Chat con estado

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

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

chat_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}:chat"

data_agent_id = "DATA_AGENT_ID"
conversation_id = "CONVERSATION_ID"

# Construct the payload
chat_payload = {
    "parent": f"projects/{billing_project}/locations/{location}",
    # "credentials": looker_credentials,
    "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}",
        }
    }
}

# 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_ID: Es el ID del agente de datos, como se define en el bloque de código de muestra en Crea un agente de datos.
CONVERSATION_ID: Es un identificador único para la conversación.
Se usó Make a bar graph for the top 5 states by the total number of airports como instrucción de ejemplo.

Chat sin estado

En los siguientes ejemplos, se muestra cómo enviar solicitudes sin estado con una referencia de agente de datos o un contexto intercalado.

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

En el siguiente código de muestra, se indica cómo hacerle una pregunta sin estado a la API con el agente de datos que definiste en los pasos anteriores. En este ejemplo, se usa una función auxiliar get_stream para transmitir la respuesta.

chat_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}:chat"

data_agent_id = "DATA_AGENT_ID"

# Construct the payload
chat_payload = {
    "parent": f"projects/{billing_project}/locations/{location}",
    # "credentials": looker_credentials,
    "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}",
    }
}

# 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_ID: Es el ID del agente de datos, como se define en el bloque de código de muestra en Crea un agente de datos.
Se usó Make a bar graph for the top 5 states by the total number of airports como instrucción de ejemplo.

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

En el siguiente código de muestra, se indica cómo hacer una pregunta sin estado a la API con 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 el análisis avanzado con Python de forma opcional incluyendo el parámetro options en la carga útil de la solicitud. Consulta la página Recurso de REST: projects.locations.dataAgents para obtener más información sobre el parámetro options y las opciones que puedes configurar para la conversación.

chat_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}:chat"

# Construct the payload
chat_payload = {
    "parent": f"projects/{billing_project}/locations/{location}",
    # "credentials": looker_credentials,
    "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)

En el siguiente código de muestra, se indica cómo hacer una pregunta sin estado a la API con un contexto intercalado y una fuente de datos de AlloyDB como ejemplo.

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

# The natural language question to ask the data agent
user_prompt = "What is the average loan amount in the US?"  # Replace with your question

# Construct the payload
chat_payload = {
    "parent": f"projects/{billing_project}/locations/{location}",
    # "credentials": looker_credentials,
    "messages": [
        {
            "userMessage": {
                "text": user_prompt
            }
        }
    ],
    "data_agent_context": {
          "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
    }
}

print(f"Sending prompt to :chat: '{user_prompt}'")
print(f"Endpoint: {chat_url}")
print(f"Payload: {json.dumps(chat_payload, indent=2)}")

get_stream(chat_url, chat_payload)

Crea una conversación de varios turnos sin estado

Para hacer preguntas adicionales 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. Las conversaciones de varios turnos te permiten enviar preguntas de seguimiento que se basan en turnos de conversación anteriores.

En el siguiente ejemplo, se define y se llama a una función multi_turn_conversation para administrar el contexto de la conversación almacenando los mensajes en una lista. La función multi_turn_conversation llama a la función get_stream_multi_turn para analizar la respuesta de la API de transmisión y actualizar la lista conversation_messages. Este proceso conserva el contexto para los turnos posteriores. Para obtener el código de get_stream_multi_turn y otras funciones auxiliares obligatorias, consulta Funciones auxiliares de chat de transmisión.

chat_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}:chat"

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

data_agent_id = "DATA_AGENT_ID"

# 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/{location}",
      # "credentials": looker_credentials,
      "messages": conversation_messages,
      # Use a data agent reference
      "data_agent_context": {
          "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
      },
      # 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_ID 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 muestra, se indica cómo enviar una solicitud inicial y, luego, una solicitud de seguimiento que se base 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 define mejor.

Define funciones auxiliares

En la siguiente sección, se incluyen definiciones de funciones auxiliares que se usan en las muestras de código anteriores para analizar las respuestas de la API y mostrar los resultados.

Funciones auxiliares de chat de transmisión

Las siguientes funciones analizan y muestran la respuesta de transmisión de la API para las solicitudes de chat estándar (de un solo turno).

Mostrar funciones auxiliares 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']
      full_text = "".join(parts)
      if "\n" not in full_text and len(full_text) > 80:
        wrapped_text = textwrap.fill(full_text, width=80)
        print(wrapped_text)
      else:
        print(full_text)

    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']))
        if 'question' in query:
          print('Question: {}'.format(query['question']))
        if 'datasources' in query:
          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 = ''

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

Establece la configuración inicial y la autenticación

Autentícate en Looker

Cómo usar claves de API

Usa tokens de acceso

Autentícate en AlloyDB para PostgreSQL

Autentícate en Cloud SQL para MySQL y Cloud SQL para PostgreSQL

Autentícate en Spanner

Conectándose a una fuente de datos

Cómo conectarse a los datos de Looker

Conéctate a los datos de BigQuery

Conéctate a tablas de BigQuery

Conéctate a BigQuery Graph

Conéctate a los datos de Data Studio

Conéctate a los datos de AlloyDB

Conéctate a los datos de Cloud SQL para MySQL y Cloud SQL para PostgreSQL

Conéctate a los datos de Spanner

Crear un agente de datos

Síncrona

Asíncrono

Cómo crear una conversación

Administra agentes de datos y conversaciones

Obtener un agente de datos

Enumera los agentes de datos

Enumera los agentes de datos accesibles

Actualiza un agente de datos

Síncrona

Asíncrono

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

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

Borra un agente de datos

Síncrona

Asíncrono

Obtén una conversación

Mostrar lista de conversaciones

Enumera los mensajes de una conversación

Borrar una conversación

Usa la API para hacer preguntas

Haz preguntas con el chat con estado y sin estado

Chat con estado

Envía una solicitud de chat con estado con 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 de varios turnos sin estado

Define funciones auxiliares

Funciones auxiliares de chat de transmisión

Mostrar funciones auxiliares para transmitir respuestas de chat

Crea un agente de datos con HTTP y Python