Module livekit.plugins.assemblyai
AssemblyAI plugin for LiveKit Agents
Streaming speech-to-text and streaming text-to-speech from AssemblyAI.
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-6-pro']" = 'universal-3-6-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-6-pro", ] = "universal-3-6-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-6-pro' instead.") model = "universal-3-6-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_codesinstead (it accepts a single code directly). Mutually exclusive withlanguage_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 exceedsmin_turn_silencewithout ending the turn). Leave unset to use AssemblyAI's server defaults: enabled, except whenspeaker_labelsis 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_contextvalues) 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 viaupdate_options. agent_context_carryover- Deprecated, use
AgentSession(stt_context_options={"forward_chat_context": ...})instead. On the Universal-3 Pro family, assistant replies are carried intoagent_contextby default; passFalseto 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_silencestill 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_modelGet 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")) language_confidence = data.get("language_confidence") # 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 # Surface the server's per-turn scores on SpeechData.metadata, reachable # from the SpeechEvent stream (e.g. in an `stt_node` override) without # subclassing the stream. Note that UserInputTranscribedEvent does not # carry metadata, so session-level listeners won't see these. Each key is # attached only when the message carries it, so models that don't emit a # field are unaffected. # # end_of_turn_confidence: on Universal-3.5 Pro (and later) this rises from 0 # toward 1 across the partials emitted while a turn is held open between # min_turn_silence and max_turn_silence, letting callers threshold it to # trigger preemptive/eager LLM generation before the final arrives; it is 1.0 # on the final. # # language_confidence: sent when the `language_detection` connection param is # on (this plugin's default for the multilingual and U3 Pro models) and scores # the `language_code` on the same message. The confidence score for the detected # language, between 0 (low confidence) and 1 (high confidence). Only populated when # language detection is enabled and an utterance is complete or turn is final. turn_metadata: dict[str, Any] = {} if end_of_turn_confidence is not None: turn_metadata["end_of_turn_confidence"] = end_of_turn_confidence if language_confidence is not None: turn_metadata["language_confidence"] = language_confidence speech_metadata = turn_metadata or 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, metadata=speech_metadata, ) ], ) 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, metadata=speech_metadata, ) ], ) 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, metadata=speech_metadata, ) ], ) 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.0Helper 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_atUnix 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_idThe 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)
class SynthesizeStream (*,
tts: TTS,
conn_options: APIConnectOptions)-
Expand source code
class SynthesizeStream(tts.SynthesizeStream): def __init__(self, *, tts: TTS, conn_options: APIConnectOptions): super().__init__(tts=tts, conn_options=conn_options) self._tts: TTS = tts self._opts = replace(tts._opts) async def _run(self, output_emitter: tts.AudioEmitter) -> None: request_id = utils.shortuuid() output_emitter.initialize( request_id=request_id, sample_rate=self._opts.sample_rate, num_channels=1, mime_type="audio/pcm", stream=True, ) # Acquire before any text arrives, so the handshake overlaps the LLM's first token # (text waits in the input channel meanwhile). try: conn = await self._tts._acquire(timeout=self._conn_options.timeout) except APIError as e: self._tts._note_retry_after(e) raise except Exception as e: raise APIConnectionError("failed to acquire an AssemblyAI TTS connection") from e self._acquire_time = self._tts._pool.last_acquire_time self._connection_reused = self._tts._pool.last_connection_reused if conn.session_id: output_emitter._note_provider_request_id(conn.session_id) sent_stream = self._tts._sentence_tokenizer.stream() if self._tts._stream_pacer: sent_stream = self._tts._stream_pacer.wrap( sent_stream=sent_stream, audio_emitter=output_emitter ) async def _input_task() -> None: async for data in self._input_ch: if isinstance(data, self._FlushSentinel): sent_stream.flush() else: sent_stream.push_text(data) sent_stream.end_input() run = _StreamRun(conn=conn, listener=conn.attach(), emitter=output_emitter, stream=self) tasks = [ asyncio.create_task(_input_task()), asyncio.create_task(run.run(sent_stream)), ] discard = True try: await asyncio.gather(*tasks) discard = False except asyncio.CancelledError: # Interrupted (barge-in). Cancel whatever the server still has for us and # hand the socket back: frames it already sent for this stream are dropped # by the reader until `Cancelled` confirms the server moved on. if run.has_outstanding: with contextlib.suppress(APIError): conn.cancel() discard = False raise except APIError as e: self._tts._note_retry_after(e) raise except Exception as e: raise APIConnectionError( f"AssemblyAI TTS stream failed (session {conn.session_id})" ) from e finally: await utils.aio.gracefully_cancel(*tasks) await sent_stream.aclose() conn.detach(run.listener) self._tts._release(conn, discard=discard)Helper class that provides a standard way to create an ABC using inheritance.
Ancestors
- livekit.agents.tts.tts.SynthesizeStream
- abc.ABC
class TTS (*,
voice: str = '987db62e-a2c2-569a-9c4d-82460cdea1bd',
language: NotGivenOr[str] = NOT_GIVEN,
model: str = 'assemblyai-tts-1',
sample_rate: int = 24000,
word_timestamps: bool = True,
api_key: NotGivenOr[str] = NOT_GIVEN,
base_url: str = 'wss://tts.us.assemblyai.com',
tokenizer: NotGivenOr[tokenize.SentenceTokenizer] = NOT_GIVEN,
text_pacing: tts.SentenceStreamPacer | bool = False,
http_session: aiohttp.ClientSession | None = None)-
Expand source code
class TTS(tts.TTS): def __init__( self, *, voice: str = DEFAULT_VOICE, language: NotGivenOr[str] = NOT_GIVEN, model: str = DEFAULT_MODEL, sample_rate: int = 24000, word_timestamps: bool = True, api_key: NotGivenOr[str] = NOT_GIVEN, base_url: str = DEFAULT_BASE_URL, tokenizer: NotGivenOr[tokenize.SentenceTokenizer] = NOT_GIVEN, text_pacing: tts.SentenceStreamPacer | bool = False, http_session: aiohttp.ClientSession | None = None, ) -> None: """Create a new instance of AssemblyAI Streaming TTS. See https://www.assemblyai.com/docs/tts/getting-started/quickstart for more information on the API. Args: voice: The voice, by its catalog ``id`` (a UUID): a voice from the library (see https://www.assemblyai.com/docs/tts/voices) or a custom (cloned) voice. A slug such as ``"aai-linda-3"`` also works, but slugs can change and ids do not. Defaults to Keegan. The voice must be paired with ``model``; a custom voice can only be used on a host in the data zone that created it. language: The language to synthesize, as a code (``"es"``, ``"es-MX"``) or an English name (``"spanish"``). Defaults to English; with a custom voice, to the language of the voice's reference clip. model: The model to synthesize with, by name or alias. Defaults to ``"assemblyai-tts-1"``. sample_rate: Output sample rate in Hz, one of 8000, 16000, 22050, 24000, 44100 or 48000. Defaults to 24000, the service's native rate. word_timestamps: Request per-word timings and publish them as an aligned transcript. Defaults to True. api_key: AssemblyAI API key. Defaults to the ``ASSEMBLYAI_API_KEY`` environment variable. base_url: The Streaming TTS host. Defaults to the US data zone; use ``"wss://tts.eu.assemblyai.com"`` to keep text and audio in the EU. tokenizer: Splits the LLM output into the requests sent to the server; each sentence is synthesized as its own request. Defaults to ``livekit.agents.tokenize.blingfire.SentenceTokenizer``. text_pacing: Stream pacer for the TTS. Set to True to use the default pacer, False to disable. http_session: An existing aiohttp ClientSession to use. """ super().__init__( capabilities=tts.TTSCapabilities( streaming=True, aligned_transcript=word_timestamps, ), sample_rate=sample_rate, num_channels=1, ) if sample_rate not in SUPPORTED_SAMPLE_RATES: raise ValueError( f"sample_rate must be one of {SUPPORTED_SAMPLE_RATES}, but got {sample_rate}" ) 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._opts = _TTSOptions( voice=_require(voice, "voice"), language=_language_name(language) if is_given(language) else None, model=_require(model, "model"), sample_rate=sample_rate, word_timestamps=word_timestamps, api_key=assemblyai_api_key, base_url=base_url.rstrip("/"), ) self._session = http_session self._streams = weakref.WeakSet[SynthesizeStream]() # monotonic time before which no session is opened, see _note_retry_after() self._connect_not_before = 0.0 self._sentence_tokenizer = ( tokenizer if is_given(tokenizer) else tokenize.blingfire.SentenceTokenizer() ) self._stream_pacer: tts.SentenceStreamPacer | None = None if text_pacing is True: self._stream_pacer = tts.SentenceStreamPacer() elif isinstance(text_pacing, tts.SentenceStreamPacer): self._stream_pacer = text_pacing # Session lifetime is enforced from each socket's own `expires_at`, see _acquire(). self._pool = utils.ConnectionPool[_Connection]( connect_cb=self._connect, close_cb=self._close, ) @property def model(self) -> str: return self._opts.model @property def provider(self) -> str: return "AssemblyAI" def update_options( self, *, voice: NotGivenOr[str] = NOT_GIVEN, language: NotGivenOr[str | None] = NOT_GIVEN, model: NotGivenOr[str] = NOT_GIVEN, ) -> None: """Change the voice, language or model of the sessions opened from now on. The configuration is fixed when a session opens, so pooled sockets are retired: every stream that has not connected yet, including one already created, opens a new session with these options. A stream already speaking keeps its voice until it finishes. Args: voice: A catalog voice id or slug. language: A language code or English name, or None for the server's default. model: A model name or alias. """ opts = self._opts new_voice, new_language, new_model = opts.voice, opts.language, opts.model if is_given(voice): new_voice = _require(voice, "voice") if is_given(language): new_language = _language_name(language) if language is not None else None if is_given(model): new_model = _require(model, "model") new = (new_voice, new_language, new_model) if new != (opts.voice, opts.language, opts.model): opts.voice, opts.language, opts.model = new self._pool.invalidate() def synthesize( self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS ) -> tts.ChunkedStream: # a one-shot synthesis runs over the pooled socket too, skipping a handshake return self._synthesize_with_stream(text, conn_options=conn_options) def stream( self, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS ) -> SynthesizeStream: stream = SynthesizeStream(tts=self, conn_options=conn_options) self._streams.add(stream) return stream def prewarm(self) -> None: self._pool.prewarm() async def aclose(self) -> None: for stream in list(self._streams): await stream.aclose() self._streams.clear() await self._pool.aclose() def _ensure_session(self) -> aiohttp.ClientSession: if not self._session: self._session = utils.http_context.http_session() return self._session def _note_retry_after(self, error: APIError) -> None: """Hold off new sessions for as long as the server asked (`retry_after_seconds`).""" body = error.body if isinstance(error, APIStatusError) else None retry_after = body.get("retry_after_seconds") if isinstance(body, dict) else None if isinstance(retry_after, (int, float)) and retry_after > 0: self._connect_not_before = max(self._connect_not_before, time.monotonic() + retry_after) async def _connect(self, timeout: float) -> _Connection: # Every refused reconnect still counts against the per-minute session allowance, # so retrying sooner than asked can turn a transient error into a 3009. if (wait := self._connect_not_before - time.monotonic()) > 0: if wait > timeout: # longer than a reply can wait: stop the retries (a FallbackAdapter moves on) raise APIConnectionError( f"AssemblyAI TTS asked to wait {wait:.1f}s before opening a session", retryable=False, ) await asyncio.sleep(wait) opts = self._opts params: dict[str, Any] = { "model_id": opts.model, "voice_id": opts.voice, "sample_rate": opts.sample_rate, "encoding": "pcm_s16le", "word_boundaries": "true" if opts.word_timestamps else "false", } if opts.language is not None: # left out, the server speaks the voice's own language params["language"] = opts.language url = f"{opts.base_url}/v1/ws?{urlencode(params)}" try: ws = await asyncio.wait_for( # the key goes in raw: the service rejects "Bearer <key>" as an invalid key self._ensure_session().ws_connect(url, headers={"Authorization": opts.api_key}), timeout, ) except asyncio.TimeoutError: raise APITimeoutError("timed out connecting to AssemblyAI TTS") from None except aiohttp.ClientResponseError as e: # RequestInfo carries the request headers, including the API key. raise APIStatusError( message=e.message, status_code=e.status, request_id=None, body=None ) from None except Exception as e: raise APIConnectionError("failed to connect to AssemblyAI TTS") from e # credentials and parameters are checked after the upgrade: a failure arrives as # an Error frame in place of Begin try: try: msg = await ws.receive(timeout=timeout) except asyncio.TimeoutError: raise APITimeoutError("timed out waiting for AssemblyAI TTS Begin") from None begin = _parse_frame(msg) if begin is None: raise APIStatusError( "AssemblyAI TTS connection closed before Begin", status_code=ws.close_code or -1, retryable=True, ) if begin.get("type") == "Error": error = _error_from_frame(begin, session_id=None) self._note_retry_after(error) raise error if begin.get("type") != "Begin": raise APIConnectionError( f"expected Begin from AssemblyAI TTS, got {begin.get('type')!r}" ) except BaseException: await ws.close() raise conn = _Connection(ws, begin) logger.debug( "connected to AssemblyAI TTS", extra={"session_id": conn.session_id, "configuration": conn.configuration}, ) if opts.word_timestamps and not conn.word_boundaries: logger.warning( "AssemblyAI TTS cannot supply word timings for this session, " "the transcript will be paced per sentence", extra={"session_id": conn.session_id}, ) return conn async def _close(self, conn: _Connection) -> None: await conn.aclose() async def _acquire(self, *, timeout: float) -> _Connection: while True: conn = await self._pool.get(timeout=timeout) if conn.usable: return conn self._pool.remove(conn) if conn.error is not None: self._note_retry_after(conn.error) if not self._pool.last_connection_reused: # unusable from the start: another session would only spend a session start raise conn.error or APIConnectionError( f"AssemblyAI TTS session {conn.session_id} was unusable as soon as it opened" ) # closed by the server while idle, or too close to its expires_at def _release(self, conn: _Connection, *, discard: bool) -> None: if discard or not conn.usable: self._pool.remove(conn) else: self._pool.put(conn)Helper class that provides a standard way to create an ABC using inheritance.
Create a new instance of AssemblyAI Streaming TTS.
See https://www.assemblyai.com/docs/tts/getting-started/quickstart for more information on the API.
Args
voice- The voice, by its catalog
id(a UUID): a voice from the library (see https://www.assemblyai.com/docs/tts/voices) or a custom (cloned) voice. A slug such as"aai-linda-3"also works, but slugs can change and ids do not. Defaults to Keegan. The voice must be paired withmodel; a custom voice can only be used on a host in the data zone that created it. language- The language to synthesize, as a code (
"es","es-MX") or an English name ("spanish"). Defaults to English; with a custom voice, to the language of the voice's reference clip. model- The model to synthesize with, by name or alias. Defaults to
"assemblyai-tts-1". sample_rate- Output sample rate in Hz, one of 8000, 16000, 22050, 24000, 44100 or 48000. Defaults to 24000, the service's native rate.
word_timestamps- Request per-word timings and publish them as an aligned transcript. Defaults to True.
api_key- AssemblyAI API key. Defaults to the
ASSEMBLYAI_API_KEYenvironment variable. base_url- The Streaming TTS host. Defaults to the US data zone; use
"wss://tts.eu.assemblyai.com"to keep text and audio in the EU. tokenizer- Splits the LLM output into the requests sent to the server; each
sentence is synthesized as its own request. Defaults to
SentenceTokenizer. text_pacing- Stream pacer for the TTS. Set to True to use the default pacer, False to disable.
http_session- An existing aiohttp ClientSession to use.
Ancestors
- livekit.agents.tts.tts.TTS
- abc.ABC
- EventEmitter
- typing.Generic
Instance variables
prop model : str-
Expand source code
@property def model(self) -> str: return self._opts.modelGet the model name/identifier for this TTS instance.
Returns
The model name if available, "unknown" otherwise.
Note
Plugins should override this property to provide their model information.
prop provider : str-
Expand source code
@property def provider(self) -> str: return "AssemblyAI"Get the provider name/identifier for this TTS instance.
Returns
The provider name if available, "unknown" otherwise.
Note
Plugins should override this property to provide their provider information.
Methods
async def aclose(self) ‑> None-
Expand source code
async def aclose(self) -> None: for stream in list(self._streams): await stream.aclose() self._streams.clear() await self._pool.aclose() def prewarm(self) ‑> None-
Expand source code
def prewarm(self) -> None: self._pool.prewarm()Pre-warm connection to the TTS service
def stream(self,
*,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> livekit.plugins.assemblyai.tts.SynthesizeStream-
Expand source code
def stream( self, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS ) -> SynthesizeStream: stream = SynthesizeStream(tts=self, conn_options=conn_options) self._streams.add(stream) return stream def synthesize(self,
text: str,
*,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> livekit.agents.tts.tts.ChunkedStream-
Expand source code
def synthesize( self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS ) -> tts.ChunkedStream: # a one-shot synthesis runs over the pooled socket too, skipping a handshake return self._synthesize_with_stream(text, conn_options=conn_options) def update_options(self,
*,
voice: NotGivenOr[str] = NOT_GIVEN,
language: NotGivenOr[str | None] = NOT_GIVEN,
model: NotGivenOr[str] = NOT_GIVEN) ‑> None-
Expand source code
def update_options( self, *, voice: NotGivenOr[str] = NOT_GIVEN, language: NotGivenOr[str | None] = NOT_GIVEN, model: NotGivenOr[str] = NOT_GIVEN, ) -> None: """Change the voice, language or model of the sessions opened from now on. The configuration is fixed when a session opens, so pooled sockets are retired: every stream that has not connected yet, including one already created, opens a new session with these options. A stream already speaking keeps its voice until it finishes. Args: voice: A catalog voice id or slug. language: A language code or English name, or None for the server's default. model: A model name or alias. """ opts = self._opts new_voice, new_language, new_model = opts.voice, opts.language, opts.model if is_given(voice): new_voice = _require(voice, "voice") if is_given(language): new_language = _language_name(language) if language is not None else None if is_given(model): new_model = _require(model, "model") new = (new_voice, new_language, new_model) if new != (opts.voice, opts.language, opts.model): opts.voice, opts.language, opts.model = new self._pool.invalidate()Change the voice, language or model of the sessions opened from now on.
The configuration is fixed when a session opens, so pooled sockets are retired: every stream that has not connected yet, including one already created, opens a new session with these options. A stream already speaking keeps its voice until it finishes.
Args
voice- A catalog voice id or slug.
language- A language code or English name, or None for the server's default.
model- A model name or alias.
Inherited members