Mulai menggunakan Live API menggunakan WebSockets

Tutorial ini menunjukkan cara terhubung ke Live API menggunakan WebSockets. Dalam tutorial ini, Anda akan menyiapkan project untuk menggunakan Live API dengan WebSocket, mengirim file audio ke model, dan menerima audio sebagai respons. Google Cloud

Untuk mengetahui informasi selengkapnya tentang WebSocket, lihat dokumentasi WebSockets API.

Sebelum memulai

Sebelum dapat mengirim permintaan, Anda harus menyiapkan autentikasi dengan Vertex AI. Anda dapat menyiapkan autentikasi menggunakan kunci API atau menggunakan kredensial default aplikasi (ADC).

Untuk tutorial ini, cara tercepat untuk memulai adalah dengan menggunakan kunci API:

Untuk mengetahui petunjuk tentang cara menyiapkan autentikasi menggunakan ADC, lihat panduan memulai kami.

Menginstal library WebSockets

Jalankan perintah berikut untuk menginstal library websockets:

pip install websockets

Menyiapkan variabel lingkungan

Tetapkan variabel lingkungan untuk project ID dan lokasi Anda. Ganti PROJECT_ID dengan Google Cloud project ID Anda.

export GOOGLE_CLOUD_PROJECT=PROJECT_ID
export GOOGLE_CLOUD_LOCATION=global

Mulai sesi audio

Contoh berikut membuat sesi, melakukan streaming audio dari file, dan mencetak ukuran potongan audio yang diterima dalam respons.

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())

Langkah berikutnya