Google は AI 技術を使用して、コンテンツをご希望の言語に翻訳しています。AI 翻訳には誤りが含まれる場合があります。

A2A でデータエージェントをオーケストレートする

Google Cloud の Conversational Analytics API は、オープンな Agent-to-Agent（A2A）プロトコルを実装しています。これにより、マルチエージェントワークフローのエージェントは、機能の検出、分析クエリの委任、実行可能な SQL クエリやグラフの可視化などの構造化されたレスポンスのストリーミングを行うことができます。

BigQuery と Looker の Conversational Analytics API に組み込まれているデータエージェントにクエリを実行するには、API リクエストでデータセットコンテキストを渡します。また、組織のビジネスロジックで構成されたカスタムデータエージェントにクエリを実行することもできます。

データエージェントを直接構築してクエリするには、Python SDK を使用してデータエージェントを構築するまたは HTTP を使用してデータエージェントを構築するをご覧ください。

Gemini for Google Cloud がデータを使用する方法とタイミングに関する説明をご覧ください。

データエージェントオーケストレーションの仕組み

会話分析 API データエージェントをアプリケーションまたはマルチエージェントシステムに統合すると、オーケストレーションワークフローは次のオペレーションに従います。

オーケストレーターエージェントは、分析クエリを委任する前にエージェントカードを検査して、データエージェントの機能とスキルを検出します。
オーケストレーターエージェントは、リクエストでデータソースを指定して組み込みデータエージェント（agents/bigquery-ca または agents/looker-ca）をクエリするか、ドメインビジネスロジックで構成されたカスタムデータエージェント（dataAgents/DATA_AGENT_ID）をクエリするメッセージを送信します。
データエージェントはリクエストを処理し、必要なクエリを実行して、結果を完了した回答として返すか、推論の進行状況と構造化されたアーティファクト（実行可能な SQL や Vega-Lite グラフの仕様など）をストリーミングします。

始める前に

始める前に、次の前提条件を満たす必要があります。

Google Cloud プロジェクトで Conversational Analytics API、BigQuery API、Looker API を有効にします。
必要な IAM ロールと権限があることを確認します。
Conversational Analytics API に対して認証を行い、クライアントライブラリをインストールするか、認証トークンを取得します。

必要なロール

A2A 経由でデータエージェントを検出してクエリするために必要な権限を取得するには、プロジェクトに対する次の IAM ロールを付与するよう管理者に依頼してください。

Gemini データ分析 Data Agent User （roles/geminidataanalytics.dataAgentUser）
ステートレスクエリの場合: Gemini データ分析データエージェントステートレスユーザー（roles/geminidataanalytics.dataAgentStatelessUser）

ロールの付与については、プロジェクト、フォルダ、組織へのアクセス権の管理をご覧ください。

必要な権限は、カスタムロールや他の事前定義ロールから取得することもできます。

基盤となるデータソースをクエリするには、ターゲットの BigQuery データセット（roles/bigquery.dataViewer など）または Looker Explore に対する読み取り権限も必要です。

エージェントの機能を確認する

オーケストレーターエージェントまたはクライアントアプリケーションは、クエリをデータエージェントに委任する前に、エージェントカードを調べて、説明、使用可能なスキル、サポートされている拡張機能などの機能と構成を確認できます。エージェントカードは、組み込みデータエージェント（agents/bigquery-ca と agents/looker-ca）とカスタムデータエージェント（dataAgents/DATA_AGENT_ID）の両方で getCard メソッドを使用して取得できます。

エージェントカードを取得する

次のコードサンプルは、エージェントカードを取得する方法を示しています。これらのサンプルでは、組み込みの BigQuery データエージェント（agents/bigquery-ca）を例として使用していますが、リクエストでエージェントリソース名を変更することで、組み込みの Looker エージェント（agents/looker-ca）またはカスタムデータエージェント（dataAgents/DATA_AGENT_ID）のカードを取得できます。

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)

前のサンプルでは、値を次のように置き換えます。

PROJECT_ID: Google Cloud プロジェクトの ID
LOCATION: エージェントリソースのロケーション（us、us-east4、eu、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"

前のサンプルでは、値を次のように置き換えます。

PROJECT_ID: Google Cloud プロジェクトの ID
LOCATION: エージェントリソースのロケーション（us、us-east4、eu、global など）

エージェントカードの構造を理解する

リクエストが成功すると、メタデータ、サポートされているスキル、拡張機能を含むエージェントカードオブジェクトが返されます。

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

エージェントカードには次のフィールドが含まれます。

name: データエージェントの表示名
description: データエージェントの分析機能の概要
protocolVersion: エンドポイントでサポートされている A2A プロトコルのバージョン（1.0 など）
skills: データエージェントが実行できるタスク。サンプルプロンプト（examples）やサポートされているデータ形式（inputModes、outputModes。text/plain、application/json など）が含まれます。
capabilities.streaming: データエージェントが stream メソッドによるリアルタイムストリーミングをサポートしているかどうかを示すブール値
capabilities.extensions: データエージェントがサポートする A2A 拡張機能（bigquery_context/v1、stateless/v1、kms/v1 など）
defaultInputModes と defaultOutputModes: リクエストとレスポンスのペイロードのデフォルトのデータ形式（text/plain や application/json など）

データエージェントにメッセージを送信する

データエージェントにメッセージを送信するには、send メソッドを使用します。データエージェントはリクエストを処理し、データに対して必要な SQL クエリを生成して実行し、生成されたデータアーティファクトとともに自然言語の回答を返します。

メッセージを送信する

メッセージを送信するときに、リソースパスでターゲットデータエージェントを指定します。

組み込みのデータエージェント（agents/bigquery-ca または agents/looker-ca）の場合は、bigquery_context/v1 または looker_context/v1 拡張機能を使用して、metadata フィールドでターゲットテーブルまたは Explore の参照を渡します。
カスタムデータエージェント（dataAgents/DATA_AGENT_ID）の場合、コンテキスト、スキーマ、指示はエージェントリソースで直接構成されるため、metadata フィールドは省略します。

会話履歴を Google Cloudに保存せずにクエリを処理するには、リクエストに stateless/v1 拡張機能を含めます。顧客管理の暗号鍵を使用して保存された会話データとメタデータを暗号化するには、Cloud KMS 鍵名とともに kms/v1 拡張機能を渡します。詳細については、顧客管理の暗号鍵（CMEK）をご覧ください。

次のコードサンプルは、組み込みの BigQuery データエージェントにメッセージを送信する方法を示しています。

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)

前のサンプルでは、値を次のように置き換えます。

PROJECT_ID: Google Cloud プロジェクトの ID
LOCATION: エージェントリソースのロケーション（us、us-east4、eu、global など）
CONVERSATION_ID:（省略可）継続する既存の会話セッションの ID
What are the top 5 countries where our users are located?: データエージェントに尋ねる自然言語の質問
DATASET_PROJECT_ID: BigQuery データセットを含む Google Cloud プロジェクトの ID（例: bigquery-public-data）
DATASET_ID: BigQuery データセットの ID（例: thelook_ecommerce）
TABLE_ID: BigQuery テーブルの ID（例: users）
KEY_RING: （省略可）CMEK を使用する場合の Cloud KMS キーリングの名前
KEY_NAME: （省略可）CMEK を使用する場合の Cloud KMS 暗号鍵の名前

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"

前のサンプルでは、値を次のように置き換えます。

PROJECT_ID: Google Cloud プロジェクトの ID
LOCATION: エージェントリソースのロケーション（us、us-east4、eu、global など）
What are the top 5 countries where our users are located?: データエージェントに尋ねる自然言語の質問
DATASET_PROJECT_ID: BigQuery データセットを含む Google Cloud プロジェクトの ID（例: bigquery-public-data）
DATASET_ID: BigQuery データセットの ID（例: thelook_ecommerce）
TABLE_ID: BigQuery テーブルの ID（例: users）

レスポンスの構造を理解する

リクエストが成功すると、最終ステータス、会話 ID、生成されたアーティファクトを含む task オブジェクトが返されます。

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

レスポンスには次のキーフィールドが含まれます。

task.id: 実行タスクの一意の識別子
task.contextId: 会話のリソースパス。セッションを続行するために、後続のリクエストの message.contextId フィールドで渡します。
task.status.state: タスクの実行状態（TASK_STATE_COMPLETED など）
task.artifacts[]: データエージェントによって生成される構造化されたアセット（自然言語の回答（Final response）、実行可能な SQL クエリ（Generated SQL）、表形式の結果行（Data result）など）

データエージェントからのレスポンスをストリーミングする

データエージェントがクエリを推論する際にリアルタイムの更新を受信するには、stream メソッドを使用します。レスポンスは、中間的な考えと増分アーティファクト（artifact_update）（生成された SQL クエリや Vega-Lite グラフ仕様など）を含むステータス更新（status_update）をストリーミングします。

ストリーミングリクエストは、会話の継続（context_id）、ステートレス処理（stateless/v1）、顧客管理の暗号鍵（kms/v1）もサポートしています。

ストリーミングメッセージを送信する

次のコードサンプルは、組み込みの BigQuery データエージェントからイベントをストリーミングする方法を示しています。カスタムデータエージェント（dataAgents/DATA_AGENT_ID）からストリーミングするには、カスタムエージェントのリソースパスをターゲットにして、metadata フィールドを省略します。

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)

前のサンプルでは、値を次のように置き換えます。

PROJECT_ID: Google Cloud プロジェクトの ID
LOCATION: エージェントリソースのロケーション（us、us-east4、eu、global など）
What are the top 5 countries where our users are located? Please show a pie chart.: データエージェントに尋ねる自然言語の質問
DATASET_PROJECT_ID: BigQuery データセットを含む Google Cloud プロジェクトの ID（例: bigquery-public-data）
DATASET_ID: BigQuery データセットの ID（例: thelook_ecommerce）
TABLE_ID: BigQuery テーブルの ID（例: users）
KEY_RING: （省略可）CMEK を使用する場合の Cloud KMS キーリングの名前
KEY_NAME: （省略可）CMEK を使用する場合の Cloud KMS 暗号鍵の名前

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"

前のサンプルでは、値を次のように置き換えます。

PROJECT_ID: Google Cloud プロジェクトの ID
LOCATION: エージェントリソースのロケーション（us、us-east4、eu、global など）
What are the top 5 countries where our users are located? Please show a pie chart.: データエージェントに尋ねる自然言語の質問
DATASET_PROJECT_ID: BigQuery データセットを含む Google Cloud プロジェクトの ID（例: bigquery-public-data）
DATASET_ID: BigQuery データセットの ID（例: thelook_ecommerce）
TABLE_ID: BigQuery テーブルの ID（例: users）

ストリーミングレスポンスの構造を理解する

ストリーミングリクエストを送信すると、サーバーはイベントオブジェクト（StreamResponse）のストリームを返します。各イベントには、ステータスの更新またはアーティファクトの更新が含まれます。

ステータス更新イベントは、データエージェントがクエリを処理する際に、中間的な進行状況の通知と思考メッセージを配信します。

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

アーティファクト更新イベントは、実行可能な SQL クエリ、自然言語の回答、グラフの仕様などの構造化された出力オブジェクトを配信します。

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

ストリーミングレスポンスには、次のキーフィールドが含まれます。

statusUpdate.status.state: タスクの中間状態または最終状態（TASK_STATE_WORKING や TASK_STATE_COMPLETED など）
statusUpdate.status.message.parts[]: 実行中に生成される思考または進行状況の説明
artifactUpdate.artifact: データエージェントによって生成される構造化アセット（Vega-Lite グラフ仕様（data）や SQL クエリ（text）など）
artifactUpdate.lastChunk: アーティファクトストリームが完了したかどうかを示すブール値のフラグ

返された Vega または Vega-Lite 仕様を Python またはフロントエンドアプリケーションでレンダリングするには、エージェントの回答を可視化としてレンダリングするをご覧ください。

次のステップ

Vega-Lite と Altair を使用してエージェントのレスポンスを可視化としてレンダリングする方法を確認する。
作成されたコンテキストでエージェントの動作をガイドする方法を学習して、ビジネスルールと検証済みクエリを構成します。
マルチエージェントシステムのアーキテクチャ統合パターンを確認する。
Gemini Data Analytics API REST リファレンスを確認します。

A2A でデータ エージェントをオーケストレートする コレクションでコンテンツを整理 必要に応じて、コンテンツの保存と分類を行います。

データ エージェント オーケストレーションの仕組み

始める前に

必要なロール

エージェントの機能を確認する

エージェント カードを取得する

Python SDK

HTTP

エージェント カードの構造を理解する

データ エージェントにメッセージを送信する

メッセージを送信する

Python SDK

HTTP

レスポンスの構造を理解する

データ エージェントからのレスポンスをストリーミングする

ストリーミング メッセージを送信する

Python SDK

HTTP

ストリーミング レスポンスの構造を理解する

次のステップ

A2A でデータエージェントをオーケストレートする

データエージェントオーケストレーションの仕組み

エージェントカードを取得する

エージェントカードの構造を理解する

データエージェントにメッセージを送信する

データエージェントからのレスポンスをストリーミングする

ストリーミングメッセージを送信する

ストリーミングレスポンスの構造を理解する