wehub-resource-sync
992 lines
40 KiB
Python
992 lines
40 KiB
Python
# Copyright 2023 LiveKit, Inc.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import dataclasses
|
|
import json
|
|
import os
|
|
import time
|
|
import weakref
|
|
from collections import Counter
|
|
from collections.abc import Sequence
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
import aiohttp
|
|
|
|
from livekit import rtc
|
|
from livekit.agents import (
|
|
DEFAULT_API_CONNECT_OPTIONS,
|
|
APIConnectionError,
|
|
APIConnectOptions,
|
|
APIStatusError,
|
|
APITimeoutError,
|
|
LanguageCode,
|
|
stt,
|
|
utils,
|
|
)
|
|
from livekit.agents.types import (
|
|
NOT_GIVEN,
|
|
NotGivenOr,
|
|
)
|
|
from livekit.agents.utils import AudioBuffer, is_given
|
|
from livekit.agents.voice.io import TimedString
|
|
|
|
from ._utils import PeriodicCollector, _to_deepgram_url
|
|
from .log import logger
|
|
from .models import DeepgramLanguages, DeepgramModels
|
|
|
|
|
|
@dataclass
|
|
class STTOptions:
|
|
language: LanguageCode | None
|
|
detect_language: bool
|
|
interim_results: bool
|
|
punctuate: bool
|
|
model: DeepgramModels | str
|
|
smart_format: bool
|
|
no_delay: bool
|
|
endpointing_ms: int
|
|
enable_diarization: bool
|
|
filler_words: bool
|
|
sample_rate: int
|
|
num_channels: int
|
|
keywords: list[tuple[str, float]]
|
|
keyterm: str | Sequence[str]
|
|
profanity_filter: bool
|
|
redact: str | list[str]
|
|
endpoint_url: str
|
|
vad_events: bool = True
|
|
numerals: bool = False
|
|
mip_opt_out: bool = False
|
|
tags: NotGivenOr[list[str]] = NOT_GIVEN
|
|
utterance_end_ms: int | None = None
|
|
dictation: bool = False
|
|
replace: dict[str, str] | None = None
|
|
search: list[str] | None = None
|
|
|
|
|
|
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 config
|
|
|
|
|
|
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
|
|
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)
|
|
|
|
@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)
|
|
|
|
|
|
def live_transcription_to_speech_data(
|
|
language: str, data: dict, *, is_final: bool, start_time_offset: float
|
|
) -> list[stt.SpeechData]:
|
|
dg_alts = data["channel"]["alternatives"]
|
|
|
|
speech_data = []
|
|
for alt in dg_alts:
|
|
if is_final:
|
|
speakers = [word["speaker"] for word in alt["words"] if "speaker" in word]
|
|
speaker = Counter(speakers).most_common(1)[0][0] if speakers else None
|
|
else:
|
|
# interim result doesn't have correct speaker information?
|
|
speaker = None
|
|
|
|
sd = stt.SpeechData(
|
|
language=LanguageCode(language),
|
|
start_time=next((word.get("start", 0) for word in alt["words"]), 0) + start_time_offset,
|
|
end_time=next((word.get("end", 0) for word in alt["words"]), 0) + start_time_offset,
|
|
confidence=alt["confidence"],
|
|
text=alt["transcript"],
|
|
speaker_id=f"S{speaker}" if speaker is not None else None,
|
|
words=[
|
|
TimedString(
|
|
text=word.get("word", ""),
|
|
start_time=word.get("start", 0) + start_time_offset,
|
|
end_time=word.get("end", 0) + start_time_offset,
|
|
start_time_offset=start_time_offset,
|
|
)
|
|
for word in alt["words"]
|
|
]
|
|
if alt["words"]
|
|
else None,
|
|
)
|
|
if language == "multi" and "languages" in alt:
|
|
sd.language = LanguageCode(alt["languages"][0]) # TODO: handle multiple languages
|
|
speech_data.append(sd)
|
|
return speech_data
|
|
|
|
|
|
def prerecorded_transcription_to_speech_event(
|
|
language: str | None, # language should be None when 'detect_language' is enabled
|
|
data: dict,
|
|
) -> stt.SpeechEvent:
|
|
# We only support one channel for now
|
|
request_id = data["metadata"]["request_id"]
|
|
channel: dict = data["results"]["channels"][0]
|
|
dg_alts = channel["alternatives"]
|
|
|
|
# Use the detected language if enabled
|
|
# https://developers.deepgram.com/docs/language-detection
|
|
detected_language = LanguageCode(channel.get("detected_language", ""))
|
|
|
|
return stt.SpeechEvent(
|
|
request_id=request_id,
|
|
type=stt.SpeechEventType.FINAL_TRANSCRIPT,
|
|
alternatives=[
|
|
stt.SpeechData(
|
|
language=LanguageCode(language or detected_language),
|
|
start_time=alt["words"][0]["start"] if alt["words"] else 0,
|
|
end_time=alt["words"][-1]["end"] if alt["words"] else 0,
|
|
confidence=alt["confidence"],
|
|
text=alt["transcript"],
|
|
words=[
|
|
TimedString(
|
|
text=word.get("word", ""),
|
|
start_time=word.get("start", 0),
|
|
end_time=word.get("end", 0),
|
|
)
|
|
for word in alt["words"]
|
|
],
|
|
)
|
|
for alt in dg_alts
|
|
],
|
|
)
|
|
|
|
|
|
def _validate_model(
|
|
model: DeepgramModels | str, language: NotGivenOr[DeepgramLanguages | str]
|
|
) -> DeepgramModels | str:
|
|
en_only_models = {
|
|
"nova-2-meeting",
|
|
"nova-2-phonecall",
|
|
"nova-2-finance",
|
|
"nova-2-conversationalai",
|
|
"nova-2-voicemail",
|
|
"nova-2-video",
|
|
"nova-2-medical",
|
|
"nova-2-drivethru",
|
|
"nova-2-automotive",
|
|
}
|
|
if is_given(language) and language not in ("en-US", "en") and model in en_only_models:
|
|
logger.warning(
|
|
f"{model} does not support language {language}, falling back to nova-2-general"
|
|
)
|
|
return "nova-2-general"
|
|
return model
|
|
|
|
|
|
def _validate_tags(tags: list[str]) -> list[str]:
|
|
for tag in tags:
|
|
if len(tag) > 128:
|
|
raise ValueError("tag must be no more than 128 characters")
|
|
return tags
|
|
|
|
|
|
def _validate_keyterm(
|
|
model: DeepgramModels | str,
|
|
language: NotGivenOr[DeepgramLanguages | str],
|
|
keyterm: NotGivenOr[str | list[str]],
|
|
keywords: NotGivenOr[list[tuple[str, float]]],
|
|
) -> None:
|
|
"""
|
|
Validating keyterm and keywords for model compatibility.
|
|
See: https://developers.deepgram.com/docs/keyterm and https://developers.deepgram.com/docs/keywords
|
|
"""
|
|
if model.startswith("nova-3") and is_given(keywords):
|
|
raise ValueError(
|
|
"Keywords is only available for use with Nova-2, Nova-1, Enhanced, and "
|
|
"Base speech to text models. For Nova-3, use Keyterm Prompting."
|
|
)
|
|
|
|
if is_given(keyterm) and (not model.startswith("nova-3")):
|
|
raise ValueError(
|
|
"Keyterm Prompting is only available for transcription using the Nova-3 Model. "
|
|
"To boost recognition of keywords using another model, use the Keywords feature."
|
|
)
|