diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py index 907f6b73250..8963ebd5c75 100644 --- a/litellm/litellm_core_utils/realtime_streaming.py +++ b/litellm/litellm_core_utils/realtime_streaming.py @@ -7,6 +7,7 @@ from dataclasses import dataclass from enum import Enum, auto from typing import TYPE_CHECKING, Any, Final, NoReturn, Protocol, TypedDict, cast +from openai.types.realtime.realtime_audio_formats import AudioPCM, AudioPCMA, AudioPCMU, RealtimeAudioFormats from pydantic import TypeAdapter, ValidationError from typing_extensions import ReadOnly @@ -113,23 +114,17 @@ def _decode_json_object(payload: str) -> Mapping[str, object]: return json.loads(payload) -_TRANSLATION_AUDIO_FORMAT: Final = TypeAdapter(str | Mapping[str, object] | None) +_TRANSLATION_AUDIO_FORMAT: Final = TypeAdapter(str | RealtimeAudioFormats | None) -def _translation_audio_bytes_per_second(audio_format: str | Mapping[str, object] | None) -> float | None: - if isinstance(audio_format, str): - return 8000.0 if audio_format in ("g711_ulaw", "g711_alaw") else None - if audio_format is None: - return None - format_type: Final = audio_format.get("type") - rate: Final = audio_format.get("rate") - if format_type in ("audio/pcmu", "audio/pcma") and rate is None: - return 8000.0 - if not isinstance(rate, (int, float)) or rate <= 0: - return None - if format_type == "audio/pcm": - return float(rate) * 2 - return float(rate) if format_type in ("audio/pcmu", "audio/pcma") else None +def _translation_audio_bytes_per_second(audio_format: str | RealtimeAudioFormats | None) -> float | None: + match audio_format: + case "pcm16" | AudioPCM(): + return 48000.0 + case "g711_ulaw" | "g711_alaw" | AudioPCMU() | AudioPCMA(): + return 8000.0 + case _: + return None class RealtimeEventNormalizer(Protocol): @@ -176,7 +171,7 @@ class RealTimeStreaming: self._is_translation_session = translation_session self._translation_input_seconds = 0.0 self._translation_input_bytes_per_second = 48000.0 - self._translation_output_audio_bytes = 0 + self._translation_output_seconds = 0.0 self._translation_output_bytes_per_second = 48000.0 self._translation_usage_finalized = False @@ -473,11 +468,7 @@ class RealTimeStreaming: if reported_input_seconds is not None else self._translation_input_seconds or None ) - synthetic_output_seconds: Final = ( - self._translation_output_audio_bytes / self._translation_output_bytes_per_second - if self._translation_output_audio_bytes > 0 - else None - ) + synthetic_output_seconds: Final = self._translation_output_seconds or None resolved_output_seconds: Final = output_seconds if output_seconds is not None else synthetic_output_seconds if input_seconds is not None or resolved_output_seconds is not None: if self._should_store_message(event_obj): @@ -527,7 +518,7 @@ class RealTimeStreaming: decoded: Final = base64.b64decode(delta, validate=True) except (ValueError, TypeError): return - self._translation_output_audio_bytes += len(decoded) + self._translation_output_seconds += len(decoded) / self._translation_output_bytes_per_second def _capture_translation_input_audio(self, message: str) -> None: if not self._is_translation_session: @@ -536,7 +527,6 @@ class RealTimeStreaming: event: Final = _decode_json_object(message) except (json.JSONDecodeError, TypeError): return - self._capture_translation_audio_formats(event) if event.get("type") != "session.input_audio_buffer.append" or not isinstance(audio := event.get("audio"), str): return try: @@ -546,6 +536,8 @@ class RealTimeStreaming: self._translation_input_seconds += len(decoded) / self._translation_input_bytes_per_second def _capture_translation_audio_formats(self, event_obj: Mapping[str, object]) -> None: + if event_obj.get("type") not in ("session.created", "session.updated"): + return session: Final = event_obj.get("session") if not isinstance(session, dict): return @@ -581,15 +573,18 @@ class RealTimeStreaming: ): self._translation_usage_finalized = True return - if self._translation_output_audio_bytes == 0 and self._translation_input_seconds == 0: + if self._translation_output_seconds == 0 and self._translation_input_seconds == 0: return - output_seconds: Final = self._translation_output_audio_bytes / self._translation_output_bytes_per_second synthetic_usage: Final = ( OpenAIRealtimeTranslationDurationUsage( - type="duration", input_seconds=self._translation_input_seconds, output_seconds=output_seconds + type="duration", + input_seconds=self._translation_input_seconds, + output_seconds=self._translation_output_seconds, ) if self._translation_input_seconds > 0 - else OpenAIRealtimeTranslationDurationUsage(type="duration", output_seconds=output_seconds) + else OpenAIRealtimeTranslationDurationUsage( + type="duration", output_seconds=self._translation_output_seconds + ) ) self.messages.append(OpenAIRealtimeTranslationClosedEvent(type="session.closed", usage=synthetic_usage)) self._translation_usage_finalized = True diff --git a/litellm/proxy/agent_endpoints/auth/managed_authorization.py b/litellm/proxy/agent_endpoints/auth/managed_authorization.py index 7e9526e411f..932aa74e668 100644 --- a/litellm/proxy/agent_endpoints/auth/managed_authorization.py +++ b/litellm/proxy/agent_endpoints/auth/managed_authorization.py @@ -110,11 +110,19 @@ def managed_inference_request( cli_model: str | None, path_model: object = None, query_model: object = None, + intent: str | None = None, ) -> dict[str, object]: from litellm.proxy.auth.route_checks import RouteChecks if route in _MANAGED_REALTIME_ROUTES: - model: Final = query_model or body.get("model") + from litellm.proxy.common_utils.http_parsing_utils import resolve_realtime_route_model + + requested_model: Final = query_model if query_model is not None else body.get("model") + model: Final = ( + resolve_realtime_route_model(None, intent, route.endswith("/realtime/translations")) + if requested_model is None + else requested_model + ) if not isinstance(model, str) or not model: raise_identity_failure( AgentIdentityFailure(message="Managed inference requires an explicit or configured model") diff --git a/litellm/proxy/auth/user_api_key_auth.py b/litellm/proxy/auth/user_api_key_auth.py index 87bba4ecedd..935feb582c4 100644 --- a/litellm/proxy/auth/user_api_key_auth.py +++ b/litellm/proxy/auth/user_api_key_auth.py @@ -3347,6 +3347,7 @@ async def _authorize_authenticated_request( user_model, request.path_params.get("model") or request.path_params.get("model_name"), request.query_params.get("model"), + intent=request.query_params.get("intent"), ) if user_api_key_auth_obj.managed_agent_policy is not None else request_data diff --git a/litellm/proxy/common_utils/http_parsing_utils.py b/litellm/proxy/common_utils/http_parsing_utils.py index 2ef4dff9569..7efc524b08a 100644 --- a/litellm/proxy/common_utils/http_parsing_utils.py +++ b/litellm/proxy/common_utils/http_parsing_utils.py @@ -30,6 +30,16 @@ _BINARY_CONTENT_TYPES: Final[frozenset[str]] = frozenset({"application/x-protobu _ANNOTATION_QUALIFIERS: Final[frozenset[object]] = frozenset({Annotated, NotRequired, ReadOnly, Required}) +def resolve_realtime_route_model(model: str | None, intent: str | None, is_translation: bool) -> str | None: + if model is not None: + return model + if is_translation: + return "gpt-realtime-translate" + if intent == "transcription": + return "gpt-realtime-whisper" + return None + + def resolve_inference_model( body_model: object, settings: Mapping[str, object], diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 4007b471a94..5f04d49faa5 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -445,6 +445,9 @@ from litellm.proxy.common_utils.http_parsing_utils import ( get_form_data, resolve_inference_model, ) +from litellm.proxy.common_utils.http_parsing_utils import ( + resolve_realtime_route_model as _resolve_realtime_route_model, +) from litellm.proxy.common_utils.load_config_utils import get_config_from_bucket from litellm.proxy.common_utils.model_deprecation import collect_model_deprecations from litellm.proxy.common_utils.model_listing_utils import ( @@ -13042,20 +13045,6 @@ async def _reject_realtime_session( await _release_realtime_max_parallel_slot(user_api_key_dict) -def _resolve_realtime_route_model( - model: str | None, - intent: str | None, - is_translation: bool, -) -> str | None: - if model is not None: - return model - if is_translation: - return "gpt-realtime-translate" - if intent == "transcription": - return "gpt-realtime-whisper" - return None - - def _resolve_realtime_upstream_query_model( model: str | None, intent: str | None, diff --git a/tests/unit/litellm_core_utils/test_realtime_streaming.py b/tests/unit/litellm_core_utils/test_realtime_streaming.py index 27501105188..1820726a579 100644 --- a/tests/unit/litellm_core_utils/test_realtime_streaming.py +++ b/tests/unit/litellm_core_utils/test_realtime_streaming.py @@ -1,4 +1,5 @@ import asyncio +import base64 import json from collections.abc import Coroutine, Mapping from dataclasses import dataclass @@ -2943,14 +2944,14 @@ def test_translation_audio_duration_is_finalized_once(event_type: str): @pytest.mark.asyncio @pytest.mark.parametrize("output_bytes", (0, 48000)) -@pytest.mark.parametrize("format_event_type", ("session.created", "session.update")) +@pytest.mark.parametrize("format_event_type", ("session.created", "session.updated")) @pytest.mark.parametrize( "audio_format,bytes_per_second", ( ("pcm16", 48000), ("g711_ulaw", 8000), ("g711_alaw", 8000), - ({"type": "audio/pcm", "rate": 16000}, 32000), + ({"type": "audio/pcm", "rate": 24000}, 48000), ({"type": "audio/pcmu", "rate": 8000}, 8000), ({"type": "audio/pcma", "rate": 8000}, 8000), ({"type": "audio/pcmu"}, 8000), @@ -2975,10 +2976,7 @@ async def test_translation_disconnect_bills_sent_input_audio( "type": format_event_type, "session": {"audio": {"input": {"format": audio_format}}}, } - if format_event_type == "session.update": - await streaming._send_to_backend(json.dumps(format_event)) - else: - streaming._capture_translation_output_audio(format_event) + streaming._capture_translation_output_audio(format_event) await streaming._send_to_backend( json.dumps( { @@ -3001,6 +2999,96 @@ async def test_translation_disconnect_bills_sent_input_audio( ] +@pytest.mark.asyncio +@pytest.mark.parametrize("rate", (24000, 1, 0, -1, float("nan"), float("inf"))) +async def test_translation_client_format_updates_do_not_change_billed_duration(rate: int | float) -> None: + backend: Final = MagicMock() + backend.send = AsyncMock() + streaming: Final = RealTimeStreaming( + websocket=_ga_client_ws(), backend_ws=backend, logging_obj=MagicMock(), translation_session=True + ) + streaming._capture_translation_output_audio( + { + "type": "session.created", + "session": {"audio": {"input": {"format": "g711_ulaw"}, "output": {"format": "g711_ulaw"}}}, + } + ) + input_event: Final = json.dumps( + {"type": "session.input_audio_buffer.append", "audio": base64.b64encode(bytes(8000)).decode()} + ) + output_event: Final = {"type": "session.output_audio.delta", "delta": base64.b64encode(bytes(8000)).decode()} + await streaming._send_to_backend(input_event) + streaming._capture_translation_output_audio(output_event) + await streaming._send_to_backend( + json.dumps( + { + "type": "session.update", + "session": { + "audio": { + "input": {"format": {"type": "audio/pcm", "rate": rate}}, + "output": {"format": {"type": "audio/pcm", "rate": rate}}, + } + }, + } + ) + ) + await streaming._send_to_backend(input_event) + streaming._capture_translation_output_audio(output_event) + streaming._finalize_translation_usage() + + assert streaming.messages == [ + {"type": "session.closed", "usage": {"type": "duration", "input_seconds": 2.0, "output_seconds": 2.0}} + ] + + +@pytest.mark.asyncio +async def test_translation_confirmed_format_changes_preserve_previous_audio_duration() -> None: + backend: Final = MagicMock() + backend.send = AsyncMock() + streaming: Final = RealTimeStreaming( + websocket=_ga_client_ws(), backend_ws=backend, logging_obj=MagicMock(), translation_session=True + ) + await streaming._send_to_backend( + json.dumps({"type": "session.input_audio_buffer.append", "audio": base64.b64encode(bytes(48000)).decode()}) + ) + streaming._capture_translation_output_audio( + {"type": "session.output_audio.delta", "delta": base64.b64encode(bytes(48000)).decode()} + ) + streaming._capture_translation_output_audio( + { + "type": "session.updated", + "session": {"audio": {"input": {"format": "g711_alaw"}, "output": {"format": "g711_alaw"}}}, + } + ) + await streaming._send_to_backend( + json.dumps({"type": "session.input_audio_buffer.append", "audio": base64.b64encode(bytes(8000)).decode()}) + ) + streaming._capture_translation_output_audio( + {"type": "session.output_audio.delta", "delta": base64.b64encode(bytes(8000)).decode()} + ) + streaming._finalize_translation_usage() + + assert streaming.messages == [ + {"type": "session.closed", "usage": {"type": "duration", "input_seconds": 2.0, "output_seconds": 2.0}} + ] + + +@pytest.mark.parametrize("rate", (1, 0, -1, float("nan"), float("inf"), 10**400)) +def test_translation_invalid_provider_format_preserves_audio_duration(rate: int | float) -> None: + streaming: Final = RealTimeStreaming( + websocket=_ga_client_ws(), backend_ws=MagicMock(), logging_obj=MagicMock(), translation_session=True + ) + output_event: Final = {"type": "session.output_audio.delta", "delta": base64.b64encode(bytes(48000)).decode()} + streaming._capture_translation_output_audio(output_event) + streaming._capture_translation_output_audio( + {"type": "session.updated", "session": {"audio": {"output": {"format": {"type": "audio/pcm", "rate": rate}}}}} + ) + streaming._capture_translation_output_audio(output_event) + streaming._finalize_translation_usage() + + assert streaming.messages == [{"type": "session.closed", "usage": {"type": "duration", "output_seconds": 2.0}}] + + @pytest.mark.asyncio async def test_translation_failed_audio_send_is_not_billed() -> None: backend: Final = MagicMock() @@ -3085,7 +3173,9 @@ def test_translation_does_not_duplicate_provider_duration_usage(): model="gpt-realtime-translate", translation_session=True, ) - streaming._translation_output_audio_bytes = 48000 + streaming._capture_translation_output_audio( + {"type": "session.output_audio.delta", "delta": base64.b64encode(bytes(48000)).decode()} + ) streaming.messages.append({"type": "session.closed", "usage": {"type": "duration", "output_seconds": 0.5}}) streaming._finalize_translation_usage() @@ -3133,7 +3223,9 @@ def test_translation_preserves_input_only_provider_usage( model="gpt-realtime-translate", translation_session=True, ) - streaming._translation_output_audio_bytes = output_audio_bytes + streaming._capture_translation_output_audio( + {"type": "session.output_audio.delta", "delta": base64.b64encode(bytes(output_audio_bytes)).decode()} + ) streaming._capture_translation_output_audio( {"type": "session.closed", "usage": {"type": "duration", "input_seconds": 0.25}} ) @@ -3153,7 +3245,9 @@ def test_translation_retained_input_only_close_event_bills_captured_output(monke model="gpt-realtime-translate", translation_session=True, ) - streaming._translation_output_audio_bytes = 48000 + streaming._capture_translation_output_audio( + {"type": "session.output_audio.delta", "delta": base64.b64encode(bytes(48000)).decode()} + ) close_event: Final = {"type": "session.closed", "usage": {"type": "duration", "input_seconds": 0.25}} streaming._capture_translation_output_audio(close_event) @@ -3175,10 +3269,15 @@ def test_translation_malformed_provider_duration_uses_captured_audio(malformed_o model="gpt-realtime-translate", translation_session=True, ) - streaming._translation_output_audio_bytes = 48000 + streaming._capture_translation_output_audio( + {"type": "session.output_audio.delta", "delta": base64.b64encode(bytes(48000)).decode()} + ) streaming._capture_translation_output_audio( - {"type": "session.closed", "usage": {"type": "duration", "input_seconds": 0.25, "output_seconds": malformed_output}} + { + "type": "session.closed", + "usage": {"type": "duration", "input_seconds": 0.25, "output_seconds": malformed_output}, + } ) streaming._finalize_translation_usage() diff --git a/tests/unit/proxy/agent_endpoints/auth/test_managed_authorization.py b/tests/unit/proxy/agent_endpoints/auth/test_managed_authorization.py index 8ca11caed8e..93bc14dad8f 100644 --- a/tests/unit/proxy/agent_endpoints/auth/test_managed_authorization.py +++ b/tests/unit/proxy/agent_endpoints/auth/test_managed_authorization.py @@ -447,9 +447,6 @@ def test_managed_inference_ignores_unsupported_query_model(): "/realtime", "/v1/realtime", "/openai/v1/realtime", - "/realtime/translations", - "/v1/realtime/translations", - "/openai/v1/realtime/translations", ), ) def test_managed_realtime_requires_a_model_and_ignores_completion_defaults(route: str) -> None: @@ -465,6 +462,35 @@ def test_managed_realtime_requires_a_model_and_ignores_completion_defaults(route ) +@pytest.mark.parametrize( + "route", ("/realtime/translations", "/v1/realtime/translations", "/openai/v1/realtime/translations") +) +@pytest.mark.parametrize("query_model", (None, "requested")) +def test_managed_translation_authorizes_the_same_model_as_the_endpoint(route: str, query_model: str | None) -> None: + from litellm.proxy.agent_endpoints.auth.managed_authorization import managed_inference_request + from litellm.proxy.common_utils.http_parsing_utils import resolve_realtime_route_model + + admitted: Final = managed_inference_request( + route, {}, {"completion_model": "unrelated-default"}, "cli", query_model=query_model + ) + + assert admitted["model"] == resolve_realtime_route_model(query_model, None, is_translation=True) + assert isinstance(admitted["model"], str) and admitted["model"] + + +@pytest.mark.parametrize("route", ("/realtime", "/v1/realtime", "/openai/v1/realtime")) +def test_managed_transcription_authorizes_the_same_model_as_the_endpoint(route: str) -> None: + from litellm.proxy.agent_endpoints.auth.managed_authorization import managed_inference_request + from litellm.proxy.common_utils.http_parsing_utils import resolve_realtime_route_model + + admitted: Final = managed_inference_request( + route, {}, {"completion_model": "unrelated-default"}, "cli", intent="transcription" + ) + + assert admitted["model"] == resolve_realtime_route_model(None, "transcription", is_translation=False) + assert isinstance(admitted["model"], str) and admitted["model"] + + @pytest.mark.parametrize("mode,user", [("autonomous", None), ("delegated", "verified-human")]) def test_matching_identity_revision_and_execution_mode_pass_admission(mode: str, user: str | None) -> None: context: Final = ManagedAgentContext.model_validate( @@ -573,22 +599,41 @@ async def test_ordinary_agent_admission_preserves_legacy_authentication( @pytest.mark.parametrize( "route", - tuple(dict.fromkeys( - LiteLLMRoutes.openai_routes.value - + LiteLLMRoutes.anthropic_routes.value - + LiteLLMRoutes.google_routes.value - )), + tuple( + dict.fromkeys( + LiteLLMRoutes.openai_routes.value + LiteLLMRoutes.anthropic_routes.value + LiteLLMRoutes.google_routes.value + ) + ), ) def test_registered_inference_routes_have_an_explicit_managed_access_decision(route: str) -> None: from litellm.proxy.agent_endpoints.auth.managed_authorization import managed_agent_route_allowed normalized: Final = route.removeprefix("/openai").removeprefix("/v1beta").removeprefix("/v1") - unsupported: Final = normalized.startswith(( - "/videos", "/batches", "/files", "/fine_tuning", "/assistants", "/threads", "/utils/", - "/vector_stores", "/vector_store/", "/search", "/containers", "/skills", "/claude-code/", - "/interactions", "/agents", "/responses/{", "/responses/input_tokens", - "/realtime/client_secrets", "/realtime/calls", "/realtime/transcription_sessions", - "/realtime/translations/client_secrets", "/realtime/translations/calls", - )) or normalized in ("/models", "/cursor/models", "/cursor/v1/models") + unsupported: Final = normalized.startswith( + ( + "/videos", + "/batches", + "/files", + "/fine_tuning", + "/assistants", + "/threads", + "/utils/", + "/vector_stores", + "/vector_store/", + "/search", + "/containers", + "/skills", + "/claude-code/", + "/interactions", + "/agents", + "/responses/{", + "/responses/input_tokens", + "/realtime/client_secrets", + "/realtime/calls", + "/realtime/transcription_sessions", + "/realtime/translations/client_secrets", + "/realtime/translations/calls", + ) + ) or normalized in ("/models", "/cursor/models", "/cursor/v1/models") concrete: Final = route.split("?")[0].replace("{model}", "model").replace("{model_name:path}", "model") assert managed_agent_route_allowed(concrete, None) is not unsupported, route