Comece a usar a API Live com WebSockets

Este tutorial mostra como estabelecer ligação à API Live através de WebSockets. Neste tutorial, vai configurar um projeto para usar a API Live com WebSockets, enviar um ficheiro de áudio para o modelo e receber áudio em resposta. Google Cloud

Para mais informações sobre WebSockets, consulte a documentação da API WebSockets.

Antes de começar

Antes de poder enviar pedidos, tem de configurar a autenticação com a Vertex AI. Pode configurar a autenticação através de uma chave da API ou através das Credenciais padrão da aplicação (ADC).

Para este tutorial, a forma mais rápida de começar é usar uma chave da API:

Para obter instruções sobre como configurar a autenticação através do ADC, consulte o nosso guia de início rápido.

Instale a biblioteca WebSockets

Execute o seguinte comando para instalar a biblioteca websockets:

pip install websockets

Configure variáveis de ambiente

Defina as variáveis de ambiente para o ID do projeto e a localização. Substitua PROJECT_ID pelo ID do seu projeto.Google Cloud 

export GOOGLE_CLOUD_PROJECT=PROJECT_ID
export GOOGLE_CLOUD_LOCATION=global

Inicie uma sessão de áudio

O exemplo seguinte estabelece uma sessão, transmite áudio a partir de um ficheiro e imprime o tamanho dos fragmentos de áudio recebidos na resposta.

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

O que se segue?