A2A로 데이터 에이전트 조정

Conversational Analytics API는 Google Cloud 개방형 에이전트 간 (A2A) 프로토콜을 구현하여 멀티 에이전트 워크플로의 에이전트가 기능을 검색하고, 분석 쿼리를 위임하고, 실행 가능한 SQL 쿼리 및 차트 시각화와 같은 구조화된 응답을 스트리밍할 수 있습니다.

API 요청에 데이터 세트 컨텍스트를 전달하여 BigQuery 및 Looker용 Conversational Analytics API에 내장된 데이터 에이전트를 쿼리하거나 조직의 비즈니스 로직으로 구성된 맞춤 데이터 에이전트를 쿼리할 수 있습니다.

데이터 에이전트를 직접 빌드하고 쿼리하려면 Python SDK를 사용하여 데이터 에이전트 빌드 또는 HTTP를 사용하여 데이터 에이전트 빌드를 참고하세요.

Google Cloud 를 위한 Gemini에서 사용자 데이터를 사용하는 방법과 시점을 알아보세요.

데이터 에이전트 오케스트레이션 작동 방식

Conversational Analytics API 데이터 에이전트를 애플리케이션 또는 멀티 에이전트 시스템에 통합하면 오케스트레이션 워크플로가 다음 작업을 따릅니다.

  • 오케스트레이터 에이전트는 분석 쿼리를 위임하기 전에 에이전트 카드를 검사하여 데이터 에이전트의 기능과 스킬을 검색합니다.
  • 오케스트레이터 에이전트는 요청에 데이터 소스를 지정하여 내장 데이터 에이전트 (agents/bigquery-ca 또는 agents/looker-ca)를 쿼리하거나 도메인 비즈니스 로직으로 구성된 맞춤 데이터 에이전트 (dataAgents/DATA_AGENT_ID)를 쿼리하는 메시지를 전송합니다.
  • 데이터 에이전트는 요청을 처리하고, 필요한 쿼리를 실행하고, 결과를 반환합니다. 결과는 완전한 응답으로 반환되거나 추론 진행 상황과 구조화된 아티팩트 (예: 실행 가능한 SQL 및 Vega-Lite 차트 사양)를 스트리밍하여 반환됩니다.

시작하기 전에

시작하기 전에 다음의 사건 요건을 완료하세요.

  1. Google Cloud 프로젝트에서 Conversational Analytics API, BigQuery API, Looker API를 사용 설정합니다.
  2. 필요한 IAM 역할과 권한이 있는지 확인합니다.
  3. 대화형 분석 API에 대해 인증하고 클라이언트 라이브러리를 설치하거나 승인 토큰을 획득합니다.

필요한 역할

A2A를 통해 데이터 에이전트를 검색하고 쿼리하는 데 필요한 권한을 얻으려면 관리자에게 프로젝트에 대한 다음 IAM 역할을 부여해 달라고 요청하세요.

역할 부여에 대한 자세한 내용은 프로젝트, 폴더, 조직에 대한 액세스 관리를 참조하세요.

커스텀 역할이나 다른 사전 정의된 역할을 통해 필요한 권한을 얻을 수도 있습니다.

기본 데이터 소스를 쿼리하려면 대상 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).

응답 구조 이해하기

요청이 성공하면 최종 상태, 대화 식별자, 생성된 아티팩트가 포함된 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 메서드를 사용하세요. 대답은 생성된 SQL 쿼리 및 Vega-Lite 차트 사양과 같은 중간 생각과 증분 아티팩트 (artifact_update)를 사용하여 상태 업데이트 (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 또는 프런트엔드 애플리케이션에서 렌더링하려면 에이전트 응답을 시각화로 렌더링을 참고하세요.

다음 단계