# Copyright 2023 LiveKit, Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from __future__ import annotations import asyncio import base64 import contextlib import dataclasses import json import os import weakref from dataclasses import dataclass from typing import Any, Literal from urllib.parse import urlparse import aiohttp from livekit import rtc from livekit.agents import ( DEFAULT_API_CONNECT_OPTIONS, APIConnectionError, APIConnectOptions, APIStatusError, APITimeoutError, stt, utils, ) from livekit.agents.stt import SpeechEvent from livekit.agents.types import NOT_GIVEN, NotGivenOr from livekit.agents.utils import AudioBuffer, is_given from .gateway_adapter import build_stt_init_payload, normalize_region_override from .log import logger STTEncoding = Literal["pcm_s16le", "pcm_mulaw"] DEFAULT_BUFFER_SIZE_SECONDS = 0.064 MAX_IMMEDIATE_RETRIES = 1 # Define bytes per frame for different encoding types bytes_per_frame = { "pcm_s16le": 2, "pcm_mulaw": 1, } def _safe_error_code(exc: BaseException) -> int | None: for attr in ("status_code", "code"): value = getattr(exc, attr, None) if isinstance(value, int) and not isinstance(value, bool): return value return None def _extract_model_from_endpoint(model_endpoint: str) -> str | None: parsed = urlparse(model_endpoint) path = parsed.path.rstrip("/") marker = "/v1/stt/" marker_index = path.find(marker) if marker_index == -1: return None model_part = path[marker_index + len(marker) :] return model_part or None def _default_stt_endpoint(*, slng_base_url: str, model: str) -> str: host = slng_base_url.split(":")[0] protocol = "ws" if host in ("localhost", "127.0.0.1") else "wss" return f"{protocol}://{slng_base_url}/v1/stt/{model}" @dataclass class STTOptions: # Audio format options sample_rate: int = 16000 buffer_size_seconds: float = DEFAULT_BUFFER_SIZE_SECONDS encoding: str = "pcm_s16le" # Common SLNG streaming options (work across all models) enable_partial_transcripts: bool = True # Common VAD options (work across all models) vad_threshold: float = 0.5 vad_min_silence_duration_ms: int = 300 vad_speech_pad_ms: int = 30 # Common diarization options (work across all models) enable_diarization: bool = False min_speakers: int | None = None max_speakers: int | None = None language: str = "en" class STT(stt.STT): def __init__( self, *, api_key: str | None = None, model: str = "deepgram/nova:3", model_endpoint: str | None = None, model_endpoints: list[str] | None = None, slng_base_url: str = "api.slng.ai", region_override: str | list[str] | None = None, sample_rate: int = 16000, encoding: NotGivenOr[STTEncoding] = NOT_GIVEN, buffer_size_seconds: float = DEFAULT_BUFFER_SIZE_SECONDS, # Common SLNG options enable_partial_transcripts: bool = True, vad_threshold: float = 0.5, vad_min_silence_duration_ms: int = 300, vad_speech_pad_ms: int = 30, enable_diarization: bool = False, min_speakers: int | None = None, max_speakers: int | None = None, language: str = "en", http_session: aiohttp.ClientSession | None = None, **model_options: Any, ) -> None: """ Initialize SLNG STT. Args: api_key: SLNG API key for authentication. model: SLNG model identifier (for example "deepgram/nova:3") model_endpoint: Optional full SLNG WebSocket endpoint URL model_endpoints: Optional fallback STT endpoints slng_base_url: SLNG gateway host. Defaults to "api.slng.ai" region_override: Optional gateway region override sent as X-Region-Override. sample_rate: Audio sample rate (default: 16000) encoding: Audio encoding format buffer_size_seconds: Buffer size in seconds enable_partial_transcripts: Enable interim results vad_threshold: Voice activity detection threshold vad_min_silence_duration_ms: Min silence duration for VAD vad_speech_pad_ms: Speech padding for VAD enable_diarization: Enable speaker identification min_speakers: Minimum speakers for diarization max_speakers: Maximum speakers for diarization language: Language code (default: "en") http_session: Optional HTTP session **model_options: Model-specific options (e.g., whisper_params={"task": "translate"}) """ resolved_key = api_key or os.environ.get("SLNG_API_KEY") if not resolved_key: raise ValueError("api_key is required, or set SLNG_API_KEY environment variable") # Detect if endpoint supports streaming (WebSocket endpoints do) # - streaming=True: Supports real-time streaming (WebSocket only) # - streaming=False: HTTP batch recognition only resolved_model_endpoint = model_endpoint or _default_stt_endpoint( slng_base_url=slng_base_url, model=model, ) endpoints = list( model_endpoints or [ resolved_model_endpoint, ] ) if not endpoints: endpoints = [resolved_model_endpoint] primary_endpoint = endpoints[0] is_streaming = primary_endpoint.startswith("ws://") or primary_endpoint.startswith("wss://") super().__init__( capabilities=stt.STTCapabilities( streaming=is_streaming, interim_results=is_streaming, offline_recognize=not is_streaming, ), ) self._api_key = resolved_key self._region_override_header = normalize_region_override(region_override) self._model_endpoints = endpoints self._active_endpoint_index = 0 self._model_endpoint = endpoints[0] self._models = [_extract_model_from_endpoint(e) for e in endpoints] self._model = ( self._models[0] if self._models else _extract_model_from_endpoint(primary_endpoint) ) self._opts = STTOptions( sample_rate=sample_rate, buffer_size_seconds=buffer_size_seconds, enable_partial_transcripts=enable_partial_transcripts, vad_threshold=vad_threshold, vad_min_silence_duration_ms=vad_min_silence_duration_ms, vad_speech_pad_ms=vad_speech_pad_ms, enable_diarization=enable_diarization, min_speakers=min_speakers, max_speakers=max_speakers, language=language, ) if is_given(encoding): self._opts.encoding = encoding # Store any extra model-specific options self._model_options = model_options self._session = http_session self._streams = weakref.WeakSet[SpeechStream]() def _set_active_endpoint_index(self, index: int) -> None: """Update the active endpoint index (called by SpeechStream after successful failover).""" self._active_endpoint_index = index @property def model(self) -> str: return "slng" @property def provider(self) -> str: return "SLNG" @property def session(self) -> aiohttp.ClientSession: if not self._session: self._session = utils.http_context.http_session() return self._session async def _recognize_impl( self, buffer: AudioBuffer, *, language: NotGivenOr[str] = NOT_GIVEN, conn_options: APIConnectOptions, ) -> stt.SpeechEvent: """ HTTP batch recognition for non-streaming STT. Converts audio buffer to base64 and sends to SLNG HTTP endpoint. """ # Use language from parameter or fall back to instance default lang = language if is_given(language) else self._opts.language # Convert AudioBuffer to bytes audio_data = rtc.combine_audio_frames(buffer).data.tobytes() # Encode as base64 audio_b64 = base64.b64encode(audio_data).decode("utf-8") # Prepare request payload payload = { "audio_b64": audio_b64, "language": lang, } # Add any model-specific options if self._model_options: payload.update(self._model_options) try: async with self.session.post( self._model_endpoint, headers={ "Authorization": f"Bearer {self._api_key}", "Content-Type": "application/json", **( {"X-Region-Override": self._region_override_header} if self._region_override_header else {} ), }, json=payload, timeout=aiohttp.ClientTimeout( total=conn_options.timeout, sock_connect=conn_options.timeout ), ) as resp: if resp.status != 200: error_text = await resp.text() logger.error(f"[SLNG STT] HTTP error {resp.status}: {error_text}") raise APIStatusError( f"SLNG STT HTTP error {resp.status}: {error_text}", status_code=resp.status, ) data = await resp.json() # Extract transcription from response # Expected format: {"text": "...", "language": "en", "segments": [...]} text = data.get("text", "") detected_language = data.get("language", lang) segments = data.get("segments", []) # Calculate start and end times from segments start_time = segments[0].get("start", 0.0) if segments else 0.0 end_time = segments[-1].get("end", 0.0) if segments else 0.0 # Build SpeechEvent return stt.SpeechEvent( type=stt.SpeechEventType.FINAL_TRANSCRIPT, alternatives=[ stt.SpeechData( language=detected_language, text=text, confidence=1.0, # SLNG doesn't provide confidence in HTTP mode start_time=start_time, end_time=end_time, ) ], ) except aiohttp.ClientError as e: logger.error(f"[SLNG STT] HTTP connection error: {e}") raise APIConnectionError(f"SLNG STT HTTP connection error: {e}") from e except asyncio.TimeoutError: logger.error("[SLNG STT] HTTP request timed out") raise APITimeoutError("SLNG STT HTTP request timed out") from None except APIStatusError: raise except Exception as e: logger.error(f"[SLNG STT] HTTP unexpected error: {e}", exc_info=True) raise APIStatusError(f"SLNG STT HTTP error: {e}") from e def stream( self, *, language: NotGivenOr[str] = NOT_GIVEN, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS, ) -> SpeechStream: config = dataclasses.replace(self._opts) if is_given(language): config.language = language stream = SpeechStream( stt=self, conn_options=conn_options, opts=config, api_key=self._api_key, region_override_header=self._region_override_header, model_endpoints=self._model_endpoints, models=self._models, active_endpoint_index=self._active_endpoint_index, model_options=self._model_options, http_session=self.session, ) self._streams.add(stream) return stream def update_options( self, *, enable_partial_transcripts: NotGivenOr[bool] = NOT_GIVEN, enable_diarization: NotGivenOr[bool] = NOT_GIVEN, vad_threshold: NotGivenOr[float] = NOT_GIVEN, vad_min_silence_duration_ms: NotGivenOr[int] = NOT_GIVEN, vad_speech_pad_ms: NotGivenOr[int] = NOT_GIVEN, language: NotGivenOr[str] = NOT_GIVEN, buffer_size_seconds: NotGivenOr[float] = NOT_GIVEN, ) -> None: if is_given(enable_partial_transcripts): self._opts.enable_partial_transcripts = enable_partial_transcripts if is_given(enable_diarization): self._opts.enable_diarization = enable_diarization if is_given(vad_threshold): self._opts.vad_threshold = vad_threshold if is_given(vad_min_silence_duration_ms): self._opts.vad_min_silence_duration_ms = vad_min_silence_duration_ms if is_given(vad_speech_pad_ms): self._opts.vad_speech_pad_ms = vad_speech_pad_ms if is_given(language): self._opts.language = language if is_given(buffer_size_seconds): self._opts.buffer_size_seconds = buffer_size_seconds for stream in self._streams: stream.update_options( enable_partial_transcripts=enable_partial_transcripts, enable_diarization=enable_diarization, vad_threshold=vad_threshold, vad_min_silence_duration_ms=vad_min_silence_duration_ms, vad_speech_pad_ms=vad_speech_pad_ms, language=language, buffer_size_seconds=buffer_size_seconds, ) class SpeechStream(stt.SpeechStream): def __init__( self, *, stt: STT, opts: STTOptions, conn_options: APIConnectOptions, api_key: str, region_override_header: str | None, model_endpoints: list[str], models: list[str | None], active_endpoint_index: int, model_options: dict, http_session: aiohttp.ClientSession, ) -> None: super().__init__(stt=stt, conn_options=conn_options, sample_rate=opts.sample_rate) self._stt_parent: STT = stt self._opts = opts self._api_key = api_key self._region_override_header = region_override_header self._model_endpoints = model_endpoints self._models = models self._active_endpoint_index = active_endpoint_index self._model_options = model_options self._session = http_session self._speech_duration: float = 0 # keep a list of final transcripts to combine them inside the END_OF_SPEECH event self._final_events: list[SpeechEvent] = [] self._reconnect_event = asyncio.Event() def update_options( self, *, enable_partial_transcripts: NotGivenOr[bool] = NOT_GIVEN, enable_diarization: NotGivenOr[bool] = NOT_GIVEN, vad_threshold: NotGivenOr[float] = NOT_GIVEN, vad_min_silence_duration_ms: NotGivenOr[int] = NOT_GIVEN, vad_speech_pad_ms: NotGivenOr[int] = NOT_GIVEN, language: NotGivenOr[str] = NOT_GIVEN, buffer_size_seconds: NotGivenOr[float] = NOT_GIVEN, ) -> None: if is_given(enable_partial_transcripts): self._opts.enable_partial_transcripts = enable_partial_transcripts if is_given(enable_diarization): self._opts.enable_diarization = enable_diarization if is_given(vad_threshold): self._opts.vad_threshold = vad_threshold if is_given(vad_min_silence_duration_ms): self._opts.vad_min_silence_duration_ms = vad_min_silence_duration_ms if is_given(vad_speech_pad_ms): self._opts.vad_speech_pad_ms = vad_speech_pad_ms if is_given(language): self._opts.language = language if is_given(buffer_size_seconds): self._opts.buffer_size_seconds = buffer_size_seconds self._reconnect_event.set() def _samples_per_buffer(self) -> int: try: buffer_size_seconds = float(self._opts.buffer_size_seconds) except (TypeError, ValueError): buffer_size_seconds = DEFAULT_BUFFER_SIZE_SECONDS if buffer_size_seconds <= 0: buffer_size_seconds = DEFAULT_BUFFER_SIZE_SECONDS return max(1, round(self._opts.sample_rate * buffer_size_seconds)) async def _run(self) -> None: did_failover = False send: asyncio.Task[None] | None = None recv: asyncio.Task[None] | None = None wait_reconnect: asyncio.Task[bool] | None = None immediate_reconnect_attempts: dict[int, int] = {} def current_model() -> str | None: try: return self._models[self._active_endpoint_index] except Exception: return None def next_model() -> str | None: idx = self._active_endpoint_index + 1 if idx < len(self._models): return self._models[idx] return None async def failover(*, exc: BaseException | None) -> bool: from_model = current_model() exc_info = ( (type(exc), exc, exc.__traceback__) if exc is not None and exc.__traceback__ is not None else None ) if self._active_endpoint_index + 1 >= len(self._model_endpoints): logger.error( "STT fallback exhausted: from=%s", from_model, exc_info=exc_info, ) return False to_model = next_model() logger.warning( "STT attempt failed: switching %s -> %s", from_model, to_model, exc_info=exc_info, ) self._active_endpoint_index += 1 return True async def next_audio_frame() -> Any | None: async for item in self._input_ch: if isinstance(item, self._FlushSentinel): continue return item return None async def send_task( ws: aiohttp.ClientWebSocketResponse, *, pending_frames: list[Any] ) -> None: samples_per_buffer = self._samples_per_buffer() bytes_per_sample = bytes_per_frame.get(self._opts.encoding, 2) audio_bstream = utils.audio.AudioByteStream( sample_rate=self._opts.sample_rate, num_channels=1, samples_per_channel=samples_per_buffer, ) for frame in pending_frames: frames = audio_bstream.write(frame.data.tobytes()) for out in frames: if len(out.data) % bytes_per_sample != 0: continue await ws.send_bytes(bytes(out.data)) self._speech_duration += out.duration async for item in self._input_ch: if isinstance(item, self._FlushSentinel): frames = audio_bstream.flush() else: frames = audio_bstream.write(item.data.tobytes()) for frame in frames: if len(frame.data) % bytes_per_sample != 0: continue await ws.send_bytes(bytes(frame.data)) self._speech_duration += frame.duration async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None: speech_started = False while True: msg = await ws.receive() if msg.type in ( aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.CLOSE, aiohttp.WSMsgType.CLOSING, ): raise APIStatusError("SLNG connection closed unexpectedly") if msg.type != aiohttp.WSMsgType.TEXT: continue try: data = json.loads(msg.data) except json.JSONDecodeError: logger.debug("[SLNG STT] ignoring non-JSON text frame: %s", msg.data[:200]) continue if not isinstance(data, dict): continue msg_type = data.get("type") if msg_type in ("Metadata", "SpeechStarted", "UtteranceEnd"): continue if msg_type == "Results": is_final_value = data.get("is_final") if isinstance(is_final_value, str): is_final = is_final_value.strip().lower() in ("true", "1") else: is_final = bool(is_final_value) raw_channel = data.get("channel") channel = raw_channel if isinstance(raw_channel, dict) else {} raw_alts = channel.get("alternatives") alternatives = raw_alts if isinstance(raw_alts, list) else [] alt0 = ( alternatives[0] if alternatives and isinstance(alternatives[0], dict) else {} ) data = { "type": "final_transcript" if is_final else "partial_transcript", "transcript": alt0.get("transcript", ""), "confidence": alt0.get("confidence", 0.0), "words": alt0.get("words", []), "language": data.get("language", alt0.get("language")), } msg_type = data["type"] if msg_type == "Error": raise APIStatusError( f"SLNG STT error: {data.get('description') or data.get('message')}" ) if msg_type in ("partial_transcript", "final_transcript"): if ( msg_type == "partial_transcript" and not self._opts.enable_partial_transcripts ): continue text = data.get("transcript", "") is_final = msg_type == "final_transcript" if not text: # Empty-text final still consumed audio at the gateway; # emit the usage metric so billed audio gets reported. if is_final and self._speech_duration > 0: self._event_ch.send_nowait( stt.SpeechEvent( type=stt.SpeechEventType.RECOGNITION_USAGE, alternatives=[], recognition_usage=stt.RecognitionUsage( audio_duration=self._speech_duration, ), ) ) self._speech_duration = 0 continue confidence = data.get("confidence", 0.0) language = data.get("language", self._opts.language) words = data.get("words", []) # Emit START_OF_SPEECH on first transcript (interim or final) if not speech_started: speech_started = True self._event_ch.send_nowait( stt.SpeechEvent(type=stt.SpeechEventType.START_OF_SPEECH) ) if is_final: start_time = words[0].get("start", 0.0) if words else 0.0 end_time = words[-1].get("end", 0.0) if words else 0.0 else: start_time = 0.0 end_time = 0.0 event = stt.SpeechEvent( type=stt.SpeechEventType.FINAL_TRANSCRIPT if is_final else stt.SpeechEventType.INTERIM_TRANSCRIPT, alternatives=[ stt.SpeechData( language=language, text=text, confidence=confidence, start_time=start_time, end_time=end_time, ) ], ) self._event_ch.send_nowait(event) # Emit END_OF_SPEECH after each final transcript. # Note: the gateway may send multiple final_transcript messages # per utterance (e.g., sentence-by-sentence). Each final is # treated as a completed segment, so START/END bracket each one. if is_final: self._event_ch.send_nowait( stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH) ) speech_started = False if self._speech_duration > 0: self._event_ch.send_nowait( stt.SpeechEvent( type=stt.SpeechEventType.RECOGNITION_USAGE, alternatives=[], recognition_usage=stt.RecognitionUsage( audio_duration=self._speech_duration, ), ) ) self._speech_duration = 0 while True: send = None recv = None wait_reconnect = None first = await next_audio_frame() if first is None: return pending_frames: list[Any] = [first] endpoint = self._model_endpoints[self._active_endpoint_index] model = current_model() ws: aiohttp.ClientWebSocketResponse | None = None try: ws = await self._connect_ws(model_endpoint=endpoint, model=model) if did_failover: logger.info("STT switched to fallback: model=%s", model) # Propagate successful failover to parent so new streams # start from the working endpoint. self._stt_parent._set_active_endpoint_index(self._active_endpoint_index) did_failover = False send = asyncio.create_task(send_task(ws, pending_frames=pending_frames)) recv = asyncio.create_task(recv_task(ws)) wait_reconnect = asyncio.create_task(self._reconnect_event.wait()) tasks_group = asyncio.gather(send, recv) done, _ = await asyncio.wait( (tasks_group, wait_reconnect), return_when=asyncio.FIRST_COMPLETED, ) if wait_reconnect in done: self._reconnect_event.clear() tasks_group.cancel() await utils.aio.gracefully_cancel(send, recv, wait_reconnect) continue for task in done: task.result() await utils.aio.gracefully_cancel(wait_reconnect) return except Exception as exc: tasks = [t for t in (send, recv, wait_reconnect) if t is not None] if tasks: with contextlib.suppress(Exception): await utils.aio.gracefully_cancel(*tasks) if isinstance(exc, APIStatusError): status_code = _safe_error_code(exc) is_permanent_client_error = ( status_code is not None and 400 <= status_code < 500 and status_code != 429 ) endpoint_index = self._active_endpoint_index attempts = immediate_reconnect_attempts.get(endpoint_index, 0) if not is_permanent_client_error and attempts < MAX_IMMEDIATE_RETRIES: immediate_reconnect_attempts[endpoint_index] = attempts + 1 continue if not await failover(exc=exc): raise did_failover = True immediate_reconnect_attempts[self._active_endpoint_index] = 0 continue finally: if ws is not None: await ws.close() async def _connect_ws( self, *, model_endpoint: str, model: str | None ) -> aiohttp.ClientWebSocketResponse: # Match e2e test headers exactly - send both Authorization and X-API-Key headers = { "Authorization": f"Bearer {self._api_key}", "X-API-Key": self._api_key, } if self._region_override_header: headers["X-Region-Override"] = self._region_override_header # Don't enable compression - e2e tests work without it and compress=15 # was causing handshake errors with Deepgram Nova endpoint try: ws = await asyncio.wait_for( self._session.ws_connect(model_endpoint, headers=headers), self._conn_options.timeout, ) except (aiohttp.ClientConnectorError, asyncio.TimeoutError) as e: raise APIConnectionError("failed to connect to SLNG STT") from e init_message = build_stt_init_payload( model=model, language=self._opts.language, sample_rate=self._opts.sample_rate, encoding=self._opts.encoding, vad_threshold=self._opts.vad_threshold, vad_min_silence_duration_ms=self._opts.vad_min_silence_duration_ms, vad_speech_pad_ms=self._opts.vad_speech_pad_ms, enable_diarization=self._opts.enable_diarization, enable_partial_transcripts=self._opts.enable_partial_transcripts, min_speakers=self._opts.min_speakers, max_speakers=self._opts.max_speakers, model_options=self._model_options, ) try: await ws.send_str(json.dumps(init_message)) except Exception: await ws.close() raise return ws