WebSocket을 사용하여 Live API 시작하기

이 튜토리얼에서는 WebSocket을 사용하여 Live API에 연결하는 방법을 보여줍니다. 이 튜토리얼에서는 WebSocket과 함께 Live API를 사용하도록 Google Cloud 프로젝트를 설정하고, 모델에 오디오 파일을 전송하고, 응답으로 오디오를 수신합니다.

WebSocket에 관한 자세한 내용은 WebSocket API 문서를 참고하세요.

시작하기 전에

요청을 보내려면 Vertex AI로 인증을 설정해야 합니다. API 키를 사용하거나 애플리케이션 기본 사용자 인증 정보 (ADC)를 사용하여 인증을 설정할 수 있습니다.

이 튜토리얼에서는 API 키를 사용하여 가장 빠르게 시작할 수 있습니다.

ADC를 사용하여 인증을 설정하는 방법은 빠른 시작을 참고하세요.

WebSockets 라이브러리 설치

다음을 실행하여 websockets 라이브러리를 설치합니다.

pip install websockets

환경 변수 설정

프로젝트 ID와 위치의 환경 변수를 설정합니다. PROJECT_ID를Google Cloud 프로젝트 ID로 바꿉니다.

export GOOGLE_CLOUD_PROJECT=PROJECT_ID
export GOOGLE_CLOUD_LOCATION=global

오디오 세션 시작

다음 예에서는 세션을 설정하고, 파일에서 오디오를 스트리밍하고, 응답에서 수신된 오디오 청크의 크기를 출력합니다.

import asyncio
import websockets
import json
import base64
import os
import sys

# Replace the [PROJECT_ID] and [LOCATION] with your Google Cloud Project ID and location.
PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT")
LOCATION = os.environ.get("GOOGLE_CLOUD_LOCATION")

# Authentication
token_list = !gcloud auth application-default print-access-token
ACCESS_TOKEN = token_list[0]

# Configuration
MODEL_ID = "gemini-live-2.5-flash-preview-native-audio-09-2025"

# Construct the WSS URL
HOST = f"{LOCATION}-aiplatform.googleapis.com"
path = "google.cloud.aiplatform.v1.LlmBidiService/BidiGenerateContent"
URI = f"wss://{HOST}/ws/{path}"
MODEL_RESOURCE = f"projects/{PROJECT_ID}/locations/{LOCATION}/publishers/google/models/{MODEL_ID}"


async def main():
   headers = {"Authorization": f"Bearer {ACCESS_TOKEN}"}
  
   async with websockets.connect(URI, additional_headers=headers) as ws:
       print("Session established.")

       # Send Setup (Handshake)
       await ws.send(json.dumps({
           "setup": {
               "model": MODEL_RESOURCE,
               "generation_config": { "response_modalities": ["AUDIO"] }
           }
       }))

       # Define Tasks
       async def send_audio():
           # Download sample if missing
           if not os.path.exists("input.wav"):
               !wget -q https://storage.googleapis.com/cloud-samples-data/generative-ai/audio/where_the_nearest_train_station_is.wav -O input.wav

           with open("input.wav", "rb") as f:
               while chunk := f.read(1024):
                   msg = {
                       "realtime_input": {
                           "media_chunks": [{
                               "mime_type": "audio/pcm;rate=16000",
                               "data": base64.b64encode(chunk).decode("utf-8")
                           }]
                       }
                   }
                   await ws.send(json.dumps(msg))
                   await asyncio.sleep(0.01)
           print("Done sending audio.")

       async def receive_audio():
           async for msg in ws:
               data = json.loads(msg)
               try:
                   parts = data["serverContent"]["modelTurn"]["parts"]
                   for part in parts:
                       if "inlineData" in part:
                           b64_audio = part["inlineData"]["data"]
                           print(f"Received chunk: {len(b64_audio)} bytes")
               except KeyError:
                   pass

               if data.get("serverContent", {}).get("turnComplete"):
                   print("Turn complete. Exiting.")
                   break

       # Run Together
       await asyncio.gather(send_audio(), receive_audio())

if __name__ == "__main__":
   # Check if running in Jupyter/Colab
   if "ipykernel" in sys.modules:
       await main()
   else:
       asyncio.run(main())

다음 단계