Module livekit.plugins.smallestai

Smallest AI plugin for LiveKit Agents

See https://docs.livekit.io/agents/integrations/smallestai/ for more information.

Classes

class ChunkedStream (*,
tts: TTS,
input_text: str,
conn_options: APIConnectOptions)

Expand source code

class ChunkedStream(tts.ChunkedStream):
    """HTTP-based synthesis — used when synthesize() is called directly."""

    def __init__(self, *, tts: TTS, input_text: str, conn_options: APIConnectOptions) -> None:
        super().__init__(tts=tts, input_text=input_text, conn_options=conn_options)
        self._tts: TTS = tts
        self._opts = replace(tts._opts)

    async def _run(self, output_emitter: tts.AudioEmitter) -> None:
        try:
            data = _to_smallest_options(self._opts)
            data["text"] = self._input_text

            headers = {
                "Authorization": f"Bearer {self._opts.api_key}",
                "Content-Type": "application/json",
                "X-Source": "livekit",
                "X-LiveKit-Version": __version__,
            }
            async with self._tts._ensure_session().post(
                f"{self._opts.base_url}/tts",
                headers=headers,
                json=data,
                timeout=aiohttp.ClientTimeout(total=self._conn_options.timeout),
            ) as resp:
                if resp.status >= 400:
                    body = await resp.text()
                    raise create_api_error_from_http(body, status=resp.status)

                output_emitter.initialize(
                    request_id=utils.shortuuid(),
                    sample_rate=self._opts.sample_rate,
                    num_channels=NUM_CHANNELS,
                    mime_type=f"audio/{self._opts.output_format}",
                )

                async for chunk, _ in resp.content.iter_chunks():
                    output_emitter.push(chunk)

                output_emitter.flush()

        except asyncio.TimeoutError:
            raise APITimeoutError() from None
        except aiohttp.ClientResponseError as e:
            raise create_api_error_from_http(e.message, status=e.status) from None
        except APIStatusError:
            raise
        except Exception as e:
            raise APIConnectionError() from e

HTTP-based synthesis — used when synthesize() is called directly.

Ancestors

• livekit.agents.tts.tts.ChunkedStream
• abc.ABC

class STT (*,
model: STTModels | str = 'pulse',
language: str = 'en',
sample_rate: int = 16000,
encoding: STTEncoding | str = 'linear16',
word_timestamps: bool = True,
diarize: bool = False,
format: bool = True,
itn_normalize: bool = False,
numerals: NotGivenOr[bool] = NOT_GIVEN,
keywords: NotGivenOr[list[tuple[str, float]]] = NOT_GIVEN,
sentence_timestamps: bool = False,
redact_pii: bool = False,
redact_pci: bool = False,
endpointing: bool = True,
endpointing_timeout_ms: int | None = 300,
eou_timeout_ms: NotGivenOr[int] = NOT_GIVEN,
vad_events: bool = True,
finalize_on_flush: bool = True,
finalize_on_words: NotGivenOr[bool] = NOT_GIVEN,
max_words: NotGivenOr[int] = NOT_GIVEN,
keepalive_interval: NotGivenOr[float | None] = NOT_GIVEN,
api_key: str | None = None,
http_session: aiohttp.ClientSession | None = None,
base_url: str = 'https://api.smallest.ai/waves/v1',
stream_path: str = '/stt/live')

Expand source code

class STT(stt.STT):
    def __init__(
        self,
        *,
        model: STTModels | str = "pulse",
        language: str = "en",
        sample_rate: int = 16000,
        encoding: STTEncoding | str = "linear16",
        word_timestamps: bool = True,
        diarize: bool = False,
        format: bool = True,
        itn_normalize: bool = False,
        numerals: NotGivenOr[bool] = NOT_GIVEN,
        keywords: NotGivenOr[list[tuple[str, float]]] = NOT_GIVEN,
        sentence_timestamps: bool = False,
        redact_pii: bool = False,
        redact_pci: bool = False,
        endpointing: bool = True,
        endpointing_timeout_ms: int | None = 300,
        eou_timeout_ms: NotGivenOr[int] = NOT_GIVEN,
        vad_events: bool = True,
        finalize_on_flush: bool = True,
        finalize_on_words: NotGivenOr[bool] = NOT_GIVEN,
        max_words: NotGivenOr[int] = NOT_GIVEN,
        keepalive_interval: NotGivenOr[float | None] = NOT_GIVEN,
        api_key: str | None = None,
        http_session: aiohttp.ClientSession | None = None,
        base_url: str = SMALLEST_STT_BASE_URL,
        stream_path: str = DEFAULT_STREAM_PATH,
    ) -> None:
        """Create a new instance of Smallest AI STT.

        Args:
            model: STT model to use. ``"pulse"`` supports streaming and batch
                transcription. Other models are batch-only; ``stream()`` raises
                ``ValueError`` for them.
            language: BCP-47 language code (e.g. "en", "hi", "fr"). Use "multi" for
                automatic language detection.
            sample_rate: Audio sample rate in Hz. Supported: 8000, 16000, 22050,
                24000, 44100, 48000. Defaults to 16000. Telephony audio should be sent
                at its native 8000 Hz rather than upsampled.
            encoding: PCM encoding of the audio stream; "linear16" is the default.
            word_timestamps: Include per-word start/end timestamps and confidence
                scores on final transcripts. Defaults to True.
            diarize: Enable speaker diarization; each word carries a speaker id.
                Defaults to False.
            format: Apply punctuation and capitalization to streaming transcripts.
                Defaults to True.
            itn_normalize: Run the server's inverse text normalization so spoken
                numbers come back as digits ("twenty five" -> "25"). Independent of
                ``format``. Defaults to False.
            numerals: Legacy digits post-pass; leave unset and use ``itn_normalize``.
            keywords: Boost recognition via ``(keyword, intensifier)`` tuples, e.g.
                ``[("NVIDIA", 2.0)]``. Intensifier ~1.0 (mild) to ~5.0 (strong).
            sentence_timestamps: Include sentence-level timing as an ``"utterances"``
                list in ``SpeechData.metadata``. Defaults to False.
            redact_pii: Mask names, addresses and phone numbers with placeholder
                tokens; matches are listed in ``SpeechData.metadata["redacted_entities"]``.
                Reliable for ``language="en"`` and ``"hi"``. Defaults to False.
            redact_pci: Mask card numbers, CVVs, ZIP codes and account numbers.
                Reliable for ``language="en"`` and ``"hi"``. Defaults to False.
            endpointing: Let the server's VAD finalize the open segment once
                ``endpointing_timeout_ms`` of trailing silence has passed, so the
                last words of a turn arrive as a final without waiting for the
                blank-token endpoint at the next decode boundary. Defaults to True.
            endpointing_timeout_ms: The server VAD silence window in ms, shared by
                ``endpointing`` and ``vad_events``. Defaults to 300 so the server's final
                lands before a VAD-driven agent commits the turn (LiveKit commits about
                0.55 s after speech with its default VAD): measured against 600 on 19 call
                recordings this cut the words leaking into the next turn by 60 % at the
                same accuracy. Shorter windows start cutting inside words. ``None`` uses
                the server default (600).
            eou_timeout_ms: Silence (ms) for the transducer's blank-token endpoint.
                Unset uses the server default (800). Values below the ~1.1 s decode
                chunk make the server cut finals inside utterances; prefer
                ``endpointing`` for low latency.
            vad_events: Ask the server for ``speech_started`` / ``speech_ended``
                messages and map them to START_OF_SPEECH / END_OF_SPEECH. This is the
                only reliable end-of-speech signal: finals alone are also emitted
                mid-utterance. With it off, every final closes the turn. Defaults to True.
            finalize_on_flush: Send ``{"type": "finalize"}`` when the framework flushes
                the stream, so buffered speech is returned as a final immediately. Defaults to True.
            finalize_on_words: Cut a final every ``max_words`` words during long
                continuous speech. Unset uses the server default (on). The cut is not a
                turn end; END_OF_SPEECH still follows the VAD.
            max_words: Word count for ``finalize_on_words``. Unset uses the server default.
            keepalive_interval: Seconds between ``{"type": "keepalive"}`` messages, which
                reset the public API's inactivity timer during long pauses (it answers with
                ``{"type": "pong"}``). Defaults to 5 s on the public API and off for a
                direct server, which does not accept the message. ``None`` disables it.
            api_key: Smallest AI API key. Falls back to the SMALLEST_API_KEY
                environment variable if not provided.
            http_session: An existing aiohttp ClientSession to reuse.
            base_url: API base URL. For a directly reachable inference server pass its
                ``http://host:port`` and ``stream_path="/transcribe"``.
            stream_path: Path of the streaming endpoint under ``base_url``.
                ``"/stt/live"`` on the public API, ``"/transcribe"`` on a server.
        """
        super().__init__(
            capabilities=stt.STTCapabilities(
                streaming=model in _STREAMING_MODELS,
                interim_results=True,
                diarization=diarize,
                aligned_transcript="word" if word_timestamps else False,
            )
        )

        api_key = api_key or os.environ.get("SMALLEST_API_KEY")
        if not api_key:
            raise ValueError(
                "Smallest AI API key is required, either as argument or set "
                "SMALLEST_API_KEY environment variable"
            )

        self._opts = _STTOptions(
            model=model,
            api_key=api_key,
            language=language,
            sample_rate=sample_rate,
            encoding=encoding,
            word_timestamps=word_timestamps,
            diarize=diarize,
            format=format,
            itn_normalize=itn_normalize,
            numerals=numerals if is_given(numerals) else None,
            keywords=list(keywords) if is_given(keywords) else [],
            sentence_timestamps=sentence_timestamps,
            redact_pii=redact_pii,
            redact_pci=redact_pci,
            endpointing=endpointing,
            endpointing_timeout_ms=endpointing_timeout_ms
            if is_given(endpointing_timeout_ms)
            else None,
            eou_timeout_ms=eou_timeout_ms if is_given(eou_timeout_ms) else None,
            vad_events=vad_events,
            finalize_on_flush=finalize_on_flush,
            finalize_on_words=finalize_on_words if is_given(finalize_on_words) else None,
            max_words=max_words if is_given(max_words) else None,
            keepalive_interval=(
                keepalive_interval
                if is_given(keepalive_interval)
                else (5.0 if stream_path == DEFAULT_STREAM_PATH else None)
            ),
            base_url=base_url,
            stream_path=stream_path,
        )
        self._session = http_session
        self._streams: weakref.WeakSet[SpeechStream] = weakref.WeakSet()

    @property
    def model(self) -> str:
        return self._opts.model

    @property
    def provider(self) -> str:
        return "SmallestAI"

    def _ensure_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 = DEFAULT_API_CONNECT_OPTIONS,
    ) -> stt.SpeechEvent:
        config = self._sanitize_options(language=language)
        params: dict[str, Any] = {
            "model": config.model,
            "language": config.language,
            "encoding": config.encoding,
            "sample_rate": config.sample_rate,
            "word_timestamps": str(config.word_timestamps).lower(),
            "diarize": str(config.diarize).lower(),
        }

        try:
            async with self._ensure_session().post(
                url=f"{config.base_url}/stt/",
                headers={
                    "Authorization": f"Bearer {config.api_key}",
                    "Content-Type": "application/octet-stream",
                    "X-Source": "livekit",
                    "X-LiveKit-Version": __version__,
                },
                params=params,
                # to_wav_bytes() produces a valid WAV file; the server auto-detects format.
                data=rtc.combine_audio_frames(buffer).to_wav_bytes(),
                timeout=aiohttp.ClientTimeout(total=30, sock_connect=conn_options.timeout),
            ) as resp:
                resp.raise_for_status()
                data = await resp.json()
                return _batch_transcription_to_speech_event(config.language, data)

        except asyncio.TimeoutError as e:
            raise APITimeoutError() from e
        except aiohttp.ClientResponseError as e:
            raise APIStatusError(
                message=e.message, status_code=e.status, request_id=None, body=None
            ) from e
        except Exception as e:
            raise APIConnectionError() from e

    def stream(
        self,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> SpeechStream:
        if not self.capabilities.streaming:
            raise ValueError(
                f"{self._opts.model} does not support streaming; use recognize() for batch transcription"
            )
        config = self._sanitize_options(language=language)
        stream = SpeechStream(
            stt=self,
            conn_options=conn_options,
            opts=config,
            http_session=self._ensure_session(),
        )
        self._streams.add(stream)
        return stream

    def update_options(
        self,
        *,
        model: NotGivenOr[STTModels | str] = NOT_GIVEN,
        language: NotGivenOr[str] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding | str] = NOT_GIVEN,
        format: NotGivenOr[bool] = NOT_GIVEN,
        itn_normalize: NotGivenOr[bool] = NOT_GIVEN,
        keywords: NotGivenOr[list[tuple[str, float]]] = NOT_GIVEN,
        sentence_timestamps: NotGivenOr[bool] = NOT_GIVEN,
        redact_pii: NotGivenOr[bool] = NOT_GIVEN,
        redact_pci: NotGivenOr[bool] = NOT_GIVEN,
        endpointing: NotGivenOr[bool] = NOT_GIVEN,
        endpointing_timeout_ms: NotGivenOr[int] = NOT_GIVEN,
        eou_timeout_ms: NotGivenOr[int] = NOT_GIVEN,
        vad_events: NotGivenOr[bool] = NOT_GIVEN,
        finalize_on_flush: NotGivenOr[bool] = NOT_GIVEN,
    ) -> None:
        """Update STT options; propagates to all active streams (triggers reconnect)."""
        if is_given(model):
            self._opts.model = model
            self._capabilities.streaming = model in _STREAMING_MODELS
        self._opts = _apply_updates(
            self._opts,
            language=language,
            sample_rate=sample_rate,
            encoding=encoding,
            format=format,
            itn_normalize=itn_normalize,
            keywords=keywords,
            sentence_timestamps=sentence_timestamps,
            redact_pii=redact_pii,
            redact_pci=redact_pci,
            endpointing=endpointing,
            endpointing_timeout_ms=endpointing_timeout_ms,
            eou_timeout_ms=eou_timeout_ms,
            vad_events=vad_events,
            finalize_on_flush=finalize_on_flush,
        )
        for stream in self._streams:
            stream.update_options(
                model=model,
                language=language,
                sample_rate=sample_rate,
                encoding=encoding,
                format=format,
                itn_normalize=itn_normalize,
                keywords=keywords,
                sentence_timestamps=sentence_timestamps,
                redact_pii=redact_pii,
                redact_pci=redact_pci,
                endpointing=endpointing,
                endpointing_timeout_ms=endpointing_timeout_ms,
                eou_timeout_ms=eou_timeout_ms,
                vad_events=vad_events,
                finalize_on_flush=finalize_on_flush,
            )

    def _sanitize_options(self, *, language: NotGivenOr[str] = NOT_GIVEN) -> _STTOptions:
        config = replace(self._opts)
        if is_given(language):
            config.language = language
        return config

Helper class that provides a standard way to create an ABC using inheritance.

Create a new instance of Smallest AI STT.

Args

model

STT model to use. "pulse" supports streaming and batch transcription. Other models are batch-only; stream() raises ValueError for them.

language

BCP-47 language code (e.g. "en", "hi", "fr"). Use "multi" for automatic language detection.

sample_rate

Audio sample rate in Hz. Supported: 8000, 16000, 22050, 24000, 44100, 48000. Defaults to 16000. Telephony audio should be sent at its native 8000 Hz rather than upsampled.

encoding

PCM encoding of the audio stream; "linear16" is the default.

word_timestamps

Include per-word start/end timestamps and confidence scores on final transcripts. Defaults to True.

diarize

Enable speaker diarization; each word carries a speaker id. Defaults to False.

format

Apply punctuation and capitalization to streaming transcripts. Defaults to True.

itn_normalize

Run the server's inverse text normalization so spoken numbers come back as digits ("twenty five" -> "25"). Independent of format. Defaults to False.

numerals

Legacy digits post-pass; leave unset and use itn_normalize.

keywords

Boost recognition via (keyword, intensifier) tuples, e.g. [("NVIDIA", 2.0)]. Intensifier ~1.0 (mild) to ~5.0 (strong).

sentence_timestamps

Include sentence-level timing as an "utterances" list in SpeechData.metadata. Defaults to False.

redact_pii

Mask names, addresses and phone numbers with placeholder tokens; matches are listed in SpeechData.metadata["redacted_entities"]. Reliable for language="en" and "hi". Defaults to False.

redact_pci

Mask card numbers, CVVs, ZIP codes and account numbers. Reliable for language="en" and "hi". Defaults to False.

endpointing

Let the server's VAD finalize the open segment once endpointing_timeout_ms of trailing silence has passed, so the last words of a turn arrive as a final without waiting for the blank-token endpoint at the next decode boundary. Defaults to True.

endpointing_timeout_ms

The server VAD silence window in ms, shared by endpointing and vad_events. Defaults to 300 so the server's final lands before a VAD-driven agent commits the turn (LiveKit commits about 0.55 s after speech with its default VAD): measured against 600 on 19 call recordings this cut the words leaking into the next turn by 60 % at the same accuracy. Shorter windows start cutting inside words. None uses the server default (600).

eou_timeout_ms

Silence (ms) for the transducer's blank-token endpoint. Unset uses the server default (800). Values below the ~1.1 s decode chunk make the server cut finals inside utterances; prefer endpointing for low latency.

vad_events

Ask the server for speech_started / speech_ended messages and map them to START_OF_SPEECH / END_OF_SPEECH. This is the only reliable end-of-speech signal: finals alone are also emitted mid-utterance. With it off, every final closes the turn. Defaults to True.

finalize_on_flush

Send {"type": "finalize"} when the framework flushes the stream, so buffered speech is returned as a final immediately. Defaults to True.

finalize_on_words

Cut a final every max_words words during long continuous speech. Unset uses the server default (on). The cut is not a turn end; END_OF_SPEECH still follows the VAD.

max_words

Word count for finalize_on_words. Unset uses the server default.

keepalive_interval

Seconds between {"type": "keepalive"} messages, which reset the public API's inactivity timer during long pauses (it answers with {"type": "pong"}). Defaults to 5 s on the public API and off for a direct server, which does not accept the message. None disables it.

api_key

Smallest AI API key. Falls back to the SMALLEST_API_KEY environment variable if not provided.

http_session

An existing aiohttp ClientSession to reuse.

base_url

API base URL. For a directly reachable inference server pass its http://host:port and stream_path="/transcribe".

stream_path

Path of the streaming endpoint under base_url. "/stt/live" on the public API, "/transcribe" on a server.

Ancestors

• livekit.agents.stt.stt.STT
• abc.ABC
• EventEmitter
• typing.Generic

Instance variables

prop model : str

Expand source code

@property
def model(self) -> str:
    return self._opts.model

Get the model name/identifier for this STT instance.

Returns

The model name if available, "unknown" otherwise.

Note

Plugins should override this property to provide their model information.

prop provider : str

Expand source code

@property
def provider(self) -> str:
    return "SmallestAI"

Get the provider name/identifier for this STT instance.

Returns

The provider name if available, "unknown" otherwise.

Note

Plugins should override this property to provide their provider information.

Methods

def stream(self,
*,
language: NotGivenOr[str] = NOT_GIVEN,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> livekit.plugins.smallestai.stt.SpeechStream

Expand source code

def stream(
    self,
    *,
    language: NotGivenOr[str] = NOT_GIVEN,
    conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
) -> SpeechStream:
    if not self.capabilities.streaming:
        raise ValueError(
            f"{self._opts.model} does not support streaming; use recognize() for batch transcription"
        )
    config = self._sanitize_options(language=language)
    stream = SpeechStream(
        stt=self,
        conn_options=conn_options,
        opts=config,
        http_session=self._ensure_session(),
    )
    self._streams.add(stream)
    return stream

def update_options(self,
*,
model: NotGivenOr[STTModels | str] = NOT_GIVEN,
language: NotGivenOr[str] = NOT_GIVEN,
sample_rate: NotGivenOr[int] = NOT_GIVEN,
encoding: NotGivenOr[STTEncoding | str] = NOT_GIVEN,
format: NotGivenOr[bool] = NOT_GIVEN,
itn_normalize: NotGivenOr[bool] = NOT_GIVEN,
keywords: NotGivenOr[list[tuple[str, float]]] = NOT_GIVEN,
sentence_timestamps: NotGivenOr[bool] = NOT_GIVEN,
redact_pii: NotGivenOr[bool] = NOT_GIVEN,
redact_pci: NotGivenOr[bool] = NOT_GIVEN,
endpointing: NotGivenOr[bool] = NOT_GIVEN,
endpointing_timeout_ms: NotGivenOr[int] = NOT_GIVEN,
eou_timeout_ms: NotGivenOr[int] = NOT_GIVEN,
vad_events: NotGivenOr[bool] = NOT_GIVEN,
finalize_on_flush: NotGivenOr[bool] = NOT_GIVEN) ‑> None

Expand source code

def update_options(
    self,
    *,
    model: NotGivenOr[STTModels | str] = NOT_GIVEN,
    language: NotGivenOr[str] = NOT_GIVEN,
    sample_rate: NotGivenOr[int] = NOT_GIVEN,
    encoding: NotGivenOr[STTEncoding | str] = NOT_GIVEN,
    format: NotGivenOr[bool] = NOT_GIVEN,
    itn_normalize: NotGivenOr[bool] = NOT_GIVEN,
    keywords: NotGivenOr[list[tuple[str, float]]] = NOT_GIVEN,
    sentence_timestamps: NotGivenOr[bool] = NOT_GIVEN,
    redact_pii: NotGivenOr[bool] = NOT_GIVEN,
    redact_pci: NotGivenOr[bool] = NOT_GIVEN,
    endpointing: NotGivenOr[bool] = NOT_GIVEN,
    endpointing_timeout_ms: NotGivenOr[int] = NOT_GIVEN,
    eou_timeout_ms: NotGivenOr[int] = NOT_GIVEN,
    vad_events: NotGivenOr[bool] = NOT_GIVEN,
    finalize_on_flush: NotGivenOr[bool] = NOT_GIVEN,
) -> None:
    """Update STT options; propagates to all active streams (triggers reconnect)."""
    if is_given(model):
        self._opts.model = model
        self._capabilities.streaming = model in _STREAMING_MODELS
    self._opts = _apply_updates(
        self._opts,
        language=language,
        sample_rate=sample_rate,
        encoding=encoding,
        format=format,
        itn_normalize=itn_normalize,
        keywords=keywords,
        sentence_timestamps=sentence_timestamps,
        redact_pii=redact_pii,
        redact_pci=redact_pci,
        endpointing=endpointing,
        endpointing_timeout_ms=endpointing_timeout_ms,
        eou_timeout_ms=eou_timeout_ms,
        vad_events=vad_events,
        finalize_on_flush=finalize_on_flush,
    )
    for stream in self._streams:
        stream.update_options(
            model=model,
            language=language,
            sample_rate=sample_rate,
            encoding=encoding,
            format=format,
            itn_normalize=itn_normalize,
            keywords=keywords,
            sentence_timestamps=sentence_timestamps,
            redact_pii=redact_pii,
            redact_pci=redact_pci,
            endpointing=endpointing,
            endpointing_timeout_ms=endpointing_timeout_ms,
            eou_timeout_ms=eou_timeout_ms,
            vad_events=vad_events,
            finalize_on_flush=finalize_on_flush,
        )

Update STT options; propagates to all active streams (triggers reconnect).

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once

class SpeechStream (*,
stt: STT,
opts: _STTOptions,
conn_options: APIConnectOptions,
http_session: aiohttp.ClientSession)

Expand source code

class SpeechStream(stt.SpeechStream):
    _FINALIZE_MSG: str = json.dumps({"type": "finalize"})
    _CLOSE_STREAM_MSG: str = json.dumps({"type": "close_stream"})
    _KEEPALIVE_MSG: str = json.dumps({"type": "keepalive"})

    def __init__(
        self,
        *,
        stt: STT,
        opts: _STTOptions,
        conn_options: APIConnectOptions,
        http_session: aiohttp.ClientSession,
    ) -> None:
        super().__init__(stt=stt, conn_options=conn_options, sample_rate=opts.sample_rate)
        self._opts = opts
        self._session = http_session
        self._speaking = False
        self._session_id = ""
        self._reconnect_event = asyncio.Event()
        self._audio_duration_collector = _PeriodicCollector(
            callback=self._on_audio_duration_report, duration=5.0
        )

    def update_options(
        self,
        *,
        model: NotGivenOr[STTModels | str] = NOT_GIVEN,
        language: NotGivenOr[str] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding | str] = NOT_GIVEN,
        format: NotGivenOr[bool] = NOT_GIVEN,
        itn_normalize: NotGivenOr[bool] = NOT_GIVEN,
        keywords: NotGivenOr[list[tuple[str, float]]] = NOT_GIVEN,
        sentence_timestamps: NotGivenOr[bool] = NOT_GIVEN,
        redact_pii: NotGivenOr[bool] = NOT_GIVEN,
        redact_pci: NotGivenOr[bool] = NOT_GIVEN,
        endpointing: NotGivenOr[bool] = NOT_GIVEN,
        endpointing_timeout_ms: NotGivenOr[int] = NOT_GIVEN,
        eou_timeout_ms: NotGivenOr[int] = NOT_GIVEN,
        vad_events: NotGivenOr[bool] = NOT_GIVEN,
        finalize_on_flush: NotGivenOr[bool] = NOT_GIVEN,
    ) -> None:
        self._opts = _apply_updates(
            self._opts,
            model=model,
            language=language,
            sample_rate=sample_rate,
            encoding=encoding,
            format=format,
            itn_normalize=itn_normalize,
            keywords=keywords,
            sentence_timestamps=sentence_timestamps,
            redact_pii=redact_pii,
            redact_pci=redact_pci,
            endpointing=endpointing,
            endpointing_timeout_ms=endpointing_timeout_ms,
            eou_timeout_ms=eou_timeout_ms,
            vad_events=vad_events,
            finalize_on_flush=finalize_on_flush,
        )
        self._reconnect_event.set()

    async def _run(self) -> None:
        closing_ws = False

        async def keepalive_task(ws: aiohttp.ClientWebSocketResponse) -> None:
            # Resets the public API's inactivity timer while no audio flows; the server
            # answers with a pong that recv_task ignores.
            interval = self._opts.keepalive_interval
            if interval is None or interval <= 0:
                await asyncio.Event().wait()  # disabled: idle for the life of the connection
                return
            try:
                while True:
                    await asyncio.sleep(interval)
                    await ws.send_str(SpeechStream._KEEPALIVE_MSG)
            except (aiohttp.ClientError, ConnectionError) as e:
                # With no audio flowing this write is where a dropped socket shows first.
                if closing_ws or self._session.closed:
                    return
                raise APIConnectionError("Smallest AI STT connection closed unexpectedly") from e

        @utils.log_exceptions(logger=logger)
        async def send_task(ws: aiohttp.ClientWebSocketResponse) -> None:
            nonlocal closing_ws

            # 50 ms packets, the cadence the Pulse docs recommend.
            samples_per_chunk = self._opts.sample_rate // 20
            audio_bstream = utils.audio.AudioByteStream(
                sample_rate=self._opts.sample_rate,
                num_channels=NUM_CHANNELS,
                samples_per_channel=samples_per_chunk,
            )

            try:
                async for data in self._input_ch:
                    frames: list[rtc.AudioFrame] = []
                    flushed = False
                    if isinstance(data, rtc.AudioFrame):
                        frames.extend(audio_bstream.write(data.data.tobytes()))
                    elif isinstance(data, self._FlushSentinel):
                        frames.extend(audio_bstream.flush())
                        flushed = True

                    for frame in frames:
                        self._audio_duration_collector.push(frame.duration)
                        await ws.send_bytes(frame.data.tobytes())

                    if flushed:
                        # The framework marked a segment end: the buffered tail is on the
                        # wire, now ask for it back as a final instead of leaving it to
                        # the server's endpointing.
                        self._audio_duration_collector.flush()
                        if self._opts.finalize_on_flush:
                            await ws.send_str(SpeechStream._FINALIZE_MSG)

                # Input closed: the server flushes, answers with is_last and closes.
                closing_ws = True
                await ws.send_str(SpeechStream._CLOSE_STREAM_MSG)
            except (aiohttp.ClientError, ConnectionError) as e:
                # A dropped socket surfaces on the write side first. Expected closes just
                # return; anything else reconnects through _run, like recv_task.
                if closing_ws or self._session.closed:
                    return
                raise APIConnectionError("Smallest AI STT connection closed unexpectedly") from e

        @utils.log_exceptions(logger=logger)
        async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
            nonlocal closing_ws
            while True:
                msg = await ws.receive()
                if msg.type in (
                    aiohttp.WSMsgType.CLOSED,
                    aiohttp.WSMsgType.CLOSE,
                    aiohttp.WSMsgType.CLOSING,
                ):
                    if closing_ws or self._session.closed:
                        return
                    raise APIStatusError(
                        message="Smallest AI STT connection closed unexpectedly",
                        status_code=ws.close_code or -1,
                        body=f"{msg.data=} {msg.extra=}",
                    )

                if msg.type == aiohttp.WSMsgType.ERROR:
                    if closing_ws or self._session.closed:
                        return
                    # An unanswered heartbeat ping closes the socket from our side and
                    # arrives here, not as a close frame; ws.exception() holds the reason.
                    raise APIConnectionError("Smallest AI STT connection lost") from ws.exception()

                if msg.type != aiohttp.WSMsgType.TEXT:
                    logger.warning("unexpected Smallest AI STT message type: %s", msg.type)
                    continue

                try:
                    data = json.loads(msg.data)
                except json.JSONDecodeError:
                    logger.warning(
                        "failed to parse Smallest AI STT message", extra={"lk.pii.data": msg.data}
                    )
                    continue

                # Errors: {"type": "error", "status", "message"} from the public API,
                # {"error": "..."} from a server reached directly.
                if data.get("type") == "error" or (
                    "error" in data and "transcript" not in data and "transcription" not in data
                ):
                    # The payload may echo user content; it goes only in the PII-marked
                    # log field and the exception body, not in the message the retry
                    # logger prints.
                    logger.error("Smallest AI STT error", extra={"lk.pii.data": data})
                    raise APIStatusError(
                        message=f"Smallest AI STT error ({data.get('status') or 'error'})",
                        status_code=-1,
                        request_id=self._session_id or None,
                        body=data,
                    )
                if data.get("type") == "pong":
                    continue

                self._process_stream_event(data)

                # The session is fully flushed; nothing else will arrive.
                if data.get("is_last"):
                    return

        ws: aiohttp.ClientWebSocketResponse | None = None
        while True:
            try:
                ws = await self._connect_ws()
                tasks = [
                    asyncio.create_task(send_task(ws)),
                    asyncio.create_task(recv_task(ws)),
                ]
                # Not in the gather: recv_task ends on is_last before the socket closes, and the
                # session must not wait on a keepalive loop. Its failure still ends the session.
                keepalive = asyncio.create_task(keepalive_task(ws))
                tasks_group = asyncio.gather(*tasks)
                wait_reconnect_task = asyncio.create_task(self._reconnect_event.wait())
                try:
                    done, _ = await asyncio.wait(
                        (tasks_group, wait_reconnect_task, keepalive),
                        return_when=asyncio.FIRST_COMPLETED,
                    )
                    for task in done:
                        if task != wait_reconnect_task:
                            task.result()

                    if wait_reconnect_task not in done:
                        break

                    self._reconnect_event.clear()
                finally:
                    await utils.aio.gracefully_cancel(*tasks, wait_reconnect_task, keepalive)
                    tasks_group.cancel()
                    tasks_group.exception()
            finally:
                if ws is not None:
                    await ws.close()
                self._audio_duration_collector.flush()
                # A socket that ends mid-utterance (reconnect, drop) takes the server's
                # speech_ended with it: close the turn here so START/END stay paired.
                self._end_speech()

    def _stream_params(self) -> dict[str, Any]:
        """Query parameters for the streaming connection; unset knobs are left to the server."""
        o = self._opts
        params: dict[str, Any] = {
            "model": o.model,
            "language": o.language,
            "encoding": o.encoding,
            "sample_rate": o.sample_rate,
            "word_timestamps": str(o.word_timestamps).lower(),
            "diarize": str(o.diarize).lower(),
            "format": str(o.format).lower(),
            "itn_normalize": str(o.itn_normalize).lower(),
            "sentence_timestamps": str(o.sentence_timestamps).lower(),
            "redact_pii": str(o.redact_pii).lower(),
            "redact_pci": str(o.redact_pci).lower(),
            "endpointing": str(o.endpointing).lower(),
            "vad_events": str(o.vad_events).lower(),
        }
        if o.numerals is not None:
            params["numerals"] = str(o.numerals).lower()
        if o.endpointing_timeout_ms is not None:
            params["endpointing_timeout"] = o.endpointing_timeout_ms
        if o.eou_timeout_ms is not None:
            params["eou_timeout_ms"] = o.eou_timeout_ms
        if o.finalize_on_words is not None:
            params["finalize_on_words"] = str(o.finalize_on_words).lower()
        if o.max_words is not None:
            params["max_words"] = o.max_words
        if o.keywords:
            params["keywords"] = ",".join(f"{kw}:{boost:g}" for kw, boost in o.keywords)
        return params

    def _ws_url(self) -> str:
        base = self._opts.base_url.replace("https://", "wss://", 1).replace("http://", "ws://", 1)
        return f"{base.rstrip('/')}{self._opts.stream_path}?{urlencode(self._stream_params())}"

    async def _connect_ws(self) -> aiohttp.ClientWebSocketResponse:
        t0 = time.perf_counter()
        try:
            ws = await asyncio.wait_for(
                self._session.ws_connect(
                    self._ws_url(),
                    headers={
                        "Authorization": f"Bearer {self._opts.api_key}",
                        "X-Source": "livekit",
                        "X-LiveKit-Version": __version__,
                    },
                    # Without a heartbeat a half-open socket is never noticed: recv_task
                    # parks on receive() and the exception-driven reconnect never runs.
                    heartbeat=30.0,
                ),
                self._conn_options.timeout,
            )
        except asyncio.TimeoutError:
            raise APIConnectionError("failed to connect to Smallest AI STT") from None
        except aiohttp.ClientResponseError as e:
            # RequestInfo carries the request headers, so chaining this error or
            # formatting it puts the API key in the exception repr.
            raise APIStatusError(
                message=e.message, status_code=e.status, request_id=None, body=None
            ) from None
        except Exception as e:
            raise APIConnectionError(
                f"failed to connect to Smallest AI STT ({type(e).__name__})"
            ) from None

        self._report_connection_acquired(time.perf_counter() - t0, False)
        logger.debug("established Smallest AI STT WebSocket connection")
        return ws

    def _on_audio_duration_report(self, duration: float) -> None:
        self._event_ch.send_nowait(
            stt.SpeechEvent(
                type=stt.SpeechEventType.RECOGNITION_USAGE,
                request_id=self._session_id,
                alternatives=[],
                recognition_usage=stt.RecognitionUsage(audio_duration=duration),
            )
        )

    def _start_speech(self) -> None:
        if not self._speaking:
            self._speaking = True
            self._event_ch.send_nowait(stt.SpeechEvent(type=stt.SpeechEventType.START_OF_SPEECH))

    def _end_speech(self) -> None:
        if self._speaking:
            self._speaking = False
            self._event_ch.send_nowait(stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH))

    def _process_stream_event(self, data: dict[str, Any]) -> None:
        session_id = data.get("session_id", "")
        if session_id:
            self._session_id = session_id

        msg_type = data.get("type", "")
        # Acoustic VAD messages: the speaker started or stopped, independent of decoding.
        if msg_type == "speech_started":
            self._start_speech()
            return
        if msg_type == "speech_ended":
            self._end_speech()
            return

        # The gateway duplicates the server's `transcript` into `transcription`.
        transcript = data.get("transcript") or data.get("transcription") or ""
        is_final = bool(data.get("is_final", False))
        is_last = bool(data.get("is_last", False))
        from_finalize = bool(data.get("from_finalize", False))

        # Punctuation-only text ("." or "।" landing in its own chunk) is not a user turn.
        if _WORD_RE.search(transcript):
            # Quiet speech the server VAD missed, or vad_events off: the text opens the turn.
            self._start_speech()
            alts = _transcript_to_speech_data(
                language=self._opts.language,
                data=data,
                transcript=transcript,
                start_time_offset=self.start_time_offset,
                diarize=self._opts.diarize,
            )
            self._event_ch.send_nowait(
                stt.SpeechEvent(
                    type=(
                        stt.SpeechEventType.FINAL_TRANSCRIPT
                        if is_final
                        else stt.SpeechEventType.INTERIM_TRANSCRIPT
                    ),
                    request_id=self._session_id,
                    alternatives=alts,
                )
            )

        # A final closes the turn when it answers a finalize, ends the stream, or the
        # server says the speaker paused. Other finals are cuts inside an utterance
        # (word rollover, blank-token endpoint); with vad_events the turn end comes
        # from speech_ended, without them every final has to count as one.
        if is_final and (
            is_last
            or from_finalize
            or data.get("speech_final") is True
            or not self._opts.vad_events
        ):
            self._end_speech()

Helper class that provides a standard way to create an ABC using inheritance.

Args: sample_rate : int or None, optional The desired sample rate for the audio input. If specified, the audio input will be automatically resampled to match the given sample rate before being processed for Speech-to-Text. If not provided (None), the input will retain its original sample rate.

Ancestors

• livekit.agents.stt.stt.RecognizeStream
• abc.ABC

Methods

def update_options(self,
*,
model: NotGivenOr[STTModels | str] = NOT_GIVEN,
language: NotGivenOr[str] = NOT_GIVEN,
sample_rate: NotGivenOr[int] = NOT_GIVEN,
encoding: NotGivenOr[STTEncoding | str] = NOT_GIVEN,
format: NotGivenOr[bool] = NOT_GIVEN,
itn_normalize: NotGivenOr[bool] = NOT_GIVEN,
keywords: NotGivenOr[list[tuple[str, float]]] = NOT_GIVEN,
sentence_timestamps: NotGivenOr[bool] = NOT_GIVEN,
redact_pii: NotGivenOr[bool] = NOT_GIVEN,
redact_pci: NotGivenOr[bool] = NOT_GIVEN,
endpointing: NotGivenOr[bool] = NOT_GIVEN,
endpointing_timeout_ms: NotGivenOr[int] = NOT_GIVEN,
eou_timeout_ms: NotGivenOr[int] = NOT_GIVEN,
vad_events: NotGivenOr[bool] = NOT_GIVEN,
finalize_on_flush: NotGivenOr[bool] = NOT_GIVEN) ‑> None

Expand source code

def update_options(
    self,
    *,
    model: NotGivenOr[STTModels | str] = NOT_GIVEN,
    language: NotGivenOr[str] = NOT_GIVEN,
    sample_rate: NotGivenOr[int] = NOT_GIVEN,
    encoding: NotGivenOr[STTEncoding | str] = NOT_GIVEN,
    format: NotGivenOr[bool] = NOT_GIVEN,
    itn_normalize: NotGivenOr[bool] = NOT_GIVEN,
    keywords: NotGivenOr[list[tuple[str, float]]] = NOT_GIVEN,
    sentence_timestamps: NotGivenOr[bool] = NOT_GIVEN,
    redact_pii: NotGivenOr[bool] = NOT_GIVEN,
    redact_pci: NotGivenOr[bool] = NOT_GIVEN,
    endpointing: NotGivenOr[bool] = NOT_GIVEN,
    endpointing_timeout_ms: NotGivenOr[int] = NOT_GIVEN,
    eou_timeout_ms: NotGivenOr[int] = NOT_GIVEN,
    vad_events: NotGivenOr[bool] = NOT_GIVEN,
    finalize_on_flush: NotGivenOr[bool] = NOT_GIVEN,
) -> None:
    self._opts = _apply_updates(
        self._opts,
        model=model,
        language=language,
        sample_rate=sample_rate,
        encoding=encoding,
        format=format,
        itn_normalize=itn_normalize,
        keywords=keywords,
        sentence_timestamps=sentence_timestamps,
        redact_pii=redact_pii,
        redact_pci=redact_pci,
        endpointing=endpointing,
        endpointing_timeout_ms=endpointing_timeout_ms,
        eou_timeout_ms=eou_timeout_ms,
        vad_events=vad_events,
        finalize_on_flush=finalize_on_flush,
    )
    self._reconnect_event.set()

class TTS (*,
api_key: str | None = None,
model: TTSModels | str = 'lightning_v3.1_pro',
voice_id: str | None = None,
sample_rate: int = 24000,
speed: float = 1.0,
language: str = 'en',
output_format: TTSEncoding | str = 'pcm',
word_timestamps: bool = False,
max_buffer_flush_ms: int = 0,
use_continuations: bool = True,
max_buffer_delay_ms: int = 3000,
base_url: str = 'https://api.smallest.ai/waves/v1',
ws_url: str = 'wss://api.smallest.ai/waves/v1/tts/live',
http_session: aiohttp.ClientSession | None = None)

Expand source code

class TTS(tts.TTS):
    def __init__(
        self,
        *,
        api_key: str | None = None,
        model: TTSModels | str = "lightning_v3.1_pro",
        voice_id: str | None = None,
        sample_rate: int = 24000,
        speed: float = 1.0,
        language: str = "en",
        output_format: TTSEncoding | str = "pcm",
        word_timestamps: bool = False,
        max_buffer_flush_ms: int = 0,
        use_continuations: bool = True,
        max_buffer_delay_ms: int = 3000,
        base_url: str = SMALLEST_BASE_URL,
        ws_url: str = SMALLEST_WS_URL,
        http_session: aiohttp.ClientSession | None = None,
    ) -> None:
        """
        Create a new instance of Smallest AI Lightning TTS.

        Args:
            api_key: Your Smallest AI API key.
            model: The TTS model to use. Use "lightning_v3.1" for the standard model with
                217 voices across 12 languages, or "lightning_v3.1_pro" (default) for the
                premium pool with curated American, British, and Indian voices at 44.1 kHz.
            voice_id: The voice ID to use for synthesis. Defaults to "meher" for
                "lightning_v3.1_pro" and "sophia" for all other models. Pro voices must be
                paired with "lightning_v3.1_pro"; standard voices with "lightning_v3.1".
            sample_rate: Sample rate for the audio output. Both models are natively 44.1 kHz;
                supported rates are 8000, 16000, 24000, and 44100.
            speed: Speed of the speech synthesis (0.5–2.0).
            language: Language of the text to be synthesized. Use "auto" for automatic
                detection and code-switching. Pro supports "en", "hi", and "auto" only.
            output_format: Output format for HTTP synthesize() calls ("pcm", "mp3", "wav",
                "ulaw", "alaw"). WebSocket streaming always returns PCM.
            word_timestamps: Request per-word timing events from the server and emit them
                as timed transcript entries alongside audio. Applies to WebSocket streaming
                only; HTTP synthesize() returns raw audio without word events. Disabled by
                default. Supported on base-queue English + Hindi voices (meher, devansh,
                kartik, maithili, liam, avery); other voices silently emit no word events.
            max_buffer_flush_ms: Server-side buffer bound (milliseconds) for the legacy
                ``continue``/``flush`` WebSocket protocol (used when ``use_continuations``
                is False). As text tokens are streamed in with ``continue: true``, the
                server accumulates them and forces partial audio output once this many
                milliseconds of text have buffered, without waiting for an explicit flush.
                ``0`` (default) disables time-based forced flushing, so audio for a segment
                is produced when the segment's end-of-input flush is sent. Ignored when
                ``use_continuations`` is True.
            use_continuations: Use the ``context_id``-based continuations protocol instead
                of the legacy ``continue``/``flush`` fields. Continuations let the server
                release buffered text as soon as it hits a natural sentence boundary (or
                ``max_buffer_delay_ms`` elapses, whichever comes first) instead of holding
                everything until an explicit flush, and carry prosody across fragments by
                priming each release with the audio of the previous one. Defaults to True;
                set False to fall back to the legacy protocol.
            max_buffer_delay_ms: Upper bound (0-5000ms) on how long the server holds a
                buffered fragment waiting for a clean sentence boundary before speaking it
                anyway. Only applies when ``use_continuations`` is True. Defaults to 3000ms
                (the server default).
            base_url: Base URL for the Smallest AI HTTP API.
            ws_url: WebSocket URL for low-latency streaming synthesis.
            http_session: An existing aiohttp ClientSession to use.
        """
        super().__init__(
            capabilities=tts.TTSCapabilities(
                streaming=True,
                aligned_transcript=word_timestamps,
            ),
            sample_rate=sample_rate,
            num_channels=NUM_CHANNELS,
        )

        api_key = api_key or os.environ.get("SMALLEST_API_KEY")
        if not api_key:
            raise ValueError(
                "Smallest.ai API key is required, either as argument or set"
                " SMALLEST_API_KEY environment variable"
            )

        if voice_id is None:
            voice_id = "meher" if model == "lightning_v3.1_pro" else "sophia"

        self._opts = _TTSOptions(
            model=model,
            api_key=api_key,
            voice_id=voice_id,
            sample_rate=sample_rate,
            speed=speed,
            language=LanguageCode(language),
            output_format=output_format,
            word_timestamps=word_timestamps,
            max_buffer_flush_ms=max_buffer_flush_ms,
            use_continuations=use_continuations,
            max_buffer_delay_ms=max_buffer_delay_ms,
            base_url=base_url,
            ws_url=ws_url,
        )
        self._session = http_session
        self._pool = utils.ConnectionPool[aiohttp.ClientWebSocketResponse](
            connect_cb=self._connect_ws,
            close_cb=self._close_ws,
            max_session_duration=3600,
            mark_refreshed_on_get=False,
        )

    @property
    def model(self) -> str:
        return self._opts.model

    @property
    def provider(self) -> str:
        return "SmallestAI"

    def _ensure_session(self) -> aiohttp.ClientSession:
        if not self._session:
            self._session = utils.http_context.http_session()
        return self._session

    async def _connect_ws(self, timeout: float) -> aiohttp.ClientWebSocketResponse:
        return await asyncio.wait_for(
            self._ensure_session().ws_connect(
                self._opts.ws_url,
                headers={
                    "Authorization": f"Bearer {self._opts.api_key}",
                    "X-Source": "livekit",
                    "X-LiveKit-Version": __version__,
                },
            ),
            timeout,
        )

    async def _close_ws(self, ws: aiohttp.ClientWebSocketResponse) -> None:
        await ws.close()

    def update_options(
        self,
        *,
        model: NotGivenOr[TTSModels | str] = NOT_GIVEN,
        voice_id: NotGivenOr[str] = NOT_GIVEN,
        speed: NotGivenOr[float] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        language: NotGivenOr[str] = NOT_GIVEN,
        output_format: NotGivenOr[TTSEncoding | str] = NOT_GIVEN,
        word_timestamps: NotGivenOr[bool] = NOT_GIVEN,
        max_buffer_flush_ms: NotGivenOr[int] = NOT_GIVEN,
        use_continuations: NotGivenOr[bool] = NOT_GIVEN,
        max_buffer_delay_ms: NotGivenOr[int] = NOT_GIVEN,
    ) -> None:
        """Update TTS options."""
        if is_given(model):
            self._opts.model = model
        if is_given(voice_id):
            self._opts.voice_id = voice_id
        if is_given(speed):
            self._opts.speed = speed
        if is_given(sample_rate):
            self._opts.sample_rate = sample_rate
            # keep the base class property in sync; the framework reads tts.sample_rate
            self._sample_rate = sample_rate
        if is_given(language):
            self._opts.language = LanguageCode(language)
        if is_given(output_format):
            self._opts.output_format = output_format
        if is_given(word_timestamps):
            self._opts.word_timestamps = word_timestamps
            self._capabilities.aligned_transcript = word_timestamps
        if is_given(max_buffer_flush_ms):
            self._opts.max_buffer_flush_ms = max_buffer_flush_ms
        if is_given(use_continuations):
            self._opts.use_continuations = use_continuations
        if is_given(max_buffer_delay_ms):
            self._opts.max_buffer_delay_ms = max_buffer_delay_ms

    def synthesize(
        self,
        text: str,
        *,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> ChunkedStream:
        return ChunkedStream(tts=self, input_text=text, conn_options=conn_options)

    def stream(
        self,
        *,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> SynthesizeStream:
        return SynthesizeStream(tts=self, conn_options=conn_options)

    def prewarm(self) -> None:
        self._pool.prewarm()

    async def aclose(self) -> None:
        await self._pool.aclose()

Helper class that provides a standard way to create an ABC using inheritance.

Create a new instance of Smallest AI Lightning TTS.

Args

api_key

Your Smallest AI API key.

model

The TTS model to use. Use "lightning_v3.1" for the standard model with 217 voices across 12 languages, or "lightning_v3.1_pro" (default) for the premium pool with curated American, British, and Indian voices at 44.1 kHz.

voice_id

The voice ID to use for synthesis. Defaults to "meher" for "lightning_v3.1_pro" and "sophia" for all other models. Pro voices must be paired with "lightning_v3.1_pro"; standard voices with "lightning_v3.1".

sample_rate

Sample rate for the audio output. Both models are natively 44.1 kHz; supported rates are 8000, 16000, 24000, and 44100.

speed

Speed of the speech synthesis (0.5–2.0).

language

Language of the text to be synthesized. Use "auto" for automatic detection and code-switching. Pro supports "en", "hi", and "auto" only.

output_format

Output format for HTTP synthesize() calls ("pcm", "mp3", "wav", "ulaw", "alaw"). WebSocket streaming always returns PCM.

word_timestamps

Request per-word timing events from the server and emit them as timed transcript entries alongside audio. Applies to WebSocket streaming only; HTTP synthesize() returns raw audio without word events. Disabled by default. Supported on base-queue English + Hindi voices (meher, devansh, kartik, maithili, liam, avery); other voices silently emit no word events.

max_buffer_flush_ms

Server-side buffer bound (milliseconds) for the legacy continue/flush WebSocket protocol (used when use_continuations is False). As text tokens are streamed in with continue: true, the server accumulates them and forces partial audio output once this many milliseconds of text have buffered, without waiting for an explicit flush. 0 (default) disables time-based forced flushing, so audio for a segment is produced when the segment's end-of-input flush is sent. Ignored when use_continuations is True.

use_continuations

Use the context_id-based continuations protocol instead of the legacy continue/flush fields. Continuations let the server release buffered text as soon as it hits a natural sentence boundary (or max_buffer_delay_ms elapses, whichever comes first) instead of holding everything until an explicit flush, and carry prosody across fragments by priming each release with the audio of the previous one. Defaults to True; set False to fall back to the legacy protocol.

max_buffer_delay_ms

Upper bound (0-5000ms) on how long the server holds a buffered fragment waiting for a clean sentence boundary before speaking it anyway. Only applies when use_continuations is True. Defaults to 3000ms (the server default).

base_url

Base URL for the Smallest AI HTTP API.

ws_url

WebSocket URL for low-latency streaming synthesis.

http_session

An existing aiohttp ClientSession to use.

Ancestors

• livekit.agents.tts.tts.TTS
• abc.ABC
• EventEmitter
• typing.Generic

Instance variables

prop model : str

Expand source code

@property
def model(self) -> str:
    return self._opts.model

Get the model name/identifier for this TTS instance.

Returns

The model name if available, "unknown" otherwise.

Note

Plugins should override this property to provide their model information.

prop provider : str

Expand source code

@property
def provider(self) -> str:
    return "SmallestAI"

Get the provider name/identifier for this TTS instance.

Returns

The provider name if available, "unknown" otherwise.

Note

Plugins should override this property to provide their provider information.

Methods

async def aclose(self) ‑> None

Expand source code

async def aclose(self) -> None:
    await self._pool.aclose()

def prewarm(self) ‑> None

Expand source code

def prewarm(self) -> None:
    self._pool.prewarm()

Pre-warm connection to the TTS service

def stream(self,
*,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> livekit.plugins.smallestai.tts.SynthesizeStream

Expand source code

def stream(
    self,
    *,
    conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
) -> SynthesizeStream:
    return SynthesizeStream(tts=self, conn_options=conn_options)

def synthesize(self,
text: str,
*,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> livekit.plugins.smallestai.tts.ChunkedStream

Expand source code

def synthesize(
    self,
    text: str,
    *,
    conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
) -> ChunkedStream:
    return ChunkedStream(tts=self, input_text=text, conn_options=conn_options)

def update_options(self,
*,
model: NotGivenOr[TTSModels | str] = NOT_GIVEN,
voice_id: NotGivenOr[str] = NOT_GIVEN,
speed: NotGivenOr[float] = NOT_GIVEN,
sample_rate: NotGivenOr[int] = NOT_GIVEN,
language: NotGivenOr[str] = NOT_GIVEN,
output_format: NotGivenOr[TTSEncoding | str] = NOT_GIVEN,
word_timestamps: NotGivenOr[bool] = NOT_GIVEN,
max_buffer_flush_ms: NotGivenOr[int] = NOT_GIVEN,
use_continuations: NotGivenOr[bool] = NOT_GIVEN,
max_buffer_delay_ms: NotGivenOr[int] = NOT_GIVEN) ‑> None

Expand source code

def update_options(
    self,
    *,
    model: NotGivenOr[TTSModels | str] = NOT_GIVEN,
    voice_id: NotGivenOr[str] = NOT_GIVEN,
    speed: NotGivenOr[float] = NOT_GIVEN,
    sample_rate: NotGivenOr[int] = NOT_GIVEN,
    language: NotGivenOr[str] = NOT_GIVEN,
    output_format: NotGivenOr[TTSEncoding | str] = NOT_GIVEN,
    word_timestamps: NotGivenOr[bool] = NOT_GIVEN,
    max_buffer_flush_ms: NotGivenOr[int] = NOT_GIVEN,
    use_continuations: NotGivenOr[bool] = NOT_GIVEN,
    max_buffer_delay_ms: NotGivenOr[int] = NOT_GIVEN,
) -> None:
    """Update TTS options."""
    if is_given(model):
        self._opts.model = model
    if is_given(voice_id):
        self._opts.voice_id = voice_id
    if is_given(speed):
        self._opts.speed = speed
    if is_given(sample_rate):
        self._opts.sample_rate = sample_rate
        # keep the base class property in sync; the framework reads tts.sample_rate
        self._sample_rate = sample_rate
    if is_given(language):
        self._opts.language = LanguageCode(language)
    if is_given(output_format):
        self._opts.output_format = output_format
    if is_given(word_timestamps):
        self._opts.word_timestamps = word_timestamps
        self._capabilities.aligned_transcript = word_timestamps
    if is_given(max_buffer_flush_ms):
        self._opts.max_buffer_flush_ms = max_buffer_flush_ms
    if is_given(use_continuations):
        self._opts.use_continuations = use_continuations
    if is_given(max_buffer_delay_ms):
        self._opts.max_buffer_delay_ms = max_buffer_delay_ms

Update TTS options.

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once