Erste Schritte mit der Live API über WebSockets

In dieser Anleitung erfahren Sie, wie Sie über WebSockets eine Verbindung zur Live API herstellen. In diesem Tutorial richten Sie ein Google Cloud -Projekt ein, um die Live API mit WebSockets zu verwenden, eine Audiodatei an das Modell zu senden und Audio als Antwort zu erhalten.

Weitere Informationen zu WebSockets finden Sie in der WebSockets API-Dokumentation.

Hinweise

Bevor Sie Anfragen senden können, müssen Sie die Authentifizierung mit Vertex AI einrichten. Sie können die Authentifizierung entweder mit einem API-Schlüssel oder mit Standardanmeldedaten für Anwendungen (Application Default Credentials, ADC) einrichten.

Für diese Anleitung ist die Verwendung eines API-Schlüssels der schnellste Weg, um loszulegen:

Eine Anleitung zum Einrichten der Authentifizierung mit ADC finden Sie in unserer Schnellstartanleitung.

WebSockets-Bibliothek installieren

Führen Sie den folgenden Befehl aus, um die websockets-Bibliothek zu installieren:

pip install websockets

Umgebungsvariablen einrichten

Legen Sie Umgebungsvariablen für Ihre Projekt-ID und Ihren Standort fest. Ersetzen Sie PROJECT_ID durch IhreGoogle Cloud Projekt-ID.

export GOOGLE_CLOUD_PROJECT=PROJECT_ID
export GOOGLE_CLOUD_LOCATION=global

Audiositzung starten

Im folgenden Beispiel wird eine Sitzung eingerichtet, Audio aus einer Datei gestreamt und die Größe der in der Antwort empfangenen Audio-Chunks ausgegeben.

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

Nächste Schritte