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 (Zugriff über geminidataanalytics.googleapis.com) senden.

Das Python-Codebeispiel 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
KI-Datenagenten erstellen
Unterhaltung erstellen
KI-Datenagenten 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 die 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 KI-Agenten 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 Agenten mit API-Schlüsseln bei einer 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: der Clientschlüssel Ihres generierten Looker API-Schlüssels.

Zugriffstokens verwenden

Das folgende Python-Codebeispiel zeigt, wie Sie Ihren Agenten mit Zugriffstokens bei einer 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 KI-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 generieren, wie unter Mit der Conversational Analytics API authentifizieren und eine Verbindung zu einer Datenquelle herstellen beschrieben. Mit der Conversational Analytics API können Sie jeweils nur zu einem Looker-Explore eine Verbindung herstellen.

looker_data_source = {
    "looker": {
        "explore_references": {
            "looker_instance_uri": "https://your_company.looker.com",
            "lookml_model": "your_model",
            "explore": "your_explore",
       },
       # Do not include the following line during agent creation
       "credentials": looker_credentials
    }
}

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 den Explore enthält, zu dem Sie eine Verbindung herstellen möchten.
your_explore: der Name des Looker-Explore, der vom KI-Datenagenten abgefragt werden soll.

Verbindung zu BigQuery-Daten herstellen

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

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

bigquery_data_sources = {
    "bq": {
        "tableReferences": [
            {
                "projectId": "my_project_id",
                "datasetId": "my_dataset_id",
                "tableId": "my_table_id"
            },
            {
                "projectId": "my_project_id_2",
                "datasetId": "amy_dataset_id_2",
                "tableId": "my_table_id_2"
            },
            # Up to 10 total tables may be included
        ]
    }
}

Ersetzen Sie die Beispielwerte so:

my_project_id: die ID des Projekts in Google Cloud , das das BigQuery-Dataset und die BigQuery-Tabelle enthält, zu denen Sie eine Verbindung herstellen möchten. Wenn Sie eine Verbindung zu einem öffentlichen Dataset herstellen möchten, geben Sie bigquery-public-data an.
my_dataset_id: die ID des BigQuery-Datasets.
my_table_id: 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 Datenquellen-ID.

KI-Datenagenten erstellen

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

Der vollständige Ressourcenname des Agents. Dieser Wert enthält die Projekt-ID, den Standort und eine eindeutige Kennung für den Agenten.
Die Beschreibung des Daten-Agents.
Der Kontext des KI-Datenagents, 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/v1beta/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 KI-Datenagenten. 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 des KI-Datenagenten.

Unterhaltung erstellen

Im folgenden Beispielcode wird gezeigt, wie Sie eine Unterhaltung mit Ihrem KI-Datenagenten erstellen.

conversation_url = f"https://geminidataanalytics.googleapis.com/v1beta/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 KI-Datenagenten, wie im Beispielcodeblock unter KI-Datenagenten erstellen definiert.
conversation_1: eine eindeutige Kennung für die Unterhaltung.

KI-Datenagenten und Unterhaltungen verwalten

Die folgenden Beispielcodes zeigen, wie Sie Ihre KI-Datenagenten und Unterhaltungen mit der Conversational Analytics API verwalten. Sie können folgende Aufgaben ausführen:

KI-Datenagenten abrufen
KI-Datenagenten auflisten
KI-Datenagenten aktualisieren
IAM-Richtlinie für einen KI-Datenagenten festlegen
IAM-Richtlinie für einen KI-Datenagenten abrufen
KI-Datenagenten löschen
Unterhaltung abrufen
Unterhaltungen auflisten
Nachrichten in einer Unterhaltung auflisten

KI-Datenagenten abrufen

Im folgenden Beispielcode wird gezeigt, wie Sie einen vorhandenen KI-Datenagenten abrufen, indem Sie eine HTTP-GET-Anfrage an die URL der KI-Datenagenten-Ressource senden.

data_agent_id = "data_agent_1"
data_agent_url = f"{base_url}/v1beta/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 KI-Datenagenten, den Sie abrufen möchten.

KI-Datenagenten auflisten

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

Zum Auflisten aller KI-Agenten 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}/v1beta/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.

KI-Datenagenten aktualisieren

Im folgenden Beispielcode wird gezeigt, wie ein KI-Datenagent aktualisiert wird, indem eine HTTP-PATCH-Anfrage an die URL der KI-Datenagenten-Ressource gesendet wird. Die Anfragenutzlast enthält die neuen Werte für die Felder, die Sie ändern möchten. Die Anfrageparameter enthalten den Parameter updateMask, mit dem die zu aktualisierenden Felder angegeben werden.

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

data_agent_url = f"{base_url}/v1beta/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 KI-Datenagenten, den Sie aktualisieren möchten.
YOUR-BILLING-PROJECT: die ID Ihres Abrechnungsprojekts.
Updated description of the data agent.: Eine neue Beschreibung für den KI-Datenagenten.

IAM-Richtlinie für einen KI-Datenagenten festlegen

Wenn Sie einen KI-Agenten freigeben möchten, können Sie mit der setIamPolicy-Methode Nutzern IAM-Rollen für einen bestimmten KI-Agenten zuweisen. Im folgenden Beispielcode wird gezeigt, wie Sie einen POST-Aufruf an die URL des KI-Datenagenten mit einer Nutzlast senden, 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}/v1beta/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 KI-Datenagenten, 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 KI-Datenagenten abrufen

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

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

data_agent_url = f"{base_url}/v1beta/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 KI-Datenagenten, für den Sie die IAM-Richtlinie abrufen möchten.

KI-Datenagenten löschen

Der folgende Beispielcode zeigt, wie Sie einen KI-Datenagenten vorläufig löschen, indem Sie eine HTTP-DELETE-Anfrage an die URL der KI-Datenagenten-Ressource senden. Wenn Sie einen KI-Agenten vorläufig löschen, wird er 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}/v1beta/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 KI-Datenagenten, den Sie löschen möchten.

Unterhaltung abrufen

Im folgenden Beispielcode 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}/v1beta/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}/v1beta/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}/v1beta/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.

API zum Stellen von Fragen verwenden

Nachdem Sie einen KI-Datenagenten 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 weiterführende Fragen stellen können, die auf dem vorherigen Kontext aufbauen. Die API bietet die folgenden Methoden zum Verwalten des Unterhaltungsverlaufs:

Zustandsorientierter Chat: Google Cloud speichert und verwaltet den Unterhaltungsverlauf. Der zustandsorientierte Chat ist grundsätzlich ein Multi-Turn-Chat, da die API den Kontext aus vorangegangenen Nachrichten beibehält. Sie müssen nur Ihre aktuelle Nachricht senden, um die Unterhaltung am Laufen zu halten.
Zustandsloser Chat: Ihre Anwendung verwaltet den Unterhaltungsverlauf. Sie müssen die relevanten vorherigen Nachrichten in jede neue Nachricht einfügen. Ausführliche Beispiele zum Verwalten von Multi-Turn-Unterhaltungen im zustandslosen Modus finden Sie unter Zustandslose Multi-Turn-Unterhaltung erstellen.

Zustandsorientierter Chat

Zustandsorientierte Chatanfrage mit einem Conversation-Verweis 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 die get_stream-Hilfsfunktion verwendet, um die Antwort zu streamen.

chat_url = f"https://geminidataanalytics.googleapis.com/v1beta/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 KI-Datenagenten, wie im Beispielcodeblock unter KI-Datenagenten erstellen definiert.
conversation_1: eine eindeutige Kennung für die Unterhaltung.
Make a bar graph for the top 5 states by the total number of airports wurde als Beispielaufforderung verwendet.

Zustandsloser Chat

Zustandslose Chatanfrage mit einem KI-Datenagentenverweis senden

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

chat_url = f"https://geminidataanalytics.googleapis.com/v1beta/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 KI-Datenagenten, wie im Beispielcodeblock unter KI-Datenagenten erstellen definiert.
Make a bar graph for the top 5 states by the total number of airports wurde als Beispielaufforderung verwendet.

Zustandslose Chatanfrage mit Inline-Kontext senden

Der folgende Beispielcode zeigt, wie Sie der API mithilfe von Inline-Kontext eine zustandslose Frage 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/v1beta/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 weiterführende Fragen 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 weiterführende Fragen stellen, die auf vorherigen Beiträgen aufbauen. In der Nutzlast der Funktion können Sie auf einen KI-Datenagenten verweisen oder die Datenquelle direkt über Inline-Kontext angeben.

chat_url = f"https://geminidataanalytics.googleapis.com/v1beta/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 KI-Datenagenten, wie im Beispielcodeblock unter KI-Datenagenten erstellen definiert.

Sie können die Hilfsfunktion multi_turn_Conversation für jede Runde des Gesprächs aufrufen. Im folgenden Beispielcode wird gezeigt, 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 KI-Datenagenten gesendet werden soll.
Can you show me the results as a bar chart?: eine weiterführende Frage, die auf der vorangegangenen Frage aufbaut oder sie verfeinert.

Antworten verarbeiten

Die folgende get_stream_multi_turn-Funktion verarbeitet die Streaming API-Antwort. 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 diesem Leitfaden behandelt werden.

KI-Datenagenten 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",
        },
        # Do not include the following line during agent creation
        # "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/v1beta/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/v1beta/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/v1beta/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/v1beta/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/v1beta/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/v1beta/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

KI-Datenagenten erstellen

Unterhaltung erstellen

KI-Datenagenten und Unterhaltungen verwalten

KI-Datenagenten abrufen

KI-Datenagenten auflisten

KI-Datenagenten aktualisieren

IAM-Richtlinie für einen KI-Datenagenten festlegen

IAM-Richtlinie für einen KI-Datenagenten abrufen

KI-Datenagenten löschen

Unterhaltung abrufen

Unterhaltungen auflisten

Nachrichten in einer Unterhaltung auflisten

API zum Stellen von Fragen verwenden

Zustandsorientierter Chat

Zustandsorientierte Chatanfrage mit einem Conversation-Verweis senden

Zustandsloser Chat

Zustandslose Chatanfrage mit einem KI-Datenagentenverweis senden

Zustandslose Chatanfrage mit Inline-Kontext senden

Zustandslose Multi-Turn-Unterhaltung erstellen

Anfragen mit mehreren Turns senden

Antworten verarbeiten

Vollständiges Codebeispiel

KI-Datenagenten mit HTTP und Python erstellen

Hilfsfunktionen in Python zum Streamen von Chatantworten

Data Agent mit HTTP und Python erstellen