Orchestrer des agents de données avec A2A

L'API Conversational Analytics dans Google Cloud implémente le protocole A2A (Agent-to-Agent) ouvert, qui permet aux agents des workflows multi-agents de découvrir des fonctionnalités, de déléguer des requêtes analytiques et de diffuser des réponses structurées, telles que des requêtes SQL exécutables et des visualisations de graphiques.

Vous pouvez interroger les agents de données intégrés à l'API Conversational Analytics pour BigQuery et Looker en transmettant le contexte de l'ensemble de données dans votre requête API. Vous pouvez également interroger des agents de données personnalisés configurés avec la logique métier de votre organisation.

Pour créer des agents de données et les interroger directement, consultez Créer un agent de données à l'aide du SDK Python ou Créer un agent de données à l'aide de HTTP.

Découvrez comment et quand Gemini pour Google Cloud utilise vos données.

Fonctionnement de l'orchestration des agents de données

Lorsque vous intégrez des agents de données de l'API Conversational Analytics dans une application ou un système multi-agent, le workflow d'orchestration suit les opérations suivantes :

Avant de commencer

Avant de commencer, remplissez les conditions préalables suivantes :

  1. Activez l'API Conversational Analytics, l'API BigQuery et l'API Looker dans votre projet Google Cloud .
  2. Vérifiez que vous disposez des rôles et autorisations IAM requis.
  3. Authentifiez-vous auprès de l'API Conversational Analytics, puis installez la bibliothèque cliente ou obtenez un jeton d'autorisation.

Rôles requis

Pour obtenir les autorisations nécessaires pour découvrir et interroger des agents de données via A2A, demandez à votre administrateur de vous accorder les rôles IAM suivants sur votre projet :

Pour en savoir plus sur l'attribution de rôles, consultez Gérer l'accès aux projets, aux dossiers et aux organisations.

Vous pouvez également obtenir les autorisations requises avec des rôles personnalisés ou d'autres rôles prédéfinis.

Pour interroger les sources de données sous-jacentes, vous devez également disposer d'autorisations de lecture sur vos ensembles de données BigQuery cibles (tels que roles/bigquery.dataViewer) ou vos explorations Looker.

Découvrir les capacités des agents

Avant de déléguer des requêtes à un agent de données, un agent orchestrateur ou une application cliente peut inspecter la fiche de l'agent pour afficher ses capacités et sa configuration, comme sa description, ses compétences disponibles et les extensions compatibles. Vous pouvez récupérer des fiches d'agent à l'aide de la méthode getCard pour les agents de données intégrés (agents/bigquery-ca et agents/looker-ca) et les agents de données personnalisés (dataAgents/DATA_AGENT_ID).

Récupérer une carte d'agent

Les exemples de code suivants montrent comment récupérer une fiche d'agent. Ces exemples utilisent l'agent de données BigQuery intégré (agents/bigquery-ca) comme exemple, mais vous pouvez récupérer la fiche de l'agent Looker intégré (agents/looker-ca) ou d'un agent de données personnalisé (dataAgents/DATA_AGENT_ID) en modifiant le nom de ressource de l'agent dans votre requête :

SDK Python

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)

Dans l'exemple précédent, remplacez les valeurs comme suit :

  • PROJECT_ID : ID de votre projet Google Cloud
  • LOCATION : emplacement de la ressource de l'agent (par exemple, us, us-east4, eu ou 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"

Dans l'exemple précédent, remplacez les valeurs comme suit :

  • PROJECT_ID : ID de votre projet Google Cloud
  • LOCATION : emplacement de la ressource de l'agent (par exemple, us, us-east4, eu ou global)

Comprendre la structure de la fiche d'agent

Une requête réussie renvoie un objet de carte d'agent contenant des métadonnées, des compétences et des extensions compatibles :

{
  "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"
  ]
}

La fiche de l'agent comprend les champs suivants :

  • name : nom à afficher de l'agent de données.
  • description : résumé des capacités analytiques de l'agent de données
  • protocolVersion : version du protocole A2A compatible avec le point de terminaison (par exemple, 1.0)
  • skills : tâches que l'agent de données peut effectuer, y compris des exemples de requêtes (examples) et des formats de données acceptés (inputModes et outputModes, tels que text/plain ou application/json)
  • capabilities.streaming : valeur booléenne indiquant si l'agent de données est compatible avec le streaming en temps réel via la méthode stream
  • capabilities.extensions : extensions A2A compatibles avec l'agent de données, telles que bigquery_context/v1, stateless/v1 et kms/v1
  • defaultInputModes et defaultOutputModes : formats de données par défaut (tels que text/plain ou application/json) pour les charges utiles des requêtes et des réponses

Envoyer un message à un agent de données

Pour envoyer un message à un agent de données, utilisez la méthode send. L'agent de données traite la demande, génère et exécute les requêtes SQL requises sur vos données, puis renvoie la réponse en langage naturel ainsi que les artefacts de données générés.

Envoyer un message

Lorsque vous envoyez un message, spécifiez l'agent de données cible dans le chemin d'accès à la ressource :

  • Pour les agents de données intégrés (agents/bigquery-ca ou agents/looker-ca), transmettez les références de la table cible ou de l'exploration dans le champ metadata à l'aide de l'extension bigquery_context/v1 ou looker_context/v1.
  • Pour les agents de données personnalisés (dataAgents/DATA_AGENT_ID), omettez le champ metadata, car le contexte, les schémas et les instructions sont configurés directement sur la ressource de l'agent.

Pour traiter les requêtes sans stocker l'historique des conversations dans Google Cloud, incluez l'extension stateless/v1 dans votre requête. Pour chiffrer les métadonnées et les données de conversation stockées à l'aide d'une clé de chiffrement gérée par le client, transmettez l'extension kms/v1 avec le nom de votre clé Cloud KMS. Pour en savoir plus, consultez Clés de chiffrement gérées par le client (CMEK).

Les exemples de code suivants montrent comment envoyer un message à l'agent de données BigQuery intégré :

SDK Python

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)

Dans l'exemple précédent, remplacez les valeurs comme suit :

  • PROJECT_ID : ID de votre projet Google Cloud
  • LOCATION : emplacement de la ressource de l'agent (par exemple, us, us-east4, eu ou global)
  • CONVERSATION_ID : (facultatif) ID d'une session de conversation existante à poursuivre.
  • What are the top 5 countries where our users are located? : question en langage naturel à poser à l'agent de données
  • DATASET_PROJECT_ID : ID du projet Google Cloud contenant l'ensemble de données BigQuery (par exemple, bigquery-public-data)
  • DATASET_ID : ID de l'ensemble de données BigQuery (par exemple, thelook_ecommerce)
  • TABLE_ID : ID de la table BigQuery (par exemple, users)
  • KEY_RING : (facultatif) nom du trousseau de clés Cloud KMS lorsque vous utilisez CMEK
  • KEY_NAME : nom de la clé cryptographique Cloud KMS lorsque vous utilisez CMEK (facultatif).

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"

Dans l'exemple précédent, remplacez les valeurs comme suit :

  • PROJECT_ID : ID de votre projet Google Cloud
  • LOCATION : emplacement de la ressource de l'agent (par exemple, us, us-east4, eu ou global)
  • What are the top 5 countries where our users are located? : question en langage naturel à poser à l'agent de données
  • DATASET_PROJECT_ID : ID du projet Google Cloud contenant l'ensemble de données BigQuery (par exemple, bigquery-public-data)
  • DATASET_ID : ID de l'ensemble de données BigQuery (par exemple, thelook_ecommerce)
  • TABLE_ID : ID de la table BigQuery (par exemple, users)

Comprendre la structure de la réponse

Une requête réussie renvoie un objet task contenant l'état final, l'identifiant de la conversation et les artefacts générés :

{
  "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"
          }
        ]
      }
    ]
  }
}

La réponse inclut les champs clés suivants :

  • task.id : identifiant unique de la tâche d'exécution.
  • task.contextId : chemin d'accès à la ressource de la conversation, que vous transmettez dans le champ message.contextId des requêtes suivantes pour poursuivre la session
  • task.status.state : état d'exécution de la tâche (par exemple, TASK_STATE_COMPLETED)
  • task.artifacts[] : composants structurés générés par l'agent de données, tels que la réponse en langage naturel (Final response), la requête SQL exécutable (Generated SQL) et les lignes de résultats tabulaires (Data result).

Diffuser des réponses à partir d'un agent de données

Pour recevoir des mises à jour en temps réel à mesure que l'agent de données traite une requête, utilisez la méthode stream. La réponse diffuse des mises à jour d'état (status_update) avec des réflexions intermédiaires et des artefacts incrémentaux (artifact_update), tels que des requêtes SQL générées et des spécifications de graphiques Vega-Lite.

Les requêtes de streaming sont également compatibles avec les conversations continues (context_id), le traitement sans état (stateless/v1) et les clés de chiffrement gérées par le client (kms/v1).

Envoyer un message de streaming

Les exemples de code suivants montrent comment diffuser des événements à partir de l'agent de données BigQuery intégré. Pour diffuser des données à partir d'un agent de données personnalisé (dataAgents/DATA_AGENT_ID), ciblez le chemin d'accès à la ressource de l'agent personnalisé et omettez le champ metadata :

SDK Python

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)

Dans l'exemple précédent, remplacez les valeurs comme suit :

  • PROJECT_ID : ID de votre projet Google Cloud
  • LOCATION : emplacement de la ressource de l'agent (par exemple, us, us-east4, eu ou global)
  • What are the top 5 countries where our users are located? Please show a pie chart. : question en langage naturel à poser à l'agent de données
  • DATASET_PROJECT_ID : ID du projet Google Cloud contenant l'ensemble de données BigQuery (par exemple, bigquery-public-data)
  • DATASET_ID : ID de l'ensemble de données BigQuery (par exemple, thelook_ecommerce)
  • TABLE_ID : ID de la table BigQuery (par exemple, users)
  • KEY_RING : (facultatif) nom du trousseau de clés Cloud KMS lorsque vous utilisez CMEK
  • KEY_NAME : nom de la clé cryptographique Cloud KMS lorsque vous utilisez CMEK (facultatif).

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"

Dans l'exemple précédent, remplacez les valeurs comme suit :

  • PROJECT_ID : ID de votre projet Google Cloud
  • LOCATION : emplacement de la ressource de l'agent (par exemple, us, us-east4, eu ou global)
  • What are the top 5 countries where our users are located? Please show a pie chart. : question en langage naturel à poser à l'agent de données
  • DATASET_PROJECT_ID : ID du projet Google Cloud contenant l'ensemble de données BigQuery (par exemple, bigquery-public-data)
  • DATASET_ID : ID de l'ensemble de données BigQuery (par exemple, thelook_ecommerce)
  • TABLE_ID : ID de la table BigQuery (par exemple, users)

Comprendre la structure de la réponse de streaming

Lorsque vous envoyez une requête de flux, le serveur renvoie un flux d'objets d'événement (StreamResponse). Chaque événement contient une mise à jour de l'état ou une mise à jour d'artefact.

Les événements de mise à jour de l'état fournissent des notifications de progression intermédiaires et des messages de réflexion à mesure que l'agent de données traite votre requête :

{
  "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."
          }
        ]
      }
    }
  }
}

Les événements de mise à jour des artefacts fournissent des objets de sortie structurés, tels que des requêtes SQL exécutables, des réponses en langage naturel ou des spécifications de graphiques :

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

La réponse de streaming inclut les champs clés suivants :

  • statusUpdate.status.state : état intermédiaire ou final de la tâche (par exemple, TASK_STATE_WORKING ou TASK_STATE_COMPLETED)
  • statusUpdate.status.message.parts[] : descriptions de la réflexion ou de la progression émises lors de l'exécution
  • artifactUpdate.artifact : élément structuré généré par l'agent de données, tel qu'une spécification de graphique Vega-Lite (data) ou une requête SQL (text)
  • artifactUpdate.lastChunk : indicateur booléen indiquant si le flux d'artefacts est terminé

Pour afficher la spécification Vega ou Vega-Lite renvoyée dans des applications Python ou de frontend, consultez Afficher une réponse d'agent sous forme de visualisation.

Étapes suivantes