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 ou Looker Studio
Créer un agent de données
Créer une conversation
Gérer les agents de données et les conversations
Utiliser l'API pour poser des questions
Créer une conversation multitour sans état

La version complète de l'exemple de code est disponible à la fin de la page, ainsi que les fonctions d'assistance utilisées pour diffuser la réponse de l'API.

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 et les instructions système

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

from google.colab import auth
auth.authenticate_user()

access_token = !gcloud auth application-default print-access-token
headers = {
    "Authorization": f"Bearer {access_token[0]}",
    "Content-Type": "application/json",
}

billing_project = 'YOUR-BILLING-PROJECT'
system_instruction = 'YOUR-SYSTEM-INSTRUCTIONS'

Remplacez les exemples de valeurs comme suit :

YOUR-BILLING-PROJECT : ID du projet de facturation dans lequel vous avez activé les API requises.
YOUR-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": "YOUR-LOOKER-CLIENT-ID",
            "client_secret": "YOUR-LOOKER-CLIENT-SECRET",
        }
    }
}

Remplacez les exemples de valeurs comme suit :

YOUR-LOOKER-CLIENT-ID : ID client de la clé API Looker générée.
YOUR-LOOKER-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": "YOUR-TOKEN",
        }
    }
}

Remplacez les exemples de valeurs comme suit :

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

Se connecter à une source de données

Les exemples de code Python suivants montrent comment définir la source de données Looker, BigQuery ou Looker Studio que votre agent utilisera.

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 sur la page S'authentifier et se connecter à une source de données avec l'API Conversational Analytics.

looker_data_source = {
    "looker": {
        "explore_references": {
            "looker_instance_uri": "https://your_company.looker.com",
            "lookml_model": "your_model",
            "explore": "your_explore",
       },
    }
}

Remplacez les exemples de valeurs comme suit :

https://your_company.looker.com : URL complète de votre instance Looker.
your_model : nom du modèle LookML qui inclut l'exploration à laquelle vous souhaitez vous connecter.
your_explore : nom de l'exploration Looker que l'agent de données doit interroger.

Se connecter aux données BigQuery

L'API Conversational Analytics vous permet de vous connecter à un maximum de 10 tables BigQuery à la fois pour les interroger.

L'exemple de code suivant définit une connexion à une table BigQuery.

bigquery_data_sources = {
    "bq": {
        "tableReferences": [
            {
                "projectId": "bigquery-public-data",
                "datasetId": "san_francisco",
                "tableId": "street_trees",
            }
        ]
    }
}

Remplacez les exemples de valeurs comme suit :

bigquery-public-data : 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.
san_francisco : ID de l'ensemble de données BigQuery.
street_trees : ID de la table BigQuery.

Se connecter aux données Looker Studio

L'exemple de code suivant définit une connexion à une source de données Looker 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.

Créer un agent de données

L'exemple de code suivant montre comment créer l'agent de données 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 également activer l'analyse avancée avec Python en incluant le paramètre options dans la charge utile de la requête.

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

data_agent_id = "data_agent_1"

data_agent_payload = {
      "name": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}", # Optional
      "description": "This is the description of data_agent_1.", # Optional

      "data_analytics_agent": {
          "published_context": {
              "datasource_references": bigquery_data_sources,
              "system_instruction": system_instruction,
              # Optional: To enable advanced analysis with Python, include the following options block:
              "options": {
                  "analysis": {
                      "python": {
                          "enabled": True
                      }
                  }
              }
          }
      }
  }

params = {"data_agent_id": data_agent_id} # Optional

data_agent_response = requests.post(
    data_agent_url, params=params, json=data_agent_payload, headers=headers
)

if data_agent_response.status_code == 200:
    print("Data Agent created successfully!")
    print(json.dumps(data_agent_response.json(), indent=2))
else:
    print(f"Error creating Data Agent: {data_agent_response.status_code}")
    print(data_agent_response.text)

Remplacez les exemples de valeurs comme suit :

data_agent_1 : 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.
This is the description of data_agent_1. : 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.

conversation_url = f"https://geminidataanalytics.googleapis.com/v1alpha/projects/{billing_project}/locations/{location}/conversations"

data_agent_id = "data_agent_1"
conversation_id = "conversation_1"

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

conversation_response = requests.post(conversation_url, headers=headers, params=params, json=conversation_payload)

if conversation_response.status_code == 200:
    print("Conversation created successfully!")
    print(json.dumps(conversation_response.json(), indent=2))
else:
    print(f"Error creating Conversation: {conversation_response.status_code}")
    print(conversation_response.text)

Remplacez les exemples de valeurs comme suit :

data_agent_1 : 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_1 : identifiant unique de la conversation.

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
Mettre à jour un agent de données
Définir la stratégie IAM pour un agent de données
Obtenir la stratégie IAM d'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_1"
data_agent_url = f"{base_url}/v1alpha/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_1 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 = "YOUR-BILLING-PROJECT"
location = "global"
data_agent_url = f"{base_url}/v1alpha/projects/{billing_project}/locations/{location}/dataAgents"

data_agent_response = requests.get(
    data_agent_url, headers=headers
)

if data_agent_response.status_code == 200:
    print("Data Agent Listed successfully!")
    print(json.dumps(data_agent_response.json(), indent=2))
else:
    print(f"Error Listing Data Agent: {data_agent_response.status_code}")

Remplacez YOUR-BILLING-PROJECT par l'ID de votre projet de facturation.

Mettre à jour un agent de données

L'exemple de code suivant montre comment mettre à jour un agent de données 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.

data_agent_id = "data_agent_1"
billing_project = "YOUR-BILLING-PROJECT"
location = "global"

data_agent_url = f"{base_url}/v1alpha/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_1 : ID de l'agent de données que vous souhaitez mettre à jour.
YOUR-BILLING-PROJECT : ID de votre projet de facturation.
Updated description of the data agent. : nouvelle description de l'agent de données.

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 = "YOUR-BILLING-PROJECT"
location = "global"
data_agent_id = "data_agent_1"
role = "roles/geminidataanalytics.dataAgentEditor"
users = "222larabrown@gmail.com, cloudysanfrancisco@gmail.com"

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

YOUR-BILLING-PROJECT : ID de votre projet de facturation.
data_agent_1 : 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.

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 = "YOUR-BILLING-PROJECT"
location = "global"
data_agent_id = "data_agent_1"

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

YOUR-BILLING-PROJECT : ID de votre projet de facturation.
data_agent_1 : ID de l'agent de données pour lequel vous souhaitez obtenir la stratégie IAM.

Supprimer un agent de données

L'exemple de code suivant montre comment supprimer un agent de données de façon réversible 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.

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
data_agent_id = "data_agent_1"

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

YOUR-BILLING-PROJECT : ID de votre projet de facturation.
data_agent_1 : 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 = "YOUR-BILLING-PROJECT"
location = "global"
conversation_id = "conversation_1"

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

YOUR-BILLING-PROJECT : ID de votre projet de facturation.
conversation_1 : 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 = "YOUR-BILLING-PROJECT"
location = "global"
conversation_url = f"{base_url}/v1alpha/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 YOUR-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 = "YOUR-BILLING-PROJECT"
location = "global"

conversation_id = "conversation_1"

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

YOUR-BILLING-PROJECT : ID de votre projet de facturation.
conversation_1 : ID de la conversation pour laquelle vous souhaitez lister les messages.

Utiliser l'API pour poser des questions

Une fois que vous avez créé un agent de données et une conversation, vous pouvez interroger vos données.

L'API Conversational Analytics est compatible avec les conversations multitours, qui permettent aux utilisateurs de poser des questions complémentaires en s'appuyant sur le contexte précédent. L'API fournit les méthodes suivantes pour gérer l'historique de la conversation :

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.

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"https://geminidataanalytics.googleapis.com/v1alpha/projects/{billing_project}/locations/{location}:chat"

data_agent_id = "data_agent_1"
conversation_id = "conversation_1"

# Construct the payload
chat_payload = {
    "parent": f"projects/{billing_project}/locations/global",
    "messages": [
        {
            "userMessage": {
                "text": "Make a bar graph for the top 5 states by the total number of airports"
            }
        }
    ],
    "conversation_reference": {
        "conversation": f"projects/{billing_project}/locations/{location}/conversations/{conversation_id}",
        "data_agent_context": {
            "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
            # "credentials": looker_credentials
        }
    }
}

# Call the get_stream function to stream the response
get_stream(chat_url, chat_payload)

Remplacez les exemples de valeurs comme suit :

data_agent_1 : 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_1 : 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

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"https://geminidataanalytics.googleapis.com/v1alpha/projects/{billing_project}/locations/{location}:chat"

data_agent_id = "data_agent_1"

# Construct the payload
chat_payload = {
    "parent": f"projects/{billing_project}/locations/global",
    "messages": [
        {
            "userMessage": {
                "text": "Make a bar graph for the top 5 states by the total number of airports"
            }
        }
    ],
    "data_agent_context": {
        "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
        # "credentials": looker_credentials
    }
}

# Call the get_stream function to stream the response
get_stream(chat_url, chat_payload)

Remplacez les exemples de valeurs comme suit :

data_agent_1 : 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.

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

# Construct the payload
chat_payload = {
    "parent": f"projects/{billing_project}/locations/global",
    "messages": [
        {
            "userMessage": {
                "text": "Make a bar graph for the top 5 states by the total number of airports"
            }
        }
    ],
    "inline_context": {
        "datasource_references": bigquery_data_sources,
          # Optional: To enable advanced analysis with Python, include the following options block:
          "options": {
              "analysis": {
                  "python": {
                      "enabled": True
                  }
              }
          }
    }
}

# Call the get_stream function to stream the response
get_stream(chat_url, chat_payload)

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 sections suivantes expliquent comment définir et appeler des fonctions d'assistance pour créer une conversation multitour :

Envoyer des requêtes multitours
Traiter les réponses

Envoyer des requêtes multitours

La fonction d'assistance multi_turn_Conversation suivante gère le contexte de la conversation en stockant les messages dans une liste. Vous pouvez ainsi poser des questions complémentaires qui s'appuient sur les tours précédents. Dans la charge utile de la fonction, vous pouvez faire référence à un agent de données ou fournir directement la source de données en utilisant le contexte intégré.

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

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

data_agent_id = "data_agent_1"

# Helper function for calling the API
def multi_turn_Conversation(msg):

  userMessage = {
      "userMessage": {
          "text": msg
      }
  }

  # Send a multi-turn request by including previous turns and the new message
  conversation_messages.append(userMessage)

  # Construct the payload
  chat_payload = {
      "parent": f"projects/{billing_project}/locations/global",
      "messages": conversation_messages,
      # Use a data agent reference
      "data_agent_context": {
          "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
          # "credentials": looker_credentials
      },
      # Use inline context
      # "inline_context": {
      #     "datasource_references": bigquery_data_sources,
      # }
  }

  # Call the get_stream_multi_turn helper function to stream the response
  get_stream_multi_turn(chat_url, chat_payload, conversation_messages)

Dans l'exemple précédent, remplacez data_agent_1 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.

Traiter les réponses

La fonction get_stream_multi_turn suivante traite la réponse de l'API qui est diffusée. Cette fonction est semblable à la fonction d'assistance get_stream, mais elle stocke la réponse dans la liste conversation_messages afin d'enregistrer le contexte de la conversation pour le prochain tour.

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

Exemple de code de bout en bout

L'exemple de code suivant peut être développé et contient toutes les tâches abordées dans ce guide.

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

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

    from google.colab import auth
    auth.authenticate_user()

    access_token = !gcloud auth application-default print-access-token
    headers = {
        "Authorization": f"Bearer {access_token[0]}",
        "Content-Type": "application/json",
    }

    ################### Data source details ###################

    billing_project = "your_billing_project"
    location = "global"
    system_instruction = "Help the user in analyzing their data"


    # BigQuery data source
    bigquery_data_sources = {
        "bq": {
        "tableReferences": [
            {
            "projectId": "bigquery-public-data",
            "datasetId": "san_francisco",
            "tableId": "street_trees"
            }
        ]
        }
    }

    # Looker data source
    looker_credentials = {
        "oauth": {
            "secret": {
            "client_id": "your_looker_client_id",
            "client_secret": "your_looker_client_secret",
            }
        }
    }
    # # To use access_token for authentication, uncomment the following looker_credentials code block and comment out the previous looker_credentials code block.
    # looker_credentials = {
    #     "oauth": {
    #         "token": {
    #           "access_token": "your_looker_access_token",
    #         }
    #     }
    # }
    looker_data_source = {
        "looker": {
        "explore_references": {
            "looker_instance_uri": "https://my_company.looker.com",
            "lookml_model": "my_model",
            "explore": "my_explore",
        },
        # "credentials": looker_credentials
    }

    # Looker Studio data source
    looker_studio_data_source = {
        "studio":{
            "studio_references":
            [
                {
                "datasource_id": "your_studio_datasource_id"
                }
            ]
        }
    }

    ################### Create data agent ###################
    data_agent_url = f"https://geminidataanalytics.googleapis.com/v1alpha/projects/{billing_project}/locations/{location}/dataAgents"

    data_agent_id = "data_agent_1"

    data_agent_payload = {
        "name": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}", # Optional
        "description": "This is the description of data_agent.", # Optional

        "data_analytics_agent": {
            "published_context": {
                "datasource_references": bigquery_data_sources,
                "system_instruction": system_instruction,
                # Optional: To enable advanced analysis with Python, include the following options block:
                "options": {
                    "analysis": {
                        "python": {
                            "enabled": True
                        }
                    }
                }
            }
        }
    }

    params = {"data_agent_id": data_agent_id} # Optional

    data_agent_response = requests.post(
        data_agent_url, params=params, json=data_agent_payload, headers=headers
    )

    if data_agent_response.status_code == 200:
        print("Data Agent created successfully!")
        print(json.dumps(data_agent_response.json(), indent=2))
    else:
        print(f"Error creating Data Agent: {data_agent_response.status_code}")
        print(data_agent_response.text)


    ################### Create conversation ###################

    conversation_url = f"https://geminidataanalytics.googleapis.com/v1alpha/projects/{billing_project}/locations/{location}/conversations"

    data_agent_id = "data_agent_1"
    conversation_id = "conversation _1"

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

    conversation_response = requests.post(conversation_url, headers=headers, params=params, json=conversation_payload)

    if conversation_response.status_code == 200:
        print("Conversation created successfully!")
        print(json.dumps(conversation_response.json(), indent=2))
    else:
        print(f"Error creating Conversation: {conversation_response.status_code}")
        print(conversation_response.text)


    ################### Chat with the API by using conversation (stateful) ####################

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

    data_agent_id = "data_agent_1"
    conversation_id = "conversation _1"

    # Construct the payload
    chat_payload = {
        "parent": f"projects/{billing_project}/locations/global",
        "messages": [
            {
                "userMessage": {
                    "text": "Make a bar graph for the top 5 states by the total number of airports"
                }
            }
        ],
        "conversation_reference": {
            "conversation": f"projects/{billing_project}/locations/{location}/conversations/{conversation_id}",
            "data_agent_context": {
                "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
                # "credentials": looker_credentials
            }
        }
    }

    # Call the get_stream function to stream the response
    get_stream(chat_url, chat_payload)

    ################### Chat with the API by using dataAgents (stateless) ####################

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

    data_agent_id = "data_agent_1"

    # Construct the payload
    chat_payload = {
        "parent": f"projects/{billing_project}/locations/global",
        "messages": [
            {
                "userMessage": {
                    "text": "Make a bar graph for the top 5 states by the total number of airports"
                }
            }
        ],
        "data_agent_context": {
            "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
            # "credentials": looker_credentials
        }
    }

    # Call the get_stream function to stream the response
    get_stream(chat_url, chat_payload)

    ################### Chat with the API by using inline context (stateless) ####################

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

    # Construct the payload
    chat_payload = {
        "parent": f"projects/{billing_project}/locations/global",
        "messages": [
            {
                "userMessage": {
                    "text": "Make a bar graph for the top 5 states by the total number of airports"
                }
            }
        ],
        "inline_context": {
            "datasource_references": bigquery_data_sources,
            # Optional - if wanting to use advanced analysis with python
            "options": {
                "analysis": {
                    "python": {
                        "enabled": True
                    }
                }
            }
        }
    }

    # Call the get_stream function to stream the response
    get_stream(chat_url, chat_payload)

    ################### Multi-turn conversation ###################

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

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

    data_agent_id = "data_agent_1"

    # Helper function for calling the API
    def multi_turn_Conversation(msg):

      userMessage = {
          "userMessage": {
              "text": msg
          }
      }

      # Send a multi-turn request by including previous turns and the new message
      conversation_messages.append(userMessage)

      # Construct the payload
      chat_payload = {
          "parent": f"projects/{billing_project}/locations/global",
          "messages": conversation_messages,
          # Use a data agent reference
          "data_agent_context": {
              "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
              # "credentials": looker_credentials
          },
          # Use inline context
          # "inline_context": {
          #     "datasource_references": bigquery_data_sources,
          # }
      }

      # Call the get_stream_multi_turn helper function to stream the response
      get_stream_multi_turn(chat_url, chat_payload, conversation_messages)

    # Send first-turn request
    multi_turn_Conversation("Which species of tree is most prevalent?")

    # Send follow-up-turn request
    multi_turn_Conversation("Can you show me the results as a bar chart?")

L'exemple de code suivant peut être développé et contient les fonctions d'assistance Python utilisées pour diffuser les réponses du chat.

Fonctions Python 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']
      print(''.join(parts))

    def get_property(data, field_name, default = ''):
      return data[field_name] if field_name in data else default

    def display_schema(data):
      fields = data['fields']
      df = pd.DataFrame({
        "Column": map(lambda field: get_property(field, 'name'), fields),
        "Type": map(lambda field: get_property(field, 'type'), fields),
        "Description": map(lambda field: get_property(field, 'description', '-'), fields),
        "Mode": map(lambda field: get_property(field, 'mode'), fields)
      })
      display(df)

    def display_section_title(text):
      display(HTML('<h2>{}</h2>'.format(text)))

    def format_bq_table_ref(table_ref):
      return '{}.{}.{}'.format(table_ref['projectId'], table_ref['datasetId'], table_ref['tableId'])

    def format_looker_table_ref(table_ref):
      return 'lookmlModel: {}, explore: {}, lookerInstanceUri: {}'.format(table_ref['lookmlModel'], table_ref['explore'], table_ref['lookerInstanceUri'])

    def display_datasource(datasource):
      source_name = ''

      if 'studioDatasourceId' in datasource:
        source_name = datasource['studioDatasourceId']
      elif 'lookerExploreReference' in datasource:
        source_name = format_looker_table_ref(datasource['lookerExploreReference'])
      else:
        source_name = format_bq_table_ref(datasource['bigqueryTableReference'])

      print(source_name)
      display_schema(datasource['schema'])

    def handle_schema_response(resp):
      if 'query' in resp:
        print(resp['query']['question'])
      elif 'result' in resp:
        display_section_title('Schema resolved')
        print('Data sources:')
        for datasource in resp['result']['datasources']:
          display_datasource(datasource)

    def handle_data_response(resp):
      if 'query' in resp:
        query = resp['query']
        display_section_title('Retrieval query')
        print('Query name: {}'.format(query['name']))
        print('Question: {}'.format(query['question']))
        print('Data sources:')
        for datasource in query['datasources']:
          display_datasource(datasource)
      elif 'generatedSql' in resp:
        display_section_title('SQL generated')
        print(resp['generatedSql'])
      elif 'result' in resp:
        display_section_title('Data retrieved')

        fields = map(lambda field: get_property(field, 'name'), resp['result']['schema']['fields'])
        dict = {}

        for field in fields:
          dict[field] = map(lambda el: get_property(el, field), resp['result']['data'])

        display(pd.DataFrame(dict))

    def handle_chart_response(resp):
      if 'query' in resp:
        print(resp['query']['instructions'])
      elif 'result' in resp:
        vegaConfig = resp['result']['vegaConfig']
        alt.Chart.from_json(json_lib.dumps(vegaConfig)).display();

    def handle_error(resp):
      display_section_title('Error')
      print('Code: {}'.format(resp['code']))
      print('Message: {}'.format(resp['message']))

    def get_stream(url, json):
      s = requests.Session()

      acc = ''

      with s.post(url, json=json, headers=headers, stream=True) as resp:
        for line in resp.iter_lines():
          if not line:
            continue

          decoded_line = str(line, encoding='utf-8')

          if decoded_line == '[{':
            acc = '{'
          elif decoded_line == '}]':
            acc += '}'
          elif decoded_line == ',':
            continue
          else:
            acc += decoded_line

          if not is_json(acc):
            continue

          data_json = json_lib.loads(acc)

          if not 'systemMessage' in data_json:
            if 'error' in data_json:
                handle_error(data_json['error'])
            continue

          if 'text' in data_json['systemMessage']:
            handle_text_response(data_json['systemMessage']['text'])
          elif 'schema' in data_json['systemMessage']:
            handle_schema_response(data_json['systemMessage']['schema'])
          elif 'data' in data_json['systemMessage']:
            handle_data_response(data_json['systemMessage']['data'])
          elif 'chart' in data_json['systemMessage']:
            handle_chart_response(data_json['systemMessage']['chart'])
          else:
            colored_json = highlight(acc, lexers.JsonLexer(), formatters.TerminalFormatter())
            print(colored_json)
            print('\n')
            acc = ''

    def get_stream_multi_turn(url, json, conversation_messages):
        s = requests.Session()

        acc = ''

        with s.post(url, json=json, headers=headers, stream=True) as resp:
            for line in resp.iter_lines():
                if not line:
                    continue

                decoded_line = str(line, encoding='utf-8')

                if decoded_line == '[{':
                    acc = '{'
                elif decoded_line == '}]':
                    acc += '}'
                elif decoded_line == ',':
                    continue
                else:
                    acc += decoded_line

                if not is_json(acc):
                    continue

                data_json = json_lib.loads(acc)
                # Store the response that will be used in the next iteration
                conversation_messages.append(data_json)

                if not 'systemMessage' in data_json:
                    if 'error' in data_json:
                        handle_error(data_json['error'])
                    continue

                if 'text' in data_json['systemMessage']:
                    handle_text_response(data_json['systemMessage']['text'])
                elif 'schema' in data_json['systemMessage']:
                    handle_schema_response(data_json['systemMessage']['schema'])
                elif 'data' in data_json['systemMessage']:
                    handle_data_response(data_json['systemMessage']['data'])
                elif 'chart' in data_json['systemMessage']:
                    handle_chart_response(data_json['systemMessage']['chart'])
                else:
                    colored_json = highlight(acc, lexers.JsonLexer(), formatters.TerminalFormatter())
                    print(colored_json)
                print('\n')
                acc = ''

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

Se connecter à une source de données

Se connecter aux données Looker

Se connecter aux données BigQuery

Se connecter aux données Looker Studio

Créer un agent de données

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

Mettre à jour un agent de données

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

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

Supprimer un agent de données

Obtenir une conversation

Lister les conversations

Lister les messages d'une conversation

Utiliser l'API pour poser des questions

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

Envoyer des requêtes multitours

Traiter les réponses

Exemple de code de bout en bout

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

Fonctions Python d'assistance pour diffuser les réponses du chat

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