Créer un agent de données à l'aide du SDK Python

Cette page vous explique comment utiliser le SDK Python pour envoyer des requêtes à l'API Conversational Analytics. L'exemple de code Python montre comment effectuer les tâches suivantes :

S'authentifier et configurer l'environnement

Pour utiliser le SDK Python pour l'API Conversational Analytics, suivez les instructions du notebook Colaboratory du SDK de l'API Conversational Analytics concernant le téléchargement et l'installation du SDK. Notez que la méthode de téléchargement et le contenu du notebook Colab du SDK sont susceptibles d'être modifiés.

Après avoir suivi les instructions de configuration dans le notebook, vous pouvez utiliser le code suivant pour importer les bibliothèques SDK requises, authentifier votre compte Google dans un environnement Colaboratory et initialiser un client pour effectuer des requêtes API :

from google.colab import auth
auth.authenticate_user()

from google.cloud import geminidataanalytics

data_agent_client = geminidataanalytics.DataAgentServiceClient()
data_chat_client = geminidataanalytics.DataChatServiceClient()

Spécifier le projet de facturation et les instructions système

L'exemple de code Python suivant définit le projet de facturation et les instructions système qui sont utilisées tout au long de votre script :

# Billing project
billing_project = "my_project_name"

# System instructions
system_instruction = "Help the user analyze their data."

Remplacez les exemples de valeurs comme suit :

  • my_project_name : ID de votre projet de facturation pour lequel les API requises sont activées.
  • Help the user analyze their data. : 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.

Se connecter à une source de données

Les exemples de code Python suivants montrent comment définir les détails de connexion pour la source de données Looker, BigQuery ou Looker Studio que votre agent interrogera pour répondre aux questions.

Se connecter aux données Looker

Les exemples de code suivants montrent comment définir les détails d'une connexion à une exploration Looker avec des clés API ou un jeton d'accès.

Clés API

Vous pouvez établir une connexion avec une instance Looker à l'aide de clés API Looker générées, comme décrit dans S'authentifier et se connecter à une source de données avec l'API Conversational Analytics.

looker_client_id = "my_looker_client_id"
looker_client_secret = "my_looker_client_secret"
looker_instance_uri = "https://my_company.looker.com"
lookml_model = "my_model"
explore = "my_explore"

looker_explore_reference = geminidataanalytics.LookerExploreReference()
looker_explore_reference.looker_instance_uri = looker_instance_uri
looker_explore_reference.lookml_model = lookml_model
looker_explore_reference.explore = explore

credentials = geminidataanalytics.Credentials()
credentials.oauth.secret.client_id = looker_client_id
credentials.oauth.secret.client_secret = looker_client_secret

datasource_references = geminidataanalytics.DatasourceReferences()
datasource_references.looker.explore_references = [looker_explore_reference]

Remplacez les exemples de valeurs comme suit :

  • my_looker_client_id : ID client de la clé API Looker générée.
  • my_looker_client_secret : code secret du client pour la clé API Looker générée.
  • https://my_company.looker.com : URL complète de votre instance Looker.
  • my_model : nom du modèle LookML qui inclut l'exploration à laquelle vous souhaitez vous connecter.
  • my_explore : nom de l'exploration Looker que l'agent de données doit interroger.

Jeton d'accès

Vous pouvez établir une connexion avec une instance Looker à l'aide d'un jeton d'accès, comme décrit dans S'authentifier et se connecter à une source de données avec l'API Conversational Analytics.

looker_access_token = "my_access_token"
looker_instance_uri = "https://my_company.looker.com"
lookml_model = "my_model"
explore = "my_explore"

looker_explore_reference = geminidataanalytics.LookerExploreReference()
looker_explore_reference.looker_instance_uri = looker_instance_uri
looker_explore_reference.lookml_model = lookml_model
looker_explore_reference.explore = explore

credentials = geminidataanalytics.Credentials()
credentials.oauth.token.access_token = looker_access_token

datasource_references = geminidataanalytics.DatasourceReferences()
datasource_references.looker.explore_references = [looker_explore_reference]

Remplacez les exemples de valeurs comme suit :

  • my_access_token : valeur access_token que vous générez pour vous authentifier auprès de Looker.
  • https://my_company.looker.com : URL complète de votre instance Looker.
  • my_model : nom du modèle LookML qui inclut l'exploration à laquelle vous souhaitez vous connecter.
  • my_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 seule table BigQuery.

bq_project_id = "my_project_id"
bq_dataset_id = "my_dataset_id"
bq_table_id = "my_table_id"

bigquery_table_reference = geminidataanalytics.BigQueryTableReference()
bigquery_table_reference.project_id = bq_project_id
bigquery_table_reference.dataset_id = bq_dataset_id
bigquery_table_reference.table_id = bq_table_id

# Connect to your data source
datasource_references = geminidataanalytics.DatasourceReferences()
datasource_references.bq.table_references = [bigquery_table_reference]

Remplacez les exemples de valeurs comme suit :

  • my_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.
  • my_dataset_id : ID de l'ensemble de données BigQuery. Par exemple, san_francisco.
  • my_table_id : ID de la table BigQuery. Par exemple, street_trees.

Se connecter aux données Looker Studio

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

studio_datasource_id = "my_datasource_id"

studio_references = geminidataanalytics.StudioDatasourceReference()
studio_references.datasource_id = studio_datasource_id

## Connect to your data source
datasource_references.studio.studio_references = [studio_references]

Dans l'exemple précédent, remplacez my_datasource_id par l'ID de la source de données.

Configurer le contexte pour un chat avec ou sans état

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'exemple de code Python suivant montre comment configurer le contexte pour un chat avec état ou sans état :

  • Chat avec état : Google Cloud stocke et gère l'historique des conversations. 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 des conversations. Vous devez inclure l'intégralité de l'historique des conversations dans chaque nouveau message. Pour obtenir des exemples détaillés sur la gestion des conversations multitours en mode sans état, consultez Créer une conversation multitour sans état.

Chat avec état

L'exemple de code suivant configure le contexte pour le chat avec état, où Google Cloud stocke et gère l'historique des conversations. Vous pouvez également activer l'analyse avancée avec Python en incluant la ligne published_context.options.analysis.python.enabled = True dans l'exemple de code suivant.

# Set up context for stateful chat
published_context = geminidataanalytics.Context()
published_context.system_instruction = system_instruction
published_context.datasource_references = datasource_references
# Optional: To enable advanced analysis with Python, include the following line:
published_context.options.analysis.python.enabled = True

Chat sans état

L'exemple de code suivant configure le contexte pour le chat sans état, où vous devez envoyer l'intégralité de l'historique de la conversation à chaque message. Vous pouvez également activer l'analyse avancée avec Python en incluant la ligne inline_context.options.analysis.python.enabled = True dans l'exemple de code suivant.

# Set up context for stateless chat
# datasource_references.looker.credentials = credentials
inline_context = geminidataanalytics.Context()
inline_context.system_instruction = system_instruction
inline_context.datasource_references = datasource_references
# Optional: To enable advanced analysis with Python, include the following line:
inline_context.options.analysis.python.enabled = True

Créer un agent de données

L'exemple de code Python suivant effectue une requête API pour créer un agent de données, que vous pouvez ensuite utiliser pour discuter de vos données. L'agent de données est configuré avec la source de données, les instructions système et le contexte spécifiés.

data_agent_id = "data_agent_1"

data_agent = geminidataanalytics.DataAgent()
data_agent.data_analytics_agent.published_context = published_context
data_agent.name = f"projects/{billing_project}/locations/global/dataAgents/{data_agent_id}" # Optional

request = geminidataanalytics.CreateDataAgentRequest(
    parent=f"projects/{billing_project}/locations/global",
    data_agent_id=data_agent_id, # Optional
    data_agent=data_agent,
)

try:
    data_agent_client.create_data_agent(request=request)
    print("Data Agent created")
except Exception as e:
    print(f"Error creating Data Agent: {e}")

Dans l'exemple précédent, remplacez la valeur data_agent_1 par un identifiant unique pour l'agent de données.

Créer une conversation

L'exemple de code Python suivant effectue une requête API pour créer une conversation.

# Initialize request arguments
data_agent_id = "data_agent_1"
conversation_id = "conversation_1"

conversation = geminidataanalytics.Conversation()
conversation.agents = [f'projects/{billing_project}/locations/global/dataAgents/{data_agent_id}']
conversation.name = f"projects/{billing_project}/locations/global/conversations/{conversation_id}"

request = geminidataanalytics.CreateConversationRequest(
    parent=f"projects/{billing_project}/locations/global",
    conversation_id=conversation_id,
    conversation=conversation,
)

# Make the request
response = data_chat_client.create_conversation(request=request)

# Handle the response
print(response)

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

L'exemple de code Python suivant montre comment effectuer une requête API pour récupérer un agent de données que vous avez créé précédemment.

# Initialize request arguments
data_agent_id = "data_agent_1"
request = geminidataanalytics.GetDataAgentRequest(
    name=f"projects/{billing_project}/locations/global/dataAgents/{data_agent_id}",
)

# Make the request
response = data_agent_client.get_data_agent(request=request)

# Handle the response
print(response)

Dans l'exemple précédent, remplacez la valeur data_agent_1 par l'identifiant unique 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 appelant la méthode list_data_agents. 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"
request = geminidataanalytics.ListDataAgentsRequest(
    parent=f"projects/{billing_project}/locations/global",
)

# Make the request
page_result = data_agent_client.list_data_agents(request=request)

# Handle the response
for response in page_result:
    print(response)

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 appelant la méthode update_data_agent sur la ressource de l'agent de données. La requête nécessite un objet DataAgent qui inclut les nouvelles valeurs des champs que vous souhaitez modifier, ainsi qu'un paramètre update_mask qui accepte un objet FieldMask pour spécifier les champs à mettre à jour.

Pour mettre à jour un agent de données, vous devez disposer de l'autorisation IAM geminidataanalytics.dataAgents.update sur l'agent. Pour en savoir plus sur les rôles IAM incluant cette autorisation, consultez la liste des rôles prédéfinis.

data_agent_id = "data_agent_1"
billing_project = "YOUR-BILLING-PROJECT"
data_agent = geminidataanalytics.DataAgent()
data_agent.data_analytics_agent.published_context = published_context
data_agent.name = f"projects/{billing_project}/locations/global/dataAgents/{data_agent_id}"
data_agent.description = "Updated description of the data agent."

update_mask = field_mask_pb2.FieldMask(paths=['description', 'data_analytics_agent.published_context'])

request = geminidataanalytics.UpdateDataAgentRequest(
    data_agent=data_agent,
    update_mask=update_mask,
)

try:
    # Make the request
    data_agent_client.update_data_agent(request=request)
    print("Data Agent Updated")
except Exception as e:
    print(f"Error updating Data Agent: {e}")

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. : description de l'agent de données mis à jour.

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

Pour partager un agent, vous pouvez utiliser la méthode set_iam_policy afin d'attribuer des rôles IAM aux utilisateurs sur un agent spécifique. La requête inclut des liaisons qui spécifient 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"

resource = f"projects/{billing_project}/locations/global/dataAgents/{data_agent_id}"

# Construct the IAM policy
binding = policy_pb2.Binding(
    role=role,
    members= [f"user:{i.strip()}" for i in users.split(",")]
)

policy = policy_pb2.Policy(bindings=[binding])

# Create the request
request = iam_policy_pb2.SetIamPolicyRequest(
    resource=resource,
    policy=policy
)

# Send the request
try:
    response = data_agent_client.set_iam_policy(request=request)
    print("IAM Policy set successfully!")
    print(f"Response: {response}")
except Exception as e:
    print(f"Error setting IAM policy: {e}")

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 utiliser la méthode get_iam_policy pour extraire la stratégie IAM d'un agent de données. La requête spécifie le chemin d'accès à la ressource de l'agent de données.

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

resource = f"projects/{billing_project}/locations/global/dataAgents/{data_agent_id}"
request = iam_policy_pb2.GetIamPolicyRequest(
            resource=resource,
        )
try:
    response = data_agent_client.get_iam_policy(request=request)
    print("IAM Policy fetched successfully!")
    print(f"Response: {response}")
except Exception as e:
    print(f"Error setting IAM policy: {e}")

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 utiliser la méthode delete_data_agent pour effectuer une suppression réversible d'un agent de données. Lorsque vous supprimez un agent de façon réversible, il est supprimé, mais peut toujours être récupéré sous un délai de 30 jours. La requête spécifie l'URL de la ressource de l'agent de données.

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

request = geminidataanalytics.DeleteDataAgentRequest(
    name=f"projects/{billing_project}/locations/global/dataAgents/{data_agent_id}",
)

try:
    # Make the request
    data_agent_client.delete_data_agent(request=request)
    print("Data Agent Deleted")
except Exception as e:
    print(f"Error deleting Data Agent: {e}")

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 utiliser la méthode get_conversation pour extraire des informations sur une conversation existante. La requête spécifie le chemin d'accès à la ressource de la conversation.

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
conversation_id = "conversation_1"

request = geminidataanalytics.GetConversationRequest(
    name = f"projects/{billing_project}/locations/global/conversations/{conversation_id}"
)

# Make the request
response = data_chat_client.get_conversation(request=request)

# Handle the response
print(response)

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 appelant la méthode list_conversations. La requête spécifie l'URL de la ressource parente, qui correspond au projet et à l'emplacement (par exemple, projects/my-project/locations/global).

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"
request = geminidataanalytics.ListConversationsRequest(
    parent=f"projects/{billing_project}/locations/global",
)

# Make the request
response = data_chat_client.list_conversations(request=request)

# Handle the response
print(response)

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 utiliser la méthode list_messages pour extraire tous les messages d'une conversation. La requête spécifie le chemin d'accès à la ressource 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"

request = geminidataanalytics.ListMessagesRequest(
    parent=f"projects/{billing_project}/locations/global/conversations/{conversation_id}",
)

# Make the request
response = data_chat_client.list_messages(request=request)

# Handle the response
print(response)

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, l'exemple de code Python suivant vous permet d'envoyer une requête à l'agent. Le code utilise le contexte que vous configurez pour le chat avec ou sans état. L'API renvoie un flux de messages qui représentent les étapes suivies par l'agent pour répondre à la requête.

Chat avec état

Envoyer une requête de chat avec état avec une référence Conversation

Vous pouvez envoyer une requête de chat avec état à l'agent de données en référençant une ressource Conversation que vous avez créée précédemment.

# Create a request that contains a single user message (your question)
question = "Which species of tree is most prevalent?"
messages = [geminidataanalytics.Message()]
messages[0].user_message.text = question

data_agent_id = "data_agent_1"
conversation_id = "conversation_1"

# Create a conversation_reference
conversation_reference = geminidataanalytics.ConversationReference()
conversation_reference.conversation = f"projects/{billing_project}/locations/global/conversations/{conversation_id}"
conversation_reference.data_agent_context.data_agent = f"projects/{billing_project}/locations/global/dataAgents/{data_agent_id}"
# conversation_reference.data_agent_context.credentials = credentials

# Form the request
request = geminidataanalytics.ChatRequest(
    parent = f"projects/{billing_project}/locations/global",
    messages = messages,
    conversation_reference = conversation_reference
)

# Make the request
stream = data_chat_client.chat(request=request)

# Handle the response
for response in stream:
    show_message(response)

Remplacez les exemples de valeurs comme suit :

  • Which species of tree is most prevalent? : question en langage naturel à envoyer à l'agent de données.
  • data_agent_1 : identifiant unique de l'agent de données, tel que défini dans Créer un agent de données.
  • conversation_1 : identifiant unique de la conversation, tel que défini dans Créer une conversation.

Chat sans état

Les exemples de code suivants montrent comment envoyer une requête à l'agent de données lorsque vous avez configuré le contexte pour le chat sans état. Vous pouvez envoyer des requêtes sans état en référençant une ressource DataAgent définie précédemment ou en utilisant un contexte intégré dans la requête.

Envoyer une requête de chat sans état avec une référence DataAgent

Vous pouvez envoyer une requête à l'agent de données en référençant une ressource DataAgent que vous avez créée précédemment.

# Create a request that contains a single user message (your question)
question = "Which species of tree is most prevalent?"
messages = [geminidataanalytics.Message()]
messages[0].user_message.text = question

data_agent_id = "data_agent_1"

data_agent_context = geminidataanalytics.DataAgentContext()
data_agent_context.data_agent = f"projects/{billing_project}/locations/global/dataAgents/{data_agent_id}"
# data_agent_context.credentials = credentials

# Form the request
request = geminidataanalytics.ChatRequest(
    parent=f"projects/{billing_project}/locations/global",
    messages=messages,
    data_agent_context = data_agent_context
)

# Make the request
stream = data_chat_client.chat(request=request)

# Handle the response
for response in stream:
    show_message(response)

Remplacez les exemples de valeurs comme suit :

  • Which species of tree is most prevalent? : question en langage naturel à envoyer à l'agent de données.
  • data_agent_1 : identifiant unique de l'agent de données, tel que défini dans Créer un agent de données.

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

L'exemple de code suivant montre comment utiliser le paramètre inline_context pour fournir du contexte directement dans votre requête de chat sans état.

# Create a request that contains a single user message (your question)
question = "Which species of tree is most prevalent?"
messages = [geminidataanalytics.Message()]
messages[0].user_message.text = question

request = geminidataanalytics.ChatRequest(
    inline_context=inline_context,
    parent=f"projects/{billing_project}/locations/global",
    messages=messages,
)

# Make the request
stream = data_chat_client.chat(request=request)

# Handle the response
for response in stream:
    show_message(response)

Dans l'exemple précédent, remplacez Which species of tree is most prevalent? par une question en langage naturel à envoyer à l'agent de données.

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. L'exemple suivant montre comment créer une conversation multitour en référençant un agent de données ou en utilisant un contexte intégré pour fournir directement la source de données.

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

data_agent_id = "data_agent_1"

# Use data agent context
data_agent_context = geminidataanalytics.DataAgentContext()
data_agent_context.data_agent = f"projects/{billing_project}/locations/global/dataAgents/{data_agent_id}"
# data_agent_context.credentials = credentials

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

    message = geminidataanalytics.Message()
    message.user_message.text = msg

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

    request = geminidataanalytics.ChatRequest(
        parent=f"projects/{billing_project}/locations/global",
        messages=conversation_messages,
        # Use data agent context
        data_agent_context=data_agent_context,
        # Use inline context
        # inline_context=inline_context,
    )

    # Make the request
    stream = data_chat_client.chat(request=request)

    # Handle the response
    for response in stream:
      show_message(response)
      conversation_messages.append(response)

# Send the 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 :

  • data_agent_1 : identifiant unique de l'agent de données, tel que défini dans l'exemple de bloc de code dans Créer un agent de données.
  • 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

L'exemple de code suivant contient des définitions de fonctions d'assistance utilisées dans les exemples de code précédents. Ces fonctions permettent d'analyser la réponse de l'API et d'afficher les résultats.

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

import proto
from google.protobuf.json_format import MessageToDict, MessageToJson

def handle_text_response(resp):
  parts = getattr(resp, 'parts')
  print(''.join(parts))

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

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

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

def format_bq_table_ref(table_ref):
  return '{}.{}.{}'.format(table_ref.project_id, table_ref.dataset_id, table_ref.table_id)

def display_datasource(datasource):
  source_name = ''
  if 'studio_datasource_id' in datasource:
   source_name = getattr(datasource, 'studio_datasource_id')
  elif 'looker_explore_reference' in datasource:
   source_name = format_looker_table_ref(getattr(datasource, 'looker_explore_reference'))
  else:
    source_name = format_bq_table_ref(getattr(datasource, 'bigquery_table_reference'))

  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 'generated_sql' in resp:
    display_section_title('SQL generated')
    print(resp.generated_sql)
  elif 'result' in resp:
    display_section_title('Data retrieved')

    fields = [field.name for field in resp.result.schema.fields]
    d = {}
    for el in resp.result.data:
      for field in fields:
        if field in d:
          d[field].append(el[field])
        else:
          d[field] = [el[field]]

    display(pd.DataFrame(d))

def handle_chart_response(resp):
  def _value_to_dict(v):
    if isinstance(v, proto.marshal.collections.maps.MapComposite):
      return _map_to_dict(v)
    elif isinstance(v, proto.marshal.collections.RepeatedComposite):
      return [_value_to_dict(el) for el in v]
    elif isinstance(v, (int, float, str, bool)):
      return v
    else:
      return MessageToDict(v)

  def _map_to_dict(d):
    out = {}
    for k in d:
      if isinstance(d[k], proto.marshal.collections.maps.MapComposite):
        out[k] = _map_to_dict(d[k])
      else:
        out[k] = _value_to_dict(d[k])
    return out

  if 'query' in resp:
    print(resp.query.instructions)
  elif 'result' in resp:
    vegaConfig = resp.result.vega_config
    vegaConfig_dict = _map_to_dict(vegaConfig)
    alt.Chart.from_json(json_lib.dumps(vegaConfig_dict)).display();

def show_message(msg):
  m = msg.system_message
  if 'text' in m:
    handle_text_response(getattr(m, 'text'))
  elif 'schema' in m:
    handle_schema_response(getattr(m, 'schema'))
  elif 'data' in m:
    handle_data_response(getattr(m, 'data'))
  elif 'chart' in m:
    handle_chart_response(getattr(m, 'chart'))
  print('\n')