scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-17 23:51:30 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 67

feat(realtime): add Meta Muse Voice transcription

Browse source
This commit is contained in:
Young Han 2026-09-02 13:22:47 -07:00 • committed by ryan-crabbe-berri
parent 19c8553052
commit b82b31a44f
15 changed files with 2446 additions and 57 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

87
litellm/litellm_core_utils/realtime_streaming.py
View file

@ -19,7 +19,7 @@ from litellm.types.llms.openai import (
OpenAIRealtimeStreamResponseBaseObject,
OpenAIRealtimeStreamSessionEvents,
)
from litellm.types.realtime import ALL_DELTA_TYPES
from litellm.types.realtime import ALL_DELTA_TYPES, RealtimeInputAudioTranscriptionUsage
from .litellm_logging import Logging as LiteLLMLogging
from .realtime_errors import client_close_code, realtime_error_event, websocket_close_reason
@ -116,6 +116,10 @@ class RealtimeEventNormalizer(Protocol):
def patch_outgoing_session(self, session: dict) -> dict: ...
class RealtimeUsageProvider(Protocol):
def unbilled_usage_on_session_close(self, model: str) -> RealtimeInputAudioTranscriptionUsage | None: ...
DefaultLoggedRealTimeEventTypes: Final = [
"session.created",
"response.create",
@ -139,6 +143,8 @@ class RealTimeStreaming:
force_transcription_model: str | None = None,
event_normalizer: RealtimeEventNormalizer | None = None,
logging_worker: _LoggingWorker = GLOBAL_LOGGING_WORKER,
usage_provider: RealtimeUsageProvider | None = None,
exclude_private_content_from_logs: bool = False,
):
self.websocket: _ClientWebSocket = websocket
self.backend_ws = backend_ws
@ -200,6 +206,10 @@ class RealTimeStreaming:
self._is_transcription_session: bool = force_transcription_model is not None
# Optional per-provider GA event normalizer (e.g. XAIRealtimeNormalizer).
self._event_normalizer = event_normalizer
self._usage_provider: RealtimeUsageProvider | None = (
usage_provider if usage_provider is not None else provider_config
)
self._exclude_private_content_from_logs = exclude_private_content_from_logs
# Per-connection caps for pre-setup audio frames (message count + total bytes).
_MAX_BUFFERED_MESSAGES: int = 200
@ -237,7 +247,7 @@ class RealTimeStreaming:
def _should_store_message(
self,
message_obj: dict | OpenAIRealtimeEvents,
message_obj: dict[str, Any] | OpenAIRealtimeEvents, # mutable-ok: existing realtime event contract
) -> bool:
_msg_type: Final = message_obj["type"] if "type" in message_obj else None
if self.logged_real_time_event_types == "*":
@ -246,16 +256,54 @@ class RealTimeStreaming:
return True
return False
def _message_for_logging(
self,
message_obj: dict[str, Any], # mutable-ok: existing realtime event contract
) -> dict[str, Any]: # mutable-ok: logging stores concrete event dictionaries
if not self._exclude_private_content_from_logs:
return message_obj
logged_message: dict[str, Any] = { # mutable-ok: incrementally builds the sanitized event copy
key: message_obj[key]
for key in (
"type",
"event_id",
"item_id",
"response_id",
"conversation_id",
"session_id",
"content_index",
"output_index",
"model",
"mode",
"usage",
)
if key in message_obj
}
session: Final = message_obj.get("session")
if isinstance(session, dict):
logged_session: Final[dict[str, Any]] = { # mutable-ok: sanitized JSON session snapshot
key: session[key] for key in ("id", "model", "mode", "type") if key in session
}
if logged_session:
logged_message["session"] = logged_session
return logged_message
def store_message(self, message: str | bytes | dict | OpenAIRealtimeEvents):
"""Store message in list"""
if isinstance(message, bytes):
message = message.decode("utf-8")
if isinstance(message, dict):
# TypedDict union members do not narrow to plain dict for mypy.
message_obj: dict[str, Any] = cast(dict[str, Any], message)
parsed_message_obj: dict[str, Any] = cast( # cast-ok: TypedDict events are JSON dictionaries
dict[str, Any], message
)
else:
message_obj = cast(dict[str, Any], json.loads(cast(str, message)))
self._collect_tool_calls_from_response_done(cast(dict, message_obj))
parsed_message_obj = cast( # cast-ok: parsed realtime events are JSON dictionaries
dict[str, Any], json.loads(message)
)
if not self._exclude_private_content_from_logs:
self._collect_tool_calls_from_response_done(parsed_message_obj)
message_obj: Final = self._message_for_logging(parsed_message_obj)
if not self._should_store_message(message_obj):
return
try:
@ -273,6 +321,8 @@ class RealTimeStreaming:
def _collect_user_input_from_client_event(self, message: str | dict) -> None:
"""Extract user text content from client WebSocket events for spend logging."""
if self._exclude_private_content_from_logs:
return
try:
if isinstance(message, str):
msg_obj = json.loads(message)
@ -309,6 +359,8 @@ class RealTimeStreaming:
def _collect_user_input_from_backend_event(self, event_obj: dict | OpenAIRealtimeEvents) -> None:
"""Extract user voice transcription from backend events for spend logging."""
if self._exclude_private_content_from_logs:
return
try:
event_type: Final = event_obj.get("type", "")
if event_type == "conversation.item.input_audio_transcription.completed":
@ -364,9 +416,9 @@ class RealTimeStreaming:
pass
def _flush_unbilled_transcription_usage(self) -> None:
if self.provider_config is None:
if self._usage_provider is None:
return
usage: Final = self.provider_config.unbilled_usage_on_session_close(self.model)
usage: Final = self._usage_provider.unbilled_usage_on_session_close(self.model)
if usage is None:
return
flush_event: Final = (
@ -403,12 +455,27 @@ class RealTimeStreaming:
except (AttributeError, TypeError):
pass
def _input_for_logging(
self,
message: str | dict, # mutable-ok: existing realtime input contract
) -> str | dict: # mutable-ok: logging stores concrete event dictionaries
if not self._exclude_private_content_from_logs:
return message
try:
parsed_message: Final[object] = message if isinstance(message, dict) else json.loads(message)
except (json.JSONDecodeError, TypeError):
return {} # mutable-ok: empty JSON logging payload
if not isinstance(parsed_message, dict):
return {} # mutable-ok: empty JSON logging payload
return self._message_for_logging(parsed_message)
def store_input(self, message: str | dict):
"""Store input message"""
self.input_message = message if isinstance(message, dict) else {}
logged_message: Final[str | dict] = self._input_for_logging(message) # mutable-ok: logging payload
self.input_message = logged_message if isinstance(logged_message, dict) else {}
self._collect_user_input_from_client_event(message)
if self.logging_obj:
self.logging_obj.pre_call(input=message, api_key="")
self.logging_obj.pre_call(input=logged_message, api_key="")
async def log_messages(self):
"""Log messages in list"""
@ -1009,6 +1076,8 @@ class RealTimeStreaming:
self.store_message(event_str)
self._capture_transcription_usage(event)
await self._send_event_to_client(event, event_str)
if self._is_transcription_session:
continue
blocked = await self.run_realtime_guardrails(
cast(str, transcript),
item_id=cast(str | None, event.get("item_id")),

3
litellm/llms/meta/__init__.py Normal file
View file

@ -0,0 +1,3 @@
from .realtime import MetaRealtime, MuseRealtimeAdapter
__all__ = ("MetaRealtime", "MuseRealtimeAdapter")

10
litellm/llms/meta/realtime/__init__.py Normal file
View file

@ -0,0 +1,10 @@
from .handler import MetaRealtime, MuseRealtimeAdapter
from .transformation import MuseEventTransformer, MuseProtocolError, MuseSessionConfig
__all__ = (
"MetaRealtime",
"MuseEventTransformer",
"MuseProtocolError",
"MuseRealtimeAdapter",
"MuseSessionConfig",
)

661
litellm/llms/meta/realtime/handler.py Normal file
View file

@ -0,0 +1,661 @@
from __future__ import annotations
import asyncio
import base64
import binascii
import contextlib
import json
import time
import uuid
from collections.abc import Awaitable, Callable, Mapping
from typing import Final, Protocol
from urllib.parse import urlparse, urlunparse
from pydantic import JsonValue, TypeAdapter, ValidationError
from litellm._logging import verbose_proxy_logger
from litellm.constants import REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
from litellm.litellm_core_utils.realtime_streaming import RealTimeStreaming
from litellm.llms.custom_httpx.http_handler import get_shared_realtime_ssl_context
from litellm.types.realtime import RealtimeInputAudioTranscriptionUsage, RealtimeQueryParams
from .transformation import (
MUSE_MODEL,
MuseEventTransformer,
MuseProtocolError,
MuseSessionConfig,
encode_event,
error_event,
parse_session_update,
session_created_event,
session_updated_event,
)
DEFAULT_MUSE_REALTIME_URL: Final = "wss://api.meta.ai/v1/asr/realtime"
_MAX_AUDIO_BACKLOG_SECONDS: Final = 4
_MAX_PENDING_PROVIDER_EVENTS: Final = 256
_JSON_ADAPTER: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue)
class _ProviderWebSocket(Protocol):
async def send(self, message: str | bytes) -> None: ...
async def recv(self, decode: bool | None = None) -> str | bytes: ...
async def close(self, code: int = 1000, reason: str = "") -> None: ...
class _ClientWebSocketExceptions(Protocol):
ConnectionClosed: type[Exception]
class _ClientWebSocket(Protocol):
exceptions: _ClientWebSocketExceptions
@property
def scope(self) -> Mapping[str, object]: ...
async def send_text(self, data: str) -> None: ...
async def receive_text(self) -> str: ...
async def close(self, code: int = 1000, reason: str | None = None) -> None: ...
class WebSocketConnect(Protocol):
def __call__(
self,
url: str,
*,
open_timeout: float,
max_size: int | None,
ssl: object | None,
) -> Awaitable[_ProviderWebSocket]: ...
class MuseAdapterError(RuntimeError):
def __init__(self, message: str, *, close_code: int) -> None:
super().__init__(message)
self.close_code: Final = close_code
class MuseRealtimeAdapter:
def __init__(
self,
*,
model: str,
api_key: str,
api_base: str | None = None,
timeout: float | None = None,
websocket_connect: WebSocketConnect | None = None,
monotonic: Callable[[], float] = time.monotonic,
sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
terminate_client: Callable[[int], Awaitable[None]] | None = None,
) -> None:
if model.removeprefix("meta/") != MUSE_MODEL:
raise ValueError("unsupported Meta realtime model")
self._model: Final = model.removeprefix("meta/")
self._access_token: Final = normalize_access_token(api_key)
self._url: Final = build_muse_realtime_url(api_base)
self._timeout: Final = timeout or 10.0
self._websocket_connect = websocket_connect
self._monotonic: Final = monotonic
self._sleep: Final = sleep
self._terminate_client: Final = terminate_client
self._provider_ws: _ProviderWebSocket | None = None
self._config: MuseSessionConfig | None = None
self._session_id: str = f"sess_{uuid.uuid4().hex}"
self._events: Final[asyncio.Queue[str | BaseException]] = asyncio.Queue(maxsize=_MAX_PENDING_PROVIDER_EVENTS)
self._events.put_nowait(encode_event(session_created_event(self._model, self._session_id)))
self._transformer: Final = MuseEventTransformer()
self._audio_condition: Final = asyncio.Condition()
self._pending_audio: bytearray = bytearray()
self._audio_generation: int = 0
self._flush_requested: bool = False
self._end_requested: bool = False
self._end_stream_sent: bool = False
self._audio_consumed: bool = False
self._closed: bool = False
self._resources_closed: bool = False
self._sender_task: asyncio.Task[None] | None = None
self._receiver_task: asyncio.Task[None] | None = None
self.close_code: int = 1000
self.close_reason: str = "Session closed"
async def send(self, message: str | bytes) -> None:
if self._closed:
raise MuseAdapterError("Meta Muse realtime session is closed", close_code=self.close_code)
if isinstance(message, bytes):
await self._reject("invalid_request_error", "invalid_event", "Client events must be JSON text")
return
try:
event: Final = _parse_client_event(message)
event_type: Final = event.get("type")
if event_type in ("session.update", "transcription_session.update"):
await self._handle_session_update(message)
return
if event_type == "input_audio_buffer.append":
await self._handle_audio_append(event)
return
if event_type == "input_audio_buffer.clear":
await self._clear_audio()
return
if event_type == "input_audio_buffer.commit":
await self._commit_audio()
return
if event_type == "input_audio_buffer.end":
await self._end_audio()
return
await self._emit(
error_event(
"invalid_request_error",
"unsupported_event",
f"Event type {event_type!r} is not supported for Meta Muse transcription",
)
)
except MuseProtocolError as exc:
await self._reject("invalid_request_error", "invalid_event", str(exc))
async def recv(self, decode: bool | None = None) -> str | bytes:
event: Final = await self._events.get()
if isinstance(event, BaseException):
close_code: Final = _exception_close_code(event) if isinstance(event, Exception) else 1011
if self._terminate_client is not None:
await self._terminate_client(close_code)
raise event
return event.encode("utf-8") if decode is False else event
async def close(self, code: int = 1000, reason: str = "") -> None:
if self._resources_closed:
return
self._closed = True
self._resources_closed = True
self.close_code = sanitize_close_code(code)
self.close_reason = safe_close_reason(self.close_code)
async with self._audio_condition:
self._end_requested = True
self._audio_condition.notify_all()
tasks: Final = tuple(task for task in (self._sender_task, self._receiver_task) if task is not None)
for task in tasks:
task.cancel()
if tasks:
await asyncio.gather(*tasks, return_exceptions=True)
provider_ws: Final = self._provider_ws
if provider_ws is not None:
with contextlib.suppress(Exception):
await provider_ws.close(code=self.close_code, reason=self.close_reason)
def unbilled_usage_on_session_close(self, model: str) -> RealtimeInputAudioTranscriptionUsage | None:
return self._transformer.take_unbilled_usage()
async def _handle_session_update(self, message: str) -> None:
config: Final = parse_session_update(message, self._model)
if self._config is not None:
if config != self._config:
await self._reject(
"invalid_request_error",
"session_configuration_locked",
"Meta Muse session configuration cannot change after setup",
)
return
await self._emit(session_updated_event(config, self._session_id))
return
await self._connect(config)
async def _connect(self, config: MuseSessionConfig) -> None:
connector: Final = self._websocket_connect or _default_websocket_connect
last_error: Exception | None = None # rebind-ok: records the latest bounded handshake attempt
for attempt in range(2):
provider_ws: _ProviderWebSocket | None = None
try:
provider_ws = await connector(
self._url,
open_timeout=self._timeout,
max_size=REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES,
ssl=_ssl_config(self._url),
)
await provider_ws.send(json.dumps(config.handshake(self._access_token), separators=(",", ":")))
raw_ack: str | bytes = await asyncio.wait_for( # rebind-ok: one response per handshake attempt
provider_ws.recv(), timeout=self._timeout
)
session_id: str = _parse_handshake_ack(raw_ack) # rebind-ok: one ID per handshake attempt
self._provider_ws = provider_ws
self._config = config
self._transformer.configure(config)
self._session_id = session_id
self._sender_task = asyncio.create_task(self._send_audio(), name="meta-muse-realtime-send")
self._receiver_task = asyncio.create_task(self._receive_events(), name="meta-muse-realtime-receive")
await self._emit(session_updated_event(config, session_id))
return
except asyncio.CancelledError:
if provider_ws is not None:
with contextlib.suppress(Exception):
await provider_ws.close()
raise
except Exception as exc: # noqa: BLE001 # connector implementations expose heterogeneous transport errors
last_error = exc
if provider_ws is not None:
with contextlib.suppress(Exception):
await provider_ws.close()
close_code: int = _exception_close_code(exc) # rebind-ok: classified per handshake attempt
retryable_transport_error: bool = not isinstance( # rebind-ok: classified per handshake attempt
exc, (MuseAdapterError, MuseProtocolError)
)
if attempt == 0 and retryable_transport_error and close_code in (1011, 1013):
continue
self.close_code = close_code
self.close_reason = safe_close_reason(close_code)
await self._emit(
error_event(
"server_error" if close_code != 1008 else "invalid_request_error",
"provider_connection_error",
"Meta Muse realtime handshake failed",
)
)
await self._events.put(MuseAdapterError("Meta Muse realtime handshake failed", close_code=close_code))
await self._mark_terminated(close_code)
return
assert last_error is not None
raise MuseAdapterError("Meta Muse realtime handshake failed", close_code=1011)
async def _handle_audio_append(self, event: Mapping[str, JsonValue]) -> None:
config: Final = self._require_configured()
if self._end_requested or self._end_stream_sent:
await self._reject("invalid_request_error", "input_ended", "Audio input has already ended")
return
audio_value: Final = event.get("audio")
if not isinstance(audio_value, str):
await self._reject("invalid_request_error", "invalid_audio", "Audio must be a base64 string")
return
try:
audio: Final = base64.b64decode(audio_value, validate=True)
except (binascii.Error, ValueError):
await self._reject("invalid_request_error", "invalid_audio", "Audio must be valid base64")
return
if len(audio) % 2:
await self._reject("invalid_request_error", "invalid_audio", "PCM16 audio must contain complete samples")
return
if not audio:
return
max_backlog_bytes: Final = config.bytes_per_second * _MAX_AUDIO_BACKLOG_SECONDS
if len(audio) > max_backlog_bytes:
await self._reject(
"invalid_request_error",
"audio_backlog_exceeded",
"Audio append exceeds the four-second Muse backlog limit",
)
return
async with self._audio_condition:
await self._audio_condition.wait_for(
lambda: self._closed or len(self._pending_audio) + len(audio) <= max_backlog_bytes
)
if self._closed:
raise MuseAdapterError("Meta Muse realtime session is closed", close_code=self.close_code)
self._pending_audio.extend(audio)
self._audio_condition.notify_all()
async def _clear_audio(self) -> None:
self._require_configured()
async with self._audio_condition:
self._pending_audio.clear()
self._audio_generation += 1
self._flush_requested = False
self._audio_condition.notify_all()
await self._emit(
{ # mutable-ok: OpenAI-compatible JSON event
"type": "input_audio_buffer.cleared",
"event_id": f"event_{uuid.uuid4().hex}",
}
)
async def _commit_audio(self) -> None:
config: Final = self._require_configured()
previous_item_id, item_id = self._transformer.commit_item()
async with self._audio_condition:
self._flush_requested = True
if config.mode == "PUSH_TO_TALK":
self._end_requested = True
self._audio_condition.notify_all()
await self._emit(
{ # mutable-ok: OpenAI-compatible JSON event
"type": "input_audio_buffer.committed",
"event_id": f"event_{uuid.uuid4().hex}",
"previous_item_id": previous_item_id,
"item_id": item_id,
}
)
async def _end_audio(self) -> None:
self._require_configured()
async with self._audio_condition:
self._flush_requested = True
self._end_requested = True
self._audio_condition.notify_all()
async def _send_audio(self) -> None:
config: Final = self._require_configured()
provider_ws: Final = self._require_provider_ws()
pacing_origin: float | None = None # rebind-ok: initialized when the first packet is ready
sent_duration: float = 0.0 # rebind-ok: absolute pacing clock advances after each packet
try:
while True:
packet, pacing_origin, ended = await self._next_audio_packet(
config,
pacing_origin,
sent_duration,
)
if ended:
break
if packet is None:
continue
await provider_ws.send(packet)
self._audio_consumed = True
sent_duration += len(packet) / config.bytes_per_second
await self._send_end_stream()
except asyncio.CancelledError:
raise
except Exception as exc: # noqa: BLE001 # WebSocket implementations expose heterogeneous transport errors
await self._fail_provider(exc, phase="audio send")
async def _next_audio_packet(
self,
config: MuseSessionConfig,
pacing_origin: float | None,
sent_duration: float,
) -> tuple[bytes | None, float | None, bool]:
async with self._audio_condition:
await self._audio_condition.wait_for(
lambda: (
self._closed
or len(self._pending_audio) >= config.packet_bytes
or (self._flush_requested and bool(self._pending_audio))
or (self._end_requested and not self._pending_audio)
)
)
if self._closed or (self._end_requested and not self._pending_audio):
return None, pacing_origin, True
packet_size: Final = min(config.packet_bytes, len(self._pending_audio))
if packet_size < config.packet_bytes and not self._flush_requested:
return None, pacing_origin, False
generation: Final = self._audio_generation
current_time: Final = self._monotonic()
effective_origin: Final = (
current_time - sent_duration
if pacing_origin is None or current_time > pacing_origin + sent_duration
else pacing_origin
)
deadline: Final = effective_origin + sent_duration
delay: Final = deadline - self._monotonic()
if delay > 0:
await self._sleep(delay)
async with self._audio_condition:
if generation != self._audio_generation:
return None, effective_origin, False
actual_size: Final = min(packet_size, len(self._pending_audio))
packet: Final = bytes(self._pending_audio[:actual_size])
del self._pending_audio[:actual_size]
if not self._pending_audio:
self._flush_requested = False
self._audio_condition.notify_all()
return packet or None, effective_origin, False
async def _send_end_stream(self) -> None:
if self._end_stream_sent:
return
provider_ws: Final = self._require_provider_ws()
await provider_ws.send('{"type":"endStream"}')
self._end_stream_sent = True
async def _receive_events(self) -> None:
provider_ws: Final = self._require_provider_ws()
try:
while True:
raw: str | bytes = await provider_ws.recv() # rebind-ok: one provider frame per iteration
if not isinstance(raw, str):
raise MuseProtocolError("provider returned a non-text event")
for event in self._transformer.transform(raw):
await self._emit(event)
except asyncio.CancelledError:
raise
except Exception as exc: # noqa: BLE001 # provider close exceptions vary by WebSocket implementation
close_code: Final = _exception_close_code(exc)
if close_code == 1000 and self._end_stream_sent:
await self._mark_terminated(1000)
await self._events.put(MuseAdapterError("Meta Muse realtime session completed", close_code=1000))
return
failure: Final = MuseAdapterError(
"Meta Muse realtime closed before input ended",
close_code=1011 if close_code == 1000 else close_code,
)
await self._fail_provider(failure, phase="receive")
async def _fail_provider(self, exc: Exception, *, phase: str) -> None:
close_code: Final = _exception_close_code(exc)
self.close_code = close_code
self.close_reason = safe_close_reason(close_code)
await self._emit(
error_event(
"server_error",
"provider_connection_error",
f"Meta Muse realtime {phase} failed",
)
)
await self._events.put(MuseAdapterError(f"Meta Muse realtime {phase} failed", close_code=close_code))
await self._mark_terminated(close_code)
async def _mark_terminated(self, close_code: int) -> None:
self._closed = True
self.close_code = sanitize_close_code(close_code)
self.close_reason = safe_close_reason(self.close_code)
async with self._audio_condition:
self._audio_condition.notify_all()
async def _terminate(self, close_code: int) -> None:
await self._mark_terminated(close_code)
if self._terminate_client is not None:
await self._terminate_client(self.close_code)
async def _reject(self, error_type: str, code: str, message: str) -> None:
self.close_code = 1008
self.close_reason = safe_close_reason(1008)
await self._emit(error_event(error_type, code, message))
await self._events.put(MuseAdapterError(message, close_code=1008))
await self._mark_terminated(1008)
async def _emit(self, event: Mapping[str, object]) -> None:
await self._events.put(encode_event(event))
def _require_configured(self) -> MuseSessionConfig:
if self._config is None:
raise MuseProtocolError("send session.update before audio events")
return self._config
def _require_provider_ws(self) -> _ProviderWebSocket:
if self._provider_ws is None:
raise MuseProtocolError("Meta Muse provider connection is not ready")
return self._provider_ws
class MetaRealtime:
async def async_realtime(
self,
model: str,
websocket: _ClientWebSocket,
logging_obj: LiteLLMLogging,
api_base: str | None = None,
api_key: str | None = None,
client: object | None = None,
timeout: float | None = None,
query_params: RealtimeQueryParams | None = None,
user_api_key_dict: object | None = None,
litellm_metadata: Mapping[str, object] | None = None,
websocket_connect: WebSocketConnect | None = None,
monotonic: Callable[[], float] = time.monotonic,
sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
**kwargs: object, # kwargs-ok: realtime dispatcher forwards provider-neutral options
) -> None:
if api_key is None or not api_key.strip():
await _send_client_error_and_close(websocket, "Meta Model API key is required")
return
try:
adapter: Final = MuseRealtimeAdapter(
model=model,
api_key=api_key,
api_base=api_base,
timeout=timeout,
websocket_connect=websocket_connect,
monotonic=monotonic,
sleep=sleep,
terminate_client=lambda code: _close_client(websocket, code),
)
except ValueError:
await _send_client_error_and_close(websocket, "Invalid Meta Muse realtime configuration")
return
realtime_streaming: Final = RealTimeStreaming(
websocket,
adapter, # pyright: ignore[reportArgumentType] # raw adapter intentionally matches the websocket surface
logging_obj,
model=model,
user_api_key_dict=user_api_key_dict,
request_data={ # mutable-ok: relay request metadata payload
"litellm_metadata": dict(litellm_metadata or {}) # mutable-ok: relay owns its metadata copy
},
force_transcription_model=model,
usage_provider=adapter,
exclude_private_content_from_logs=True,
)
try:
await realtime_streaming.bidirectional_forward()
except MuseAdapterError as exc:
adapter.close_code = exc.close_code
adapter.close_reason = safe_close_reason(exc.close_code)
except Exception: # noqa: BLE001 # relay errors are normalized before closing the accepted client socket
adapter.close_code = 1011
adapter.close_reason = safe_close_reason(1011)
verbose_proxy_logger.exception("Meta Muse realtime session failed")
finally:
await adapter.close(code=adapter.close_code)
await _close_client(websocket, adapter.close_code)
def normalize_access_token(api_key: str) -> str:
stripped: Final = api_key.strip()
if not stripped:
raise ValueError("Meta Model API key is required")
parts: Final = stripped.split(None, 1)
if parts[0].casefold() == "bearer":
if len(parts) != 2 or not parts[1].strip():
raise ValueError("Meta Model API key must include a token after Bearer")
return f"Bearer {parts[1].strip()}"
return f"Bearer {stripped}"
def build_muse_realtime_url(api_base: str | None) -> str:
if api_base is None:
return DEFAULT_MUSE_REALTIME_URL
parsed: Final = urlparse(api_base.strip())
scheme: Final = "wss" if parsed.scheme == "https" else parsed.scheme
if (
scheme != "wss"
or not parsed.hostname
or parsed.username is not None
or parsed.password is not None
or parsed.fragment
):
raise ValueError("Meta api_base must be an absolute wss:// or https:// URL without credentials or a fragment")
netloc: Final = f"{parsed.hostname}:{parsed.port}" if parsed.port is not None else parsed.hostname
return urlunparse((scheme, netloc, "/v1/asr/realtime", "", "", ""))
def sanitize_close_code(code: int | None) -> int:
if code is not None and code in (1000, 1008, 1011, 1013):
return code
return 1011
def safe_close_reason(code: int) -> str:
return { # mutable-ok: immutable-by-convention close-reason lookup
1000: "Session closed",
1008: "Invalid realtime transcription request",
1011: "Realtime transcription service error",
1013: "Realtime transcription service unavailable",
}.get(code, "Realtime transcription service error")
def _parse_client_event(payload: str) -> Mapping[str, JsonValue]:
try:
value: Final = _JSON_ADAPTER.validate_json(payload)
except ValidationError:
raise MuseProtocolError("invalid JSON object") from None
if not isinstance(value, dict):
raise MuseProtocolError("message must be a JSON object")
event_type: Final = value.get("type")
if not isinstance(event_type, str) or not event_type:
raise MuseProtocolError("message type must be a non-empty string")
return value
def _parse_handshake_ack(raw: str | bytes) -> str:
if not isinstance(raw, str):
raise MuseProtocolError("provider returned a non-text handshake response")
message: Final = _parse_json_object(raw)
if message.get("type") == "error":
raise MuseAdapterError("Meta Muse realtime handshake was rejected", close_code=1008)
session_id: Final = message.get("sessionId")
if not isinstance(session_id, str) or not session_id.strip():
raise MuseProtocolError("provider returned an invalid handshake response")
return session_id.strip()
def _parse_json_object(payload: str) -> Mapping[str, JsonValue]:
try:
value: Final = _JSON_ADAPTER.validate_json(payload)
except ValidationError:
raise MuseProtocolError("invalid provider JSON object") from None
if not isinstance(value, dict):
raise MuseProtocolError("provider message must be a JSON object")
return value
def _exception_close_code(exc: Exception) -> int:
code: Final = getattr(exc, "code", None)
if isinstance(exc, MuseAdapterError):
return sanitize_close_code(exc.close_code)
return sanitize_close_code(code if isinstance(code, int) else None)
def _ssl_config(url: str) -> object | None:
if not url.startswith("wss://"):
return None
config: Final = get_shared_realtime_ssl_context()
return True if config is False else config
async def _default_websocket_connect(
url: str,
*,
open_timeout: float,
max_size: int | None,
ssl: object | None,
) -> _ProviderWebSocket:
import websockets
connection: Final = await websockets.connect(
url,
open_timeout=open_timeout,
max_size=max_size,
ssl=ssl, # pyright: ignore[reportArgumentType] # shared SSL helper returns the library-supported union
)
return connection
async def _send_client_error_and_close(websocket: _ClientWebSocket, message: str) -> None:
with contextlib.suppress(Exception):
await websocket.send_text(encode_event(error_event("invalid_request_error", "invalid_configuration", message)))
await _close_client(websocket, 1008)
async def _close_client(websocket: _ClientWebSocket, code: int) -> None:
with contextlib.suppress(Exception):
await websocket.close(code=sanitize_close_code(code), reason=safe_close_reason(sanitize_close_code(code)))

619
litellm/llms/meta/realtime/transformation.py Normal file
View file

@ -0,0 +1,619 @@
from __future__ import annotations
import json
import math
import uuid
from collections import OrderedDict, deque
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Final, Literal, TypeAlias
from pydantic import JsonValue, TypeAdapter, ValidationError
from litellm.types.realtime import RealtimeInputAudioTranscriptionUsage
MUSE_MODEL: Final = "muse-voice-transcribe-1.0"
SUPPORTED_SAMPLE_RATES: Final = frozenset((16_000, 24_000))
SUPPORTED_MODES: Final = frozenset(("PUSH_TO_TALK", "ENDPOINTING", "DIARIZATION"))
SUPPORTED_LANGUAGES: Final = (
"Arabic",
"Bengali",
"Dutch",
"English",
"French",
"German",
"Hebrew",
"Hindi",
"Indonesian",
"Italian",
"Japanese",
"Kannada",
"Korean",
"Malay",
"Mandarin Chinese",
"Marathi",
"Polish",
"Portuguese",
"Spanish",
"Tagalog",
"Tamil",
"Telugu",
"Thai",
"Turkish",
"Vietnamese",
)
_LANGUAGE_NAMES: Final = { # mutable-ok: immutable-by-convention language lookup table
language.casefold(): language for language in SUPPORTED_LANGUAGES
}
_LANGUAGE_CODES: Final = { # mutable-ok: immutable-by-convention language lookup table
"ar": "Arabic",
"bn": "Bengali",
"de": "German",
"en": "English",
"es": "Spanish",
"fil": "Tagalog",
"fr": "French",
"he": "Hebrew",
"hi": "Hindi",
"id": "Indonesian",
"it": "Italian",
"iw": "Hebrew",
"ja": "Japanese",
"kn": "Kannada",
"ko": "Korean",
"ms": "Malay",
"mr": "Marathi",
"nl": "Dutch",
"pl": "Polish",
"pt": "Portuguese",
"ta": "Tamil",
"te": "Telugu",
"th": "Thai",
"tl": "Tagalog",
"tr": "Turkish",
"vi": "Vietnamese",
"zh": "Mandarin Chinese",
}
_JSON_ADAPTER: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue)
OpenAIEvent: TypeAlias = Mapping[str, object]
class MuseProtocolError(ValueError):
pass
@dataclass(frozen=True, slots=True)
class MuseSessionConfig:
model: str
mode: Literal["PUSH_TO_TALK", "ENDPOINTING", "DIARIZATION"]
sample_rate: Literal[16000, 24000]
keywords: tuple[str, ...]
language_bias: tuple[str, ...]
@property
def audio_encoding(self) -> Literal["PCM_16KHZ", "PCM_24KHZ"]:
return "PCM_16KHZ" if self.sample_rate == 16_000 else "PCM_24KHZ"
@property
def bytes_per_second(self) -> int:
return self.sample_rate * 2
@property
def packet_bytes(self) -> int:
return self.bytes_per_second * 80 // 1000
def handshake(self, access_token: str) -> Mapping[str, object]:
base: Final[Mapping[str, object]] = { # mutable-ok: JSON wire payload
"mode": self.mode,
"authorization": {"accessToken": access_token}, # mutable-ok: JSON wire payload
"audioEncoding": self.audio_encoding,
"model": self.model,
"partialMode": "CUMULATIVE",
"emitAudioProgress": True,
}
payload: dict[str, object] = dict(base) # mutable-ok: incrementally builds JSON wire payload
if self.keywords:
payload["keywords"] = list(self.keywords) # mutable-ok: JSON arrays require concrete lists
if self.language_bias:
payload["languageBias"] = list(self.language_bias) # mutable-ok: JSON arrays require concrete lists
return payload
def openai_session(self, session_id: str) -> Mapping[str, object]:
turn_detection: Final[Mapping[str, object] | None] = (
None if self.mode == "PUSH_TO_TALK" else {"type": "server_vad"} # mutable-ok: JSON wire payload
)
transcription: dict[str, object] = { # mutable-ok: incrementally builds JSON wire payload
"model": self.model,
}
if self.language_bias:
transcription["language"] = self.language_bias[0]
transcription["language_bias"] = list( # mutable-ok: JSON arrays require concrete lists
self.language_bias
)
if self.keywords:
transcription["keywords"] = list(self.keywords) # mutable-ok: JSON arrays require concrete lists
return { # mutable-ok: JSON wire payload
"id": session_id,
"object": "realtime.transcription_session",
"type": "transcription",
"model": self.model,
"audio": { # mutable-ok: JSON wire payload
"input": { # mutable-ok: JSON wire payload
"format": {"type": "audio/pcm", "rate": self.sample_rate}, # mutable-ok: JSON wire payload
"transcription": transcription,
"turn_detection": turn_detection,
}
},
}
@dataclass(slots=True)
class _TurnState:
item_id: str | None = None
started: bool = False
start_emitted: bool = False
latest_partial: str | None = None
emitted_partial: str = ""
final_text: str | None = None
completed_signal: bool = False
completed_emitted: bool = False
stopped: bool = False
stopped_emitted: bool = False
speaker: str | None = None
def _json_object(payload: str) -> Mapping[str, JsonValue]:
try:
value: Final = _JSON_ADAPTER.validate_json(payload)
except ValidationError:
raise MuseProtocolError("invalid JSON object") from None
if not isinstance(value, dict):
raise MuseProtocolError("message must be a JSON object")
return value
def _mapping(value: JsonValue | None, name: str) -> Mapping[str, JsonValue]:
if value is None:
return {} # mutable-ok: empty JSON object
if not isinstance(value, dict):
raise MuseProtocolError(f"{name} must be an object")
return value
def _string(value: JsonValue | None, name: str) -> str | None:
if value is None:
return None
if not isinstance(value, str):
raise MuseProtocolError(f"{name} must be a string")
return value
def _normalize_model(model: str) -> str:
return model.removeprefix("meta/").strip()
def normalize_language(language: str) -> str:
value: Final = language.strip()
if not value:
raise MuseProtocolError("language must be non-empty")
documented_name: Final = _LANGUAGE_NAMES.get(value.casefold())
if documented_name is not None:
return documented_name
primary: Final = value.replace("_", "-").split("-", 1)[0].casefold()
mapped_name: Final = _LANGUAGE_CODES.get(primary)
if mapped_name is None:
raise MuseProtocolError("unsupported Muse Voice language")
return mapped_name
def _normalize_string_sequence(value: JsonValue | None, name: str) -> tuple[str, ...]:
if value is None:
return ()
if not isinstance(value, list):
raise MuseProtocolError(f"{name} must be an array of strings")
normalized: list[str] = [] # mutable-ok: deduplicates validated language hints before freezing
for entry in value:
if not isinstance(entry, str) or not entry.strip():
raise MuseProtocolError(f"{name} entries must be non-empty strings")
item: str = entry.strip() # rebind-ok: normalized once for each hint
if item not in normalized:
normalized.append(item)
return tuple(normalized)
def _normalize_language_sequence(value: JsonValue | None) -> tuple[str, ...]:
return tuple(dict.fromkeys(normalize_language(item) for item in _normalize_string_sequence(value, "language_bias")))
def _parse_sample_rate(session: Mapping[str, JsonValue]) -> Literal[16000, 24000]:
beta_format: Final = session.get("input_audio_format")
audio: Final = _mapping(session.get("audio"), "session.audio")
audio_input: Final = _mapping(audio.get("input"), "session.audio.input")
ga_format: Final = audio_input.get("format")
if beta_format is not None and ga_format is not None:
raise MuseProtocolError("input audio format must use either beta or GA layout")
if beta_format is not None:
if beta_format != "pcm16":
raise MuseProtocolError("Muse Voice requires pcm16 input audio")
return 24_000
if ga_format is None:
return 24_000
if isinstance(ga_format, str):
if ga_format != "pcm16":
raise MuseProtocolError("Muse Voice requires audio/pcm input audio")
return 24_000
format_mapping: Final = _mapping(ga_format, "session.audio.input.format")
if format_mapping.get("type") != "audio/pcm":
raise MuseProtocolError("Muse Voice requires audio/pcm input audio")
channels: Final = format_mapping.get("channels", 1)
if isinstance(channels, bool) or channels != 1:
raise MuseProtocolError("Muse Voice requires mono input audio")
rate: Final = format_mapping.get("rate", 24_000)
if isinstance(rate, bool) or not isinstance(rate, int) or rate not in SUPPORTED_SAMPLE_RATES:
raise MuseProtocolError("Muse Voice supports PCM16 at 16000 Hz or 24000 Hz")
return rate
def _parse_mode(
session: Mapping[str, JsonValue], audio_input: Mapping[str, JsonValue]
) -> Literal["PUSH_TO_TALK", "ENDPOINTING", "DIARIZATION"]:
explicit: Final = session.get("mode")
if explicit is not None:
if not isinstance(explicit, str) or explicit.upper() not in SUPPORTED_MODES:
raise MuseProtocolError("unsupported Muse Voice mode")
normalized_mode: Final = explicit.upper()
if normalized_mode == "PUSH_TO_TALK":
return "PUSH_TO_TALK"
if normalized_mode == "DIARIZATION":
return "DIARIZATION"
return "ENDPOINTING"
turn_detection_present: Final = "turn_detection" in session or "turn_detection" in audio_input
turn_detection: Final = session.get("turn_detection", audio_input.get("turn_detection"))
if turn_detection_present and turn_detection is None:
return "PUSH_TO_TALK"
if turn_detection is None:
return "ENDPOINTING"
turn_detection_mapping: Final = _mapping(turn_detection, "turn_detection")
if turn_detection_mapping.get("type") not in (None, "server_vad"):
raise MuseProtocolError("Muse Voice supports server_vad turn detection or null")
return "ENDPOINTING"
def parse_session_update(payload: str, expected_model: str) -> MuseSessionConfig:
message: Final = _json_object(payload)
if message.get("type") not in ("session.update", "transcription_session.update"):
raise MuseProtocolError("expected session.update")
session: Final = _mapping(message.get("session"), "session")
if not session:
raise MuseProtocolError("session.update requires a session object")
session_type: Final = session.get("type")
if session_type not in (None, "transcription", "realtime"):
raise MuseProtocolError("Muse Voice supports transcription sessions only")
audio: Final = _mapping(session.get("audio"), "session.audio")
audio_input: Final = _mapping(audio.get("input"), "session.audio.input")
beta_transcription: Final = session.get("input_audio_transcription")
ga_transcription: Final = audio_input.get("transcription")
if beta_transcription is not None and ga_transcription is not None:
raise MuseProtocolError("input transcription must use either beta or GA layout")
transcription: Final = _mapping(
beta_transcription if beta_transcription is not None else ga_transcription,
"input audio transcription",
)
requested_model: Final = _string(transcription.get("model"), "transcription model")
normalized_model: Final = _normalize_model(expected_model)
if normalized_model != MUSE_MODEL:
raise MuseProtocolError("unsupported Meta realtime model")
if requested_model is not None and _normalize_model(requested_model) != normalized_model:
raise MuseProtocolError("realtime session model cannot be changed")
language_value: Final = _string(transcription.get("language"), "language")
explicit_bias: Final = _normalize_language_sequence(transcription.get("language_bias"))
language_bias: Final = tuple(
dict.fromkeys((normalize_language(language_value), *explicit_bias))
if language_value is not None
else explicit_bias
)
keywords: Final = _normalize_string_sequence(transcription.get("keywords"), "keywords")
return MuseSessionConfig(
model=normalized_model,
mode=_parse_mode(session, audio_input),
sample_rate=_parse_sample_rate(session),
keywords=keywords,
language_bias=language_bias,
)
def session_created_event(model: str, session_id: str) -> OpenAIEvent:
normalized_model: Final = _normalize_model(model)
default_config: Final = MuseSessionConfig(
model=normalized_model,
mode="ENDPOINTING",
sample_rate=24_000,
keywords=(),
language_bias=(),
)
return { # mutable-ok: OpenAI-compatible JSON event
"type": "session.created",
"event_id": f"event_{uuid.uuid4().hex}",
"session": default_config.openai_session(session_id),
}
def session_updated_event(config: MuseSessionConfig, session_id: str) -> OpenAIEvent:
return { # mutable-ok: OpenAI-compatible JSON event
"type": "session.updated",
"event_id": f"event_{uuid.uuid4().hex}",
"session": config.openai_session(session_id),
}
def error_event(error_type: str, code: str, message: str) -> OpenAIEvent:
return { # mutable-ok: OpenAI-compatible JSON event
"type": "error",
"event_id": f"event_{uuid.uuid4().hex}",
"error": { # mutable-ok: nested OpenAI-compatible error object
"type": error_type,
"code": code,
"message": message,
},
}
class MuseEventTransformer:
def __init__(self, *, completed_turn_limit: int = 128) -> None:
self._turns: OrderedDict[str, _TurnState] = OrderedDict() # mutable-ok: ordered active-turn state
self._active_turn_id: str | None = None
self._mode: Literal["PUSH_TO_TALK", "ENDPOINTING", "DIARIZATION"] = "ENDPOINTING"
self._completed_turn_ids: set[str] = set() # mutable-ok: bounded completed-turn membership
self._completed_turn_order: deque[str] = deque( # mutable-ok: bounded completion eviction order
maxlen=completed_turn_limit
)
self._completed_turn_limit: Final = completed_turn_limit
self._pending_item_ids: deque[str] = deque() # mutable-ok: FIFO commit correlation state
self._last_committed_item_id: str | None = None
self._last_audio_processed_ms: float = 0.0
self._unassigned_usage_seconds: float = 0.0
def configure(self, config: MuseSessionConfig) -> None:
self._mode = config.mode
def transform(self, payload: str) -> tuple[OpenAIEvent, ...]:
message: Final = _json_object(payload)
event_type: Final = message.get("type")
if event_type == "error":
return (error_event("server_error", "provider_error", "Meta Muse realtime transcription failed"),)
if event_type == "audioProgress":
self._update_audio_progress(message)
return ()
if event_type == "speechStart":
self._speech_start(message)
elif event_type == "transcript":
self._transcript(message)
elif event_type == "speaker":
self._speaker(message)
elif event_type == "speechEnd":
self._speech_end(message)
elif event_type == "speechComplete":
self._speech_complete(message)
else:
return ()
return self._drain()
def commit_item(self) -> tuple[str | None, str]:
previous_item_id: Final = self._last_committed_item_id
provider_turn_id: Final = self._active_turn_id
active_turn: Final = self._turns.get(provider_turn_id) if provider_turn_id is not None else None
item_id: Final = (
active_turn.item_id or provider_turn_id
if active_turn is not None and provider_turn_id is not None
else f"item_{uuid.uuid4().hex}"
)
if active_turn is not None:
active_turn.item_id = item_id
else:
self._pending_item_ids.append(item_id)
self._last_committed_item_id = item_id
return previous_item_id, item_id
def take_unbilled_usage(self) -> RealtimeInputAudioTranscriptionUsage | None:
seconds: Final = self._unassigned_usage_seconds
if seconds <= 0:
return None
self._unassigned_usage_seconds = 0.0
return {"type": "duration", "seconds": seconds} # mutable-ok: typed usage wire payload
def _turn(self, turn_id: str) -> _TurnState:
if turn_id in self._completed_turn_ids:
raise _CompletedTurn
turn: Final = self._turns.get(turn_id)
if turn is not None:
return turn
created: Final = _TurnState(item_id=self._pending_item_ids.popleft() if self._pending_item_ids else turn_id)
self._turns[turn_id] = created
return created
def _speech_start(self, message: Mapping[str, JsonValue]) -> None:
turn_id: Final = self._required_turn_id(message, "speechStart")
try:
turn: Final = self._turn(turn_id)
except _CompletedTurn:
return
turn.started = True
self._active_turn_id = turn_id
def _transcript(self, message: Mapping[str, JsonValue]) -> None:
transcript: Final = message.get("transcript")
if not isinstance(transcript, str):
raise MuseProtocolError("transcript event has invalid transcript")
if not transcript and message.get("turnId") is None and self._active_turn_id is None:
return
turn_id: Final = self._transcript_turn_id(message)
try:
turn: Final = self._turn(turn_id)
except _CompletedTurn:
return
final: Final = message.get("final") is True
if final:
turn.final_text = transcript
turn.completed_signal = True
if self._mode == "PUSH_TO_TALK":
turn.stopped = True
if self._active_turn_id == turn_id:
self._active_turn_id = None
return
if turn.final_text is None:
turn.latest_partial = transcript
def _speaker(self, message: Mapping[str, JsonValue]) -> None:
turn_id: Final = (
self._required_turn_id(message, "speaker") if message.get("turnId") is not None else self._active_turn_id
)
if turn_id is None:
raise MuseProtocolError("speaker event arrived outside an active turn")
label: Final = message.get("label")
if not isinstance(label, str) or not label.strip():
raise MuseProtocolError("speaker event has invalid label")
try:
turn: Final = self._turn(turn_id)
except _CompletedTurn:
return
turn.speaker = label.strip()
def _speech_end(self, message: Mapping[str, JsonValue]) -> None:
turn_id: Final = self._required_turn_id(message, "speechEnd")
try:
turn: Final = self._turn(turn_id)
except _CompletedTurn:
return
turn.stopped = True
if self._active_turn_id == turn_id:
self._active_turn_id = None
def _speech_complete(self, message: Mapping[str, JsonValue]) -> None:
turn_id: Final = self._required_turn_id(message, "speechComplete")
transcript: Final = message.get("transcript")
if not isinstance(transcript, str):
raise MuseProtocolError("speechComplete event has invalid transcript")
try:
turn: Final = self._turn(turn_id)
except _CompletedTurn:
return
turn.final_text = transcript
turn.completed_signal = True
def _update_audio_progress(self, message: Mapping[str, JsonValue]) -> None:
processed_ms: Final = message.get("audioProcessedMs")
if (
isinstance(processed_ms, bool)
or not isinstance(processed_ms, (int, float))
or not math.isfinite(processed_ms)
or processed_ms < 0
):
raise MuseProtocolError("audioProgress event has invalid audioProcessedMs")
if processed_ms <= self._last_audio_processed_ms:
return
self._unassigned_usage_seconds += (float(processed_ms) - self._last_audio_processed_ms) / 1000
self._last_audio_processed_ms = float(processed_ms)
def _drain(self) -> tuple[OpenAIEvent, ...]:
events: list[OpenAIEvent] = [] # mutable-ok: ordered events are frozen to a tuple before return
while self._turns:
turn_id: str = next(iter(self._turns)) # rebind-ok: selects the next ordered turn
turn: _TurnState = self._turns[turn_id] # rebind-ok: state for the selected turn
has_content: bool = ( # rebind-ok: evaluated for the selected turn
turn.latest_partial is not None or turn.final_text is not None
)
item_id: str = turn.item_id or turn_id # rebind-ok: selected for each ordered turn
if (turn.started or has_content) and not turn.start_emitted:
turn.start_emitted = True
events.append(self._speech_event("input_audio_buffer.speech_started", item_id))
if turn.latest_partial is not None and turn.final_text is None:
delta: str = self._new_suffix( # rebind-ok: computed for the selected turn
turn.emitted_partial, turn.latest_partial
)
if delta:
turn.emitted_partial = turn.latest_partial
events.append(
{ # mutable-ok: OpenAI-compatible JSON event
"type": "conversation.item.input_audio_transcription.delta",
"event_id": f"event_{uuid.uuid4().hex}",
"item_id": item_id,
"content_index": 0,
"delta": delta,
}
)
if turn.stopped and not turn.stopped_emitted:
turn.stopped_emitted = True
events.append(self._speech_event("input_audio_buffer.speech_stopped", item_id))
if turn.final_text is not None and turn.stopped_emitted and not turn.completed_emitted:
turn.completed_emitted = True
usage: RealtimeInputAudioTranscriptionUsage | None = ( # rebind-ok: usage assigned per turn
self.take_unbilled_usage()
)
completed_event: dict[str, object] = { # mutable-ok: incrementally builds OpenAI JSON event
"type": "conversation.item.input_audio_transcription.completed",
"event_id": f"event_{uuid.uuid4().hex}",
"item_id": item_id,
"content_index": 0,
"transcript": turn.final_text,
}
if turn.speaker is not None:
completed_event["speaker"] = turn.speaker
if usage is not None:
completed_event["usage"] = usage
events.append(completed_event)
if not (turn.completed_emitted and (turn.stopped or turn.completed_signal)):
break
del self._turns[turn_id]
self._remember_completed(turn_id)
return tuple(events)
def _remember_completed(self, turn_id: str) -> None:
if turn_id in self._completed_turn_ids:
return
if len(self._completed_turn_order) >= self._completed_turn_limit:
self._completed_turn_ids.discard(self._completed_turn_order.popleft())
self._completed_turn_order.append(turn_id)
self._completed_turn_ids.add(turn_id)
def _transcript_turn_id(self, message: Mapping[str, JsonValue]) -> str:
if message.get("turnId") is not None:
return self._required_turn_id(message, "transcript")
if self._active_turn_id is not None:
return self._active_turn_id
if self._mode != "PUSH_TO_TALK":
raise MuseProtocolError("transcript event is missing turnId outside an active turn")
turn_id: Final = f"item_{uuid.uuid4().hex}"
self._active_turn_id = turn_id
return turn_id
@staticmethod
def _required_turn_id(message: Mapping[str, JsonValue], event: str) -> str:
value: Final = message.get("turnId")
if isinstance(value, bool) or not isinstance(value, (str, int)):
raise MuseProtocolError(f"{event} event has invalid turnId")
turn_id: Final = str(value).strip()
if not turn_id:
raise MuseProtocolError(f"{event} event has invalid turnId")
return turn_id
@staticmethod
def _speech_event(event_type: str, turn_id: str) -> OpenAIEvent:
return { # mutable-ok: OpenAI-compatible JSON event
"type": event_type,
"event_id": f"event_{uuid.uuid4().hex}",
"item_id": turn_id,
}
@staticmethod
def _new_suffix(previous: str, current: str) -> str:
if current.startswith(previous):
return current[len(previous) :]
return ""
class _CompletedTurn(Exception):
pass
def encode_event(event: Mapping[str, object]) -> str:
return json.dumps(event, separators=(",", ":"))

2
litellm/llms/openai_like/providers.json
View file

@ -173,7 +173,7 @@
"api_key_env": "META_API_KEY",
"api_base_env": "META_API_BASE",
"base_class": "openai_gpt",
"supported_endpoints": ["/v1/chat/completions", "/v1/responses", "/v1/messages"]
"supported_endpoints": ["/v1/chat/completions", "/v1/responses", "/v1/messages", "/v1/realtime"]
},
"cognition": {
"base_url": "https://api.cognition.ai/v1",

27
litellm/model_prices_and_context_window_backup.json
View file

@ -34717,6 +34717,21 @@
"supports_web_search": true,
"supports_xhigh_reasoning_effort": true
},
"meta/muse-voice-transcribe-1.0": {
"litellm_provider": "meta",
"mode": "audio_transcription",
"source": "https://dev.meta.ai/docs/speech-to-text",
"supported_endpoints": [
"/v1/realtime"
],
"supported_modalities": [
"audio"
],
"supported_output_modalities": [
"text"
],
"supports_audio_input": true
},
"meta_llama/Llama-3.3-70B-Instruct": {
"litellm_provider": "meta_llama",
"max_input_tokens": 128000,
@ -59098,9 +59113,9 @@
"litellm_provider": "wandb",
"mode": "chat",
"supports_reasoning": true,
"input_cost_per_token": 0.00000131,
"output_cost_per_token": 0.00000396,
"cache_read_input_token_cost": 0.000000044,
"input_cost_per_token": 1.31e-06,
"output_cost_per_token": 3.96e-06,
"cache_read_input_token_cost": 4.4e-08,
"supports_prompt_caching": true,
"source": "https://wandb.ai/site/pricing/tokens/"
},
@ -59108,9 +59123,9 @@
"litellm_provider": "wandb",
"mode": "chat",
"supports_reasoning": true,
"input_cost_per_token": 0.0000001,
"output_cost_per_token": 0.00000015,
"cache_read_input_token_cost": 0.00000005,
"input_cost_per_token": 1e-07,
"output_cost_per_token": 1.5e-07,
"cache_read_input_token_cost": 5e-08,
"supports_prompt_caching": true,
"source": "https://wandb.ai/site/pricing/tokens/"
},

34
litellm/realtime_api/main.py
View file

@ -333,7 +333,7 @@ async def _resolve_vertex_access_token_bounded(
@wrapper_client
async def _arealtime(
async def _arealtime( # noqa: C901 # central dispatcher branches once per supported realtime provider
model: str,
websocket: "WebSocket", # fastapi websocket
api_base: str | None = None,
@ -391,7 +391,37 @@ async def _arealtime(
model=model,
provider=LlmProviders(_custom_llm_provider),
)
if provider_config is not None:
if _custom_llm_provider == LlmProviders.META.value:
if model != "muse-voice-transcribe-1.0":
raise ValueError(f"Unsupported Meta realtime model: {model}")
if query_params is None or query_params.get("intent") != "transcription":
raise ValueError("Meta Muse Voice realtime requires intent=transcription")
from litellm.llms.meta.realtime.handler import MetaRealtime
meta_api_key: Final = get_secret_str("META_API_KEY")
dynamic_key_override: Final = dynamic_api_key if dynamic_api_key != meta_api_key else None
resolved_meta_api_key: Final = (
api_key
or litellm_params.api_key
or dynamic_key_override
or get_secret_str("MODEL_API_KEY")
or dynamic_api_key
or meta_api_key
)
await MetaRealtime().async_realtime(
model=model,
websocket=websocket,
logging_obj=litellm_logging_obj,
api_base=dynamic_api_base or litellm_params.api_base or api_base,
api_key=resolved_meta_api_key,
client=client,
timeout=timeout,
query_params=query_params,
user_api_key_dict=kwargs.get("user_api_key_dict"),
litellm_metadata=_build_litellm_metadata(kwargs),
)
elif provider_config is not None:
await base_llm_http_handler.async_realtime(
model=model,
websocket=websocket,

2
litellm/types/llms/openai.py
View file

@ -2202,6 +2202,8 @@ class OpenAIRealtimeInputAudioTranscriptionCompleted(TypedDict):
item_id: ReadOnly[str]
content_index: ReadOnly[int]
transcript: ReadOnly[str]
usage: NotRequired[ReadOnly[Mapping[str, object]]]
speaker: NotRequired[ReadOnly[str]]
class OpenAIRealtimeUsageTokenDetails(TypedDict):

12
litellm/types/realtime.py
View file

@ -169,9 +169,19 @@ class RealtimeInputAudioTranscriptionUsageInputTokenDetails(TypedDict):
audio_tokens: ReadOnly[int]
class RealtimeInputAudioTranscriptionUsage(TypedDict):
class RealtimeInputAudioTranscriptionTokenUsage(TypedDict):
type: ReadOnly[Literal["tokens"]]
input_tokens: ReadOnly[int]
output_tokens: ReadOnly[int]
total_tokens: ReadOnly[int]
input_token_details: ReadOnly[RealtimeInputAudioTranscriptionUsageInputTokenDetails]
class RealtimeInputAudioTranscriptionDurationUsage(TypedDict):
type: ReadOnly[Literal["duration"]]
seconds: ReadOnly[float]
RealtimeInputAudioTranscriptionUsage = (
RealtimeInputAudioTranscriptionTokenUsage | RealtimeInputAudioTranscriptionDurationUsage
)

27
model_prices_and_context_window.json
View file

@ -34717,6 +34717,21 @@
"supports_web_search": true,
"supports_xhigh_reasoning_effort": true
},
"meta/muse-voice-transcribe-1.0": {
"litellm_provider": "meta",
"mode": "audio_transcription",
"source": "https://dev.meta.ai/docs/speech-to-text",
"supported_endpoints": [
"/v1/realtime"
],
"supported_modalities": [
"audio"
],
"supported_output_modalities": [
"text"
],
"supports_audio_input": true
},
"meta_llama/Llama-3.3-70B-Instruct": {
"litellm_provider": "meta_llama",
"max_input_tokens": 128000,
@ -59098,9 +59113,9 @@
"litellm_provider": "wandb",
"mode": "chat",
"supports_reasoning": true,
"input_cost_per_token": 0.00000131,
"output_cost_per_token": 0.00000396,
"cache_read_input_token_cost": 0.000000044,
"input_cost_per_token": 1.31e-06,
"output_cost_per_token": 3.96e-06,
"cache_read_input_token_cost": 4.4e-08,
"supports_prompt_caching": true,
"source": "https://wandb.ai/site/pricing/tokens/"
},
@ -59108,9 +59123,9 @@
"litellm_provider": "wandb",
"mode": "chat",
"supports_reasoning": true,
"input_cost_per_token": 0.0000001,
"output_cost_per_token": 0.00000015,
"cache_read_input_token_cost": 0.00000005,
"input_cost_per_token": 1e-07,
"output_cost_per_token": 1.5e-07,
"cache_read_input_token_cost": 5e-08,
"supports_prompt_caching": true,
"source": "https://wandb.ai/site/pricing/tokens/"
},

191
tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
View file

@ -10,8 +10,6 @@ from websockets.exceptions import ConnectionClosed
from websockets.frames import Close
import litellm
from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.litellm_core_utils.realtime_streaming import (
REALTIME_SESSION_SUCCESS_LOGGED_KEY,
@ -20,10 +18,6 @@ from litellm.litellm_core_utils.realtime_streaming import (
)
from litellm.llms.xai.realtime.transformation import XAIRealtimeNormalizer
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.llms.openai import (
OpenAIRealtimeStreamResponseBaseObject,
OpenAIRealtimeStreamSessionEvents,
)
def _make_transcript_event(text: str, item_id: str = "item_x") -> bytes:
@ -161,6 +155,7 @@ async def test_backend_to_client_send_text_receives_str_not_bytes():
logging_obj = MagicMock()
logging_obj.async_success_handler = AsyncMock()
logging_obj.success_handler = MagicMock()
logging_obj.dispatch_success_handlers = AsyncMock()
streaming = RealTimeStreaming(client_ws, backend_ws, logging_obj)
await streaming.backend_to_client_send_messages()
@ -812,7 +807,6 @@ async def test_transcription_captured_in_backend_to_client():
Test that conversation.item.input_audio_transcription.completed events
from the backend are captured as user input during the WebSocket session.
"""
import litellm
client_ws = MagicMock()
client_ws.send_text = AsyncMock()
@ -838,6 +832,7 @@ async def test_transcription_captured_in_backend_to_client():
logging_obj.model_call_details = {"messages": "default-message-value"}
logging_obj.async_success_handler = AsyncMock()
logging_obj.success_handler = MagicMock()
logging_obj.dispatch_success_handlers = AsyncMock()
streaming = RealTimeStreaming(client_ws, backend_ws, logging_obj)
await streaming.backend_to_client_send_messages()
@ -883,6 +878,7 @@ async def test_transcription_session_captures_usage_and_skips_response_create():
logging_obj.model_call_details = {}
logging_obj.async_success_handler = AsyncMock()
logging_obj.success_handler = MagicMock()
logging_obj.dispatch_success_handlers = AsyncMock()
streaming = RealTimeStreaming(client_ws, backend_ws, logging_obj)
await streaming.backend_to_client_send_messages()
@ -1100,7 +1096,6 @@ def test_capture_transcription_usage_deduplicates_when_already_stored():
When the event is already in messages (logged via store_message), it must not
be appended a second time by _capture_transcription_usage.
"""
import litellm
streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock())
# Add the event type to the default logged list so _should_store_message returns True.
@ -1409,7 +1404,6 @@ async def test_realtime_guardrail_blocks_prompt_injection(monkeypatch: pytest.Mo
)
@pytest.mark.asyncio
async def test_realtime_guardrail_allows_clean_transcript(monkeypatch: pytest.MonkeyPatch):
"""
@ -1466,7 +1460,6 @@ async def test_realtime_guardrail_allows_clean_transcript(monkeypatch: pytest.Mo
assert len(response_creates) == 1, f"Clean transcript should trigger response.create, got: {sent_to_backend}"
@pytest.mark.asyncio
async def test_realtime_text_input_guardrail_blocks_and_returns_error(monkeypatch: pytest.MonkeyPatch):
"""
@ -1560,7 +1553,6 @@ async def test_realtime_text_input_guardrail_blocks_and_returns_error(monkeypatc
assert len(original_items) == 0, f"Blocked item should not be forwarded to backend, got: {original_items}"
@pytest.mark.asyncio
async def test_realtime_function_call_output_guardrail_blocks_and_returns_error(monkeypatch: pytest.MonkeyPatch):
"""
@ -1649,7 +1641,6 @@ async def test_realtime_function_call_output_guardrail_blocks_and_returns_error(
assert "test@example.com" not in sanitized_item["output"]
@pytest.mark.asyncio
async def test_realtime_function_call_output_guardrail_allows_clean_output(monkeypatch: pytest.MonkeyPatch):
"""
@ -1714,7 +1705,6 @@ async def test_realtime_function_call_output_guardrail_allows_clean_output(monke
assert len(forwarded) == 1, f"Clean function_call_output should be forwarded, got: {forwarded}"
@pytest.mark.asyncio
async def test_realtime_text_input_guardrail_uses_pre_call_mode(monkeypatch: pytest.MonkeyPatch):
"""
@ -1750,7 +1740,6 @@ async def test_realtime_text_input_guardrail_uses_pre_call_mode(monkeypatch: pyt
)
@pytest.mark.asyncio
async def test_realtime_session_created_injects_session_update_for_audio_guardrail(monkeypatch: pytest.MonkeyPatch):
"""
@ -1807,7 +1796,6 @@ async def test_realtime_session_created_injects_session_update_for_audio_guardra
)
@pytest.mark.asyncio
async def test_realtime_session_created_does_not_inject_session_update_for_pre_call_only(
monkeypatch: pytest.MonkeyPatch,
@ -1852,7 +1840,6 @@ async def test_realtime_session_created_does_not_inject_session_update_for_pre_c
assert len(session_updates) == 0, f"pre_call-only guardrail must not inject session.update, got: {sent_to_backend}"
@pytest.mark.asyncio
async def test_pre_call_and_post_call_guardrails_do_not_disable_server_vad(monkeypatch: pytest.MonkeyPatch):
"""Model Armor-style pre_call + post_call must not gate audio VAD."""
@ -1868,17 +1855,17 @@ async def test_pre_call_and_post_call_guardrails_do_not_disable_server_vad(monke
litellm,
"callbacks",
[
ModelArmorStyleGuardrail(
guardrail_name="model_armor_all_pre_call",
event_hook=GuardrailEventHooks.pre_call,
default_on=False,
),
ModelArmorStyleGuardrail(
guardrail_name="model_armor_all_post_call",
event_hook=GuardrailEventHooks.post_call,
default_on=False,
),
],
ModelArmorStyleGuardrail(
guardrail_name="model_armor_all_pre_call",
event_hook=GuardrailEventHooks.pre_call,
default_on=False,
),
ModelArmorStyleGuardrail(
guardrail_name="model_armor_all_post_call",
event_hook=GuardrailEventHooks.post_call,
default_on=False,
),
],
)
client_ws = MagicMock()
@ -1902,7 +1889,6 @@ async def test_pre_call_and_post_call_guardrails_do_not_disable_server_vad(monke
assert streaming._has_audio_transcription_guardrails() is False
@pytest.mark.asyncio
async def test_end_session_after_n_fails_closes_connection(monkeypatch: pytest.MonkeyPatch):
"""
@ -1949,7 +1935,6 @@ async def test_end_session_after_n_fails_closes_connection(monkeypatch: pytest.M
assert streaming._violation_count == 2
@pytest.mark.asyncio
async def test_on_violation_end_session_closes_on_first_fail(monkeypatch: pytest.MonkeyPatch):
"""
@ -1995,7 +1980,6 @@ async def test_on_violation_end_session_closes_on_first_fail(monkeypatch: pytest
assert streaming._violation_count == 1
@pytest.mark.asyncio
async def test_provider_path_suppresses_duplicate_session_created_after_synthetic():
client_ws = MagicMock()
@ -2956,7 +2940,9 @@ async def test_log_messages_routes_async_logging_through_bounded_worker():
mock_worker.ensure_initialized_and_enqueue.assert_called_once()
enqueued = mock_worker.ensure_initialized_and_enqueue.call_args
assert (enqueued.args or tuple(enqueued.kwargs.values()))[0] is logging_obj.dispatch_success_handlers.return_value
assert (enqueued.args or tuple(enqueued.kwargs.values()))[
0
] is logging_obj.dispatch_success_handlers.return_value
logging_obj.dispatch_success_handlers.assert_called_once_with(streaming.messages, prefer_async_handlers=True)
logging_obj.success_handler.assert_not_called()
# the bare create_task path must no longer be used for success logging
@ -3041,6 +3027,7 @@ async def test_session_close_flushes_unbilled_transcription_usage():
logging_obj: Final = MagicMock()
logging_obj.async_success_handler = AsyncMock()
logging_obj.success_handler = MagicMock()
logging_obj.dispatch_success_handlers = AsyncMock()
usage: Final[RealtimeInputAudioTranscriptionUsage] = {
"type": "tokens",
@ -3116,6 +3103,7 @@ async def test_session_close_flush_noop_without_unbilled_usage():
logging_obj: Final = MagicMock()
logging_obj.async_success_handler = AsyncMock()
logging_obj.success_handler = MagicMock()
logging_obj.dispatch_success_handlers = AsyncMock()
provider_config: Final = MagicMock()
provider_config.unbilled_usage_on_session_close = MagicMock(return_value=None)
@ -3412,3 +3400,144 @@ async def test_refused_session_does_not_stamp_the_reservation_ownership_marker()
assert session.logging.logged_failures == (upstream_close,)
assert REALTIME_SESSION_SUCCESS_LOGGED_KEY not in session.logging.model_call_details
@pytest.mark.asyncio
async def test_separate_usage_provider_flushes_duration_once_without_client_event():
from typing import Final
client_ws: Final = MagicMock()
client_ws.send_text = AsyncMock()
backend_ws: Final = MagicMock()
backend_ws.recv = AsyncMock(side_effect=ConnectionClosed(None, None))
logging_obj: Final = MagicMock()
logging_obj.model_call_details = {}
logging_obj.dispatch_success_handlers = AsyncMock()
usage_provider: Final = MagicMock()
usage_provider.unbilled_usage_on_session_close.return_value = {
"type": "duration",
"seconds": 0.75,
}
streaming: Final = RealTimeStreaming(
client_ws,
backend_ws,
logging_obj,
model="muse-voice-transcribe-1.0",
usage_provider=usage_provider,
)
await streaming.backend_to_client_send_messages()
usage_provider.unbilled_usage_on_session_close.assert_called_once_with("muse-voice-transcribe-1.0")
duration_events: Final = tuple(
message
for message in streaming.messages
if isinstance(message, dict) and message.get("usage") == {"type": "duration", "seconds": 0.75}
)
assert len(duration_events) == 1
assert client_ws.send_text.await_count == 0
logging_obj.dispatch_success_handlers.assert_called_once_with(streaming.messages, prefer_async_handlers=True)
@pytest.mark.asyncio
async def test_transformed_transcription_completion_never_sends_response_create():
from typing import Final
completed_event: Final = {
"type": "conversation.item.input_audio_transcription.completed",
"event_id": "event_1",
"item_id": "turn_1",
"content_index": 0,
"transcript": "private transcript",
"usage": {"type": "duration", "seconds": 0.5},
}
provider_config: Final = MagicMock()
provider_config.requires_session_configuration.return_value = True
provider_config.transform_realtime_response.return_value = {
"response": completed_event,
"current_output_item_id": None,
"current_response_id": None,
"current_delta_chunks": None,
"current_conversation_id": None,
"current_item_chunks": None,
"current_delta_type": None,
"session_configuration_request": None,
}
provider_config.transform_realtime_request.return_value = (json.dumps({"type": "response.create"}),)
provider_config.is_setup_message.return_value = False
provider_config.is_content_message.return_value = False
client_ws: Final = MagicMock()
client_ws.send_text = AsyncMock()
backend_ws: Final = MagicMock()
backend_ws.send = AsyncMock()
streaming: Final = RealTimeStreaming(
client_ws,
backend_ws,
MagicMock(),
provider_config=provider_config,
model="muse-voice-transcribe-1.0",
force_transcription_model="muse-voice-transcribe-1.0",
)
await streaming._handle_provider_config_message("{}")
assert json.loads(client_ws.send_text.await_args.args[0]) == completed_event
backend_ws.send.assert_not_awaited()
def test_private_logging_excludes_audio_transcript_hints_and_provider_body(monkeypatch: pytest.MonkeyPatch):
from typing import Final
monkeypatch.setattr(litellm, "logged_real_time_event_types", "*")
logging_obj: Final = MagicMock()
logging_obj.model_call_details = {}
streaming: Final = RealTimeStreaming(
MagicMock(),
MagicMock(),
logging_obj,
model="muse-voice-transcribe-1.0",
exclude_private_content_from_logs=True,
)
audio: Final = "cHJpdmF0ZS1hdWRpbw=="
transcript: Final = "private transcript"
keyword: Final = "private keyword"
provider_body: Final = "private provider body"
streaming.store_input(
json.dumps(
{
"type": "session.update",
"session": {
"type": "transcription",
"model": "muse-voice-transcribe-1.0",
"mode": "ENDPOINTING",
"audio": {"input": {"transcription": {"keywords": [keyword]}}},
},
}
)
)
streaming.store_input(json.dumps({"type": "input_audio_buffer.append", "audio": audio}))
streaming.store_message(
{
"type": "conversation.item.input_audio_transcription.completed",
"event_id": "event_1",
"item_id": "turn_1",
"transcript": transcript,
"provider_body": provider_body,
"usage": {"type": "duration", "seconds": 1.0},
}
)
logged_inputs: Final = tuple(call.kwargs["input"] for call in logging_obj.pre_call.call_args_list)
serialized: Final = json.dumps({"inputs": logged_inputs, "messages": streaming.messages})
assert audio not in serialized
assert transcript not in serialized
assert keyword not in serialized
assert provider_body not in serialized
assert "muse-voice-transcribe-1.0" in serialized
assert "ENDPOINTING" in serialized
assert "turn_1" in serialized
assert '"seconds": 1.0' in serialized
assert streaming.input_messages == []

449
tests/test_litellm/llms/meta/realtime/test_meta_realtime_handler.py Normal file
View file

@ -0,0 +1,449 @@
import asyncio
import base64
import json
from collections.abc import Callable
from typing import Final
from unittest.mock import AsyncMock, MagicMock
import pytest
from litellm.llms.meta.realtime.handler import (
DEFAULT_MUSE_REALTIME_URL,
MetaRealtime,
MuseAdapterError,
MuseRealtimeAdapter,
build_muse_realtime_url,
normalize_access_token,
safe_close_reason,
sanitize_close_code,
)
from litellm.llms.meta.realtime.transformation import MUSE_MODEL
class FakeProviderWebSocket:
def __init__(self, session_id: str = "provider-session") -> None:
self.sent: list[str | bytes] = []
self.close_calls: list[tuple[int, str]] = []
self._session_id: Final = session_id
self._recv_count = 0
self._closed = asyncio.Event()
async def send(self, message: str | bytes) -> None:
self.sent.append(message)
async def recv(self, decode: bool | None = None) -> str | bytes:
self._recv_count += 1
if self._recv_count == 1:
return json.dumps({"sessionId": self._session_id})
await self._closed.wait()
raise MuseAdapterError("closed", close_code=1000)
async def close(self, code: int = 1000, reason: str = "") -> None:
self.close_calls.append((code, reason))
self._closed.set()
class DelayedAckWebSocket(FakeProviderWebSocket):
def __init__(self) -> None:
super().__init__()
self.ack_release = asyncio.Event()
async def recv(self, decode: bool | None = None) -> str | bytes:
self._recv_count += 1
if self._recv_count == 1:
await self.ack_release.wait()
return json.dumps({"sessionId": self._session_id})
await self._closed.wait()
raise MuseAdapterError("closed", close_code=1000)
async def _wait_until(predicate: Callable[[], bool]) -> None:
for _ in range(100):
if predicate():
return
await asyncio.sleep(0)
raise AssertionError("condition did not become true")
def _session_update(*, rate: int = 24_000, mode: str = "ENDPOINTING") -> str:
return json.dumps(
{
"type": "session.update",
"session": {
"type": "transcription",
"mode": mode,
"audio": {
"input": {
"format": {"type": "audio/pcm", "rate": rate, "channels": 1},
"transcription": {"model": MUSE_MODEL},
}
},
},
}
)
async def _configured_adapter(
*,
rate: int = 24_000,
mode: str = "ENDPOINTING",
provider_ws: FakeProviderWebSocket | None = None,
monotonic: Callable[[], float] = lambda: 10.0,
sleep: Callable[[float], object] | None = None,
) -> tuple[MuseRealtimeAdapter, FakeProviderWebSocket, dict[str, object]]:
ws: Final = provider_ws or FakeProviderWebSocket()
connect_call: Final[dict[str, object]] = {}
async def connect(url: str, **kwargs: object) -> FakeProviderWebSocket:
connect_call.update({"url": url, **kwargs})
return ws
async def no_sleep(_: float) -> None:
return None
adapter: Final = MuseRealtimeAdapter(
model=f"meta/{MUSE_MODEL}",
api_key=" raw-token ",
websocket_connect=connect,
monotonic=monotonic,
sleep=sleep or no_sleep,
)
created: Final = json.loads(await adapter.recv())
assert created["type"] == "session.created"
await adapter.send(_session_update(rate=rate, mode=mode))
updated: Final = json.loads(await adapter.recv())
assert updated["type"] == "session.updated"
return adapter, ws, connect_call
@pytest.mark.parametrize(
("api_key", "expected"),
[
("token", "Bearer token"),
(" Bearer token ", "Bearer token"),
("bearer token", "Bearer token"),
],
)
def test_normalize_access_token_emits_exactly_one_bearer_prefix(api_key: str, expected: str):
assert normalize_access_token(api_key) == expected
@pytest.mark.parametrize("api_key", ["", " ", "Bearer", " bearer "])
def test_normalize_access_token_rejects_missing_token(api_key: str):
with pytest.raises(ValueError, match=r"token|key is required"):
normalize_access_token(api_key)
def test_build_muse_realtime_url_uses_fixed_secure_path():
assert build_muse_realtime_url(None) == DEFAULT_MUSE_REALTIME_URL
assert build_muse_realtime_url("https://example.test/custom/path?ignored=yes") == (
"wss://example.test/v1/asr/realtime"
)
assert build_muse_realtime_url("wss://example.test:8443/other") == ("wss://example.test:8443/v1/asr/realtime")
@pytest.mark.parametrize(
"api_base",
[
"http://example.test",
"ws://example.test",
"wss://user:pass@example.test",
"wss://example.test/path#fragment",
"not-a-url",
],
)
def test_build_muse_realtime_url_rejects_insecure_or_ambiguous_overrides(api_base: str):
with pytest.raises(ValueError, match="absolute wss:// or https://"):
build_muse_realtime_url(api_base)
@pytest.mark.asyncio
async def test_handshake_contains_bearer_only_in_json_body_and_waits_for_ack():
provider_ws: Final = DelayedAckWebSocket()
connect_call: Final[dict[str, object]] = {}
async def connect(url: str, **kwargs: object) -> DelayedAckWebSocket:
connect_call.update({"url": url, **kwargs})
return provider_ws
adapter: Final = MuseRealtimeAdapter(
model=MUSE_MODEL,
api_key="Bearer private-token",
websocket_connect=connect,
)
await adapter.recv()
update_task: Final = asyncio.create_task(adapter.send(_session_update()))
await _wait_until(lambda: len(provider_ws.sent) == 1)
assert connect_call["url"] == DEFAULT_MUSE_REALTIME_URL
assert "additional_headers" not in connect_call
handshake: Final = json.loads(provider_ws.sent[0])
assert handshake["authorization"] == {"accessToken": "Bearer private-token"}
assert handshake["audioEncoding"] == "PCM_24KHZ"
assert not update_task.done()
assert not any(isinstance(frame, bytes) for frame in provider_ws.sent)
provider_ws.ack_release.set()
await update_task
updated: Final = json.loads(await adapter.recv())
assert updated["type"] == "session.updated"
assert updated["session"]["id"] == "provider-session"
await adapter.close()
@pytest.mark.asyncio
@pytest.mark.parametrize(("rate", "packet_bytes"), [(16_000, 2_560), (24_000, 3_840)])
async def test_audio_is_strictly_decoded_and_packetized_as_raw_pcm(rate: int, packet_bytes: int):
adapter, provider_ws, _ = await _configured_adapter(rate=rate)
pcm: Final = (b"\xff\xfe\x00\x80" * (packet_bytes // 2))[: packet_bytes * 2]
await adapter.send(json.dumps({"type": "input_audio_buffer.append", "audio": base64.b64encode(pcm).decode()}))
await _wait_until(lambda: sum(isinstance(frame, bytes) for frame in provider_ws.sent) == 2)
binary_frames: Final = tuple(frame for frame in provider_ws.sent if isinstance(frame, bytes))
assert binary_frames == (pcm[:packet_bytes], pcm[packet_bytes:])
assert b"\xff\xfe" in pcm
await adapter.close()
@pytest.mark.asyncio
@pytest.mark.parametrize(
("audio", "expected_message"),
[
("not base64!", "valid base64"),
(base64.b64encode(b"\x00").decode(), "complete samples"),
],
)
async def test_invalid_base64_or_odd_pcm_is_rejected(audio: str, expected_message: str):
adapter, provider_ws, _ = await _configured_adapter()
await adapter.send(json.dumps({"type": "input_audio_buffer.append", "audio": audio}))
error: Final = json.loads(await adapter.recv())
assert error["type"] == "error"
assert error["error"]["code"] == "invalid_audio"
assert expected_message in error["error"]["message"]
assert adapter.close_code == 1008
assert not any(isinstance(frame, bytes) for frame in provider_ws.sent)
await adapter.close()
@pytest.mark.asyncio
async def test_absolute_pacing_delays_only_audio_ahead_of_wall_time():
sleeps: Final[list[float]] = []
async def record_sleep(delay: float) -> None:
sleeps.append(delay)
adapter, provider_ws, _ = await _configured_adapter(monotonic=lambda: 10.0, sleep=record_sleep)
pcm: Final = b"\x01\x02" * 3_840
await adapter.send(json.dumps({"type": "input_audio_buffer.append", "audio": base64.b64encode(pcm).decode()}))
await _wait_until(lambda: sum(isinstance(frame, bytes) for frame in provider_ws.sent) == 2)
assert sleeps == pytest.approx([0.08])
await adapter.close()
@pytest.mark.asyncio
async def test_append_larger_than_four_seconds_is_rejected_without_dropping_prefix():
adapter, provider_ws, _ = await _configured_adapter(rate=16_000)
oversized_pcm: Final = b"\x00\x00" * (16_000 * 4 + 1)
await adapter.send(
json.dumps({"type": "input_audio_buffer.append", "audio": base64.b64encode(oversized_pcm).decode()})
)
error: Final = json.loads(await adapter.recv())
assert error["error"]["code"] == "audio_backlog_exceeded"
assert adapter.close_code == 1008
assert not any(isinstance(frame, bytes) for frame in provider_ws.sent)
await adapter.close()
@pytest.mark.asyncio
async def test_clear_discards_only_unsent_audio():
adapter, provider_ws, _ = await _configured_adapter()
old_pcm: Final = b"\x01\x02" * 100
new_pcm: Final = b"\x03\x04" * 100
await adapter.send(json.dumps({"type": "input_audio_buffer.append", "audio": base64.b64encode(old_pcm).decode()}))
await adapter.send(_event("input_audio_buffer.clear"))
cleared: Final = json.loads(await adapter.recv())
await adapter.send(json.dumps({"type": "input_audio_buffer.append", "audio": base64.b64encode(new_pcm).decode()}))
await adapter.send(_event("input_audio_buffer.commit"))
await _wait_until(lambda: any(isinstance(frame, bytes) for frame in provider_ws.sent))
assert cleared["type"] == "input_audio_buffer.cleared"
assert tuple(frame for frame in provider_ws.sent if isinstance(frame, bytes)) == (new_pcm,)
assert '{"type":"endStream"}' not in provider_ws.sent
await adapter.close()
@pytest.mark.asyncio
async def test_endpointing_commit_flushes_partial_packet_without_ending_stream():
adapter, provider_ws, _ = await _configured_adapter(mode="ENDPOINTING")
pcm: Final = b"\x01\x02" * 100
await adapter.send(json.dumps({"type": "input_audio_buffer.append", "audio": base64.b64encode(pcm).decode()}))
await adapter.send(_event("input_audio_buffer.commit"))
committed: Final = json.loads(await adapter.recv())
await _wait_until(lambda: any(isinstance(frame, bytes) for frame in provider_ws.sent))
assert committed["type"] == "input_audio_buffer.committed"
assert committed["item_id"].startswith("item_")
assert committed["previous_item_id"] is None
assert tuple(frame for frame in provider_ws.sent if isinstance(frame, bytes)) == (pcm,)
assert '{"type":"endStream"}' not in provider_ws.sent
await adapter.close()
@pytest.mark.asyncio
@pytest.mark.parametrize(
("mode", "terminal_event"),
[("PUSH_TO_TALK", "input_audio_buffer.commit"), ("ENDPOINTING", "input_audio_buffer.end")],
)
async def test_commit_or_end_sends_end_stream_exactly_once(mode: str, terminal_event: str):
adapter, provider_ws, _ = await _configured_adapter(mode=mode)
pcm: Final = b"\x01\x02" * 100
await adapter.send(json.dumps({"type": "input_audio_buffer.append", "audio": base64.b64encode(pcm).decode()}))
await adapter.send(_event(terminal_event))
await adapter.send(_event("input_audio_buffer.end"))
await _wait_until(lambda: '{"type":"endStream"}' in provider_ws.sent)
assert tuple(frame for frame in provider_ws.sent if isinstance(frame, bytes)) == (pcm,)
assert provider_ws.sent.count('{"type":"endStream"}') == 1
await adapter.close()
@pytest.mark.asyncio
async def test_response_create_is_returned_as_error_and_never_sent_upstream():
adapter, provider_ws, _ = await _configured_adapter()
await adapter.send(_event("response.create"))
error: Final = json.loads(await adapter.recv())
assert error["type"] == "error"
assert error["error"]["code"] == "unsupported_event"
assert not any(isinstance(frame, str) and "response.create" in frame for frame in provider_ws.sent)
await adapter.close()
@pytest.mark.asyncio
async def test_close_codes_and_reasons_are_sanitized_without_secret_leakage():
adapter, provider_ws, _ = await _configured_adapter()
secret: Final = "Bearer private-token"
await adapter.close(code=4001, reason=f"provider rejected {secret}")
assert adapter.close_code == 1011
assert adapter.close_reason == "Realtime transcription service error"
assert provider_ws.close_calls == [(1011, "Realtime transcription service error")]
assert secret not in json.dumps(provider_ws.close_calls)
assert sanitize_close_code(1013) == 1013
assert safe_close_reason(1008) == "Invalid realtime transcription request"
@pytest.mark.asyncio
async def test_handshake_failure_reports_only_exception_type():
secret: Final = "private-token"
async def failing_connect(url: str, **kwargs: object) -> FakeProviderWebSocket:
raise RuntimeError(f"failed with {secret}")
adapter: Final = MuseRealtimeAdapter(
model=MUSE_MODEL,
api_key=secret,
websocket_connect=failing_connect,
)
await adapter.recv()
await adapter.send(_session_update())
error = json.loads(await adapter.recv())
assert error["type"] == "error"
assert error["error"]["message"] == "Meta Muse realtime handshake failed"
assert secret not in json.dumps(error)
with pytest.raises(MuseAdapterError) as exc_info:
await adapter.recv()
assert exc_info.value.close_code == 1011
assert secret not in str(exc_info.value)
@pytest.mark.asyncio
async def test_meta_realtime_missing_credentials_closes_client_with_policy_code():
client_ws: Final = MagicMock()
client_ws.send_text = AsyncMock()
client_ws.close = AsyncMock()
await MetaRealtime().async_realtime(
model=MUSE_MODEL,
websocket=client_ws,
logging_obj=MagicMock(),
api_key=None,
)
client_ws.close.assert_awaited_once_with(
code=1008,
reason="Invalid realtime transcription request",
)
sent_error: Final = json.loads(client_ws.send_text.await_args.args[0])
assert sent_error["type"] == "error"
assert sent_error["error"]["code"] == "invalid_configuration"
@pytest.mark.asyncio
async def test_meta_realtime_invalid_constructor_input_sends_error_before_close():
client_ws: Final = MagicMock()
client_ws.send_text = AsyncMock()
client_ws.close = AsyncMock()
await MetaRealtime().async_realtime(
model=MUSE_MODEL,
websocket=client_ws,
logging_obj=MagicMock(),
api_key="Bearer",
)
sent_error: Final = json.loads(client_ws.send_text.await_args.args[0])
assert sent_error["error"]["message"] == "Invalid Meta Muse realtime configuration"
client_ws.close.assert_awaited_once_with(
code=1008,
reason="Invalid realtime transcription request",
)
@pytest.mark.asyncio
async def test_meta_realtime_enables_private_logging_usage_and_model_enforcement(monkeypatch: pytest.MonkeyPatch):
captured: Final[dict[str, object]] = {}
class CapturingStreaming:
def __init__(self, websocket, backend_ws, logging_obj, **kwargs):
captured.update({"websocket": websocket, "backend_ws": backend_ws, "logging_obj": logging_obj, **kwargs})
async def bidirectional_forward(self) -> None:
return None
client_ws: Final = MagicMock()
client_ws.send_text = AsyncMock()
client_ws.close = AsyncMock()
monkeypatch.setattr("litellm.llms.meta.realtime.handler.RealTimeStreaming", CapturingStreaming)
await MetaRealtime().async_realtime(
model=MUSE_MODEL,
websocket=client_ws,
logging_obj=MagicMock(),
api_key="private-token",
)
adapter: Final = captured["backend_ws"]
assert isinstance(adapter, MuseRealtimeAdapter)
assert captured["force_transcription_model"] == MUSE_MODEL
assert captured["usage_provider"] is adapter
assert captured["exclude_private_content_from_logs"] is True
client_ws.close.assert_awaited_once_with(code=1000, reason="Session closed")
def _event(event_type: str) -> str:
return json.dumps({"type": event_type})

299
tests/test_litellm/llms/meta/realtime/test_meta_realtime_transformation.py Normal file
View file

@ -0,0 +1,299 @@
import json
import pytest
from litellm.llms.meta.realtime.transformation import (
MUSE_MODEL,
MuseEventTransformer,
MuseProtocolError,
encode_event,
normalize_language,
parse_session_update,
session_created_event,
session_updated_event,
)
def _event(event_type: str, **fields: object) -> str:
return json.dumps({"type": event_type, **fields})
def test_beta_session_builds_authenticated_24khz_handshake_with_hints():
config = parse_session_update(
_event(
"session.update",
session={
"type": "transcription",
"input_audio_format": "pcm16",
"turn_detection": {"type": "server_vad"},
"input_audio_transcription": {
"model": "meta/muse-voice-transcribe-1.0",
"language": "en-US",
"language_bias": ["Spanish", "english", "French"],
"keywords": [" Muse ", "LiteLLM", "Muse"],
"prompt": "must not become a keyword",
},
},
),
"meta/muse-voice-transcribe-1.0",
)
assert config.sample_rate == 24_000
assert config.packet_bytes == 3_840
assert config.mode == "ENDPOINTING"
assert config.language_bias == ("English", "Spanish", "French")
assert config.keywords == ("Muse", "LiteLLM")
assert config.handshake("Bearer token") == {
"mode": "ENDPOINTING",
"authorization": {"accessToken": "Bearer token"},
"audioEncoding": "PCM_24KHZ",
"model": MUSE_MODEL,
"partialMode": "CUMULATIVE",
"emitAudioProgress": True,
"keywords": ["Muse", "LiteLLM"],
"languageBias": ["English", "Spanish", "French"],
}
assert "must not become a keyword" not in json.dumps(config.handshake("Bearer token"))
def test_ga_session_accepts_16khz_mono_push_to_talk():
config = parse_session_update(
_event(
"session.update",
session={
"type": "transcription",
"audio": {
"input": {
"format": {"type": "audio/pcm", "rate": 16000, "channels": 1},
"turn_detection": None,
"transcription": {"model": MUSE_MODEL, "language": "zh-Hans"},
}
},
},
),
MUSE_MODEL,
)
assert config.sample_rate == 16_000
assert config.packet_bytes == 2_560
assert config.mode == "PUSH_TO_TALK"
assert config.language_bias == ("Mandarin Chinese",)
assert config.handshake("Bearer token")["audioEncoding"] == "PCM_16KHZ"
@pytest.mark.parametrize(
("source", "expected"),
[
("EN_us", "English"),
("mandarin chinese", "Mandarin Chinese"),
("fil-PH", "Tagalog"),
("iw-IL", "Hebrew"),
("pt-BR", "Portuguese"),
],
)
def test_language_normalization_uses_official_muse_names(source: str, expected: str):
assert normalize_language(source) == expected
@pytest.mark.parametrize(
("session", "message"),
[
({"input_audio_format": "g711_ulaw"}, "requires pcm16"),
({"audio": {"input": {"format": {"type": "audio/pcm", "rate": 8000}}}}, "16000 Hz or 24000 Hz"),
(
{"audio": {"input": {"format": {"type": "audio/pcm", "rate": 24000, "channels": 2}}}},
"requires mono",
),
(
{"input_audio_format": "pcm16", "audio": {"input": {"format": {"type": "audio/pcm"}}}},
"either beta or GA layout",
),
({"input_audio_transcription": {"model": "other-model"}}, "cannot be changed"),
({"input_audio_transcription": {"keywords": ["valid", ""]}}, "non-empty strings"),
({"input_audio_transcription": {"language": "xx"}}, "unsupported Muse Voice language"),
],
)
def test_session_rejects_unsupported_audio_model_and_hints(session: dict[str, object], message: str):
with pytest.raises(MuseProtocolError, match=message):
parse_session_update(_event("session.update", session={"type": "transcription", **session}), MUSE_MODEL)
def test_session_events_expose_openai_transcription_shapes():
config = parse_session_update(
_event(
"session.update",
session={
"mode": "DIARIZATION",
"audio": {
"input": {
"format": {"type": "audio/pcm", "rate": 24000},
"transcription": {"model": MUSE_MODEL, "language": "ja", "keywords": ["Meta"]},
}
},
},
),
MUSE_MODEL,
)
created = session_created_event(MUSE_MODEL, "session-before-handshake")
updated = session_updated_event(config, "provider-session")
assert created["type"] == "session.created"
assert created["session"]["type"] == "transcription"
assert updated["type"] == "session.updated"
assert updated["session"]["id"] == "provider-session"
assert updated["session"]["audio"]["input"]["transcription"] == {
"model": MUSE_MODEL,
"language": "Japanese",
"keywords": ["Meta"],
"language_bias": ["Japanese"],
}
def test_turnless_empty_silence_transcript_is_ignored():
transformer = MuseEventTransformer()
assert transformer.transform(_event("transcript", transcript="", final=True)) == ()
def test_transcript_without_speech_start_synthesizes_start_before_delta():
transformer = MuseEventTransformer()
events = transformer.transform(_event("transcript", turnId="turn-1", transcript="hello", final=False))
assert [event["type"] for event in events] == [
"input_audio_buffer.speech_started",
"conversation.item.input_audio_transcription.delta",
]
def test_cumulative_partials_emit_only_extensions_and_final_is_authoritative():
transformer = MuseEventTransformer()
started = transformer.transform(_event("speechStart", turnId="turn-1"))
first = transformer.transform(_event("transcript", turnId="turn-1", transcript="hello", final=False))
extension = transformer.transform(_event("transcript", turnId="turn-1", transcript="hello world", final=False))
rewrite = transformer.transform(_event("transcript", turnId="turn-1", transcript="hullo world", final=False))
assert transformer.transform(_event("speechComplete", turnId="turn-1", transcript="hullo world")) == ()
completed = transformer.transform(_event("speechEnd", turnId="turn-1"))
assert [event["type"] for event in started] == ["input_audio_buffer.speech_started"]
assert first[0]["delta"] == "hello"
assert extension[0]["delta"] == " world"
assert rewrite == ()
assert completed[0]["type"] == "input_audio_buffer.speech_stopped"
assert completed[1]["type"] == "conversation.item.input_audio_transcription.completed"
assert completed[1]["item_id"] == "turn-1"
assert completed[1]["transcript"] == "hullo world"
def test_completed_transcript_waits_for_speech_stopped():
transformer = MuseEventTransformer()
transformer.transform(_event("speechStart", turnId="turn-1"))
assert transformer.transform(_event("speechComplete", turnId="turn-1", transcript="done")) == ()
released = transformer.transform(_event("speechEnd", turnId="turn-1"))
assert [event["type"] for event in released] == [
"input_audio_buffer.speech_stopped",
"conversation.item.input_audio_transcription.completed",
]
def test_overlapping_turns_are_emitted_in_provider_turn_order():
transformer = MuseEventTransformer()
transformer.transform(_event("speechStart", turnId="turn-a"))
transformer.transform(_event("speechStart", turnId="turn-b"))
assert transformer.transform(_event("transcript", turnId="turn-b", transcript="second", final=False)) == ()
assert transformer.transform(_event("speechComplete", turnId="turn-a", transcript="first")) == ()
released = transformer.transform(_event("speechEnd", turnId="turn-a"))
assert [(event["type"], event["item_id"]) for event in released] == [
("input_audio_buffer.speech_stopped", "turn-a"),
("conversation.item.input_audio_transcription.completed", "turn-a"),
("input_audio_buffer.speech_started", "turn-b"),
("conversation.item.input_audio_transcription.delta", "turn-b"),
]
assert transformer.transform(_event("speechComplete", turnId="turn-b", transcript="second final")) == ()
final_b = transformer.transform(_event("speechEnd", turnId="turn-b"))
assert final_b[0]["type"] == "input_audio_buffer.speech_stopped"
assert final_b[1]["item_id"] == "turn-b"
assert final_b[1]["transcript"] == "second final"
def test_committed_item_id_is_used_for_next_provider_turn():
transformer = MuseEventTransformer()
previous_item_id, item_id = transformer.commit_item()
started = transformer.transform(_event("speechStart", turnId="provider-turn"))
transformer.transform(_event("speechComplete", turnId="provider-turn", transcript="hello"))
completed = transformer.transform(_event("speechEnd", turnId="provider-turn"))
assert previous_item_id is None
assert started[0]["item_id"] == item_id
assert completed[-1]["item_id"] == item_id
def test_commit_after_speech_start_reuses_active_item_id():
transformer = MuseEventTransformer()
started = transformer.transform(_event("speechStart", turnId="provider-turn"))
previous_item_id, item_id = transformer.commit_item()
transformer.transform(_event("speechComplete", turnId="provider-turn", transcript="hello"))
completed = transformer.transform(_event("speechEnd", turnId="provider-turn"))
assert previous_item_id is None
assert item_id == "provider-turn"
assert started[0]["item_id"] == item_id
assert completed[-1]["item_id"] == item_id
def test_speaker_and_positive_audio_progress_deltas_attach_to_next_completion():
transformer = MuseEventTransformer()
transformer.transform(_event("audioProgress", audioProcessedMs=1000))
transformer.transform(_event("audioProgress", audioProcessedMs=750))
transformer.transform(_event("audioProgress", audioProcessedMs=1600))
transformer.transform(_event("speaker", turnId=42, label=" Speaker 2 "))
transformer.transform(_event("speechComplete", turnId=42, transcript="hello"))
completed = transformer.transform(_event("speechEnd", turnId=42))
assert completed[-1]["speaker"] == "Speaker 2"
assert completed[-1]["usage"] == {"type": "duration", "seconds": 1.6}
assert transformer.take_unbilled_usage() is None
def test_trailing_audio_progress_is_returned_once():
transformer = MuseEventTransformer()
transformer.transform(_event("audioProgress", audioProcessedMs=250))
assert transformer.take_unbilled_usage() == {"type": "duration", "seconds": 0.25}
assert transformer.take_unbilled_usage() is None
def test_completed_turn_tombstone_suppresses_late_duplicates():
transformer = MuseEventTransformer()
transformer.transform(_event("speechComplete", turnId="turn-1", transcript="done"))
assert transformer.transform(_event("speechComplete", turnId="turn-1", transcript="duplicate")) == ()
assert transformer.transform(_event("speaker", turnId="turn-1", label="late")) == ()
def test_provider_error_is_sanitized_and_encodable():
token = "private-token"
provider_body = f"authorization failed for Bearer {token}"
transformed = MuseEventTransformer().transform(
_event("error", code="AUTH", message=provider_body, request={"accessToken": token})
)
encoded = encode_event(transformed[0])
assert json.loads(encoded)["error"] == {
"type": "server_error",
"code": "provider_error",
"message": "Meta Muse realtime transcription failed",
}
assert token not in encoded
assert provider_body not in encoded

80
tests/test_litellm/realtime_api/test_main.py
View file

@ -4,7 +4,6 @@ from types import TracebackType
from typing import Final
from unittest.mock import MagicMock, patch
import pytest
import litellm
@ -152,6 +151,85 @@ async def test_vertex_credential_resolution_bounds_a_thread_offloaded_refresh():
assert time.monotonic() - start < 5
@pytest.mark.asyncio
async def test_meta_realtime_rejects_missing_transcription_intent(monkeypatch: pytest.MonkeyPatch):
def mock_get_llm_provider(model, api_base, api_key):
return model.removeprefix("meta/"), "meta", api_key, api_base
monkeypatch.setattr(realtime_main, "get_llm_provider", mock_get_llm_provider)
with pytest.raises(ValueError, match="requires intent=transcription"):
await realtime_main._arealtime.__wrapped__(
model="meta/muse-voice-transcribe-1.0",
websocket=MagicMock(),
litellm_logging_obj=FakeLogging(),
query_params={"model": "meta/muse-voice-transcribe-1.0"},
)
@pytest.mark.asyncio
async def test_meta_realtime_rejects_unsupported_model_before_connecting(monkeypatch: pytest.MonkeyPatch):
def mock_get_llm_provider(model, api_base, api_key):
return model.removeprefix("meta/"), "meta", api_key, api_base
monkeypatch.setattr(realtime_main, "get_llm_provider", mock_get_llm_provider)
with pytest.raises(ValueError, match="Unsupported Meta realtime model: other-model"):
await realtime_main._arealtime.__wrapped__(
model="meta/other-model",
websocket=MagicMock(),
litellm_logging_obj=FakeLogging(),
query_params={"model": "meta/other-model", "intent": "transcription"},
)
@pytest.mark.asyncio
@pytest.mark.parametrize(
("explicit_key", "model_key", "meta_key", "expected"),
[
("explicit", "model-env", "meta-env", "explicit"),
(None, "model-env", "meta-env", "model-env"),
(None, None, "meta-env", "meta-env"),
],
)
async def test_meta_realtime_credential_precedence_is_forwarded_to_handler(
monkeypatch: pytest.MonkeyPatch,
explicit_key: str | None,
model_key: str | None,
meta_key: str | None,
expected: str,
):
captured: dict[str, object] = {}
def mock_get_llm_provider(model, api_base, api_key):
return model.removeprefix("meta/"), "meta", meta_key, api_base
def mock_get_secret_str(name: str):
return {"MODEL_API_KEY": model_key, "META_API_KEY": meta_key}.get(name)
async def mock_async_realtime(self, **kwargs):
captured.update(kwargs)
monkeypatch.setattr(realtime_main, "get_llm_provider", mock_get_llm_provider)
monkeypatch.setattr(realtime_main, "get_secret_str", mock_get_secret_str)
monkeypatch.setattr(
"litellm.llms.meta.realtime.handler.MetaRealtime.async_realtime",
mock_async_realtime,
)
await realtime_main._arealtime.__wrapped__(
model="meta/muse-voice-transcribe-1.0",
websocket=MagicMock(),
litellm_logging_obj=FakeLogging(),
api_key=explicit_key,
query_params={"model": "meta/muse-voice-transcribe-1.0", "intent": "transcription"},
)
assert captured["model"] == "muse-voice-transcribe-1.0"
assert captured["api_key"] == expected
assert captured["query_params"] == {"model": "muse-voice-transcribe-1.0", "intent": "transcription"}
@pytest.mark.asyncio
async def test_arealtime_vertex_branch_resolves_credentials_under_a_bound(monkeypatch):
"""The wiring half of the regression: the vertex branch of _arealtime must