feat(vertex_ai): Vertex AI Gemini Live via unified /realtime endpoint

docs/my-website/docs/providers/vertex_realtime.md

@@ -0,0 +1,203 @@
+# Vertex AI Gemini Live - Realtime API
+
+Use Vertex AI's Gemini Live API (BidiGenerateContent) through LiteLLM's unified `/realtime` endpoint, which speaks the OpenAI Realtime protocol.
+
+| Feature | Supported |
+|---------|-----------|
+| Proxy (`/realtime`) | ✅ |
+| Voice in / Voice out | ✅ |
+| Text in / Text out | ✅ |
+| Server VAD | ✅ |
+| Output transcription | ✅ |
+
+## Setup
+
+### 1. Auth
+
+LiteLLM uses your Google Cloud credentials (OAuth2 Bearer token), not an API key.
+
+```bash
+gcloud auth application-default login
+```
+
+Or set a service-account key file:
+
+```bash
+export GOOGLE_APPLICATION_CREDENTIALS=/path/to/sa-key.json
+```
+
+### 2. Proxy config
+
+```yaml
+model_list:
+  - model_name: vertex-gemini-live
+    litellm_params:
+      model: vertex_ai/gemini-2.0-flash-live-001
+      vertex_project: your-gcp-project-id
+      vertex_location: us-east4   # or any supported region, or "global"
+
+general_settings:
+  master_key: sk-your-key
+```
+
+### 3. Start the proxy
+
+```bash
+litellm --config config.yaml --port 4000
+```
+
+## Usage
+
+### Python (websockets)
+
+```python
+import asyncio
+import json
+import websockets
+
+PROXY_URL = "ws://localhost:4000/realtime?model=vertex-gemini-live"
+API_KEY = "sk-your-key"
+
+async def main():
+    async with websockets.connect(
+        PROXY_URL,
+        additional_headers={"api-key": API_KEY},
+    ) as ws:
+        # Wait for session.created
+        event = json.loads(await ws.recv())
+        print(f"session.created: {event['session']['id']}")
+
+        # Send a text message
+        await ws.send(json.dumps({
+            "type": "conversation.item.create",
+            "item": {
+                "type": "message",
+                "role": "user",
+                "content": [{"type": "input_text", "text": "Say hello in one sentence."}],
+            },
+        }))
+
+        # Collect the response
+        async for raw in ws:
+            ev = json.loads(raw)
+            t = ev.get("type", "")
+            if t == "response.text.delta":
+                print(ev.get("delta", ""), end="", flush=True)
+            elif t == "response.done":
+                print("\n[done]")
+                break
+
+asyncio.run(main())
+```
+
+### Node.js
+
+```js
+const WebSocket = require("ws");
+
+const ws = new WebSocket(
+  "ws://localhost:4000/realtime?model=vertex-gemini-live",
+  { headers: { "api-key": "sk-your-key" } }
+);
+
+ws.on("open", () => {
+  ws.send(JSON.stringify({
+    type: "conversation.item.create",
+    item: {
+      type: "message",
+      role: "user",
+      content: [{ type: "input_text", text: "Say hello." }],
+    },
+  }));
+});
+
+ws.on("message", (data) => {
+  const ev = JSON.parse(data);
+  if (ev.type === "response.text.delta") process.stdout.write(ev.delta);
+  if (ev.type === "response.done") ws.close();
+});
+```
+
+### OpenAI SDK (Python)
+
+```python
+import asyncio
+from openai import AsyncOpenAI
+
+client = AsyncOpenAI(
+    base_url="http://localhost:4000",
+    api_key="sk-your-key",
+)
+
+async def main():
+    async with client.beta.realtime.connect(
+        model="vertex-gemini-live"
+    ) as conn:
+        await conn.session.update(session={"modalities": ["text"]})
+
+        await conn.conversation.item.create(
+            item={
+                "type": "message",
+                "role": "user",
+                "content": [{"type": "input_text", "text": "Say hello."}],
+            }
+        )
+
+        async for event in conn:
+            if event.type == "response.text.delta":
+                print(event.delta, end="", flush=True)
+            elif event.type == "response.done":
+                print()
+                break
+
+asyncio.run(main())
+```
+
+## Voice in / Voice out
+
+For a complete voice example see [`voice_realtime_test.py`](https://github.com/BerriAI/litellm/blob/main/voice_realtime_test.py).
+
+Key settings for audio:
+- Microphone input: **16 kHz** PCM16 (`audio/pcm;rate=16000`)
+- Speaker output: **24 kHz** PCM16 (Vertex AI returns audio at 24 kHz)
+- Server VAD is enabled by default with 800 ms silence threshold
+
+```python
+# session.update with server VAD — the proxy ignores this for Vertex AI
+# because VAD is already configured in the initial setup message.
+await ws.send(json.dumps({
+    "type": "session.update",
+    "session": {
+        "modalities": ["audio"],
+        "turn_detection": {"type": "server_vad", "silence_duration_ms": 800},
+    },
+}))
+```
+
+## Supported OpenAI Realtime Events
+
+**Client → Proxy (→ Vertex AI)**
+
+| OpenAI event | Notes |
+|---|---|
+| `input_audio_buffer.append` | Forwarded as `realtime_input.audio` |
+| `conversation.item.create` | Forwarded as `realtime_input.text` |
+| `session.update` | Silently ignored — Vertex AI does not support mid-session reconfiguration |
+| `response.create` | Silently ignored — Vertex AI responds automatically after each turn |
+
+**Vertex AI → Proxy (→ Client)**
+
+| OpenAI event emitted | Vertex AI source |
+|---|---|
+| `session.created` | Synthesized after `setupComplete` |
+| `response.text.delta` | `serverContent.modelTurn.parts[].text` |
+| `response.audio.delta` | `serverContent.modelTurn.parts[].inlineData` |
+| `response.audio_transcript.delta` | `serverContent.outputTranscription.text` |
+| `conversation.item.input_audio_transcription.completed` | `serverContent.inputTranscription.text` |
+| `response.done` | `serverContent.turnComplete` |
+
+## Limitations
+
+- `session.update` is not forwarded (Vertex AI only accepts one setup message per connection).
+- Tool calling / function calling is not yet supported.
+- Audio transcription requires `outputAudioTranscription: {}` to be set in the initial setup (done automatically by LiteLLM).

docs/my-website/sidebars.js

@@ -758,6 +758,7 @@ const sidebars = {
             "providers/vertex_batch",
             "providers/vertex_ocr",
             "providers/vertex_ai_agent_engine",
+            "providers/vertex_realtime",
           ]
         },
         {

litellm/litellm_core_utils/realtime_streaming.py

@@ -145,7 +145,9 @@ def _collect_user_input_from_client_event(
         except (json.JSONDecodeError, AttributeError, TypeError):
             pass
 
-    def _collect_user_input_from_backend_event(self, event_obj: dict) -> None:
+    def _collect_user_input_from_backend_event(
+        self, event_obj: Union[dict, OpenAIRealtimeEvents]
+    ) -> None:
         """Extract user voice transcription from backend events for spend logging."""
         try:
             event_type = event_obj.get("type", "")
@@ -162,7 +164,7 @@ def _collect_user_input_from_backend_event(self, event_obj: dict) -> None:
             pass
 
     def _collect_tool_calls_from_response_done(
-        self, event_obj: dict
+        self, event_obj: Union[dict, OpenAIRealtimeEvents]
     ) -> None:
         """Extract function_call items from response.done events for spend logging."""
         try:
@@ -211,6 +213,23 @@ async def log_messages(self):
             ## SYNC LOGGING
             executor.submit(self.logging_obj.success_handler(self.messages))
 
+    async def _send_to_backend(self, message: str) -> None:
+        """Send a message to the backend WebSocket.
+
+        If a provider_config is set the message is first passed through
+        transform_realtime_request so that provider-specific translation
+        (e.g. dropping session.update for Vertex AI) is applied even for
+        guardrail-injected messages.
+        """
+        if self.provider_config:
+            transformed = self.provider_config.transform_realtime_request(
+                message, self.model, self.session_configuration_request
+            )
+            for msg in transformed:
+                await self.backend_ws.send(msg)
+        else:
+            await self.backend_ws.send(message)
+
     def _has_realtime_guardrails(self) -> bool:
         """Return True if any callback is registered for realtime_input_transcription."""
         from litellm.integrations.custom_guardrail import CustomGuardrail
@@ -276,9 +295,9 @@ async def run_realtime_guardrails(
                     safe_msg = str(e) or "I'm sorry, that request was blocked by the content filter."
                 # Cancel any in-flight response before speaking the warning.
                 # This handles the race where create_response fired before we could intercept.
-                await self.backend_ws.send(json.dumps({"type": "response.cancel"}))
-                # Ask OpenAI to speak the warning — TTS audio plays naturally in the client
-                await self.backend_ws.send(
+                await self._send_to_backend(json.dumps({"type": "response.cancel"}))
+                # Ask the model to speak the warning — TTS audio plays naturally in the client
+                await self._send_to_backend(
                     json.dumps(
                         {
                             "type": "response.create",
@@ -333,7 +352,7 @@ async def _handle_provider_config_message(self, raw_response) -> None:
             ## GUARDRAIL: inject create_response=false on session.created
             if isinstance(event, dict) and event.get("type") == "session.created":
                 if self._has_realtime_guardrails():
-                    await self.backend_ws.send(
+                    await self._send_to_backend(
                         json.dumps(
                             {
                                 "type": "session.update",
@@ -362,7 +381,7 @@ async def _handle_provider_config_message(self, raw_response) -> None:
                     transcript, item_id=event.get("item_id")
                 )
                 if not blocked:
-                    await self.backend_ws.send(
+                    await self._send_to_backend(
                         json.dumps({"type": "response.create"})
                     )
                 continue
@@ -383,7 +402,7 @@ async def _handle_raw_backend_message(self, raw_response) -> bool:
                 # set create_response=false so the LLM never auto-responds
                 # before our guardrail has a chance to run.
                 if self._has_realtime_guardrails():
-                    await self.backend_ws.send(
+                    await self._send_to_backend(
                         json.dumps(
                             {
                                 "type": "session.update",
@@ -416,7 +435,7 @@ async def _handle_raw_backend_message(self, raw_response) -> bool:
                 )
                 if not blocked:
                     # Clean — trigger LLM response
-                    await self.backend_ws.send(
+                    await self._send_to_backend(
                         json.dumps({"type": "response.create"})
                     )
                 return True
@@ -437,7 +456,13 @@ async def backend_to_client_send_messages(self):
                     raw_response = await self.backend_ws.recv()  # type: ignore[assignment]
 
                 if self.provider_config:
-                    await self._handle_provider_config_message(raw_response)
+                    try:
+                        await self._handle_provider_config_message(raw_response)
+                    except Exception as e:
+                        verbose_logger.exception(
+                            f"Error processing backend message, skipping: {e}"
+                        )
+                        continue
                 else:
                     handled = await self._handle_raw_backend_message(raw_response)
                     if handled:

litellm/llms/custom_httpx/llm_http_handler.py

@@ -4678,13 +4678,23 @@ async def async_realtime(
                 max_size=REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES,
                 ssl=ssl_context,
             ) as backend_ws:
+                # Auto-send session setup if the provider requires it
+                # (e.g. Gemini/Vertex AI Live needs a `setup` message before any realtime_input)
+                _session_config: Optional[str] = None
+                if provider_config.requires_session_configuration():
+                    _session_config = provider_config.session_configuration_request(model)
+                    if _session_config:
+                        await backend_ws.send(_session_config)
+
                 realtime_streaming = RealTimeStreaming(
                     websocket,
                     cast(ClientConnection, backend_ws),
                     logging_obj,
                     provider_config,
                     model,
                 )
+                if _session_config:
+                    realtime_streaming.session_configuration_request = _session_config
                 await realtime_streaming.bidirectional_forward()
 
         except websockets.exceptions.InvalidStatusCode as e:  # type: ignore