Diese Seite wurde von der Cloud Translation API übersetzt.

Data Agent mit HTTP und Python erstellen

Auf dieser Seite wird beschrieben, wie Sie mit Python HTTP-Anfragen an die Conversational Analytics API (über geminidataanalytics.googleapis.com) senden.

Der Beispiel-Python-Code auf dieser Seite zeigt, wie Sie die folgenden Aufgaben ausführen:

Erste Einstellungen und Authentifizierung konfigurieren
Verbindung zu einer Looker-, BigQuery- oder Looker Studio-Datenquelle herstellen
Daten-Agent erstellen
Unterhaltung erstellen
Data Agents und Unterhaltungen verwalten
API zum Stellen von Fragen verwenden
Zustandslose Multi-Turn-Unterhaltung erstellen

Eine vollständige Version des Beispielcodes finden Sie am Ende der Seite, zusammen mit den Hilfsfunktionen, die zum Streamen der API-Antwort verwendet werden.

Erste Einstellungen und Authentifizierung konfigurieren

Der folgende Python-Beispielcode führt diese Aufgaben aus:

Importiert die erforderlichen Python-Bibliotheken
Zugriffstoken für die HTTP-Authentifizierung mit der Google Cloud CLI abrufen
Definiert Variablen für das Abrechnungsprojekt und Systemanweisungen

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'

Ersetzen Sie die Beispielwerte so:

YOUR-BILLING-PROJECT: Die ID des Abrechnungsprojekts, in dem Sie die erforderlichen APIs aktiviert haben.
YOUR-SYSTEM-INSTRUCTIONS: Systemanweisungen, um das Verhalten des Agents zu steuern und an Ihre Datenanforderungen anzupassen. Sie können beispielsweise Systemanweisungen verwenden, um geschäftliche Begriffe zu definieren, die Länge der Antworten zu steuern oder die Datenformatierung festzulegen. Idealerweise definieren Sie Systemanweisungen im empfohlenen YAML-Format, das unter Effektive Systemanweisungen schreiben beschrieben wird, um detaillierte und strukturierte Anleitungen zu geben.

Bei Looker authentifizieren

Wenn Sie eine Verbindung zu einer Looker-Datenquelle herstellen möchten, müssen Sie sich bei der Looker-Instanz authentifizieren.

API-Schlüssel verwenden

Das folgende Python-Codebeispiel zeigt, wie Sie Ihren Agent mit API-Schlüsseln für eine Looker-Instanz authentifizieren.

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

Ersetzen Sie die Beispielwerte so:

YOUR-LOOKER-CLIENT-ID: Die Client-ID Ihres generierten Looker-API-Schlüssels.
YOUR-LOOKER-CLIENT-SECRET: Das Client-Secret Ihres generierten Looker API-Schlüssels.

Zugriffstokens verwenden

Das folgende Python-Codebeispiel zeigt, wie Sie Ihren Agent mit Zugriffstokens für eine Looker-Instanz authentifizieren.

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

Ersetzen Sie die Beispielwerte so:

YOUR-TOKEN: Der access_token-Wert, den Sie zur Authentifizierung bei Looker generieren.

Mit einer Datenquelle verbinden

Die folgenden Python-Codebeispiele zeigen, wie Sie die Looker-, BigQuery- oder Looker Studio-Datenquelle für Ihren Agenten definieren.

Verbindung zu Looker-Daten herstellen

Im folgenden Beispielcode wird eine Verbindung zu einem Looker-Explore definiert. Wenn Sie eine Verbindung zu einer Looker-Instanz herstellen möchten, müssen Sie Looker-API-Schlüssel generiert haben, wie unter Authentifizieren und Verbinden mit einer Datenquelle über die Conversational Analytics API beschrieben.

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

Ersetzen Sie die Beispielwerte so:

https://your_company.looker.com: Die vollständige URL Ihrer Looker-Instanz.
your_model: Der Name des LookML-Modells, das das Explore enthält, mit dem Sie eine Verbindung herstellen möchten.
your_explore: Der Name des Looker-Explores, das vom Data-Agent abgefragt werden soll.

Verbindung zu BigQuery-Daten herstellen

Mit der Conversational Analytics API können Sie gleichzeitig eine Verbindung zu bis zu 10 BigQuery-Tabellen herstellen und Abfragen ausführen.

Im folgenden Beispielcode wird eine Verbindung zu einer BigQuery-Tabelle definiert.

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

Ersetzen Sie die Beispielwerte so:

bigquery-public-data: Die ID des Google Cloud -Projekts, das das BigQuery-Dataset und die Tabelle enthält, mit denen Sie eine Verbindung herstellen möchten. Wenn Sie eine Verbindung zu einem öffentlichen Dataset herstellen möchten, geben Sie bigquery-public-data an.
san_francisco: Die ID des BigQuery-Datasets.
street_trees: Die ID der BigQuery-Tabelle.

Verbindung zu Looker Studio-Daten herstellen

Im folgenden Beispielcode wird eine Verbindung zu einer Looker Studio-Datenquelle definiert.

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

Ersetzen Sie studio_datasource_id durch die ID der Datenquelle.

Data-Agent erstellen

Im folgenden Beispielcode wird gezeigt, wie der Daten-Agent erstellt wird, indem eine HTTP-POST-Anfrage an den Endpunkt für die Erstellung von Daten-Agents gesendet wird. Die Nutzlast der Anfrage enthält die folgenden Details:

Der vollständige Ressourcenname für den Agent. Dieser Wert enthält die Projekt-ID, den Standort und eine eindeutige Kennung für den Agent.
Die Beschreibung des Daten-Agents.
Der Kontext des Daten-Agents, einschließlich der Systembeschreibung (definiert in Erste Einstellungen und Authentifizierung konfigurieren) und der Datenquelle, die der Agent verwendet (definiert in Mit einer Datenquelle verbinden).

Optional können Sie auch die erweiterte Analyse mit Python aktivieren, indem Sie den Parameter options in die Anfrage-Nutzlast einfügen.

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)

Ersetzen Sie die Beispielwerte so:

data_agent_1: Eine eindeutige Kennung für den Daten-Agent. Dieser Wert wird im Ressourcennamen des Agents und als URL-Abfrageparameter data_agent_id verwendet.
This is the description of data_agent_1.: Eine Beschreibung für den Daten-Agent.

Unterhaltung erstellen

Der folgende Beispielcode zeigt, wie Sie eine Unterhaltung mit Ihrem Daten-Agent erstellen.

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)

Ersetzen Sie die Beispielwerte so:

data_agent_1: Die ID des Daten-Agents, wie im Beispielcodeblock unter Daten-Agent erstellen definiert.
conversation_1: Eine eindeutige ID für die Unterhaltung.

Daten-Agents und Unterhaltungen verwalten

Die folgenden Codebeispiele zeigen, wie Sie Ihre Data Agents und Unterhaltungen mit der Conversational Analytics API verwalten. Sie können folgende Aufgaben ausführen:

Daten-Agent abrufen
Daten-Agents auflisten
Daten-Agent aktualisieren
IAM-Richtlinie für einen Daten-Agent festlegen
IAM-Richtlinie für einen Daten-Agent abrufen
Daten-Agent löschen
Unterhaltung starten
Unterhaltungen auflisten
Nachrichten in einer Unterhaltung auflisten

Daten-Agent abrufen

Der folgende Beispielcode zeigt, wie Sie einen vorhandenen Daten-Agent abrufen, indem Sie eine HTTP-GET-Anfrage an die URL der Daten-Agent-Ressource senden.

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)

Ersetzen Sie im vorherigen Beispiel data_agent_1 durch die ID des Daten-Agents, den Sie abrufen möchten.

Daten-Agents auflisten

Im folgenden Code wird gezeigt, wie Sie alle Daten-Agents für ein bestimmtes Projekt auflisten, indem Sie eine HTTP-GET-Anfrage an den dataAgents-Endpunkt senden.

Zum Auflisten aller Agents benötigen Sie die Berechtigung geminidataanalytics.dataAgents.list für das Projekt. Weitere Informationen dazu, welche IAM-Rollen diese Berechtigung enthalten, finden Sie in der Liste der vordefinierten Rollen.

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

Ersetzen Sie YOUR-BILLING-PROJECT durch die ID Ihres Abrechnungsprojekts.

Daten-Agent aktualisieren

Der folgende Beispielcode zeigt, wie Sie einen Daten-Agent aktualisieren, indem Sie eine HTTP-PATCH-Anfrage an die Ressourcen-URL des Daten-Agents senden. Die Nutzlast der Anfrage enthält die neuen Werte für die Felder, die Sie ändern möchten. Die Anfrageparameter enthalten einen updateMask-Parameter, der die zu aktualisierenden Felder angibt.

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)

Ersetzen Sie die Beispielwerte so:

data_agent_1: Die ID des Daten-Agents, den Sie aktualisieren möchten.
YOUR-BILLING-PROJECT: Die ID Ihres Abrechnungsprojekts.
Updated description of the data agent.: Eine neue Beschreibung für den Daten-Agent.

IAM-Richtlinie für einen Daten-Agent festlegen

Wenn Sie einen Agent freigeben möchten, können Sie mit der Methode setIamPolicy Nutzern IAM-Rollen für einen bestimmten Agent zuweisen. Der folgende Beispielcode zeigt, wie Sie einen POST-Aufruf an die Daten-Agent-URL mit einer Nutzlast ausführen, die Bindungen enthält. Die Bindung gibt an, welche Rollen welchen Nutzern zugewiesen werden sollen.

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)

Ersetzen Sie die Beispielwerte so:

YOUR-BILLING-PROJECT: Die ID Ihres Abrechnungsprojekts.
data_agent_1: Die ID des Daten-Agents, für den Sie die IAM-Richtlinie festlegen möchten.
222larabrown@gmail.com, cloudysanfrancisco@gmail.com: Eine durch Kommas getrennte Liste von Nutzer-E-Mail-Adressen, denen Sie die angegebene Rolle zuweisen möchten.

IAM-Richtlinie für einen Daten-Agent abrufen

Der folgende Beispielcode zeigt, wie Sie die IAM-Richtlinie für einen Daten-Agent abrufen, indem Sie eine HTTP-POST-Anfrage an die Daten-Agent-URL senden. Die Nutzlast der Anfrage enthält den Pfad des Daten-Agents.

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)

Ersetzen Sie die Beispielwerte so:

YOUR-BILLING-PROJECT: Die ID Ihres Abrechnungsprojekts.
data_agent_1: Die ID des Daten-Agents, für den Sie die IAM-Richtlinie abrufen möchten.

Data Agent löschen

Im folgenden Beispielcode wird gezeigt, wie ein Daten-Agent vorläufig gelöscht wird, indem eine HTTP-DELETE-Anfrage an die Ressourcen-URL des Daten-Agents gesendet wird. Beim vorläufigen Löschen wird der Agent gelöscht, kann aber innerhalb von 30 Tagen wiederhergestellt werden.

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)

Ersetzen Sie die Beispielwerte so:

YOUR-BILLING-PROJECT: Die ID Ihres Abrechnungsprojekts.
data_agent_1: Die ID des Daten-Agents, den Sie löschen möchten.

Unterhaltung abrufen

Im folgenden Codebeispiel wird gezeigt, wie Sie eine vorhandene Unterhaltung abrufen, indem Sie eine HTTP-GET-Anfrage an die URL der Unterhaltungsressource senden.

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)

Ersetzen Sie die Beispielwerte so:

YOUR-BILLING-PROJECT: Die ID Ihres Abrechnungsprojekts.
conversation_1: Die ID der Unterhaltung, die Sie abrufen möchten.

Unterhaltungen auflisten

Im folgenden Beispielcode wird gezeigt, wie Sie Unterhaltungen für ein bestimmtes Projekt auflisten, indem Sie eine HTTP-GET-Anfrage an den Endpunkt conversations senden.

Standardmäßig werden mit dieser Methode die Unterhaltungen zurückgegeben, die Sie erstellt haben. Administratoren (Nutzer mit der IAM-Rolle cloudaicompanion.topicAdmin) können alle Unterhaltungen im Projekt sehen.

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)

Ersetzen Sie YOUR-BILLING-PROJECT durch die ID des Abrechnungsprojekts, in dem Sie die erforderlichen APIs aktiviert haben.

Nachrichten in einer Unterhaltung auflisten

Der folgende Beispielcode zeigt, wie Sie alle Nachrichten in einer Unterhaltung auflisten, indem Sie eine HTTP-GET-Anfrage an den messages-Endpunkt der Unterhaltung senden.

Zum Auflisten von Nachrichten benötigen Sie die Berechtigung cloudaicompanion.topics.get für die Unterhaltung.

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)

Ersetzen Sie die Beispielwerte so:

YOUR-BILLING-PROJECT: Die ID Ihres Abrechnungsprojekts.
conversation_1: Die ID der Unterhaltung, für die Sie Nachrichten auflisten möchten.

Fragen über die API stellen

Nachdem Sie einen Daten-Agenten und eine Unterhaltung erstellt haben, können Sie Fragen zu Ihren Daten stellen.

Die Conversational Analytics API unterstützt Multi-Turn-Unterhaltungen, bei denen Nutzer Folgefragen stellen können, die auf dem vorherigen Kontext aufbauen. Die API bietet die folgenden Methoden zum Verwalten des Unterhaltungsverlaufs:

Statusbehafteter Chat: Google Cloud Speichert und verwaltet den Unterhaltungsverlauf. Der zustandsbehaftete Chat ist von Natur aus ein Multi-Turn-Chat, da die API den Kontext aus früheren Nachrichten beibehält. Sie müssen nur die aktuelle Nachricht für jeden Zug senden.
Zustandsloser Chat: Ihre Anwendung verwaltet den Unterhaltungsverlauf. Sie müssen jeder neuen Nachricht die relevanten vorherigen Nachrichten beifügen. Ausführliche Beispiele für die Verwaltung von wechselseitigen Unterhaltungen im zustandslosen Modus finden Sie unter Zustandslose wechselseitige Unterhaltung erstellen.

Zustandsorientierter Chat

Statusbehaftete Chatanfrage mit Konversationsreferenz senden

Im folgenden Beispielcode wird gezeigt, wie Sie der API Fragen mit der conversation stellen, die Sie in den vorherigen Schritten definiert haben. In diesem Beispiel wird eine get_stream-Hilfsfunktion verwendet, um die Antwort zu streamen.

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)

Ersetzen Sie die Beispielwerte so:

data_agent_1: Die ID des Daten-Agents, wie im Beispielcodeblock unter Daten-Agent erstellen definiert.
conversation_1: Eine eindeutige ID für die Unterhaltung.
Make a bar graph for the top 5 states by the total number of airports wurde als Beispiel-Prompt verwendet.

Zustandsloser Chat

Eine statuslose Chatanfrage mit einer Daten-Agent-Referenz senden

Im folgenden Beispielcode wird gezeigt, wie Sie der API eine zustandslose Frage stellen. Dazu verwenden Sie den Data Agent, den Sie in den vorherigen Schritten definiert haben. In diesem Beispiel wird eine get_stream-Hilfsfunktion verwendet, um die Antwort zu streamen.

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)

Ersetzen Sie die Beispielwerte so:

data_agent_1: Die ID des Daten-Agents, wie im Beispielcodeblock unter Daten-Agent erstellen definiert.
Make a bar graph for the top 5 states by the total number of airports wurde als Beispiel-Prompt verwendet.

Statuslose Chatanfrage mit Inline-Kontext senden

Der folgende Beispielcode zeigt, wie Sie der API eine zustandslose Frage mit Inline-Kontext stellen. In diesem Beispiel wird eine get_stream-Hilfsfunktion verwendet, um die Antwort zu streamen. Als Beispiel wird eine BigQuery-Datenquelle verwendet.

Optional können Sie auch die erweiterte Analyse mit Python aktivieren, indem Sie den Parameter options in die Anfrage-Nutzlast einfügen.

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)

Zustandslose Multi-Turn-Unterhaltung erstellen

Wenn Sie in einer zustandslosen Unterhaltung Folgefragen stellen möchten, muss Ihre Anwendung den Kontext der Unterhaltung verwalten, indem sie den gesamten Nachrichtenverlauf mit jeder neuen Anfrage sendet. In den folgenden Abschnitten wird gezeigt, wie Sie Hilfsfunktionen definieren und aufrufen, um eine Unterhaltung mit mehreren Zügen zu erstellen:

Anfragen mit mehreren Turns senden
Antworten verarbeiten

Anfragen mit mehreren Turns senden

Die folgende Hilfsfunktion multi_turn_Conversation verwaltet den Unterhaltungskontext, indem Nachrichten in einer Liste gespeichert werden. So können Sie Folgefragen stellen, die auf vorherigen Beiträgen aufbauen. In der Nutzlast der Funktion können Sie auf einen Daten-Agent verweisen oder die Datenquelle direkt über Inline-Kontext angeben.

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)

Ersetzen Sie im vorherigen Beispiel data_agent_1 durch die ID des Daten-Agents, wie im Beispielcodeblock unter Daten-Agent erstellen definiert.

Sie können die Hilfsfunktion multi_turn_Conversation für jede Runde des Gesprächs aufrufen. Das folgende Codebeispiel zeigt, wie Sie eine erste Anfrage und dann eine Folgeanfrage senden, die auf der vorherigen Antwort aufbaut.

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

Ersetzen Sie im vorherigen Beispiel die Beispielwerte so:

Which species of tree is most prevalent?: Eine Frage in natürlicher Sprache, die an den Data Agent gesendet werden soll.
Can you show me the results as a bar chart?: Eine Folgefrage, die auf der vorherigen Frage aufbaut oder sie verfeinert.

Antworten verarbeiten

Die folgende get_stream_multi_turn-Funktion verarbeitet die Antwort der Streaming API. Diese Funktion ähnelt der get_stream-Hilfsfunktion, speichert die Antwort jedoch in der Liste conversation_messages, um den Unterhaltungskontext für den nächsten Zug zu speichern.

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

Vollständiges Codebeispiel

Das folgende aufklappbare Codebeispiel enthält alle Aufgaben, die in dieser Anleitung behandelt werden.

Daten-Agent mit HTTP und Python erstellen

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

Das folgende aufklappbare Codebeispiel enthält die Python-Hilfsfunktionen, die zum Streamen von Chatantworten verwendet werden.

Hilfsfunktionen in Python zum Streamen von Chatantworten

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

Data Agent mit HTTP und Python erstellen Mit Sammlungen den Überblick behalten Sie können Inhalte basierend auf Ihren Einstellungen speichern und kategorisieren.

Erste Einstellungen und Authentifizierung konfigurieren

Bei Looker authentifizieren

API-Schlüssel verwenden

Zugriffstokens verwenden

Mit einer Datenquelle verbinden

Verbindung zu Looker-Daten herstellen

Verbindung zu BigQuery-Daten herstellen

Verbindung zu Looker Studio-Daten herstellen

Data-Agent erstellen

Unterhaltung erstellen

Daten-Agents und Unterhaltungen verwalten

Daten-Agent abrufen

Daten-Agents auflisten

Daten-Agent aktualisieren

IAM-Richtlinie für einen Daten-Agent festlegen

IAM-Richtlinie für einen Daten-Agent abrufen

Data Agent löschen

Unterhaltung abrufen

Unterhaltungen auflisten

Nachrichten in einer Unterhaltung auflisten

Fragen über die API stellen

Zustandsorientierter Chat

Statusbehaftete Chatanfrage mit Konversationsreferenz senden

Zustandsloser Chat

Eine statuslose Chatanfrage mit einer Daten-Agent-Referenz senden

Statuslose Chatanfrage mit Inline-Kontext senden

Zustandslose Multi-Turn-Unterhaltung erstellen

Anfragen mit mehreren Turns senden

Antworten verarbeiten

Vollständiges Codebeispiel

Daten-Agent mit HTTP und Python erstellen

Hilfsfunktionen in Python zum Streamen von Chatantworten

Data Agent mit HTTP und Python erstellen