Daten-Agents mit A2A orchestrieren

Die Conversational Analytics API Google Cloud implementiert das offene Agent-to-Agent-Protokoll (A2A). Damit können Agenten in Multi-Agenten-Workflows Funktionen ermitteln, Analyseanfragen delegieren und strukturierte Antworten streamen, z. B. ausführbare SQL-Abfragen und Diagrammvisualisierungen.

Sie können integrierte Daten-KI-Agenten der Conversational Analytics API für BigQuery und Looker abfragen, indem Sie in Ihrer API-Anfrage den Dataset-Kontext übergeben. Alternativ können Sie benutzerdefinierte Daten-KI-Agenten abfragen, die mit der Geschäftslogik Ihrer Organisation konfiguriert sind.

Informationen zum direkten Erstellen und Abfragen von KI-Datenagenten finden Sie unter KI-Datenagenten mit dem Python SDK erstellen oder KI-Datenagenten mit HTTP erstellen.

Weitere Informationen dazu, wie und wann Gemini for Google Cloud Ihre Daten verwendet

Funktionsweise der Orchestrierung von Daten-KI-Agenten

Wenn Sie Conversational Analytics API-KI-Datenagenten in eine Anwendung oder ein Multi-Agenten-System einbinden, folgt der Orchestrierungsworkflow diesen Vorgängen:

  • Der Orchestrator-Agent erkennt die Funktionen und Skills eines Daten-Agents, indem er seine Agentenkarte prüft, bevor er Analyseanfragen delegiert.
  • Der Orchestrator-Agent sendet eine Nachricht, um entweder einen integrierten Daten-Agenten (agents/bigquery-ca oder agents/looker-ca) abzufragen, indem Datenquellen in der Anfrage angegeben werden, oder einen benutzerdefinierten Daten-Agenten (dataAgents/DATA_AGENT_ID), der mit der Geschäftslogik der Domain konfiguriert ist.
  • Der Daten-Agent verarbeitet die Anfrage, führt die erforderlichen Abfragen aus und gibt die Ergebnisse zurück – entweder als vollständige Beantwortung oder durch Streaming von Begründungsfortschritt und strukturierten Artefakten (z. B. ausführbarer SQL-Code und Vega-Lite-Diagrammspezifikationen).

Hinweis

Bevor Sie beginnen, müssen die folgenden Voraussetzungen erfüllt sein:

  1. Aktivieren Sie die Conversational Analytics API, die BigQuery API und die Looker API in Ihrem Google Cloud -Projekt.
  2. Prüfen Sie, ob Sie die erforderlichen IAM-Rollen und ‑Berechtigungen haben.
  3. Authentifizieren Sie sich für die Conversational Analytics API und installieren Sie die Clientbibliothek oder rufen Sie ein Autorisierungstoken ab.

Erforderliche Rollen

Bitten Sie Ihren Administrator, Ihnen die folgenden IAM-Rollen für das Projekt zuzuweisen, um die Berechtigungen zu erhalten, die Sie zum Ermitteln und Abfragen von Data Agents über A2A benötigen:

Weitere Informationen zum Zuweisen von Rollen finden Sie unter Zugriff auf Projekte, Ordner und Organisationen verwalten.

Sie können die erforderlichen Berechtigungen auch über benutzerdefinierte Rollen oder andere vordefinierte Rollen erhalten.

Wenn Sie zugrunde liegende Datenquellen abfragen möchten, benötigen Sie außerdem Leseberechtigungen für Ihre BigQuery-Zieldatasets (z. B. roles/bigquery.dataViewer) oder Looker-Explores.

Funktionen von KI-Agenten entdecken

Bevor Anfragen an einen Daten-Agenten delegiert werden, kann ein Orchestrator-Agent oder eine Clientanwendung die Agentenkarte prüfen, um die Funktionen und die Konfiguration des Agenten zu sehen, z. B. die Beschreibung, die verfügbaren Skills und die unterstützten Erweiterungen. Sie können Agentenkarten mit der Methode getCard für integrierte Datenagenten (agents/bigquery-ca und agents/looker-ca) und benutzerdefinierte Datenagenten (dataAgents/DATA_AGENT_ID) abrufen.

Agentenkarte abrufen

Die folgenden Codebeispiele zeigen, wie Sie eine Agentenkarte abrufen. In diesen Beispielen wird der integrierte BigQuery-Datenagent (agents/bigquery-ca) verwendet. Sie können die Karte für den integrierten Looker-Agenten (agents/looker-ca) oder einen benutzerdefinierten Datenagenten (dataAgents/DATA_AGENT_ID) abrufen, indem Sie den Ressourcennamen des Agenten in Ihrer Anfrage ändern:

Python SDK

from google.cloud import geminidataanalytics_v1

client = geminidataanalytics_v1.DataA2AServiceClient()

agent_name = "projects/PROJECT_ID/locations/LOCATION/agents/bigquery-ca"

request = geminidataanalytics_v1.GetAgentCardRequest(tenant=agent_name)
card = client.get_agent_card(request=request)

print(card)

Ersetzen Sie im vorherigen Beispiel die Werte so:

  • PROJECT_ID: die ID Ihres Google Cloud -Projekts
  • LOCATION: Der Standort der Agent-Ressource, z. B. us, us-east4, eu oder global.

HTTP

curl -X GET \
  -H "Authorization: Bearer $(gcloud auth application-default print-access-token)" \
  "https://geminidataanalytics.googleapis.com/v1/a2a/projects/PROJECT_ID/locations/LOCATION/agents/bigquery-ca/v1/card"

Ersetzen Sie im vorherigen Beispiel die Werte so:

  • PROJECT_ID: die ID Ihres Google Cloud -Projekts
  • LOCATION: Der Standort der Agent-Ressource, z. B. us, us-east4, eu oder global.

Struktur der Agent-Karte

Bei einer erfolgreichen Anfrage wird ein Agent-Kartenobjekt mit Metadaten, unterstützten Skills und Erweiterungen zurückgegeben:

{
  "name": "BigQuery Conversational Analytics Agent",
  "description": "This agent can answer questions about your data using BigQuery.",
  "protocolVersion": "1.0",
  "skills": [
    {
      "id": "data-analysis",
      "name": "Data Analysis",
      "description": "Provides data analysis assistance",
      "examples": [
        "What is the total sales for the last 3 months?"
      ],
      "inputModes": [
        "text/plain"
      ],
      "outputModes": [
        "text/plain",
        "application/json"
      ]
    }
  ],
  "capabilities": {
    "streaming": true,
    "extensions": [
      {
        "uri": "https://docs.cloud.google.com/gemini/docs/conversational-analytics-api/reference/a2a/extensions/bigquery_context/v1",
        "description": "Google Data Analytics BigQuery Context extension"
      }
    ]
  },
  "defaultInputModes": [
    "text/plain"
  ],
  "defaultOutputModes": [
    "text/plain",
    "application/json"
  ]
}

Die Agent-Karte enthält die folgenden Felder:

  • name: der Anzeigename des KI-Datenagenten
  • description: eine Zusammenfassung der Analysefunktionen des KI-Datenagenten
  • protocolVersion: Die Version des A2A-Protokolls, die vom Endpunkt unterstützt wird (z. B. 1.0).
  • skills: die Aufgaben, die der KI-Datenagent ausführen kann, einschließlich Beispielprompts (examples) und unterstützter Datenformate (inputModes und outputModes, z. B. text/plain oder application/json)
  • capabilities.streaming: ein boolescher Wert, der angibt, ob der KI-Datenagent Echtzeit-Streaming über die stream-Methode unterstützt.
  • capabilities.extensions: Die A2A-Erweiterungen, die der KI-Datenagent unterstützt, z. B. bigquery_context/v1, stateless/v1 und kms/v1
  • defaultInputModes und defaultOutputModes: die Standarddatenformate (z. B. text/plain oder application/json) für Anfrage- und Antwortnutzlasten

Nachricht an einen KI-Datenagenten senden

Verwenden Sie die Methode send, um eine Nachricht an einen Daten-KI-Agenten zu senden. Der Daten-KI-Agent verarbeitet die Anfrage, generiert und führt die erforderlichen SQL-Abfragen für Ihre Daten aus und gibt die Antwort in natürlicher Sprache zusammen mit den generierten Datenartefakten zurück.

Nachricht senden

Wenn Sie eine Nachricht senden, geben Sie den Zieldatenagenten im Ressourcenpfad an:

  • Übergeben Sie für integrierte Daten-KI-Agenten (agents/bigquery-ca oder agents/looker-ca) Zieltabellen- oder Explore-Referenzen im Feld metadata mit der Erweiterung bigquery_context/v1 oder looker_context/v1.
  • Lassen Sie bei benutzerdefinierten KI-Datenagenten (dataAgents/DATA_AGENT_ID) das Feld metadata weg, da Kontext, Schemas und Anweisungen direkt in der Agent-Ressource konfiguriert werden.

Wenn Sie Anfragen verarbeiten möchten, ohne den Unterhaltungsverlauf in Google Cloudzu speichern, fügen Sie die Erweiterung stateless/v1 in Ihre Anfrage ein. Wenn Sie gespeicherte Unterhaltungsdaten und Metadaten mit einem kundenverwalteten Verschlüsselungsschlüssel verschlüsseln möchten, übergeben Sie die kms/v1-Erweiterung mit dem Namen Ihres Cloud KMS-Schlüssels. Weitere Informationen finden Sie unter Vom Kunden verwaltete Verschlüsselungsschlüssel (CMEK).

Die folgenden Codebeispiele zeigen, wie Sie eine Nachricht an den integrierten BigQuery-Datenagenten senden:

Python SDK

from google.cloud import geminidataanalytics_v1

client = geminidataanalytics_v1.DataA2AServiceClient()

agent_name = "projects/PROJECT_ID/locations/LOCATION/agents/bigquery-ca"

request = geminidataanalytics_v1.SendMessageRequest(
    tenant=agent_name,
    message=geminidataanalytics_v1.Message(
        role="ROLE_USER",
        # Optional: Pass context_id to continue an existing conversation
        # context_id="projects/PROJECT_ID/locations/LOCATION/conversations/CONVERSATION_ID",
        parts=[
            geminidataanalytics_v1.Part(
                text="What are the top 5 countries where our users are located?"
            )
        ],
    ),
    configuration=geminidataanalytics_v1.SendMessageConfiguration(
        return_immediately=False
    ),
    metadata={
        "https://docs.cloud.google.com/gemini/docs/conversational-analytics-api/reference/a2a/extensions/bigquery_context/v1": {
            "datasource_references": {
                "bq": {
                    "tableReferences": [
                        {
                            "projectId": "DATASET_PROJECT_ID",
                            "datasetId": "DATASET_ID",
                            "tableId": "TABLE_ID",
                        }
                    ]
                }
            }
        },
        # Optional: Process queries without storing conversation history in Google Cloud
        # "https://docs.cloud.google.com/gemini/docs/conversational-analytics-api/reference/a2a/extensions/stateless/v1": {},
        # Optional: Encrypt conversation history and metadata with a customer-managed encryption key (CMEK)
        # "https://docs.cloud.google.com/gemini/docs/conversational-analytics-api/reference/a2a/extensions/kms/v1": {
        #     "kmsKey": "projects/PROJECT_ID/locations/LOCATION/keyRings/KEY_RING/cryptoKeys/KEY_NAME"
        # },
    },
)

response = client.send_message(request=request)

print(response)

Ersetzen Sie im vorherigen Beispiel die Werte so:

  • PROJECT_ID: die ID Ihres Google Cloud -Projekts
  • LOCATION: Der Standort der Agent-Ressource, z. B. us, us-east4, eu oder global.
  • CONVERSATION_ID: (Optional) Die ID einer vorhandenen Unterhaltungssitzung, die fortgesetzt werden soll.
  • What are the top 5 countries where our users are located?: die Frage in natürlicher Sprache, die an den KI-Datenagenten gestellt werden soll
  • DATASET_PROJECT_ID: die ID des Projekts in Google Cloud , das das BigQuery-Dataset enthält (z. B. bigquery-public-data)
  • DATASET_ID: die ID des BigQuery-Datasets (z. B. thelook_ecommerce)
  • TABLE_ID: die ID der BigQuery-Tabelle (z. B. users)
  • KEY_RING: (Optional) Der Name des Cloud KMS-Schlüsselbunds bei Verwendung von CMEK
  • KEY_NAME: (Optional) Der Name des Cloud KMS-Kryptoschlüssels bei Verwendung von CMEK

HTTP

curl -X POST \
  -H "Authorization: Bearer $(gcloud auth application-default print-access-token)" \
  -H "Content-Type: application/json; charset=utf-8" \
  -H "A2A-Extensions: https://docs.cloud.google.com/gemini/docs/conversational-analytics-api/reference/a2a/extensions/bigquery_context/v1" \
  -d '{
    "message": {
      "role": "ROLE_USER",
      "parts": [
        {
          "text": "What are the top 5 countries where our users are located?"
        }
      ]
    },
    "configuration": {
      "return_immediately": false
    },
    "metadata": {
      "https://docs.cloud.google.com/gemini/docs/conversational-analytics-api/reference/a2a/extensions/bigquery_context/v1": {
        "datasource_references": {
          "bq": {
            "tableReferences": [
              {
                "projectId": "DATASET_PROJECT_ID",
                "datasetId": "DATASET_ID",
                "tableId": "TABLE_ID"
              }
            ]
          }
        }
      }
    }
  }' \
  "https://geminidataanalytics.googleapis.com/v1/a2a/projects/PROJECT_ID/locations/LOCATION/agents/bigquery-ca/v1/message:send"

Ersetzen Sie im vorherigen Beispiel die Werte so:

  • PROJECT_ID: die ID Ihres Google Cloud -Projekts
  • LOCATION: Der Standort der Agent-Ressource, z. B. us, us-east4, eu oder global.
  • What are the top 5 countries where our users are located?: die Frage in natürlicher Sprache, die an den KI-Datenagenten gestellt werden soll
  • DATASET_PROJECT_ID: die ID des Projekts in Google Cloud , das das BigQuery-Dataset enthält (z. B. bigquery-public-data)
  • DATASET_ID: die ID des BigQuery-Datasets (z. B. thelook_ecommerce)
  • TABLE_ID: die ID der BigQuery-Tabelle (z. B. users)

Antwortstruktur verstehen

Bei einer erfolgreichen Anfrage wird ein task-Objekt zurückgegeben, das den endgültigen Status, die Konversations-ID und generierte Artefakte enthält:

{
  "task": {
    "id": "ab12d1f2-e170-4c4f-aff4-be466c6beaaa",
    "contextId": "projects/my-project/locations/us/conversations/conv-67890",
    "status": {
      "state": "TASK_STATE_COMPLETED"
    },
    "artifacts": [
      {
        "artifactId": "synthetic-8a368e8e-378a-4b80-9f07-0b1a397c221d",
        "name": "Final response",
        "description": "Final response from the agent.",
        "parts": [
          {
            "text": "The top 5 countries where our users are located are China (33,783), the United States (22,701), Brasil (14,620), South Korea (5,302), and France (4,645)."
          }
        ]
      },
      {
        "artifactId": "synthetic-50483808-b231-4b30-a859-2c30d0355a8d",
        "name": "Generated SQL",
        "description": "Generated SQL from the agent.",
        "parts": [
          {
            "text": "SELECT country, COUNT(DISTINCT id) AS user_count FROM `bigquery-public-data.thelook_ecommerce.users` GROUP BY country ORDER BY user_count DESC LIMIT 5",
            "mediaType": "text/x-sql"
          }
        ]
      }
    ]
  }
}

Die Antwort umfasst die folgenden Schlüsselfelder:

  • task.id: die eindeutige Kennung für die Ausführungsaufgabe
  • task.contextId: Der Ressourcenpfad der Unterhaltung, den Sie in nachfolgenden Anfragen im Feld message.contextId übergeben, um die Sitzung fortzusetzen.
  • task.status.state: Der Ausführungsstatus der Aufgabe, z. B. TASK_STATE_COMPLETED.
  • task.artifacts[]: Strukturierte Assets, die vom Daten-KI-Agenten generiert werden, z. B. die Antwort in natürlicher Sprache (Final response), die ausführbare SQL-Abfrage (Generated SQL) und die tabellarischen Ergebniszeilen (Data result)

Antworten von einem Daten-Agent streamen

Wenn Sie Echtzeit-Updates erhalten möchten, während der Daten-KI-Agent eine Anfrage bearbeitet, verwenden Sie die Methode stream. Die Antwort enthält Statusaktualisierungen (status_update) mit Zwischenüberlegungen und inkrementellen Artefakten (artifact_update), z. B. generierte SQL-Abfragen und Vega-Lite-Diagrammspezifikationen.

Streaminganfragen unterstützen auch fortlaufende Unterhaltungen (context_id), die zustandslose Verarbeitung (stateless/v1) und vom Kunden verwaltete Verschlüsselungsschlüssel (kms/v1).

Streamingnachricht senden

In den folgenden Codebeispielen wird gezeigt, wie Sie Ereignisse vom integrierten BigQuery-Datenagenten streamen. Wenn Sie Daten von einem benutzerdefinierten Daten-KI-Agenten (dataAgents/DATA_AGENT_ID) streamen möchten, geben Sie den Ressourcenpfad des benutzerdefinierten Agenten an und lassen Sie das Feld metadata weg:

Python SDK

from google.cloud import geminidataanalytics_v1

client = geminidataanalytics_v1.DataA2AServiceClient()

agent_name = "projects/PROJECT_ID/locations/LOCATION/agents/bigquery-ca"

request = geminidataanalytics_v1.SendMessageRequest(
    tenant=agent_name,
    message=geminidataanalytics_v1.Message(
        role="ROLE_USER",
        parts=[
            geminidataanalytics_v1.Part(
                text="What are the top 5 countries where our users are located? Please show a pie chart."
            )
        ],
    ),
    metadata={
        "https://docs.cloud.google.com/gemini/docs/conversational-analytics-api/reference/a2a/extensions/bigquery_context/v1": {
            "datasource_references": {
                "bq": {
                    "tableReferences": [
                        {
                            "projectId": "DATASET_PROJECT_ID",
                            "datasetId": "DATASET_ID",
                            "tableId": "TABLE_ID",
                        }
                    ]
                }
            }
        },
        # Optional: Process queries without storing conversation history in Google Cloud
        # "https://docs.cloud.google.com/gemini/docs/conversational-analytics-api/reference/a2a/extensions/stateless/v1": {},
        # Optional: Encrypt conversation history and metadata with a customer-managed encryption key (CMEK)
        # "https://docs.cloud.google.com/gemini/docs/conversational-analytics-api/reference/a2a/extensions/kms/v1": {
        #     "kmsKey": "projects/PROJECT_ID/locations/LOCATION/keyRings/KEY_RING/cryptoKeys/KEY_NAME"
        # },
    },
)

# Stream response events
stream = client.send_streaming_message(request=request)

for chunk in stream:
  print(chunk)

Ersetzen Sie im vorherigen Beispiel die Werte so:

  • PROJECT_ID: die ID Ihres Google Cloud -Projekts
  • LOCATION: Der Standort der Agent-Ressource, z. B. us, us-east4, eu oder global.
  • What are the top 5 countries where our users are located? Please show a pie chart.: die Frage in natürlicher Sprache, die an den KI-Datenagenten gestellt werden soll
  • DATASET_PROJECT_ID: die ID des Projekts in Google Cloud , das das BigQuery-Dataset enthält (z. B. bigquery-public-data)
  • DATASET_ID: die ID des BigQuery-Datasets (z. B. thelook_ecommerce)
  • TABLE_ID: die ID der BigQuery-Tabelle (z. B. users)
  • KEY_RING: (Optional) Der Name des Cloud KMS-Schlüsselbunds bei Verwendung von CMEK
  • KEY_NAME: (Optional) Der Name des Cloud KMS-Kryptoschlüssels bei Verwendung von CMEK

HTTP

curl -X POST \
  -N \
  -H "Authorization: Bearer $(gcloud auth application-default print-access-token)" \
  -H "Content-Type: application/json; charset=utf-8" \
  -H "Accept: text/event-stream, application/json" \
  -H "A2A-Extensions: https://docs.cloud.google.com/gemini/docs/conversational-analytics-api/reference/a2a/extensions/bigquery_context/v1" \
  -d '{
    "message": {
      "role": "ROLE_USER",
      "parts": [
        {
          "text": "What are the top 5 countries where our users are located? Please show a pie chart."
        }
      ]
    },
    "metadata": {
      "https://docs.cloud.google.com/gemini/docs/conversational-analytics-api/reference/a2a/extensions/bigquery_context/v1": {
        "datasource_references": {
          "bq": {
            "tableReferences": [
              {
                "projectId": "DATASET_PROJECT_ID",
                "datasetId": "DATASET_ID",
                "tableId": "TABLE_ID"
              }
            ]
          }
        }
      }
    }
  }' \
  "https://geminidataanalytics.googleapis.com/v1/a2a/projects/PROJECT_ID/locations/LOCATION/agents/bigquery-ca/v1/message:stream"

Ersetzen Sie im vorherigen Beispiel die Werte so:

  • PROJECT_ID: die ID Ihres Google Cloud -Projekts
  • LOCATION: Der Standort der Agent-Ressource, z. B. us, us-east4, eu oder global.
  • What are the top 5 countries where our users are located? Please show a pie chart.: die Frage in natürlicher Sprache, die an den KI-Datenagenten gestellt werden soll
  • DATASET_PROJECT_ID: die ID des Projekts in Google Cloud , das das BigQuery-Dataset enthält (z. B. bigquery-public-data)
  • DATASET_ID: die ID des BigQuery-Datasets (z. B. thelook_ecommerce)
  • TABLE_ID: die ID der BigQuery-Tabelle (z. B. users)

Struktur der Streaming-Antwort

Wenn Sie eine Streaminganfrage senden, gibt der Server einen Stream von Ereignisobjekten (StreamResponse) zurück. Jedes Ereignis enthält entweder eine Statusaktualisierung oder eine Artefaktaktualisierung.

Statusaktualisierungsereignisse liefern Zwischenbenachrichtigungen zum Fortschritt und Gedankenmeldungen, während der Datenagent Ihre Anfrage bearbeitet:

{
  "statusUpdate": {
    "taskId": "f41cd8e3-e665-460c-aceb-7b337f1848ef",
    "status": {
      "state": "TASK_STATE_WORKING",
      "message": {
        "role": "ROLE_AGENT",
        "parts": [
          {
            "text": "Analyzing context"
          },
          {
            "text": "Retrieved context for 1 table."
          }
        ]
      }
    }
  }
}

Bei Ereignissen zur Aktualisierung von Artefakten werden strukturierte Ausgabeartefakte wie ausführbare SQL-Abfragen, Antworten in natürlicher Sprache oder Diagrammspezifikationen bereitgestellt:

{
  "artifactUpdate": {
    "taskId": "f41cd8e3-e665-460c-aceb-7b337f1848ef",
    "artifact": {
      "artifactId": "synthetic-7ca98286-0a15-4ca0-a8bc-f14dc231b3ba",
      "name": "Chart result",
      "description": "Chart visualization generated by the data agent.",
      "parts": [
        {
          "data": {
            "title": "Top 5 Countries by User Population",
            "mark": "arc",
            "encoding": {
              "color": {
                "field": "country",
                "type": "nominal"
              },
              "theta": {
                "field": "user_count",
                "type": "quantitative"
              }
            },
            "data": {
              "values": [
                {
                  "country": "China",
                  "user_count": 33783
                },
                {
                  "country": "United States",
                  "user_count": 22701
                },
                {
                  "country": "Brasil",
                  "user_count": 14620
                },
                {
                  "country": "South Korea",
                  "user_count": 5302
                },
                {
                  "country": "France",
                  "user_count": 4645
                }
              ]
            }
          }
        }
      }
    },
    "lastChunk": true
  }
}

Die Streaming-Antwort enthält die folgenden Schlüsselfelder:

  • statusUpdate.status.state: Der Zwischen- oder Endstatus der Aufgabe, z. B. TASK_STATE_WORKING oder TASK_STATE_COMPLETED.
  • statusUpdate.status.message.parts[]: Gedanken- oder Fortschrittsbeschreibungen, die während der Ausführung ausgegeben werden
  • artifactUpdate.artifact: Das strukturierte Asset, das vom Daten-Agent generiert wird, z. B. eine Vega-Lite-Diagrammspezifikation (data) oder eine SQL-Abfrage (text)
  • artifactUpdate.lastChunk: Ein boolesches Flag, das angibt, ob der Artefaktstream vollständig ist.

Informationen zum Rendern der zurückgegebenen Vega- oder Vega-Lite-Spezifikation in Python- oder Frontend-Anwendungen finden Sie unter Antwort eines KI-Agenten als Visualisierung rendern.

Nächste Schritte