diff --git a/litellm/litellm_core_utils/health_check_helpers.py b/litellm/litellm_core_utils/health_check_helpers.py index b19bcd941fa..940730eeb9f 100644 --- a/litellm/litellm_core_utils/health_check_helpers.py +++ b/litellm/litellm_core_utils/health_check_helpers.py @@ -31,7 +31,13 @@ EXTRACTION_HEALTH_CHECK_FIELDS: Final[Mapping[str, Mapping[str, str]]] = Mapping ) -def native_health_check_mode(model: str, custom_llm_provider: str | None) -> Literal["anthropic_messages"] | None: +def native_health_check_mode( + model: str, custom_llm_provider: str | None +) -> Literal["anthropic_messages", "realtime"] | None: + if custom_llm_provider == "soniox": + from litellm.llms.soniox.common_utils import soniox_health_check_mode + + return soniox_health_check_mode(model) if custom_llm_provider != "bedrock_mantle": return None from litellm.llms.bedrock_mantle.common_utils import mantle_health_check_mode diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py index 927caec3350..484bacdc72c 100644 --- a/litellm/litellm_core_utils/realtime_streaming.py +++ b/litellm/litellm_core_utils/realtime_streaming.py @@ -20,6 +20,7 @@ from litellm.constants import ( REALTIME_SESSION_SUCCESS_LOGGED_KEY, ) from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER +from litellm.llms.base_llm.realtime.transcription_protocol import RealtimeTranscriptionProtocolError from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig, RealtimeBackend from litellm.types.llms.openai import ( OpenAIRealtimeEvents, @@ -1281,24 +1282,29 @@ class RealTimeStreaming: verbose_logger.exception("Connection closed in backend to client send messages - %s", e) close: Final = backend_close_from(e) self._flush_unbilled_transcription_usage() - if self._backend_refused_session(close): - await self.log_backend_refusal(e) - else: - await self.log_messages() + await self._log_session_end(e if self._backend_refused_session(close) else None) return close except asyncio.CancelledError: self._flush_unbilled_transcription_usage() - await self.log_messages() + await self._log_session_end(None) raise except Exception as e: verbose_logger.exception("Error in backend to client send messages: %s", e) self._flush_unbilled_transcription_usage() - await self.log_messages() + await self._log_session_end(None) return BackendClose(code=1011, reason="proxy failed while relaying the upstream websocket") def _backend_refused_session(self, close: BackendClose) -> bool: return close.code != 1000 and not self._backend_sent_frames + async def _log_session_end(self, transport_refusal: Exception | None) -> None: + provider_refusal: Final = None if self.provider_config is None else self.provider_config.session_refusal() + refusal: Final = provider_refusal or transport_refusal + if refusal is None: + await self.log_messages() + return + await self.log_backend_refusal(refusal) + async def log_backend_refusal(self, error: Exception) -> None: if not self.logging_obj: return @@ -1691,7 +1697,13 @@ class RealTimeStreaming: # ``_maybe_send_guardrail_turn_detection_update`` would retry. if msg_type == "session.update" and self.provider_config is None: await self._send_session_gate_once() - sent = await self._send_to_backend(message) + try: + sent = await self._send_to_backend(message) + except RealtimeTranscriptionProtocolError as e: + await self.websocket.send_text(realtime_error_event(str(e), error_type="invalid_request_error")) + if self._uses_deferred_backend_setup(): + raise + continue if guardrail_turn_detection_injected and sent: self._guardrail_turn_detection_update_sent = True diff --git a/litellm/llms/base_llm/realtime/transformation.py b/litellm/llms/base_llm/realtime/transformation.py index 1f4ad29fa74..f1539d16b30 100644 --- a/litellm/llms/base_llm/realtime/transformation.py +++ b/litellm/llms/base_llm/realtime/transformation.py @@ -4,7 +4,6 @@ from types import TracebackType from typing import TYPE_CHECKING, Any, Protocol import httpx -from typing_extensions import Self from litellm.types.llms.openai import OpenAIRealtimeStreamSessionEvents from litellm.types.realtime import ( @@ -24,7 +23,7 @@ else: class RealtimeBackend(Protocol): - async def __aenter__(self) -> Self: ... + async def __aenter__(self) -> object: ... async def __aexit__( self, @@ -97,9 +96,15 @@ class BaseRealtimeConfig(ABC): def unbilled_usage_on_session_close(self, model: str) -> RealtimeInputAudioTranscriptionUsage | None: return None + def session_refusal(self) -> Exception | None: + return None + async def open_backend(self, url: str, headers: Mapping[str, str]) -> RealtimeBackend | None: return None + def wrap_backend(self, backend: RealtimeBackend) -> RealtimeBackend: + return backend + def transform_session_created_event( self, model: str, diff --git a/litellm/llms/custom_httpx/llm_http_handler.py b/litellm/llms/custom_httpx/llm_http_handler.py index 4c0f84a003d..8a9ff62cfae 100644 --- a/litellm/llms/custom_httpx/llm_http_handler.py +++ b/litellm/llms/custom_httpx/llm_http_handler.py @@ -6923,7 +6923,7 @@ class BaseLLMHTTPHandler: try: ssl_context: Final = realtime_ssl_for_url(url) provider_backend: Final = await provider_config.open_backend(url, headers) - backend_ws: Final = ( + backend_ws: Final = provider_config.wrap_backend( provider_backend if provider_backend is not None else await self._open_realtime_backend_ws(websockets, url, headers, ssl_context) @@ -6961,9 +6961,7 @@ class BaseLLMHTTPHandler: await backend_ws.send(_session_config) realtime_streaming.session_configuration_request = _session_config - # For providers that defer setup until client session.update, optionally - # send synthetic session.created to unblock clients waiting on connect. - if not provider_config.requires_session_configuration(): + if not _session_config: synthetic_session: Final = provider_config.transform_session_created_event( model=model, logging_session_id=logging_obj.litellm_trace_id, diff --git a/litellm/llms/soniox/common_utils.py b/litellm/llms/soniox/common_utils.py index 8d83b4c9218..796f6ca12e4 100644 --- a/litellm/llms/soniox/common_utils.py +++ b/litellm/llms/soniox/common_utils.py @@ -3,7 +3,7 @@ Shared utilities for the Soniox provider (https://soniox.com). """ from collections.abc import Mapping, Sequence -from typing import Final, TypeAlias +from typing import Final, Literal, TypeAlias from litellm.litellm_core_utils.audio_utils.subtitle_utils import ( SubtitleToken, @@ -17,6 +17,8 @@ SonioxToken: TypeAlias = Mapping[str, object] # Soniox API base URL. SONIOX_API_BASE: Final[str] = "https://api.soniox.com" +SONIOX_REALTIME_MODEL_PREFIX: Final = "stt-rt-" + # Default polling interval in seconds when waiting for an async transcription # to finish. Mirrors the Soniox SDK default. SONIOX_DEFAULT_POLL_INTERVAL: Final[float] = 1.0 @@ -54,6 +56,10 @@ class SonioxException(BaseLLMException): """Provider-specific exception class for Soniox.""" +def soniox_health_check_mode(model: str) -> Literal["realtime"] | None: + return "realtime" if model.startswith(SONIOX_REALTIME_MODEL_PREFIX) else None + + def get_soniox_api_key(api_key: str | None = None) -> str | None: """Resolve the Soniox API key from arg or env var.""" # Local import to avoid a circular import: litellm.secret_managers.main diff --git a/litellm/llms/soniox/realtime/__init__.py b/litellm/llms/soniox/realtime/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/litellm/llms/soniox/realtime/transformation.py b/litellm/llms/soniox/realtime/transformation.py new file mode 100644 index 00000000000..ed7590b6f39 --- /dev/null +++ b/litellm/llms/soniox/realtime/transformation.py @@ -0,0 +1,609 @@ +import asyncio +import time +from collections import deque +from collections.abc import Awaitable, Callable, Mapping +from itertools import chain, groupby +from types import MappingProxyType, TracebackType +from typing import Final +from urllib.parse import urlparse, urlunparse + +from pydantic import BaseModel, ConfigDict, JsonValue, ValidationError +from typing_extensions import NotRequired, ReadOnly, Required, Self, TypedDict + +from litellm import verbose_logger +from litellm._uuid import uuid +from litellm.constants import REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES +from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj +from litellm.llms.base_llm.realtime.transcription_protocol import ( + SESSION_UPDATE_EVENT_TYPES, + RealtimeTranscriptionProtocolError, + TranscriptionSessionUpdate, + completed_event, + decode_pcm16_append, + delta_event, + duration_usage, + error_event, + json_object, + parse_transcription_session_update, + speech_event, + transcription_session, + transcription_session_created_event, + uses_server_vad, +) +from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig, RealtimeBackend +from litellm.llms.custom_httpx.http_handler import realtime_ssl_for_url +from litellm.llms.soniox.common_utils import SonioxException, get_soniox_api_key +from litellm.types.llms.openai import ( + OpenAIRealtimeErrorEvent, + OpenAIRealtimeEvents, + OpenAIRealtimeInputAudioTranscriptionCompleted, + OpenAIRealtimeInputAudioTranscriptionDelta, + OpenAIRealtimeTranscriptionSessionCreated, +) +from litellm.types.realtime import ( + RealtimeInputAudioTranscriptionUsage, + RealtimeResponseTransformInput, + RealtimeResponseTypedDict, +) + +DEFAULT_SONIOX_REALTIME_URL: Final = "wss://stt-rt.soniox.com/transcribe-websocket" +WEBSOCKET_SCHEMES: Final = MappingProxyType({"wss": "wss", "ws": "ws", "https": "wss", "http": "ws"}) +DEFAULT_SAMPLE_RATE: Final = 24_000 +KEEPALIVE_INTERVAL_SECONDS: Final = 5.0 +_MAX_PENDING_FRAMES: Final = 64 +_FINALIZE: Final = '{"type":"finalize"}' +_KEEPALIVE: Final = '{"type":"keepalive"}' +_END_STREAM: Final = "" +_HEALTH_CHECK_SILENCE: Final = bytes(DEFAULT_SAMPLE_RATE // 5) +_CONTROL_FRAMES: Final = frozenset((_FINALIZE, _KEEPALIVE, _END_STREAM)) +SESSION_STARTED_FRAME: Final = '{"type":"litellm.soniox.session_started"}' +_ENDPOINT_TOKEN: Final = "" +_FINALIZED_TOKEN: Final = "" + + +class SonioxProtocolError(RealtimeTranscriptionProtocolError): + pass + + +class SonioxRealtimeOptions(BaseModel): + model_config = ConfigDict(frozen=True, extra="ignore") + + language_hints: tuple[str, ...] | None = None + language_hints_strict: bool | None = None + enable_language_identification: bool | None = None + enable_speaker_diarization: bool | None = None + context: JsonValue | None = None + translation: JsonValue | None = None + client_reference_id: str | None = None + max_endpoint_delay_ms: int | None = None + endpoint_sensitivity: float | None = None + endpoint_latency_adjustment_level: int | None = None + + +class SonioxStartRequest(SonioxRealtimeOptions): + model: str + audio_format: str + sample_rate: int + num_channels: int + enable_endpoint_detection: bool + + +class SonioxSpeechLabels(TypedDict, total=False): + speaker: ReadOnly[str] + language: ReadOnly[str] + + +class SonioxTranscriptSegment(SonioxSpeechLabels): + text: Required[ReadOnly[str]] + + +class SonioxTranscriptionDelta(OpenAIRealtimeInputAudioTranscriptionDelta, SonioxSpeechLabels): + pass + + +class SonioxTranscriptionCompleted(OpenAIRealtimeInputAudioTranscriptionCompleted): + segments: NotRequired[ReadOnly[tuple[SonioxTranscriptSegment, ...]]] + + +class _SonioxToken(BaseModel): + model_config = ConfigDict(frozen=True) + + text: str + is_final: bool = False + translation_status: str | None = None + speaker: str | None = None + language: str | None = None + + @property + def labelled(self) -> bool: + return self.speaker is not None or self.language is not None + + +class _SonioxResponse(BaseModel): + model_config = ConfigDict(frozen=True) + + tokens: tuple[_SonioxToken, ...] = () + total_audio_proc_ms: int | None = None + error_code: int | None = None + error_message: str | None = None + finished: bool = False + + +def _sample_rate(update: TranscriptionSessionUpdate | None) -> int: + audio_format: Final = None if update is None else update.audio_format + if audio_format is None: + return DEFAULT_SAMPLE_RATE + if not audio_format.is_pcm16: + raise SonioxProtocolError("Soniox realtime requires pcm16 input audio") + if audio_format.channels not in (None, 1): + raise SonioxProtocolError("Soniox realtime requires mono input audio") + if audio_format.rate is None: + return DEFAULT_SAMPLE_RATE + if audio_format.rate <= 0: + raise SonioxProtocolError("sample rate must be positive") + return audio_format.rate + + +def _language_hints( + language: str | None, configured: tuple[str, ...] | None, strict: bool | None +) -> tuple[str, ...] | None: + if not language or (strict and configured): + return configured + return (language, *(hint for hint in configured or () if hint != language)) + + +def _uses_endpoint_detection(update: TranscriptionSessionUpdate, drop_params: bool) -> bool: + return uses_server_vad(update, "Soniox realtime", drop_params, SonioxProtocolError) + + +def _ignored_settings(update: TranscriptionSessionUpdate) -> tuple[str, ...]: + vad_tuning: Final = (f"turn_detection.{key}" for key in sorted(update.turn_detection or {}) if key != "type") + return (*update.unsupported_transcription_keys, *vad_tuning) + + +def build_start_request( + model: str, update: TranscriptionSessionUpdate | None, options: SonioxRealtimeOptions, drop_params: bool = False +) -> SonioxStartRequest: + return SonioxStartRequest( + **options.model_dump(exclude={"language_hints"}), + language_hints=_language_hints( + None if update is None else update.language, options.language_hints, options.language_hints_strict + ), + model=model, + audio_format="pcm_s16le", + sample_rate=_sample_rate(update), + num_channels=1, + enable_endpoint_detection=update is None or _uses_endpoint_detection(update, drop_params), + ) + + +def _session_created(session_id: str, start_request: SonioxStartRequest) -> OpenAIRealtimeTranscriptionSessionCreated: + session: Final = transcription_session( + session_id=session_id, + model=start_request.model, + sample_rate=start_request.sample_rate, + language=start_request.language_hints[0] if start_request.language_hints else None, + server_vad=start_request.enable_endpoint_detection, + ) + return transcription_session_created_event(session) + + +def _changes_started_rate(update: TranscriptionSessionUpdate, started: SonioxStartRequest) -> bool: + audio_format: Final = update.audio_format + if audio_format is None: + return False + requested_rate: Final = _sample_rate(update) + return audio_format.rate is not None and requested_rate != started.sample_rate + + +def _changes_started_stream(update: TranscriptionSessionUpdate, started: SonioxStartRequest, drop_params: bool) -> bool: + turn_detection_set: Final = update.turn_detection is not None or update.turn_detection_disabled + return ( + _changes_started_rate(update, started) + or ( + update.language is not None + and _language_hints(update.language, started.language_hints, started.language_hints_strict) + != started.language_hints + ) + or (turn_detection_set and _uses_endpoint_detection(update, drop_params) != started.enable_endpoint_detection) + ) + + +def build_soniox_realtime_url(api_base: str | None) -> str: + if api_base is None: + return DEFAULT_SONIOX_REALTIME_URL + parsed: Final = urlparse(api_base) + scheme: Final = WEBSOCKET_SCHEMES.get(parsed.scheme) + if scheme is None: + raise ValueError("Soniox realtime api_base must be a wss://, ws://, https://, or http:// URL") + return urlunparse(parsed._replace(scheme=scheme)) + + +def _labels(token: _SonioxToken) -> SonioxSpeechLabels: + speaker: Final[SonioxSpeechLabels] = {} if token.speaker is None else {"speaker": token.speaker} + language: Final[SonioxSpeechLabels] = {} if token.language is None else {"language": token.language} + return {**speaker, **language} + + +def _delta(item_id: str, token: _SonioxToken) -> SonioxTranscriptionDelta: + return {**delta_event(item_id, token.text), **_labels(token)} + + +def _segment(run: tuple[_SonioxToken, ...]) -> SonioxTranscriptSegment: + return {**_labels(run[0]), "text": "".join(token.text for token in run).strip()} + + +def _segments(tokens: tuple[_SonioxToken, ...]) -> tuple[SonioxTranscriptSegment, ...]: + return tuple(_segment(tuple(run)) for _, run in groupby(tokens, key=lambda token: (token.speaker, token.language))) + + +def _new_item_id() -> str: + return f"item_{uuid.uuid4().hex}" + + +def _completed( + item_id: str, tokens: tuple[_SonioxToken, ...], usage: RealtimeInputAudioTranscriptionUsage | None +) -> OpenAIRealtimeInputAudioTranscriptionCompleted: + completed: Final = completed_event(item_id, "".join(token.text for token in tokens).strip(), usage) + if not any(token.labelled for token in tokens): + return completed + labelled: Final[SonioxTranscriptionCompleted] = {**completed, "segments": _segments(tokens)} + return labelled + + +def _soniox_error(response: _SonioxResponse) -> SonioxException | None: + if response.error_code is None and response.error_message is None: + return None + return SonioxException( + status_code=response.error_code or 500, + message=f"Soniox realtime error {response.error_code}: {response.error_message}", + ) + + +class SonioxEventTransformer: + def __init__(self, *, drops_originals: bool) -> None: + self._drops_originals: Final = drops_originals + self._item_id: str | None = None + self._final_tokens: Final[list[_SonioxToken]] = [] # mutable-ok: a turn's tokens append in O(1) each + self._finalizes: Final[deque[bool]] = deque() # mutable-ok: pending finalizes queue and pop in O(1) each + self._oldest_commit_answered: bool = False + self._processed_ms: int = 0 + self._billed_ms: int = 0 + self._refusal: SonioxException | None = None + + @property + def refusal(self) -> SonioxException | None: + return self._refusal + + def transform(self, payload: str) -> tuple[OpenAIRealtimeEvents, ...]: + try: + response: Final = _SonioxResponse.model_validate_json(payload) + except ValidationError: + raise SonioxProtocolError("Soniox returned an invalid realtime response") from None + self._bill(response.total_audio_proc_ms) + error: Final = _soniox_error(response) + if error is not None: + return (self._report(error),) + return tuple(chain.from_iterable(self._token_events(token) for token in response.tokens)) + + def queue_finalize(self, *, discards_turn: bool) -> None: + if discards_turn and not self._finalizes: + self._item_id = None + self._final_tokens.clear() + self._finalizes.append(discards_turn) + + def take_unbilled_usage(self, stream_ms: int = 0) -> RealtimeInputAudioTranscriptionUsage | None: + billable_ms: Final = max(self._processed_ms, stream_ms) + unbilled_ms: Final = billable_ms - self._billed_ms + if unbilled_ms <= 0: + return None + self._billed_ms = billable_ms + return duration_usage(unbilled_ms / 1000) + + def _report(self, error: SonioxException) -> OpenAIRealtimeErrorEvent: + if self._processed_ms == 0: + self._refusal = error + return error_event(error.message) + + def _bill(self, total_audio_proc_ms: int | None) -> None: + if total_audio_proc_ms is not None and total_audio_proc_ms > self._processed_ms: + self._processed_ms = total_audio_proc_ms + + def _token_events(self, token: _SonioxToken) -> tuple[OpenAIRealtimeEvents, ...]: + if token.text == _FINALIZED_TOKEN: + return self._finalized() + if self._discarding: + return () + if token.text == _ENDPOINT_TOKEN: + return self._end_turn() + started: Final = self._start() + if not token.is_final or (self._drops_originals and token.translation_status == "original"): + return started + self._final_tokens.append(token) + return (*started, _delta(self._current_item_id(), token)) + + @property + def _discarding(self) -> bool: + return bool(self._finalizes) and self._finalizes[0] + + def _finalized(self) -> tuple[OpenAIRealtimeEvents, ...]: + discarded: Final = self._finalizes.popleft() if self._finalizes else None + answered: Final = self._oldest_commit_answered + self._oldest_commit_answered = False + if discarded: + return () + if self._item_id is not None: + return self._complete(speech_stopped=False) + if discarded is None or answered: + return () + return (_completed(_new_item_id(), (), self.take_unbilled_usage()),) + + def _end_turn(self) -> tuple[OpenAIRealtimeEvents, ...]: + events: Final = self._complete(speech_stopped=True) + if events and self._finalizes: + self._oldest_commit_answered = True + return events + + def _start(self) -> tuple[OpenAIRealtimeEvents, ...]: + if self._item_id is not None: + return () + item_id: Final = _new_item_id() + self._item_id = item_id + return (speech_event("input_audio_buffer.speech_started", item_id),) + + def _current_item_id(self) -> str: + assert self._item_id is not None + return self._item_id + + def _complete(self, *, speech_stopped: bool) -> tuple[OpenAIRealtimeEvents, ...]: + item_id: Final = self._item_id + if item_id is None: + return () + tokens: Final = tuple(self._final_tokens) + self._item_id = None + self._final_tokens.clear() + stopped: Final[tuple[OpenAIRealtimeEvents, ...]] = ( + (speech_event("input_audio_buffer.speech_stopped", item_id),) if speech_stopped else () + ) + return (*stopped, _completed(item_id, tokens, self.take_unbilled_usage())) + + +class SonioxRealtimeBackend: + def __init__( + self, + inner: RealtimeBackend, + *, + keepalive_interval: float = KEEPALIVE_INTERVAL_SECONDS, + clock: Callable[[], float] = time.monotonic, + sleep: Callable[[float], Awaitable[None]] = asyncio.sleep, + ) -> None: + self._inner: Final = inner + self._interval: Final = keepalive_interval + self._clock: Final = clock + self._sleep: Final = sleep + self._frames: Final[asyncio.Queue[str | bytes | Exception]] = asyncio.Queue(maxsize=_MAX_PENDING_FRAMES) + self._last_sent: float = clock() + self._started: bool = False + self._ended: bool = False + self._tasks: tuple[asyncio.Task[None], ...] = () + + async def __aenter__(self) -> Self: + await self._inner.__aenter__() + self._tasks = (asyncio.create_task(self._pump()), asyncio.create_task(self._keepalive())) + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + self._stop() + await self._inner.__aexit__(exc_type, exc_value, traceback) + + async def send(self, message: str | bytes) -> None: + self._last_sent = self._clock() + self._ended = self._ended or message == _END_STREAM + await self._inner.send(message) + if not self._started and isinstance(message, str) and message not in _CONTROL_FRAMES: + self._started = True + await self._frames.put(SESSION_STARTED_FRAME) + + async def recv(self, decode: bool | None = None) -> str | bytes: + frame: Final = await self._frames.get() + if isinstance(frame, Exception): + raise frame + return frame + + async def close(self) -> None: + self._stop() + await self._inner.close() + + def _stop(self) -> None: + for task in self._tasks: + task.cancel() + + async def _pump(self) -> None: + while True: + try: + frame = await self._inner.recv() + except Exception as e: # noqa: BLE001 # recv() re-raises it, so an upstream failure never hangs the reader + await self._frames.put(e) + return + await self._frames.put(frame) + + async def _keepalive(self) -> None: + while not self._ended: + await self._sleep(self._interval) + if self._ended or self._clock() - self._last_sent < self._interval: + continue + try: + await self.send(_KEEPALIVE) + except Exception as e: # noqa: BLE001 # stops only keepalives; the pump reports the broken socket + verbose_logger.debug("Soniox realtime: stopping keepalive after send failed: %s", e) + return + + +class SonioxRealtimeConfig(BaseRealtimeConfig): + def __init__( + self, + *, + options: SonioxRealtimeOptions | None = None, + wrap: Callable[[RealtimeBackend], RealtimeBackend] = SonioxRealtimeBackend, + clock: Callable[[], float] = time.monotonic, + drop_params: bool = False, + ) -> None: + self._options: Final = options or SonioxRealtimeOptions() + self._drop_params: Final = drop_params + self._wrap: Final = wrap + self._clock: Final = clock + self._transformer: Final = SonioxEventTransformer(drops_originals=self._options.translation is not None) + self._start_request: SonioxStartRequest | None = None + self._session_id: str = f"sess_{uuid.uuid4().hex}" + self._started_at: float | None = None + + @classmethod + def from_litellm_params(cls, litellm_params: Mapping[str, object], *, drop_params: bool = False) -> Self: + return cls(options=SonioxRealtimeOptions.model_validate(dict(litellm_params)), drop_params=drop_params) + + def requires_session_configuration(self) -> bool: + return True + + def validate_environment( + self, + headers: dict[str, str], # mutable-ok: BaseRealtimeConfig contract + model: str, + api_key: str | None = None, + ) -> dict[str, str]: # mutable-ok: BaseRealtimeConfig contract + resolved_key: Final = get_soniox_api_key(api_key) + if not resolved_key: + raise ValueError("Missing Soniox API key. Set SONIOX_API_KEY or api_key in the deployment litellm_params") + return {**headers, "Authorization": f"Bearer {resolved_key}"} + + def get_complete_url(self, api_base: str | None, model: str, api_key: str | None = None) -> str: + return build_soniox_realtime_url(api_base) + + def wrap_backend(self, backend: RealtimeBackend) -> RealtimeBackend: + return self._wrap(backend) + + def transform_session_created_event( + self, + model: str, + logging_session_id: str, + session_configuration_request: str | None = None, + ) -> OpenAIRealtimeTranscriptionSessionCreated: + self._session_id = logging_session_id + return _session_created(logging_session_id, build_start_request(model, None, self._options)) + + def transform_realtime_request( + self, + message: str, + model: str, + session_configuration_request: str | None = None, + ) -> tuple[str | bytes, ...]: + request: Final = json_object(message, SonioxProtocolError) + event_type: Final = request.get("type") + if event_type is None: + raise SonioxProtocolError("Soniox realtime takes OpenAI realtime events, and this message has no type") + if event_type in SESSION_UPDATE_EVENT_TYPES: + update: Final = parse_transcription_session_update(message, SonioxProtocolError) + ignored: Final = _ignored_settings(update) + if ignored: + verbose_logger.warning("Soniox realtime: ignoring unsupported session settings %s", ignored) + return self._start(model, update) + if event_type == "input_audio_buffer.append": + audio: Final = decode_pcm16_append(request.get("audio"), error=SonioxProtocolError) + return (*self._start(model, None), audio) + if event_type == "input_audio_buffer.commit": + return self._finalize(model, discards_turn=False) + if event_type == "input_audio_buffer.clear": + return () if self._start_request is None else self._finalize(model, discards_turn=True) + if event_type == "input_audio_buffer.end": + return (*self._start(model, None), _END_STREAM) + verbose_logger.debug("Soniox realtime: dropping unsupported client event %s", event_type) + return () + + def _finalize(self, model: str, *, discards_turn: bool) -> tuple[str, ...]: + start: Final = self._start(model, None) + self._transformer.queue_finalize(discards_turn=discards_turn) + return (*start, _FINALIZE) + + def session_refusal(self) -> Exception | None: + return self._transformer.refusal + + def unbilled_usage_on_session_close(self, model: str) -> RealtimeInputAudioTranscriptionUsage | None: + started_at: Final = self._started_at + if self._transformer.refusal is not None: + return None + if started_at is None: + return self._transformer.take_unbilled_usage() + return self._transformer.take_unbilled_usage(int((self._clock() - started_at) * 1000)) + + def transform_realtime_response( + self, + message: str | bytes, + model: str, + logging_obj: LiteLLMLoggingObj, + realtime_response_transform_input: RealtimeResponseTransformInput, + ) -> RealtimeResponseTypedDict: + payload: Final = message.decode("utf-8") if isinstance(message, bytes) else message + result: Final[RealtimeResponseTypedDict] = { + "response": list(self._backend_events(payload)), + "current_output_item_id": realtime_response_transform_input.get("current_output_item_id"), + "current_response_id": realtime_response_transform_input.get("current_response_id"), + "current_delta_chunks": realtime_response_transform_input.get("current_delta_chunks"), + "current_conversation_id": realtime_response_transform_input.get("current_conversation_id"), + "current_item_chunks": realtime_response_transform_input.get("current_item_chunks"), + "current_delta_type": realtime_response_transform_input.get("current_delta_type"), + "session_configuration_request": realtime_response_transform_input.get("session_configuration_request"), + } + return result + + def _backend_events(self, payload: str) -> tuple[OpenAIRealtimeEvents, ...]: + if payload != SESSION_STARTED_FRAME: + return self._transformer.transform(payload) + assert self._start_request is not None + return (_session_created(self._session_id, self._start_request),) + + def _start(self, model: str, update: TranscriptionSessionUpdate | None) -> tuple[str, ...]: + started: Final = self._start_request + if started is not None: + if update is not None and _changes_started_stream(update, started, self._drop_params): + raise SonioxProtocolError("Soniox realtime can't change the session once the stream has started") + return () + start_request: Final = build_start_request(model, update, self._options, self._drop_params) + self._start_request = start_request + self._started_at = self._clock() + return (start_request.model_dump_json(exclude_none=True),) + + +def _finishes_stream(frame: str | bytes) -> bool: + response: Final = _SonioxResponse.model_validate_json(frame) + error: Final = _soniox_error(response) + if error is not None: + raise error + return response.finished + + +async def soniox_realtime_health_check( + model: str, api_base: str | None, api_key: str | None, options: SonioxRealtimeOptions +) -> bool: + import websockets + + config: Final = SonioxRealtimeConfig() + url: Final = config.get_complete_url(api_base, model) + headers: Final = config.validate_environment({}, model, api_key) + start_request: Final = build_start_request(model, None, options) + async with websockets.connect( + url, + additional_headers=headers, + max_size=REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES, + ssl=realtime_ssl_for_url(url), + ) as backend: + await backend.send(start_request.model_dump_json(exclude_none=True)) + await backend.send(_HEALTH_CHECK_SILENCE) + await backend.send(_END_STREAM) + async for frame in backend: + if _finishes_stream(frame): + return True + raise SonioxException(status_code=502, message="Soniox closed the realtime stream without finishing it") diff --git a/litellm/model_prices_and_context_window_backup.json b/litellm/model_prices_and_context_window_backup.json index 4fad4857c60..1256fef9181 100644 --- a/litellm/model_prices_and_context_window_backup.json +++ b/litellm/model_prices_and_context_window_backup.json @@ -61177,6 +61177,126 @@ ], "supports_audio_input": true }, + "soniox/stt-rt-v5": { + "input_cost_per_second": 3.333e-05, + "litellm_provider": "soniox", + "metadata": { + "calculation": "$0.12/3600 seconds = $0.00003333 per second", + "original_pricing_per_hour": 0.12 + }, + "mode": "audio_transcription", + "source": "https://soniox.com/pricing", + "supported_endpoints": [ + "/v1/realtime" + ], + "supported_modalities": [ + "audio" + ], + "supported_output_modalities": [ + "text" + ], + "supports_audio_input": true + }, + "soniox/stt-rt-preview": { + "input_cost_per_second": 3.333e-05, + "litellm_provider": "soniox", + "metadata": { + "calculation": "$0.12/3600 seconds = $0.00003333 per second", + "original_pricing_per_hour": 0.12 + }, + "mode": "audio_transcription", + "source": "https://soniox.com/pricing", + "supported_endpoints": [ + "/v1/realtime" + ], + "supported_modalities": [ + "audio" + ], + "supported_output_modalities": [ + "text" + ], + "supports_audio_input": true + }, + "soniox/stt-rt-preview-v2": { + "input_cost_per_second": 3.333e-05, + "litellm_provider": "soniox", + "metadata": { + "calculation": "$0.12/3600 seconds = $0.00003333 per second", + "original_pricing_per_hour": 0.12 + }, + "mode": "audio_transcription", + "source": "https://soniox.com/pricing", + "supported_endpoints": [ + "/v1/realtime" + ], + "supported_modalities": [ + "audio" + ], + "supported_output_modalities": [ + "text" + ], + "supports_audio_input": true + }, + "soniox/stt-rt-v3": { + "input_cost_per_second": 3.333e-05, + "litellm_provider": "soniox", + "metadata": { + "calculation": "$0.12/3600 seconds = $0.00003333 per second", + "original_pricing_per_hour": 0.12 + }, + "mode": "audio_transcription", + "source": "https://soniox.com/pricing", + "supported_endpoints": [ + "/v1/realtime" + ], + "supported_modalities": [ + "audio" + ], + "supported_output_modalities": [ + "text" + ], + "supports_audio_input": true + }, + "soniox/stt-rt-v3-preview": { + "input_cost_per_second": 3.333e-05, + "litellm_provider": "soniox", + "metadata": { + "calculation": "$0.12/3600 seconds = $0.00003333 per second", + "original_pricing_per_hour": 0.12 + }, + "mode": "audio_transcription", + "source": "https://soniox.com/pricing", + "supported_endpoints": [ + "/v1/realtime" + ], + "supported_modalities": [ + "audio" + ], + "supported_output_modalities": [ + "text" + ], + "supports_audio_input": true + }, + "soniox/stt-rt-v4": { + "input_cost_per_second": 3.333e-05, + "litellm_provider": "soniox", + "metadata": { + "calculation": "$0.12/3600 seconds = $0.00003333 per second", + "original_pricing_per_hour": 0.12 + }, + "mode": "audio_transcription", + "source": "https://soniox.com/pricing", + "supported_endpoints": [ + "/v1/realtime" + ], + "supported_modalities": [ + "audio" + ], + "supported_output_modalities": [ + "text" + ], + "supports_audio_input": true + }, "tensormesh/Qwen/Qwen3.5-397B-A17B-FP8": { "litellm_provider": "tensormesh", "mode": "chat", diff --git a/litellm/provider_endpoints_support_backup.json b/litellm/provider_endpoints_support_backup.json index 4188866de08..9f5989e765c 100644 --- a/litellm/provider_endpoints_support_backup.json +++ b/litellm/provider_endpoints_support_backup.json @@ -3724,7 +3724,7 @@ "ocr": false, "search": false, "moderations": false, - "realtime": false, + "realtime": true, "video_generations": false, "batches": false, "messages": false, diff --git a/litellm/proxy/public_endpoints/provider_create_fields.json b/litellm/proxy/public_endpoints/provider_create_fields.json index 4237517508b..04e6c5d8ba9 100644 --- a/litellm/proxy/public_endpoints/provider_create_fields.json +++ b/litellm/proxy/public_endpoints/provider_create_fields.json @@ -3612,7 +3612,7 @@ "key": "api_key", "label": "Soniox API Key", "placeholder": null, - "tooltip": "Currently only the async Speech-to-Text REST API (api.soniox.com) is supported. Realtime STT (stt-rt.soniox.com) and TTS (tts-rt.soniox.com) are not yet available.", + "tooltip": "Supports async Speech-to-Text (api.soniox.com) on /v1/audio/transcriptions and realtime Speech-to-Text (stt-rt.soniox.com) on /v1/realtime. TTS (tts-rt.soniox.com) is not yet available.", "required": true, "field_type": "password", "options": null, diff --git a/litellm/realtime_api/README.md b/litellm/realtime_api/README.md index d810de2f24f..987e8be1ed1 100644 --- a/litellm/realtime_api/README.md +++ b/litellm/realtime_api/README.md @@ -4,6 +4,6 @@ Supported endpoints: - WebSocket: `/v1/realtime` (with `intent=transcription` for transcription-only sessions) - HTTP: `/v1/realtime/client_secrets`, `/v1/realtime/transcription_sessions` -Supported providers: OpenAI, Azure OpenAI, Bedrock, Vertex AI, xAI. +Supported providers: OpenAI, Azure OpenAI, Bedrock, Vertex AI, xAI, Gemini, Meta, Soniox. For user-facing documentation and usage examples, see the litellm-docs repo. \ No newline at end of file diff --git a/litellm/realtime_api/main.py b/litellm/realtime_api/main.py index e9b534bdd89..919fd54e7d6 100644 --- a/litellm/realtime_api/main.py +++ b/litellm/realtime_api/main.py @@ -7,6 +7,7 @@ from dataclasses import dataclass from types import MappingProxyType from typing import TYPE_CHECKING, Any, Final, Literal, NoReturn, cast +from pydantic import TypeAdapter from typing_extensions import ReadOnly, TypedDict, assert_never import litellm @@ -55,6 +56,7 @@ from ..llms.openai.realtime.handler import ( confirm_session_started, ) from ..llms.openai.workload_identity import openai_static_api_key, resolve_openai_bearer_token_async +from ..llms.soniox.realtime.transformation import SonioxRealtimeOptions, soniox_realtime_health_check from ..llms.vertex_ai.audio_transcription.realtime_transformation import is_vertex_speech_to_text_model from ..llms.vertex_ai.realtime.transformation import VertexAIRealtimeConfig, vertex_realtime_config from ..llms.vertex_ai.vertex_llm_base import VertexBase @@ -75,6 +77,7 @@ vertex_llm_base: Final = VertexBase() base_llm_http_handler = BaseLLMHTTPHandler() _EMPTY_MODEL_PARAMS: Final[Mapping[str, object]] = MappingProxyType({}) _EMPTY_AUTH_HEADERS: Final[Mapping[str, str]] = MappingProxyType({}) +_DEPLOYMENT_PARAMS: Final = TypeAdapter(Mapping[str, object]) def _model_params_with_stored_credentials(model_params: Mapping[str, object]) -> Mapping[str, object]: @@ -527,6 +530,7 @@ async def _arealtime( custom_llm_provider=_custom_llm_provider, ) + deployment_params: Final = _DEPLOYMENT_PARAMS.validate_python(kwargs) if live_session_start is not None: await _alive_session( model=model, @@ -537,7 +541,7 @@ async def _arealtime( api_base=dynamic_api_base or litellm_params.api_base, api_key=dynamic_api_key, litellm_metadata=_build_litellm_metadata(kwargs), - litellm_params=kwargs, + litellm_params=deployment_params, ) return @@ -548,6 +552,7 @@ async def _arealtime( model=model, provider=LlmProviders(_custom_llm_provider), drop_params=drop_params, + litellm_params=deployment_params, ) if provider_config is not None: await base_llm_http_handler.async_realtime( @@ -882,6 +887,13 @@ async def realtime_health_check( ) elif custom_llm_provider == "openai": url = openai_realtime.construct_url(api_base=openai_api_base, query_params={"model": model}) + elif custom_llm_provider == "soniox": + return await soniox_realtime_health_check( + model=model, + api_base=resolved_api_base, + api_key=health_check_api_key, + options=SonioxRealtimeOptions.model_validate(dict(resolved_params)), + ) elif custom_llm_provider == "xai": url = xai_realtime.construct_url( api_base=resolved_api_base or "https://api.x.ai/v1", query_params={"model": model} diff --git a/litellm/utils.py b/litellm/utils.py index a656d8ebca3..342a2f12310 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -10759,6 +10759,7 @@ class ProviderConfigManager: model: str, provider: LlmProviders, drop_params: bool = False, + litellm_params: Mapping[str, object] = MappingProxyType({}), ) -> BaseRealtimeConfig | None: if LlmProviders.GEMINI == provider: from litellm.llms.gemini.realtime.transformation import GeminiRealtimeConfig @@ -10768,6 +10769,10 @@ class ProviderConfigManager: from litellm.llms.meta.realtime.transformation import MetaRealtimeConfig return MetaRealtimeConfig(drop_params=drop_params) + if LlmProviders.SONIOX == provider: + from litellm.llms.soniox.realtime.transformation import SonioxRealtimeConfig + + return SonioxRealtimeConfig.from_litellm_params(litellm_params, drop_params=drop_params) return None @staticmethod diff --git a/model_prices_and_context_window.json b/model_prices_and_context_window.json index 4fad4857c60..1256fef9181 100644 --- a/model_prices_and_context_window.json +++ b/model_prices_and_context_window.json @@ -61177,6 +61177,126 @@ ], "supports_audio_input": true }, + "soniox/stt-rt-v5": { + "input_cost_per_second": 3.333e-05, + "litellm_provider": "soniox", + "metadata": { + "calculation": "$0.12/3600 seconds = $0.00003333 per second", + "original_pricing_per_hour": 0.12 + }, + "mode": "audio_transcription", + "source": "https://soniox.com/pricing", + "supported_endpoints": [ + "/v1/realtime" + ], + "supported_modalities": [ + "audio" + ], + "supported_output_modalities": [ + "text" + ], + "supports_audio_input": true + }, + "soniox/stt-rt-preview": { + "input_cost_per_second": 3.333e-05, + "litellm_provider": "soniox", + "metadata": { + "calculation": "$0.12/3600 seconds = $0.00003333 per second", + "original_pricing_per_hour": 0.12 + }, + "mode": "audio_transcription", + "source": "https://soniox.com/pricing", + "supported_endpoints": [ + "/v1/realtime" + ], + "supported_modalities": [ + "audio" + ], + "supported_output_modalities": [ + "text" + ], + "supports_audio_input": true + }, + "soniox/stt-rt-preview-v2": { + "input_cost_per_second": 3.333e-05, + "litellm_provider": "soniox", + "metadata": { + "calculation": "$0.12/3600 seconds = $0.00003333 per second", + "original_pricing_per_hour": 0.12 + }, + "mode": "audio_transcription", + "source": "https://soniox.com/pricing", + "supported_endpoints": [ + "/v1/realtime" + ], + "supported_modalities": [ + "audio" + ], + "supported_output_modalities": [ + "text" + ], + "supports_audio_input": true + }, + "soniox/stt-rt-v3": { + "input_cost_per_second": 3.333e-05, + "litellm_provider": "soniox", + "metadata": { + "calculation": "$0.12/3600 seconds = $0.00003333 per second", + "original_pricing_per_hour": 0.12 + }, + "mode": "audio_transcription", + "source": "https://soniox.com/pricing", + "supported_endpoints": [ + "/v1/realtime" + ], + "supported_modalities": [ + "audio" + ], + "supported_output_modalities": [ + "text" + ], + "supports_audio_input": true + }, + "soniox/stt-rt-v3-preview": { + "input_cost_per_second": 3.333e-05, + "litellm_provider": "soniox", + "metadata": { + "calculation": "$0.12/3600 seconds = $0.00003333 per second", + "original_pricing_per_hour": 0.12 + }, + "mode": "audio_transcription", + "source": "https://soniox.com/pricing", + "supported_endpoints": [ + "/v1/realtime" + ], + "supported_modalities": [ + "audio" + ], + "supported_output_modalities": [ + "text" + ], + "supports_audio_input": true + }, + "soniox/stt-rt-v4": { + "input_cost_per_second": 3.333e-05, + "litellm_provider": "soniox", + "metadata": { + "calculation": "$0.12/3600 seconds = $0.00003333 per second", + "original_pricing_per_hour": 0.12 + }, + "mode": "audio_transcription", + "source": "https://soniox.com/pricing", + "supported_endpoints": [ + "/v1/realtime" + ], + "supported_modalities": [ + "audio" + ], + "supported_output_modalities": [ + "text" + ], + "supports_audio_input": true + }, "tensormesh/Qwen/Qwen3.5-397B-A17B-FP8": { "litellm_provider": "tensormesh", "mode": "chat", diff --git a/provider_endpoints_support.json b/provider_endpoints_support.json index 4188866de08..9f5989e765c 100644 --- a/provider_endpoints_support.json +++ b/provider_endpoints_support.json @@ -3724,7 +3724,7 @@ "ocr": false, "search": false, "moderations": false, - "realtime": false, + "realtime": true, "video_generations": false, "batches": false, "messages": false, diff --git a/tests/unit/litellm_core_utils/test_health_check_helpers.py b/tests/unit/litellm_core_utils/test_health_check_helpers.py index 44b36a355c6..6979d72ff14 100644 --- a/tests/unit/litellm_core_utils/test_health_check_helpers.py +++ b/tests/unit/litellm_core_utils/test_health_check_helpers.py @@ -795,9 +795,12 @@ async def test_ahealth_check_keeps_the_noul_probe_for_other_decisions_providers( ("us.anthropic.claude-haiku-4-5-20251001-v1:0", "bedrock", None), ("claude-haiku-4-5", "anthropic", None), ("anthropic.claude-haiku-4-5", None, None), + ("stt-rt-v5", "soniox", "realtime"), + ("stt-rt-preview", "soniox", "realtime"), + ("stt-async-v5", "soniox", None), ), ) -def test_native_health_check_mode_is_messages_only_for_mantle_claude( +def test_native_health_check_mode_routes_mantle_claude_to_messages_and_soniox_realtime_models_to_realtime( model: str, custom_llm_provider: str | None, expected: str | None ) -> None: assert native_health_check_mode(model=model, custom_llm_provider=custom_llm_provider) == expected diff --git a/tests/unit/litellm_core_utils/test_realtime_streaming.py b/tests/unit/litellm_core_utils/test_realtime_streaming.py index 7d85da3f756..a8543ebcfc2 100644 --- a/tests/unit/litellm_core_utils/test_realtime_streaming.py +++ b/tests/unit/litellm_core_utils/test_realtime_streaming.py @@ -24,6 +24,7 @@ from litellm.litellm_core_utils.realtime_streaming import ( client_sent_openai_beta_realtime_header, ) from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj +from litellm.llms.base_llm.realtime.transcription_protocol import RealtimeTranscriptionProtocolError from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig from litellm.llms.xai.realtime.transformation import XAIRealtimeNormalizer from litellm.llms.xai.realtime.usage_meter import XAIRealtimeUsageMeter @@ -827,9 +828,7 @@ async def test_usage_meter_snapshot_stored_in_model_call_details(): backend_ws.send = AsyncMock() backend_ws.recv = AsyncMock( side_effect=[ - json.dumps( - {"type": "session.created", "session": {"turn_detection": {"type": None}}} - ).encode(), + json.dumps({"type": "session.created", "session": {"turn_detection": {"type": None}}}).encode(), json.dumps( {"type": "response.output_audio.delta", "delta": base64.b64encode(b"x" * 24000).decode()} ).encode(), @@ -3187,6 +3186,7 @@ async def test_session_close_flushes_unbilled_transcription_usage(): provider_config: Final = MagicMock() provider_config.transform_realtime_response = MagicMock(return_value=transcript_frame) provider_config.unbilled_usage_on_session_close = MagicMock(return_value=usage) + provider_config.session_refusal = MagicMock(return_value=None) streaming: Final = RealTimeStreaming( client_ws, @@ -3320,10 +3320,14 @@ def _backend_ws_closing_with(*frames: bytes | Exception) -> MagicMock: return backend_ws -def _relay_session(client_ws: MagicMock, backend_ws: MagicMock) -> _RelaySession: +def _relay_session( + client_ws: MagicMock, backend_ws: MagicMock, provider_config: BaseRealtimeConfig | None = None +) -> _RelaySession: logging: Final = _RecordingLogging() worker: Final = _InlineLoggingWorker() - streaming: Final = RealTimeStreaming(client_ws, backend_ws, logging, model="gpt-realtime", logging_worker=worker) + streaming: Final = RealTimeStreaming( + client_ws, backend_ws, logging, provider_config=provider_config, model="gpt-realtime", logging_worker=worker + ) return _RelaySession(streaming=streaming, logging=logging, worker=worker) @@ -3408,6 +3412,103 @@ async def test_upstream_refusal_before_any_frame_logs_a_failure_not_a_success(): assert session.logging.logged_sessions == () +class _RefusingRealtimeConfig(_PassthroughRealtimeConfig): + def __init__(self, refusal: Exception | None) -> None: + self._refusal: Final = refusal + + def requires_session_configuration(self) -> bool: + return True + + def session_refusal(self) -> Exception | None: + return self._refusal + + +@pytest.mark.asyncio +async def test_a_session_the_provider_refused_in_band_logs_a_failure_even_after_a_normal_close(): + refusal: Final = ValueError("Soniox realtime error 401: Invalid API key.") + session: Final = _relay_session( + _client_ws_that_never_sends(), + _backend_ws_closing_with( + json.dumps({"type": "error", "error": {"message": str(refusal)}}).encode(), + ConnectionClosed(Close(1000, ""), None), + ), + _RefusingRealtimeConfig(refusal), + ) + + await session.run() + + assert session.logging.logged_failures == (refusal,) + assert session.logging.logged_sessions == () + + +@pytest.mark.asyncio +async def test_a_provider_session_without_a_refusal_logs_a_success_after_a_normal_close(): + session: Final = _relay_session( + _client_ws_that_never_sends(), + _backend_ws_closing_with( + json.dumps({"type": "session.created", "session": {"id": "sess_1"}}).encode(), + ConnectionClosed(Close(1000, ""), None), + ), + _RefusingRealtimeConfig(None), + ) + + await session.run() + + assert session.logging.logged_failures == () + assert len(session.logging.logged_sessions) == 1 + + +class _TranscriptionConfigRejectingSessionUpdates(_PassthroughRealtimeConfig): + def __init__(self, *, defers_setup: bool) -> None: + self._defers_setup: Final = defers_setup + + def requires_session_configuration(self) -> bool: + return not self._defers_setup + + def transform_realtime_request( + self, message: str, model: str, session_configuration_request: str | None = None + ) -> Sequence[str | bytes]: + if json.loads(message).get("type") == "session.update": + raise RealtimeTranscriptionProtocolError("requires pcm16 input audio") + return (message,) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "defers_setup,client_frames_read,audio_reaches_backend", + ((True, 1, False), (False, 3, True)), + ids=["deferred_setup_session_ends", "eager_setup_session_continues"], +) +async def test_a_rejected_transcription_session_update_sends_an_error_and_ends_only_a_deferred_setup_session( + defers_setup: bool, client_frames_read: int, audio_reaches_backend: bool +): + append: Final = json.dumps({"type": "input_audio_buffer.append", "audio": "AAE="}) + client_ws: Final = MagicMock() + client_ws.headers = {} + client_ws.receive_text = AsyncMock( + side_effect=[json.dumps({"type": "session.update", "session": {"type": "transcription"}}), append, RuntimeError] + ) + client_ws.send_text = AsyncMock() + backend_ws: Final = MagicMock() + backend_ws.send = AsyncMock() + streaming: Final = RealTimeStreaming( + client_ws, + backend_ws, + _RecordingLogging(), + provider_config=_TranscriptionConfigRejectingSessionUpdates(defers_setup=defers_setup), + model="chirp_3", + logging_worker=_InlineLoggingWorker(), + ) + + await asyncio.wait_for(streaming.client_ack_messages(), timeout=2) + + assert _error_events_sent_to(client_ws) == [ + {"type": "error", "error": {"type": "invalid_request_error", "message": "requires pcm16 input audio"}} + ] + assert client_ws.receive_text.await_count == client_frames_read + assert (append in [call.args[0] for call in backend_ws.send.await_args_list]) is audio_reaches_backend + + @pytest.mark.asyncio async def test_upstream_refusal_after_a_synthetic_session_created_still_logs_a_failure(): """LIT-6973: deferred Gemini Live setup stores a synthetic ``session.created`` before @@ -4056,9 +4157,7 @@ _XAI_TYPED_VAD_SESSION: Final = { pytest.param(({"type": "realtime", "instructions": "hi"},), id="no-turn-detection"), pytest.param(({"type": "realtime", "audio": {"input": {"turn_detection": None}}},), id="push-to-talk"), pytest.param((_XAI_TYPED_VAD_SESSION,), id="typed-server-vad"), - pytest.param( - ({"type": "realtime", "instructions": "hi"}, _XAI_TYPED_VAD_SESSION), id="later-typed-server-vad" - ), + pytest.param(({"type": "realtime", "instructions": "hi"}, _XAI_TYPED_VAD_SESSION), id="later-typed-server-vad"), ], ) async def test_xai_session_update_gate_leaves_out_the_vad_type( @@ -4070,9 +4169,9 @@ async def test_xai_session_update_gate_leaves_out_the_vad_type( monkeypatch, *client_sessions, event_normalizer=XAIRealtimeNormalizer() ) - assert [session["audio"]["input"]["turn_detection"] for session in forwarded] == [ - {"create_response": False} - ] * (len(client_sessions) + 1) + assert [session["audio"]["input"]["turn_detection"] for session in forwarded] == [{"create_response": False}] * ( + len(client_sessions) + 1 + ) @pytest.mark.asyncio @@ -4206,7 +4305,9 @@ async def test_client_never_sees_the_error_caused_by_the_guardrail_cancel(monkey await RealTimeStreaming(client_ws, backend_ws, logging_obj).backend_to_client_send_messages() client_errors: Final = [ - event["error"] for call in client_ws.send_text.await_args_list if (event := json.loads(call.args[0]))["type"] == "error" + event["error"] + for call in client_ws.send_text.await_args_list + if (event := json.loads(call.args[0]))["type"] == "error" ] assert [(error["code"], error.get("event_id")) for error in client_errors] == [ ("content_policy_violation", None), diff --git a/tests/unit/llms/custom_httpx/test_llm_http_handler.py b/tests/unit/llms/custom_httpx/test_llm_http_handler.py index d8326536d03..dca21db5b6f 100644 --- a/tests/unit/llms/custom_httpx/test_llm_http_handler.py +++ b/tests/unit/llms/custom_httpx/test_llm_http_handler.py @@ -4911,6 +4911,160 @@ async def test_async_realtime_bridges_a_transcription_session_through_the_provid assert [bytes(request.audio) for request in speech_client.requests[1:]] == [b"\x00\x01" * 800, b"\x00\x01" * 800] +class _FakeSonioxSocket: + def __init__(self) -> None: + self.sent: Final[list[str | bytes]] = [] + self._responses: Final[asyncio.Queue[str]] = asyncio.Queue() + + async def __aenter__(self) -> "_FakeSonioxSocket": + return self + + async def __aexit__(self, exc_type, exc_value, traceback) -> None: + return None + + async def send(self, message: str | bytes) -> None: + self.sent.append(message) + if message == '{"type":"finalize"}': + tokens = [{"text": "Hello", "is_final": True}, {"text": "", "is_final": True}] + await self._responses.put(json.dumps({"tokens": tokens, "total_audio_proc_ms": 1200})) + + async def recv(self, decode: bool | None = None) -> str: + return await self._responses.get() + + async def close(self) -> None: + return None + + +@pytest.mark.asyncio +async def test_async_realtime_bridges_a_soniox_transcription_through_the_wrapped_websocket(): + import websockets.exceptions # noqa: F401 # binds the submodule so async_realtime's except clause resolves, as in the proxy process + + from litellm.llms.soniox.realtime.transformation import SonioxRealtimeBackend, SonioxRealtimeConfig + + upstream = _FakeSonioxSocket() + wrapped: list[object] = [] + provider_config = SonioxRealtimeConfig( + wrap=lambda backend: wrapped.append(backend) or SonioxRealtimeBackend(backend) + ) + audio = b"\x00\x01" * 800 + client_ws = _ScriptedClientWebSocket( + [ + json.dumps( + { + "type": "session.update", + "session": { + "type": "transcription", + "audio": { + "input": { + "format": {"type": "audio/pcm", "rate": 16000}, + "transcription": {"model": "stt-rt-v5", "language": "en"}, + "turn_detection": None, + } + }, + }, + } + ), + json.dumps({"type": "input_audio_buffer.append", "audio": base64.b64encode(audio).decode()}), + json.dumps({"type": "input_audio_buffer.commit"}), + ], + last_event_type="conversation.item.input_audio_transcription.completed", + ) + logging_obj = Mock() + logging_obj.litellm_trace_id = "trace_1" + logging_obj.model_call_details = {} + logging_obj.dispatch_success_handlers = AsyncMock() + logging_obj.dispatch_failure_handlers = AsyncMock() + handler = BaseLLMHTTPHandler() + + with patch.object(handler, "_open_realtime_backend_ws", AsyncMock(return_value=upstream)) as dial: + await handler.async_realtime( + model="stt-rt-v5", + websocket=client_ws, + logging_obj=logging_obj, + provider_config=provider_config, + headers={}, + api_key="soniox-key", + query_params={"model": "stt-rt-v5", "intent": "transcription"}, + ) + + assert dial.await_args.args[1:3] == ( + "wss://stt-rt.soniox.com/transcribe-websocket", + {"Authorization": "Bearer soniox-key"}, + ) + assert wrapped == [upstream] + assert json.loads(upstream.sent[0]) == { + "model": "stt-rt-v5", + "audio_format": "pcm_s16le", + "sample_rate": 16000, + "num_channels": 1, + "enable_endpoint_detection": False, + "language_hints": ["en"], + } + assert upstream.sent[1:] == [audio, '{"type":"finalize"}'] + events = client_ws.sent_events() + assert [event["type"] for event in events] == [ + "session.created", + "session.updated", + "input_audio_buffer.speech_started", + "conversation.item.input_audio_transcription.delta", + "conversation.item.input_audio_transcription.completed", + ] + assert events[-1]["transcript"] == "Hello" + assert events[-1]["usage"] == {"type": "duration", "seconds": 1.2} + + +@pytest.mark.asyncio +async def test_async_realtime_soniox_rejects_a_bad_session_update_and_still_transcribes_audio_sent_without_one(): + import websockets.exceptions # noqa: F401 # binds the submodule so async_realtime's except clause resolves, as in the proxy process + + from litellm.llms.soniox.realtime.transformation import SonioxRealtimeConfig + + upstream = _FakeSonioxSocket() + audio = b"\x00\x01" * 800 + client_ws = _ScriptedClientWebSocket( + [ + json.dumps( + { + "type": "session.update", + "session": {"type": "transcription", "audio": {"input": {"format": {"type": "audio/pcmu"}}}}, + } + ), + json.dumps({"type": "input_audio_buffer.append", "audio": base64.b64encode(audio).decode()}), + json.dumps({"type": "input_audio_buffer.commit"}), + ], + last_event_type="conversation.item.input_audio_transcription.completed", + ) + logging_obj = Mock() + logging_obj.litellm_trace_id = "trace_1" + logging_obj.model_call_details = {} + logging_obj.dispatch_success_handlers = AsyncMock() + logging_obj.dispatch_failure_handlers = AsyncMock() + handler = BaseLLMHTTPHandler() + + with patch.object(handler, "_open_realtime_backend_ws", AsyncMock(return_value=upstream)): + await handler.async_realtime( + model="stt-rt-v5", + websocket=client_ws, + logging_obj=logging_obj, + provider_config=SonioxRealtimeConfig(), + headers={}, + api_key="soniox-key", + query_params={"model": "stt-rt-v5", "intent": "transcription"}, + ) + + assert json.loads(upstream.sent[0])["sample_rate"] == 24_000 + assert upstream.sent[1:] == [audio, '{"type":"finalize"}'] + events = client_ws.sent_events() + errors = [event for event in events if event["type"] == "error"] + assert errors == [ + { + "type": "error", + "error": {"type": "invalid_request_error", "message": "Soniox realtime requires pcm16 input audio"}, + } + ] + assert events[-1]["transcript"] == "Hello" + + @pytest.mark.asyncio async def test_responses_agentic_followup_does_not_repeat_request_params_from_plan_kwargs(monkeypatch): """A plan whose kwargs repeat a request param must not crash the Responses follow-up with a duplicate keyword""" diff --git a/tests/unit/llms/soniox/realtime/__init__.py b/tests/unit/llms/soniox/realtime/__init__.py new file mode 100644 index 00000000000..b503dff241f --- /dev/null +++ b/tests/unit/llms/soniox/realtime/__init__.py @@ -0,0 +1 @@ +"""Soniox realtime tests.""" diff --git a/tests/unit/llms/soniox/realtime/test_soniox_realtime_transformation.py b/tests/unit/llms/soniox/realtime/test_soniox_realtime_transformation.py new file mode 100644 index 00000000000..653f73ce22c --- /dev/null +++ b/tests/unit/llms/soniox/realtime/test_soniox_realtime_transformation.py @@ -0,0 +1,807 @@ +import asyncio +import base64 +import json +from typing import Final +from unittest.mock import MagicMock + +import pytest + +from litellm.llms.soniox.common_utils import SonioxException +from litellm.llms.soniox.realtime.transformation import ( + DEFAULT_SONIOX_REALTIME_URL, + SESSION_STARTED_FRAME, + SonioxEventTransformer, + SonioxRealtimeBackend, + SonioxProtocolError, + SonioxRealtimeConfig, + SonioxRealtimeOptions, +) +from litellm.types.llms.openai import OpenAIRealtimeEvents +from litellm.types.realtime import RealtimeResponseTransformInput +from litellm.types.utils import LlmProviders +from litellm.utils import ProviderConfigManager + +MODEL: Final = "stt-rt-v5" +EMPTY_TRANSFORM_INPUT: Final[RealtimeResponseTransformInput] = { + "session_configuration_request": None, + "current_output_item_id": None, + "current_response_id": None, + "current_delta_chunks": None, + "current_item_chunks": None, + "current_conversation_id": None, + "current_delta_type": None, +} +PCM: Final = b"\x01\x02" * 160 + + +def _event(event_type: str, **fields: object) -> str: + return json.dumps({"type": event_type, **fields}) + + +def _session_update(audio_input: dict[str, object]) -> str: + return _event("session.update", session={"type": "transcription", "audio": {"input": audio_input}}) + + +def _append() -> str: + return _event("input_audio_buffer.append", audio=base64.b64encode(PCM).decode()) + + +def _start_request(frames: tuple[str | bytes, ...]) -> dict[str, object]: + assert isinstance(frames[0], str) + return json.loads(frames[0]) + + +def _response(tokens: list[dict[str, object]], total_audio_proc_ms: int | None = None) -> str: + return json.dumps({"tokens": tokens, "total_audio_proc_ms": total_audio_proc_ms}) + + +def _token(text: str, is_final: bool = True, **fields: object) -> dict[str, object]: + return {"text": text, "is_final": is_final, **fields} + + +def _types(events: tuple[OpenAIRealtimeEvents, ...]) -> list[object]: + return [event["type"] for event in events] + + +def test_session_update_starts_the_stream_with_client_audio_and_deployment_options(): + config = SonioxRealtimeConfig( + options=SonioxRealtimeOptions(language_hints=("en", "fr"), enable_speaker_diarization=True) + ) + + frames = config.transform_realtime_request( + _session_update( + { + "format": {"type": "audio/pcm", "rate": 16_000}, + "transcription": {"model": "client-chosen-model", "language": "fr"}, + "turn_detection": None, + } + ), + MODEL, + ) + + assert _start_request(frames) == { + "model": MODEL, + "audio_format": "pcm_s16le", + "sample_rate": 16_000, + "num_channels": 1, + "enable_endpoint_detection": False, + "language_hints": ["fr", "en"], + "enable_speaker_diarization": True, + } + + +def test_an_empty_transcription_language_keeps_the_deployment_language_hints(): + config = SonioxRealtimeConfig(options=SonioxRealtimeOptions(language_hints=("fa",))) + + frames = config.transform_realtime_request(_session_update({"transcription": {"language": ""}}), MODEL) + + assert _start_request(frames)["language_hints"] == ["fa"] + + +def test_a_client_language_cannot_widen_strict_deployment_language_hints(): + config = SonioxRealtimeConfig(options=SonioxRealtimeOptions(language_hints=("fa",), language_hints_strict=True)) + + frames = config.transform_realtime_request(_session_update({"transcription": {"language": "en"}}), MODEL) + + assert _start_request(frames)["language_hints"] == ["fa"] + assert config.transform_realtime_request(_session_update({"transcription": {"language": "en"}}), MODEL) == () + + +def test_a_soniox_native_config_frame_is_rejected_without_starting_the_stream(): + config = SonioxRealtimeConfig() + + with pytest.raises(SonioxProtocolError): + config.transform_realtime_request(json.dumps({"model": "stt-rt-v5", "audio_format": "auto"}), MODEL) + assert _start_request(config.transform_realtime_request(_append(), MODEL))["sample_rate"] == 24_000 + + +def test_audio_before_session_update_starts_a_default_stream_once_and_rejects_a_later_change(): + config = SonioxRealtimeConfig() + + first = config.transform_realtime_request(_append(), MODEL) + second = config.transform_realtime_request(_append(), MODEL) + with pytest.raises(SonioxProtocolError, match="once the stream has started"): + config.transform_realtime_request(_session_update({"turn_detection": None}), MODEL) + + assert _start_request(first) == { + "model": MODEL, + "audio_format": "pcm_s16le", + "sample_rate": 24_000, + "num_channels": 1, + "enable_endpoint_detection": True, + } + assert first[1:] == (PCM,) + assert second == (PCM,) + assert config.transform_realtime_request(_append(), MODEL) == (PCM,) + + +@pytest.mark.parametrize( + ("event_type", "control_frame"), + [("input_audio_buffer.commit", '{"type":"finalize"}'), ("input_audio_buffer.end", "")], +) +def test_buffer_control_events_map_to_soniox_control_frames(event_type: str, control_frame: str): + config = SonioxRealtimeConfig() + config.transform_realtime_request(_append(), MODEL) + + assert config.transform_realtime_request(_event(event_type), MODEL) == (control_frame,) + + +@pytest.mark.parametrize("event_type", ["response.create", "input_audio_buffer.clear"]) +def test_events_with_nothing_to_send_upstream_are_dropped_without_starting_the_stream(event_type: str): + config = SonioxRealtimeConfig() + + assert config.transform_realtime_request(_event(event_type), MODEL) == () + assert _start_request(config.transform_realtime_request(_append(), MODEL))["model"] == MODEL + + +def _events(config: SonioxRealtimeConfig, tokens: list[dict[str, object]]) -> tuple[OpenAIRealtimeEvents, ...]: + return tuple( + config.transform_realtime_response(_response(tokens), MODEL, MagicMock(), EMPTY_TRANSFORM_INPUT)["response"] + ) + + +def test_clear_discards_the_open_turn_and_everything_soniox_finalizes_for_it(): + config = SonioxRealtimeConfig() + config.transform_realtime_request(_append(), MODEL) + _events(config, [_token("Hel")]) + + assert config.transform_realtime_request(_event("input_audio_buffer.clear"), MODEL) == ('{"type":"finalize"}',) + cleared = _events(config, [_token("lo"), _token(""), _token("")]) + config.transform_realtime_request(_event("input_audio_buffer.commit"), MODEL) + kept = _events(config, [_token("Bye"), _token("")]) + + assert cleared == () + assert kept[-1]["transcript"] == "Bye" + + +def test_clear_after_a_commit_keeps_the_committed_turn_and_drops_only_audio_sent_after_it(): + config = SonioxRealtimeConfig() + config.transform_realtime_request(_append(), MODEL) + _events(config, [_token("One")]) + config.transform_realtime_request(_event("input_audio_buffer.commit"), MODEL) + config.transform_realtime_request(_event("input_audio_buffer.clear"), MODEL) + + committed = _events(config, [_token("")]) + cleared = _events(config, [_token(" two"), _token("")]) + later = _events(config, [_token("Three"), _token("")]) + + assert committed[-1]["transcript"] == "One" + assert cleared == () + assert later[-1]["transcript"] == "Three" + + +@pytest.mark.parametrize( + "audio_input", + [ + {"format": {"type": "audio/pcmu"}}, + {"format": {"type": "audio/pcm", "rate": 24_000, "channels": 2}}, + {"format": {"type": "audio/pcm", "rate": 0}}, + ], +) +def test_unsupported_audio_formats_are_rejected_without_starting_the_stream(audio_input: dict[str, object]): + config = SonioxRealtimeConfig() + + with pytest.raises(SonioxProtocolError): + config.transform_realtime_request(_session_update(audio_input), MODEL) + assert _start_request(config.transform_realtime_request(_append(), MODEL))["sample_rate"] == 24_000 + + +@pytest.mark.parametrize( + ("audio_input", "sample_rate"), + [({"format": {"type": "audio/pcm"}}, 24_000), ({"input_audio_format": "pcm16"}, 24_000)], +) +def test_pcm16_without_a_rate_streams_at_the_openai_default_rate(audio_input: dict[str, object], sample_rate: int): + config = SonioxRealtimeConfig() + + frames = config.transform_realtime_request(_session_update(audio_input), MODEL) + + assert _start_request(frames)["sample_rate"] == sample_rate + + +def test_validate_environment_sends_the_key_as_a_bearer_header(monkeypatch): + monkeypatch.setenv("SONIOX_API_KEY", "env-key") + config = SonioxRealtimeConfig() + + assert config.validate_environment({"x": "y"}, MODEL, "deployment-key") == { + "x": "y", + "Authorization": "Bearer deployment-key", + } + assert config.validate_environment({}, MODEL) == {"Authorization": "Bearer env-key"} + + +def test_validate_environment_rejects_a_missing_key(monkeypatch): + monkeypatch.delenv("SONIOX_API_KEY", raising=False) + + with pytest.raises(ValueError, match="SONIOX_API_KEY"): + SonioxRealtimeConfig().validate_environment({}, MODEL) + + +def test_get_complete_url_defaults_to_soniox_and_opens_a_websocket_for_http_bases(): + config = SonioxRealtimeConfig() + + assert config.get_complete_url(None, MODEL) == DEFAULT_SONIOX_REALTIME_URL + assert config.get_complete_url("wss://proxy.example/ws", MODEL) == "wss://proxy.example/ws" + assert config.get_complete_url("https://proxy.example/ws?region=eu", MODEL) == "wss://proxy.example/ws?region=eu" + assert config.get_complete_url("http://localhost:8080/ws", MODEL) == "ws://localhost:8080/ws" + with pytest.raises(ValueError, match="wss://"): + config.get_complete_url("ftp://proxy.example/ws", MODEL) + + +def test_endpoint_detection_emits_one_openai_turn_billed_by_processed_audio(): + transformer = SonioxEventTransformer(drops_originals=False) + + provisional = transformer.transform(_response([_token("Hel", is_final=False)], total_audio_proc_ms=500)) + finals = transformer.transform( + _response([_token("Hello"), _token(" world"), _token(" next", is_final=False)], total_audio_proc_ms=1500) + ) + endpoint = transformer.transform(_response([_token("")], total_audio_proc_ms=2000)) + + assert _types(provisional) == ["input_audio_buffer.speech_started"] + item_id = provisional[0]["item_id"] + assert [(event["type"], event.get("delta")) for event in finals] == [ + ("conversation.item.input_audio_transcription.delta", "Hello"), + ("conversation.item.input_audio_transcription.delta", " world"), + ] + assert _types(endpoint) == [ + "input_audio_buffer.speech_stopped", + "conversation.item.input_audio_transcription.completed", + ] + assert {event["item_id"] for event in (*finals, *endpoint)} == {item_id} + assert all("speaker" not in event and "language" not in event for event in finals) + assert endpoint[1]["transcript"] == "Hello world" + assert "segments" not in endpoint[1] + assert endpoint[1]["usage"] == {"type": "duration", "seconds": 2.0} + assert transformer.take_unbilled_usage() is None + + +def test_manual_finalization_completes_the_turn_without_speech_stopped_and_starts_a_new_item(): + transformer = SonioxEventTransformer(drops_originals=False) + + first = transformer.transform(_response([_token("One"), _token("")])) + second = transformer.transform(_response([_token("Two"), _token("")])) + + assert _types(first) == [ + "input_audio_buffer.speech_started", + "conversation.item.input_audio_transcription.delta", + "conversation.item.input_audio_transcription.completed", + ] + assert first[-1]["transcript"] == "One" + assert "usage" not in first[-1] + assert second[-1]["transcript"] == "Two" + assert second[0]["item_id"] != first[0]["item_id"] + + +def test_a_commit_without_speech_answers_with_an_empty_turn_billed_for_the_silence(): + config = SonioxRealtimeConfig() + config.transform_realtime_request(_append(), MODEL) + config.transform_realtime_request(_event("input_audio_buffer.commit"), MODEL) + + events = tuple( + config.transform_realtime_response( + _response([_token("")], total_audio_proc_ms=2000), MODEL, MagicMock(), EMPTY_TRANSFORM_INPUT + )["response"] + ) + + assert _types(events) == ["conversation.item.input_audio_transcription.completed"] + assert events[0]["transcript"] == "" + assert events[0]["usage"] == {"type": "duration", "seconds": 2.0} + assert _events(config, [_token("Next"), _token("")])[0]["item_id"] != events[0]["item_id"] + + +def test_a_commit_soniox_answers_after_its_endpoint_already_ended_the_turn_adds_no_second_turn(): + config = SonioxRealtimeConfig() + config.transform_realtime_request(_append(), MODEL) + _events(config, [_token("Hello")]) + config.transform_realtime_request(_event("input_audio_buffer.commit"), MODEL) + + events = _events(config, [_token(""), _token("")]) + + assert [event["transcript"] for event in events if "transcript" in event] == ["Hello"] + + +def test_a_server_vad_turn_answers_only_the_oldest_pending_commit(): + config = SonioxRealtimeConfig() + config.transform_realtime_request(_append(), MODEL) + _events(config, [_token("Hello")]) + config.transform_realtime_request(_event("input_audio_buffer.commit"), MODEL) + config.transform_realtime_request(_append(), MODEL) + config.transform_realtime_request(_event("input_audio_buffer.commit"), MODEL) + + events = _events(config, [_token(""), _token(""), _token("")]) + + assert [event["transcript"] for event in events if "transcript" in event] == ["Hello", ""] + + +def test_a_clear_without_speech_still_answers_nothing(): + config = SonioxRealtimeConfig() + config.transform_realtime_request(_append(), MODEL) + config.transform_realtime_request(_event("input_audio_buffer.clear"), MODEL) + + assert _events(config, [_token("")]) == () + + +def test_translation_sessions_transcribe_only_the_translated_tokens(): + transformer = SonioxEventTransformer(drops_originals=True) + + events = transformer.transform( + _response( + [ + _token("Hola", translation_status="original"), + _token("Hello", translation_status="translation"), + _token(""), + ] + ) + ) + + assert [event.get("delta") for event in events if "delta" in event] == ["Hello"] + assert events[-1]["transcript"] == "Hello" + + +def test_translation_sessions_keep_speech_soniox_leaves_untranslated(): + transformer = SonioxEventTransformer(drops_originals=True) + + events = transformer.transform( + _response( + [_token("Hola", translation_status="none"), _token(" amigos", translation_status="none"), _token("")] + ) + ) + + assert events[-1]["transcript"] == "Hola amigos" + + +def test_diarized_tokens_label_each_delta_and_split_the_completed_turn_into_speaker_segments(): + transformer = SonioxEventTransformer(drops_originals=False) + + events = transformer.transform( + _response( + [ + _token("Hi", speaker="1", language="en"), + _token(" doctor", speaker="1", language="en"), + _token(" Salam", speaker="2", language="fa"), + _token(" Hi", speaker="1", language="en"), + _token(""), + ] + ) + ) + + assert [(event["delta"], event["speaker"], event["language"]) for event in events if "delta" in event] == [ + ("Hi", "1", "en"), + (" doctor", "1", "en"), + (" Salam", "2", "fa"), + (" Hi", "1", "en"), + ] + assert events[-1]["transcript"] == "Hi doctor Salam Hi" + assert events[-1]["segments"] == ( + {"speaker": "1", "language": "en", "text": "Hi doctor"}, + {"speaker": "2", "language": "fa", "text": "Salam"}, + {"speaker": "1", "language": "en", "text": "Hi"}, + ) + + +def test_translation_sessions_announce_speech_on_the_first_spoken_token_before_any_translation(): + transformer = SonioxEventTransformer(drops_originals=True) + + events = transformer.transform(_response([_token("Hol", is_final=False, translation_status="original")])) + + assert _types(events) == ["input_audio_buffer.speech_started"] + + +def test_provider_errors_surface_as_openai_error_events(): + transformer = SonioxEventTransformer(drops_originals=False) + + events = transformer.transform(json.dumps({"error_code": 401, "error_message": "Invalid API key."})) + + assert events == ( + {"type": "error", "error": {"type": "server_error", "message": "Soniox realtime error 401: Invalid API key."}}, + ) + + +def test_malformed_provider_frames_are_rejected(): + with pytest.raises(SonioxProtocolError): + SonioxEventTransformer(drops_originals=False).transform('{"tokens": "nope"}') + + +def test_audio_processed_after_the_last_turn_is_billed_on_session_close(): + config = SonioxRealtimeConfig() + config.transform_realtime_response( + _response([_token("Hi"), _token("")], total_audio_proc_ms=1000), MODEL, MagicMock(), EMPTY_TRANSFORM_INPUT + ) + config.transform_realtime_response( + _response([], total_audio_proc_ms=3500), MODEL, MagicMock(), EMPTY_TRANSFORM_INPUT + ) + + assert config.unbilled_usage_on_session_close(MODEL) == {"type": "duration", "seconds": 2.5} + assert config.unbilled_usage_on_session_close(MODEL) is None + + +def test_translation_option_switches_the_config_to_translated_transcripts(): + config = SonioxRealtimeConfig( + options=SonioxRealtimeOptions(translation={"type": "one_way", "target_language": "en"}) + ) + + result = config.transform_realtime_response( + _response([_token("Hola", translation_status="original"), _token("Hello", translation_status="translation")]), + MODEL, + MagicMock(), + EMPTY_TRANSFORM_INPUT, + ) + + assert [event.get("delta") for event in result["response"] if "delta" in event] == ["Hello"] + + +def test_stream_time_since_the_start_request_is_billed_on_close_because_soniox_charges_the_whole_stream(): + now = [100.0] + config = SonioxRealtimeConfig(clock=lambda: now[0]) + config.wrap_backend(_RecordingBackend()) + now[0] = 110.0 + config.transform_realtime_request(_append(), MODEL) + config.transform_realtime_response( + _response([_token("Hi"), _token("")], total_audio_proc_ms=5000), MODEL, MagicMock(), EMPTY_TRANSFORM_INPUT + ) + now[0] = 130.5 + + assert config.unbilled_usage_on_session_close(MODEL) == {"type": "duration", "seconds": 15.5} + assert config.unbilled_usage_on_session_close(MODEL) is None + + +def _refusal_after(processed_ms: int, error_code: int = 401) -> SonioxRealtimeConfig: + now = [100.0] + config = SonioxRealtimeConfig(clock=lambda: now[0]) + config.transform_realtime_request(_append(), MODEL) + if processed_ms: + config.transform_realtime_response( + _response([_token("Hi")], total_audio_proc_ms=processed_ms), MODEL, MagicMock(), EMPTY_TRANSFORM_INPUT + ) + config.transform_realtime_response( + json.dumps({"error_code": error_code, "error_message": "Soniox error.", "total_audio_proc_ms": processed_ms}), + MODEL, + MagicMock(), + EMPTY_TRANSFORM_INPUT, + ) + now[0] = 103.0 + return config + + +@pytest.mark.parametrize("error_code", [400, 401, 402, 403, 408, 429, 500, 503]) +def test_an_error_before_any_audio_is_processed_is_a_refused_session_that_bills_nothing(error_code: int): + config = _refusal_after(processed_ms=0, error_code=error_code) + + refusal = config.session_refusal() + assert isinstance(refusal, SonioxException) + assert refusal.status_code == error_code + assert config.unbilled_usage_on_session_close(MODEL) is None + + +def test_an_error_after_audio_was_processed_keeps_the_session_a_billed_success(): + config = _refusal_after(processed_ms=1500) + + assert config.session_refusal() is None + assert config.unbilled_usage_on_session_close(MODEL) == {"type": "duration", "seconds": 3.0} + + +def test_a_connection_that_never_started_a_stream_bills_nothing(): + now = [100.0] + config = SonioxRealtimeConfig(clock=lambda: now[0]) + config.wrap_backend(_RecordingBackend()) + now[0] = 160.0 + + assert config.unbilled_usage_on_session_close(MODEL) is None + + +def test_soniox_opts_out_of_deferred_setup_so_client_frames_reach_it_in_order(): + assert SonioxRealtimeConfig().requires_session_configuration() + + +def test_a_session_update_matching_the_started_stream_is_a_no_op(): + config = SonioxRealtimeConfig() + config.transform_realtime_request(_append(), MODEL) + + assert ( + config.transform_realtime_request(_session_update({"format": {"type": "audio/pcm", "rate": 24_000}}), MODEL) + == () + ) + + +def _stream_started_at_16khz_in_french_without_vad() -> SonioxRealtimeConfig: + config = SonioxRealtimeConfig() + config.transform_realtime_request( + _session_update( + { + "format": {"type": "audio/pcm", "rate": 16_000}, + "transcription": {"language": "fr"}, + "turn_detection": None, + } + ), + MODEL, + ) + return config + + +@pytest.mark.parametrize( + "follow_up", + [ + {"transcription": {"language": "fr"}}, + {"format": {"type": "audio/pcm", "rate": 16_000}}, + {"format": {"type": "audio/pcm"}}, + {"turn_detection": None}, + ], +) +def test_a_partial_follow_up_update_that_restates_the_stream_is_a_no_op(follow_up: dict[str, object]): + config = _stream_started_at_16khz_in_french_without_vad() + + assert config.transform_realtime_request(_session_update(follow_up), MODEL) == () + + +def test_a_follow_up_update_with_an_unsupported_encoding_is_still_rejected(): + config = _stream_started_at_16khz_in_french_without_vad() + + with pytest.raises(SonioxProtocolError, match="requires pcm16"): + config.transform_realtime_request(_session_update({"format": {"type": "audio/pcmu"}}), MODEL) + + +@pytest.mark.parametrize( + "follow_up", + [ + {"transcription": {"language": "de"}}, + {"format": {"type": "audio/pcm", "rate": 24_000}}, + {"turn_detection": {"type": "server_vad"}}, + ], +) +def test_a_partial_follow_up_update_that_changes_a_sent_setting_is_rejected(follow_up: dict[str, object]): + config = _stream_started_at_16khz_in_french_without_vad() + + with pytest.raises(SonioxProtocolError, match="once the stream has started"): + config.transform_realtime_request(_session_update(follow_up), MODEL) + + +def test_session_created_event_advertises_the_deployment_model_for_cost_tracking(): + config = SonioxRealtimeConfig(options=SonioxRealtimeOptions(language_hints=("de",))) + + event = config.transform_session_created_event(MODEL, "session_1") + + assert event["session"]["id"] == "session_1" + assert event["session"]["audio"]["input"]["transcription"] == {"model": MODEL, "language": "de"} + + +def test_backend_session_start_is_announced_with_the_negotiated_session(): + config = SonioxRealtimeConfig() + config.transform_session_created_event(MODEL, "session_1") + config.transform_realtime_request( + _session_update( + { + "format": {"type": "audio/pcm", "rate": 16_000}, + "transcription": {"language": "it"}, + "turn_detection": None, + } + ), + MODEL, + ) + + result = config.transform_realtime_response(SESSION_STARTED_FRAME, MODEL, MagicMock(), EMPTY_TRANSFORM_INPUT) + + assert _types(tuple(result["response"])) == ["session.created"] + assert result["response"][0]["session"]["id"] == "session_1" + assert result["response"][0]["session"]["audio"]["input"] == { + "format": {"type": "audio/pcm", "rate": 16_000}, + "transcription": {"model": MODEL, "language": "it"}, + "turn_detection": None, + } + + +def test_provider_config_manager_passes_only_soniox_options_from_deployment_params(): + config = ProviderConfigManager.get_provider_realtime_config( + MODEL, + LlmProviders.SONIOX, + litellm_params={ + "api_key": "secret", + "model": "soniox/stt-rt-v5", + "audio_format": "mulaw", + "language_hints": ["pl"], + "context": {"terms": ["LiteLLM"]}, + "max_endpoint_delay_ms": 800, + }, + ) + + assert isinstance(config, SonioxRealtimeConfig) + assert _start_request(config.transform_realtime_request(_append(), MODEL)) == { + "model": MODEL, + "audio_format": "pcm_s16le", + "sample_rate": 24_000, + "num_channels": 1, + "enable_endpoint_detection": True, + "language_hints": ["pl"], + "context": {"terms": ["LiteLLM"]}, + "max_endpoint_delay_ms": 800, + } + + +SEMANTIC_VAD: Final = {"turn_detection": {"type": "semantic_vad", "eagerness": "high"}} + + +def _soniox_config(drop_params: bool) -> SonioxRealtimeConfig: + config = ProviderConfigManager.get_provider_realtime_config(MODEL, LlmProviders.SONIOX, drop_params=drop_params) + assert isinstance(config, SonioxRealtimeConfig) + return config + + +def test_unsupported_turn_detection_is_rejected_without_drop_params(): + config = _soniox_config(drop_params=False) + + with pytest.raises(SonioxProtocolError, match="Set drop_params to use server_vad instead"): + config.transform_realtime_request(_session_update(SEMANTIC_VAD), MODEL) + + +def test_unsupported_turn_detection_falls_back_to_endpoint_detection_with_drop_params(): + config = _soniox_config(drop_params=True) + + frames = config.transform_realtime_request(_session_update(SEMANTIC_VAD), MODEL) + + assert _start_request(frames)["enable_endpoint_detection"] is True + assert config.transform_realtime_request(_session_update(SEMANTIC_VAD), MODEL) == () + + +def test_a_follow_up_with_unsupported_turn_detection_is_rejected_without_drop_params(): + config = _soniox_config(drop_params=False) + config.transform_realtime_request(_append(), MODEL) + + with pytest.raises(SonioxProtocolError, match="Set drop_params to use server_vad instead"): + config.transform_realtime_request(_session_update(SEMANTIC_VAD), MODEL) + + +class _RecordingBackend: + def __init__(self) -> None: + self.sent: Final[list[str | bytes]] = [] + self.upstream: Final[asyncio.Queue[str | Exception]] = asyncio.Queue() + self.entered = False + self.exited = False + self.closed = False + + async def __aenter__(self) -> "_RecordingBackend": + self.entered = True + return self + + async def __aexit__(self, exc_type, exc_value, traceback) -> None: + self.exited = True + + async def send(self, message: str | bytes) -> None: + self.sent.append(message) + + async def recv(self, decode: bool | None = None) -> str | bytes: + frame = await self.upstream.get() + if isinstance(frame, Exception): + raise frame + return frame + + async def close(self) -> None: + self.closed = True + + +class _BrokenBackend(_RecordingBackend): + async def send(self, message: str | bytes) -> None: + raise ConnectionError("upstream gone") + + +@pytest.mark.asyncio +async def test_keepalive_is_sent_only_while_the_client_is_idle_and_stops_after_end_of_stream(): + now = [0.0] + ticks: asyncio.Queue[float] = asyncio.Queue() + + async def sleep(_: float) -> None: + now[0] = await ticks.get() + + async def tick(at: float) -> None: + await ticks.put(at) + for _ in range(5): + await asyncio.sleep(0) + + inner = _RecordingBackend() + backend = SonioxRealtimeBackend(inner, keepalive_interval=5.0, clock=lambda: now[0], sleep=sleep) + + async with backend: + await tick(5.0) + assert inner.sent == ['{"type":"keepalive"}'] + now[0] = 7.0 + await backend.send(PCM) + await tick(10.0) + assert inner.sent == ['{"type":"keepalive"}', PCM] + await backend.send("") + await tick(30.0) + assert inner.sent == ['{"type":"keepalive"}', PCM, ""] + + assert inner.entered and inner.exited + + +@pytest.mark.asyncio +async def test_backend_announces_the_session_once_the_start_request_is_sent_while_relaying_upstream_frames(): + inner = _RecordingBackend() + backend = SonioxRealtimeBackend(inner) + + async with backend: + pending = asyncio.create_task(backend.recv()) + await backend.send('{"type":"keepalive"}') + await asyncio.sleep(0) + assert not pending.done() + await backend.send('{"model":"stt-rt-v5"}') + assert await pending == SESSION_STARTED_FRAME + await backend.send('{"model":"ignored-second-config"}') + await inner.upstream.put('{"tokens":[]}') + assert await backend.recv() == '{"tokens":[]}' + + +@pytest.mark.asyncio +async def test_backend_relays_upstream_frames_that_arrive_before_the_session_starts(): + inner = _RecordingBackend() + + async with SonioxRealtimeBackend(inner) as backend: + await inner.upstream.put('{"error_code":401,"error_message":"Invalid API key."}') + + assert await backend.recv() == '{"error_code":401,"error_message":"Invalid API key."}' + + +@pytest.mark.asyncio +async def test_a_cancelled_read_loses_no_frames_and_upstream_errors_reach_the_reader(): + inner = _RecordingBackend() + + async with SonioxRealtimeBackend(inner) as backend: + cancelled = asyncio.create_task(backend.recv()) + await asyncio.sleep(0) + cancelled.cancel() + await asyncio.gather(cancelled, return_exceptions=True) + await inner.upstream.put('{"tokens":[]}') + await inner.upstream.put(ConnectionError("upstream closed")) + + assert await backend.recv() == '{"tokens":[]}' + with pytest.raises(ConnectionError, match="upstream closed"): + await backend.recv() + + +@pytest.mark.asyncio +async def test_closing_the_backend_cancels_pending_reads_and_closes_upstream(): + inner = _RecordingBackend() + backend = SonioxRealtimeBackend(inner) + await backend.__aenter__() + pending = asyncio.create_task(backend.recv()) + await backend.send('{"model":"stt-rt-v5"}') + assert await pending == SESSION_STARTED_FRAME + + await backend.close() + await inner.upstream.put('{"tokens":[]}') + await asyncio.sleep(0) + + assert inner.closed + assert inner.upstream.qsize() == 1 + + +@pytest.mark.asyncio +async def test_keepalive_stops_after_the_upstream_rejects_a_send(): + sleeps: list[float] = [] + + async def sleep(seconds: float) -> None: + sleeps.append(seconds) + await asyncio.sleep(0) + + async with SonioxRealtimeBackend(_BrokenBackend(), keepalive_interval=0.0, sleep=sleep): + for _ in range(5): + await asyncio.sleep(0) + + assert sleeps == [0.0] diff --git a/tests/unit/llms/test_get_complete_url_trailing_slash.py b/tests/unit/llms/test_get_complete_url_trailing_slash.py index cd778c20d9b..43a74e56028 100644 --- a/tests/unit/llms/test_get_complete_url_trailing_slash.py +++ b/tests/unit/llms/test_get_complete_url_trailing_slash.py @@ -143,6 +143,7 @@ API_BASE_IS_THE_FULL_URL: Final[Mapping[str, str]] = MappingProxyType( "duckduckgo.search.transformation.DuckDuckGoSearchConfig", "google_pse.search.transformation.GooglePSESearchConfig", "searchapi.search.transformation.SearchAPIConfig", + "soniox.realtime.transformation.SonioxRealtimeConfig", "tinyfish.search.transformation.TinyfishSearchConfig", "vertex_ai.text_to_speech.transformation.VertexAITextToSpeechConfig", "vertex_ai.vertex_ai_partner_models.anthropic.messages.transformation." diff --git a/tests/unit/realtime_api/test_main.py b/tests/unit/realtime_api/test_main.py index 5478b807691..46777b32319 100644 --- a/tests/unit/realtime_api/test_main.py +++ b/tests/unit/realtime_api/test_main.py @@ -618,7 +618,11 @@ async def test_bearer_health_check_sends_the_secret_behind_an_oidc_credential( monkeypatch.setattr( litellm, "credential_list", - [CredentialItem(credential_name="voice-sa", credential_values={"api_key": f"oidc/file/{token}"}, credential_info={})], + [ + CredentialItem( + credential_name="voice-sa", credential_values={"api_key": f"oidc/file/{token}"}, credential_info={} + ) + ], ) connect: Final = _CapturingConnect() with patch("websockets.connect", connect): @@ -1447,3 +1451,98 @@ async def test_openai_realtime_health_check_sends_the_federated_bearer_for_a_key model_params={"model": "openai/gpt-realtime", **federated_deployment}, ) assert connect.kwargs["additional_headers"] == {"Authorization": "Bearer federated-bearer"} + + +@pytest.mark.asyncio +async def test_arealtime_routes_soniox_with_the_deployment_options(monkeypatch): + from litellm.llms.soniox.realtime.transformation import SonioxRealtimeConfig + + captured: dict[str, object] = {} + + async def mock_async_realtime(**kwargs): + captured.update(kwargs) + + monkeypatch.setattr(realtime_main.base_llm_http_handler, "async_realtime", mock_async_realtime) + await realtime_main._arealtime.__wrapped__( + model="soniox/stt-rt-v5", + websocket=MagicMock(), + litellm_logging_obj=FakeLogging(), + query_params={"model": "soniox/stt-rt-v5", "intent": "transcription"}, + api_key="deployment-key", + language_hints=["de"], + ) + + provider_config = captured["provider_config"] + assert isinstance(provider_config, SonioxRealtimeConfig) + assert captured["model"] == "stt-rt-v5" + assert captured["api_key"] == "deployment-key" + start_request = provider_config.transform_realtime_request( + json.dumps({"type": "input_audio_buffer.commit"}), "stt-rt-v5" + )[0] + assert json.loads(start_request)["language_hints"] == ["de"] + + +async def _serve_soniox(reply: str | None) -> tuple[object, list[object]]: + import websockets + + received: list[object] = [] + + async def handler(connection) -> None: + received.append(connection.request.headers["Authorization"]) + received.append(json.loads(await connection.recv())) + received.append(await connection.recv()) + received.append(await connection.recv()) + if reply is not None: + await connection.send(reply) + + return await websockets.serve(handler, "127.0.0.1", 0), received + + +@pytest.mark.asyncio +async def test_soniox_realtime_health_check_starts_and_ends_a_stream_on_the_deployment_websocket(): + server, received = await _serve_soniox(json.dumps({"tokens": [], "finished": True})) + port = next(iter(server.sockets)).getsockname()[1] + async with server: + healthy = await realtime_main.realtime_health_check( + model="stt-rt-v5", + custom_llm_provider="soniox", + api_key="soniox-key", + api_base=f"http://127.0.0.1:{port}", + model_params={"model": "soniox/stt-rt-v5", "language_hints": ["fa", "en"]}, + ) + + assert healthy is True + assert received[0] == "Bearer soniox-key" + assert received[1]["model"] == "stt-rt-v5" + assert received[1]["language_hints"] == ["fa", "en"] + assert received[2] == bytes(len(received[2])) + assert len(received[2]) > 0 + assert received[3] == "" + + +@pytest.mark.asyncio +async def test_soniox_realtime_health_check_fails_when_soniox_refuses_the_stream(): + from litellm.llms.soniox.common_utils import SonioxException + + server, _ = await _serve_soniox(json.dumps({"error_code": 401, "error_message": "Invalid API key."})) + port = next(iter(server.sockets)).getsockname()[1] + async with server: + with pytest.raises(SonioxException) as refused: + await realtime_main.realtime_health_check( + model="stt-rt-v5", custom_llm_provider="soniox", api_key="bad-key", api_base=f"ws://127.0.0.1:{port}" + ) + + assert refused.value.status_code == 401 + + +@pytest.mark.asyncio +async def test_soniox_realtime_health_check_fails_when_soniox_closes_without_finishing_the_stream(): + from litellm.llms.soniox.common_utils import SonioxException + + server, _ = await _serve_soniox(None) + port = next(iter(server.sockets)).getsockname()[1] + async with server: + with pytest.raises(SonioxException): + await realtime_main.realtime_health_check( + model="stt-rt-v5", custom_llm_provider="soniox", api_key="soniox-key", api_base=f"ws://127.0.0.1:{port}" + )