Module livekit.agents.inference.stt

Classes

class AssemblyaiOptions (*args, **kwargs)

Expand source code

class AssemblyaiOptions(TypedDict, total=False):
    format_turns: bool  # default: False
    end_of_turn_confidence_threshold: float  # default: 0.01
    min_end_of_turn_silence_when_confident: int  # default: 0
    max_turn_silence: int  # default: not specified
    keyterms_prompt: list[str]  # default: not specified
    language_detection: bool
    inactivity_timeout: float  # seconds
    prompt: str  # default: not specified (u3-rt-pro only, mutually exclusive with keyterms_prompt)
    speaker_labels: bool  # when True, enables speaker diarization (default off)
    agent_context: str  # context to bias recognition (u3-rt-pro only, max 1750 chars)
    previous_context_n_turns: int  # prior turns carried as context; 0 disables (u3-rt-pro only)
    voice_focus: Literal["near-field", "far-field"]  # isolate primary voice (u3-rt-pro only)
    voice_focus_threshold: float  # background suppression strength (u3-rt-pro only)
    mode: Literal["min_latency", "balanced", "max_accuracy"]  # accuracy/latency preset (u3-rt-pro)

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

Ancestors

• builtins.dict

Class variables

var agent_context : str

var end_of_turn_confidence_threshold : float

var format_turns : bool

var inactivity_timeout : float

var keyterms_prompt : list[str]

var language_detection : bool

var max_turn_silence : int

var min_end_of_turn_silence_when_confident : int

var mode : Literal['min_latency', 'balanced', 'max_accuracy']

var previous_context_n_turns : int

var prompt : str

var speaker_labels : bool

var voice_focus : Literal['near-field', 'far-field']

var voice_focus_threshold : float

class CartesiaOptions (*args, **kwargs)

Expand source code

class CartesiaOptions(TypedDict, total=False):
    min_volume: float  # ink-whisper only; default: not specified
    max_silence_duration_secs: float  # ink-whisper only; default: not specified
    # Turn-detection tuning for turn-detecting models (e.g. ink-2). The gateway
    # validates these against Cartesia's documented ranges.
    turn_start_threshold: float  # range 0.5-0.9, default 0.8
    turn_eager_end_threshold: float  # range 0.3-0.6, default 0.4
    turn_end_threshold: float  # range 0.05-0.5, default 0.2
    turn_end_timeout_ms: int  # range 640-11200, default 5600
    keyterm: str | list[str]  # up to 100 terms totaling 1200 characters

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

Ancestors

• builtins.dict

Class variables

var keyterm : str | list[str]

var max_silence_duration_secs : float

var min_volume : float

var turn_eager_end_threshold : float

var turn_end_threshold : float

var turn_end_timeout_ms : int

var turn_start_threshold : float

class DeepgramFluxOptions (*args, **kwargs)

Expand source code

class DeepgramFluxOptions(TypedDict, total=False):
    eager_eot_threshold: float  # range 0.3-0.9, default: 0.5
    eot_threshold: float  # range 0.5-0.9
    eot_timeout_ms: int
    keyterm: str | list[str]
    mip_opt_out: bool  # always True for LiveKit Inference; the passed value is ignored
    tag: str | list[str]
    detect_language: bool
    # flux-general-multi only: biases toward these languages without restricting
    # detection; takes precedence over `language`
    language_hint: str | list[str]

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

Ancestors

• builtins.dict

Class variables

var detect_language : bool

var eager_eot_threshold : float

var eot_threshold : float

var eot_timeout_ms : int

var keyterm : str | list[str]

var language_hint : str | list[str]

var mip_opt_out : bool

var tag : str | list[str]

class DeepgramOptions (*args, **kwargs)

Expand source code

class DeepgramOptions(TypedDict, total=False):
    filler_words: bool  # default: True
    interim_results: bool  # default: True
    endpointing: int  # default: 25 (ms)
    punctuate: bool  # default: True
    smart_format: bool
    keywords: list[tuple[str, float]]
    keyterm: str | list[str]
    profanity_filter: bool
    numerals: bool
    mip_opt_out: bool  # always True for LiveKit Inference; the passed value is ignored
    vad_events: bool  # default: False
    diarize: bool  # when True, enables speaker diarization (default off)
    dictation: bool
    detect_language: bool
    no_delay: bool  # default: True
    utterance_end: bool
    redact: str | list[str]
    replace: str | list[str]
    search: str | list[str]
    tag: str | list[str]
    channels: int
    version: str
    callback: str
    callback_method: str
    extra: str

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

Ancestors

• builtins.dict

Class variables

var callback : str

var callback_method : str

var channels : int

var detect_language : bool

var diarize : bool

var dictation : bool

var endpointing : int

var extra : str

var filler_words : bool

var interim_results : bool

var keyterm : str | list[str]

var keywords : list[tuple[str, float]]

var mip_opt_out : bool

var no_delay : bool

var numerals : bool

var profanity_filter : bool

var punctuate : bool

var redact : str | list[str]

var replace : str | list[str]

var search : str | list[str]

var smart_format : bool

var tag : str | list[str]

var utterance_end : bool

var vad_events : bool

var version : str

class FallbackModel (*args, **kwargs)

Expand source code

class FallbackModel(TypedDict, total=False):
    """Inference Fallback Adapter: configuration for a fallback STT model that runs server-side in LiveKit Inference, providing automatic fallback between providers.

    Extra fields are passed through to the provider.

    Example:
        >>> FallbackModel(model="deepgram/nova-3", extra_kwargs={"keyterm": ["livekit"]})
    """

    model: Required[str]
    """Model name (e.g. "deepgram/nova-3", "assemblyai/universal-streaming", "cartesia/ink-whisper")."""

    extra_kwargs: dict[str, Any]
    """Extra configuration for the model."""

Inference Fallback Adapter: configuration for a fallback STT model that runs server-side in LiveKit Inference, providing automatic fallback between providers.

Extra fields are passed through to the provider.

Example

>>> FallbackModel(model="deepgram/nova-3", extra_kwargs={"keyterm": ["livekit"]})

Ancestors

• builtins.dict

Class variables

var extra_kwargs : dict[str, typing.Any]

Extra configuration for the model.

var model : str

Model name (e.g. "deepgram/nova-3", "assemblyai/universal-streaming", "cartesia/ink-whisper").

class GoogleOptions (*args, **kwargs)

Expand source code

class GoogleOptions(TypedDict, total=False):
    # Mirrors the Live API's AudioTranscriptionConfig. Omit language_codes, or pass an
    # empty list, to let the model detect the language itself.
    # https://ai.google.dev/gemini-api/docs/live-api/live-transcribe
    language_codes: list[str]  # BCP-47 codes, e.g. ["en-US", "es-ES"]
    custom_vocabulary: list[str]  # up to 1000 terms that bias recognition

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

Ancestors

• builtins.dict

Class variables

var custom_vocabulary : list[str]

var language_codes : list[str]

class InworldOptions (*args, **kwargs)

Expand source code

class InworldOptions(TypedDict, total=False):
    enable_voice_profile: bool  # default: True
    voice_profile_top_n: int  # range 1-20, default 10
    include_word_timestamps: bool  # default: True
    audio_encoding: Literal["LINEAR16", "AUTO_DETECT"]  # default: LINEAR16
    inactivity_timeout_seconds: int  # >= 0; 0 disables
    end_of_turn_confidence_threshold: float  # range 0.0-1.0, default 0.5
    min_end_of_turn_silence_when_confident: int  # >= 0 (ms)
    prompts: list[str]
    vad_threshold: float  # range 0.0-1.0, default 0.5

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

Ancestors

• builtins.dict

Class variables

var audio_encoding : Literal['LINEAR16', 'AUTO_DETECT']

var enable_voice_profile : bool

var end_of_turn_confidence_threshold : float

var inactivity_timeout_seconds : int

var include_word_timestamps : bool

var min_end_of_turn_silence_when_confident : int

var prompts : list[str]

var vad_threshold : float

var voice_profile_top_n : int

class OpenAIOptions (*args, **kwargs)

Expand source code

class OpenAIOptions(TypedDict, total=False):
    # Mirrors the transcription config of an OpenAI Realtime transcription session.
    # https://developers.openai.com/api/docs/guides/realtime-transcription
    languages: list[str]  # expected input languages; overrides `language`
    prompt: str  # describes the recording or its setting, up to 1024 characters
    keywords: list[str]  # literal terms the audio may contain
    delay: Literal["minimal", "low", "medium", "high", "xhigh"]  # latency vs. accuracy
    noise_reduction: Literal["near_field", "far_field"]

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

Ancestors

• builtins.dict

Class variables

var delay : Literal['minimal', 'low', 'medium', 'high', 'xhigh']

var keywords : list[str]

var languages : list[str]

var noise_reduction : Literal['near_field', 'far_field']

var prompt : str

class STT (model: NotGivenOr[STTModels | str] = NOT_GIVEN,
*,
language: NotGivenOr[str] = NOT_GIVEN,
base_url: NotGivenOr[str] = NOT_GIVEN,
encoding: NotGivenOr[STTEncoding] = NOT_GIVEN,
sample_rate: NotGivenOr[int] = NOT_GIVEN,
api_key: NotGivenOr[str] = NOT_GIVEN,
api_secret: NotGivenOr[str] = NOT_GIVEN,
http_session: aiohttp.ClientSession | None = None,
extra_kwargs: NotGivenOr[dict[str, Any] | CartesiaOptions | DeepgramOptions | DeepgramFluxOptions | AssemblyaiOptions | XaiOptions | SpeechmaticsOptions | InworldOptions | GoogleOptions | OpenAIOptions] = NOT_GIVEN,
fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN)

Expand source code

class STT(stt.STT):
    @overload
    def __init__(
        self,
        model: CartesiaModels,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[CartesiaOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None: ...

    @overload
    def __init__(
        self,
        model: DeepgramModels,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[DeepgramOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None: ...

    @overload
    def __init__(
        self,
        model: DeepgramFluxModels,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[DeepgramFluxOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None: ...

    @overload
    def __init__(
        self,
        model: AssemblyAIModels,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[AssemblyaiOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None: ...

    @overload
    def __init__(
        self,
        model: XaiModels,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[XaiOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None: ...

    @overload
    def __init__(
        self,
        model: SpeechmaticsModels,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[SpeechmaticsOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
        vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN,
    ) -> None: ...

    @overload
    def __init__(
        self,
        model: InworldModels,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[InworldOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None: ...

    @overload
    def __init__(
        self,
        model: GoogleModels,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[GoogleOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None: ...

    @overload
    def __init__(
        self,
        model: OpenAIModels,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[OpenAIOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
        vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN,
    ) -> None: ...

    @overload
    def __init__(
        self,
        model: str,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None: ...

    def __init__(
        self,
        model: NotGivenOr[STTModels | str] = NOT_GIVEN,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[
            dict[str, Any]
            | CartesiaOptions
            | DeepgramOptions
            | DeepgramFluxOptions
            | AssemblyaiOptions
            | XaiOptions
            | SpeechmaticsOptions
            | InworldOptions
            | GoogleOptions
            | OpenAIOptions
        ] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
        vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN,
    ) -> None:
        """Livekit Cloud Inference STT

        Args:
            model (STTModels | str, optional): STT model to use, in "provider/model[:language]" format.
            language (str, optional): Language of the STT model.
            encoding (STTEncoding, optional): Encoding of the STT model.
            sample_rate (int, optional): Sample rate of the STT model. Defaults to 24000 for
                OpenAI models, which accept no other rate, and to 16000 otherwise.
            base_url (str, optional): LIVEKIT_URL, if not provided, read from environment variable.
            api_key (str, optional): LIVEKIT_API_KEY, if not provided, read from environment variable.
            api_secret (str, optional): LIVEKIT_API_SECRET, if not provided, read from environment variable.
            http_session (aiohttp.ClientSession, optional): HTTP session to use.
            extra_kwargs (dict, optional): Extra kwargs to pass to the STT model.
            fallback (FallbackModelType, optional): Fallback models - either a list of model names,
                a list of FallbackModel instances.
            conn_options (APIConnectOptions, optional): Connection options for request attempts.
            vad (VAD, optional): External Voice Activity Detector. When provided, each audio
                frame is forwarded to the VAD and `session.finalize` is sent to the inference
                gateway on end of speech. Only applicable to models that detect no turns
                themselves: the Speechmatics RT models (enhanced/standard) and OpenAI
                gpt-live-transcribe. They get a default VAD when none is passed.
        """
        # Infer diarization capability from provider-specific extra_kwargs
        # keys (see _DIARIZATION_EXTRA_KEYS). xAI uses "diarize" (same as
        # Deepgram); AssemblyAI uses "speaker_labels".
        diarization_enabled = _diarization_enabled(
            dict(extra_kwargs) if is_given(extra_kwargs) else None
        )

        # Parse language from model string if provided: "provider/model:language"
        if is_given(model) and isinstance(model, str):
            parsed_model, parsed_language = _parse_model_string(model)
            model = parsed_model
            if is_given(parsed_language) and not is_given(language):
                language = parsed_language

        vad = _resolve_vad_for_model(model, vad if is_given(vad) else None)
        resolved_sample_rate = (
            sample_rate
            if is_given(sample_rate)
            else (_required_sample_rate(model) or DEFAULT_SAMPLE_RATE)
        )
        _warn_if_sample_rate_refused(model, resolved_sample_rate)

        fallback_models: NotGivenOr[list[FallbackModel]] = NOT_GIVEN
        if is_given(fallback):
            fallback_models = _normalize_fallback(fallback)
        models = [model]
        if is_given(fallback_models):
            models.extend(item["model"] for item in fallback_models)
        aligned_transcript: Literal["word", False] = (
            "word" if all(_aligned_transcript_for_model(item) for item in models) else False
        )

        # chat_context follows the model's native support; the session decides whether to forward
        super().__init__(
            capabilities=stt.STTCapabilities(
                streaming=True,
                interim_results=True,
                diarization=diarization_enabled,
                aligned_transcript=aligned_transcript,
                offline_recognize=False,
                keyterms=_keyterms_extra_for_model(model) is not None,
                chat_context=_supports_agent_context_carryover(model),
            ),
        )

        lk_base_url = base_url if is_given(base_url) else get_default_inference_url()

        lk_api_key = (
            api_key
            if is_given(api_key)
            else os.getenv("LIVEKIT_INFERENCE_API_KEY", os.getenv("LIVEKIT_API_KEY", ""))
        )
        if not lk_api_key:
            raise ValueError(
                "api_key is required, either as argument or set LIVEKIT_API_KEY environmental variable"
            )

        lk_api_secret = (
            api_secret
            if is_given(api_secret)
            else os.getenv("LIVEKIT_INFERENCE_API_SECRET", os.getenv("LIVEKIT_API_SECRET", ""))
        )
        if not lk_api_secret:
            raise ValueError(
                "api_secret is required, either as argument or set LIVEKIT_API_SECRET environmental variable"
            )

        self._opts = STTOptions(
            model=model,
            language=LanguageCode(language) if isinstance(language, str) else language,
            encoding=encoding if is_given(encoding) else DEFAULT_ENCODING,
            sample_rate=resolved_sample_rate,
            base_url=lk_base_url,
            api_key=lk_api_key,
            api_secret=lk_api_secret,
            extra_kwargs=dict(extra_kwargs) if is_given(extra_kwargs) else {},
            fallback=fallback_models,
            conn_options=conn_options if is_given(conn_options) else DEFAULT_API_CONNECT_OPTIONS,
        )

        self._session = http_session
        self._vad = vad
        # a default sample rate follows the model; an explicit one is the caller's choice
        self._explicit_sample_rate = is_given(sample_rate)
        self._session_keyterms: list[str] = []  # framework-managed; merged into extra_kwargs
        self._streams = weakref.WeakSet[SpeechStream]()

    @classmethod
    def from_model_string(cls, model: str) -> STT:
        """Create a STT instance from a model string

        Args:
            model (str): STT model to use, in "provider/model[:language]" format

        Returns:
            STT: STT instance
        """
        model_name, language = _parse_model_string(model)
        return cls(model=model_name, language=language)

    @property
    def model(self) -> str:
        return self._opts.model if is_given(self._opts.model) else "unknown"

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

    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: utils.AudioBuffer,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        conn_options: APIConnectOptions,
    ) -> stt.SpeechEvent:
        raise NotImplementedError(
            "LiveKit Inference STT does not support batch recognition, use stream() instead"
        )

    def stream(
        self,
        *,
        language: NotGivenOr[STTLanguages | str] = NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> SpeechStream:
        """Create a streaming transcription session."""
        options = self._sanitize_options(language=language)
        stream = SpeechStream(
            stt=self,
            opts=options,
            conn_options=conn_options,
            vad_instance=self._vad,
        )
        self._streams.add(stream)
        return stream

    def update_options(
        self,
        *,
        model: NotGivenOr[STTModels | str] = NOT_GIVEN,
        language: NotGivenOr[STTLanguages | str] = NOT_GIVEN,
        extra: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
    ) -> None:
        """Update STT configuration options."""
        # a ":language" suffix belongs to the model that carries it, so it is applied to
        # this STT but not forced onto a live stream that keeps an earlier model
        explicit_language = language
        if is_given(model):
            # Mirror __init__: strip ":language" suffix and apply if not overridden.
            if isinstance(model, str):
                parsed_model, parsed_language = _parse_model_string(model)
                model = parsed_model
                if is_given(parsed_language) and not is_given(language):
                    language = parsed_language

            self._opts.model = model
            self._vad = _resolve_vad_for_model(model, self._vad)
            if not self._explicit_sample_rate:
                # the default rate follows the model; a live stream's own rate stays
                # fixed at its creation, but streams created from now on must work
                self._opts.sample_rate = _required_sample_rate(model) or DEFAULT_SAMPLE_RATE
            _warn_if_sample_rate_refused(model, self._opts.sample_rate)
            models = [self._opts.model]
            if is_given(self._opts.fallback):
                models.extend(item["model"] for item in self._opts.fallback)
            self._capabilities = replace(
                self._capabilities,
                keyterms=_keyterms_extra_for_model(self._opts.model) is not None,
                chat_context=_supports_agent_context_carryover(self._opts.model),
                aligned_transcript=(
                    "word" if all(_aligned_transcript_for_model(item) for item in models) else False
                ),
            )
        if is_given(language):
            self._opts.language = LanguageCode(language)
        if is_given(extra):
            # keep this a pure user baseline: framework session keyterms are merged in
            # per stream at forward time, so a later clear can actually remove them
            self._opts.extra_kwargs.update(extra)
            self._capabilities = replace(
                self._capabilities,
                diarization=_diarization_enabled(self._opts.extra_kwargs),
            )

        for stream in self._streams:
            stream.update_options(model=model, language=explicit_language, extra=extra)

    def _update_session_keyterms(self, keyterms: list[str]) -> None:
        if keyterms == self._session_keyterms:
            return
        keyterm_extra = _keyterms_extra_for_model(
            self._opts.model, extra_kwargs=self._opts.extra_kwargs, session_keyterms=keyterms
        )
        if keyterm_extra is None:
            super()._update_session_keyterms(keyterms)  # warn-and-skip for unsupported models
            return

        self._session_keyterms = list(keyterms)
        # Each stream formats the terms for the model it is actually running (a refused
        # switch can leave it on a different provider than the parent). inference applies
        # extra live via session.update, so a speaking stream defers to END_OF_SPEECH
        # because the gateway may reconnect upstream when the keyterms change.
        for stream in self._streams:
            stream._on_session_keyterms_changed()

    def _push_conversation_item(self, ev: ConversationItemAddedEvent) -> None:
        if (
            (chat_item := ev.item).type == "message"
            and chat_item.role == "assistant"
            and (text := chat_item.text_content)
        ):
            if len(text) > _ASSEMBLYAI_MAX_AGENT_CONTEXT_CHARS:
                logger.debug(
                    "truncating agent_context carryover from %d to %d chars",
                    len(text),
                    _ASSEMBLYAI_MAX_AGENT_CONTEXT_CHARS,
                )
                text = text[-_ASSEMBLYAI_MAX_AGENT_CONTEXT_CHARS:]
            self.update_options(extra={"agent_context": text})

    def _sanitize_options(
        self, *, language: NotGivenOr[STTLanguages | str] = NOT_GIVEN
    ) -> STTOptions:
        """Create a sanitized copy of options with language override if provided."""
        options = replace(self._opts)
        options.extra_kwargs = dict(options.extra_kwargs)

        if is_given(language):
            options.language = LanguageCode(language)

        return options

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

Livekit Cloud Inference STT

Args

model : STTModels | str, optional

STT model to use, in "provider/model[:language]" format.

language : str, optional

Language of the STT model.

encoding : STTEncoding, optional

Encoding of the STT model.

sample_rate : int, optional

Sample rate of the STT model. Defaults to 24000 for OpenAI models, which accept no other rate, and to 16000 otherwise.

base_url : str, optional

LIVEKIT_URL, if not provided, read from environment variable.

api_key : str, optional

LIVEKIT_API_KEY, if not provided, read from environment variable.

api_secret : str, optional

LIVEKIT_API_SECRET, if not provided, read from environment variable.

http_session : aiohttp.ClientSession, optional

HTTP session to use.

extra_kwargs : dict, optional

Extra kwargs to pass to the STT model.

fallback : FallbackModelType, optional

Fallback models - either a list of model names, a list of FallbackModel instances.

conn_options : APIConnectOptions, optional

Connection options for request attempts.

vad : VAD, optional

External Voice Activity Detector. When provided, each audio frame is forwarded to the VAD and session.finalize is sent to the inference gateway on end of speech. Only applicable to models that detect no turns themselves: the Speechmatics RT models (enhanced/standard) and OpenAI gpt-live-transcribe. They get a default VAD when none is passed.

Ancestors

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

Static methods

def from_model_string(model: str) ‑> STT

Create a STT instance from a model string

Args

model : str

STT model to use, in "provider/model[:language]" format

Returns

STT

STT instance

Instance variables

prop model : str

Expand source code

@property
def model(self) -> str:
    return self._opts.model if is_given(self._opts.model) else "unknown"

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 "livekit"

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[STTLanguages | str] = NOT_GIVEN,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> SpeechStream

Expand source code

def stream(
    self,
    *,
    language: NotGivenOr[STTLanguages | str] = NOT_GIVEN,
    conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
) -> SpeechStream:
    """Create a streaming transcription session."""
    options = self._sanitize_options(language=language)
    stream = SpeechStream(
        stt=self,
        opts=options,
        conn_options=conn_options,
        vad_instance=self._vad,
    )
    self._streams.add(stream)
    return stream

Create a streaming transcription session.

def update_options(self,
*,
model: NotGivenOr[STTModels | str] = NOT_GIVEN,
language: NotGivenOr[STTLanguages | str] = NOT_GIVEN,
extra: NotGivenOr[dict[str, Any]] = NOT_GIVEN) ‑> None

Expand source code

def update_options(
    self,
    *,
    model: NotGivenOr[STTModels | str] = NOT_GIVEN,
    language: NotGivenOr[STTLanguages | str] = NOT_GIVEN,
    extra: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
) -> None:
    """Update STT configuration options."""
    # a ":language" suffix belongs to the model that carries it, so it is applied to
    # this STT but not forced onto a live stream that keeps an earlier model
    explicit_language = language
    if is_given(model):
        # Mirror __init__: strip ":language" suffix and apply if not overridden.
        if isinstance(model, str):
            parsed_model, parsed_language = _parse_model_string(model)
            model = parsed_model
            if is_given(parsed_language) and not is_given(language):
                language = parsed_language

        self._opts.model = model
        self._vad = _resolve_vad_for_model(model, self._vad)
        if not self._explicit_sample_rate:
            # the default rate follows the model; a live stream's own rate stays
            # fixed at its creation, but streams created from now on must work
            self._opts.sample_rate = _required_sample_rate(model) or DEFAULT_SAMPLE_RATE
        _warn_if_sample_rate_refused(model, self._opts.sample_rate)
        models = [self._opts.model]
        if is_given(self._opts.fallback):
            models.extend(item["model"] for item in self._opts.fallback)
        self._capabilities = replace(
            self._capabilities,
            keyterms=_keyterms_extra_for_model(self._opts.model) is not None,
            chat_context=_supports_agent_context_carryover(self._opts.model),
            aligned_transcript=(
                "word" if all(_aligned_transcript_for_model(item) for item in models) else False
            ),
        )
    if is_given(language):
        self._opts.language = LanguageCode(language)
    if is_given(extra):
        # keep this a pure user baseline: framework session keyterms are merged in
        # per stream at forward time, so a later clear can actually remove them
        self._opts.extra_kwargs.update(extra)
        self._capabilities = replace(
            self._capabilities,
            diarization=_diarization_enabled(self._opts.extra_kwargs),
        )

    for stream in self._streams:
        stream.update_options(model=model, language=explicit_language, extra=extra)

Update STT configuration options.

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once

class STTOptions (model: NotGivenOr[STTModels | str],
language: NotGivenOr[LanguageCode],
encoding: STTEncoding,
sample_rate: int,
base_url: str,
api_key: str,
api_secret: str,
extra_kwargs: dict[str, Any],
fallback: NotGivenOr[list[FallbackModel]],
conn_options: NotGivenOr[APIConnectOptions])

Expand source code

@dataclass
class STTOptions:
    model: NotGivenOr[STTModels | str]
    language: NotGivenOr[LanguageCode]
    encoding: STTEncoding
    sample_rate: int
    base_url: str
    api_key: str
    api_secret: str
    extra_kwargs: dict[str, Any]
    fallback: NotGivenOr[list[FallbackModel]]
    conn_options: NotGivenOr[APIConnectOptions]

STTOptions(model: 'NotGivenOr[STTModels | str]', language: 'NotGivenOr[LanguageCode]', encoding: 'STTEncoding', sample_rate: 'int', base_url: 'str', api_key: 'str', api_secret: 'str', extra_kwargs: 'dict[str, Any]', fallback: 'NotGivenOr[list[FallbackModel]]', conn_options: 'NotGivenOr[APIConnectOptions]')

Instance variables

var api_key : str

var api_secret : str

var base_url : str

var conn_options : livekit.agents.types.APIConnectOptions | livekit.agents.types.NotGiven

var encoding : Literal['pcm_s16le']

var extra_kwargs : dict[str, typing.Any]

var fallback : list[FallbackModel] | livekit.agents.types.NotGiven

var language : livekit.agents.language.LanguageCode | livekit.agents.types.NotGiven

var model : Literal['deepgram/nova-3', 'deepgram/nova-3-medical', 'deepgram/nova-2', 'deepgram/nova-2-medical', 'deepgram/nova-2-conversationalai', 'deepgram/nova-2-phonecall'] | Literal['deepgram/flux-general', 'deepgram/flux-general-en', 'deepgram/flux-general-multi'] | Literal['cartesia/ink-whisper', 'cartesia/ink-2'] | Literal['assemblyai/universal-streaming', 'assemblyai/universal-streaming-multilingual', 'assemblyai/u3-rt-pro', 'assemblyai/universal-3-5-pro', 'assemblyai/universal-3-6-pro'] | Literal['xai/stt-1'] | Literal['speechmatics/enhanced', 'speechmatics/standard', 'speechmatics/linden-1'] | Literal['inworld/inworld-stt-1'] | Literal['google/gemini-3.5-transcribe-live'] | Literal['openai/gpt-live-transcribe'] | Literal['auto'] | str | livekit.agents.types.NotGiven

var sample_rate : int

class SpeechStream (*,
stt: STT,
opts: STTOptions,
conn_options: APIConnectOptions,
vad_instance: vad.VAD | None = None)

Expand source code

class SpeechStream(stt.SpeechStream):
    def __init__(
        self,
        *,
        stt: STT,
        opts: STTOptions,
        conn_options: APIConnectOptions,
        vad_instance: vad.VAD | None = None,
    ) -> None:
        super().__init__(stt=stt, conn_options=conn_options, sample_rate=opts.sample_rate)
        self._stt: STT = stt
        self._opts = opts
        self._request_id = str(utils.shortuuid("stt_request_"))

        self._speaking = False
        # keyterm extra set while the user is speaking; applied at END_OF_SPEECH (latest wins).
        # inference applies live, but the gateway may reconnect upstream, so defer to a calm moment.
        self._pending_extra: dict[str, Any] | None = None
        self._speech_duration: float = 0
        self._ws: aiohttp.ClientWebSocketResponse | None = None
        self._vad: vad.VAD | None = vad_instance
        self._session_update_tasks: set[asyncio.Task[None]] = set()

    def update_options(
        self,
        *,
        model: NotGivenOr[STTModels | str] = NOT_GIVEN,
        language: NotGivenOr[STTLanguages | str] = NOT_GIVEN,
        extra: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
    ) -> None:
        """Update streaming transcription options.

        When the WebSocket is live, a mid-stream session.update is sent so providers
        that support it (e.g. AssemblyAI, Deepgram Flux) can apply changes without
        reconnecting. Unsupported providers ignore the message.

        A live session's provider, audio rate, and client endpointing are fixed when it is
        created: the gateway does not switch them through session.update. This stream
        therefore keeps the model it was created with, so its VAD (and the `session.finalize`
        it sends) still matches the running model, and framework keyterms are formatted for
        that model. Streams created after a parent switch start on the new model. The
        requested model is still forwarded, as the other session settings are, but a
        language or extra written for a different provider is left to those new streams.
        """
        changed_provider = False
        if is_given(model):
            refused = self._warn_if_rate_refused(model)
            # Mirror __init__/STT.update_options: strip ":language" suffix. It belongs to the
            # model carrying it, so it is not applied when this stream cannot run that model
            # or when the model belongs to a provider the live session stays off.
            parsed_language: NotGivenOr[LanguageCode] = NOT_GIVEN
            if isinstance(model, str):
                parsed_model, parsed_language = _parse_model_string(model)
                model = parsed_model
            changed_provider = _provider_for_model(model) != _provider_for_model(self._opts.model)
            if (
                is_given(parsed_language)
                and not is_given(language)
                and not refused
                and not changed_provider
            ):
                language = parsed_language

        if is_given(language):
            self._opts.language = LanguageCode(language)

        forward_extra: NotGivenOr[dict[str, Any]] = NOT_GIVEN
        if is_given(extra) and not changed_provider:
            # a pure user baseline: framework keyterms are merged in at forward time, so a
            # later clear can actually remove them. An extra written alongside another
            # provider's model is for that model, not for the one this session is running.
            self._opts.extra_kwargs.update(extra)
            forward_extra = self._forward_extra(extra)

        if not (is_given(model) or is_given(language) or is_given(forward_extra)):
            return
        self._send_live_update(model=model, language=language, extra=forward_extra)

    def _warn_if_rate_refused(self, model: STTModels | str) -> bool:
        """A live session's rate is fixed at creation, so a model needing another leaves
        the session on the model it is running and only future streams get the switch.
        Returns whether this stream refused the model."""
        required = _required_sample_rate(model)
        if required is not None and required != self._opts.sample_rate:
            logger.warning(
                "live stream stays on %r: %r takes audio only at %d Hz, this stream was "
                "created at %d Hz",
                self._opts.model,
                model,
                required,
                self._opts.sample_rate,
            )
            return True
        return False

    def _forward_extra(self, extra: dict[str, Any]) -> NotGivenOr[dict[str, Any]]:
        """The user's extra delta, with the running model's framework keyterms folded in when
        the delta sets that model's own keyterm key, so the merge cannot shorten them. Any
        payload deferred to end of speech is left alone so the change is not applied mid-turn."""
        keyterm_key = _keyterms_key_for_model(self._opts.model)
        if keyterm_key is not None and keyterm_key in extra:
            # the user edited this session's keyterms explicitly: send the whole value and
            # supersede anything deferred to end of speech
            self._pending_extra = None
            return {**extra, **self._keyterm_payload()}
        return dict(extra)

    def _keyterm_payload(self) -> dict[str, Any]:
        """This stream's keyterms under its own running model's key, formatted fresh."""
        return (
            _keyterms_extra_for_model(
                self._opts.model,
                extra_kwargs=self._opts.extra_kwargs,
                session_keyterms=self._stt._session_keyterms,
            )
            or {}
        )

    def _send_live_update(
        self,
        *,
        model: NotGivenOr[STTModels | str] = NOT_GIVEN,
        language: NotGivenOr[STTLanguages | str] = NOT_GIVEN,
        extra: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
    ) -> None:
        """Send a sparse session.update for the current session, if one is live."""
        if self._ws is None or self._ws.closed:
            return
        settings: dict[str, Any] = {}
        if is_given(model):
            settings["model"] = model
        if is_given(language):
            settings["language"] = str(LanguageCode(language))
        if is_given(extra):
            settings["extra"] = extra
        if not settings:
            return
        update_msg = {
            "type": "session.update",
            "settings": settings,
        }
        # Hold the task: the loop only weakly references it, and a collected one
        # means self._opts moved on while the server was never told.
        task = asyncio.create_task(self._send_session_update(update_msg))
        self._session_update_tasks.add(task)
        task.add_done_callback(self._session_update_tasks.discard)

    def _on_session_keyterms_changed(self) -> None:
        """Re-send this stream's keyterms under its own model's key, formatted fresh so a
        cleared list is actually removed from the live session."""
        extra = self._keyterm_payload()
        if not extra:
            return
        # inference applies extra live via session.update; defer to END_OF_SPEECH since
        # the gateway may reconnect upstream when the keyterms change
        if self._speaking:
            self._pending_extra = extra
        else:
            self._send_live_update(extra=extra)

    def _on_end_of_speech(self) -> None:
        if self._pending_extra is not None:
            # send directly: the deferred payload is already formatted and must not be
            # stored as user baseline, or a later clear could not remove it
            self._send_live_update(extra=self._pending_extra)
            self._pending_extra = None

    async def _send_session_update(self, msg: dict[str, Any]) -> None:
        try:
            if self._ws is not None and not self._ws.closed:
                await self._ws.send_str(json.dumps(msg))
        except Exception:
            logger.debug("failed to send session.update, ws may be closing")

    @utils.log_exceptions(logger=logger)
    async def _vad_finalize_task(
        self, ws: aiohttp.ClientWebSocketResponse, stream: vad.VADStream
    ) -> None:
        async for ev in stream:
            if ev.type != vad.VADEventType.END_OF_SPEECH:
                continue
            if ws.closed:
                return
            try:
                await ws.send_str(json.dumps({"type": "session.finalize"}))
            except Exception:
                logger.debug("failed to send session.finalize from VAD, ws may be closing")
                return

    async def _run(self) -> None:
        """Main loop for streaming transcription."""
        input_ended = asyncio.Event()
        transcript_timeout = utils.aio.sleep(FINALIZATION_TIMEOUT)
        session_closed = False
        http_session = self._stt._ensure_session()
        # one VAD stream per session: the session's model (and so whether it needs
        # client endpointing) is fixed for its lifetime
        vad_stream: vad.VADStream | None = self._vad.stream() if self._vad is not None else None

        async def wait_for_inactivity() -> None:
            await input_ended.wait()
            await transcript_timeout

        @utils.log_exceptions(logger=logger)
        async def send_task(ws: aiohttp.ClientWebSocketResponse) -> None:
            audio_bstream = utils.audio.AudioByteStream(
                sample_rate=self._opts.sample_rate,
                num_channels=1,
                samples_per_channel=self._opts.sample_rate // 20,  # 50ms
            )

            try:
                async for ev in self._input_ch:
                    frames: list[rtc.AudioFrame] = []
                    if isinstance(ev, rtc.AudioFrame):
                        if vad_stream is not None:
                            vad_stream.push_frame(ev)
                        frames.extend(audio_bstream.push(ev.data))
                    elif isinstance(ev, self._FlushSentinel):
                        frames.extend(audio_bstream.flush())

                    for frame in frames:
                        self._speech_duration += frame.duration
                        audio_bytes = frame.data.tobytes()
                        base64_audio = base64.b64encode(audio_bytes).decode("utf-8")
                        audio_msg = {
                            "type": "input_audio",
                            "audio": base64_audio,
                        }
                        await ws.send_str(json.dumps(audio_msg))

                if vad_stream is not None:
                    # the VAD may have ended on its own already (e.g. it closed its stream
                    # early); that must not stop session.finalize from going out
                    with contextlib.suppress(RuntimeError):
                        vad_stream.end_input()

                input_ended.set()
                finalize_msg = {
                    "type": "session.finalize",
                }
                await ws.send_str(json.dumps(finalize_msg))
            except (aiohttp.ClientError, ConnectionError) as e:
                if input_ended.is_set() or http_session.closed:
                    return
                raise APIConnectionError(
                    "LiveKit Inference STT connection closed unexpectedly"
                ) from e

        @utils.log_exceptions(logger=logger)
        async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
            nonlocal session_closed
            while True:
                msg = await ws.receive()
                if msg.type in (
                    aiohttp.WSMsgType.CLOSED,
                    aiohttp.WSMsgType.CLOSE,
                    aiohttp.WSMsgType.CLOSING,
                ):
                    if input_ended.is_set() or http_session.closed:
                        return
                    raise APIStatusError(
                        message="LiveKit Inference STT connection closed unexpectedly",
                        status_code=ws.close_code or -1,
                        retryable=True,
                    )

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

                data = json.loads(msg.data)
                msg_type = data.get("type")
                if msg_type in ("interim_transcript", "preflight_transcript", "final_transcript"):
                    # Remember finals received before input ends; only the drain starts the timer.
                    with contextlib.suppress(utils.aio.SleepFinished):
                        transcript_timeout.reset(
                            FINAL_TRANSCRIPT_INACTIVITY_TIMEOUT
                            if msg_type == "final_transcript"
                            else None
                        )
                if msg_type == "session.created":
                    pass
                elif msg_type == "start_of_speech":
                    self._process_start_of_speech()
                elif msg_type == "interim_transcript":
                    self._process_transcript(data, is_final=False)
                elif msg_type == "preflight_transcript":
                    self._process_preflight_transcript(data)
                elif msg_type == "final_transcript":
                    self._process_transcript(data, is_final=True)
                elif msg_type == "session.finalized":
                    # Acknowledgments are optional and may precede trailing transcripts.
                    pass
                elif msg_type == "session.closed":
                    session_closed = True
                    if not input_ended.is_set():
                        raise APIStatusError(
                            message="LiveKit Inference STT session closed before input ended",
                            status_code=-1,
                            retryable=True,
                        )
                    return
                elif msg_type == "error":
                    logger.error(
                        "received error from LiveKit Inference STT",
                        extra={"lk.pii.event": data},
                    )
                    code = data.get("code", -1)
                    raise APIError(
                        "LiveKit Inference STT returned an error",
                        body={"code": code},
                        retryable=not input_ended.is_set(),
                    )

        ws: aiohttp.ClientWebSocketResponse | None = None
        try:
            ws = await self._connect_ws(http_session)
            self._ws = ws
            receiver = asyncio.create_task(recv_task(ws))
            inactivity = asyncio.create_task(wait_for_inactivity())
            tasks = [
                asyncio.create_task(send_task(ws)),
                receiver,
                inactivity,
            ]
            if vad_stream is not None:
                tasks.append(asyncio.create_task(self._vad_finalize_task(ws, vad_stream)))
            try:
                pending = set(tasks)
                while True:
                    done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
                    for task in done:
                        task.result()
                    if receiver in done or inactivity in done:
                        break
            finally:
                transcript_timeout.cancel()
                await utils.aio.gracefully_cancel(*tasks)
        finally:
            self._ws = None
            if self._session_update_tasks:
                await utils.aio.gracefully_cancel(*self._session_update_tasks)
                self._session_update_tasks.clear()
            if ws is not None:
                try:
                    if not session_closed and not ws.closed:
                        await ws.send_str(json.dumps({"type": "session.close"}))
                except Exception:
                    logger.debug("failed to send session.close, ws may be closing")
                await ws.close()
            if vad_stream is not None:
                await vad_stream.aclose()

    async def _connect_ws(
        self, http_session: aiohttp.ClientSession
    ) -> aiohttp.ClientWebSocketResponse:
        """Connect to the LiveKit Inference STT WebSocket."""
        params: dict[str, Any] = {
            "settings": {
                "sample_rate": str(self._opts.sample_rate),
                "encoding": self._opts.encoding,
                # merge the framework session keyterms into the user's extra_kwargs keyterm key
                "extra": {
                    **self._opts.extra_kwargs,
                    **(
                        _keyterms_extra_for_model(
                            self._opts.model,
                            extra_kwargs=self._opts.extra_kwargs,
                            session_keyterms=self._stt._session_keyterms,
                        )
                        or {}
                    ),
                },
            },
        }

        if self._opts.model and self._opts.model != "auto":
            params["model"] = self._opts.model

        if self._opts.language:
            params["settings"]["language"] = self._opts.language

        if self._opts.fallback:
            models = [
                {"model": m.get("model"), "extra": m.get("extra_kwargs")}
                for m in self._opts.fallback
            ]
            params["fallback"] = {"models": models}

        if self._opts.conn_options:
            params["connection"] = {
                "timeout": self._opts.conn_options.timeout,
                "retries": self._opts.conn_options.max_retry,
            }

        base_url = self._opts.base_url
        if base_url.startswith(("http://", "https://")):
            base_url = base_url.replace("http", "ws", 1)
        headers = {
            **get_inference_headers(),
            "Authorization": f"Bearer {create_access_token(self._opts.api_key, self._opts.api_secret)}",
        }
        session_id = headers.get(HEADER_SESSION_ID)
        try:
            ws = await asyncio.wait_for(
                http_session.ws_connect(
                    f"{base_url}/stt?{urlencode({'model': self._opts.model})}", headers=headers
                ),
                self._conn_options.timeout,
            )
            params["type"] = "session.create"
            await ws.send_str(json.dumps(params))
            if session_id is not None:
                self._request_id = session_id
        except aiohttp.ClientResponseError as e:
            raise create_api_error_from_http(e.message, status=e.status) from e
        except asyncio.TimeoutError as e:
            raise APITimeoutError("LiveKit Inference STT connection timed out.") from e
        except aiohttp.ClientConnectorError as e:
            raise APIConnectionError("failed to connect to LiveKit Inference STT") from e
        return ws

    def _build_speech_data(self, data: dict) -> stt.SpeechData:
        language = LanguageCode(data.get("language", self._opts.language or "en"))
        words = data.get("words", []) or []
        # The gateway carries provider-specific data on the `extra` field
        # of the transcript message. We surface it on SpeechData.metadata
        extra = data.get("extra")
        metadata = extra if isinstance(extra, dict) and extra else None
        return stt.SpeechData(
            language=language,
            start_time=self.start_time_offset + data.get("start", 0),
            end_time=self.start_time_offset + data.get("start", 0) + data.get("duration", 0),
            confidence=data.get("confidence", 1.0),
            text=data.get("transcript", ""),
            speaker_id=data.get("speaker_id"),
            words=[
                TimedString(
                    text=word.get("word", ""),
                    start_time=word.get("start", 0) + self.start_time_offset,
                    end_time=word.get("end", 0) + self.start_time_offset,
                    start_time_offset=self.start_time_offset,
                    confidence=word.get("confidence", 0.0),
                    speaker_id=word.get("speaker_id"),
                )
                for word in words
            ],
            metadata=metadata,
        )

    def _process_preflight_transcript(self, data: dict) -> None:
        text = data.get("transcript", "")
        if not text or not self._speaking:
            return

        speech_data = self._build_speech_data(data)
        request_id = data.get("request_id", self._request_id)
        event = stt.SpeechEvent(
            type=stt.SpeechEventType.PREFLIGHT_TRANSCRIPT,
            request_id=request_id,
            alternatives=[speech_data],
        )
        self._event_ch.send_nowait(event)

    def _process_start_of_speech(self) -> None:
        """Onset reported by a provider that detects it server-side (e.g. Cartesia Ink-2).

        Without this the first transcript has to stand in for onset, which lands about
        800ms late because it waits for a word to be decoded.
        """
        if self._speaking:
            return
        self._speaking = True
        self._event_ch.send_nowait(stt.SpeechEvent(type=stt.SpeechEventType.START_OF_SPEECH))

    def _process_transcript(self, data: dict, is_final: bool) -> None:
        request_id = data.get("request_id", self._request_id)
        text = data.get("transcript", "")

        if not text and not is_final:
            return
        self._process_start_of_speech()

        speech_data = self._build_speech_data(data)

        if is_final:
            if self._speech_duration > 0:
                self._event_ch.send_nowait(
                    stt.SpeechEvent(
                        type=stt.SpeechEventType.RECOGNITION_USAGE,
                        request_id=request_id,
                        recognition_usage=stt.RecognitionUsage(
                            audio_duration=self._speech_duration,
                        ),
                    )
                )
                self._speech_duration = 0

            event = stt.SpeechEvent(
                type=stt.SpeechEventType.FINAL_TRANSCRIPT,
                request_id=request_id,
                alternatives=[speech_data],
            )
            self._event_ch.send_nowait(event)

            if self._speaking:
                self._speaking = False
                end_event = stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH)
                self._event_ch.send_nowait(end_event)
                self._on_end_of_speech()
        else:
            event = stt.SpeechEvent(
                type=stt.SpeechEventType.INTERIM_TRANSCRIPT,
                request_id=request_id,
                alternatives=[speech_data],
            )
            self._event_ch.send_nowait(event)

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[STTLanguages | str] = NOT_GIVEN,
extra: NotGivenOr[dict[str, Any]] = NOT_GIVEN) ‑> None

Expand source code

def update_options(
    self,
    *,
    model: NotGivenOr[STTModels | str] = NOT_GIVEN,
    language: NotGivenOr[STTLanguages | str] = NOT_GIVEN,
    extra: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
) -> None:
    """Update streaming transcription options.

    When the WebSocket is live, a mid-stream session.update is sent so providers
    that support it (e.g. AssemblyAI, Deepgram Flux) can apply changes without
    reconnecting. Unsupported providers ignore the message.

    A live session's provider, audio rate, and client endpointing are fixed when it is
    created: the gateway does not switch them through session.update. This stream
    therefore keeps the model it was created with, so its VAD (and the `session.finalize`
    it sends) still matches the running model, and framework keyterms are formatted for
    that model. Streams created after a parent switch start on the new model. The
    requested model is still forwarded, as the other session settings are, but a
    language or extra written for a different provider is left to those new streams.
    """
    changed_provider = False
    if is_given(model):
        refused = self._warn_if_rate_refused(model)
        # Mirror __init__/STT.update_options: strip ":language" suffix. It belongs to the
        # model carrying it, so it is not applied when this stream cannot run that model
        # or when the model belongs to a provider the live session stays off.
        parsed_language: NotGivenOr[LanguageCode] = NOT_GIVEN
        if isinstance(model, str):
            parsed_model, parsed_language = _parse_model_string(model)
            model = parsed_model
        changed_provider = _provider_for_model(model) != _provider_for_model(self._opts.model)
        if (
            is_given(parsed_language)
            and not is_given(language)
            and not refused
            and not changed_provider
        ):
            language = parsed_language

    if is_given(language):
        self._opts.language = LanguageCode(language)

    forward_extra: NotGivenOr[dict[str, Any]] = NOT_GIVEN
    if is_given(extra) and not changed_provider:
        # a pure user baseline: framework keyterms are merged in at forward time, so a
        # later clear can actually remove them. An extra written alongside another
        # provider's model is for that model, not for the one this session is running.
        self._opts.extra_kwargs.update(extra)
        forward_extra = self._forward_extra(extra)

    if not (is_given(model) or is_given(language) or is_given(forward_extra)):
        return
    self._send_live_update(model=model, language=language, extra=forward_extra)

Update streaming transcription options.

When the WebSocket is live, a mid-stream session.update is sent so providers that support it (e.g. AssemblyAI, Deepgram Flux) can apply changes without reconnecting. Unsupported providers ignore the message.

A live session's provider, audio rate, and client endpointing are fixed when it is created: the gateway does not switch them through session.update. This stream therefore keeps the model it was created with, so its VAD (and the session.finalize it sends) still matches the running model, and framework keyterms are formatted for that model. Streams created after a parent switch start on the new model. The requested model is still forwarded, as the other session settings are, but a language or extra written for a different provider is left to those new streams.

class SpeechmaticsOptions (*args, **kwargs)

Expand source code

class SpeechmaticsOptions(TypedDict, total=False):
    domain: str  # e.g. "finance"
    output_locale: str  # BCP-47 locale for output formatting
    max_delay: float  # 0.7-4.0 seconds, default 1.0; RT only
    max_delay_mode: str  # "flexible" | "fixed"; RT only
    diarization: str  # "none" | "speaker" | "channel" | "channel_and_speaker_change" | "speaker_change"; non-"none" enables diarization
    speaker_sensitivity: float  # 0.0-1.0
    max_speakers: int
    prefer_current_speaker: bool
    enable_partials: bool  # default True (overridden by gateway)
    enable_entities: bool  # RT only
    punctuation_overrides: dict[str, Any]
    additional_vocab: list[dict[str, Any]]  # [{"content": ..., "sounds_like": [...]}]
    end_of_utterance_silence_trigger: float  # seconds of silence before final; RT only
    audio_filtering_config: dict[str, Any]  # RT only
    transcript_filtering_config: dict[str, Any]

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

Ancestors

• builtins.dict

Class variables

var additional_vocab : list[dict[str, typing.Any]]

var audio_filtering_config : dict[str, typing.Any]

var diarization : str

var domain : str

var enable_entities : bool

var enable_partials : bool

var end_of_utterance_silence_trigger : float

var max_delay : float

var max_delay_mode : str

var max_speakers : int

var output_locale : str

var prefer_current_speaker : bool

var punctuation_overrides : dict[str, typing.Any]

var speaker_sensitivity : float

var transcript_filtering_config : dict[str, typing.Any]

class XaiOptions (*args, **kwargs)

Expand source code

class XaiOptions(TypedDict, total=False):
    diarize: bool  # when True, enables speaker diarization (default off)
    endpointing: int  # silence duration in ms before utterance-final (0-5000)
    format: bool  # enables Inverse Text Normalization (e.g. "one hundred dollars" -> "$100"); requires language
    interim_results: bool  # default True; set False to opt out of interim transcripts

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

Ancestors

• builtins.dict

Class variables

var diarize : bool

var endpointing : int

var format : bool

var interim_results : bool