Inizia a utilizzare l'API Live con WebSocket

Questo tutorial mostra come connettersi all'API Live utilizzando i WebSocket. In questo tutorial configurerai un progetto Google Cloud per utilizzare l'API Live con WebSocket, invierai un file audio al modello e riceverai l'audio in risposta.

Per maggiori informazioni su WebSocket, consulta la documentazione dell'API WebSocket.

Prima di iniziare

Prima di poter inviare richieste, devi configurare l'autenticazione con Vertex AI. Puoi configurare l'autenticazione utilizzando una chiave API o le credenziali predefinite dell'applicazione (ADC).

Per questo tutorial, il modo più rapido per iniziare è utilizzare una chiave API:

Per istruzioni sulla configurazione dell'autenticazione tramite ADC, consulta la nostra guida rapida.

Installa la libreria WebSocket

Esegui questo comando per installare la libreria websockets:

pip install websockets

Imposta le variabili di ambiente

Imposta le variabili di ambiente per l'ID progetto e la località. Sostituisci PROJECT_ID con l'Google Cloud ID progetto.

export GOOGLE_CLOUD_PROJECT=PROJECT_ID
export GOOGLE_CLOUD_LOCATION=global

Avviare una sessione audio

L'esempio seguente stabilisce una sessione, riproduce in streaming l'audio da un file e stampa le dimensioni dei blocchi audio ricevuti nella risposta.

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

Passaggi successivi