Module livekit.plugins.deepgram
Deepgram plugin for LiveKit Agents
Support for speech-to-text with Deepgram.
See https://docs.livekit.io/agents/integrations/stt/deepgram/ for more information.
Classes
class STT (*,
model: DeepgramModels | str = 'nova-3',
language: DeepgramLanguages | str = 'en-US',
detect_language: bool = False,
interim_results: bool = True,
punctuate: bool = True,
smart_format: bool = False,
sample_rate: int = 16000,
no_delay: bool = True,
endpointing_ms: int = 25,
enable_diarization: bool = False,
filler_words: bool = True,
keywords: NotGivenOr[list[tuple[str, float]]] = NOT_GIVEN,
keyterm: NotGivenOr[str | list[str]] = NOT_GIVEN,
tags: NotGivenOr[list[str]] = NOT_GIVEN,
profanity_filter: bool = False,
redact: NotGivenOr[str | list[str]] = NOT_GIVEN,
api_key: NotGivenOr[str] = NOT_GIVEN,
http_session: aiohttp.ClientSession | None = None,
base_url: str = 'https://api.deepgram.com/v1/listen',
numerals: bool = False,
mip_opt_out: bool = False,
vad_events: bool = True,
utterance_end_ms: int | None = None,
dictation: bool = False,
replace: dict[str, str] | None = None,
search: list[str] | None = None,
keyterms: NotGivenOr[list[str]] = NOT_GIVEN)-
Expand source code
class STT(stt.STT): def __init__( self, *, model: DeepgramModels | str = "nova-3", language: DeepgramLanguages | str = "en-US", detect_language: bool = False, interim_results: bool = True, punctuate: bool = True, smart_format: bool = False, sample_rate: int = 16000, no_delay: bool = True, endpointing_ms: int = 25, enable_diarization: bool = False, # enable filler words by default to improve turn detector accuracy filler_words: bool = True, keywords: NotGivenOr[list[tuple[str, float]]] = NOT_GIVEN, keyterm: NotGivenOr[str | list[str]] = NOT_GIVEN, tags: NotGivenOr[list[str]] = NOT_GIVEN, profanity_filter: bool = False, redact: NotGivenOr[str | list[str]] = NOT_GIVEN, api_key: NotGivenOr[str] = NOT_GIVEN, http_session: aiohttp.ClientSession | None = None, base_url: str = "https://api.deepgram.com/v1/listen", numerals: bool = False, mip_opt_out: bool = False, vad_events: bool = True, utterance_end_ms: int | None = None, dictation: bool = False, replace: dict[str, str] | None = None, search: list[str] | None = None, # deprecated keyterms: NotGivenOr[list[str]] = NOT_GIVEN, ) -> None: """Create a new instance of Deepgram STT. Args: model: The Deepgram model to use for speech recognition. Defaults to "nova-3". language: The language code for recognition. Defaults to "en-US". detect_language: Whether to enable automatic language detection. Defaults to False. interim_results: Whether to return interim (non-final) transcription results. Defaults to True. punctuate: Whether to add punctuations to the transcription. Defaults to True. Turn detector will work better with punctuations. smart_format: Whether to apply smart formatting to numbers, dates, etc. Defaults to False. sample_rate: The sample rate of the audio in Hz. Defaults to 16000. no_delay: When smart_format is used, ensures it does not wait for sequence to be complete before returning results. Defaults to True. endpointing_ms: Time in milliseconds of silence to consider end of speech. Set to 0 to disable. Defaults to 25. filler_words: Whether to include filler words (um, uh, etc.) in transcription. Defaults to True. keywords: List of tuples containing keywords and their boost values for improved recognition. Each tuple should be (keyword: str, boost: float). Defaults to None. `keywords` does not work with Nova-3 models. Use `keyterm` instead. keyterm: str or list of str of key terms to improve recognition accuracy. Defaults to None. `keyterm` is only supported by Nova-3 models. tags: List of tags to add to the requests for usage reporting. Defaults to NOT_GIVEN. profanity_filter: Whether to filter profanity from the transcription. Defaults to False. redact: Redact sensitive information from the transcription. Accepts a single value or list of values. Supported values: "pci", "numbers", "ssn", "true" (redact all). See https://developers.deepgram.com/docs/redaction for details. api_key: Your Deepgram API key. If not provided, will look for DEEPGRAM_API_KEY environment variable. http_session: Optional aiohttp ClientSession to use for requests. base_url: The base URL for Deepgram API. Defaults to "https://api.deepgram.com/v1/listen". numerals: Whether to include numerals in the transcription. Defaults to False. mip_opt_out: Whether to take part in the model improvement program vad_events: Whether to enable VAD (Voice Activity Detection) events. When enabled, SpeechStarted events are sent when speech is detected. Defaults to True. utterance_end_ms: Duration of silence in milliseconds to detect the end of an utterance and emit an UtteranceEnd event. Requires interim_results=True. See https://developers.deepgram.com/docs/understand-endpointing-interim-results dictation: Whether to enable dictation mode which converts spoken punctuation commands (e.g. "comma", "period") into punctuation marks. Defaults to False. See https://developers.deepgram.com/reference/speech-to-text/listen-streaming#query-dictation replace: Dictionary of terms to replace in the transcript, where keys are the original terms and values are the replacements (e.g. {"hello": "hi"}). See https://developers.deepgram.com/reference/speech-to-text/listen-streaming#query-replace search: List of terms to search for in the transcript. Matched terms are returned with confidence scores in the response. See https://developers.deepgram.com/reference/speech-to-text/listen-streaming#query-search Raises: ValueError: If no API key is provided or found in environment variables. Note: The api_key must be set either through the constructor argument or by setting the DEEPGRAM_API_KEY environmental variable. """ # noqa: E501 super().__init__( capabilities=stt.STTCapabilities( streaming=True, interim_results=interim_results, diarization=enable_diarization, aligned_transcript="word", keyterms=True, ) ) deepgram_api_key = api_key if is_given(api_key) else os.environ.get("DEEPGRAM_API_KEY") if not deepgram_api_key: raise ValueError( "Deepgram API key is required, either as argument or set" " DEEPGRAM_API_KEY environment variable" ) self._api_key = deepgram_api_key model = _validate_model(model, language) if is_given(keyterms): logger.warning( "`keyterms` is deprecated, use `keyterm` instead for consistency with Deepgram API." ) keyterm = keyterms _validate_keyterm(model, language, keyterm, keywords) self._opts = STTOptions( language=LanguageCode(language) if language else None, detect_language=detect_language, interim_results=interim_results, punctuate=punctuate, model=model, smart_format=smart_format, no_delay=no_delay, endpointing_ms=endpointing_ms, enable_diarization=enable_diarization, filler_words=filler_words, sample_rate=sample_rate, num_channels=1, keywords=keywords if is_given(keywords) else [], keyterm=([keyterm] if isinstance(keyterm, str) else list(keyterm)) if is_given(keyterm) else [], profanity_filter=profanity_filter, redact=redact if is_given(redact) else [], numerals=numerals, mip_opt_out=mip_opt_out, vad_events=vad_events, tags=_validate_tags(tags) if is_given(tags) else [], endpoint_url=base_url, utterance_end_ms=utterance_end_ms, dictation=dictation, replace=replace, search=search, ) # user keyterms; _opts.keyterm holds the effective set (user + session) self._user_keyterm: list[str] = list(self._opts.keyterm) self._session_keyterms: list[str] = [] self._session = http_session self._streams = weakref.WeakSet[SpeechStream]() @property def model(self) -> str: return self._opts.model @property def provider(self) -> str: return "Deepgram" def _ensure_session(self) -> aiohttp.ClientSession: if not self._session: self._session = utils.http_context.http_session() return self._session async def _recognize_impl( self, buffer: AudioBuffer, *, language: NotGivenOr[DeepgramLanguages | str] = NOT_GIVEN, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS, ) -> stt.SpeechEvent: config = self._sanitize_options(language=language) recognize_config = { "model": str(config.model), "punctuate": config.punctuate, "detect_language": config.detect_language, "smart_format": config.smart_format, "keywords": self._opts.keywords, "profanity_filter": config.profanity_filter, "numerals": config.numerals, "mip_opt_out": config.mip_opt_out, } if self._opts.keyterm: recognize_config["keyterm"] = self._opts.keyterm if config.redact: recognize_config["redact"] = config.redact if config.enable_diarization: logger.warning("speaker diarization is not supported in non-streaming mode, ignoring") if config.language: recognize_config["language"] = config.language try: async with self._ensure_session().post( url=_to_deepgram_url(recognize_config, self._opts.endpoint_url, websocket=False), data=rtc.combine_audio_frames(buffer).to_wav_bytes(), headers={ "Authorization": f"Token {self._api_key}", "Accept": "application/json", "Content-Type": "audio/wav", }, timeout=aiohttp.ClientTimeout( total=30, sock_connect=conn_options.timeout, ), ) as res: return prerecorded_transcription_to_speech_event( config.language, await res.json(), ) except asyncio.TimeoutError as e: raise APITimeoutError() from e except aiohttp.ClientResponseError as e: raise APIStatusError( message=e.message, status_code=e.status, request_id=None, body=None, ) from e except Exception as e: raise APIConnectionError() from e def stream( self, *, language: NotGivenOr[DeepgramLanguages | str] = NOT_GIVEN, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS, ) -> SpeechStream: config = self._sanitize_options(language=language) stream = SpeechStream( stt=self, conn_options=conn_options, opts=config, api_key=self._api_key, http_session=self._ensure_session(), base_url=self._opts.endpoint_url, ) self._streams.add(stream) return stream def update_options( self, *, language: NotGivenOr[DeepgramLanguages | str] = NOT_GIVEN, model: NotGivenOr[DeepgramModels | str] = NOT_GIVEN, interim_results: NotGivenOr[bool] = NOT_GIVEN, punctuate: NotGivenOr[bool] = NOT_GIVEN, smart_format: NotGivenOr[bool] = NOT_GIVEN, sample_rate: NotGivenOr[int] = NOT_GIVEN, no_delay: NotGivenOr[bool] = NOT_GIVEN, endpointing_ms: NotGivenOr[int] = NOT_GIVEN, enable_diarization: NotGivenOr[bool] = NOT_GIVEN, filler_words: NotGivenOr[bool] = NOT_GIVEN, keywords: NotGivenOr[list[tuple[str, float]]] = NOT_GIVEN, keyterm: NotGivenOr[str | list[str]] = NOT_GIVEN, profanity_filter: NotGivenOr[bool] = NOT_GIVEN, redact: NotGivenOr[str | list[str]] = NOT_GIVEN, numerals: NotGivenOr[bool] = NOT_GIVEN, mip_opt_out: NotGivenOr[bool] = NOT_GIVEN, vad_events: NotGivenOr[bool] = NOT_GIVEN, tags: NotGivenOr[list[str]] = NOT_GIVEN, endpoint_url: NotGivenOr[str] = NOT_GIVEN, utterance_end_ms: NotGivenOr[int | None] = NOT_GIVEN, dictation: NotGivenOr[bool] = NOT_GIVEN, replace: NotGivenOr[dict[str, str] | None] = NOT_GIVEN, search: NotGivenOr[list[str] | None] = NOT_GIVEN, # deprecated keyterms: NotGivenOr[list[str]] = NOT_GIVEN, ) -> None: if is_given(language): self._opts.language = LanguageCode(language) if is_given(model): self._opts.model = _validate_model( model, language if is_given(language) else (self._opts.language or NOT_GIVEN) ) if is_given(interim_results): self._opts.interim_results = interim_results if is_given(punctuate): self._opts.punctuate = punctuate if is_given(smart_format): self._opts.smart_format = smart_format if is_given(sample_rate): self._opts.sample_rate = sample_rate if is_given(no_delay): self._opts.no_delay = no_delay if is_given(endpointing_ms): self._opts.endpointing_ms = endpointing_ms if is_given(enable_diarization): self._opts.enable_diarization = enable_diarization if is_given(filler_words): self._opts.filler_words = filler_words if is_given(keywords): self._opts.keywords = keywords if is_given(keyterms): logger.warning( "`keyterms` is deprecated, use `keyterm` instead for consistency with Deepgram API." ) keyterm = keyterms if is_given(keyterm): self._user_keyterm = [keyterm] if isinstance(keyterm, str) else list(keyterm) keyterm = list(dict.fromkeys([*self._user_keyterm, *self._session_keyterms])) self._opts.keyterm = keyterm if is_given(profanity_filter): self._opts.profanity_filter = profanity_filter if is_given(redact): self._opts.redact = redact if is_given(numerals): self._opts.numerals = numerals if is_given(mip_opt_out): self._opts.mip_opt_out = mip_opt_out if is_given(vad_events): self._opts.vad_events = vad_events if is_given(tags): self._opts.tags = _validate_tags(tags) if is_given(endpoint_url): self._opts.endpoint_url = endpoint_url if is_given(utterance_end_ms): self._opts.utterance_end_ms = utterance_end_ms if is_given(dictation): self._opts.dictation = dictation if is_given(replace): self._opts.replace = replace if is_given(search): self._opts.search = search for stream in self._streams: stream.update_options( language=language, model=model, interim_results=interim_results, punctuate=punctuate, smart_format=smart_format, sample_rate=sample_rate, no_delay=no_delay, endpointing_ms=endpointing_ms, filler_words=filler_words, keywords=keywords, keyterm=keyterm, profanity_filter=profanity_filter, redact=redact, numerals=numerals, mip_opt_out=mip_opt_out, vad_events=vad_events, endpoint_url=endpoint_url, utterance_end_ms=utterance_end_ms, dictation=dictation, replace=replace, search=search, ) 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_keyterm, *keyterms])) self._opts.keyterm = merged for stream in self._streams: if stream._speaking: # defer the reconnect to the end of the utterance so we don't cut it off stream._pending_keyterm = merged else: stream.update_options(keyterm=merged) def _sanitize_options( self, *, language: NotGivenOr[DeepgramLanguages | str] = NOT_GIVEN ) -> STTOptions: config = dataclasses.replace(self._opts) if is_given(language): config.language = LanguageCode(language) if config.detect_language: config.language = None return configHelper class that provides a standard way to create an ABC using inheritance.
Create a new instance of Deepgram STT.
Args
model- The Deepgram model to use for speech recognition. Defaults to "nova-3".
language- The language code for recognition. Defaults to "en-US".
detect_language- Whether to enable automatic language detection. Defaults to False.
interim_results- Whether to return interim (non-final) transcription results. Defaults to True.
punctuate- Whether to add punctuations to the transcription. Defaults to True. Turn detector will work better with punctuations.
smart_format- Whether to apply smart formatting to numbers, dates, etc. Defaults to False.
sample_rate- The sample rate of the audio in Hz. Defaults to 16000.
no_delay- When smart_format is used, ensures it does not wait for sequence to be complete before returning results. Defaults to True.
endpointing_ms- Time in milliseconds of silence to consider end of speech. Set to 0 to disable. Defaults to 25.
filler_words- Whether to include filler words (um, uh, etc.) in transcription. Defaults to True.
keywords- List of tuples containing keywords and their boost values for improved recognition.
Each tuple should be (keyword: str, boost: float). Defaults to None.
keywordsdoes not work with Nova-3 models. Usekeyterminstead. keyterm- str or list of str of key terms to improve recognition accuracy. Defaults to None.
keytermis only supported by Nova-3 models. tags- List of tags to add to the requests for usage reporting. Defaults to NOT_GIVEN.
profanity_filter- Whether to filter profanity from the transcription. Defaults to False.
redact- Redact sensitive information from the transcription. Accepts a single value or list of values. Supported values: "pci", "numbers", "ssn", "true" (redact all). See https://developers.deepgram.com/docs/redaction for details.
api_key- Your Deepgram API key. If not provided, will look for DEEPGRAM_API_KEY environment variable.
http_session- Optional aiohttp ClientSession to use for requests.
base_url- The base URL for Deepgram API. Defaults to "https://api.deepgram.com/v1/listen".
numerals- Whether to include numerals in the transcription. Defaults to False.
mip_opt_out- Whether to take part in the model improvement program
vad_events- Whether to enable VAD (Voice Activity Detection) events. When enabled, SpeechStarted events are sent when speech is detected. Defaults to True.
utterance_end_ms- Duration of silence in milliseconds to detect the end of an utterance and emit an UtteranceEnd event. Requires interim_results=True. See https://developers.deepgram.com/docs/understand-endpointing-interim-results
dictation- Whether to enable dictation mode which converts spoken punctuation commands (e.g. "comma", "period") into punctuation marks. Defaults to False. See https://developers.deepgram.com/reference/speech-to-text/listen-streaming#query-dictation
replace- Dictionary of terms to replace in the transcript, where keys are the original terms and values are the replacements (e.g. {"hello": "hi"}). See https://developers.deepgram.com/reference/speech-to-text/listen-streaming#query-replace
search- List of terms to search for in the transcript. Matched terms are returned with confidence scores in the response. See https://developers.deepgram.com/reference/speech-to-text/listen-streaming#query-search
Raises
ValueError- If no API key is provided or found in environment variables.
Note
The api_key must be set either through the constructor argument or by setting the DEEPGRAM_API_KEY environmental variable.
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.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 "Deepgram"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[DeepgramLanguages | str] = NOT_GIVEN,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> livekit.plugins.deepgram.stt.SpeechStream-
Expand source code
def stream( self, *, language: NotGivenOr[DeepgramLanguages | str] = NOT_GIVEN, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS, ) -> SpeechStream: config = self._sanitize_options(language=language) stream = SpeechStream( stt=self, conn_options=conn_options, opts=config, api_key=self._api_key, http_session=self._ensure_session(), base_url=self._opts.endpoint_url, ) self._streams.add(stream) return stream def update_options(self,
*,
language: NotGivenOr[DeepgramLanguages | str] = NOT_GIVEN,
model: NotGivenOr[DeepgramModels | str] = NOT_GIVEN,
interim_results: NotGivenOr[bool] = NOT_GIVEN,
punctuate: NotGivenOr[bool] = NOT_GIVEN,
smart_format: NotGivenOr[bool] = NOT_GIVEN,
sample_rate: NotGivenOr[int] = NOT_GIVEN,
no_delay: NotGivenOr[bool] = NOT_GIVEN,
endpointing_ms: NotGivenOr[int] = NOT_GIVEN,
enable_diarization: NotGivenOr[bool] = NOT_GIVEN,
filler_words: NotGivenOr[bool] = NOT_GIVEN,
keywords: NotGivenOr[list[tuple[str, float]]] = NOT_GIVEN,
keyterm: NotGivenOr[str | list[str]] = NOT_GIVEN,
profanity_filter: NotGivenOr[bool] = NOT_GIVEN,
redact: NotGivenOr[str | list[str]] = NOT_GIVEN,
numerals: NotGivenOr[bool] = NOT_GIVEN,
mip_opt_out: NotGivenOr[bool] = NOT_GIVEN,
vad_events: NotGivenOr[bool] = NOT_GIVEN,
tags: NotGivenOr[list[str]] = NOT_GIVEN,
endpoint_url: NotGivenOr[str] = NOT_GIVEN,
utterance_end_ms: NotGivenOr[int | None] = NOT_GIVEN,
dictation: NotGivenOr[bool] = NOT_GIVEN,
replace: NotGivenOr[dict[str, str] | None] = NOT_GIVEN,
search: NotGivenOr[list[str] | None] = NOT_GIVEN,
keyterms: NotGivenOr[list[str]] = NOT_GIVEN) ‑> None-
Expand source code
def update_options( self, *, language: NotGivenOr[DeepgramLanguages | str] = NOT_GIVEN, model: NotGivenOr[DeepgramModels | str] = NOT_GIVEN, interim_results: NotGivenOr[bool] = NOT_GIVEN, punctuate: NotGivenOr[bool] = NOT_GIVEN, smart_format: NotGivenOr[bool] = NOT_GIVEN, sample_rate: NotGivenOr[int] = NOT_GIVEN, no_delay: NotGivenOr[bool] = NOT_GIVEN, endpointing_ms: NotGivenOr[int] = NOT_GIVEN, enable_diarization: NotGivenOr[bool] = NOT_GIVEN, filler_words: NotGivenOr[bool] = NOT_GIVEN, keywords: NotGivenOr[list[tuple[str, float]]] = NOT_GIVEN, keyterm: NotGivenOr[str | list[str]] = NOT_GIVEN, profanity_filter: NotGivenOr[bool] = NOT_GIVEN, redact: NotGivenOr[str | list[str]] = NOT_GIVEN, numerals: NotGivenOr[bool] = NOT_GIVEN, mip_opt_out: NotGivenOr[bool] = NOT_GIVEN, vad_events: NotGivenOr[bool] = NOT_GIVEN, tags: NotGivenOr[list[str]] = NOT_GIVEN, endpoint_url: NotGivenOr[str] = NOT_GIVEN, utterance_end_ms: NotGivenOr[int | None] = NOT_GIVEN, dictation: NotGivenOr[bool] = NOT_GIVEN, replace: NotGivenOr[dict[str, str] | None] = NOT_GIVEN, search: NotGivenOr[list[str] | None] = NOT_GIVEN, # deprecated keyterms: NotGivenOr[list[str]] = NOT_GIVEN, ) -> None: if is_given(language): self._opts.language = LanguageCode(language) if is_given(model): self._opts.model = _validate_model( model, language if is_given(language) else (self._opts.language or NOT_GIVEN) ) if is_given(interim_results): self._opts.interim_results = interim_results if is_given(punctuate): self._opts.punctuate = punctuate if is_given(smart_format): self._opts.smart_format = smart_format if is_given(sample_rate): self._opts.sample_rate = sample_rate if is_given(no_delay): self._opts.no_delay = no_delay if is_given(endpointing_ms): self._opts.endpointing_ms = endpointing_ms if is_given(enable_diarization): self._opts.enable_diarization = enable_diarization if is_given(filler_words): self._opts.filler_words = filler_words if is_given(keywords): self._opts.keywords = keywords if is_given(keyterms): logger.warning( "`keyterms` is deprecated, use `keyterm` instead for consistency with Deepgram API." ) keyterm = keyterms if is_given(keyterm): self._user_keyterm = [keyterm] if isinstance(keyterm, str) else list(keyterm) keyterm = list(dict.fromkeys([*self._user_keyterm, *self._session_keyterms])) self._opts.keyterm = keyterm if is_given(profanity_filter): self._opts.profanity_filter = profanity_filter if is_given(redact): self._opts.redact = redact if is_given(numerals): self._opts.numerals = numerals if is_given(mip_opt_out): self._opts.mip_opt_out = mip_opt_out if is_given(vad_events): self._opts.vad_events = vad_events if is_given(tags): self._opts.tags = _validate_tags(tags) if is_given(endpoint_url): self._opts.endpoint_url = endpoint_url if is_given(utterance_end_ms): self._opts.utterance_end_ms = utterance_end_ms if is_given(dictation): self._opts.dictation = dictation if is_given(replace): self._opts.replace = replace if is_given(search): self._opts.search = search for stream in self._streams: stream.update_options( language=language, model=model, interim_results=interim_results, punctuate=punctuate, smart_format=smart_format, sample_rate=sample_rate, no_delay=no_delay, endpointing_ms=endpointing_ms, filler_words=filler_words, keywords=keywords, keyterm=keyterm, profanity_filter=profanity_filter, redact=redact, numerals=numerals, mip_opt_out=mip_opt_out, vad_events=vad_events, endpoint_url=endpoint_url, utterance_end_ms=utterance_end_ms, dictation=dictation, replace=replace, search=search, )
Inherited members
class STTv2 (*,
model: V2Models | str = 'flux-general-en',
sample_rate: int = 16000,
eager_eot_threshold: NotGivenOr[float] = NOT_GIVEN,
eot_threshold: NotGivenOr[float] = NOT_GIVEN,
eot_timeout_ms: NotGivenOr[int] = NOT_GIVEN,
keyterm: NotGivenOr[str | list[str]] = NOT_GIVEN,
tags: NotGivenOr[list[str]] = NOT_GIVEN,
language_hint: NotGivenOr[list[str]] = NOT_GIVEN,
api_key: NotGivenOr[str] = NOT_GIVEN,
http_session: aiohttp.ClientSession | None = None,
base_url: str = 'wss://api.deepgram.com/v2/listen',
mip_opt_out: bool = False,
keyterms: NotGivenOr[list[str]] = NOT_GIVEN)-
Expand source code
class STTv2(stt.STT): def __init__( self, *, model: V2Models | str = "flux-general-en", sample_rate: int = 16000, eager_eot_threshold: NotGivenOr[float] = NOT_GIVEN, eot_threshold: NotGivenOr[float] = NOT_GIVEN, eot_timeout_ms: NotGivenOr[int] = NOT_GIVEN, keyterm: NotGivenOr[str | list[str]] = NOT_GIVEN, tags: NotGivenOr[list[str]] = NOT_GIVEN, language_hint: NotGivenOr[list[str]] = NOT_GIVEN, api_key: NotGivenOr[str] = NOT_GIVEN, http_session: aiohttp.ClientSession | None = None, base_url: str = "wss://api.deepgram.com/v2/listen", mip_opt_out: bool = False, # deprecated keyterms: NotGivenOr[list[str]] = NOT_GIVEN, ) -> None: """Create a new instance of Deepgram STT. Args: model: The Deepgram model to use for speech recognition. Defaults to "flux-general-en". sample_rate: The sample rate of the audio in Hz. Defaults to 16000. eager_eot_threshold: The threshold for eager end of turn to enable preemptive generation. Disabled by default. Set to 0.3-0.9 to enable preemptive generation. eot_threshold: The threshold for end of speech detection, ranges 0.5-0.9. Defaults to 0.7. If using eager_eot_threshold, set this higher to allow a higher eager value. eot_timeout_ms: The timeout for end of speech detection. Defaults to 3000. keyterm: str or list of str of key terms to improve recognition accuracy. Defaults to None. tags: List of tags to add to the requests for usage reporting. Defaults to NOT_GIVEN. language_hint: List of str of language hints to bias the model for improved accuracy. Only usable with `flux-general-multi`. Defaults to NOT_GIVEN. api_key: Your Deepgram API key. If not provided, will look for DEEPGRAM_API_KEY environment variable. http_session: Optional aiohttp ClientSession to use for requests. base_url: The base URL for Deepgram API. Defaults to "https://api.deepgram.com/v1/listen". mip_opt_out: Whether to take part in the model improvement program Raises: ValueError: If no API key is provided or found in environment variables. Note: The api_key must be set either through the constructor argument or by setting the DEEPGRAM_API_KEY environmental variable. """ # noqa: E501 super().__init__( capabilities=stt.STTCapabilities( streaming=True, interim_results=True, aligned_transcript="word", offline_recognize=False, keyterms=True, ) ) deepgram_api_key = api_key if is_given(api_key) else os.environ.get("DEEPGRAM_API_KEY") if not deepgram_api_key: raise ValueError("Deepgram API key is required") self._api_key = deepgram_api_key if is_given(keyterms): logger.warning( "`keyterms` is deprecated, use `keyterm` instead for consistency with Deepgram API." ) keyterm = keyterms if is_given(eager_eot_threshold): effective_eot = eot_threshold if is_given(eot_threshold) else 0.7 if eager_eot_threshold > effective_eot: raise ValueError( f"eager_eot_threshold ({eager_eot_threshold}) must be less than or equal to eot_threshold " f"({effective_eot}); increase eot_threshold (max 0.9) to use a higher eager value" ) if language_hint and model != "flux-general-multi": logger.warning( "`language_hint` is only supported by `flux-general-multi` and will be ignored for model '%s'", model, ) self._opts = STTOptions( model=model, sample_rate=sample_rate, keyterm=([keyterm] if isinstance(keyterm, str) else list(keyterm)) if is_given(keyterm) else [], mip_opt_out=mip_opt_out, tags=_validate_tags(tags) if is_given(tags) else [], language_hint=language_hint if is_given(language_hint) else [], eager_eot_threshold=eager_eot_threshold, eot_threshold=eot_threshold, eot_timeout_ms=eot_timeout_ms, endpoint_url=base_url, ) # user keyterms; _opts.keyterm holds the effective set (user + session) self._user_keyterm: list[str] = list(self._opts.keyterm) self._session_keyterms: list[str] = [] self._session = http_session self._streams = weakref.WeakSet[SpeechStreamv2]() def _ensure_session(self) -> aiohttp.ClientSession: if not self._session: self._session = utils.http_context.http_session() return self._session async def _recognize_impl( self, buffer: AudioBuffer, *, language: NotGivenOr[str] = NOT_GIVEN, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS, ) -> stt.SpeechEvent: raise NotImplementedError( "V2 API does not support non-streaming recognize. Use with a StreamAdapter" ) @property def model(self) -> str: return self._opts.model @property def provider(self) -> str: return "Deepgram" def stream( self, *, language: NotGivenOr[str] = NOT_GIVEN, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS, ) -> SpeechStreamv2: stream = SpeechStreamv2( stt=self, conn_options=conn_options, opts=self._opts, api_key=self._api_key, http_session=self._ensure_session(), base_url=self._opts.endpoint_url, ) self._streams.add(stream) return stream def update_options( self, *, model: NotGivenOr[V2Models | str] = NOT_GIVEN, sample_rate: NotGivenOr[int] = NOT_GIVEN, eager_eot_threshold: NotGivenOr[float] = NOT_GIVEN, eot_threshold: NotGivenOr[float] = NOT_GIVEN, eot_timeout_ms: NotGivenOr[int] = NOT_GIVEN, keyterm: NotGivenOr[str | list[str]] = NOT_GIVEN, mip_opt_out: NotGivenOr[bool] = NOT_GIVEN, tags: NotGivenOr[list[str]] = NOT_GIVEN, language_hint: NotGivenOr[list[str]] = NOT_GIVEN, endpoint_url: NotGivenOr[str] = NOT_GIVEN, # deprecated keyterms: NotGivenOr[list[str]] = NOT_GIVEN, ) -> None: effective_eager = ( eager_eot_threshold if is_given(eager_eot_threshold) else self._opts.eager_eot_threshold ) effective_eot = ( eot_threshold if is_given(eot_threshold) else (self._opts.eot_threshold if is_given(self._opts.eot_threshold) else 0.7) ) if is_given(effective_eager) and effective_eager > effective_eot: raise ValueError( f"eager_eot_threshold ({effective_eager}) must be less than or equal to eot_threshold ({effective_eot})" ) if is_given(model): self._opts.model = model if is_given(sample_rate): self._opts.sample_rate = sample_rate if is_given(eot_threshold): self._opts.eot_threshold = eot_threshold if is_given(eot_timeout_ms): self._opts.eot_timeout_ms = eot_timeout_ms if is_given(keyterms): logger.warning( "`keyterms` is deprecated, use `keyterm` instead for consistency with Deepgram API." ) keyterm = keyterms if is_given(keyterm): self._user_keyterm = [keyterm] if isinstance(keyterm, str) else list(keyterm) keyterm = list(dict.fromkeys([*self._user_keyterm, *self._session_keyterms])) self._opts.keyterm = keyterm if is_given(mip_opt_out): self._opts.mip_opt_out = mip_opt_out if is_given(tags): self._opts.tags = _validate_tags(tags) if is_given(language_hint): self._opts.language_hint = language_hint if language_hint and self._opts.model != "flux-general-multi": logger.warning( "`language_hint` is only supported by `flux-general-multi` and will be ignored for model '%s'", self._opts.model, ) if is_given(endpoint_url): self._opts.endpoint_url = endpoint_url if is_given(eager_eot_threshold): self._opts.eager_eot_threshold = eager_eot_threshold for stream in self._streams: stream.update_options( model=model, sample_rate=sample_rate, eot_threshold=eot_threshold, eot_timeout_ms=eot_timeout_ms, keyterm=keyterm, mip_opt_out=mip_opt_out, endpoint_url=endpoint_url, tags=tags, language_hint=language_hint, eager_eot_threshold=eager_eot_threshold, ) 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_keyterm, *keyterms])) self._opts.keyterm = merged for stream in self._streams: # tuned in-band, safe to apply mid-utterance stream.update_options(keyterm=merged)Helper class that provides a standard way to create an ABC using inheritance.
Create a new instance of Deepgram STT.
Args
model- The Deepgram model to use for speech recognition. Defaults to "flux-general-en".
sample_rate- The sample rate of the audio in Hz. Defaults to 16000.
eager_eot_threshold- The threshold for eager end of turn to enable preemptive generation. Disabled by default. Set to 0.3-0.9 to enable preemptive generation.
eot_threshold- The threshold for end of speech detection, ranges 0.5-0.9. Defaults to 0.7. If using eager_eot_threshold, set this higher to allow a higher eager value.
eot_timeout_ms- The timeout for end of speech detection. Defaults to 3000.
keyterm- str or list of str of key terms to improve recognition accuracy. Defaults to None.
tags- List of tags to add to the requests for usage reporting. Defaults to NOT_GIVEN.
language_hint- List of str of language hints to bias the model for improved accuracy. Only usable with
flux-general-multi. Defaults to NOT_GIVEN. api_key- Your Deepgram API key. If not provided, will look for DEEPGRAM_API_KEY environment variable.
http_session- Optional aiohttp ClientSession to use for requests.
base_url- The base URL for Deepgram API. Defaults to "https://api.deepgram.com/v1/listen".
mip_opt_out- Whether to take part in the model improvement program
Raises
ValueError- If no API key is provided or found in environment variables.
Note
The api_key must be set either through the constructor argument or by setting the DEEPGRAM_API_KEY environmental variable.
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.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 "Deepgram"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.deepgram.stt_v2.SpeechStreamv2-
Expand source code
def stream( self, *, language: NotGivenOr[str] = NOT_GIVEN, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS, ) -> SpeechStreamv2: stream = SpeechStreamv2( stt=self, conn_options=conn_options, opts=self._opts, api_key=self._api_key, http_session=self._ensure_session(), base_url=self._opts.endpoint_url, ) self._streams.add(stream) return stream def update_options(self,
*,
model: NotGivenOr[V2Models | str] = NOT_GIVEN,
sample_rate: NotGivenOr[int] = NOT_GIVEN,
eager_eot_threshold: NotGivenOr[float] = NOT_GIVEN,
eot_threshold: NotGivenOr[float] = NOT_GIVEN,
eot_timeout_ms: NotGivenOr[int] = NOT_GIVEN,
keyterm: NotGivenOr[str | list[str]] = NOT_GIVEN,
mip_opt_out: NotGivenOr[bool] = NOT_GIVEN,
tags: NotGivenOr[list[str]] = NOT_GIVEN,
language_hint: NotGivenOr[list[str]] = NOT_GIVEN,
endpoint_url: NotGivenOr[str] = NOT_GIVEN,
keyterms: NotGivenOr[list[str]] = NOT_GIVEN) ‑> None-
Expand source code
def update_options( self, *, model: NotGivenOr[V2Models | str] = NOT_GIVEN, sample_rate: NotGivenOr[int] = NOT_GIVEN, eager_eot_threshold: NotGivenOr[float] = NOT_GIVEN, eot_threshold: NotGivenOr[float] = NOT_GIVEN, eot_timeout_ms: NotGivenOr[int] = NOT_GIVEN, keyterm: NotGivenOr[str | list[str]] = NOT_GIVEN, mip_opt_out: NotGivenOr[bool] = NOT_GIVEN, tags: NotGivenOr[list[str]] = NOT_GIVEN, language_hint: NotGivenOr[list[str]] = NOT_GIVEN, endpoint_url: NotGivenOr[str] = NOT_GIVEN, # deprecated keyterms: NotGivenOr[list[str]] = NOT_GIVEN, ) -> None: effective_eager = ( eager_eot_threshold if is_given(eager_eot_threshold) else self._opts.eager_eot_threshold ) effective_eot = ( eot_threshold if is_given(eot_threshold) else (self._opts.eot_threshold if is_given(self._opts.eot_threshold) else 0.7) ) if is_given(effective_eager) and effective_eager > effective_eot: raise ValueError( f"eager_eot_threshold ({effective_eager}) must be less than or equal to eot_threshold ({effective_eot})" ) if is_given(model): self._opts.model = model if is_given(sample_rate): self._opts.sample_rate = sample_rate if is_given(eot_threshold): self._opts.eot_threshold = eot_threshold if is_given(eot_timeout_ms): self._opts.eot_timeout_ms = eot_timeout_ms if is_given(keyterms): logger.warning( "`keyterms` is deprecated, use `keyterm` instead for consistency with Deepgram API." ) keyterm = keyterms if is_given(keyterm): self._user_keyterm = [keyterm] if isinstance(keyterm, str) else list(keyterm) keyterm = list(dict.fromkeys([*self._user_keyterm, *self._session_keyterms])) self._opts.keyterm = keyterm if is_given(mip_opt_out): self._opts.mip_opt_out = mip_opt_out if is_given(tags): self._opts.tags = _validate_tags(tags) if is_given(language_hint): self._opts.language_hint = language_hint if language_hint and self._opts.model != "flux-general-multi": logger.warning( "`language_hint` is only supported by `flux-general-multi` and will be ignored for model '%s'", self._opts.model, ) if is_given(endpoint_url): self._opts.endpoint_url = endpoint_url if is_given(eager_eot_threshold): self._opts.eager_eot_threshold = eager_eot_threshold for stream in self._streams: stream.update_options( model=model, sample_rate=sample_rate, eot_threshold=eot_threshold, eot_timeout_ms=eot_timeout_ms, keyterm=keyterm, mip_opt_out=mip_opt_out, endpoint_url=endpoint_url, tags=tags, language_hint=language_hint, eager_eot_threshold=eager_eot_threshold, )
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): _KEEPALIVE_MSG: str = json.dumps({"type": "KeepAlive"}) _CLOSE_MSG: str = json.dumps({"type": "CloseStream"}) _FINALIZE_MSG: str = json.dumps({"type": "Finalize"}) def __init__( self, *, stt: STT, opts: STTOptions, conn_options: APIConnectOptions, api_key: str, http_session: aiohttp.ClientSession, base_url: str, ) -> None: if opts.detect_language or opts.language is None: raise ValueError( "language detection is not supported in streaming mode, " "please disable it and specify a language" ) 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._opts.endpoint_url = base_url self._speaking = False self._audio_duration_collector = PeriodicCollector( callback=self._on_audio_duration_report, duration=5.0, ) self._request_id = "" self._reconnect_event = asyncio.Event() # keyterms set while the user is speaking; applied at END_OF_SPEECH (latest wins) self._pending_keyterm: list[str] | None = None # Track how much duration has already been reported so we can emit # the connection-lifetime remainder on close, matching what Deepgram # actually bills (which includes WebSocket open/teardown overhead # beyond the pushed audio frames). self._reported_duration: float = 0.0 def update_options( self, *, language: NotGivenOr[DeepgramLanguages | str] = NOT_GIVEN, model: NotGivenOr[DeepgramModels | str] = NOT_GIVEN, interim_results: NotGivenOr[bool] = NOT_GIVEN, punctuate: NotGivenOr[bool] = NOT_GIVEN, smart_format: NotGivenOr[bool] = NOT_GIVEN, sample_rate: NotGivenOr[int] = NOT_GIVEN, no_delay: NotGivenOr[bool] = NOT_GIVEN, endpointing_ms: NotGivenOr[int] = NOT_GIVEN, enable_diarization: NotGivenOr[bool] = NOT_GIVEN, filler_words: NotGivenOr[bool] = NOT_GIVEN, keywords: NotGivenOr[list[tuple[str, float]]] = NOT_GIVEN, keyterm: NotGivenOr[str | list[str]] = NOT_GIVEN, profanity_filter: NotGivenOr[bool] = NOT_GIVEN, redact: NotGivenOr[str | list[str]] = NOT_GIVEN, numerals: NotGivenOr[bool] = NOT_GIVEN, mip_opt_out: NotGivenOr[bool] = NOT_GIVEN, vad_events: NotGivenOr[bool] = NOT_GIVEN, tags: NotGivenOr[list[str]] = NOT_GIVEN, endpoint_url: NotGivenOr[str] = NOT_GIVEN, utterance_end_ms: NotGivenOr[int | None] = NOT_GIVEN, dictation: NotGivenOr[bool] = NOT_GIVEN, replace: NotGivenOr[dict[str, str] | None] = NOT_GIVEN, search: NotGivenOr[list[str] | None] = NOT_GIVEN, # deprecated keyterms: NotGivenOr[list[str]] = NOT_GIVEN, ) -> None: if is_given(language): self._opts.language = LanguageCode(language) if is_given(model): self._opts.model = _validate_model( model, language if is_given(language) else (self._opts.language or NOT_GIVEN) ) if is_given(interim_results): self._opts.interim_results = interim_results if is_given(punctuate): self._opts.punctuate = punctuate if is_given(smart_format): self._opts.smart_format = smart_format if is_given(sample_rate): self._opts.sample_rate = sample_rate if is_given(no_delay): self._opts.no_delay = no_delay if is_given(endpointing_ms): self._opts.endpointing_ms = endpointing_ms if is_given(enable_diarization): self._opts.enable_diarization = enable_diarization if is_given(filler_words): self._opts.filler_words = filler_words if is_given(keywords): self._opts.keywords = keywords if is_given(keyterms): logger.warning( "`keyterms` is deprecated, use `keyterm` instead for consistency with Deepgram API." ) keyterm = keyterms if is_given(keyterm): self._opts.keyterm = keyterm self._pending_keyterm = None if is_given(profanity_filter): self._opts.profanity_filter = profanity_filter if is_given(redact): self._opts.redact = redact if is_given(numerals): self._opts.numerals = numerals if is_given(mip_opt_out): self._opts.mip_opt_out = mip_opt_out if is_given(vad_events): self._opts.vad_events = vad_events if is_given(tags): self._opts.tags = _validate_tags(tags) if is_given(endpoint_url): self._opts.endpoint_url = endpoint_url if is_given(utterance_end_ms): self._opts.utterance_end_ms = utterance_end_ms if is_given(dictation): self._opts.dictation = dictation if is_given(replace): self._opts.replace = replace if is_given(search): self._opts.search = search self._reconnect_event.set() def _on_end_of_speech(self) -> None: if self._pending_keyterm is not None: self.update_options(keyterm=self._pending_keyterm) self._pending_keyterm = None async def _run(self) -> None: closing_ws = False async def keepalive_task(ws: aiohttp.ClientWebSocketResponse) -> None: # if we want to keep the connection alive even if no audio is sent, # Deepgram expects a keepalive message. # https://developers.deepgram.com/reference/listen-live#stream-keepalive try: while True: await ws.send_str(SpeechStream._KEEPALIVE_MSG) await asyncio.sleep(5) except Exception as e: logger.warning(f"Deepgram keepalive task exited: {e}") return @utils.log_exceptions(logger=logger) async def send_task(ws: aiohttp.ClientWebSocketResponse) -> None: nonlocal closing_ws # forward audio to deepgram in chunks of 50ms samples_50ms = self._opts.sample_rate // 20 audio_bstream = utils.audio.AudioByteStream( sample_rate=self._opts.sample_rate, num_channels=self._opts.num_channels, samples_per_channel=samples_50ms, ) has_ended = False try: async for data in self._input_ch: 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) await ws.send_bytes(frame.data.tobytes()) if has_ended: self._audio_duration_collector.flush() await ws.send_str(SpeechStream._FINALIZE_MSG) has_ended = False # tell deepgram we are done sending audio/inputs closing_ws = True await ws.send_str(SpeechStream._CLOSE_MSG) except (aiohttp.ClientError, ConnectionError) as e: # a mid-write socket drop surfaces here as a raw connection error. # if the close is expected (aclose or the http session closing) just # return; otherwise re-raise as a retryable APIError so _main_task # reconnects, symmetric with recv_task. if closing_ws or self._session.closed: return raise APIConnectionError("deepgram 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, ): # close is expected, see SpeechStream.aclose # or when the agent session ends, the http session is closed if closing_ws or self._session.closed: return # this will trigger a reconnection, see the _run loop raise APIStatusError( message="deepgram 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 deepgram message type %s", msg.type) continue try: self._process_stream_event(json.loads(msg.data)) except Exception: logger.exception("failed to process deepgram message") ws: aiohttp.ClientWebSocketResponse | None = None while True: conn_start_time = 0.0 try: ws = await self._connect_ws() conn_start_time = time.perf_counter() 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, ) # propagate exceptions from completed tasks 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 the exception finally: if ws is not None: await ws.close() # Deepgram bills WebSocket lifetime, not just audio # frames pushed. Emit the remainder between the # connection's wall-clock lifetime and the frame # durations we've already reported so usage reflects # what the provider actually charges for. if conn_start_time: self._audio_duration_collector.flush() lifetime = time.perf_counter() - conn_start_time remainder = lifetime - self._reported_duration if remainder > 0: self._on_audio_duration_report(remainder) self._reported_duration = 0.0 async def _connect_ws(self) -> aiohttp.ClientWebSocketResponse: live_config: dict[str, Any] = { "model": self._opts.model, "punctuate": self._opts.punctuate, "smart_format": self._opts.smart_format, "no_delay": self._opts.no_delay, "interim_results": self._opts.interim_results, "encoding": "linear16", "vad_events": self._opts.vad_events, "sample_rate": self._opts.sample_rate, "channels": self._opts.num_channels, "endpointing": False if self._opts.endpointing_ms == 0 else self._opts.endpointing_ms, "filler_words": self._opts.filler_words, "profanity_filter": self._opts.profanity_filter, "numerals": self._opts.numerals, "mip_opt_out": self._opts.mip_opt_out, } if self._opts.enable_diarization: live_config["diarize"] = True if self._opts.keywords: live_config["keywords"] = self._opts.keywords if self._opts.keyterm: live_config["keyterm"] = self._opts.keyterm if self._opts.utterance_end_ms is not None: live_config["utterance_end_ms"] = self._opts.utterance_end_ms if self._opts.dictation: live_config["dictation"] = True if self._opts.replace: live_config["replace"] = self._opts.replace if self._opts.search: live_config["search"] = self._opts.search if self._opts.language: live_config["language"] = self._opts.language if self._opts.redact: live_config["redact"] = self._opts.redact if self._opts.tags: live_config["tag"] = self._opts.tags t0 = time.perf_counter() try: ws = await asyncio.wait_for( self._session.ws_connect( _to_deepgram_url(live_config, base_url=self._opts.endpoint_url, websocket=True), headers={"Authorization": f"Token {self._api_key}"}, ), self._conn_options.timeout, ) self._report_connection_acquired(time.perf_counter() - t0, False) ws_headers = { k: v for k, v in ws._response.headers.items() if k.startswith("dg-") or k == "Date" } logger.debug( "Established new Deepgram STT WebSocket connection:", extra={"headers": ws_headers}, ) except (aiohttp.ClientConnectorError, asyncio.TimeoutError) as e: raise APIConnectionError("failed to connect to deepgram") from e return ws def _on_audio_duration_report(self, duration: float) -> None: self._reported_duration += duration usage_event = stt.SpeechEvent( type=stt.SpeechEventType.RECOGNITION_USAGE, request_id=self._request_id, alternatives=[], recognition_usage=stt.RecognitionUsage(audio_duration=duration), ) self._event_ch.send_nowait(usage_event) def _process_stream_event(self, data: dict) -> None: assert self._opts.language is not None if data["type"] == "SpeechStarted": # This is a normal case. Deepgram's SpeechStarted events # are not correlated with speech_final or utterance end. # It's possible that we receive two in a row without an endpoint # It's also possible we receive a transcript without a SpeechStarted event. if self._speaking: return self._speaking = True start_event = stt.SpeechEvent(type=stt.SpeechEventType.START_OF_SPEECH) self._event_ch.send_nowait(start_event) # see this page: # https://developers.deepgram.com/docs/understand-endpointing-interim-results#using-endpointing-speech_final # for more information about the different types of events elif data["type"] == "Results": metadata = data["metadata"] request_id = metadata["request_id"] is_final_transcript = data["is_final"] is_endpoint = data["speech_final"] self._request_id = request_id alts = live_transcription_to_speech_data( self._opts.language, data, is_final=is_final_transcript, start_time_offset=self.start_time_offset, ) # If, for some reason, we didn't get a SpeechStarted event but we got # a transcript with text, we should start speaking. It's rare but has # been observed. if len(alts) > 0 and alts[0].text: if not self._speaking: self._speaking = True start_event = stt.SpeechEvent(type=stt.SpeechEventType.START_OF_SPEECH) self._event_ch.send_nowait(start_event) if is_final_transcript: final_event = stt.SpeechEvent( type=stt.SpeechEventType.FINAL_TRANSCRIPT, request_id=request_id, alternatives=alts, ) self._event_ch.send_nowait(final_event) else: interim_event = stt.SpeechEvent( type=stt.SpeechEventType.INTERIM_TRANSCRIPT, request_id=request_id, alternatives=alts, ) self._event_ch.send_nowait(interim_event) # if we receive an endpoint, only end the speech if # we either had a SpeechStarted event or we have a seen # a non-empty transcript (deepgram doesn't have a SpeechEnded event) if is_endpoint and self._speaking: self._speaking = False self._event_ch.send_nowait(stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH)) self._on_end_of_speech() elif data["type"] == "UtteranceEnd": # Fired when utterance_end_ms is set and the configured silence duration has elapsed. # https://developers.deepgram.com/docs/understand-endpointing-interim-results if self._speaking: self._speaking = False self._event_ch.send_nowait(stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH)) self._on_end_of_speech() elif data["type"] == "Metadata": pass # metadata is too noisy else: logger.warning("received unexpected message from deepgram %s", data)Helper class that provides a standard way to create an ABC using inheritance.
Args: sample_rate : int or None, optional The desired sample rate for the audio input. If specified, the audio input will be automatically resampled to match the given sample rate before being processed for Speech-to-Text. If not provided (None), the input will retain its original sample rate.
Ancestors
- livekit.agents.stt.stt.RecognizeStream
- abc.ABC
Methods
def update_options(self,
*,
language: NotGivenOr[DeepgramLanguages | str] = NOT_GIVEN,
model: NotGivenOr[DeepgramModels | str] = NOT_GIVEN,
interim_results: NotGivenOr[bool] = NOT_GIVEN,
punctuate: NotGivenOr[bool] = NOT_GIVEN,
smart_format: NotGivenOr[bool] = NOT_GIVEN,
sample_rate: NotGivenOr[int] = NOT_GIVEN,
no_delay: NotGivenOr[bool] = NOT_GIVEN,
endpointing_ms: NotGivenOr[int] = NOT_GIVEN,
enable_diarization: NotGivenOr[bool] = NOT_GIVEN,
filler_words: NotGivenOr[bool] = NOT_GIVEN,
keywords: NotGivenOr[list[tuple[str, float]]] = NOT_GIVEN,
keyterm: NotGivenOr[str | list[str]] = NOT_GIVEN,
profanity_filter: NotGivenOr[bool] = NOT_GIVEN,
redact: NotGivenOr[str | list[str]] = NOT_GIVEN,
numerals: NotGivenOr[bool] = NOT_GIVEN,
mip_opt_out: NotGivenOr[bool] = NOT_GIVEN,
vad_events: NotGivenOr[bool] = NOT_GIVEN,
tags: NotGivenOr[list[str]] = NOT_GIVEN,
endpoint_url: NotGivenOr[str] = NOT_GIVEN,
utterance_end_ms: NotGivenOr[int | None] = NOT_GIVEN,
dictation: NotGivenOr[bool] = NOT_GIVEN,
replace: NotGivenOr[dict[str, str] | None] = NOT_GIVEN,
search: NotGivenOr[list[str] | None] = NOT_GIVEN,
keyterms: NotGivenOr[list[str]] = NOT_GIVEN) ‑> None-
Expand source code
def update_options( self, *, language: NotGivenOr[DeepgramLanguages | str] = NOT_GIVEN, model: NotGivenOr[DeepgramModels | str] = NOT_GIVEN, interim_results: NotGivenOr[bool] = NOT_GIVEN, punctuate: NotGivenOr[bool] = NOT_GIVEN, smart_format: NotGivenOr[bool] = NOT_GIVEN, sample_rate: NotGivenOr[int] = NOT_GIVEN, no_delay: NotGivenOr[bool] = NOT_GIVEN, endpointing_ms: NotGivenOr[int] = NOT_GIVEN, enable_diarization: NotGivenOr[bool] = NOT_GIVEN, filler_words: NotGivenOr[bool] = NOT_GIVEN, keywords: NotGivenOr[list[tuple[str, float]]] = NOT_GIVEN, keyterm: NotGivenOr[str | list[str]] = NOT_GIVEN, profanity_filter: NotGivenOr[bool] = NOT_GIVEN, redact: NotGivenOr[str | list[str]] = NOT_GIVEN, numerals: NotGivenOr[bool] = NOT_GIVEN, mip_opt_out: NotGivenOr[bool] = NOT_GIVEN, vad_events: NotGivenOr[bool] = NOT_GIVEN, tags: NotGivenOr[list[str]] = NOT_GIVEN, endpoint_url: NotGivenOr[str] = NOT_GIVEN, utterance_end_ms: NotGivenOr[int | None] = NOT_GIVEN, dictation: NotGivenOr[bool] = NOT_GIVEN, replace: NotGivenOr[dict[str, str] | None] = NOT_GIVEN, search: NotGivenOr[list[str] | None] = NOT_GIVEN, # deprecated keyterms: NotGivenOr[list[str]] = NOT_GIVEN, ) -> None: if is_given(language): self._opts.language = LanguageCode(language) if is_given(model): self._opts.model = _validate_model( model, language if is_given(language) else (self._opts.language or NOT_GIVEN) ) if is_given(interim_results): self._opts.interim_results = interim_results if is_given(punctuate): self._opts.punctuate = punctuate if is_given(smart_format): self._opts.smart_format = smart_format if is_given(sample_rate): self._opts.sample_rate = sample_rate if is_given(no_delay): self._opts.no_delay = no_delay if is_given(endpointing_ms): self._opts.endpointing_ms = endpointing_ms if is_given(enable_diarization): self._opts.enable_diarization = enable_diarization if is_given(filler_words): self._opts.filler_words = filler_words if is_given(keywords): self._opts.keywords = keywords if is_given(keyterms): logger.warning( "`keyterms` is deprecated, use `keyterm` instead for consistency with Deepgram API." ) keyterm = keyterms if is_given(keyterm): self._opts.keyterm = keyterm self._pending_keyterm = None if is_given(profanity_filter): self._opts.profanity_filter = profanity_filter if is_given(redact): self._opts.redact = redact if is_given(numerals): self._opts.numerals = numerals if is_given(mip_opt_out): self._opts.mip_opt_out = mip_opt_out if is_given(vad_events): self._opts.vad_events = vad_events if is_given(tags): self._opts.tags = _validate_tags(tags) if is_given(endpoint_url): self._opts.endpoint_url = endpoint_url if is_given(utterance_end_ms): self._opts.utterance_end_ms = utterance_end_ms if is_given(dictation): self._opts.dictation = dictation if is_given(replace): self._opts.replace = replace if is_given(search): self._opts.search = search self._reconnect_event.set()
class SpeechStreamv2 (*,
stt: STTv2,
opts: STTOptions,
conn_options: APIConnectOptions,
api_key: str,
http_session: aiohttp.ClientSession,
base_url: str)-
Expand source code
class SpeechStreamv2(stt.SpeechStream): # _KEEPALIVE_MSG: str = json.dumps({"type": "KeepAlive"}) _CLOSE_MSG: str = json.dumps({"type": "CloseStream"}) # _FINALIZE_MSG: str = json.dumps({"type": "Finalize"}) def __init__( self, *, stt: STTv2, 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._opts.endpoint_url = base_url self._speaking = False self._audio_duration_collector = PeriodicCollector( callback=self._on_audio_duration_report, duration=5.0, ) self._request_id = "" self._reconnect_event = asyncio.Event() # active connection for in-band Configure updates; None while disconnected self._ws: aiohttp.ClientWebSocketResponse | None = None self._reconfigure_atask: asyncio.Task[None] | None = None def update_options( self, *, model: NotGivenOr[V2Models | str] = NOT_GIVEN, sample_rate: NotGivenOr[int] = NOT_GIVEN, eot_threshold: NotGivenOr[float] = NOT_GIVEN, eot_timeout_ms: NotGivenOr[int] = NOT_GIVEN, keyterm: NotGivenOr[str | list[str]] = NOT_GIVEN, mip_opt_out: NotGivenOr[bool] = NOT_GIVEN, tags: NotGivenOr[list[str]] = NOT_GIVEN, language_hint: NotGivenOr[list[str]] = NOT_GIVEN, endpoint_url: NotGivenOr[str] = NOT_GIVEN, eager_eot_threshold: NotGivenOr[float] = NOT_GIVEN, # deprecated keyterms: NotGivenOr[list[str]] = NOT_GIVEN, ) -> None: if is_given(model): self._opts.model = model if is_given(sample_rate): self._opts.sample_rate = sample_rate if is_given(eot_threshold): self._opts.eot_threshold = eot_threshold if is_given(eot_timeout_ms): self._opts.eot_timeout_ms = eot_timeout_ms if is_given(keyterms): logger.warning( "`keyterms` is deprecated, use `keyterm` instead for consistency with Deepgram API." ) keyterm = keyterms if is_given(keyterm): self._opts.keyterm = keyterm if is_given(mip_opt_out): self._opts.mip_opt_out = mip_opt_out if is_given(tags): self._opts.tags = _validate_tags(tags) if is_given(language_hint): self._opts.language_hint = language_hint if is_given(endpoint_url): self._opts.endpoint_url = endpoint_url if is_given(eager_eot_threshold): self._opts.eager_eot_threshold = eager_eot_threshold # these only take effect on a fresh connection needs_reconnect = any( is_given(opt) for opt in (model, sample_rate, mip_opt_out, tags, endpoint_url) ) if needs_reconnect: # reconnect carries the latest options self._reconnect_event.set() return # send only changed fields; Flux keeps omitted ones unchanged # https://developers.deepgram.com/docs/flux/configure thresholds: dict[str, Any] = {} if is_given(eager_eot_threshold): thresholds["eager_eot_threshold"] = eager_eot_threshold if is_given(eot_threshold): thresholds["eot_threshold"] = eot_threshold if is_given(eot_timeout_ms): thresholds["eot_timeout_ms"] = eot_timeout_ms changed_options: dict[str, Any] = {} if thresholds: changed_options["thresholds"] = thresholds if is_given(keyterm): # keyterms replaces the whole list, so send the full effective set changed_options["keyterms"] = self._opts.keyterm if is_given(language_hint): changed_options["language_hints"] = self._opts.language_hint if changed_options: # chain off the previous send so deltas reach the server in order self._reconfigure_atask = asyncio.create_task( self._send_configure(changed_options, self._reconfigure_atask) ) async def _send_configure( self, options: dict[str, Any], prev: asyncio.Task[None] | None ) -> None: if prev is not None: await asyncio.gather(prev, return_exceptions=True) ws = self._ws if ws is None or ws.closed: # not connected; next connection carries the latest options return try: await ws.send_str(json.dumps({"type": "Configure", **options})) except Exception: # closing; next connection carries the latest options logger.debug("failed to send Configure to deepgram") async def _run(self) -> None: closing_ws = False # async def keepalive_task(ws: aiohttp.ClientWebSocketResponse) -> None: # # if we want to keep the connection alive even if no audio is sent, # # Deepgram expects a keepalive message. # # https://developers.deepgram.com/reference/listen-live#stream-keepalive # try: # while True: # await ws.send_str(SpeechStream._KEEPALIVE_MSG) # await asyncio.sleep(5) # except Exception: # return @utils.log_exceptions(logger=logger) async def send_task(ws: aiohttp.ClientWebSocketResponse) -> None: nonlocal closing_ws # forward audio to deepgram in chunks of 50ms samples_50ms = self._opts.sample_rate // 20 audio_bstream = utils.audio.AudioByteStream( sample_rate=self._opts.sample_rate, num_channels=1, samples_per_channel=samples_50ms, ) has_ended = False try: async for data in self._input_ch: 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) await ws.send_bytes(frame.data.tobytes()) if has_ended: self._audio_duration_collector.flush() has_ended = False # tell deepgram we are done sending audio/inputs closing_ws = True await ws.send_str(SpeechStreamv2._CLOSE_MSG) except (aiohttp.ClientError, ConnectionError) as e: # a mid-write socket drop surfaces here as a raw connection error. # if the close is expected (aclose or the http session closing) just # return; otherwise re-raise as a retryable APIError so _main_task # reconnects, symmetric with recv_task. if closing_ws or self._session.closed: return raise APIConnectionError("deepgram 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, ): # close is expected, see SpeechStream.aclose # or when the agent session ends, the http session is closed if closing_ws or self._session.closed: return # this will trigger a reconnection, see the _run loop raise APIStatusError( message="deepgram 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 deepgram message type %s", msg.type) continue try: self._process_stream_event(json.loads(msg.data)) except Exception: logger.exception("failed to process deepgram message") ws: aiohttp.ClientWebSocketResponse | None = None while True: try: ws = await self._connect_ws() # expose the connection for in-band Configure updates self._ws = ws 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, ) # propagate exceptions from completed tasks 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 the exception finally: self._ws = None if self._reconfigure_atask is not None: await utils.aio.gracefully_cancel(self._reconfigure_atask) self._reconfigure_atask = None if ws is not None: await ws.close() async def _connect_ws(self) -> aiohttp.ClientWebSocketResponse: live_config: dict[str, Any] = { "model": self._opts.model, "sample_rate": self._opts.sample_rate, "encoding": "linear16", "mip_opt_out": self._opts.mip_opt_out, } if self._opts.eager_eot_threshold: live_config["eager_eot_threshold"] = self._opts.eager_eot_threshold if self._opts.eot_threshold: live_config["eot_threshold"] = self._opts.eot_threshold if self._opts.eot_timeout_ms: live_config["eot_timeout_ms"] = self._opts.eot_timeout_ms if self._opts.keyterm: live_config["keyterm"] = self._opts.keyterm if self._opts.tags: live_config["tag"] = self._opts.tags if self._opts.language_hint: live_config["language_hint"] = self._opts.language_hint try: ws = await asyncio.wait_for( self._session.ws_connect( _to_deepgram_url(live_config, base_url=self._opts.endpoint_url, websocket=True), headers={"Authorization": f"Token {self._api_key}"}, heartbeat=30.0, ), self._conn_options.timeout, ) ws_headers = { k: v for k, v in ws._response.headers.items() if k.startswith("dg-") or k == "Date" } logger.debug( "Established new Deepgram STT WebSocket connection:", extra={"headers": ws_headers}, ) except (aiohttp.ClientConnectorError, asyncio.TimeoutError) as e: raise APIConnectionError("failed to connect to deepgram") from e return ws def _on_audio_duration_report(self, duration: float) -> None: usage_event = stt.SpeechEvent( type=stt.SpeechEventType.RECOGNITION_USAGE, request_id=self._request_id, alternatives=[], recognition_usage=stt.RecognitionUsage(audio_duration=duration), ) self._event_ch.send_nowait(usage_event) def _send_transcript_event(self, event_type: stt.SpeechEventType, data: dict) -> None: alts = _parse_transcription(self._opts.language, data, self.start_time_offset) if alts: event = stt.SpeechEvent( type=event_type, request_id=self._request_id, alternatives=alts, ) self._event_ch.send_nowait(event) def _process_stream_event(self, data: dict) -> None: assert self._opts.language is not None if request_id := data.get("request_id"): self._request_id = request_id if data["type"] == "TurnInfo": event_type = data["event"] if event_type == "StartOfTurn": if self._speaking: return self._speaking = True start_event = stt.SpeechEvent(type=stt.SpeechEventType.START_OF_SPEECH) self._event_ch.send_nowait(start_event) self._send_transcript_event(stt.SpeechEventType.INTERIM_TRANSCRIPT, data) elif event_type == "Update": if not self._speaking: return self._send_transcript_event(stt.SpeechEventType.INTERIM_TRANSCRIPT, data) elif event_type == "EagerEndOfTurn": # technically, a pause in speech is detected. for lifecycle purposes, # we are assuming the user is still speaking and sending a preflight event to # start preemptive synthesis. if not self._speaking: return self._send_transcript_event(stt.SpeechEventType.PREFLIGHT_TRANSCRIPT, data) elif event_type == "TurnResumed": # sending interim transcript will abort eager end of turn self._send_transcript_event(stt.SpeechEventType.INTERIM_TRANSCRIPT, data) elif event_type == "EndOfTurn": if not self._speaking: return self._speaking = False self._send_transcript_event(stt.SpeechEventType.FINAL_TRANSCRIPT, data) end_event = stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH) self._event_ch.send_nowait(end_event) elif data["type"] == "ConfigureSuccess": logger.debug("deepgram applied Configure update", extra={"data": data}) elif data["type"] == "ConfigureFailure": logger.warning("deepgram rejected Configure update", extra={"data": data}) elif data["type"] == "Error": logger.warning("deepgram sent an error", extra={"data": data}) desc = data.get("description") or "unknown error from deepgram" code = -1 raise APIStatusError(message=desc, status_code=code)Helper class that provides a standard way to create an ABC using inheritance.
Args: sample_rate : int or None, optional The desired sample rate for the audio input. If specified, the audio input will be automatically resampled to match the given sample rate before being processed for Speech-to-Text. If not provided (None), the input will retain its original sample rate.
Ancestors
- livekit.agents.stt.stt.RecognizeStream
- abc.ABC
Methods
def update_options(self,
*,
model: NotGivenOr[V2Models | str] = NOT_GIVEN,
sample_rate: NotGivenOr[int] = NOT_GIVEN,
eot_threshold: NotGivenOr[float] = NOT_GIVEN,
eot_timeout_ms: NotGivenOr[int] = NOT_GIVEN,
keyterm: NotGivenOr[str | list[str]] = NOT_GIVEN,
mip_opt_out: NotGivenOr[bool] = NOT_GIVEN,
tags: NotGivenOr[list[str]] = NOT_GIVEN,
language_hint: NotGivenOr[list[str]] = NOT_GIVEN,
endpoint_url: NotGivenOr[str] = NOT_GIVEN,
eager_eot_threshold: NotGivenOr[float] = NOT_GIVEN,
keyterms: NotGivenOr[list[str]] = NOT_GIVEN) ‑> None-
Expand source code
def update_options( self, *, model: NotGivenOr[V2Models | str] = NOT_GIVEN, sample_rate: NotGivenOr[int] = NOT_GIVEN, eot_threshold: NotGivenOr[float] = NOT_GIVEN, eot_timeout_ms: NotGivenOr[int] = NOT_GIVEN, keyterm: NotGivenOr[str | list[str]] = NOT_GIVEN, mip_opt_out: NotGivenOr[bool] = NOT_GIVEN, tags: NotGivenOr[list[str]] = NOT_GIVEN, language_hint: NotGivenOr[list[str]] = NOT_GIVEN, endpoint_url: NotGivenOr[str] = NOT_GIVEN, eager_eot_threshold: NotGivenOr[float] = NOT_GIVEN, # deprecated keyterms: NotGivenOr[list[str]] = NOT_GIVEN, ) -> None: if is_given(model): self._opts.model = model if is_given(sample_rate): self._opts.sample_rate = sample_rate if is_given(eot_threshold): self._opts.eot_threshold = eot_threshold if is_given(eot_timeout_ms): self._opts.eot_timeout_ms = eot_timeout_ms if is_given(keyterms): logger.warning( "`keyterms` is deprecated, use `keyterm` instead for consistency with Deepgram API." ) keyterm = keyterms if is_given(keyterm): self._opts.keyterm = keyterm if is_given(mip_opt_out): self._opts.mip_opt_out = mip_opt_out if is_given(tags): self._opts.tags = _validate_tags(tags) if is_given(language_hint): self._opts.language_hint = language_hint if is_given(endpoint_url): self._opts.endpoint_url = endpoint_url if is_given(eager_eot_threshold): self._opts.eager_eot_threshold = eager_eot_threshold # these only take effect on a fresh connection needs_reconnect = any( is_given(opt) for opt in (model, sample_rate, mip_opt_out, tags, endpoint_url) ) if needs_reconnect: # reconnect carries the latest options self._reconnect_event.set() return # send only changed fields; Flux keeps omitted ones unchanged # https://developers.deepgram.com/docs/flux/configure thresholds: dict[str, Any] = {} if is_given(eager_eot_threshold): thresholds["eager_eot_threshold"] = eager_eot_threshold if is_given(eot_threshold): thresholds["eot_threshold"] = eot_threshold if is_given(eot_timeout_ms): thresholds["eot_timeout_ms"] = eot_timeout_ms changed_options: dict[str, Any] = {} if thresholds: changed_options["thresholds"] = thresholds if is_given(keyterm): # keyterms replaces the whole list, so send the full effective set changed_options["keyterms"] = self._opts.keyterm if is_given(language_hint): changed_options["language_hints"] = self._opts.language_hint if changed_options: # chain off the previous send so deltas reach the server in order self._reconfigure_atask = asyncio.create_task( self._send_configure(changed_options, self._reconfigure_atask) )
class TTS (*,
model: TTSModels | str = 'aura-2-andromeda-en',
encoding: str = 'linear16',
sample_rate: int = 24000,
bit_rate: int | None = None,
api_key: str | None = None,
base_url: str = 'https://api.deepgram.com/v1/speak',
word_tokenizer: NotGivenOr[tokenize.WordTokenizer] = NOT_GIVEN,
http_session: aiohttp.ClientSession | None = None,
mip_opt_out: bool = False)-
Expand source code
class TTS(tts.TTS): def __init__( self, *, model: TTSModels | str = "aura-2-andromeda-en", encoding: str = "linear16", sample_rate: int = 24000, bit_rate: int | None = None, api_key: str | None = None, base_url: str = BASE_URL, word_tokenizer: NotGivenOr[tokenize.WordTokenizer] = NOT_GIVEN, http_session: aiohttp.ClientSession | None = None, mip_opt_out: bool = False, ) -> None: """ Create a new instance of Deepgram TTS. Args: model (TTSModels | str): TTS model to use. Defaults to "aura-2-andromeda-en". See https://developers.deepgram.com/docs/tts-models for available models. encoding (str): Audio encoding to use. Defaults to "linear16". sample_rate (int): Sample rate of audio. Defaults to 24000. bit_rate (int | None): Bit rate for compressed encodings (e.g. mp3). Defaults to None. See https://developers.deepgram.com/reference/text-to-speech-api#query-bit_rate api_key (str): Deepgram API key. If not provided, will look for DEEPGRAM_API_KEY in environment. base_url (str): Base URL for Deepgram TTS API. Defaults to "https://api.deepgram.com/v1/speak" word_tokenizer (tokenize.WordTokenizer): Tokenizer for processing text. Defaults to basic WordTokenizer. http_session (aiohttp.ClientSession): Optional aiohttp session to use for requests. """ # noqa: E501 super().__init__( capabilities=tts.TTSCapabilities(streaming=True), sample_rate=sample_rate, num_channels=NUM_CHANNELS, ) api_key = api_key or os.environ.get("DEEPGRAM_API_KEY") if not api_key: raise ValueError("Deepgram API key required. Set DEEPGRAM_API_KEY or provide api_key.") if not is_given(word_tokenizer): word_tokenizer = tokenize.basic.WordTokenizer(ignore_punctuation=False) self._opts = _TTSOptions( model=model, encoding=encoding, sample_rate=sample_rate, bit_rate=bit_rate, word_tokenizer=word_tokenizer, base_url=base_url, api_key=api_key, mip_opt_out=mip_opt_out, ) self._session = http_session self._streams = weakref.WeakSet[SynthesizeStream]() self._pool = utils.ConnectionPool[aiohttp.ClientWebSocketResponse]( connect_cb=self._connect_ws, close_cb=self._close_ws, max_session_duration=3600, # 1 hour mark_refreshed_on_get=False, ) @property def model(self) -> str: return self._opts.model @property def provider(self) -> str: return "Deepgram" async def _connect_ws(self, timeout: float) -> aiohttp.ClientWebSocketResponse: session = self._ensure_session() config: dict = { "encoding": self._opts.encoding, "model": self._opts.model, "sample_rate": self._opts.sample_rate, "mip_opt_out": self._opts.mip_opt_out, } if self._opts.bit_rate is not None: config["bit_rate"] = self._opts.bit_rate ws = await asyncio.wait_for( session.ws_connect( _to_deepgram_url(config, self._opts.base_url, websocket=True), headers={"Authorization": f"Token {self._opts.api_key}"}, ), timeout, ) ws_headers = { k: v for k, v in ws._response.headers.items() if k.startswith("dg-") or k == "Date" } logger.debug( "Established new Deepgram TTS WebSocket connection:", extra={"headers": ws_headers}, ) return ws async def _close_ws(self, ws: aiohttp.ClientWebSocketResponse) -> None: try: # Send Flush and Close messages to ensure Deepgram processes all remaining audio # and properly terminates the session, preventing lingering TTS sessions await ws.send_str(SynthesizeStream._FLUSH_MSG) await ws.send_str(SynthesizeStream._CLOSE_MSG) # Wait for server acknowledgment to prevent race conditions and ensure # proper cleanup, avoiding 429 Too Many Requests errors from lingering sessions try: await asyncio.wait_for(ws.receive(), timeout=1.0) except asyncio.TimeoutError: pass except Exception as e: logger.warning(f"Error during WebSocket close sequence: {e}") finally: await ws.close() def _ensure_session(self) -> aiohttp.ClientSession: if not self._session: self._session = utils.http_context.http_session() return self._session def update_options( self, *, model: NotGivenOr[TTSModels | str] = NOT_GIVEN, encoding: NotGivenOr[str] = NOT_GIVEN, sample_rate: NotGivenOr[int] = NOT_GIVEN, bit_rate: NotGivenOr[int | None] = NOT_GIVEN, ) -> None: """ Args: model (TTSModels | str): TTS model to use. encoding (str): Audio encoding to use. sample_rate (int): Sample rate of audio in Hz. bit_rate (int | None): Bit rate for compressed encodings (e.g. mp3). See https://developers.deepgram.com/reference/text-to-speech-api#query-bit_rate """ connection_params_changed = False if is_given(model): self._opts.model = model connection_params_changed = True if is_given(encoding): self._opts.encoding = encoding connection_params_changed = True if is_given(sample_rate): self._opts.sample_rate = sample_rate self._sample_rate = sample_rate # keep base class property in sync connection_params_changed = True if is_given(bit_rate): self._opts.bit_rate = bit_rate connection_params_changed = True if connection_params_changed: # These params are baked into the WebSocket URL at connection time, so any # existing pooled connection must be invalidated to avoid serving audio at # the wrong rate/encoding. self._pool.invalidate() 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 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()Helper class that provides a standard way to create an ABC using inheritance.
Create a new instance of Deepgram TTS.
Args
model:TTSModels | str- TTS model to use. Defaults to "aura-2-andromeda-en". See https://developers.deepgram.com/docs/tts-models for available models.
encoding:str- Audio encoding to use. Defaults to "linear16".
sample_rate:int- Sample rate of audio. Defaults to 24000.
bit_rate:int | None- Bit rate for compressed encodings (e.g. mp3). Defaults to None. See https://developers.deepgram.com/reference/text-to-speech-api#query-bit_rate
api_key:str- Deepgram API key. If not provided, will look for DEEPGRAM_API_KEY in environment.
base_url:str- Base URL for Deepgram TTS API. Defaults to "https://api.deepgram.com/v1/speak"
word_tokenizer:tokenize.WordTokenizer- Tokenizer for processing text. Defaults to basic WordTokenizer.
http_session:aiohttp.ClientSession- Optional aiohttp session to use for requests.
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 "Deepgram"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.deepgram.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.deepgram.tts.ChunkedStream-
Expand source code
def synthesize( self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS ) -> ChunkedStream: return ChunkedStream(tts=self, input_text=text, conn_options=conn_options) def update_options(self,
*,
model: NotGivenOr[TTSModels | str] = NOT_GIVEN,
encoding: NotGivenOr[str] = NOT_GIVEN,
sample_rate: NotGivenOr[int] = NOT_GIVEN,
bit_rate: NotGivenOr[int | None] = NOT_GIVEN) ‑> None-
Expand source code
def update_options( self, *, model: NotGivenOr[TTSModels | str] = NOT_GIVEN, encoding: NotGivenOr[str] = NOT_GIVEN, sample_rate: NotGivenOr[int] = NOT_GIVEN, bit_rate: NotGivenOr[int | None] = NOT_GIVEN, ) -> None: """ Args: model (TTSModels | str): TTS model to use. encoding (str): Audio encoding to use. sample_rate (int): Sample rate of audio in Hz. bit_rate (int | None): Bit rate for compressed encodings (e.g. mp3). See https://developers.deepgram.com/reference/text-to-speech-api#query-bit_rate """ connection_params_changed = False if is_given(model): self._opts.model = model connection_params_changed = True if is_given(encoding): self._opts.encoding = encoding connection_params_changed = True if is_given(sample_rate): self._opts.sample_rate = sample_rate self._sample_rate = sample_rate # keep base class property in sync connection_params_changed = True if is_given(bit_rate): self._opts.bit_rate = bit_rate connection_params_changed = True if connection_params_changed: # These params are baked into the WebSocket URL at connection time, so any # existing pooled connection must be invalidated to avoid serving audio at # the wrong rate/encoding. self._pool.invalidate()Args
model:TTSModels | str- TTS model to use.
encoding:str- Audio encoding to use.
sample_rate:int- Sample rate of audio in Hz.
bit_rate:int | None- Bit rate for compressed encodings (e.g. mp3). See https://developers.deepgram.com/reference/text-to-speech-api#query-bit_rate
Inherited members