Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

Créer un agent de données à l'aide de HTTP et Python

Cette page explique comment utiliser Python pour envoyer des requêtes HTTP à l'API Conversational Analytics (accessible via geminidataanalytics.googleapis.com).

L'exemple de code Python sur cette page montre comment effectuer les tâches suivantes :

Configurer les paramètres initiaux et l'authentification
Se connecter à une source de données Looker, BigQuery, Data Studio, AlloyDB pour PostgreSQL, Cloud SQL ou Spanner
Créez un agent de données.
Créez une conversation.
Gérer les agents de données et les conversations
Utilisez l'API pour poser des questions.
Créer une conversation multitour sans état
Définissez des fonctions d'assistance.

Configurer les paramètres initiaux et l'authentification

L'exemple de code Python suivant effectue les tâches ci-dessous :

Importe les bibliothèques Python requises.
Obtient un jeton d'accès pour l'authentification HTTP à l'aide de Google Cloud CLI.
Définit les variables pour le projet de facturation, l'emplacement et les instructions système.
Définit le point de terminaison de l'API en fonction de 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"

Remplacez les exemples de valeurs comme suit :

BILLING_PROJECT : ID du Google Cloud projet dans lequel vous avez activé les API requises.
LOCATION : emplacement de stockage de vos ressources. Pour utiliser le point de terminaison mondial, spécifiez global. Pour répondre aux exigences de résidence des données, spécifiez une région unique (par exemple, us-east4) ou une région multiple (par exemple, eu ou us) acceptée. Pour en savoir plus, consultez Résidence des données.

Important : L'emplacement dans le chemin d'accès de votre requête doit correspondre à celui du point de terminaison que vous appelez. Par exemple, si vous appelez le point de terminaison geminidataanalytics.us.rep.googleapis.com, spécifiez us dans le chemin d'accès de votre requête (par exemple, /locations/us).
SYSTEM_INSTRUCTIONS : instructions système pour guider le comportement de l'agent et le personnaliser en fonction de vos besoins en données. Par exemple, vous pouvez utiliser des instructions système pour définir des termes métier, contrôler la longueur des réponses ou définir le format des données. Dans l'idéal, définissez les instructions système en utilisant le format YAML recommandé dans Rédiger des instructions système efficaces pour fournir des conseils détaillés et structurés.

S'authentifier auprès de Looker

Si vous prévoyez de vous connecter à une source de données Looker, vous devrez vous authentifier auprès de l'instance Looker.

Utiliser des clés API

L'exemple de code Python suivant montre comment authentifier votre agent auprès d'une instance Looker à l'aide de clés API.

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

Remplacez les exemples de valeurs comme suit :

CLIENT_ID : ID client de la clé API Looker générée.
CLIENT_SECRET : code secret du client pour la clé API Looker générée.

Utiliser des jetons d'accès

L'exemple de code Python suivant montre comment authentifier votre agent auprès d'une instance Looker à l'aide de jetons d'accès.

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

Remplacez les exemples de valeurs comme suit :

ACCESS_TOKEN : valeur access_token que vous générez pour vous authentifier auprès de Looker.

S'authentifier auprès d'AlloyDB pour PostgreSQL

Si vous prévoyez de vous connecter à une source de données AlloyDB, utilisez la gcloud CLI pour vous authentifier auprès de l'instance AlloyDB.

Activez l'authentification Identity and Access Management (IAM) sur AlloyDB en activant l'option alloydb.iam_authentication sur l'instance AlloyDB.
Attribuez des rôles IAM. Le compte principal (utilisateur ou compte de service) qui appelle l'API Conversational Analytics doit disposer des autorisations IAM appropriées dans votre projetGoogle Cloud . Cela inclut généralement des rôles tels que alloydb.databaseUser.
Créez un utilisateur de base de données correspondant. Créez un utilisateur AlloyDB qui reflète le compte principal IAM, en utilisant l'adresse e-mail du compte utilisateur ou du compte de service comme nom d'utilisateur.

Pour en savoir plus, consultez Authentification IAM pour les bases de données.

S'authentifier auprès de Cloud SQL pour MySQL et Cloud SQL pour PostgreSQL

Si vous prévoyez de vous connecter à une source de données Cloud SQL pour MySQL ou Cloud SQL pour PostgreSQL, vous devez vous authentifier auprès de l'instance Cloud SQL.

Activez l'authentification IAM pour les bases de données. Assurez-vous que le flag cloudsql.iam_authentication est défini sur on dans votre instance Cloud SQL pour MySQL. Vous pouvez l'ajouter lorsque vous créez une instance ou lorsque vous corrigez une instance existante.
```
gcloud sql instances patch INSTANCE_NAME --database-flags
cloudsql.iam_authentication=on
```
Attribuez les rôles IAM requis.

Le compte principal (utilisateur ou compte de service) qui tente de se connecter doit disposer des autorisations IAM cloudsql.instances.login et cloudsql.instances.connect sur le projet. Pour accorder ces rôles et autorisations, procédez comme suit :
1. Attribuez le rôle roles/cloudsql.instanceUser. Cela inclut les autorisations de se connecter à l'instance (cloudsql.instances.login) et de se connecter à l'aide du proxy (cloudsql.instances.connect).
2. Vous pouvez également attribuer un rôle personnalisé contenant au moins cloudsql.instances.login et cloudsql.instances.connect.
Ajoutez le compte principal IAM en tant qu'utilisateur de la base de données. Créez un utilisateur dans l'instance Cloud SQL qui correspond à l'adresse e-mail du principal IAM.

Pour un compte de service, exécutez la commande suivante :
```
gcloud sql users create SERVICE_ACCOUNT_EMAIL --instance=INSTANCE_NAME
--type=cloud_iam_service_account
```
Pour un compte utilisateur, exécutez la commande suivante :
```
gcloud sql users create USER_EMAIL --instance=INSTANCE_NAME
--type=cloud_iam_user
```
Accordez des droits pour une base de données. La connexion avec IAM authentifie l'utilisateur auprès de la base de données, mais n'accorde pas automatiquement d'autorisations dans la base de données. Vous devez utiliser des commandes PostgreSQL standards, comme GRANT, pour accorder à ce nouvel utilisateur de base de données des autorisations sur des objets spécifiques tels que des tables et des schémas.

Connectez-vous en tant que superutilisateur (comme l'utilisateur postgres par défaut) et exécutez les commandes suivantes :

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

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

Connectez-vous à l'aide de l'authentification IAM.

Les clients se connectent généralement via le proxy d'authentification Cloud SQL ou une bibliothèque de connecteurs, qui gère l'échange de jetons. Le nom d'utilisateur de la base de données correspond à l'adresse e-mail du principal IAM. Les mots de passe ne sont pas utilisés. Un jeton OAuth 2.0, généralement géré automatiquement par le proxy d'authentification Cloud SQL ou par Google Cloud CLI, est utilisé à la place.

L'exemple suivant utilise psql et gcloud CLI pour générer un jeton :
```
PGPASSWORD=$(gcloud sql generate-login-token) psql --host=HOST_IP \
--username=IAM_PRINCIPAL_EMAIL --dbname=DATABASE_NAME\
```
Pour en savoir plus, consultez la page Authentification IAM.

Remplacez les exemples de valeurs comme suit :
- INSTANCE_NAME : ID de l'instance Cloud SQL que vous souhaitez corriger pour activer l'authentification IAM pour les bases de données.
- SERVICE_ACCOUNT_EMAIL : adresse e-mail du compte de service.
- USER_EMAIL : adresse e-mail du compte utilisateur.
- HOST_IP : adresse IP de l'instance Cloud SQL.
- IAM_PRINCIPAL_EMAIL : adresse e-mail du compte principal IAM (compte de service ou utilisateur, par exemple).
- DATABASE_NAME : nom de la base de données à laquelle se connecter.

S'authentifier auprès de Spanner

Si vous prévoyez de vous connecter à une source de données Spanner, vous devez vous authentifier auprès de l'instance Spanner.

Attribuez des rôles IAM. Le compte principal (utilisateur ou compte de service) qui appelle l'API Conversational Analytics doit disposer des autorisations IAM appropriées dans votre projetGoogle Cloud . Cela inclut généralement des rôles tels que databaseUser.
Créez des rôles de base de données abstraits correspondants dans Spanner à l'aide du langage de définition de données (LDD).
Utilisez des stratégies IAM pour permettre aux comptes principaux IAM d'assumer ces rôles de base de données.

Pour en savoir plus, consultez Privilèges de contrôle des accès ultraprécis.

Vous connecter à une source de données

Les sections suivantes expliquent comment définir les détails de connexion pour les sources de données de votre agent. Votre agent peut se connecter aux données de différentes manières :

Looker
Data Studio
BigQuery
Données AlloyDB
Données Cloud SQL pour MySQL
Données Cloud SQL pour PostgreSQL
Données Spanner

Se connecter aux données Looker

L'exemple de code suivant définit une connexion à une exploration Looker. Pour établir une connexion à une instance Looker, vérifiez que vous avez généré des clés API Looker, comme décrit dans S'authentifier et se connecter à une source de données. L'API Conversational Analytics vous permet de vous connecter à un maximum de cinq explorations Looker à la fois.

Lorsque vous vous connectez à une source de données Looker, tenez compte des points suivants :

Vous pouvez interroger n'importe quelle exploration incluse dans une conversation.
Un agent ne peut interroger qu'une seule exploration à la fois. Il n'est pas possible d'effectuer des requêtes dans plusieurs Explorations simultanément.
Un agent peut interroger plusieurs explorations dans la même conversation.
Un agent peut interroger plusieurs explorations dans une conversation qui inclut des questions en plusieurs parties ou des questions de suivi.

Par exemple, un utilisateur associe deux explorations, l'une appelée cat-explore et l'autre dog-explore. L'utilisateur saisit la question "Qu'est-ce qui est le plus grand : le nombre de chats ou le nombre de chiens ?". Cela créerait deux requêtes : une pour compter le nombre de chats dans cat-explore et une pour compter le nombre de chiens dans dog-explore. L'agent compare le nombre des deux requêtes après les avoir exécutées.

L'exemple de code suivant définit une connexion à plusieurs explorations Looker. Pour améliorer les performances de l'agent, vous pouvez éventuellement fournir des requêtes de référence en tant que contexte structuré pour vos explorations. Pour en savoir plus, consultez Définir le contexte de l'agent de données pour les sources de données Looker.

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

Remplacez les exemples de valeurs comme suit :

INSTANCE_URI : URL complète de votre instance Looker (par exemple, https://mycompany.looker.com).
MODEL : nom du modèle LookML qui inclut l'exploration à laquelle vous souhaitez vous connecter.
EXPLORE : nom de l'exploration Looker que l'agent de données doit interroger.
MODEL_2 : nom du deuxième modèle LookML qui inclut l'exploration à laquelle vous souhaitez vous connecter. Vous pouvez répéter cette variable pour d'autres modèles, jusqu'à cinq explorations.
EXPLORE_2 : nom de l'exploration Looker supplémentaire que l'agent de données doit interroger. Vous pouvez répéter cette variable pour inclure jusqu'à cinq explorations.

Se connecter aux données BigQuery

L'API Conversational Analytics ne limite pas le nombre de tables BigQuery auxquelles vous pouvez vous connecter. Toutefois, si vous vous connectez à un grand nombre de tables, la précision peut diminuer ou vous risquez de dépasser la limite de jetons d'entrée du modèle.

Les sections suivantes expliquent comment se connecter aux données BigQuery à l'aide de références de table ou de références de graphique.

Se connecter à des tables BigQuery

L'exemple de code suivant définit une connexion à plusieurs tables BigQuery et inclut des exemples de champs de contexte structurés facultatifs. Pour améliorer les performances de l'agent, vous pouvez éventuellement fournir un contexte structuré pour vos tables BigQuery, comme des descriptions de tables et de colonnes, des synonymes, des tags, des exemples de requêtes et des fonctions définies par l'utilisateur. Pour en savoir plus, consultez Définir le contexte de l'agent de données pour les sources de données BigQuery.

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"
            },
        ]
    }
}

Remplacez les exemples de valeurs comme suit :

PROJECT_ID : ID du projet Google Cloud contenant l'ensemble de données et la table BigQuery auxquels vous souhaitez vous connecter. Pour vous connecter à un ensemble de données public, spécifiez bigquery-public-data.
DATASET_ID : ID de l'ensemble de données BigQuery.
TABLE_ID : ID de la table BigQuery.
TABLE_DESCRIPTION : description facultative du contenu et de l'objectif de la table.
COLUMN_NAME : nom d'une colonne de la table pour laquelle vous fournissez une description facultative.
COLUMN_DESCRIPTION : description facultative du contenu et de l'objectif de la colonne.

Se connecter à BigQuery Graph

Vous pouvez également vous connecter à BigQuery Graph en fournissant une référence de graphique dans le champ propertyGraphReferences. Cette fonctionnalité est disponible en version bêta.

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

Remplacez les exemples de valeurs comme suit :

GRAPH_PROJECT_ID : ID du projet Google Cloud contenant le graphique BigQuery.
GRAPH_DATASET_ID : ID de l'ensemble de données BigQuery contenant le graphique.
GRAPH_ID : ID du graphique BigQuery.

Se connecter aux données Data Studio

L'exemple de code suivant définit une connexion à une source de données Data Studio.

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

Remplacez studio_datasource_id par l'ID de la source de données.

Se connecter aux données AlloyDB

L'API Conversational Analytics utilise un champ database_reference dans le champ alloydb pour se connecter à votre cluster AlloyDB.

L'exemple suivant définit une connexion à une base de données 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"
          # }

    }
}

Remplacez les exemples de valeurs comme suit :

PROJECT_ID : ID du projet Google Cloud contenant le cluster AlloyDB.
REGION : région où réside le cluster AlloyDB, par exemple us-central1.
CLUSTER_ID : ID du cluster AlloyDB.
INSTANCE_ID : ID de l'instance AlloyDB.
DATABASE : nom de la base de données cible, par exemple financial.
TABLE_1_ID, TABLE_2_ID : facultatif. Liste des tables de la base de données que l'agent de données est invité à utiliser (par exemple, "loan", "client", "disp").
billing_project : ID du projet de facturation dans lequel vous avez activé les API requises.
location : emplacement de votre ensemble de contexte (par exemple, global).
your_context_set_id : ID de votre ensemble de contexte si vous avez créé un contexte avancé pour l'agent.

Se connecter aux données Cloud SQL pour MySQL et Cloud SQL pour PostgreSQL

L'API Conversational Analytics utilise un champ database_reference dans le champ sql pour se connecter à votre instance Cloud SQL.

L'exemple suivant définit une connexion à une base de données Cloud SQL pour MySQL ou Cloud SQL pour 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"
          # }
    }
}

Remplacez les exemples de valeurs comme suit :

PROJECT_ID : ID du projet Google Cloud contenant l'instance Cloud SQL.
REGION : région dans laquelle réside l'instance Cloud SQL (par exemple, us-central1).
CLUSTER_ID : ID du cluster Cloud SQL.
INSTANCE_ID : ID de l'instance Cloud SQL.
DATABASE_ID : nom de la base de données cible, par exemple financial.
TABLE_1_ID, TABLE_2_ID : facultatif. Liste des tables de la base de données que l'agent de données est invité à utiliser (par exemple, "loan", "client", "disp").
billing_project : ID du projet de facturation dans lequel vous avez activé les API requises.
location : emplacement de votre ensemble de contexte (par exemple, global).
your_context_set_id : ID de votre ensemble de contexte si vous avez créé un contexte avancé pour l'agent.

Se connecter aux données Spanner

L'API Conversational Analytics utilise un champ database_reference dans le champ spanner pour se connecter à votre instance Spanner.

L'exemple suivant définit une connexion à une base de données 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"
        # }
    }
}

Remplacez les exemples de valeurs comme suit :

PROJECT_ID : ID du projet Google Cloud contenant l'instance Spanner.
INSTANCE_ID : ID de l'instance Spanner.
DATABASE : nom de la base de données cible, par exemple financial.
TABLE_1_ID, TABLE_2_ID : liste des tables de la base de données que l'agent de données est invité à utiliser (par exemple, "loan", "client", "disp").
billing_project : ID du projet de facturation dans lequel vous avez activé les API requises.
location : emplacement de votre ensemble de contexte (par exemple, global).
your_context_set_id : si vous avez créé un contexte avancé pour l'agent, ID de votre ensemble de contexte.

Créer un agent de données

L'exemple de code suivant montre comment créer l'agent de données de manière synchrone ou asynchrone en envoyant une requête HTTP POST au point de terminaison de création de l'agent de données. La charge utile de la requête inclut les informations suivantes :

Nom complet de la ressource pour l'agent : cette valeur inclut l'ID du projet, l'emplacement et un identifiant unique pour l'agent.
Description de l'agent de données.
Contexte de l'agent de données, y compris la description du système (définie dans la section Configurer les paramètres initiaux et l'authentification) et la source de données utilisée par l'agent (définie dans la section Se connecter à une source de données).

Vous pouvez éventuellement protéger l'agent de données à l'aide de clés de chiffrement gérées par le client (CMEK) en fournissant un kms_key lors de la création. Le chiffrement CMEK n'est compatible qu'avec les sources de données Looker. Pour en savoir plus, consultez Clés de chiffrement gérées par le client (CMEK).

Vous pouvez également activer l'analyse avancée avec Python en incluant le paramètre options dans la charge utile de la requête. Consultez la ressource REST projects.locations.dataAgents pour en savoir plus sur le paramètre options et les options que vous pouvez configurer pour la conversation.

Synchrone

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

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
  data_agent_payload["kms_key"] = f"projects/{kms_project_id}/locations/{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)

Asynchrone

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

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
  data_agent_payload["kms_key"] = f"projects/{kms_project_id}/locations/{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)

Remplacez les exemples de valeurs comme suit :

DATA_AGENT_ID : identifiant unique de l'agent de données. Cette valeur est utilisée dans le nom de ressource de l'agent et comme paramètre de requête d'URL data_agent_id.
KMS_PROJECT_ID : Si vous utilisez une clé CMEK, ID du projet dans lequel la clé est hébergée. Si vous ne le fournissez pas, la valeur par défaut est votre projet de facturation.
KEY_RING_NAME : si vous utilisez CMEK, nom de votre trousseau de clés Cloud KMS.
KEY_NAME : si vous utilisez CMEK, nom de votre clé Cloud KMS.
This is the description of the data agent. : description de l'agent de données.

Créer une conversation

L'exemple de code suivant montre comment créer une conversation avec votre agent de données.

Vous pouvez éventuellement protéger la conversation avec une clé CMEK en fournissant un kms_key lors de la création. Le chiffrement CMEK n'est compatible qu'avec les sources de données Looker. Pour en savoir plus, consultez Clés de chiffrement gérées par le client (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

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
  conversation_payload["kms_key"] = f"projects/{kms_project_id}/locations/{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)

Remplacez les exemples de valeurs comme suit :

DATA_AGENT_ID : ID de l'agent de données, tel que défini dans l'exemple de bloc de code de la section Créer un agent de données.
CONVERSATION_ID : identifiant unique de la conversation.
KMS_PROJECT_ID : Si vous utilisez une clé CMEK, ID du projet dans lequel la clé est hébergée. Si vous ne le fournissez pas, la valeur par défaut est votre projet de facturation.
KEY_RING_NAME : si vous utilisez CMEK, nom de votre trousseau de clés Cloud KMS.
KEY_NAME : si vous utilisez CMEK, nom de votre clé Cloud KMS.

Gérer les agents de données et les conversations

Les exemples de code suivants montrent comment gérer vos agents de données et vos conversations à l'aide de l'API Conversational Analytics. Vous pouvez effectuer les tâches suivantes :

Obtenir un agent de données
Lister les agents de données
Lister les agents de données accessibles
Mettre à jour un agent de données
Obtenir la stratégie IAM d'un agent de données
Définir la stratégie IAM pour un agent de données
Supprimer un agent de données
Obtenir une conversation
Lister les conversations
Lister les messages d'une conversation

Obtenir un agent de données

L'exemple de code suivant montre comment récupérer un agent de données existant en envoyant une requête HTTP GET à l'URL de la ressource de l'agent de données.

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)

Dans l'exemple précédent, remplacez DATA_AGENT_ID par l'ID de l'agent de données que vous souhaitez récupérer.

Lister les agents de données

Le code suivant montre comment lister tous les agents de données d'un projet donné en envoyant une requête HTTP GET au point de terminaison dataAgents.

Pour recenser tous les agents, vous devez disposer de l'autorisation geminidataanalytics.dataAgents.list sur le projet. Pour en savoir plus sur les rôles IAM incluant cette autorisation, consultez la liste des rôles prédéfinis.

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

Remplacez BILLING_PROJECT par l'ID de votre projet de facturation.

Lister les agents de données accessibles

Le code suivant montre comment lister tous les agents de données accessibles pour un projet donné en envoyant une requête HTTP GET au point de terminaison 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}")

Remplacez les exemples de valeurs comme suit :

BILLING_PROJECT : ID de votre projet de facturation.
CREATOR_FILTER : filtre à appliquer en fonction du créateur de l'agent de données. Les valeurs possibles sont NONE (par défaut), CREATOR_ONLY et NOT_CREATOR_ONLY.

Mettre à jour un agent de données

L'exemple de code suivant montre comment mettre à jour un agent de données de manière synchrone ou asynchrone en envoyant une requête HTTP PATCH à l'URL de la ressource de l'agent de données. La charge utile de la requête inclut les nouvelles valeurs des champs que vous souhaitez modifier, et les paramètres de requête incluent un paramètre updateMask qui spécifie les champs à mettre à jour.

Synchrone

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)

Asynchrone

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)

Remplacez les exemples de valeurs comme suit :

DATA_AGENT_ID : ID de l'agent de données que vous souhaitez mettre à jour.
BILLING_PROJECT : ID de votre projet de facturation.
Updated description of the data agent. : nouvelle description de l'agent de données.

Obtenir la stratégie IAM d'un agent de données

L'exemple de code suivant montre comment récupérer la stratégie IAM d'un agent de données en envoyant une requête HTTP POST à l'URL de l'agent de données. La charge utile de la requête inclut le chemin d'accès de l'agent de données.

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)

Remplacez les exemples de valeurs comme suit :

BILLING_PROJECT : ID de votre projet de facturation.
DATA_AGENT_ID : ID de l'agent de données pour lequel vous souhaitez obtenir la stratégie IAM.

Définir la stratégie IAM pour un agent de données

Pour partager un agent, vous pouvez utiliser la méthode setIamPolicy afin d'attribuer des rôles IAM aux utilisateurs sur un agent spécifique. L'exemple de code suivant montre comment effectuer un appel POST à l'URL de l'agent de données avec une charge utile incluant des liaisons. La liaison spécifie les rôles à attribuer à chaque utilisateur.

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)

Remplacez les exemples de valeurs comme suit :

BILLING_PROJECT : ID de votre projet de facturation.
DATA_AGENT_ID : ID de l'agent de données pour lequel vous souhaitez définir la stratégie IAM.
222larabrown@gmail.com, cloudysanfrancisco@gmail.com : liste des adresses e-mail des utilisateurs auxquels vous souhaitez attribuer le rôle spécifié, séparées par une virgule.

Supprimer un agent de données

L'exemple de code suivant montre comment supprimer un agent de données de façon réversible de manière synchrone ou asynchrone en envoyant une requête HTTP DELETE à l'URL de la ressource de l'agent de données. Avec la suppression réversible, l'agent est supprimé mais peut être récupéré dans un délai de 30 jours.

Synchrone

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)

Asynchrone

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)

Remplacez les exemples de valeurs comme suit :

BILLING_PROJECT : ID de votre projet de facturation.
DATA_AGENT_ID : ID de l'agent de données que vous souhaitez supprimer.

Obtenir une conversation

L'exemple de code suivant montre comment récupérer une conversation existante en envoyant une requête HTTP GET à l'URL de la ressource de la conversation.

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)

Remplacez les exemples de valeurs comme suit :

BILLING_PROJECT : ID de votre projet de facturation.
CONVERSATION_ID : ID de la conversation que vous souhaitez extraire.

Lister les conversations

L'exemple de code suivant montre comment lister les conversations d'un projet donné en envoyant une requête HTTP GET au point de terminaison conversations.

Par défaut, cette méthode renvoie les conversations que vous avez créées. Les administrateurs (utilisateurs disposant du rôle IAM cloudaicompanion.topicAdmin) peuvent voir toutes les conversations du projet.

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)

Remplacez BILLING_PROJECT par l'ID du projet de facturation dans lequel vous avez activé les API requises.

Lister les messages d'une conversation

L'exemple de code suivant montre comment lister tous les messages d'une conversation en envoyant une requête HTTP GET au point de terminaison messages de la conversation.

Pour lister les messages, vous devez disposer de l'autorisation cloudaicompanion.topics.get sur la conversation.

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)

Remplacez les exemples de valeurs comme suit :

BILLING_PROJECT : ID de votre projet de facturation.
CONVERSATION_ID : ID de la conversation pour laquelle vous souhaitez lister les messages.

Supprimer une conversation

L'exemple de code suivant montre comment supprimer une conversation en envoyant une requête HTTP DELETE à l'URL de la ressource de la conversation. Les administrateurs (utilisateurs disposant du rôle Identity and Access Management cloudaicompanion.topicAdmin) ou les utilisateurs disposant de l'autorisation IAM cloudaicompanion.topics.delete peuvent supprimer les conversations du projet.

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)

Remplacez les exemples de valeurs comme suit :

BILLING_PROJECT : ID de votre projet de facturation.
CONVERSATION_ID : ID de la conversation que vous souhaitez supprimer.

Utiliser l'API pour poser des questions

Une fois que vous avez créé un agent de données et, pour le chat avec état, une conversation, vous pouvez envoyer des requêtes à l'agent.

Lorsque vous envoyez une requête, l'API renvoie un flux d'objets Message. Ce flux peut contenir différents types de messages, y compris du texte, des tableaux de données et des graphiques. Les messages peuvent fournir des informations sur le raisonnement de l'agent, rendre compte de sa progression ou donner la réponse finale. L'objectif de chaque message est indiqué par sa valeur TextType :

THOUGHT : affiche le processus de réflexion interne de l'agent lorsqu'il planifie sa réponse à votre requête. Les messages THOUGHT fournissent des informations détaillées sur le processus de raisonnement et de prise de décision de l'agent. Ils se composent de deux parties : parts[0], qui est le résumé de la pensée, et parts[1], qui est le texte complet de la pensée.
PROGRESS : indique la progression de l'agent sur une action, comme la récupération de données ou un outil en cours d'appel. Cette valeur n'est renvoyée que pour les sources de données Looker et comporte deux parties : parts[0] correspond au résumé et parts[1] au texte complet de la progression.
FINAL_RESPONSE : fournit la réponse finale à votre requête.

Les exemples de la section suivante utilisent les fonctions d'assistance définies dans Fonctions d'assistance Python pour diffuser les réponses du chat afin de traiter et d'afficher chaque message de la réponse diffusée de l'API. Pour savoir comment afficher ces messages dans une interface utilisateur lorsque vous utilisez des sources de données Looker, consultez Afficher les réponses de l'agent pour les sources de données Looker.

Poser des questions avec le chat avec ou sans état

Vous pouvez interagir avec l'API de différentes manières :

Chat avec état : Google Cloud stocke et gère l'historique de la conversation. Le chat avec état est intrinsèquement multitour, car l'API conserve le contexte des messages précédents. Vous n'avez besoin d'envoyer que le message actuel pour chaque tour.
Chat sans état : votre application gère l'historique de la conversation. Vous devez inclure les précédents messages pertinents dans chaque nouveau message. Pour obtenir des exemples détaillés de la gestion des conversations multitours en mode sans état, consultez la section Créer une conversation multitour sans état.

Les exemples suivants montrent comment utiliser l'API pour le chat avec et sans état en envoyant une requête POST au point de terminaison :chat.

Attention : Les exemples de code de cette section utilisent la bibliothèque Python standard requests pour illustrer le streaming HTTP. Les clients HTTP ou REST standards ne peuvent pas lire les trailers HTTP/2. Si une erreur fatale (comme un délai avant expiration) se produit lors du streaming, l'API peut renvoyer un message d'erreur générique, tel que 503 Connection reset by peer ou 500 Internal Server Error. Pour les applications de production, nous vous recommandons vivement d'utiliser le SDK Python, qui utilise gRPC pour analyser correctement ces erreurs de fin.

Chat avec état

Envoyer une requête de chat avec état et référence à une conversation

L'exemple de code suivant montre comment poser des questions à l'API en utilisant la conversation que vous avez définie lors des étapes précédentes. Cet exemple utilise une fonction d'assistance get_stream pour diffuser la réponse.

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)

Remplacez les exemples de valeurs comme suit :

DATA_AGENT_ID : ID de l'agent de données, tel que défini dans l'exemple de bloc de code de la section Créer un agent de données.
CONVERSATION_ID : identifiant unique de la conversation.
La valeur Make a bar graph for the top 5 states by the total number of airports a été utilisée comme exemple de prompt.

Chat sans état

Les exemples suivants montrent comment envoyer des requêtes sans état en utilisant une référence d'agent de données ou un contexte intégré.

Envoyer une requête de chat sans état avec référence à un agent de données

L'exemple de code suivant montre comment poser une question sans état à l'API en utilisant l'agent de données que vous avez défini lors des étapes précédentes. Cet exemple utilise une fonction d'assistance get_stream pour diffuser la réponse.

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)

Remplacez les exemples de valeurs comme suit :

DATA_AGENT_ID : ID de l'agent de données, tel que défini dans l'exemple de bloc de code de la section Créer un agent de données.
La valeur Make a bar graph for the top 5 states by the total number of airports a été utilisée comme exemple de prompt.

Envoyer une requête de chat sans état avec un contexte intégré

L'exemple de code suivant montre comment poser une question sans état à l'API à l'aide du contexte intégré. Cet exemple utilise une fonction d'assistance get_stream pour diffuser la réponse et une source de données BigQuery comme exemple.

Vous pouvez également activer l'analyse avancée avec Python en incluant le paramètre options dans la charge utile de la requête. Pour en savoir plus sur le paramètre options et les options que vous pouvez configurer pour la conversation, consultez la page Ressource REST : projects.locations.dataAgents.

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)

L'exemple de code suivant montre comment poser une question sans état à l'API à l'aide du contexte intégré et d'une source de données AlloyDB comme exemple.

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)

Créer une conversation multitour sans état

Pour qu'il soit possible de poser des questions complémentaires dans une conversation sans état, votre application doit gérer le contexte de la conversation en envoyant l'intégralité de l'historique des messages à chaque nouvelle requête. Les conversations multitours vous permettent de poser des questions complémentaires qui s'appuient sur les tours précédents.

L'exemple suivant définit et appelle une fonction multi_turn_conversation pour gérer le contexte de la conversation en stockant les messages dans une liste. La fonction multi_turn_conversation appelle la fonction get_stream_multi_turn pour analyser la réponse de l'API de streaming et mettre à jour la liste conversation_messages. Ce processus préserve le contexte pour les tours suivants. Pour obtenir le code de get_stream_multi_turn et d'autres fonctions d'assistance requises, consultez Fonctions d'assistance pour le chat en streaming.

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)

Dans l'exemple précédent, remplacez DATA_AGENT_ID par l'ID de l'agent de données, tel que défini dans l'exemple de bloc de code de la section Créer un agent de données.

Vous pouvez appeler la fonction d'assistance multi_turn_conversation pour chaque tour de conversation. L'exemple de code suivant montre comment envoyer une requête initiale, puis une requête complémentaire qui s'appuie sur la réponse précédente :

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

Dans l'exemple précédent, remplacez les exemples de valeurs comme suit :

Which species of tree is most prevalent? : question en langage naturel à envoyer à l'agent de données.
Can you show me the results as a bar chart? : question complémentaire qui s'appuie sur la question précédente ou la précise.

Définir des fonctions d'assistance

La section suivante contient des définitions de fonctions d'assistance utilisées dans les exemples de code précédents pour analyser les réponses de l'API et afficher les résultats.

Fonctions d'assistance pour le streaming du chat

Les fonctions suivantes analysent et affichent la réponse de l'API qui est diffusée pour les requêtes de chat standards (à tour unique).

Afficher les fonctions d'assistance pour diffuser les réponses du 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 = ''

Créer un agent de données à l'aide de HTTP et Python Restez organisé à l'aide des collections Enregistrez et classez les contenus selon vos préférences.

Configurer les paramètres initiaux et l'authentification

S'authentifier auprès de Looker

Utiliser des clés API

Utiliser des jetons d'accès

S'authentifier auprès d'AlloyDB pour PostgreSQL

S'authentifier auprès de Cloud SQL pour MySQL et Cloud SQL pour PostgreSQL

S'authentifier auprès de Spanner

Vous connecter à une source de données

Se connecter aux données Looker

Se connecter aux données BigQuery

Se connecter à des tables BigQuery

Se connecter à BigQuery Graph

Se connecter aux données Data Studio

Se connecter aux données AlloyDB

Se connecter aux données Cloud SQL pour MySQL et Cloud SQL pour PostgreSQL

Se connecter aux données Spanner

Créer un agent de données

Synchrone

Asynchrone

Créer une conversation

Gérer les agents de données et les conversations

Obtenir un agent de données

Lister les agents de données

Lister les agents de données accessibles

Mettre à jour un agent de données

Synchrone

Asynchrone

Obtenir la stratégie IAM d'un agent de données

Définir la stratégie IAM pour un agent de données

Supprimer un agent de données

Synchrone

Asynchrone

Obtenir une conversation

Lister les conversations

Lister les messages d'une conversation

Supprimer une conversation

Utiliser l'API pour poser des questions

Poser des questions avec le chat avec ou sans état

Chat avec état

Envoyer une requête de chat avec état et référence à une conversation

Chat sans état

Envoyer une requête de chat sans état avec référence à un agent de données

Envoyer une requête de chat sans état avec un contexte intégré

Créer une conversation multitour sans état

Définir des fonctions d'assistance

Fonctions d'assistance pour le streaming du chat

Afficher les fonctions d'assistance pour diffuser les réponses du chat

Créer un agent de données à l'aide de HTTP et Python