mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-30 06:43:27 +00:00
feat(soniox): add realtime speech-to-text on /v1/realtime (#44822)
* feat(soniox): add realtime speech-to-text on /v1/realtime Maps OpenAI realtime transcription sessions onto the Soniox websocket API (stt-rt-v5): server VAD uses Soniox endpoint detection, commit sends a manual finalize, deployment litellm_params pass Soniox options through, translation sessions transcribe the translated tokens, idle streams get keepalives, and cost is tracked from processed audio duration. * fix(soniox): bill realtime usage from integer milliseconds * fix(soniox): bound session-start queue and cover realtime backend edge paths * fix(soniox): bill realtime stream time, start sessions on connect and surface client protocol errors Soniox charges for the whole open stream, so session close now bills the connection's open time. The backend announces the session on connect, which flushes audio sent before any session.update, and a single upstream pump makes recv safe to cancel. Invalid client frames now get an OpenAI error event instead of a silent disconnect * fix(soniox): skip deferred setup so client frames reach Soniox in order Drops the connect marker frame, which counted as an upstream frame and hid refusals, and could start a buffered stream at the default rate before the client's session.update. Soniox now opts out of deferred setup, the handler announces a synthetic session.created whenever no setup went upstream, a session.update that would change a started stream gets an error event, and stream time is billed only once a stream started * fix(soniox): judge a follow-up session.update only on the settings it sends A partial follow-up session.update was rebuilt with defaults for the settings it left out, so restating part of a 16 kHz or VAD-off stream was rejected as a change * fix(soniox): treat an omitted rate in a follow-up session.update as unchanged * chore(soniox): drop comments that restate the code * fix(soniox): return speaker labels, log refused sessions as failures, and probe realtime health Deltas carry speaker and language when diarization or language identification is on, and the completed event lists segments per speaker and language. A session Soniox refuses before any audio is processed is logged as a failure with no spend. Billing starts at the start request. Deferred-setup providers (Meta, Vertex Chirp) send the error event and still end the session. Soniox realtime models get a websocket /health probe and cost rows for the stt-rt-v5 aliases. * fix(soniox): bill failed streams, probe health with deployment options, and keep token accumulation linear Only request rejections (400, 401, 402, 403, 429) before any processed audio count as a refused session, so a 408 or 5xx still bills the open stream. The /health probe starts Soniox with the deployment's options, sends 100 ms of silence, and passes only on Soniox's finished frame. A turn's final tokens append in constant time instead of copying the tuple per token. * fix(soniox): log every Soniox error before any processed audio as a failed session * fix(soniox): keep untranslated speech, support input_audio_buffer.clear, ignore an empty language * fix(soniox): reject frames without a type, keep strict language hints, queue finalizes in a deque * fix(soniox): answer a commit with no speech with an empty completed turn * fix(soniox): let a server VAD turn answer a commit that is still in flight * fix(soniox): let a server VAD turn answer only the oldest pending commit --------- Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
Dan Lemon
GitHub
parent
2a61f12f78
commit
cd58014315
22 changed files with 2093 additions and 34 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
0
litellm/llms/soniox/realtime/__init__.py
Normal file
0
litellm/llms/soniox/realtime/__init__.py
Normal file
609
litellm/llms/soniox/realtime/transformation.py
Normal file
609
litellm/llms/soniox/realtime/transformation.py
Normal file
|
|
@ -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 = "<end>"
|
||||
_FINALIZED_TOKEN: Final = "<fin>"
|
||||
|
||||
|
||||
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")
|
||||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -3724,7 +3724,7 @@
|
|||
"ocr": false,
|
||||
"search": false,
|
||||
"moderations": false,
|
||||
"realtime": false,
|
||||
"realtime": true,
|
||||
"video_generations": false,
|
||||
"batches": false,
|
||||
"messages": false,
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
@ -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}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -3724,7 +3724,7 @@
|
|||
"ocr": false,
|
||||
"search": false,
|
||||
"moderations": false,
|
||||
"realtime": false,
|
||||
"realtime": true,
|
||||
"video_generations": false,
|
||||
"batches": false,
|
||||
"messages": false,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
|
|
|
|||
|
|
@ -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": "<fin>", "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"""
|
||||
|
|
|
|||
1
tests/unit/llms/soniox/realtime/__init__.py
Normal file
1
tests/unit/llms/soniox/realtime/__init__.py
Normal file
|
|
@ -0,0 +1 @@
|
|||
"""Soniox realtime tests."""
|
||||
|
|
@ -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("<end>"), _token("<fin>")])
|
||||
config.transform_realtime_request(_event("input_audio_buffer.commit"), MODEL)
|
||||
kept = _events(config, [_token("Bye"), _token("<fin>")])
|
||||
|
||||
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("<fin>")])
|
||||
cleared = _events(config, [_token(" two"), _token("<fin>")])
|
||||
later = _events(config, [_token("Three"), _token("<end>")])
|
||||
|
||||
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("<end>")], 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("<fin>")]))
|
||||
second = transformer.transform(_response([_token("Two"), _token("<fin>")]))
|
||||
|
||||
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("<fin>")], 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("<end>")])[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("<end>"), _token("<fin>")])
|
||||
|
||||
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("<end>"), _token("<fin>"), _token("<fin>")])
|
||||
|
||||
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("<fin>")]) == ()
|
||||
|
||||
|
||||
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("<end>"),
|
||||
]
|
||||
)
|
||||
)
|
||||
|
||||
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("<end>")]
|
||||
)
|
||||
)
|
||||
|
||||
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("<end>"),
|
||||
]
|
||||
)
|
||||
)
|
||||
|
||||
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("<end>")], 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("<end>")], 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]
|
||||
|
|
@ -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."
|
||||
|
|
|
|||
|
|
@ -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}"
|
||||
)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue