Orchestrare gli agenti di dati con A2A

L'API Conversational Analytics in Google Cloud implementa il protocollo Agent-to-Agent (A2A), che consente agli agenti nei flussi di lavoro multi-agente di scoprire funzionalità, delegare query analitiche e trasmettere in streaming risposte strutturate, come query SQL eseguibili e visualizzazioni di grafici.

Puoi eseguire query sugli agenti di dati integrati nell'API Conversational Analytics per BigQuery e Looker passando il contesto del set di dati nella richiesta API oppure puoi eseguire query sugli agenti di dati personalizzati configurati con la logica di business della tua organizzazione.

Per creare ed eseguire query sugli agenti dati direttamente, consulta Creare un agente dati utilizzando l'SDK Python o Creare un agente dati utilizzando HTTP.

Scopri come e quando Gemini per Google Cloud utilizza i tuoi dati.

Come funziona l'orchestrazione degli agenti di dati

Quando integri gli agenti di dati dell'API Conversational Analytics in un'applicazione o in un sistema multi-agente, il flusso di lavoro di orchestrazione segue queste operazioni:

Prima di iniziare

Prima di iniziare, completa i seguenti prerequisiti:

  1. Abilita l'API Conversational Analytics, l'API BigQuery e l'API Looker nel tuo progetto Google Cloud .
  2. Verifica di disporre dei ruoli e delle autorizzazioni IAM richiesti.
  3. Autenticati nell'API Conversational Analytics e installa la libreria client o ottieni un token di autorizzazione.

Ruoli obbligatori

Per ottenere le autorizzazioni necessarie per rilevare ed eseguire query sugli agenti di dati tramite A2A, chiedi all'amministratore di concederti i seguenti ruoli IAM sul progetto:

Per saperne di più sulla concessione dei ruoli, consulta Gestisci l'accesso a progetti, cartelle e organizzazioni.

Potresti anche riuscire a ottenere le autorizzazioni richieste tramite i ruoli personalizzati o altri ruoli predefiniti.

Per eseguire query sulle origini dati sottostanti, devi disporre anche delle autorizzazioni di lettura sui set di dati BigQuery di destinazione (ad esempio roles/bigquery.dataViewer) o sulle esplorazioni di Looker.

Scopri le funzionalità degli agenti

Prima di delegare le query a un agente dati, un agente orchestratore o un'applicazione client può esaminare la scheda dell'agente per visualizzarne le funzionalità e la configurazione, ad esempio la descrizione, le competenze disponibili e le estensioni supportate. Puoi recuperare le schede degli agenti utilizzando il metodo getCard sia per gli agenti di dati integrati (agents/bigquery-ca e agents/looker-ca) sia per gli agenti di dati personalizzati (dataAgents/DATA_AGENT_ID).

Recuperare una scheda Agente

I seguenti esempi di codice mostrano come recuperare una scheda dell'agente. Questi esempi utilizzano come esempio l'agente di dati BigQuery integrato (agents/bigquery-ca), ma puoi recuperare la scheda per l'agente Looker integrato (agents/looker-ca) o un agente di dati personalizzato (dataAgents/DATA_AGENT_ID) modificando il nome della risorsa dell'agente nella richiesta:

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)

Nell'esempio precedente, sostituisci i valori nel seguente modo:

  • PROJECT_ID: l'ID del tuo Google Cloud progetto
  • LOCATION: la posizione della risorsa agente (ad esempio us, us-east4, eu o 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"

Nell'esempio precedente, sostituisci i valori nel seguente modo:

  • PROJECT_ID: l'ID del tuo Google Cloud progetto
  • LOCATION: la posizione della risorsa agente (ad esempio us, us-east4, eu o global)

Comprendere la struttura della scheda dell'agente

Una richiesta riuscita restituisce un oggetto scheda dell'agente che contiene metadati, competenze supportate ed estensioni:

{
  "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 scheda dell'agente include i seguenti campi:

  • name: il nome visualizzato dell'agente dati
  • description: un riepilogo delle funzionalità di analisi dell'agente di dati
  • protocolVersion: la versione del protocollo A2A supportata dall'endpoint (ad esempio 1.0)
  • skills: le attività che l'agente dati può svolgere, inclusi prompt di esempio (examples) e formati di dati supportati (inputModes e outputModes, ad esempio text/plain o application/json)
  • capabilities.streaming: un valore booleano che indica se l'agente dati supporta lo streaming in tempo reale tramite il metodo stream
  • capabilities.extensions: le estensioni da app ad app supportate dall'agente dati, ad esempio bigquery_context/v1, stateless/v1 e kms/v1
  • defaultInputModes e defaultOutputModes: i formati di dati predefiniti (ad esempio text/plain o application/json) per i payload di richieste e risposte

Inviare un messaggio a un agente per i dati

Per inviare un messaggio a un agente dei dati, utilizza il metodo send. L'agente dati elabora la richiesta, genera ed esegue le query SQL richieste sui tuoi dati e restituisce la risposta in linguaggio naturale insieme agli artefatti di dati generati.

Invia un messaggio

Quando invii un messaggio, specifica l'agente dati di destinazione nel percorso della risorsa:

  • Per gli agenti dati integrati (agents/bigquery-ca o agents/looker-ca), passa i riferimenti alla tabella di destinazione o a Esplora nel campo metadata utilizzando l'estensione bigquery_context/v1 o looker_context/v1.
  • Per gli agenti di dati personalizzati (dataAgents/DATA_AGENT_ID), ometti il campo metadata perché il contesto, gli schemi e le istruzioni vengono configurati direttamente nella risorsa agente.

Per elaborare le query senza memorizzare la cronologia delle conversazioni in Google Cloud, includi l'estensione stateless/v1 nella richiesta. Per criptare i metadati e i dati delle conversazioni archiviati utilizzando una chiave di crittografia gestita dal cliente, trasmetti l'estensione kms/v1 con il nome della chiave Cloud KMS. Per saperne di più, consulta Chiavi di crittografia gestite dal cliente (CMEK).

I seguenti esempi di codice mostrano come inviare un messaggio all'agente dati BigQuery integrato:

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)

Nell'esempio precedente, sostituisci i valori come segue:

  • PROJECT_ID: l'ID del tuo Google Cloud progetto
  • LOCATION: la posizione della risorsa agente (ad esempio us, us-east4, eu o global)
  • CONVERSATION_ID: (facoltativo) l'ID di una sessione di conversazione esistente da continuare
  • What are the top 5 countries where our users are located?: la domanda in linguaggio naturale da porre all'agente di dati
  • DATASET_PROJECT_ID: l'ID del progetto Google Cloud che contiene il set di dati BigQuery (ad esempio, bigquery-public-data)
  • DATASET_ID: l'ID del set di dati BigQuery (ad esempio, thelook_ecommerce)
  • TABLE_ID: l'ID della tabella BigQuery (ad esempio, users)
  • KEY_RING: (facoltativo) il nome della raccolta di chiavi Cloud KMS quando utilizzi CMEK
  • KEY_NAME: (facoltativo) il nome della chiave di crittografia Cloud KMS quando si utilizza 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"

Nell'esempio precedente, sostituisci i valori come segue:

  • PROJECT_ID: l'ID del tuo Google Cloud progetto
  • LOCATION: la posizione della risorsa agente (ad esempio us, us-east4, eu o global)
  • What are the top 5 countries where our users are located?: la domanda in linguaggio naturale da porre all'agente di dati
  • DATASET_PROJECT_ID: l'ID del progetto Google Cloud che contiene il set di dati BigQuery (ad esempio, bigquery-public-data)
  • DATASET_ID: l'ID del set di dati BigQuery (ad esempio, thelook_ecommerce)
  • TABLE_ID: l'ID della tabella BigQuery (ad esempio, users)

Comprendere la struttura della risposta

Una richiesta riuscita restituisce un oggetto task che contiene lo stato finale, l'identificatore della conversazione e gli artefatti generati:

{
  "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 risposta include i seguenti campi chiave:

  • task.id: l'identificatore univoco dell'attività di esecuzione
  • task.contextId: il percorso della risorsa della conversazione, che viene passato nel campo message.contextId delle richieste successive per continuare la sessione
  • task.status.state: lo stato di esecuzione dell'attività (ad esempio TASK_STATE_COMPLETED)
  • task.artifacts[]: asset strutturati generati dall'agente dati, come la risposta in linguaggio naturale (Final response), la query SQL eseguibile (Generated SQL) e le righe dei risultati tabellari (Data result)

Trasmettere in streaming le risposte di un agente dati

Per ricevere aggiornamenti in tempo reale mentre l'agente dati esamina una query, utilizza il metodo stream. Il flusso di risposta aggiorna lo stato (status_update) con pensieri intermedi e artefatti incrementali (artifact_update), come query SQL generate e specifiche dei grafici Vega-Lite.

Le richieste di streaming supportano anche le conversazioni continue (context_id), l'elaborazione stateless (stateless/v1) e le chiavi di crittografia gestite dal cliente (kms/v1).

Inviare un messaggio di streaming

I seguenti esempi di codice mostrano come trasmettere in streaming gli eventi dall'agente di dati BigQuery integrato. Per eseguire lo streaming da un agente di dati personalizzato (dataAgents/DATA_AGENT_ID), scegli come target il percorso della risorsa dell'agente personalizzato e ometti il campo 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)

Nell'esempio precedente, sostituisci i valori come segue:

  • PROJECT_ID: l'ID del tuo Google Cloud progetto
  • LOCATION: la posizione della risorsa agente (ad esempio us, us-east4, eu o global)
  • What are the top 5 countries where our users are located? Please show a pie chart.: la domanda in linguaggio naturale da porre all'agente dati
  • DATASET_PROJECT_ID: l'ID del progetto Google Cloud che contiene il set di dati BigQuery (ad esempio, bigquery-public-data)
  • DATASET_ID: l'ID del set di dati BigQuery (ad esempio, thelook_ecommerce)
  • TABLE_ID: l'ID della tabella BigQuery (ad esempio, users)
  • KEY_RING: (facoltativo) il nome della raccolta di chiavi Cloud KMS quando utilizzi CMEK
  • KEY_NAME: (facoltativo) il nome della chiave di crittografia Cloud KMS quando si utilizza 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"

Nell'esempio precedente, sostituisci i valori come segue:

  • PROJECT_ID: l'ID del tuo Google Cloud progetto
  • LOCATION: la posizione della risorsa agente (ad esempio us, us-east4, eu o global)
  • What are the top 5 countries where our users are located? Please show a pie chart.: la domanda in linguaggio naturale da porre all'agente di dati
  • DATASET_PROJECT_ID: l'ID del progetto Google Cloud che contiene il set di dati BigQuery (ad esempio, bigquery-public-data)
  • DATASET_ID: l'ID del set di dati BigQuery (ad esempio, thelook_ecommerce)
  • TABLE_ID: l'ID della tabella BigQuery (ad esempio, users)

Informazioni sulla struttura della risposta di streaming

Quando invii una richiesta di streaming, il server restituisce un flusso di oggetti evento (StreamResponse). Ogni evento contiene un aggiornamento dello stato o un aggiornamento dell'artefatto.

Gli eventi di aggiornamento dello stato forniscono notifiche di avanzamento intermedie e messaggi di pensiero man mano che l'agente di dati elabora la query:

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

Gli eventi di aggiornamento degli artefatti forniscono oggetti di output strutturati, come query SQL eseguibili, risposte in linguaggio naturale o specifiche dei grafici:

{
  "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 risposta di streaming include i seguenti campi chiave:

  • statusUpdate.status.state: lo stato intermedio o finale dell'attività (ad esempio TASK_STATE_WORKING o TASK_STATE_COMPLETED)
  • statusUpdate.status.message.parts[]: descrizioni di pensieri o progressi emesse durante l'esecuzione
  • artifactUpdate.artifact: l'asset strutturato generato dall'agente dati, ad esempio una specifica del grafico Vega-Lite (data) o una query SQL (text)
  • artifactUpdate.lastChunk: un flag booleano che indica se il flusso di artefatti è completo

Per eseguire il rendering della specifica Vega o Vega-Lite restituita in Python o nelle applicazioni frontend, consulta Eseguire il rendering della risposta di un agente come visualizzazione.

Passaggi successivi