Module livekit.plugins.assemblyai

AssemblyAI plugin for LiveKit Agents

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

Classes

class STT (*,
api_key: NotGivenOr[str] = NOT_GIVEN,
sample_rate: int = 16000,
encoding: "Literal['pcm_s16le', 'pcm_mulaw']" = 'pcm_s16le',
model: "Literal['universal-streaming-english', 'universal-streaming-multilingual', 'u3-rt-pro', 'u3-rt-pro-beta-1', 'u3-pro', 'universal-3-5-pro']" = 'universal-3-5-pro',
language_detection: NotGivenOr[bool] = NOT_GIVEN,
language_code: NotGivenOr[str] = NOT_GIVEN,
language_codes: NotGivenOr[str | list[str]] = NOT_GIVEN,
end_of_turn_confidence_threshold: NotGivenOr[float] = NOT_GIVEN,
min_turn_silence: NotGivenOr[int] = NOT_GIVEN,
max_turn_silence: NotGivenOr[int] = NOT_GIVEN,
format_turns: NotGivenOr[bool] = NOT_GIVEN,
continuous_partials: NotGivenOr[bool] = NOT_GIVEN,
interruption_delay: NotGivenOr[int] = NOT_GIVEN,
keyterms_prompt: NotGivenOr[list[str]] = NOT_GIVEN,
prompt: NotGivenOr[str] = NOT_GIVEN,
agent_context: NotGivenOr[str] = NOT_GIVEN,
previous_context_n_turns: NotGivenOr[int] = NOT_GIVEN,
vad_threshold: NotGivenOr[float] = NOT_GIVEN,
speaker_labels: NotGivenOr[bool] = NOT_GIVEN,
max_speakers: NotGivenOr[int] = NOT_GIVEN,
domain: NotGivenOr[str] = NOT_GIVEN,
voice_focus: "NotGivenOr[Literal['near-field', 'far-field']]" = NOT_GIVEN,
voice_focus_threshold: NotGivenOr[float] = NOT_GIVEN,
mode: "NotGivenOr[Literal['min_latency', 'balanced', 'max_accuracy']]" = NOT_GIVEN,
http_session: aiohttp.ClientSession | None = None,
buffer_size_seconds: float = 0.05,
base_url: str = 'wss://streaming.assemblyai.com',
min_end_of_turn_silence_when_confident: NotGivenOr[int] = NOT_GIVEN,
agent_context_carryover: NotGivenOr[bool] = NOT_GIVEN)
Expand source code
class STT(stt.STT):
    def __init__(
        self,
        *,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        sample_rate: int = 16000,
        encoding: Literal["pcm_s16le", "pcm_mulaw"] = "pcm_s16le",
        model: Literal[
            "universal-streaming-english",
            "universal-streaming-multilingual",
            "u3-rt-pro",
            "u3-rt-pro-beta-1",
            "u3-pro",
            "universal-3-5-pro",
        ] = "universal-3-5-pro",
        language_detection: NotGivenOr[bool] = NOT_GIVEN,
        language_code: NotGivenOr[str] = NOT_GIVEN,
        language_codes: NotGivenOr[str | list[str]] = NOT_GIVEN,
        end_of_turn_confidence_threshold: NotGivenOr[float] = NOT_GIVEN,
        min_turn_silence: NotGivenOr[int] = NOT_GIVEN,
        max_turn_silence: NotGivenOr[int] = NOT_GIVEN,
        format_turns: NotGivenOr[bool] = NOT_GIVEN,
        continuous_partials: NotGivenOr[bool] = NOT_GIVEN,
        interruption_delay: NotGivenOr[int] = NOT_GIVEN,
        keyterms_prompt: NotGivenOr[list[str]] = NOT_GIVEN,
        prompt: NotGivenOr[str] = NOT_GIVEN,
        agent_context: NotGivenOr[str] = NOT_GIVEN,
        previous_context_n_turns: NotGivenOr[int] = NOT_GIVEN,
        vad_threshold: NotGivenOr[float] = NOT_GIVEN,
        speaker_labels: NotGivenOr[bool] = NOT_GIVEN,
        max_speakers: NotGivenOr[int] = NOT_GIVEN,
        domain: NotGivenOr[str] = NOT_GIVEN,
        voice_focus: NotGivenOr[Literal["near-field", "far-field"]] = NOT_GIVEN,
        voice_focus_threshold: NotGivenOr[float] = NOT_GIVEN,
        mode: NotGivenOr[Literal["min_latency", "balanced", "max_accuracy"]] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        buffer_size_seconds: float = 0.05,
        base_url: str = "wss://streaming.assemblyai.com",
        # Deprecated — use min_turn_silence instead
        min_end_of_turn_silence_when_confident: NotGivenOr[int] = NOT_GIVEN,
        agent_context_carryover: NotGivenOr[bool] = NOT_GIVEN,
    ):
        """
        Args:
            base_url: The AssemblyAI streaming endpoint base URL. Use the EU endpoint
                (wss://streaming.eu.assemblyai.com) for streaming in the EU. Defaults to
                wss://streaming.assemblyai.com.
                See https://www.assemblyai.com/docs/universal-streaming for more details.
            vad_threshold: The threshold for voice activity detection (VAD). A value between
                0 and 1 that determines how sensitive the VAD is. Lower values make the VAD
                more sensitive (detects quieter speech). Higher values make it less sensitive.
                Defaults to 0.4.
            language_code: Deprecated — use ``language_codes`` instead (it accepts a
                single code directly). Mutually exclusive with ``language_codes``.
            language_codes: Steer transcription toward one or more expected languages.
                Accepts a single code ('es') or a list (['en', 'es']). Each entry
                accepts any common format ('en', 'en-US', 'english') and is normalized
                to a bare ISO 639-1 code before being sent; duplicates after
                normalization are dropped, preserving order. One code biases the model
                toward that language — several codes, toward that set — instead of
                automatically detecting/code-switching across all supported languages.
                At most 10 codes; 'multi' (the unsteered multilingual default) cannot
                be combined with other codes. Leave unset to use the model's default
                multilingual behavior. Only supported with the Universal-3 Pro family
                models. Can be updated mid-session via ``update_options``; pass an
                empty list (or empty string) there to clear steering back to the model
                default. At construction an empty value is equivalent to leaving it
                unset.
            min_turn_silence: Minimum silence in ms before a confident end-of-turn is finalized.
            min_end_of_turn_silence_when_confident: Deprecated. Use min_turn_silence instead.
            continuous_partials: Whether to emit additional partial transcripts during long
                turns at a steady ~3 second cadence, on top of the baseline partials
                (one at 750 ms after turn start, configurable via `interruption_delay`,
                and one each time silence exceeds `min_turn_silence` without ending the
                turn). Leave unset to use AssemblyAI's server defaults: enabled, except
                when `speaker_labels` is on, where the server disables it so turns break
                cleanly at speaker changes. Only supported with the Universal-3 Pro
                family models.
            interruption_delay: How soon the first early partial is emitted, in ms.
                Range 0–1000, default 500. Lower values produce faster time-to-first-token
                for barge-in; higher values produce more confident first partials. Only
                supported with the Universal-3 Pro family models.
            agent_context: Free-text context describing what the agent said, used to bias
                transcription of the user's reply. Set at construction or updated per-turn
                via `update_options(agent_context=...)`. Only supported with the
                Universal-3 Pro family models (max 1750 characters; longer values raise
                ValueError). When chat-context carryover is on (the default on the Universal-3
                Pro family) each assistant reply replaces this value automatically; disable it
                to manage this manually.
            previous_context_n_turns: Maximum number of prior conversation entries (user
                transcripts and any `agent_context` values) carried forward as context for
                each transcription. Set to 0 to disable automatic context carryover
                entirely; leave unset to use the server default (recommended). Range 0–100.
                Only supported with the Universal-3 Pro family models. Set at construction
                (connect) time only; it cannot be changed via `update_options`.
            agent_context_carryover: Deprecated, use
                ``AgentSession(stt_context_options={"forward_chat_context": ...})`` instead.
                On the Universal-3 Pro family, assistant replies are carried into ``agent_context``
                by default; pass ``False`` to opt out. On other models it is off. Replies longer
                than the 1750-character server cap are truncated (keeping the tail) before
                being sent.
            voice_focus: Voice Focus isolates the primary voice and suppresses background
                noise (chatter, keyboard clicks, fan hum, room echo) before the audio reaches
                the model. Use 'near-field' for headsets, handsets, and close-talking
                microphones; use 'far-field' for conference rooms, laptop mics, and other
                distant-mic setups. Only supported with the Universal-3 Pro family models.
                Set at construction (connect) time only.
                See https://www.assemblyai.com/docs/streaming/voice-focus.
            voice_focus_threshold: Controls how aggressively background audio is suppressed,
                a float between 0.0 and 1.0 (higher is more aggressive). Only takes effect
                alongside `voice_focus`. Only supported with the Universal-3 Pro family
                models. Set at construction (connect) time only.
            mode: Accuracy/latency preset for the Universal-3 Pro family: 'min_latency'
                (fastest time-to-text), 'balanced' (the server default, recommended for
                voice agents), or 'max_accuracy' (highest accuracy, for scribes/post-call).
                The model applies its own per-mode silence tuning. To let that tuning take
                effect, the plugin suppresses its default 100ms min/max turn-silence windows
                when a mode is set; values you pass explicitly for `min_turn_silence` /
                `max_turn_silence` still take precedence over the mode's defaults.
                Leave unset to use the server default. Only supported with the Universal-3 Pro
                family models. Set at construction (connect) time only.
        """
        if is_given(language_code) and is_given(language_codes):
            raise ValueError(
                "language_code and language_codes are mutually exclusive; "
                "use language_codes (it accepts a single code directly)"
            )
        if is_given(language_code):
            logger.warning("'language_code' is deprecated, use 'language_codes' instead.")
        # An explicit empty value ([] or "") is equivalent to unset at
        # construction — the param is omitted from the connect query either
        # way — so it is not subject to the U3-Pro-family gate below.
        if is_given(language_codes) and not language_codes:
            language_codes = NOT_GIVEN
        if is_given(agent_context):
            _validate_agent_context(agent_context)

        # agent_context carryover is only available on the u3-rt-pro family ("u3-pro" is
        # normalized to "u3-rt-pro" below), where it is on by default; the session's
        # stt_context_options.forward_chat_context toggle is the supported way to control it.
        supports_carryover = model in _U3_PRO_MODELS or model == "u3-pro"
        if is_given(agent_context_carryover):
            logger.warning(
                "'agent_context_carryover' is deprecated, use "
                "AgentSession(stt_context_options={'forward_chat_context': ...}) instead."
            )
            if agent_context_carryover and not supports_carryover:
                logger.warning(
                    "agent_context_carryover is enabled but model %r does not support it; ignoring",
                    model,
                )
        # on by default for supported models; an explicit agent_context_carryover=False opts out
        carryover_enabled = supports_carryover and (
            agent_context_carryover if is_given(agent_context_carryover) else True
        )
        super().__init__(
            capabilities=stt.STTCapabilities(
                streaming=True,
                interim_results=True,
                aligned_transcript="word",
                offline_recognize=False,
                diarization=is_given(speaker_labels) and speaker_labels is True,
                keyterms=True,
                chat_context=carryover_enabled,
            ),
        )
        if model == "u3-pro":
            logger.warning("'u3-pro' is deprecated, use 'universal-3-5-pro' instead.")
            model = "universal-3-5-pro"

        # These parameters are only supported by the Universal-3 Pro family of models.
        if model not in _U3_PRO_MODELS:
            _u3_pro_only_params = {
                "prompt": prompt,
                "agent_context": agent_context,
                "previous_context_n_turns": previous_context_n_turns,
                "continuous_partials": continuous_partials,
                "interruption_delay": interruption_delay,
                "voice_focus": voice_focus,
                "voice_focus_threshold": voice_focus_threshold,
                "mode": mode,
                "language_code": language_code,
                "language_codes": language_codes,
            }
            for _param_name, _param_value in _u3_pro_only_params.items():
                if is_given(_param_value):
                    raise ValueError(
                        f"The {_param_name!r} parameter is only supported with the "
                        f"{', '.join(_U3_PRO_MODELS)} models."
                    )

        self._base_url = base_url
        assemblyai_api_key = api_key if is_given(api_key) else os.environ.get("ASSEMBLYAI_API_KEY")
        if not assemblyai_api_key:
            raise ValueError(
                "AssemblyAI API key is required. "
                "Pass one in via the `api_key` parameter, "
                "or set it as the `ASSEMBLYAI_API_KEY` environment variable"
            )
        self._api_key = assemblyai_api_key

        # Handle deprecated min_end_of_turn_silence_when_confident
        if is_given(min_end_of_turn_silence_when_confident):
            logger.warning(
                "'min_end_of_turn_silence_when_confident' is deprecated, "
                "use 'min_turn_silence' instead."
            )
            if not is_given(min_turn_silence):
                min_turn_silence = min_end_of_turn_silence_when_confident

        # we want to minimize latency as much as possible, it's ok if the phrase arrives in multiple final transcripts
        # designed to work with LK's end of turn models.
        # Skip this default when a `mode` preset is selected so the server's
        # per-mode silence tuning governs instead of being overridden by 100.
        if not is_given(min_turn_silence) and not is_given(mode):
            min_turn_silence = 100

        # Normalize to bare ISO 639-1 codes (e.g. "es-ES" / "Spanish" -> "es"),
        # the form AssemblyAI's language steering expects. The singular
        # language_code is shorthand for a one-element list.
        normalized_language_codes: NotGivenOr[list[str]] = NOT_GIVEN
        if is_given(language_code):
            normalized_language_codes = _normalize_language_codes(language_code)
        elif is_given(language_codes):
            normalized_language_codes = _normalize_language_codes(language_codes)

        self._opts = STTOptions(
            sample_rate=sample_rate,
            buffer_size_seconds=buffer_size_seconds,
            encoding=encoding,
            speech_model=model,
            language_detection=language_detection,
            language_codes=normalized_language_codes,
            end_of_turn_confidence_threshold=end_of_turn_confidence_threshold,
            min_turn_silence=min_turn_silence,
            max_turn_silence=max_turn_silence,
            format_turns=format_turns,
            continuous_partials=continuous_partials,
            interruption_delay=interruption_delay,
            keyterms_prompt=keyterms_prompt,
            prompt=prompt,
            agent_context=agent_context,
            previous_context_n_turns=previous_context_n_turns,
            vad_threshold=vad_threshold,
            speaker_labels=speaker_labels,
            max_speakers=max_speakers,
            domain=domain,
            voice_focus=voice_focus,
            voice_focus_threshold=voice_focus_threshold,
            mode=mode,
        )
        self._session = http_session
        # user keyterms; _opts.keyterms_prompt holds the effective set (user + session)
        self._user_keyterms: list[str] = list(keyterms_prompt or [])
        self._session_keyterms: list[str] = []
        self._streams = weakref.WeakSet[SpeechStream]()

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

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

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

    async def _recognize_impl(
        self,
        buffer: AudioBuffer,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        conn_options: APIConnectOptions,
    ) -> stt.SpeechEvent:
        raise NotImplementedError("Not implemented")

    def stream(
        self,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> SpeechStream:
        config = dataclasses.replace(self._opts)
        stream = SpeechStream(
            stt=self,
            conn_options=conn_options,
            opts=config,
            api_key=self._api_key,
            http_session=self.session,
            base_url=self._base_url,
        )
        self._streams.add(stream)
        return stream

    def update_options(
        self,
        *,
        buffer_size_seconds: NotGivenOr[float] = NOT_GIVEN,
        end_of_turn_confidence_threshold: NotGivenOr[float] = NOT_GIVEN,
        min_turn_silence: NotGivenOr[int] = NOT_GIVEN,
        max_turn_silence: NotGivenOr[int] = NOT_GIVEN,
        prompt: NotGivenOr[str] = NOT_GIVEN,
        agent_context: NotGivenOr[str] = NOT_GIVEN,
        keyterms_prompt: NotGivenOr[list[str]] = NOT_GIVEN,
        language_codes: NotGivenOr[str | list[str]] = NOT_GIVEN,
        vad_threshold: NotGivenOr[float] = NOT_GIVEN,
        continuous_partials: NotGivenOr[bool] = NOT_GIVEN,
        interruption_delay: NotGivenOr[int] = NOT_GIVEN,
        # Deprecated — use min_turn_silence instead
        min_end_of_turn_silence_when_confident: NotGivenOr[int] = NOT_GIVEN,
    ) -> None:
        if is_given(min_end_of_turn_silence_when_confident):
            logger.warning(
                "'min_end_of_turn_silence_when_confident' is deprecated, "
                "use 'min_turn_silence' instead."
            )
            if not is_given(min_turn_silence):
                min_turn_silence = min_end_of_turn_silence_when_confident

        # Validate/normalize before mutating any option so a ValueError from a
        # bad value cannot leave _opts partially updated.
        if is_given(language_codes):
            if self._opts.speech_model not in _U3_PRO_MODELS:
                raise ValueError(
                    "The 'language_codes' parameter is only supported with the "
                    f"{', '.join(_U3_PRO_MODELS)} models."
                )
            language_codes = _normalize_language_codes(language_codes)
        if is_given(agent_context):
            _validate_agent_context(agent_context)

        if is_given(buffer_size_seconds):
            self._opts.buffer_size_seconds = buffer_size_seconds
        if is_given(end_of_turn_confidence_threshold):
            self._opts.end_of_turn_confidence_threshold = end_of_turn_confidence_threshold
        if is_given(min_turn_silence):
            self._opts.min_turn_silence = min_turn_silence
        if is_given(max_turn_silence):
            self._opts.max_turn_silence = max_turn_silence
        if is_given(prompt):
            self._opts.prompt = prompt
        if is_given(agent_context):
            self._opts.agent_context = agent_context
        if is_given(keyterms_prompt):
            self._user_keyterms = list(keyterms_prompt)
            # re-merge with the active session keyterms so a user update doesn't drop them
            keyterms_prompt = list(dict.fromkeys([*self._user_keyterms, *self._session_keyterms]))
            self._opts.keyterms_prompt = keyterms_prompt
        if is_given(language_codes):
            self._opts.language_codes = language_codes
        if is_given(vad_threshold):
            self._opts.vad_threshold = vad_threshold
        if is_given(continuous_partials):
            self._opts.continuous_partials = continuous_partials
        if is_given(interruption_delay):
            self._opts.interruption_delay = interruption_delay

        for stream in self._streams:
            stream.update_options(
                buffer_size_seconds=buffer_size_seconds,
                end_of_turn_confidence_threshold=end_of_turn_confidence_threshold,
                min_turn_silence=min_turn_silence,
                max_turn_silence=max_turn_silence,
                prompt=prompt,
                agent_context=agent_context,
                keyterms_prompt=keyterms_prompt,
                language_codes=language_codes,
                vad_threshold=vad_threshold,
                continuous_partials=continuous_partials,
                interruption_delay=interruption_delay,
            )

    def _update_session_keyterms(self, keyterms: list[str]) -> None:
        if keyterms == self._session_keyterms:
            return
        self._session_keyterms = list(keyterms)
        merged = list(dict.fromkeys([*self._user_keyterms, *keyterms]))
        self._opts.keyterms_prompt = merged
        # applied live via the stream's UpdateConfiguration (no reconnect)
        for stream in self._streams:
            stream.update_options(keyterms_prompt=merged)

    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) > _MAX_AGENT_CONTEXT_CHARS:
                # Keep the tail: the end of the reply (the question posed to
                # the user) has the most biasing value for their next utterance.
                logger.debug(
                    "truncating agent_context carryover from %d to %d chars",
                    len(text),
                    _MAX_AGENT_CONTEXT_CHARS,
                )
                text = text[-_MAX_AGENT_CONTEXT_CHARS:]
            self.update_options(agent_context=text)

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

Args

base_url
The AssemblyAI streaming endpoint base URL. Use the EU endpoint (wss://streaming.eu.assemblyai.com) for streaming in the EU. Defaults to wss://streaming.assemblyai.com. See https://www.assemblyai.com/docs/universal-streaming for more details.
vad_threshold
The threshold for voice activity detection (VAD). A value between 0 and 1 that determines how sensitive the VAD is. Lower values make the VAD more sensitive (detects quieter speech). Higher values make it less sensitive. Defaults to 0.4.
language_code
Deprecated — use language_codes instead (it accepts a single code directly). Mutually exclusive with language_codes.
language_codes
Steer transcription toward one or more expected languages. Accepts a single code ('es') or a list (['en', 'es']). Each entry accepts any common format ('en', 'en-US', 'english') and is normalized to a bare ISO 639-1 code before being sent; duplicates after normalization are dropped, preserving order. One code biases the model toward that language — several codes, toward that set — instead of automatically detecting/code-switching across all supported languages. At most 10 codes; 'multi' (the unsteered multilingual default) cannot be combined with other codes. Leave unset to use the model's default multilingual behavior. Only supported with the Universal-3 Pro family models. Can be updated mid-session via update_options; pass an empty list (or empty string) there to clear steering back to the model default. At construction an empty value is equivalent to leaving it unset.
min_turn_silence
Minimum silence in ms before a confident end-of-turn is finalized.
min_end_of_turn_silence_when_confident
Deprecated. Use min_turn_silence instead.
continuous_partials
Whether to emit additional partial transcripts during long turns at a steady ~3 second cadence, on top of the baseline partials (one at 750 ms after turn start, configurable via interruption_delay, and one each time silence exceeds min_turn_silence without ending the turn). Leave unset to use AssemblyAI's server defaults: enabled, except when speaker_labels is on, where the server disables it so turns break cleanly at speaker changes. Only supported with the Universal-3 Pro family models.
interruption_delay
How soon the first early partial is emitted, in ms. Range 0–1000, default 500. Lower values produce faster time-to-first-token for barge-in; higher values produce more confident first partials. Only supported with the Universal-3 Pro family models.
agent_context
Free-text context describing what the agent said, used to bias transcription of the user's reply. Set at construction or updated per-turn via update_options(agent_context=...). Only supported with the Universal-3 Pro family models (max 1750 characters; longer values raise ValueError). When chat-context carryover is on (the default on the Universal-3 Pro family) each assistant reply replaces this value automatically; disable it to manage this manually.
previous_context_n_turns
Maximum number of prior conversation entries (user transcripts and any agent_context values) carried forward as context for each transcription. Set to 0 to disable automatic context carryover entirely; leave unset to use the server default (recommended). Range 0–100. Only supported with the Universal-3 Pro family models. Set at construction (connect) time only; it cannot be changed via update_options.
agent_context_carryover
Deprecated, use AgentSession(stt_context_options={"forward_chat_context": ...}) instead. On the Universal-3 Pro family, assistant replies are carried into agent_context by default; pass False to opt out. On other models it is off. Replies longer than the 1750-character server cap are truncated (keeping the tail) before being sent.
voice_focus
Voice Focus isolates the primary voice and suppresses background noise (chatter, keyboard clicks, fan hum, room echo) before the audio reaches the model. Use 'near-field' for headsets, handsets, and close-talking microphones; use 'far-field' for conference rooms, laptop mics, and other distant-mic setups. Only supported with the Universal-3 Pro family models. Set at construction (connect) time only. See https://www.assemblyai.com/docs/streaming/voice-focus.
voice_focus_threshold
Controls how aggressively background audio is suppressed, a float between 0.0 and 1.0 (higher is more aggressive). Only takes effect alongside voice_focus. Only supported with the Universal-3 Pro family models. Set at construction (connect) time only.
mode
Accuracy/latency preset for the Universal-3 Pro family: 'min_latency' (fastest time-to-text), 'balanced' (the server default, recommended for voice agents), or 'max_accuracy' (highest accuracy, for scribes/post-call). The model applies its own per-mode silence tuning. To let that tuning take effect, the plugin suppresses its default 100ms min/max turn-silence windows when a mode is set; values you pass explicitly for min_turn_silence / max_turn_silence still take precedence over the mode's defaults. Leave unset to use the server default. Only supported with the Universal-3 Pro family models. Set at construction (connect) time only.

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.speech_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 "AssemblyAI"

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.

prop session : aiohttp.ClientSession
Expand source code
@property
def session(self) -> aiohttp.ClientSession:
    if not self._session:
        self._session = utils.http_context.http_session()
    return self._session

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.assemblyai.stt.SpeechStream
Expand source code
def stream(
    self,
    *,
    language: NotGivenOr[str] = NOT_GIVEN,
    conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
) -> SpeechStream:
    config = dataclasses.replace(self._opts)
    stream = SpeechStream(
        stt=self,
        conn_options=conn_options,
        opts=config,
        api_key=self._api_key,
        http_session=self.session,
        base_url=self._base_url,
    )
    self._streams.add(stream)
    return stream
def update_options(self,
*,
buffer_size_seconds: NotGivenOr[float] = NOT_GIVEN,
end_of_turn_confidence_threshold: NotGivenOr[float] = NOT_GIVEN,
min_turn_silence: NotGivenOr[int] = NOT_GIVEN,
max_turn_silence: NotGivenOr[int] = NOT_GIVEN,
prompt: NotGivenOr[str] = NOT_GIVEN,
agent_context: NotGivenOr[str] = NOT_GIVEN,
keyterms_prompt: NotGivenOr[list[str]] = NOT_GIVEN,
language_codes: NotGivenOr[str | list[str]] = NOT_GIVEN,
vad_threshold: NotGivenOr[float] = NOT_GIVEN,
continuous_partials: NotGivenOr[bool] = NOT_GIVEN,
interruption_delay: NotGivenOr[int] = NOT_GIVEN,
min_end_of_turn_silence_when_confident: NotGivenOr[int] = NOT_GIVEN) ‑> None
Expand source code
def update_options(
    self,
    *,
    buffer_size_seconds: NotGivenOr[float] = NOT_GIVEN,
    end_of_turn_confidence_threshold: NotGivenOr[float] = NOT_GIVEN,
    min_turn_silence: NotGivenOr[int] = NOT_GIVEN,
    max_turn_silence: NotGivenOr[int] = NOT_GIVEN,
    prompt: NotGivenOr[str] = NOT_GIVEN,
    agent_context: NotGivenOr[str] = NOT_GIVEN,
    keyterms_prompt: NotGivenOr[list[str]] = NOT_GIVEN,
    language_codes: NotGivenOr[str | list[str]] = NOT_GIVEN,
    vad_threshold: NotGivenOr[float] = NOT_GIVEN,
    continuous_partials: NotGivenOr[bool] = NOT_GIVEN,
    interruption_delay: NotGivenOr[int] = NOT_GIVEN,
    # Deprecated — use min_turn_silence instead
    min_end_of_turn_silence_when_confident: NotGivenOr[int] = NOT_GIVEN,
) -> None:
    if is_given(min_end_of_turn_silence_when_confident):
        logger.warning(
            "'min_end_of_turn_silence_when_confident' is deprecated, "
            "use 'min_turn_silence' instead."
        )
        if not is_given(min_turn_silence):
            min_turn_silence = min_end_of_turn_silence_when_confident

    # Validate/normalize before mutating any option so a ValueError from a
    # bad value cannot leave _opts partially updated.
    if is_given(language_codes):
        if self._opts.speech_model not in _U3_PRO_MODELS:
            raise ValueError(
                "The 'language_codes' parameter is only supported with the "
                f"{', '.join(_U3_PRO_MODELS)} models."
            )
        language_codes = _normalize_language_codes(language_codes)
    if is_given(agent_context):
        _validate_agent_context(agent_context)

    if is_given(buffer_size_seconds):
        self._opts.buffer_size_seconds = buffer_size_seconds
    if is_given(end_of_turn_confidence_threshold):
        self._opts.end_of_turn_confidence_threshold = end_of_turn_confidence_threshold
    if is_given(min_turn_silence):
        self._opts.min_turn_silence = min_turn_silence
    if is_given(max_turn_silence):
        self._opts.max_turn_silence = max_turn_silence
    if is_given(prompt):
        self._opts.prompt = prompt
    if is_given(agent_context):
        self._opts.agent_context = agent_context
    if is_given(keyterms_prompt):
        self._user_keyterms = list(keyterms_prompt)
        # re-merge with the active session keyterms so a user update doesn't drop them
        keyterms_prompt = list(dict.fromkeys([*self._user_keyterms, *self._session_keyterms]))
        self._opts.keyterms_prompt = keyterms_prompt
    if is_given(language_codes):
        self._opts.language_codes = language_codes
    if is_given(vad_threshold):
        self._opts.vad_threshold = vad_threshold
    if is_given(continuous_partials):
        self._opts.continuous_partials = continuous_partials
    if is_given(interruption_delay):
        self._opts.interruption_delay = interruption_delay

    for stream in self._streams:
        stream.update_options(
            buffer_size_seconds=buffer_size_seconds,
            end_of_turn_confidence_threshold=end_of_turn_confidence_threshold,
            min_turn_silence=min_turn_silence,
            max_turn_silence=max_turn_silence,
            prompt=prompt,
            agent_context=agent_context,
            keyterms_prompt=keyterms_prompt,
            language_codes=language_codes,
            vad_threshold=vad_threshold,
            continuous_partials=continuous_partials,
            interruption_delay=interruption_delay,
        )

Inherited members

class SpeechStream (*,
stt: STT,
opts: STTOptions,
conn_options: APIConnectOptions,
api_key: str,
http_session: aiohttp.ClientSession,
base_url: str)
Expand source code
class SpeechStream(stt.SpeechStream):
    # Used to close websocket
    _CLOSE_MSG: str = json.dumps({"type": "Terminate"})

    def __init__(
        self,
        *,
        stt: STT,
        opts: STTOptions,
        conn_options: APIConnectOptions,
        api_key: str,
        http_session: aiohttp.ClientSession,
        base_url: str,
    ) -> None:
        super().__init__(stt=stt, conn_options=conn_options, sample_rate=opts.sample_rate)

        self._opts = opts
        self._api_key = api_key
        self._session = http_session
        self._base_url = base_url
        self._speech_duration: float = 0
        self._last_preflight_start_time: float = 0
        self._config_update_queue: asyncio.Queue[dict] = asyncio.Queue()
        self._session_id: str | None = None
        self._expires_at: int | None = None
        self._last_frame_sent_at: float | None = None

    @property
    def session_id(self) -> str | None:
        """The AssemblyAI session ID. Set when the WebSocket connection is established
        (before any speech events). None until the connection completes.
        Share this with the AssemblyAI team when reporting issues."""
        return self._session_id

    @property
    def expires_at(self) -> int | None:
        """Unix timestamp when the AssemblyAI session expires. Set alongside session_id
        when the WebSocket connection is established."""
        return self._expires_at

    def update_options(
        self,
        *,
        buffer_size_seconds: NotGivenOr[float] = NOT_GIVEN,
        end_of_turn_confidence_threshold: NotGivenOr[float] = NOT_GIVEN,
        min_turn_silence: NotGivenOr[int] = NOT_GIVEN,
        max_turn_silence: NotGivenOr[int] = NOT_GIVEN,
        prompt: NotGivenOr[str] = NOT_GIVEN,
        agent_context: NotGivenOr[str] = NOT_GIVEN,
        keyterms_prompt: NotGivenOr[list[str]] = NOT_GIVEN,
        language_codes: NotGivenOr[str | list[str]] = NOT_GIVEN,
        vad_threshold: NotGivenOr[float] = NOT_GIVEN,
        continuous_partials: NotGivenOr[bool] = NOT_GIVEN,
        interruption_delay: NotGivenOr[int] = NOT_GIVEN,
        # Deprecated — use min_turn_silence instead
        min_end_of_turn_silence_when_confident: NotGivenOr[int] = NOT_GIVEN,
    ) -> None:
        if is_given(min_end_of_turn_silence_when_confident):
            logger.warning(
                "'min_end_of_turn_silence_when_confident' is deprecated, "
                "use 'min_turn_silence' instead."
            )
            if not is_given(min_turn_silence):
                min_turn_silence = min_end_of_turn_silence_when_confident

        # Validate/normalize before mutating any option so a ValueError from a
        # bad value cannot leave _opts partially updated.
        if is_given(language_codes):
            if self._opts.speech_model not in _U3_PRO_MODELS:
                raise ValueError(
                    "The 'language_codes' parameter is only supported with the "
                    f"{', '.join(_U3_PRO_MODELS)} models."
                )
            language_codes = _normalize_language_codes(language_codes)
        if is_given(agent_context):
            _validate_agent_context(agent_context)

        if is_given(buffer_size_seconds):
            self._opts.buffer_size_seconds = buffer_size_seconds
        if is_given(end_of_turn_confidence_threshold):
            self._opts.end_of_turn_confidence_threshold = end_of_turn_confidence_threshold
        if is_given(min_turn_silence):
            self._opts.min_turn_silence = min_turn_silence
        if is_given(max_turn_silence):
            self._opts.max_turn_silence = max_turn_silence
        if is_given(prompt):
            self._opts.prompt = prompt
        if is_given(agent_context):
            self._opts.agent_context = agent_context
        if is_given(keyterms_prompt):
            self._opts.keyterms_prompt = keyterms_prompt
        if is_given(language_codes):
            self._opts.language_codes = language_codes
        if is_given(vad_threshold):
            self._opts.vad_threshold = vad_threshold
        if is_given(continuous_partials):
            self._opts.continuous_partials = continuous_partials
        if is_given(interruption_delay):
            self._opts.interruption_delay = interruption_delay

        # Send UpdateConfiguration message over the active websocket
        config_msg: dict = {"type": "UpdateConfiguration"}
        if is_given(prompt):
            config_msg["prompt"] = prompt
        if is_given(agent_context):
            config_msg["agent_context"] = agent_context
        if is_given(keyterms_prompt):
            config_msg["keyterms_prompt"] = keyterms_prompt
        if is_given(language_codes):
            config_msg["language_codes"] = language_codes
        if is_given(max_turn_silence):
            config_msg["max_turn_silence"] = max_turn_silence
        if is_given(min_turn_silence):
            config_msg["min_turn_silence"] = min_turn_silence
        if is_given(end_of_turn_confidence_threshold):
            config_msg["end_of_turn_confidence_threshold"] = end_of_turn_confidence_threshold
        if is_given(continuous_partials):
            config_msg["continuous_partials"] = continuous_partials
        if is_given(interruption_delay):
            config_msg["interruption_delay"] = interruption_delay
        if is_given(vad_threshold):
            config_msg["vad_threshold"] = vad_threshold

        if len(config_msg) > 1:
            self._config_update_queue.put_nowait(config_msg)

    def force_endpoint(self) -> None:
        """Force-finalize the current turn immediately."""
        self._config_update_queue.put_nowait({"type": "ForceEndpoint"})

    async def _run(self) -> None:
        """Run a single websocket connection to AssemblyAI."""
        closing_ws = False

        async def send_task(ws: aiohttp.ClientWebSocketResponse) -> None:
            nonlocal closing_ws
            anchored = False

            samples_per_buffer = self._opts.sample_rate // round(1 / self._opts.buffer_size_seconds)
            audio_bstream = utils.audio.AudioByteStream(
                sample_rate=self._opts.sample_rate,
                num_channels=1,
                samples_per_channel=samples_per_buffer,
            )

            # forward inputs to AssemblyAI
            # if we receive a close message, signal it to AssemblyAI and break.
            # the recv task will then make sure to process the remaining audio and stop
            try:
                async for data in self._input_ch:
                    if isinstance(data, self._FlushSentinel):
                        frames = audio_bstream.flush()
                    else:
                        frames = audio_bstream.write(data.data.tobytes())

                    for frame in frames:
                        if not anchored:
                            # Anchor the stream's wall-clock to the moment just
                            # before the first frame is sent — aligned with the
                            # server's stream-relative zero used by
                            # SpeechStarted.timestamp.
                            self.start_time = time.time()
                            anchored = True
                        self._speech_duration += frame.duration
                        await ws.send_bytes(frame.data.tobytes())
                        self._last_frame_sent_at = time.time()

                closing_ws = True
                logger.debug("AssemblyAI sending close message session=%s", self._session_id)
                await ws.send_str(SpeechStream._CLOSE_MSG)
            except (aiohttp.ClientError, ConnectionError) as e:
                if closing_ws or self._session.closed:
                    return
                raise APIConnectionError("AssemblyAI connection closed unexpectedly") from e

        async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
            nonlocal closing_ws
            consecutive_timeouts = 0
            while True:
                try:
                    msg = await asyncio.wait_for(ws.receive(), timeout=5)
                    consecutive_timeouts = 0
                except asyncio.TimeoutError:
                    if closing_ws:
                        break
                    consecutive_timeouts += 1
                    # First warning at 15s, then every 15s while silence continues.
                    # `session=None` here means WS connected but AAI never sent `Begin`.
                    if consecutive_timeouts % 3 == 0:
                        logger.warning(
                            "AssemblyAI no messages received for %ds session=%s",
                            consecutive_timeouts * 5,
                            self._session_id,
                        )
                        # If the send side is also idle, the stall is upstream
                        # of this plugin (no audio reaching us). Otherwise
                        # frames are flowing and the stall is downstream.
                        if self._last_frame_sent_at is not None:
                            send_idle_s = time.time() - self._last_frame_sent_at
                            if send_idle_s >= 15:
                                logger.warning(
                                    "AssemblyAI no audio frames sent for %.0fs session=%s",
                                    send_idle_s,
                                    self._session_id,
                                )
                    continue

                if msg.type in (
                    aiohttp.WSMsgType.CLOSED,
                    aiohttp.WSMsgType.CLOSE,
                    aiohttp.WSMsgType.CLOSING,
                ):
                    if closing_ws:  # close is expected, see SpeechStream.aclose
                        return

                    logger.warning(
                        "AssemblyAI WebSocket closed unexpectedly "
                        "session=%s code=%s data=%s extra=%s",
                        self._session_id,
                        ws.close_code,
                        msg.data,
                        msg.extra,
                    )
                    raise APIStatusError(
                        "AssemblyAI connection closed unexpectedly",
                        status_code=ws.close_code or -1,
                        body=f"{msg.data=} {msg.extra=}",
                    )

                if msg.type != aiohttp.WSMsgType.TEXT:
                    logger.error(
                        "unexpected AssemblyAI message type=%s session=%s",
                        msg.type,
                        self._session_id,
                    )
                    continue

                try:
                    self._process_stream_event(json.loads(msg.data))
                except Exception:
                    logger.exception(
                        "failed to process AssemblyAI message session=%s",
                        self._session_id,
                    )

        async def send_config_task(ws: aiohttp.ClientWebSocketResponse) -> None:
            """Send config updates and control messages immediately, independent of audio."""
            while True:
                config_msg = await self._config_update_queue.get()
                await ws.send_str(json.dumps(config_msg))

        ws: aiohttp.ClientWebSocketResponse | None = None
        try:
            ws = await self._connect_ws()
            config_task = asyncio.create_task(send_config_task(ws))
            tasks = [
                asyncio.create_task(send_task(ws)),
                asyncio.create_task(recv_task(ws)),
            ]
            try:
                await asyncio.gather(*tasks)
            finally:
                await utils.aio.gracefully_cancel(config_task, *tasks)
        finally:
            if ws is not None:
                await ws.close()

    async def _connect_ws(self) -> aiohttp.ClientWebSocketResponse:
        # Universal-3 Pro family defaults: min=100, max=min (so both 100 unless overridden).
        # When a `mode` preset is selected, leave them unset (None) unless the
        # caller set them explicitly, so the server's per-mode silence tuning is
        # not overridden by the latency-optimized 100ms default.
        min_silence: int | None
        max_silence: int | None
        if self._opts.speech_model in _U3_PRO_MODELS:
            default_min = None if is_given(self._opts.mode) else 100
            min_silence = (
                self._opts.min_turn_silence
                if is_given(self._opts.min_turn_silence)
                else default_min
            )
            max_silence = (
                self._opts.max_turn_silence
                if is_given(self._opts.max_turn_silence)
                else min_silence
            )
        else:
            min_silence = (
                self._opts.min_turn_silence if is_given(self._opts.min_turn_silence) else None
            )
            max_silence = (
                self._opts.max_turn_silence if is_given(self._opts.max_turn_silence) else None
            )

        live_config = {
            "sample_rate": self._opts.sample_rate,
            "encoding": self._opts.encoding,
            "speech_model": self._opts.speech_model,
            "format_turns": self._opts.format_turns if is_given(self._opts.format_turns) else None,
            "continuous_partials": self._opts.continuous_partials
            if is_given(self._opts.continuous_partials)
            else None,
            "interruption_delay": self._opts.interruption_delay
            if is_given(self._opts.interruption_delay)
            else None,
            "end_of_turn_confidence_threshold": self._opts.end_of_turn_confidence_threshold
            if is_given(self._opts.end_of_turn_confidence_threshold)
            else None,
            "min_turn_silence": min_silence,
            "max_turn_silence": max_silence,
            "keyterms_prompt": json.dumps(self._opts.keyterms_prompt)
            if self._opts.keyterms_prompt
            else None,
            "language_detection": self._opts.language_detection
            if is_given(self._opts.language_detection)
            else True
            if "multilingual" in self._opts.speech_model
            or self._opts.speech_model in _U3_PRO_MODELS
            else False,
            "language_codes": json.dumps(self._opts.language_codes)
            if is_given(self._opts.language_codes) and self._opts.language_codes
            else None,
            "prompt": self._opts.prompt if is_given(self._opts.prompt) else None,
            "agent_context": self._opts.agent_context
            if is_given(self._opts.agent_context)
            else None,
            "previous_context_n_turns": self._opts.previous_context_n_turns
            if is_given(self._opts.previous_context_n_turns)
            else None,
            "vad_threshold": self._opts.vad_threshold
            if is_given(self._opts.vad_threshold)
            else None,
            "speaker_labels": self._opts.speaker_labels
            if is_given(self._opts.speaker_labels)
            else None,
            "max_speakers": self._opts.max_speakers if is_given(self._opts.max_speakers) else None,
            "domain": self._opts.domain if is_given(self._opts.domain) else None,
            "voice_focus": self._opts.voice_focus if is_given(self._opts.voice_focus) else None,
            "voice_focus_threshold": self._opts.voice_focus_threshold
            if is_given(self._opts.voice_focus_threshold)
            else None,
            "mode": self._opts.mode if is_given(self._opts.mode) else None,
        }

        headers = {
            "Authorization": self._api_key,
            "Content-Type": "application/json",
            "User-Agent": "AssemblyAI/1.0 (integration=Livekit)",
        }

        filtered_config = {
            k: ("true" if v else "false") if isinstance(v, bool) else v
            for k, v in live_config.items()
            if v is not None
        }
        url = f"{self._base_url}/v3/ws?{urlencode(filtered_config)}"
        logger.debug(
            "connecting to AssemblyAI model=%s base_url=%s",
            self._opts.speech_model,
            self._base_url,
        )
        ws = await self._session.ws_connect(url, headers=headers)
        logger.debug(
            "AssemblyAI WebSocket connected status=%s",
            ws._response.status if ws._response is not None else None,
        )
        return ws

    def _process_stream_event(self, data: dict) -> None:
        message_type = data.get("type")

        if message_type == "Begin":
            self._session_id = data.get("id")
            self._expires_at = data.get("expires_at")
            logger.info(
                "AssemblyAI session started id=%s expires_at=%s",
                self._session_id,
                self._expires_at,
            )
            return

        if message_type == "SpeechStarted":
            # SpeechStarted can arrive well after actual speech onset. The
            # `timestamp` field carries the server VAD's onset time in stream-
            # relative ms. Convert to wall-clock by adding self.start_time
            # (the stream's wall-clock anchor) so the framework records an
            # accurate _speech_start_time instead of message arrival.
            timestamp_ms = data.get("timestamp")
            speech_start_time: float | None = None
            if timestamp_ms is not None:
                speech_start_time = self.start_time + timestamp_ms / 1000
            self._event_ch.send_nowait(
                stt.SpeechEvent(
                    type=stt.SpeechEventType.START_OF_SPEECH,
                    speech_start_time=speech_start_time,
                )
            )
            return

        if message_type == "Termination":
            audio_duration = data.get("audio_duration_seconds")
            session_duration = data.get("session_duration_seconds")
            logger.debug(
                "AssemblyAI session terminated session=%s audio_duration=%ss session_duration=%ss",
                self._session_id,
                audio_duration,
                session_duration,
            )
            return

        if message_type != "Turn":
            logger.debug(
                "AssemblyAI unhandled message type=%s session=%s",
                message_type,
                self._session_id,
            )
            return
        words = data.get("words", [])
        end_of_turn = data.get("end_of_turn", False)
        end_of_turn_confidence = data.get("end_of_turn_confidence")
        turn_is_formatted = data.get("turn_is_formatted", False)
        utterance = data.get("utterance", "")
        transcript = data.get("transcript", "")
        language = LanguageCode(data.get("language_code", "en"))

        # Extract speaker label for diarization (returns "A", "B", ... or "UNKNOWN")
        speaker_label = data.get("speaker_label")
        speaker_id = speaker_label if speaker_label and speaker_label != "UNKNOWN" else None

        # transcript (final) and words (interim) are cumulative
        # utterance (preflight) is chunk based
        start_time: float = 0
        end_time: float = 0
        confidence: float = 0
        # word timestamps are in milliseconds
        # https://www.assemblyai.com/docs/api-reference/streaming-api/streaming-api#receive.receiveTurn.words
        timed_words: list[TimedString] = [
            TimedString(
                text=word.get("text", ""),
                start_time=word.get("start", 0) / 1000 + self.start_time_offset,
                end_time=word.get("end", 0) / 1000 + self.start_time_offset,
                start_time_offset=self.start_time_offset,
                confidence=word.get("confidence", 0),
            )
            for word in words
        ]

        # words are cumulative
        if timed_words:
            interim_text = " ".join(word for word in timed_words)
            start_time = timed_words[0].start_time or start_time
            end_time = timed_words[-1].end_time or end_time
            confidence = sum(word.confidence or 0.0 for word in timed_words) / len(timed_words)

            interim_event = stt.SpeechEvent(
                type=stt.SpeechEventType.INTERIM_TRANSCRIPT,
                alternatives=[
                    stt.SpeechData(
                        language=language,
                        text=interim_text,
                        start_time=start_time,
                        end_time=end_time,
                        words=timed_words,
                        confidence=confidence,
                        speaker_id=speaker_id,
                    )
                ],
            )
            self._event_ch.send_nowait(interim_event)
            logger.debug(
                "interim transcript session=%s end_of_turn_confidence=%s",
                self._session_id,
                end_of_turn_confidence,
            )

        if utterance:
            if self._last_preflight_start_time == 0.0:
                self._last_preflight_start_time = start_time

            # utterance is chunk based so we need to filter the words to
            # only include the ones that are part of the current utterance
            utterance_words = [
                word
                for word in timed_words
                if is_given(word.start_time) and word.start_time >= self._last_preflight_start_time
            ]
            utterance_confidence = sum(word.confidence or 0.0 for word in utterance_words) / max(
                len(utterance_words), 1
            )

            final_event = stt.SpeechEvent(
                type=stt.SpeechEventType.PREFLIGHT_TRANSCRIPT,
                alternatives=[
                    stt.SpeechData(
                        language=language,
                        text=utterance,
                        start_time=self._last_preflight_start_time,
                        end_time=end_time,
                        words=utterance_words,
                        confidence=utterance_confidence,
                        speaker_id=speaker_id,
                    )
                ],
            )
            self._event_ch.send_nowait(final_event)
            logger.debug(
                "preflight transcript session=%s end_of_turn_confidence=%s",
                self._session_id,
                end_of_turn_confidence,
            )
            self._last_preflight_start_time = end_time

        if end_of_turn and (
            not (is_given(self._opts.format_turns) and self._opts.format_turns) or turn_is_formatted
        ):
            final_event = stt.SpeechEvent(
                type=stt.SpeechEventType.FINAL_TRANSCRIPT,
                alternatives=[
                    stt.SpeechData(
                        language=language,
                        text=transcript,
                        start_time=start_time,
                        end_time=end_time,
                        words=timed_words,
                        confidence=confidence,
                        speaker_id=speaker_id,
                    )
                ],
            )
            self._event_ch.send_nowait(final_event)
            logger.debug(
                "final transcript session=%s end_of_turn_confidence=%s",
                self._session_id,
                end_of_turn_confidence,
            )

            if words:
                first_word_start = words[0].get("start", 0)
                last_word_end = words[-1].get("end", 0)
                logger.debug(
                    "turn speech_duration=%.3fs session=%s (from word timestamps)",
                    (last_word_end - first_word_start) / 1000,
                    self._session_id,
                )

            self._event_ch.send_nowait(stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH))

            if self._speech_duration > 0.0:
                usage_event = stt.SpeechEvent(
                    type=stt.SpeechEventType.RECOGNITION_USAGE,
                    alternatives=[],
                    recognition_usage=stt.RecognitionUsage(audio_duration=self._speech_duration),
                )
                self._event_ch.send_nowait(usage_event)
                self._speech_duration = 0
                self._last_preflight_start_time = 0.0

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

Instance variables

prop expires_at : int | None
Expand source code
@property
def expires_at(self) -> int | None:
    """Unix timestamp when the AssemblyAI session expires. Set alongside session_id
    when the WebSocket connection is established."""
    return self._expires_at

Unix timestamp when the AssemblyAI session expires. Set alongside session_id when the WebSocket connection is established.

prop session_id : str | None
Expand source code
@property
def session_id(self) -> str | None:
    """The AssemblyAI session ID. Set when the WebSocket connection is established
    (before any speech events). None until the connection completes.
    Share this with the AssemblyAI team when reporting issues."""
    return self._session_id

The AssemblyAI session ID. Set when the WebSocket connection is established (before any speech events). None until the connection completes. Share this with the AssemblyAI team when reporting issues.

Methods

def force_endpoint(self) ‑> None
Expand source code
def force_endpoint(self) -> None:
    """Force-finalize the current turn immediately."""
    self._config_update_queue.put_nowait({"type": "ForceEndpoint"})

Force-finalize the current turn immediately.

def update_options(self,
*,
buffer_size_seconds: NotGivenOr[float] = NOT_GIVEN,
end_of_turn_confidence_threshold: NotGivenOr[float] = NOT_GIVEN,
min_turn_silence: NotGivenOr[int] = NOT_GIVEN,
max_turn_silence: NotGivenOr[int] = NOT_GIVEN,
prompt: NotGivenOr[str] = NOT_GIVEN,
agent_context: NotGivenOr[str] = NOT_GIVEN,
keyterms_prompt: NotGivenOr[list[str]] = NOT_GIVEN,
language_codes: NotGivenOr[str | list[str]] = NOT_GIVEN,
vad_threshold: NotGivenOr[float] = NOT_GIVEN,
continuous_partials: NotGivenOr[bool] = NOT_GIVEN,
interruption_delay: NotGivenOr[int] = NOT_GIVEN,
min_end_of_turn_silence_when_confident: NotGivenOr[int] = NOT_GIVEN) ‑> None
Expand source code
def update_options(
    self,
    *,
    buffer_size_seconds: NotGivenOr[float] = NOT_GIVEN,
    end_of_turn_confidence_threshold: NotGivenOr[float] = NOT_GIVEN,
    min_turn_silence: NotGivenOr[int] = NOT_GIVEN,
    max_turn_silence: NotGivenOr[int] = NOT_GIVEN,
    prompt: NotGivenOr[str] = NOT_GIVEN,
    agent_context: NotGivenOr[str] = NOT_GIVEN,
    keyterms_prompt: NotGivenOr[list[str]] = NOT_GIVEN,
    language_codes: NotGivenOr[str | list[str]] = NOT_GIVEN,
    vad_threshold: NotGivenOr[float] = NOT_GIVEN,
    continuous_partials: NotGivenOr[bool] = NOT_GIVEN,
    interruption_delay: NotGivenOr[int] = NOT_GIVEN,
    # Deprecated — use min_turn_silence instead
    min_end_of_turn_silence_when_confident: NotGivenOr[int] = NOT_GIVEN,
) -> None:
    if is_given(min_end_of_turn_silence_when_confident):
        logger.warning(
            "'min_end_of_turn_silence_when_confident' is deprecated, "
            "use 'min_turn_silence' instead."
        )
        if not is_given(min_turn_silence):
            min_turn_silence = min_end_of_turn_silence_when_confident

    # Validate/normalize before mutating any option so a ValueError from a
    # bad value cannot leave _opts partially updated.
    if is_given(language_codes):
        if self._opts.speech_model not in _U3_PRO_MODELS:
            raise ValueError(
                "The 'language_codes' parameter is only supported with the "
                f"{', '.join(_U3_PRO_MODELS)} models."
            )
        language_codes = _normalize_language_codes(language_codes)
    if is_given(agent_context):
        _validate_agent_context(agent_context)

    if is_given(buffer_size_seconds):
        self._opts.buffer_size_seconds = buffer_size_seconds
    if is_given(end_of_turn_confidence_threshold):
        self._opts.end_of_turn_confidence_threshold = end_of_turn_confidence_threshold
    if is_given(min_turn_silence):
        self._opts.min_turn_silence = min_turn_silence
    if is_given(max_turn_silence):
        self._opts.max_turn_silence = max_turn_silence
    if is_given(prompt):
        self._opts.prompt = prompt
    if is_given(agent_context):
        self._opts.agent_context = agent_context
    if is_given(keyterms_prompt):
        self._opts.keyterms_prompt = keyterms_prompt
    if is_given(language_codes):
        self._opts.language_codes = language_codes
    if is_given(vad_threshold):
        self._opts.vad_threshold = vad_threshold
    if is_given(continuous_partials):
        self._opts.continuous_partials = continuous_partials
    if is_given(interruption_delay):
        self._opts.interruption_delay = interruption_delay

    # Send UpdateConfiguration message over the active websocket
    config_msg: dict = {"type": "UpdateConfiguration"}
    if is_given(prompt):
        config_msg["prompt"] = prompt
    if is_given(agent_context):
        config_msg["agent_context"] = agent_context
    if is_given(keyterms_prompt):
        config_msg["keyterms_prompt"] = keyterms_prompt
    if is_given(language_codes):
        config_msg["language_codes"] = language_codes
    if is_given(max_turn_silence):
        config_msg["max_turn_silence"] = max_turn_silence
    if is_given(min_turn_silence):
        config_msg["min_turn_silence"] = min_turn_silence
    if is_given(end_of_turn_confidence_threshold):
        config_msg["end_of_turn_confidence_threshold"] = end_of_turn_confidence_threshold
    if is_given(continuous_partials):
        config_msg["continuous_partials"] = continuous_partials
    if is_given(interruption_delay):
        config_msg["interruption_delay"] = interruption_delay
    if is_given(vad_threshold):
        config_msg["vad_threshold"] = vad_threshold

    if len(config_msg) > 1:
        self._config_update_queue.put_nowait(config_msg)