Premiers pas avec l'API Live à l'aide de WebSockets

Ce tutoriel vous explique comment vous connecter à l'API Live à l'aide de WebSockets. Dans ce tutoriel, vous allez configurer un projet Google Cloud pour utiliser l'API Live avec les WebSockets, envoyer un fichier audio au modèle et recevoir de l'audio en réponse.

Pour en savoir plus sur WebSockets, consultez la documentation de l'API WebSockets.

Avant de commencer

Avant de pouvoir envoyer des requêtes, vous devez configurer l'authentification avec Vertex AI. Vous pouvez configurer l'authentification à l'aide d'une clé API ou d'identifiants par défaut de l'application (ADC).

Pour ce tutoriel, le moyen le plus rapide de commencer est d'utiliser une clé API :

Pour obtenir des instructions sur la configuration de l'authentification à l'aide d'ADC, consultez notre démarrage rapide.

Installer la bibliothèque WebSockets

Exécutez la commande suivante pour installer la bibliothèque websockets :

pip install websockets

Configurer des variables d'environnement

Définissez des variables d'environnement pour l'ID et l'emplacement de votre projet. Remplacez PROJECT_ID par l'ID du projetGoogle Cloud .

export GOOGLE_CLOUD_PROJECT=PROJECT_ID
export GOOGLE_CLOUD_LOCATION=global

Démarrer une session audio

L'exemple suivant établit une session, diffuse l'audio d'un fichier et affiche la taille des blocs audio reçus dans la réponse.

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

Étapes suivantes