Add gpt-realtime-whisper Realtime transcription support (OpenAI + Azure)

litellm/__init__.py

@@ -1350,6 +1350,7 @@ def add_known_models(model_cost_map: Optional[Dict] = None):
 from .realtime_api.main import (
     _arealtime,
     acreate_realtime_client_secret,
+    acreate_realtime_transcription_session,
     arealtime_calls,
 )
 from .responses.main import _aresponses_websocket

litellm/cost_calculator.py

@@ -2534,4 +2534,102 @@ def handle_realtime_stream_cost_calculation(
         break  # exit if we find a valid model
     total_cost = input_cost_per_token + output_cost_per_token
 
+    if any(r.get("type") == _TRANSCRIPTION_COMPLETED_EVENT_TYPE for r in results):
+        total_cost += handle_realtime_transcription_cost_calculation(
+            results=results,
+            custom_llm_provider=custom_llm_provider,
+            litellm_model_name=litellm_model_name,
+        )
+
+    return total_cost
+
+
+_TRANSCRIPTION_COMPLETED_EVENT_TYPE = (
+    "conversation.item.input_audio_transcription.completed"
+)
+
+
+def handle_realtime_transcription_cost_calculation(
+    results: OpenAIRealtimeStreamList,
+    custom_llm_provider: str,
+    litellm_model_name: str,
+) -> float:
+    """
+    Cost for realtime transcription sessions (e.g. gpt-realtime-whisper).
+
+    Transcription sessions emit no `response.done` events; instead each
+    `conversation.item.input_audio_transcription.completed` event carries a
+    `usage` object billed by the ASR model. The usage is one of:
+      - {"type": "duration", "seconds": <float>}  → priced via input_cost_per_second
+      - {"type": "tokens", "input_tokens": ...}    → priced via input/audio token cost
+    """
+    completed_events = [
+        cast(dict, result)
+        for result in results
+        if result.get("type") == _TRANSCRIPTION_COMPLETED_EVENT_TYPE
+    ]
+    if not completed_events:
+        return 0.0
+
+    model_name = (
+        _get_transcription_model_name_from_results(results) or litellm_model_name
+    )
+    try:
+        model_info = litellm.get_model_info(
+            model=model_name, custom_llm_provider=custom_llm_provider
+        )
+    except Exception:
+        model_info = {}
+
+    total_cost = 0.0
+    for event in completed_events:
+        usage = event.get("usage") or {}
+        total_cost += _transcription_usage_cost(usage, model_info)
     return total_cost
+
+
+def _get_transcription_model_name_from_results(
+    results: OpenAIRealtimeStreamList,
+) -> Optional[str]:
+    """Resolve the ASR model from a transcription_session.* / session.* event."""
+    for result in results:
+        if result.get("type") in (
+            "transcription_session.created",
+            "transcription_session.updated",
+            "session.created",
+            "session.updated",
+        ):
+            session = cast(dict, result).get("session", {}) or {}
+            transcription = (
+                (session.get("audio", {}) or {}).get("input", {}) or {}
+            ).get("transcription", {}) or session.get("input_audio_transcription", {})
+            model = (transcription or {}).get("model") or session.get("model")
+            if model:
+                return model
+    return None
+
+
+def _transcription_usage_cost(usage: dict, model_info: dict) -> float:
+    usage_type = usage.get("type")
+    if usage_type == "duration":
+        seconds = usage.get("seconds") or 0.0
+        per_second = model_info.get("input_cost_per_second") or 0.0
+        return float(seconds) * float(per_second)
+    if usage_type == "tokens":
+        input_token_details = usage.get("input_token_details") or {}
+        audio_tokens = input_token_details.get("audio_tokens") or 0
+        text_tokens = input_token_details.get("text_tokens") or 0
+        output_tokens = usage.get("output_tokens") or 0
+        audio_cost = float(audio_tokens) * float(
+            model_info.get("input_cost_per_audio_token")
+            or model_info.get("input_cost_per_token")
+            or 0.0
+        )
+        text_cost = float(text_tokens) * float(
+            model_info.get("input_cost_per_token") or 0.0
+        )
+        output_cost = float(output_tokens) * float(
+            model_info.get("output_cost_per_token") or 0.0
+        )
+        return audio_cost + text_cost + output_cost
+    return 0.0

litellm/litellm_core_utils/realtime_streaming.py

@@ -47,6 +47,7 @@ def __init__(
         user_api_key_dict: Optional[Any] = None,
         request_data: Optional[Dict] = None,
         backend_uses_beta_protocol: Optional[bool] = None,
+        force_transcription_model: Optional[str] = None,
     ):
         self.websocket = websocket
         self.backend_ws = backend_ws
@@ -100,6 +101,11 @@ def __init__(
         self._flushing_pending_messages_until_setup: bool = False
         self._pending_messages_until_setup: List[str] = []
         self._pending_messages_byte_total: int = 0
+        # Whether this is a transcription-only session (session.type == "transcription",
+        # e.g. gpt-realtime-whisper). Such sessions must not be sent response.create and
+        # their input_audio_transcription.completed usage drives duration-based cost.
+        self._force_transcription_model = force_transcription_model
+        self._is_transcription_session: bool = force_transcription_model is not None
 
     # Per-connection caps for pre-setup audio frames (message count + total bytes).
     _MAX_BUFFERED_MESSAGES: int = 200
@@ -209,6 +215,8 @@ def _collect_user_input_from_client_event(self, message: Union[str, dict]) -> No
                     self.session_tools = tools
                 # GA: session.type is required; log it for traceability but no action needed
                 verbose_logger.debug(f"Realtime session.type: {session.get('type')}")
+                if session.get("type") == "transcription":
+                    self._is_transcription_session = True
         except (json.JSONDecodeError, AttributeError, TypeError):
             pass
 
@@ -225,6 +233,55 @@ def _collect_user_input_from_backend_event(
         except (AttributeError, TypeError):
             pass
 
+    def _detect_transcription_session_from_backend(
+        self, event_obj: Union[dict, OpenAIRealtimeEvents]
+    ) -> None:
+        """Flag transcription-only sessions from backend session events."""
+        try:
+            event_type = event_obj.get("type", "")
+            if event_type in (
+                "transcription_session.created",
+                "transcription_session.updated",
+            ):
+                self._is_transcription_session = True
+            elif event_type in ("session.created", "session.updated"):
+                session = cast(dict, event_obj).get("session", {}) or {}
+                if session.get("type") == "transcription":
+                    self._is_transcription_session = True
+        except (AttributeError, TypeError):
+            pass
+
+    def _capture_transcription_usage(
+        self, event_obj: Union[dict, OpenAIRealtimeEvents]
+    ) -> None:
+        """
+        Append a usage-only transcription completed event to the logged results so
+        the cost calculator can bill it by audio duration. The default logged event
+        types exclude this event, so it is captured here directly for transcription
+        sessions rather than widening logging for every realtime session. Only the
+        type and usage are kept — the transcript is already captured separately in
+        input_messages, so it is not duplicated into the response log here.
+        """
+        try:
+            usage = event_obj.get("usage")
+            if usage is None:
+                return
+            # If this event type is already captured by store_message (e.g. the user
+            # logs all realtime events), don't append a second copy.
+            if self._should_store_message(event_obj):
+                return
+            self.messages.append(
+                cast(
+                    OpenAIRealtimeEvents,
+                    {
+                        "type": "conversation.item.input_audio_transcription.completed",
+                        "usage": usage,
+                    },
+                )
+            )
+        except (AttributeError, TypeError):
+            pass
+
     def _collect_tool_calls_from_response_done(
         self, event_obj: Union[dict, OpenAIRealtimeEvents]
     ) -> None:
@@ -285,6 +342,7 @@ async def _send_to_backend(self, message: str) -> bool:
         backend, False if the provider transformation produced no output and
         the message was effectively dropped.
         """
+        message = self._enforce_transcription_session_model(message)
         if self.provider_config:
             transformed = self.provider_config.transform_realtime_request(
                 message, self.model, self.session_configuration_request
@@ -304,6 +362,80 @@ async def _send_to_backend(self, message: str) -> bool:
         await self.backend_ws.send(message)  # type: ignore[union-attr, attr-defined]
         return True
 
+    def _enforce_transcription_session_model(self, message: str) -> str:
+        """Force client transcription session updates to the authorized model.
+
+        `/v1/realtime?intent=transcription` may intentionally omit `model` from
+        the upstream URL for Azure compatibility, but the proxy still authorizes
+        a resolved LiteLLM model before opening the backend websocket. If a
+        client later sends a transcription `session.update`, any model embedded
+        in that update must be rewritten to the same authorized model instead of
+        allowing a post-auth model/deployment switch.
+
+        Normal realtime sessions keep their independent nested transcription
+        model behavior because `_force_transcription_model` is only set for
+        transcription-intent websocket routes.
+        """
+        if self._force_transcription_model is None:
+            return message
+
+        try:
+            message_obj = json.loads(message)
+        except (json.JSONDecodeError, TypeError):
+            return message
+
+        if message_obj.get("type") not in (
+            "session.update",
+            "transcription_session.update",
+        ):
+            return message
+
+        session = message_obj.get("session")
+        if not isinstance(session, dict):
+            return message
+
+        if session.get("type") == "transcription":
+            self._is_transcription_session = True
+
+        authorized_model = self._force_transcription_model
+        changed = False
+
+        transcription = session.get("input_audio_transcription")
+        if (
+            isinstance(transcription, dict)
+            and transcription.get("model") != authorized_model
+        ):
+            session["input_audio_transcription"] = {
+                **transcription,
+                "model": authorized_model,
+            }
+            changed = True
+
+        audio = session.get("audio")
+        if isinstance(audio, dict):
+            audio_input = audio.get("input")
+            if isinstance(audio_input, dict):
+                nested_transcription = audio_input.get("transcription")
+                if (
+                    isinstance(nested_transcription, dict)
+                    and nested_transcription.get("model") != authorized_model
+                ):
+                    session["audio"] = {
+                        **audio,
+                        "input": {
+                            **audio_input,
+                            "transcription": {
+                                **nested_transcription,
+                                "model": authorized_model,
+                            },
+                        },
+                    }
+                    changed = True
+
+        if not changed:
+            return message
+        return json.dumps(message_obj)
+
     def _uses_deferred_backend_setup(self) -> bool:
         """True when setup is deferred until the client's first session.update."""
         if self.provider_config is None:
@@ -713,6 +845,8 @@ async def _handle_raw_backend_message(self, raw_response) -> bool:
         try:
             event_obj = json.loads(raw_response)
 
+            self._detect_transcription_session_from_backend(event_obj)
+
             # For audio/VAD guardrail path: once the session is ready, tell the backend
             # not to auto-respond after VAD detects end-of-speech.  We send the
             # session.created to the client FIRST so the client is always in sync, then
@@ -731,12 +865,20 @@ async def _handle_raw_backend_message(self, raw_response) -> bool:
                 event_obj.get("type")
                 == "conversation.item.input_audio_transcription.completed"
             ):
-                transcript = event_obj.get("transcript", "")
                 self._collect_user_input_from_backend_event(event_obj)
                 ## LOGGING — must happen before continue below
                 self.store_message(raw_response)
                 # Forward transcript to client so user sees what they said
                 await self.websocket.send_text(raw_response)
+
+                # Transcription-only sessions (e.g. gpt-realtime-whisper) have no
+                # assistant turn: capture audio-duration usage for cost and never
+                # trigger response.create.
+                if self._is_transcription_session:
+                    self._capture_transcription_usage(event_obj)
+                    return True
+
+                transcript = event_obj.get("transcript", "")
                 blocked = await self.run_realtime_guardrails(
                     transcript,
                     item_id=event_obj.get("item_id"),