Module livekit.plugins.elevenlabs
ElevenLabs plugin for LiveKit Agents
See https://docs.livekit.io/agents/integrations/tts/elevenlabs/ for more information.
Classes
class PronunciationDictionaryLocator (pronunciation_dictionary_id: str, version_id: str)-
Expand source code
@dataclass class PronunciationDictionaryLocator: pronunciation_dictionary_id: str version_id: strPronunciationDictionaryLocator(pronunciation_dictionary_id: 'str', version_id: 'str')
Instance variables
var pronunciation_dictionary_id : strvar version_id : str
class STT (*,
api_key: NotGivenOr[str] = NOT_GIVEN,
base_url: NotGivenOr[str] = NOT_GIVEN,
language_code: NotGivenOr[str] = NOT_GIVEN,
secondary_languages: NotGivenOr[list[str]] = NOT_GIVEN,
include_language_detection: NotGivenOr[bool] = NOT_GIVEN,
tag_audio_events: bool = True,
use_realtime: NotGivenOr[bool] = NOT_GIVEN,
sample_rate: STTRealtimeSampleRates = 16000,
audio_chunk_duration_ms: int = 50,
server_vad: NotGivenOr[VADOptions] = NOT_GIVEN,
include_timestamps: bool = False,
http_session: aiohttp.ClientSession | None = None,
model: NotGivenOr[ElevenLabsSTTModels | str] = NOT_GIVEN,
model_id: NotGivenOr[ElevenLabsSTTModels | str] = NOT_GIVEN,
keyterms: NotGivenOr[list[str]] = NOT_GIVEN,
no_verbatim: NotGivenOr[bool] = NOT_GIVEN,
enable_logging: bool = True,
previous_text: NotGivenOr[str] = NOT_GIVEN)-
Expand source code
class STT(stt.STT): def __init__( self, *, api_key: NotGivenOr[str] = NOT_GIVEN, base_url: NotGivenOr[str] = NOT_GIVEN, language_code: NotGivenOr[str] = NOT_GIVEN, secondary_languages: NotGivenOr[list[str]] = NOT_GIVEN, include_language_detection: NotGivenOr[bool] = NOT_GIVEN, tag_audio_events: bool = True, use_realtime: NotGivenOr[bool] = NOT_GIVEN, # Deprecated sample_rate: STTRealtimeSampleRates = 16000, audio_chunk_duration_ms: int = 50, server_vad: NotGivenOr[VADOptions] = NOT_GIVEN, include_timestamps: bool = False, http_session: aiohttp.ClientSession | None = None, model: NotGivenOr[ElevenLabsSTTModels | str] = NOT_GIVEN, model_id: NotGivenOr[ElevenLabsSTTModels | str] = NOT_GIVEN, # Deprecated keyterms: NotGivenOr[list[str]] = NOT_GIVEN, no_verbatim: NotGivenOr[bool] = NOT_GIVEN, enable_logging: bool = True, previous_text: NotGivenOr[str] = NOT_GIVEN, ) -> None: """ Create a new instance of ElevenLabs STT. Args: api_key (NotGivenOr[str]): ElevenLabs API key. Can be set via argument or `ELEVEN_API_KEY` environment variable. base_url (NotGivenOr[str]): Custom base URL for the API. Optional. language_code (NotGivenOr[str]): Language code for the STT model. Optional. secondary_languages (NotGivenOr[list[str]]): Additional languages that may be spoken in the audio, on top of `language_code`. Keeps the primary language hinted while still recognizing the others, which is what a code-switching speaker needs. Names and ISO-639-3 codes are accepted and normalized to what the API takes. Only supported for Scribe v2 realtime. include_language_detection (NotGivenOr[bool]): Whether the committed transcript reports the language the model actually heard. Defaults to True when no `language_code` is set and False otherwise. Turning it off while no `language_code` is set leaves the plugin with no language to report and every transcript falls back to "en". Only supported for Scribe v2 realtime. tag_audio_events (bool): Whether to tag audio events like (laughter), (footsteps), etc. in the transcription. Only supported for Scribe v1 model. Default is True. use_realtime (bool): Whether to use "scribe_v2_realtime" model for streaming mode. Default is NOT_GIVEN. Note that this flag is deprecated in favour of explicitly specifying the model id. sample_rate (STTRealtimeSampleRates): Audio sample rate in Hz. Default is 16000. audio_chunk_duration_ms (int): Duration of each outgoing realtime audio chunk in milliseconds. Must be a positive integer. Defaults to 50. Larger chunks reduce message frequency but increase buffering latency. Flushes send any shorter remaining chunk before committing. Only used for Scribe v2 realtime. server_vad (NotGivenOr[VADOptions]): Server-side VAD options, only supported for Scribe v2 realtime model. http_session (aiohttp.ClientSession | None): Custom HTTP session for API requests. Optional. model (ElevenLabsSTTModels | str): ElevenLabs STT model to use. If not specified a default model will be selected based on parameters provided. model_id (ElevenLabsSTTModels | str): Deprecated alias for `model`. Use `model` instead. keyterms (NotGivenOr[list[str]]): A list of keywords or phrases to bias the transcription towards. Supported for both Scribe v2 (batch) and Scribe v2 realtime. Batch accepts up to 1000 keyterms of at most 50 characters each; realtime accepts up to 50 keyterms of at most 20 characters each. Usage incurs additional costs. no_verbatim (NotGivenOr[bool]): When True, the model removes filler words, false starts and disfluencies from the transcript, producing cleaner output. Supported for both Scribe v2 (batch) and Scribe v2 realtime. Default is False. enable_logging (bool): Enable logging of the request. When set to false, zero retention mode will be used. Defaults to True. previous_text (NotGivenOr[str]): Preceding text context sent once on the first realtime audio chunk to improve transcription accuracy. Only supported for Scribe v2 realtime. """ if ( isinstance(audio_chunk_duration_ms, bool) or not isinstance(audio_chunk_duration_ms, int) or audio_chunk_duration_ms <= 0 ): raise ValueError("audio_chunk_duration_ms must be a positive integer") if is_given(model_id): if is_given(model): logger.warning( "both `model` and `model_id` parameters are provided. `model_id` will be ignored." ) else: logger.warning("`model_id` parameter is deprecated, use `model` instead.") model = model_id if is_given(use_realtime): if is_given(model): logger.warning( "both `use_realtime` and `model` parameters are provided. `use_realtime` will be ignored." ) else: logger.warning( "`use_realtime` parameter is deprecated. " "Specify a realtime model to enable streaming. " "Defaulting model to one based on use_realtime parameter. " ) model = "scribe_v2_realtime" if use_realtime else "scribe_v1" model = model if is_given(model) else "scribe_v1" use_realtime = model == "scribe_v2_realtime" if not use_realtime and is_given(server_vad): logger.warning("Server-side VAD is only supported for Scribe v2 realtime model") if not use_realtime and is_given(secondary_languages): logger.warning( "`secondary_languages` is only supported for Scribe v2 realtime model " "and will be ignored" ) secondary_languages = NOT_GIVEN if not use_realtime and is_given(include_language_detection): logger.warning( "`include_language_detection` is only supported for Scribe v2 realtime model " "and will be ignored" ) include_language_detection = NOT_GIVEN resolved_previous_text = previous_text if is_given(previous_text) else None if not use_realtime and resolved_previous_text is not None: logger.warning( "`previous_text` is only supported for Scribe v2 realtime model and will be ignored" ) resolved_previous_text = None super().__init__( capabilities=STTCapabilities( streaming=use_realtime, interim_results=True, aligned_transcript="word" if include_timestamps and use_realtime else False, ) ) elevenlabs_api_key = api_key if is_given(api_key) else os.environ.get("ELEVEN_API_KEY") if not elevenlabs_api_key: raise ValueError( "ElevenLabs API key is required, either as argument or " "set ELEVEN_API_KEY environmental variable" ) self._opts = STTOptions( api_key=elevenlabs_api_key, base_url=base_url if is_given(base_url) else API_BASE_URL_V1, language_code=LanguageCode(language_code) if language_code else None, secondary_languages=[LanguageCode(language) for language in secondary_languages] if is_given(secondary_languages) else NOT_GIVEN, include_language_detection=include_language_detection, tag_audio_events=tag_audio_events, sample_rate=sample_rate, audio_chunk_duration_ms=audio_chunk_duration_ms, server_vad=server_vad, include_timestamps=include_timestamps, model_id=model, keyterms=keyterms, no_verbatim=no_verbatim if is_given(no_verbatim) else False, enable_logging=enable_logging, previous_text=resolved_previous_text, ) self._session = http_session self._streams = weakref.WeakSet[SpeechStream]() @property def model(self) -> str: return self._opts.model_id @property def provider(self) -> str: return "ElevenLabs" def _ensure_session(self) -> aiohttp.ClientSession: if not self._session: self._session = http_context.http_session() return self._session async def _recognize_impl( self, buffer: AudioBuffer, *, language: NotGivenOr[str] = NOT_GIVEN, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS, ) -> stt.SpeechEvent: if is_given(language): self._opts.language_code = LanguageCode(language) wav_bytes = rtc.combine_audio_frames(buffer).to_wav_bytes() form = aiohttp.FormData() form.add_field("file", wav_bytes, filename="audio.wav", content_type="audio/x-wav") form.add_field("model_id", self._opts.model_id) form.add_field("tag_audio_events", str(self._opts.tag_audio_events).lower()) if self._opts.language_code: form.add_field("language_code", self._opts.language_code) if is_given(self._opts.keyterms): for keyterm in self._opts.keyterms: form.add_field("keyterms", keyterm) if self._opts.no_verbatim: form.add_field("no_verbatim", "true") try: async with self._ensure_session().post( _synthesize_url(self._opts), data=form, headers={AUTHORIZATION_HEADER: self._opts.api_key}, ) as response: response_json = await response.json() if response.status != 200: raise APIStatusError( message=response_json.get("detail", "Unknown ElevenLabs error"), status_code=response.status, request_id=trace_id_from_headers(response.headers), body=response_json, ) extracted_text = response_json.get("text") language_code = response_json.get("language_code") speaker_id = None start_time, end_time = 0, 0 words = response_json.get("words") if words: speaker_id = words[0].get("speaker_id", None) start_time = min(w.get("start", 0) for w in words) end_time = max(w.get("end", 0) for w in words) except asyncio.TimeoutError as e: raise APITimeoutError() from e except aiohttp.ClientResponseError as e: raise APIStatusError( message=e.message, status_code=e.status, request_id=trace_id_from_headers(e.headers), body=None, ) from e except Exception as e: raise APIConnectionError() from e normalized_language = LanguageCode(language_code or self._opts.language_code or "") return self._transcription_to_speech_event( language_code=normalized_language, text=extracted_text, start_time=start_time, end_time=end_time, speaker_id=speaker_id, words=words, ) def _transcription_to_speech_event( self, language_code: str, text: str, start_time: float, end_time: float, speaker_id: str | None, words: list[dict[str, Any]] | None = None, ) -> stt.SpeechEvent: return stt.SpeechEvent( type=SpeechEventType.FINAL_TRANSCRIPT, alternatives=[ stt.SpeechData( text=text, language=LanguageCode(language_code), speaker_id=speaker_id, start_time=start_time, end_time=end_time, confidence=_speech_confidence(words), words=[ TimedString( text=word.get("text", ""), start_time=word.get("start", 0), end_time=word.get("end", 0), ) for word in words ] if words else None, ) ], ) def update_options( self, *, tag_audio_events: NotGivenOr[bool] = NOT_GIVEN, server_vad: NotGivenOr[VADOptions] = NOT_GIVEN, keyterms: NotGivenOr[list[str]] = NOT_GIVEN, secondary_languages: NotGivenOr[list[str]] = NOT_GIVEN, no_verbatim: NotGivenOr[bool] = NOT_GIVEN, ) -> None: if is_given(tag_audio_events): self._opts.tag_audio_events = tag_audio_events if is_given(server_vad): self._opts.server_vad = server_vad if is_given(keyterms): self._opts.keyterms = keyterms if is_given(secondary_languages): if self._opts.model_id == "scribe_v2_realtime": self._opts.secondary_languages = [ LanguageCode(language) for language in secondary_languages ] else: logger.warning( "`secondary_languages` is only supported for Scribe v2 realtime model " "and will be ignored" ) secondary_languages = NOT_GIVEN if is_given(no_verbatim): self._opts.no_verbatim = no_verbatim for stream in self._streams: stream.update_options( server_vad=server_vad, no_verbatim=no_verbatim, keyterms=keyterms, secondary_languages=secondary_languages, ) def stream( self, *, language: NotGivenOr[str] = NOT_GIVEN, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS, ) -> SpeechStream: stream = SpeechStream( stt=self, opts=self._opts, conn_options=conn_options, language=LanguageCode(language) if is_given(language) else self._opts.language_code, http_session=self._ensure_session(), ) self._streams.add(stream) return streamHelper class that provides a standard way to create an ABC using inheritance.
Create a new instance of ElevenLabs STT.
Args
api_key:NotGivenOr[str]- ElevenLabs API key. Can be set via argument or
ELEVEN_API_KEYenvironment variable. base_url:NotGivenOr[str]- Custom base URL for the API. Optional.
language_code:NotGivenOr[str]- Language code for the STT model. Optional.
secondary_languages:NotGivenOr[list[str]]- Additional languages that may be spoken in
the audio, on top of
language_code. Keeps the primary language hinted while still recognizing the others, which is what a code-switching speaker needs. Names and ISO-639-3 codes are accepted and normalized to what the API takes. Only supported for Scribe v2 realtime. include_language_detection:NotGivenOr[bool]- Whether the committed transcript reports
the language the model actually heard. Defaults to True when no
language_codeis set and False otherwise. Turning it off while nolanguage_codeis set leaves the plugin with no language to report and every transcript falls back to "en". Only supported for Scribe v2 realtime. tag_audio_events:bool- Whether to tag audio events like (laughter), (footsteps), etc. in the transcription. Only supported for Scribe v1 model. Default is True.
use_realtime:bool- Whether to use "scribe_v2_realtime" model for streaming mode. Default is NOT_GIVEN. Note that this flag is deprecated in favour of explicitly specifying the model id.
sample_rate:STTRealtimeSampleRates- Audio sample rate in Hz. Default is 16000.
audio_chunk_duration_ms:int- Duration of each outgoing realtime audio chunk in milliseconds. Must be a positive integer. Defaults to 50. Larger chunks reduce message frequency but increase buffering latency. Flushes send any shorter remaining chunk before committing. Only used for Scribe v2 realtime.
server_vad:NotGivenOr[VADOptions]- Server-side VAD options, only supported for Scribe v2 realtime model.
http_session:aiohttp.ClientSession | None- Custom HTTP session for API requests. Optional.
model:ElevenLabsSTTModels | str- ElevenLabs STT model to use. If not specified a default model will be selected based on parameters provided.
model_id:ElevenLabsSTTModels | str- Deprecated alias for
model. Usemodelinstead. keyterms:NotGivenOr[list[str]]- A list of keywords or phrases to bias the transcription towards. Supported for both Scribe v2 (batch) and Scribe v2 realtime. Batch accepts up to 1000 keyterms of at most 50 characters each; realtime accepts up to 50 keyterms of at most 20 characters each. Usage incurs additional costs.
no_verbatim:NotGivenOr[bool]- When True, the model removes filler words, false starts and disfluencies from the transcript, producing cleaner output. Supported for both Scribe v2 (batch) and Scribe v2 realtime. Default is False.
enable_logging:bool- Enable logging of the request. When set to false, zero retention mode will be used. Defaults to True.
previous_text:NotGivenOr[str]- Preceding text context sent once on the first realtime audio chunk to improve transcription accuracy. Only supported for Scribe v2 realtime.
Ancestors
- livekit.agents.stt.stt.STT
- abc.ABC
- EventEmitter
- typing.Generic
Instance variables
prop model : str-
Expand source code
@property def model(self) -> str: return self._opts.model_idGet 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 "ElevenLabs"Get the provider name/identifier for this STT instance.
Returns
The provider name if available, "unknown" otherwise.
Note
Plugins should override this property to provide their provider information.
Methods
def stream(self,
*,
language: NotGivenOr[str] = NOT_GIVEN,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> livekit.plugins.elevenlabs.stt.SpeechStream-
Expand source code
def stream( self, *, language: NotGivenOr[str] = NOT_GIVEN, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS, ) -> SpeechStream: stream = SpeechStream( stt=self, opts=self._opts, conn_options=conn_options, language=LanguageCode(language) if is_given(language) else self._opts.language_code, http_session=self._ensure_session(), ) self._streams.add(stream) return stream def update_options(self,
*,
tag_audio_events: NotGivenOr[bool] = NOT_GIVEN,
server_vad: NotGivenOr[VADOptions] = NOT_GIVEN,
keyterms: NotGivenOr[list[str]] = NOT_GIVEN,
secondary_languages: NotGivenOr[list[str]] = NOT_GIVEN,
no_verbatim: NotGivenOr[bool] = NOT_GIVEN) ‑> None-
Expand source code
def update_options( self, *, tag_audio_events: NotGivenOr[bool] = NOT_GIVEN, server_vad: NotGivenOr[VADOptions] = NOT_GIVEN, keyterms: NotGivenOr[list[str]] = NOT_GIVEN, secondary_languages: NotGivenOr[list[str]] = NOT_GIVEN, no_verbatim: NotGivenOr[bool] = NOT_GIVEN, ) -> None: if is_given(tag_audio_events): self._opts.tag_audio_events = tag_audio_events if is_given(server_vad): self._opts.server_vad = server_vad if is_given(keyterms): self._opts.keyterms = keyterms if is_given(secondary_languages): if self._opts.model_id == "scribe_v2_realtime": self._opts.secondary_languages = [ LanguageCode(language) for language in secondary_languages ] else: logger.warning( "`secondary_languages` is only supported for Scribe v2 realtime model " "and will be ignored" ) secondary_languages = NOT_GIVEN if is_given(no_verbatim): self._opts.no_verbatim = no_verbatim for stream in self._streams: stream.update_options( server_vad=server_vad, no_verbatim=no_verbatim, keyterms=keyterms, secondary_languages=secondary_languages, )
Inherited members
class SpeechStream (*,
stt: STT,
opts: STTOptions,
conn_options: APIConnectOptions,
language: LanguageCode | None,
http_session: aiohttp.ClientSession)-
Expand source code
class SpeechStream(stt.SpeechStream): """Streaming speech recognition using ElevenLabs Scribe v2 realtime API""" def __init__( self, *, stt: STT, opts: STTOptions, conn_options: APIConnectOptions, language: LanguageCode | None, http_session: aiohttp.ClientSession, ) -> None: super().__init__(stt=stt, conn_options=conn_options, sample_rate=opts.sample_rate) self._opts = opts self._language = language self._session = http_session self._reconnect_event = asyncio.Event() self._speaking = False # Track if we're currently in a speech segment self._last_partial_text = "" self._audio_duration_collector = PeriodicCollector( callback=self._on_audio_duration_report, duration=5.0, ) def update_options( self, *, server_vad: NotGivenOr[VADOptions] = NOT_GIVEN, no_verbatim: NotGivenOr[bool] = NOT_GIVEN, keyterms: NotGivenOr[list[str]] = NOT_GIVEN, secondary_languages: NotGivenOr[list[str]] = NOT_GIVEN, ) -> None: if is_given(server_vad): self._opts.server_vad = server_vad self._reconnect_event.set() if is_given(no_verbatim): self._opts.no_verbatim = no_verbatim self._reconnect_event.set() if is_given(keyterms): self._opts.keyterms = keyterms self._reconnect_event.set() if is_given(secondary_languages): self._opts.secondary_languages = [ LanguageCode(language) for language in secondary_languages ] self._reconnect_event.set() def _on_audio_duration_report(self, duration: float) -> None: usage_event = stt.SpeechEvent( type=stt.SpeechEventType.RECOGNITION_USAGE, alternatives=[], recognition_usage=stt.RecognitionUsage(audio_duration=duration), ) self._event_ch.send_nowait(usage_event) @property def _server_vad(self) -> VADOptions | None: return self._opts.server_vad if is_given(self._opts.server_vad) else None @property def _language_detection(self) -> bool: """Whether the session reports the language the model actually heard. Defaults to on when no language was pinned, which is the only case where the plugin used to request it.""" if is_given(self._opts.include_language_detection): return self._opts.include_language_detection return not self._language @property def _final_message_type(self) -> str: """The committed message this session treats as the final transcript. ElevenLabs sends every commit twice and puts the word timestamps and the detected language on the delayed copy only, so that copy is the final one whenever either is asked for.""" if self._opts.include_timestamps or self._language_detection: return "committed_transcript_with_timestamps" return "committed_transcript" async def _run(self) -> None: """Run the streaming transcription session""" closing_ws = False async def keepalive_task(ws: aiohttp.ClientWebSocketResponse) -> None: try: while True: # scribe_v2_realtime model requires a keepalive message instead of a ping await asyncio.sleep(10) await ws.send_str( json.dumps( { "message_type": "input_audio_chunk", "audio_base_64": "", "commit": False, "sample_rate": self._opts.sample_rate, } ) ) except Exception: return @utils.log_exceptions(logger=logger) async def send_task(ws: aiohttp.ClientWebSocketResponse) -> None: nonlocal closing_ws # Buffer audio into chunks of the configured duration. samples_per_chunk = self._opts.sample_rate * self._opts.audio_chunk_duration_ms // 1000 audio_bstream = utils.audio.AudioByteStream( sample_rate=self._opts.sample_rate, num_channels=1, samples_per_channel=samples_per_chunk, ) has_ended = False try: async for data in self._input_ch: # Write audio bytes to the buffer and get complete chunks frames: list[rtc.AudioFrame] = [] if isinstance(data, rtc.AudioFrame): frames.extend(audio_bstream.write(data.data.tobytes())) elif isinstance(data, self._FlushSentinel): frames.extend(audio_bstream.flush()) has_ended = True for frame in frames: self._audio_duration_collector.push(frame.duration) audio_b64 = base64.b64encode(frame.data.tobytes()).decode("utf-8") await ws.send_str( json.dumps( { "message_type": "input_audio_chunk", "audio_base_64": audio_b64, "commit": False, "sample_rate": self._opts.sample_rate, } ) ) if has_ended: self._audio_duration_collector.flush() await ws.send_str( json.dumps( { "message_type": "input_audio_chunk", "audio_base_64": "", "commit": True, "sample_rate": self._opts.sample_rate, } ) ) has_ended = False closing_ws = True except (aiohttp.ClientError, ConnectionError) as e: if closing_ws or self._session.closed: return raise APIConnectionError("ElevenLabs STT connection closed unexpectedly") from e @utils.log_exceptions(logger=logger) async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None: nonlocal closing_ws while True: msg = await ws.receive() if msg.type in ( aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.CLOSE, aiohttp.WSMsgType.CLOSING, ): if closing_ws or self._session.closed: return raise APIStatusError( message="ElevenLabs STT connection closed unexpectedly", status_code=ws.close_code or -1, body=f"{msg.data=} {msg.extra=}", ) if msg.type != aiohttp.WSMsgType.TEXT: logger.warning("unexpected ElevenLabs STT message type %s", msg.type) continue try: parsed = json.loads(msg.data) self._process_stream_event(parsed) except Exception: logger.exception("failed to process ElevenLabs STT message") ws: aiohttp.ClientWebSocketResponse | None = None while True: try: ws = await self._connect_ws() self._last_partial_text = "" if self._opts.previous_text: # Must be the first input_audio_chunk on the connection. await ws.send_str( json.dumps( { "message_type": "input_audio_chunk", "audio_base_64": "", "commit": False, "sample_rate": self._opts.sample_rate, "previous_text": self._opts.previous_text, } ) ) tasks = [ asyncio.create_task(send_task(ws)), asyncio.create_task(recv_task(ws)), asyncio.create_task(keepalive_task(ws)), ] tasks_group = asyncio.gather(*tasks) wait_reconnect_task = asyncio.create_task(self._reconnect_event.wait()) try: done, _ = await asyncio.wait( (tasks_group, wait_reconnect_task), return_when=asyncio.FIRST_COMPLETED, ) for task in done: if task != wait_reconnect_task: task.result() if wait_reconnect_task not in done: break self._reconnect_event.clear() finally: await utils.aio.gracefully_cancel(*tasks, wait_reconnect_task) tasks_group.cancel() tasks_group.exception() # Retrieve exception to prevent it from being logged finally: if ws is not None: await ws.close() async def _connect_ws(self) -> aiohttp.ClientWebSocketResponse: """Establish WebSocket connection to ElevenLabs Scribe v2 API""" commit_strategy = "vad" if self._server_vad is not None else "manual" params = [ f"model_id={self._opts.model_id}", f"audio_format=pcm_{self._opts.sample_rate}", f"commit_strategy={commit_strategy}", f"enable_logging={str(self._opts.enable_logging).lower()}", ] if self._language_detection: params.append("include_language_detection=true") if (server_vad := self._server_vad) is not None: if ( vad_silence_threshold_secs := server_vad.get("vad_silence_threshold_secs") ) is not None: params.append(f"vad_silence_threshold_secs={vad_silence_threshold_secs}") if (vad_threshold := server_vad.get("vad_threshold")) is not None: params.append(f"vad_threshold={vad_threshold}") if (min_speech_duration_ms := server_vad.get("min_speech_duration_ms")) is not None: params.append(f"min_speech_duration_ms={min_speech_duration_ms}") if (min_silence_duration_ms := server_vad.get("min_silence_duration_ms")) is not None: params.append(f"min_silence_duration_ms={min_silence_duration_ms}") # the realtime API takes a bare ISO-639-1/639-3 code and rejects the session on a # region-tagged one ("ru-RU"), so both language params go on the wire without the region if self._language: params.append(f"language_code={quote(self._language.language)}") if is_given(self._opts.secondary_languages): params.extend( f"secondary_languages={quote(language.language)}" for language in self._opts.secondary_languages ) if self._opts.include_timestamps: params.append("include_timestamps=true") if self._opts.no_verbatim: params.append("no_verbatim=true") if is_given(self._opts.keyterms): params.extend(f"keyterms={quote(keyterm)}" for keyterm in self._opts.keyterms) query_string = "&".join(params) # Convert HTTPS URL to WSS base_url = self._opts.base_url.replace("https://", "wss://").replace("http://", "ws://") ws_url = f"{base_url}/speech-to-text/realtime?{query_string}" try: ws = await asyncio.wait_for( self._session.ws_connect( ws_url, headers={AUTHORIZATION_HEADER: self._opts.api_key}, ), self._conn_options.timeout, ) except aiohttp.WSServerHandshakeError as e: raise APIStatusError( message=e.message, status_code=e.status, request_id=trace_id_from_headers(e.headers), ) from e except (aiohttp.ClientConnectorError, asyncio.TimeoutError) as e: raise APIConnectionError("Failed to connect to ElevenLabs") from e return ws def _process_stream_event(self, data: dict) -> None: """Process incoming WebSocket messages from ElevenLabs""" message_type = data.get("message_type") text = data.get("text", "") words = data.get("words", []) start_time = words[0].get("start", 0) if words else 0 end_time = words[-1].get("end", 0) if words else 0 language_code = data.get("language_code") or self._language normalized_language = LanguageCode(language_code) if language_code else LanguageCode("en") # 11labs only sends word timestamps for final transcripts speech_data = stt.SpeechData( language=normalized_language, text=text, start_time=start_time + self.start_time_offset, end_time=end_time + self.start_time_offset, confidence=_speech_confidence(words), ) if words and self._opts.include_timestamps: speech_data.words = [ TimedString( text=word.get("text", ""), start_time=word.get("start", 0) + self.start_time_offset, end_time=word.get("end", 0) + self.start_time_offset, start_time_offset=self.start_time_offset, ) for word in words ] if message_type == "partial_transcript": logger.debug("Received message type partial_transcript", extra={"lk.pii.data": data}) if text and text != self._last_partial_text: self._last_partial_text = text # Send START_OF_SPEECH if we're not already speaking if not self._speaking: self._event_ch.send_nowait( stt.SpeechEvent(type=SpeechEventType.START_OF_SPEECH) ) self._speaking = True # Send INTERIM_TRANSCRIPT interim_event = stt.SpeechEvent( type=SpeechEventType.INTERIM_TRANSCRIPT, alternatives=[speech_data], ) self._event_ch.send_nowait(interim_event) # 11labs sends every commit twice; _final_message_type picks the copy this session reads elif message_type == self._final_message_type: # Final committed transcripts - these are sent to the LLM/TTS layer in LiveKit agents # and trigger agent responses (unlike partial transcripts which are UI-only) self._last_partial_text = "" if text: # Send START_OF_SPEECH if we're not already speaking if not self._speaking: self._event_ch.send_nowait( stt.SpeechEvent(type=SpeechEventType.START_OF_SPEECH) ) self._speaking = True # Send FINAL_TRANSCRIPT but keep speaking=True # Multiple commits can occur within the same speech segment final_event = stt.SpeechEvent( type=SpeechEventType.FINAL_TRANSCRIPT, alternatives=[speech_data], ) self._event_ch.send_nowait(final_event) if self._server_vad is not None: self._event_ch.send_nowait(stt.SpeechEvent(type=SpeechEventType.END_OF_SPEECH)) self._speaking = False else: # Empty commit signals end of speech segment (similar to Cartesia's is_final flag) # This groups multiple committed transcripts into one speech segment if self._speaking: self._event_ch.send_nowait(stt.SpeechEvent(type=SpeechEventType.END_OF_SPEECH)) self._speaking = False elif message_type in ("committed_transcript", "committed_transcript_with_timestamps"): # the other copy of a commit the branch above already emitted pass elif message_type == "session_started": # Session initialization message - informational only session_id = data.get("session_id", "unknown") logger.debug("Session started with ID: %s", session_id) # Error handling for known ElevenLabs error types elif message_type in ( "auth_error", "quota_exceeded", "transcriber_error", "input_error", "error", ): error_msg = data.get("message", "Unknown error") error_details = data.get("details", "") details_suffix = " - " + error_details if error_details else "" logger.error( "ElevenLabs STT error [%s]: %s%s", message_type, error_msg, details_suffix, ) raise APIConnectionError(f"{message_type}: {error_msg}{details_suffix}") else: logger.warning( "ElevenLabs STT unknown message type: %s", message_type, extra={"lk.pii.data": data}, )Streaming speech recognition using ElevenLabs Scribe v2 realtime API
Args: sample_rate : int or None, optional The desired sample rate for the audio input. If specified, the audio input will be automatically resampled to match the given sample rate before being processed for Speech-to-Text. If not provided (None), the input will retain its original sample rate.
Ancestors
- livekit.agents.stt.stt.RecognizeStream
- abc.ABC
Methods
def update_options(self,
*,
server_vad: NotGivenOr[VADOptions] = NOT_GIVEN,
no_verbatim: NotGivenOr[bool] = NOT_GIVEN,
keyterms: NotGivenOr[list[str]] = NOT_GIVEN,
secondary_languages: NotGivenOr[list[str]] = NOT_GIVEN) ‑> None-
Expand source code
def update_options( self, *, server_vad: NotGivenOr[VADOptions] = NOT_GIVEN, no_verbatim: NotGivenOr[bool] = NOT_GIVEN, keyterms: NotGivenOr[list[str]] = NOT_GIVEN, secondary_languages: NotGivenOr[list[str]] = NOT_GIVEN, ) -> None: if is_given(server_vad): self._opts.server_vad = server_vad self._reconnect_event.set() if is_given(no_verbatim): self._opts.no_verbatim = no_verbatim self._reconnect_event.set() if is_given(keyterms): self._opts.keyterms = keyterms self._reconnect_event.set() if is_given(secondary_languages): self._opts.secondary_languages = [ LanguageCode(language) for language in secondary_languages ] self._reconnect_event.set()
class TTS (*,
voice_id: str = 'hpp4J3VqNfWAUOO0d1Us',
voice_settings: NotGivenOr[VoiceSettings] = NOT_GIVEN,
model: TTSModels | str = 'eleven_turbo_v2_5',
encoding: NotGivenOr[TTSEncoding] = NOT_GIVEN,
api_key: NotGivenOr[str] = NOT_GIVEN,
base_url: NotGivenOr[str] = NOT_GIVEN,
streaming_latency: NotGivenOr[int] = NOT_GIVEN,
inactivity_timeout: int = 180,
auto_mode: NotGivenOr[bool] = NOT_GIVEN,
apply_text_normalization: "Literal['auto', 'off', 'on']" = 'auto',
apply_language_text_normalization: NotGivenOr[bool] = NOT_GIVEN,
word_tokenizer: NotGivenOr[tokenize.WordTokenizer | tokenize.SentenceTokenizer] = NOT_GIVEN,
enable_ssml_parsing: bool = False,
enable_logging: bool = True,
chunk_length_schedule: NotGivenOr[list[int]] = NOT_GIVEN,
http_session: aiohttp.ClientSession | None = None,
language: NotGivenOr[str] = NOT_GIVEN,
sync_alignment: bool = True,
preferred_alignment: "NotGivenOr[Literal['normalized', 'original']]" = NOT_GIVEN,
pronunciation_dictionary_locators: NotGivenOr[list[PronunciationDictionaryLocator]] = NOT_GIVEN)-
Expand source code
class TTS(tts.TTS): def __init__( self, *, voice_id: str = DEFAULT_VOICE_ID, voice_settings: NotGivenOr[VoiceSettings] = NOT_GIVEN, model: TTSModels | str = "eleven_turbo_v2_5", encoding: NotGivenOr[TTSEncoding] = NOT_GIVEN, api_key: NotGivenOr[str] = NOT_GIVEN, base_url: NotGivenOr[str] = NOT_GIVEN, streaming_latency: NotGivenOr[int] = NOT_GIVEN, inactivity_timeout: int = WS_INACTIVITY_TIMEOUT, auto_mode: NotGivenOr[bool] = NOT_GIVEN, apply_text_normalization: Literal["auto", "off", "on"] = "auto", apply_language_text_normalization: NotGivenOr[bool] = NOT_GIVEN, word_tokenizer: NotGivenOr[tokenize.WordTokenizer | tokenize.SentenceTokenizer] = NOT_GIVEN, enable_ssml_parsing: bool = False, enable_logging: bool = True, chunk_length_schedule: NotGivenOr[list[int]] = NOT_GIVEN, # range is [50, 500] http_session: aiohttp.ClientSession | None = None, language: NotGivenOr[str] = NOT_GIVEN, sync_alignment: bool = True, preferred_alignment: NotGivenOr[Literal["normalized", "original"]] = NOT_GIVEN, pronunciation_dictionary_locators: NotGivenOr[ list[PronunciationDictionaryLocator] ] = NOT_GIVEN, ) -> None: """ Create a new instance of ElevenLabs TTS. Args: voice_id (str): Voice ID. Defaults to `DEFAULT_VOICE_ID`. voice_settings (NotGivenOr[VoiceSettings]): Voice settings. model (TTSModels | str): TTS model to use. Defaults to "eleven_turbo_v2_5". "eleven_v3" and "eleven_v3_conversational" go through ElevenLabs' text-to-dialogue API instead (single voice per instance, same as other models). api_key (NotGivenOr[str]): ElevenLabs API key. Can be set via argument or `ELEVEN_API_KEY` environment variable. base_url (NotGivenOr[str]): Custom base URL for the API. Optional. streaming_latency (NotGivenOr[int]): Optimize for streaming latency, defaults to 0 - disabled. 4 for max latency optimizations. deprecated inactivity_timeout (int): Inactivity timeout in seconds for the websocket connection. Defaults to 300. auto_mode (bool): Reduces latency by disabling chunk schedule and buffers. Sentence tokenizer will be used to synthesize one sentence at a time. Defaults to True unless ``chunk_length_schedule`` is provided. apply_text_normalization (Literal["auto", "off", "on"]): This parameter controls text normalization with three modes: ‘auto’, ‘on’, and ‘off’. When set to ‘auto’, the system will automatically decide whether to apply text normalization (e.g., spelling out numbers). With ‘on’, text normalization will always be applied, while with ‘off’, it will be skipped. apply_language_text_normalization (bool): This parameter controls language text normalization. This helps with proper pronunciation of text in some supported languages. word_tokenizer (NotGivenOr[tokenize.WordTokenizer | tokenize.SentenceTokenizer]): Tokenizer for processing text. Defaults to basic WordTokenizer when auto_mode=False, `livekit.agents.tokenize.blingfire.SentenceTokenizer` otherwise. enable_ssml_parsing (bool): Enable SSML parsing for input text. Defaults to False. enable_logging (bool): Enable logging of the request. When set to false, zero retention mode will be used. Defaults to True. chunk_length_schedule (NotGivenOr[list[int]]): Schedule for chunk lengths, ranging from 50 to 500. Defaults are [120, 160, 250, 290]. http_session (aiohttp.ClientSession | None): Custom HTTP session for API requests. Optional. language (NotGivenOr[str]): Language code used to enforce a language for the model and text normalization. If the model does not support language overrides, it will be ignored. sync_alignment (bool): Enable sync alignment for the TTS model. Defaults to True. preferred_alignment (Literal["normalized", "original"]): Use normalized or original alignment. Defaults to "normalized", or "original" for CJK (ja, ko, zh) languages. pronunciation_dictionary_locators (NotGivenOr[list[PronunciationDictionaryLocator]]): List of pronunciation dictionary locators to use for pronunciation control. """ # noqa: E501 if not is_given(encoding): encoding = _DefaultEncoding super().__init__( capabilities=tts.TTSCapabilities( streaming=True, aligned_transcript=sync_alignment, ), sample_rate=_sample_rate_from_format(encoding), num_channels=1, ) elevenlabs_api_key = api_key if is_given(api_key) else os.environ.get("ELEVEN_API_KEY") if not elevenlabs_api_key: raise ValueError( "ElevenLabs API key is required, either as argument or set ELEVEN_API_KEY environmental variable" # noqa: E501 ) if not is_given(auto_mode): auto_mode = not is_given(chunk_length_schedule) if not is_given(word_tokenizer): word_tokenizer = ( tokenize.basic.WordTokenizer(ignore_punctuation=False) if not auto_mode else tokenize.blingfire.SentenceTokenizer() ) elif auto_mode and not isinstance(word_tokenizer, tokenize.SentenceTokenizer): logger.warning( "auto_mode is enabled, it expects full sentences or phrases, " "please provide a SentenceTokenizer instead of a WordTokenizer." ) self._opts = _TTSOptions( voice_id=voice_id, voice_settings=voice_settings, model=model, api_key=elevenlabs_api_key, base_url=base_url if is_given(base_url) else API_BASE_URL_V1, encoding=encoding, sample_rate=self.sample_rate, streaming_latency=streaming_latency, word_tokenizer=word_tokenizer, chunk_length_schedule=chunk_length_schedule, enable_ssml_parsing=enable_ssml_parsing, enable_logging=enable_logging, language=LanguageCode(language) if is_given(language) else NOT_GIVEN, inactivity_timeout=inactivity_timeout, sync_alignment=sync_alignment, auto_mode=auto_mode, apply_text_normalization=apply_text_normalization, apply_language_text_normalization=apply_language_text_normalization, preferred_alignment=preferred_alignment, pronunciation_dictionary_locators=pronunciation_dictionary_locators, ) self._session = http_session self._streams = weakref.WeakSet[SynthesizeStream]() self.__current_connection: _Connection | _DialogueConnection | None = None self._connection_lock = asyncio.Lock() self._warn_if_dialogue_model_ignores_options() @property def model(self) -> str: return self._opts.model @property def provider(self) -> str: return "ElevenLabs" def _ensure_session(self) -> aiohttp.ClientSession: if not self._session: self._session = utils.http_context.http_session() return self._session def _warn_if_dialogue_model_ignores_options(self) -> None: """Warn if options unsupported by text-to-dialogue are set for a dialogue model.""" if not is_dialogue_model(self._opts.model): return ignored = [ name for name, is_set in ( ("chunk_length_schedule", is_given(self._opts.chunk_length_schedule)), ("streaming_latency", is_given(self._opts.streaming_latency)), ("enable_ssml_parsing", self._opts.enable_ssml_parsing), ( "apply_language_text_normalization", is_given(self._opts.apply_language_text_normalization), ), ) if is_set ] if is_given(self._opts.voice_settings): settings = _strip_nones(dataclasses.asdict(self._opts.voice_settings)) ignored.extend( f"voice_settings.{name}" for name in settings if name not in _DIALOGUE_VOICE_SETTINGS_FIELDS ) if ignored: logger.warning( "model '%s' is synthesized via ElevenLabs' text-to-dialogue API, which " "does not support these options; they will be ignored: %s", self._opts.model, ", ".join(ignored), ) async def list_voices(self) -> list[Voice]: async with self._ensure_session().get( f"{self._opts.base_url}/voices", headers={AUTHORIZATION_HEADER: self._opts.api_key}, ) as resp: return _dict_to_voices_list(await resp.json()) def update_options( self, *, voice_id: NotGivenOr[str] = NOT_GIVEN, voice_settings: NotGivenOr[VoiceSettings] = NOT_GIVEN, model: NotGivenOr[TTSModels | str] = NOT_GIVEN, language: NotGivenOr[str] = NOT_GIVEN, pronunciation_dictionary_locators: NotGivenOr[ list[PronunciationDictionaryLocator] ] = NOT_GIVEN, ) -> None: """ Args: voice_id (NotGivenOr[str]): Voice ID. voice_settings (NotGivenOr[VoiceSettings]): Voice settings. model (NotGivenOr[TTSModels | str]): TTS model to use. language (NotGivenOr[str]): Language code for the TTS model. pronunciation_dictionary_locators (NotGivenOr[list[PronunciationDictionaryLocator]]): List of pronunciation dictionary locators. """ changed = False if is_given(model) and model != self._opts.model: self._opts.model = model changed = True self._warn_if_dialogue_model_ignores_options() if is_given(voice_id) and voice_id != self._opts.voice_id: self._opts.voice_id = voice_id changed = True if is_given(voice_settings): self._opts.voice_settings = voice_settings changed = True if is_given(language): language = LanguageCode(language) if language != self._opts.language: self._opts.language = language changed = True if is_given(pronunciation_dictionary_locators): self._opts.pronunciation_dictionary_locators = pronunciation_dictionary_locators changed = True if changed and self.__current_connection: self.__current_connection.mark_non_current() self.__current_connection = None async def _current_connection(self) -> tuple[_Connection | _DialogueConnection, float, bool]: """Get the current connection, creating one if needed. Returns: Tuple of (connection, acquire_time, connection_reused) """ async with self._connection_lock: if ( self.__current_connection and self.__current_connection.is_current and not self.__current_connection._closed ): return self.__current_connection, 0.0, True session = self._ensure_session() conn: _Connection | _DialogueConnection = ( _DialogueConnection(self._opts, session) if is_dialogue_model(self._opts.model) else _Connection(self._opts, session) ) t0 = time.perf_counter() await conn.connect() acquire_time = time.perf_counter() - t0 self.__current_connection = conn return conn, acquire_time, False def synthesize( self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS ) -> ChunkedStream: return ChunkedStream(tts=self, input_text=text, conn_options=conn_options) def stream( self, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS ) -> SynthesizeStream: stream = SynthesizeStream(tts=self, conn_options=conn_options) self._streams.add(stream) return stream async def aclose(self) -> None: for stream in list(self._streams): await stream.aclose() self._streams.clear() if self.__current_connection: await self.__current_connection.aclose() self.__current_connection = NoneHelper class that provides a standard way to create an ABC using inheritance.
Create a new instance of ElevenLabs TTS.
Args
voice_id:str- Voice ID. Defaults to
DEFAULT_VOICE_ID. voice_settings:NotGivenOr[VoiceSettings]- Voice settings.
model:TTSModels | str- TTS model to use. Defaults to "eleven_turbo_v2_5". "eleven_v3" and "eleven_v3_conversational" go through ElevenLabs' text-to-dialogue API instead (single voice per instance, same as other models).
api_key:NotGivenOr[str]- ElevenLabs API key. Can be set via argument or
ELEVEN_API_KEYenvironment variable. base_url:NotGivenOr[str]- Custom base URL for the API. Optional.
streaming_latency:NotGivenOr[int]- Optimize for streaming latency, defaults to 0 - disabled. 4 for max latency optimizations. deprecated
inactivity_timeout:int- Inactivity timeout in seconds for the websocket connection. Defaults to 300.
auto_mode:bool- Reduces latency by disabling chunk schedule and buffers.
Sentence tokenizer will be used to synthesize one sentence at a time.
Defaults to True unless
chunk_length_scheduleis provided. - apply_text_normalization (Literal["auto", "off", "on"]): This parameter controls text normalization with three modes: ‘auto’, ‘on’, and ‘off’. When set to ‘auto’, the system will automatically decide whether to apply text normalization (e.g., spelling out numbers). With ‘on’, text normalization will always be applied, while with ‘off’, it will be skipped.
apply_language_text_normalization:bool- This parameter controls language text normalization. This helps with proper pronunciation of text in some supported languages.
word_tokenizer:NotGivenOr[tokenize.WordTokenizer | tokenize.SentenceTokenizer]- Tokenizer for processing text. Defaults to basic WordTokenizer when auto_mode=False,
SentenceTokenizerotherwise. enable_ssml_parsing:bool- Enable SSML parsing for input text. Defaults to False.
enable_logging:bool- Enable logging of the request. When set to false, zero retention mode will be used. Defaults to True.
chunk_length_schedule:NotGivenOr[list[int]]- Schedule for chunk lengths, ranging from 50 to 500. Defaults are [120, 160, 250, 290].
http_session:aiohttp.ClientSession | None- Custom HTTP session for API requests. Optional.
language:NotGivenOr[str]- Language code used to enforce a language for the model and text normalization. If the model does not support language overrides, it will be ignored.
sync_alignment:bool- Enable sync alignment for the TTS model. Defaults to True.
- preferred_alignment (Literal["normalized", "original"]): Use normalized or original alignment. Defaults to "normalized", or "original" for CJK (ja, ko, zh) languages.
pronunciation_dictionary_locators:NotGivenOr[list[PronunciationDictionaryLocator]]- List of pronunciation dictionary locators to use for pronunciation control.
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 "ElevenLabs"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() if self.__current_connection: await self.__current_connection.aclose() self.__current_connection = None async def list_voices(self) ‑> list[livekit.plugins.elevenlabs.tts.Voice]-
Expand source code
async def list_voices(self) -> list[Voice]: async with self._ensure_session().get( f"{self._opts.base_url}/voices", headers={AUTHORIZATION_HEADER: self._opts.api_key}, ) as resp: return _dict_to_voices_list(await resp.json()) def stream(self,
*,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> livekit.plugins.elevenlabs.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.plugins.elevenlabs.tts.ChunkedStream-
Expand source code
def synthesize( self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS ) -> ChunkedStream: return ChunkedStream(tts=self, input_text=text, conn_options=conn_options) def update_options(self,
*,
voice_id: NotGivenOr[str] = NOT_GIVEN,
voice_settings: NotGivenOr[VoiceSettings] = NOT_GIVEN,
model: NotGivenOr[TTSModels | str] = NOT_GIVEN,
language: NotGivenOr[str] = NOT_GIVEN,
pronunciation_dictionary_locators: NotGivenOr[list[PronunciationDictionaryLocator]] = NOT_GIVEN) ‑> None-
Expand source code
def update_options( self, *, voice_id: NotGivenOr[str] = NOT_GIVEN, voice_settings: NotGivenOr[VoiceSettings] = NOT_GIVEN, model: NotGivenOr[TTSModels | str] = NOT_GIVEN, language: NotGivenOr[str] = NOT_GIVEN, pronunciation_dictionary_locators: NotGivenOr[ list[PronunciationDictionaryLocator] ] = NOT_GIVEN, ) -> None: """ Args: voice_id (NotGivenOr[str]): Voice ID. voice_settings (NotGivenOr[VoiceSettings]): Voice settings. model (NotGivenOr[TTSModels | str]): TTS model to use. language (NotGivenOr[str]): Language code for the TTS model. pronunciation_dictionary_locators (NotGivenOr[list[PronunciationDictionaryLocator]]): List of pronunciation dictionary locators. """ changed = False if is_given(model) and model != self._opts.model: self._opts.model = model changed = True self._warn_if_dialogue_model_ignores_options() if is_given(voice_id) and voice_id != self._opts.voice_id: self._opts.voice_id = voice_id changed = True if is_given(voice_settings): self._opts.voice_settings = voice_settings changed = True if is_given(language): language = LanguageCode(language) if language != self._opts.language: self._opts.language = language changed = True if is_given(pronunciation_dictionary_locators): self._opts.pronunciation_dictionary_locators = pronunciation_dictionary_locators changed = True if changed and self.__current_connection: self.__current_connection.mark_non_current() self.__current_connection = NoneArgs
voice_id:NotGivenOr[str]- Voice ID.
voice_settings:NotGivenOr[VoiceSettings]- Voice settings.
model:NotGivenOr[TTSModels | str]- TTS model to use.
language:NotGivenOr[str]- Language code for the TTS model.
pronunciation_dictionary_locators:NotGivenOr[list[PronunciationDictionaryLocator]]- List of pronunciation dictionary locators.
Inherited members
class Voice (id: str, name: str, category: str)-
Expand source code
@dataclass class Voice: id: str name: str category: strVoice(id: 'str', name: 'str', category: 'str')
Instance variables
var category : strvar id : strvar name : str
class VoiceSettings (stability: float,
similarity_boost: float,
style: NotGivenOr[float] = NOT_GIVEN,
speed: NotGivenOr[float] = NOT_GIVEN,
use_speaker_boost: NotGivenOr[bool] = NOT_GIVEN)-
Expand source code
@dataclass class VoiceSettings: stability: float # [0.0 - 1.0] similarity_boost: float # [0.0 - 1.0] style: NotGivenOr[float] = NOT_GIVEN # [0.0 - 1.0] speed: NotGivenOr[float] = NOT_GIVEN # [0.8 - 1.2] use_speaker_boost: NotGivenOr[bool] = NOT_GIVENVoiceSettings(stability: 'float', similarity_boost: 'float', style: 'NotGivenOr[float]' = NOT_GIVEN, speed: 'NotGivenOr[float]' = NOT_GIVEN, use_speaker_boost: 'NotGivenOr[bool]' = NOT_GIVEN)
Instance variables
var similarity_boost : floatvar speed : float | livekit.agents.types.NotGivenvar stability : floatvar style : float | livekit.agents.types.NotGivenvar use_speaker_boost : bool | livekit.agents.types.NotGiven