diff --git a/changelog/3225.changed.md b/changelog/3225.changed.md new file mode 100644 index 000000000..c5063f58e --- /dev/null +++ b/changelog/3225.changed.md @@ -0,0 +1,15 @@ +- Updated `SpeechmaticsSTTService` to use new Python Voice SDK with improved VAD, + Smart Turn capabilities, and brings dramatic improvements to latency without + any impact on accuracy. Use the `turn_detection_mode` parameter to control the + endpointing of speech, with `TurnDetectionMode.EXTERNAL` (default), + `TurnDetectionMode.ADAPTIVE`, or `TurnDetectionMode.SMART_TURN`. + ```python + stt = SpeechmaticsSTTService( + api_key=os.getenv("SPEECHMATICS_API_KEY"), + params=SpeechmaticsSTTService.InputParams( + language=Language.EN, + turn_detection_mode=SpeechmaticsSTTService.TurnDetectionMode.ADAPTIVE, + speaker_active_format="<{speaker_id}>{text}", + ), + ) + ``` diff --git a/changelog/3225.deprecated.md b/changelog/3225.deprecated.md new file mode 100644 index 000000000..162c82a82 --- /dev/null +++ b/changelog/3225.deprecated.md @@ -0,0 +1,4 @@ +- For `SpeechmaticsSTTService`, the `end_of_utterance_mode` parameter is deprecated. + Use the new `turn_detection_mode` parameter instead, with `TurnDetectionMode.EXTERNAL`, + `TurnDetectionMode.ADAPTIVE`, or `TurnDetectionMode.SMART_TURN`. The `enable_vad` + parameter is also deprecated and is inferred from the `turn_detection_mode`. diff --git a/examples/foundational/07a-interruptible-speechmatics-vad.py b/examples/foundational/07a-interruptible-speechmatics-vad.py index db3b95465..e4572160d 100644 --- a/examples/foundational/07a-interruptible-speechmatics-vad.py +++ b/examples/foundational/07a-interruptible-speechmatics-vad.py @@ -73,7 +73,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): 4. Text-to-Speech (TTS) - Low latency streaming audio synthesis - - Multiple voice options available including `sarah`, `theo`, and `megan` + - Multiple voice options available including `sarah`, `theo`, `megan` and `jack` 5. Configuration Options - `operating_point` parameter defaults to `ENHANCED` for optimal accuracy @@ -92,10 +92,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): api_key=os.getenv("SPEECHMATICS_API_KEY"), params=SpeechmaticsSTTService.InputParams( language=Language.EN, - enable_vad=True, - enable_diarization=True, - focus_speakers=["S1"], - end_of_utterance_silence_trigger=0.5, + turn_detection_mode=SpeechmaticsSTTService.TurnDetectionMode.ADAPTIVE, + # focus_speakers=["S1"], speaker_active_format="<{speaker_id}>{text}", speaker_passive_format="<{speaker_id}>{text}", ), diff --git a/examples/foundational/07a-interruptible-speechmatics.py b/examples/foundational/07a-interruptible-speechmatics.py index 981a69fdc..988d2894a 100644 --- a/examples/foundational/07a-interruptible-speechmatics.py +++ b/examples/foundational/07a-interruptible-speechmatics.py @@ -70,7 +70,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): TTS Features: - Low latency streaming audio synthesis - - Multiple voice options available including `sarah`, `theo`, and `megan` + - Multiple voice options available including `sarah`, `theo`, `megan` and `jack` For more information: - STT: https://docs.speechmatics.com/rt-api-ref @@ -83,8 +83,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): api_key=os.getenv("SPEECHMATICS_API_KEY"), params=SpeechmaticsSTTService.InputParams( language=Language.EN, - enable_diarization=True, - end_of_utterance_silence_trigger=0.5, speaker_active_format="<{speaker_id}>{text}", ), ) diff --git a/examples/foundational/13h-speechmatics-transcription.py b/examples/foundational/13h-speechmatics-transcription.py index 8b2984a84..30ee58059 100644 --- a/examples/foundational/13h-speechmatics-transcription.py +++ b/examples/foundational/13h-speechmatics-transcription.py @@ -68,7 +68,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): api_key=os.getenv("SPEECHMATICS_API_KEY"), params=SpeechmaticsSTTService.InputParams( language=Language.EN, - enable_diarization=True, speaker_active_format="<{speaker_id}>{text}", ), ) diff --git a/pyproject.toml b/pyproject.toml index f3c3ae7c5..b1c03adf9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -105,7 +105,7 @@ silero = [ "onnxruntime>=1.20.1,<2" ] simli = [ "simli-ai~=1.0.3"] soniox = [ "pipecat-ai[websockets-base]" ] soundfile = [ "soundfile~=0.13.1" ] -speechmatics = [ "speechmatics-rt>=0.5.0" ] +speechmatics = [ "speechmatics-voice[smart]>=0.2.4" ] strands = [ "strands-agents>=1.9.1,<2" ] tavus=[] together = [] diff --git a/src/pipecat/services/speechmatics/stt.py b/src/pipecat/services/speechmatics/stt.py index f18314d8d..3efa20d84 100644 --- a/src/pipecat/services/speechmatics/stt.py +++ b/src/pipecat/services/speechmatics/stt.py @@ -7,18 +7,19 @@ """Speechmatics STT service integration.""" import asyncio -import datetime import os -import re -from dataclasses import dataclass, field +import time from enum import Enum from typing import Any, AsyncGenerator -from urllib.parse import urlencode +from dotenv import load_dotenv from loguru import logger from pydantic import BaseModel +from pipecat import version as pipecat_version from pipecat.frames.frames import ( + BotStartedSpeakingFrame, + BotStoppedSpeakingFrame, CancelFrame, EndFrame, ErrorFrame, @@ -35,16 +36,19 @@ from pipecat.transcriptions.language import Language, resolve_language from pipecat.utils.tracing.service_decorators import traced_stt try: - from speechmatics.rt import ( - AsyncClient, + from speechmatics.voice import ( + AdditionalVocabEntry, + AgentClientMessageType, + AgentServerMessageType, AudioEncoding, - AudioFormat, - ClientMessageType, - ConversationConfig, + EndOfUtteranceMode, OperatingPoint, - ServerMessageType, - TranscriptionConfig, - __version__, + SpeakerFocusConfig, + SpeakerFocusMode, + SpeakerIdentifier, + VoiceAgentClient, + VoiceAgentConfig, + VoiceAgentConfigPreset, ) except ModuleNotFoundError as e: logger.error(f"Exception: {e}") @@ -54,147 +58,23 @@ except ModuleNotFoundError as e: raise Exception(f"Missing module: {e}") -class EndOfUtteranceMode(str, Enum): - """End of turn delay options for transcription.""" +load_dotenv() - NONE = "none" - FIXED = "fixed" + +class TurnDetectionMode(str, Enum): + """Endpoint and turn detection handling mode. + + How the STT engine handles the endpointing of speech. If using Pipecat's built-in endpointing, + then use `TurnDetectionMode.EXTERNAL` (default). + + To use the STT engine's built-in endpointing, then use `TurnDetectionMode.ADAPTIVE` for simple + voice activity detection or `TurnDetectionMode.SMART_TURN` for more advanced ML-based + endpointing. + """ + + EXTERNAL = "external" ADAPTIVE = "adaptive" - - -class DiarizationFocusMode(str, Enum): - """Speaker focus mode for diarization.""" - - RETAIN = "retain" - IGNORE = "ignore" - - -@dataclass -class AdditionalVocabEntry: - """Additional vocabulary entry. - - Parameters: - content: The word to add to the dictionary. - sounds_like: Similar words to the word. - """ - - content: str - sounds_like: list[str] = field(default_factory=list) - - -@dataclass -class DiarizationKnownSpeaker: - """Known speakers for speaker diarization. - - Parameters: - label: The label of the speaker. - speaker_identifiers: One or more data strings for the speaker. - """ - - label: str - speaker_identifiers: list[str] - - -@dataclass -class SpeechFragment: - """Fragment of an utterance. - - Parameters: - start_time: Start time of the fragment in seconds (from session start). - end_time: End time of the fragment in seconds (from session start). - language: Language of the fragment. Defaults to `Language.EN`. - is_eos: Whether the fragment is the end of a sentence. Defaults to `False`. - is_final: Whether the fragment is the final fragment. Defaults to `False`. - is_disfluency: Whether the fragment is a disfluency. Defaults to `False`. - is_punctuation: Whether the fragment is a punctuation. Defaults to `False`. - attaches_to: Whether the fragment attaches to the previous or next fragment (punctuation). Defaults to empty string. - content: Content of the fragment. Defaults to empty string. - speaker: Speaker of the fragment (if diarization is enabled). Defaults to `None`. - confidence: Confidence of the fragment (0.0 to 1.0). Defaults to `1.0`. - result: Raw result of the fragment from the TTS. - """ - - start_time: float - end_time: float - language: Language = Language.EN - is_eos: bool = False - is_final: bool = False - is_disfluency: bool = False - is_punctuation: bool = False - attaches_to: str = "" - content: str = "" - speaker: str | None = None - confidence: float = 1.0 - result: Any | None = None - - -@dataclass -class SpeakerFragments: - """SpeechFragment items grouped by speaker_id. - - Parameters: - speaker_id: The ID of the speaker. - is_active: Whether the speaker is active (emits frame). - timestamp: The timestamp of the frame. - language: The language of the frame. - fragments: The list of SpeechFragment items. - """ - - speaker_id: str | None = None - is_active: bool = False - timestamp: str | None = None - language: Language | None = None - fragments: list[SpeechFragment] = field(default_factory=list) - - def __str__(self): - """Return a string representation of the object.""" - return f"SpeakerFragments(speaker_id: {self.speaker_id}, timestamp: {self.timestamp}, language: {self.language}, text: {self._format_text()})" - - def _format_text(self, format: str | None = None) -> str: - """Wrap text with speaker ID in an optional f-string format. - - Args: - format: Format to wrap the text with. - - Returns: - str: The wrapped text. - """ - # Cumulative contents - content = "" - - # Assemble the text - for frag in self.fragments: - if content == "" or frag.attaches_to == "previous": - content += frag.content - else: - content += " " + frag.content - - # Format the text, if format is provided - if format is None or self.speaker_id is None: - return content - return format.format(**{"speaker_id": self.speaker_id, "text": content}) - - def _as_frame_attributes( - self, active_format: str | None = None, passive_format: str | None = None - ) -> dict[str, Any]: - """Return a dictionary of attributes for a TranscriptionFrame. - - Args: - active_format: Format to wrap the text with. - passive_format: Format to wrap the text with. Defaults to `active_format`. - - Returns: - dict[str, Any]: The dictionary of attributes. - """ - if not passive_format: - passive_format = active_format - return { - "text": self._format_text(active_format if self.is_active else passive_format), - "user_id": self.speaker_id or "", - "timestamp": self.timestamp, - "language": self.language, - "result": [frag.result for frag in self.fragments], - } + SMART_TURN = "smart_turn" class SpeechmaticsSTTService(STTService): @@ -205,85 +85,37 @@ class SpeechmaticsSTTService(STTService): and speaker diarization. """ + # Export related classes as class attributes + TurnDetectionMode = TurnDetectionMode + AudioEncoding = AudioEncoding + OperatingPoint = OperatingPoint + SpeakerFocusMode = SpeakerFocusMode + SpeakerFocusConfig = SpeakerFocusConfig + SpeakerIdentifier = SpeakerIdentifier + AdditionalVocabEntry = AdditionalVocabEntry + class InputParams(BaseModel): """Configuration parameters for Speechmatics STT service. Parameters: - operating_point: Operating point for transcription accuracy vs. latency tradeoff. It is - recommended to use OperatingPoint.ENHANCED for most use cases. Defaults to - OperatingPoint.ENHANCED. - domain: Domain for Speechmatics API. Defaults to None. language: Language code for transcription. Defaults to `Language.EN`. - output_locale: Output locale for transcription, e.g. `Language.EN_GB`. - Defaults to None. - - enable_vad: Enable VAD to trigger end of utterance detection. This should be used - without any other VAD enabled in the agent and will emit the speaker started - and stopped frames. Defaults to False. - - enable_partials: Enable partial transcriptions. When enabled, the STT engine will - emit partial word frames - useful for the visualisation of real-time transcription. - Defaults to True. - - max_delay: Maximum delay in seconds for transcription. This forces the STT engine to - speed up the processing of transcribed words and reduces the interval between partial - and final results. Lower values can have an impact on accuracy. Defaults to 1.0. - - end_of_utterance_silence_trigger: Maximum delay in seconds for end of utterance trigger. - The delay is used to wait for any further transcribed words before emitting the final - word frames. The value must be lower than max_delay. - Defaults to 0.5. - - end_of_utterance_mode: End of utterance delay mode. When ADAPTIVE is used, the delay - can be adjusted on the content of what the most recent speaker has said, such as - rate of speech and whether they have any pauses or disfluencies. When FIXED is used, - the delay is fixed to the value of `end_of_utterance_delay`. Use of NONE disables - end of utterance detection and uses a fallback timer. - Defaults to `EndOfUtteranceMode.FIXED`. - - additional_vocab: List of additional vocabulary entries. If you supply a list of - additional vocabulary entries, the this will increase the weight of the words in the - vocabulary and help the STT engine to better transcribe the words. - Defaults to []. - - punctuation_overrides: Punctuation overrides. This allows you to override the punctuation - in the STT engine. This is useful for languages that use different punctuation - than English. See documentation for more information. - Defaults to None. - - enable_diarization: Enable speaker diarization. When enabled, the STT engine will - determine and attribute words to unique speakers. The speaker_sensitivity - parameter can be used to adjust the sensitivity of diarization. - Defaults to False. - - speaker_sensitivity: Diarization sensitivity. A higher value increases the sensitivity - of diarization and helps when two or more speakers have similar voices. - Defaults to 0.5. - - max_speakers: Maximum number of speakers to detect. This forces the STT engine to cluster - words into a fixed number of speakers. It should not be used to limit the number of - speakers, unless it is clear that there will only be a known number of speakers. - Defaults to None. + turn_detection_mode: Endpoint handling, one of `TurnDetectionMode.EXTERNAL`, + `TurnDetectionMode.ADAPTIVE` and `TurnDetectionMode.SMART_TURN`. + Defaults to `TurnDetectionMode.EXTERNAL`. speaker_active_format: Formatter for active speaker ID. This formatter is used to format the text output for individual speakers and ensures that the context is clear for language models further down the pipeline. The attributes `text` and `speaker_id` are available. The system instructions for the language model may need to include any necessary instructions to handle the formatting. - Example: `@{speaker_id}: {text}`. - Defaults to transcription output. + Example: `@{speaker_id}: {text}`. Defaults to None. speaker_passive_format: Formatter for passive speaker ID. As with the speaker_active_format, the attributes `text` and `speaker_id` are available. - Example: `@{speaker_id} [background]: {text}`. - Defaults to transcription output. - - prefer_current_speaker: Prefer current speaker ID. When set to true, groups of words close - together are given extra weight to be identified as the same speaker. - Defaults to False. + Example: `@{speaker_id} [background]: {text}`. Defaults to None. focus_speakers: List of speaker IDs to focus on. When enabled, only these speakers are emitted as finalized frames and other speakers are considered passive. Words from @@ -299,11 +131,11 @@ class SpeechmaticsSTTService(STTService): `__ASSISTANT__`). Defaults to []. - focus_mode: Speaker focus mode for diarization. When set to `DiarizationFocusMode.RETAIN`, + focus_mode: Speaker focus mode for diarization. When set to `SpeakerFocusMode.RETAIN`, the STT engine will retain words spoken by other speakers (not listed in `ignore_speakers`) - and process them as passive speaker frames. When set to `DiarizationFocusMode.IGNORE`, + and process them as passive speaker frames. When set to `SpeakerFocusMode.IGNORE`, the STT engine will ignore words spoken by other speakers and they will not be processed. - Defaults to `DiarizationFocusMode.RETAIN`. + Defaults to `SpeakerFocusMode.RETAIN`. known_speakers: List of known speaker labels and identifiers. If you supply a list of labels and identifiers for speakers, then the STT engine will use them to attribute @@ -314,41 +146,100 @@ class SpeechmaticsSTTService(STTService): Refer to our examples on the format of the known_speakers parameter. Defaults to []. - chunk_size: Audio chunk size for streaming. Defaults to 160. + additional_vocab: List of additional vocabulary entries. If you supply a list of + additional vocabulary entries, the this will increase the weight of the words in the + vocabulary and help the STT engine to better transcribe the words. + Defaults to []. + audio_encoding: Audio encoding format. Defaults to AudioEncoding.PCM_S16LE. + + operating_point: Operating point for transcription accuracy vs. latency tradeoff. It is + recommended to use OperatingPoint.ENHANCED for most use cases. Default to enhanced. + + max_delay: Maximum delay in seconds for transcription. This forces the STT engine to + speed up the processing of transcribed words and reduces the interval between partial + and final results. Lower values can have an impact on accuracy. + + end_of_utterance_silence_trigger: Maximum delay in seconds for end of utterance trigger. + The delay is used to wait for any further transcribed words before emitting the final + word frames. The value must be lower than max_delay. + + end_of_utterance_max_delay: Maximum delay in seconds for end of utterance delay. + The delay is used to wait for any further transcribed words before emitting the final + word frames. The value must be greater than end_of_utterance_silence_trigger. + + punctuation_overrides: Punctuation overrides. This allows you to override the punctuation + in the STT engine. This is useful for languages that use different punctuation + than English. See documentation for more information. + + include_partials: Include partial segment fragments (words) in the output of + AddPartialSegment messages. Partial fragments from the STT will always be used for + speaker activity detection. This setting is used only for the formatted text output + of individual segments. + + enable_diarization: Enable speaker diarization. When enabled, the STT engine will + determine and attribute words to unique speakers. The speaker_sensitivity + parameter can be used to adjust the sensitivity of diarization. + + speaker_sensitivity: Diarization sensitivity. A higher value increases the sensitivity + of diarization and helps when two or more speakers have similar voices. + + max_speakers: Maximum number of speakers to detect. This forces the STT engine to cluster + words into a fixed number of speakers. It should not be used to limit the number of + speakers, unless it is clear that there will only be a known number of speakers. + + prefer_current_speaker: Prefer current speaker ID. When set to true, groups of words close + together are given extra weight to be identified as the same speaker. + + extra_params: Extra parameters to pass to the STT engine. This is a dictionary of + additional parameters that can be used to configure the STT engine. + Default to None. """ # Service configuration - operating_point: OperatingPoint = OperatingPoint.ENHANCED domain: str | None = None language: Language | str = Language.EN - output_locale: Language | str | None = None - # Features - enable_vad: bool = False - enable_partials: bool = True - max_delay: float = 1.0 - end_of_utterance_silence_trigger: float = 0.5 - end_of_utterance_mode: EndOfUtteranceMode = EndOfUtteranceMode.FIXED - additional_vocab: list[AdditionalVocabEntry] = [] - punctuation_overrides: dict | None = None + # Endpointing mode + turn_detection_mode: TurnDetectionMode = TurnDetectionMode.EXTERNAL - # Diarization - enable_diarization: bool = False - speaker_sensitivity: float = 0.5 - max_speakers: int | None = None - speaker_active_format: str = "{text}" - speaker_passive_format: str = "{text}" - prefer_current_speaker: bool = False + # Output formatting + speaker_active_format: str | None = None + speaker_passive_format: str | None = None + + # Speakers focus_speakers: list[str] = [] ignore_speakers: list[str] = [] - focus_mode: DiarizationFocusMode = DiarizationFocusMode.RETAIN - known_speakers: list[DiarizationKnownSpeaker] = [] + focus_mode: SpeakerFocusMode = SpeakerFocusMode.RETAIN + known_speakers: list[SpeakerIdentifier] = [] + + # Custom dictionary + additional_vocab: list[AdditionalVocabEntry] = [] # Audio - chunk_size: int = 160 audio_encoding: AudioEncoding = AudioEncoding.PCM_S16LE + # ------------------- + # Advanced features + # ------------------- + + # Features + operating_point: OperatingPoint | None = None + max_delay: float | None = None + end_of_utterance_silence_trigger: float | None = None + end_of_utterance_max_delay: float | None = None + punctuation_overrides: dict | None = None + include_partials: bool | None = None + + # Diarization + enable_diarization: bool | None = None + speaker_sensitivity: float | None = None + max_speakers: int | None = None + prefer_current_speaker: bool | None = None + + # Extra parameters + extra_params: dict | None = None + class UpdateParams(BaseModel): """Update parameters for Speechmatics STT service. @@ -370,23 +261,23 @@ class SpeechmaticsSTTService(STTService): `__ASSISTANT__`). Defaults to []. - focus_mode: Speaker focus mode for diarization. When set to `DiarizationFocusMode.RETAIN`, + focus_mode: Speaker focus mode for diarization. When set to `SpeakerFocusMode.RETAIN`, the STT engine will retain words spoken by other speakers (not listed in `ignore_speakers`) - and process them as passive speaker frames. When set to `DiarizationFocusMode.IGNORE`, + and process them as passive speaker frames. When set to `SpeakerFocusMode.IGNORE`, the STT engine will ignore words spoken by other speakers and they will not be processed. - Defaults to `DiarizationFocusMode.RETAIN`. + Defaults to `SpeakerFocusMode.RETAIN`. """ focus_speakers: list[str] = [] ignore_speakers: list[str] = [] - focus_mode: DiarizationFocusMode = DiarizationFocusMode.RETAIN + focus_mode: SpeakerFocusMode = SpeakerFocusMode.RETAIN def __init__( self, *, api_key: str | None = None, base_url: str | None = None, - sample_rate: int = 16000, + sample_rate: int | None = None, params: InputParams | None = None, **kwargs, ): @@ -397,7 +288,7 @@ class SpeechmaticsSTTService(STTService): `SPEECHMATICS_API_KEY` if not provided. base_url: Base URL for Speechmatics API. Uses environment variable `SPEECHMATICS_RT_URL` or defaults to `wss://eu2.rt.speechmatics.com/v2`. - sample_rate: Audio sample rate in Hz. Defaults to 16000. + sample_rate: Optional audio sample rate in Hz. params: Optional[InputParams]: Input parameters for the service. **kwargs: Additional arguments passed to STTService. """ @@ -415,60 +306,203 @@ class SpeechmaticsSTTService(STTService): if not self._base_url: raise ValueError("Missing Speechmatics base URL") - # Default parameters - self._params = params or SpeechmaticsSTTService.InputParams() + # Default params + params = params or SpeechmaticsSTTService.InputParams() # Deprecation check - _check_deprecated_args(kwargs, self._params) + self._check_deprecated_args(kwargs, params) - # Complete configuration objects - self._transcription_config: TranscriptionConfig = None - self._process_config() + # Voice agent + self._client: VoiceAgentClient | None = None + self._config: VoiceAgentConfig = self._prepare_config(params) - # STT client - self._client: AsyncClient | None = None + # Outbound frame queue + self._outbound_frames: asyncio.Queue[Frame] = asyncio.Queue() - # Current utterance speech data - self._speech_fragments: list[SpeechFragment] = [] + # Output formatting + if params.speaker_active_format is None: + params.speaker_active_format = ( + "@{speaker_id}: {text}" if params.enable_diarization else "{text}" + ) + + # Framework options + self._enable_vad: bool = self._config.end_of_utterance_mode != EndOfUtteranceMode.EXTERNAL + self._speaker_active_format: str = params.speaker_active_format + self._speaker_passive_format: str = ( + params.speaker_passive_format or params.speaker_active_format + ) + + # Metrics + self.set_model_name(self._config.operating_point.value) + + # Message queue + self._stt_msg_queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue() + self._stt_msg_task: asyncio.Task | None = None # Speaking states self._is_speaking: bool = False - - # Timing info - self._start_time: datetime.datetime | None = None - self._total_time: datetime.timedelta | None = None + self._bot_speaking: bool = False # Event handlers - if self._params.enable_diarization: + if params.enable_diarization: self._register_event_handler("on_speakers_result") - # EndOfUtterance fallback timer - self._end_of_utterance_timer: asyncio.Task | None = None + # ============================================================================ + # LIFE-CYCLE / SESSION MANAGEMENT + # ============================================================================ async def start(self, frame: StartFrame): """Called when the new session starts.""" await super().start(frame) await self._connect() + self._stt_msg_task = self.create_task(self._process_stt_messages()) async def stop(self, frame: EndFrame): """Called when the session ends.""" + await self.cancel_task(self._stt_msg_task) await super().stop(frame) await self._disconnect() async def cancel(self, frame: CancelFrame): """Called when the session is cancelled.""" + await self.cancel_task(self._stt_msg_task) await super().cancel(frame) await self._disconnect() - async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: - """Adds audio to the audio buffer and yields None.""" + async def _connect(self) -> None: + """Connect to the STT service. + + - Create STT client + - Register handlers for messages + - Connect to the client + """ + # Log the event + logger.debug(f"{self} connecting to Speechmatics STT service") + + # Update the audio sample rate + self._config.sample_rate = self.sample_rate + + # STT client + self._client: VoiceAgentClient = VoiceAgentClient( + api_key=self._api_key, + url=self._base_url, + app=f"pipecat/{pipecat_version()}", + config=self._config, + ) + + # Add message queue + def add_message(message: dict[str, Any]): + self._stt_msg_queue.put_nowait(message) + + # Add listeners + self._client.on(AgentServerMessageType.ADD_PARTIAL_SEGMENT, add_message) + self._client.on(AgentServerMessageType.ADD_SEGMENT, add_message) + + # Add listeners for VAD + if self._enable_vad: + self._client.on(AgentServerMessageType.START_OF_TURN, add_message) + self._client.on(AgentServerMessageType.END_OF_TURN, add_message) + + # Speaker result listener + if self._config.enable_diarization: + self._client.on(AgentServerMessageType.SPEAKERS_RESULT, add_message) + + # Other messages for debugging + self._client.on(AgentServerMessageType.ERROR, add_message) + self._client.on(AgentServerMessageType.WARNING, add_message) + self._client.on(AgentServerMessageType.INFO, add_message) + self._client.on(AgentServerMessageType.END_OF_TURN_PREDICTION, add_message) + self._client.on(AgentServerMessageType.END_OF_UTTERANCE, add_message) + + # Connect to the client + try: + await self._client.connect() + logger.debug(f"{self} connected") + except Exception as e: + self._client = None + await self.push_error(error_msg=f"Error connecting to STT service: {e}", exception=e) + + async def _disconnect(self) -> None: + """Disconnect from the STT service. + + - Disconnect the client + - Emit on_disconnected event handler for clients + """ + # Disconnect the client + logger.debug(f"{self} disconnecting from Speechmatics STT service") try: if self._client: - await self._client.send_audio(audio) - yield None + await self._client.disconnect() + except asyncio.TimeoutError: + logger.warning(f"{self} timeout while closing Speechmatics client connection") except Exception as e: - yield ErrorFrame(error=f"Unknown error occurred: {e}") - await self._disconnect() + await self.push_error(error_msg=f"Error closing Speechmatics client: {e}", exception=e) + finally: + self._client = None + await self._call_event_handler("on_disconnected") + + async def _process_stt_messages(self) -> None: + """Process messages from the STT client. + + Messages from the STT client are processed in a separate task to avoid blocking the main + thread. They are handled in strict order in which they are received. + """ + try: + while True: + message = await self._stt_msg_queue.get() + await self._handle_message(message) + except asyncio.CancelledError: + pass + + # ============================================================================ + # CONFIGURATION + # ============================================================================ + + def _prepare_config(self, params: InputParams) -> VoiceAgentConfig: + """Parse the InputParams into VoiceAgentConfig.""" + # Preset + config = VoiceAgentConfigPreset.load(params.turn_detection_mode.value) + + # Language + domain + config.language = self._language_to_speechmatics_language(params.language) + config.domain = params.domain + config.output_locale = self._locale_to_speechmatics_locale(config.language, params.language) + + # Speaker config + config.speaker_config = SpeakerFocusConfig( + focus_speakers=params.focus_speakers, + ignore_speakers=params.ignore_speakers, + focus_mode=params.focus_mode, + ) + config.known_speakers = params.known_speakers + + # Custom dictionary + config.additional_vocab = params.additional_vocab + + # Advanced parameters + for param in [ + "operating_point", + "max_delay", + "end_of_utterance_silence_trigger", + "end_of_utterance_max_delay", + "punctuation_overrides", + "include_partials", + "enable_diarization", + "speaker_sensitivity", + "max_speakers", + "prefer_current_speaker", + ]: + if getattr(params, param) is not None: + setattr(config, param, getattr(params, param)) + + # Extra parameters + if isinstance(params.extra_params, dict): + for key, value in params.extra_params.items(): + if hasattr(config, key): + setattr(config, key, value) + + # Return the complete config + return config def update_params( self, @@ -483,18 +517,219 @@ class SpeechmaticsSTTService(STTService): params: Update parameters for the service. """ # Check possible - if not self._params.enable_diarization: + if not self._config.enable_diarization: raise ValueError("Diarization is not enabled") - # Update the diarization configuration + # Update the existing diarization configuration if params.focus_speakers is not None: - self._params.focus_speakers = params.focus_speakers + self._config.speaker_config.focus_speakers = params.focus_speakers if params.ignore_speakers is not None: - self._params.ignore_speakers = params.ignore_speakers + self._config.speaker_config.ignore_speakers = params.ignore_speakers if params.focus_mode is not None: - self._params.focus_mode = params.focus_mode + self._config.speaker_config.focus_mode = params.focus_mode - async def send_message(self, message: ClientMessageType | str, **kwargs: Any) -> None: + # Send the update + if self._client: + self._client.update_diarization_config(self._config.speaker_config) + + # ============================================================================ + # HANDLE ENGINE MESSAGES + # ============================================================================ + + async def _handle_message(self, message: dict[str, Any]) -> None: + """Handle a message from the STT client.""" + event = message.get("message", "") + + # Handle events + match event: + case AgentServerMessageType.ADD_PARTIAL_SEGMENT: + await self._handle_partial_segment(message) + case AgentServerMessageType.ADD_SEGMENT: + await self._handle_segment(message) + case AgentServerMessageType.START_OF_TURN: + await self._handle_start_of_turn(message) + case AgentServerMessageType.END_OF_TURN: + await self._handle_end_of_turn(message) + case AgentServerMessageType.SPEAKERS_RESULT: + await self._handle_speakers_result(message) + case _: + logger.debug(f"{self} {event} -> {message}") + + async def _handle_partial_segment(self, message: dict[str, Any]) -> None: + """Handle AddPartialSegment events. + + AddPartialSegment events are triggered by Speechmatics STT when it detects a + partial segment of speech. These events provide the partial transcript for + the current speaking turn. + + Args: + message: the message payload. + """ + # Handle segments + segments: list[dict[str, Any]] = message.get("segments", []) + if segments: + await self._send_frames(segments) + + # Update metrics + await self._emit_metrics(message.get("metadata", {}).get("processing_time", 0.0)) + + async def _handle_segment(self, message: dict[str, Any]) -> None: + """Handle AddSegment events. + + AddSegment events are triggered by Speechmatics STT when it detects a + final segment of speech. These events provide the final transcript for + the current speaking turn. + + Args: + message: the message payload. + """ + # Handle segments + segments: list[dict[str, Any]] = message.get("segments", []) + if segments: + await self._send_frames(segments, finalized=True) + + async def _handle_start_of_turn(self, message: dict[str, Any]) -> None: + """Handle StartOfTurn events. + + When Speechmatics STT detects the start of a new speaking turn, a StartOfTurn + event is triggered. This triggers bot interruption to stop any ongoing speech + synthesis and signals the start of user speech detection. + + The service will: + - Send a BotInterruptionFrame upstream to stop bot speech + - Send a UserStartedSpeakingFrame downstream to notify other components + - Start metrics collection for measuring response times + + Args: + message: the message payload. + """ + logger.debug(f"{self} StartOfTurn received") + await self.push_interruption_task_frame_and_wait() + await self.broadcast_frame(UserStartedSpeakingFrame) + # await self.start_processing_metrics() + + async def _handle_end_of_turn(self, message: dict[str, Any]) -> None: + """Handle EndOfTurn events. + + EndOfTurn events are triggered by Speechmatics STT when it concludes a + speaking turn. This occurs either due to silence or reaching the + end-of-turn confidence thresholds. These events provide the final + transcript for the completed turn. + + The service will: + - Stop processing metrics collection + - Send a UserStoppedSpeakingFrame to signal turn completion + + Args: + message: the message payload. + """ + logger.debug(f"{self} EndOfTurn received") + # await self.stop_processing_metrics() + await self.broadcast_frame(UserStoppedSpeakingFrame) + + async def _handle_speakers_result(self, message: dict[str, Any]) -> None: + """Handle SpeakersResult events. + + SpeakersResult events are triggered by Speechmatics STT when it provides + speaker information for the current speaking turn. + + Args: + message: the message payload. + """ + logger.debug(f"{self} speakers result received from STT") + await self._call_event_handler("on_speakers_result", message) + + # ============================================================================ + # SEND FRAMES TO PIPELINE + # ============================================================================ + + async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process frames for VAD and metrics handling. + + Args: + frame: Frame to process. + direction: Direction of frame processing. + """ + # Forward to parent + await super().process_frame(frame, direction) + + # Track the bot + if isinstance(frame, BotStartedSpeakingFrame): + self._bot_speaking = True + elif isinstance(frame, BotStoppedSpeakingFrame): + self._bot_speaking = False + + # Force finalization + if isinstance(frame, UserStoppedSpeakingFrame): + if self._enable_vad: + logger.warning( + f"{self} UserStoppedSpeakingFrame received but internal VAD is being used" + ) + elif not self._enable_vad and self._client is not None: + self._client.finalize() + + async def _send_frames(self, segments: list[dict[str, Any]], finalized: bool = False) -> None: + """Send frames to the pipeline. + + Args: + segments: The segments to send. + finalized: Whether the data is final or partial. + """ + # Skip if no frames + if not segments: + return + + # Frames to send + frames: list[Frame] = [] + + # Create frame from segment + def attr_from_segment(segment: dict[str, Any]) -> dict[str, Any]: + # Formats the output text based on the speaker and defined formats from the config. + text = ( + self._speaker_active_format + if segment.get("is_active", True) + else self._speaker_passive_format + ).format( + **{ + "speaker_id": segment.get("speaker_id", "UU"), + "text": segment.get("text", ""), + "ts": segment.get("timestamp"), + "lang": segment.get("language"), + } + ) + + # Return the attributes for the frame + return { + "text": text, + "user_id": segment.get("speaker_id") or "", + "timestamp": segment.get("timestamp"), + "language": segment.get("language"), + "result": segment.get("results", []), + } + + # If final, then re-parse into TranscriptionFrame + if finalized: + frames += [TranscriptionFrame(**attr_from_segment(segment)) for segment in segments] + finalized_text = "|".join([s["text"] for s in segments]) + await self._handle_transcription(finalized_text, True, segments[0]["language"]) + logger.debug(f"{self} finalized transcript: {[f.text for f in frames]}") + + # Return as interim results (unformatted) + else: + frames += [ + InterimTranscriptionFrame(**attr_from_segment(segment)) for segment in segments + ] + logger.debug(f"{self} interim transcript: {[f.text for f in frames]}") + + # Send the frames + for frame in frames: + await self.push_frame(frame) + + # ============================================================================ + # PUBLIC FUNCTIONS + # ============================================================================ + + async def send_message(self, message: AgentClientMessageType | str, **kwargs: Any) -> None: """Send a message to the STT service. This sends a message to the STT service via the underlying transport. If the session @@ -508,635 +743,221 @@ class SpeechmaticsSTTService(STTService): try: payload = {"message": message} payload.update(kwargs) - logger.debug(f"Sending message to STT: {payload}") - asyncio.run_coroutine_threadsafe( - self._client.send_message(payload), self.get_event_loop() - ) + logger.debug(f"{self} sending message to STT: {payload}") + self.create_task(self._client.send_message(payload)) except Exception as e: - await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) - raise RuntimeError(f"error sending message to STT: {e}") + raise RuntimeError(f"{self} error sending message to STT: {e}") - async def _connect(self) -> None: - """Connect to the STT service.""" - # Create new STT RT client - self._client = AsyncClient( - api_key=self._api_key, - url=_get_endpoint_url(self._base_url), - ) + # ============================================================================ + # METRICS + # ============================================================================ - # Log the event - logger.debug(f"{self} Connecting to Speechmatics STT service") - - # Recognition started event - @self._client.on(ServerMessageType.RECOGNITION_STARTED) - def _evt_on_recognition_started(message: dict[str, Any]): - logger.debug(f"Recognition started (session: {message.get('id')})") - self._start_time = datetime.datetime.now(datetime.timezone.utc) - - # Partial transcript event - if self._params.enable_partials: - - @self._client.on(ServerMessageType.ADD_PARTIAL_TRANSCRIPT) - def _evt_on_partial_transcript(message: dict[str, Any]): - self._handle_transcript(message, is_final=False) - - # Final transcript event - @self._client.on(ServerMessageType.ADD_TRANSCRIPT) - def _evt_on_final_transcript(message: dict[str, Any]): - self._handle_transcript(message, is_final=True) - - # End of Utterance - if self._params.end_of_utterance_mode == EndOfUtteranceMode.FIXED: - - @self._client.on(ServerMessageType.END_OF_UTTERANCE) - def _evt_on_end_of_utterance(message: dict[str, Any]): - logger.debug("End of utterance received from STT") - asyncio.run_coroutine_threadsafe( - self._handle_end_of_utterance(), self.get_event_loop() - ) - - # Speaker Result - if self._params.enable_diarization: - - @self._client.on(ServerMessageType.SPEAKERS_RESULT) - def _evt_on_speakers_result(message: dict[str, Any]): - logger.debug("Speakers result received from STT") - asyncio.run_coroutine_threadsafe( - self._call_event_handler("on_speakers_result", message), - self.get_event_loop(), - ) - - # Start session - try: - await self._client.start_session( - transcription_config=self._transcription_config, - audio_format=AudioFormat( - encoding=self._params.audio_encoding, - sample_rate=self.sample_rate, - chunk_size=self._params.chunk_size, - ), - ) - logger.debug(f"{self} Connected to Speechmatics STT service") - await self._call_event_handler("on_connected") - except Exception as e: - await self.push_error(error_msg=f"Error connecting to Speechmatics: {e}", exception=e) - self._client = None - - async def _disconnect(self) -> None: - """Disconnect from the STT service.""" - # Disconnect the client - logger.debug(f"{self} Disconnecting from Speechmatics STT service") - try: - if self._client: - await asyncio.wait_for(self._client.close(), timeout=5.0) - logger.debug(f"{self} Disconnected from Speechmatics STT service") - except asyncio.TimeoutError: - logger.warning(f"{self} Timeout while closing Speechmatics client connection") - except Exception as e: - await self.push_error( - error_msg=f"Error disconnecting from Speechmatics: {e}", exception=e - ) - finally: - self._client = None - await self._call_event_handler("on_disconnected") - - def _process_config(self) -> None: - """Create a formatted STT transcription config. - - Creates a transcription config object based on the service parameters. Aligns - with the Speechmatics RT API transcription config. - """ - # Convert language if it's a Language enum - language = self._params.language - if isinstance(language, Language): - language = _language_to_speechmatics_language(language) - - # Convert output locale if it's a Language enum - output_locale = self._params.output_locale - if isinstance(output_locale, Language): - output_locale = _locale_to_speechmatics_locale(language, output_locale) - - # Transcription config - transcription_config = TranscriptionConfig( - language=language, - domain=self._params.domain, - output_locale=output_locale, - operating_point=self._params.operating_point, - diarization="speaker" if self._params.enable_diarization else None, - enable_partials=self._params.enable_partials, - max_delay=self._params.max_delay, - ) - - # Additional vocab - if self._params.additional_vocab: - transcription_config.additional_vocab = [ - { - "content": e.content, - **({"sounds_like": e.sounds_like} if e.sounds_like else {}), - } - for e in self._params.additional_vocab - ] - - # Diarization - if self._params.enable_diarization: - dz_cfg = {} - if self._params.speaker_sensitivity is not None: - dz_cfg["speaker_sensitivity"] = self._params.speaker_sensitivity - if self._params.prefer_current_speaker is not None: - dz_cfg["prefer_current_speaker"] = self._params.prefer_current_speaker - if self._params.known_speakers: - dz_cfg["speakers"] = { - s.label: s.speaker_identifiers for s in self._params.known_speakers - } - if self._params.max_speakers is not None: - dz_cfg["max_speakers"] = self._params.max_speakers - if dz_cfg: - transcription_config.speaker_diarization_config = dz_cfg - - # End of Utterance (for fixed) - if ( - self._params.end_of_utterance_silence_trigger - and self._params.end_of_utterance_mode == EndOfUtteranceMode.FIXED - ): - transcription_config.conversation_config = ConversationConfig( - end_of_utterance_silence_trigger=self._params.end_of_utterance_silence_trigger, - ) - - # Punctuation overrides - if self._params.punctuation_overrides: - transcription_config.punctuation_overrides = self._params.punctuation_overrides - - # Set config - self._transcription_config = transcription_config - - def _handle_transcript(self, message: dict[str, Any], is_final: bool) -> None: - """Handle the partial and final transcript events. - - Args: - message: The new Partial or Final from the STT engine. - is_final: Whether the data is final or partial. - """ - # Add the speech fragments - has_changed = self._add_speech_fragments( - message=message, - is_final=is_final, - ) - - # Skip if unchanged - if not has_changed: - return - - # Set a timer for the end of utterance - self._end_of_utterance_timer_start() - - # Send frames - asyncio.run_coroutine_threadsafe(self._send_frames(), self.get_event_loop()) - - @traced_stt - async def _handle_transcription( - self, transcript: str, is_final: bool, language: Language | None = None - ): - """Handle a transcription result with tracing.""" - pass - - def _end_of_utterance_timer_start(self): - """Start the timer for the end of utterance. - - This will use the STT's `end_of_utterance_silence_trigger` value and set - a timer to send the latest transcript to the pipeline. It is used as a - fallback from the EnfOfUtterance messages from the STT. - - Note that the `end_of_utterance_silence_trigger` will be from when the - last updated speech was received and this will likely be longer in - real world time to that inside of the STT engine. - """ - # Reset the end of utterance timer - if self._end_of_utterance_timer is not None: - self._end_of_utterance_timer.cancel() - - # Send after a delay - async def send_after_delay(delay: float): - await asyncio.sleep(delay) - logger.debug("Fallback EndOfUtterance triggered.") - asyncio.run_coroutine_threadsafe(self._handle_end_of_utterance(), self.get_event_loop()) - - # Start the timer - self._end_of_utterance_timer = asyncio.create_task( - send_after_delay(self._params.end_of_utterance_silence_trigger * 2) - ) - - async def _handle_end_of_utterance(self): - """Handle the end of utterance event. - - This will check for any running timers for end of utterance, reset them, - and then send a finalized frame to the pipeline. - """ - # Send the frames - await self._send_frames(finalized=True) - - # Reset the end of utterance timer - if self._end_of_utterance_timer: - self._end_of_utterance_timer.cancel() - self._end_of_utterance_timer = None - - async def _send_frames(self, finalized: bool = False) -> None: - """Send frames to the pipeline. - - Send speech frames to the pipeline. If VAD is enabled, then this will - also send an interruption and user started speaking frames. When the - final transcript is received, then this will send a user stopped speaking - and stop interruption frames. - - Args: - finalized: Whether the data is final or partial. - """ - # Get speech frames (InterimTranscriptionFrame) - speech_frames = self._get_frames_from_fragments() - - # Skip if no frames - if not speech_frames: - return - - # Check at least one frame is active - if not any(frame.is_active for frame in speech_frames): - return - - # Frames to send - downstream_frames: list[Frame] = [] - - # If VAD is enabled, then send a speaking frame - if self._params.enable_vad and not self._is_speaking: - logger.debug("User started speaking") - self._is_speaking = True - await self.push_interruption_task_frame_and_wait() - downstream_frames += [UserStartedSpeakingFrame()] - - # If final, then re-parse into TranscriptionFrame - if finalized: - # Reset the speech fragments - self._speech_fragments.clear() - - # Transform frames - downstream_frames += [ - TranscriptionFrame( - **frame._as_frame_attributes( - self._params.speaker_active_format, self._params.speaker_passive_format - ) - ) - for frame in speech_frames - ] - - # Log transcript(s) - logger.debug(f"Finalized transcript: {[f.text for f in downstream_frames]}") - - # Return as interim results (unformatted) - else: - downstream_frames += [ - InterimTranscriptionFrame( - **frame._as_frame_attributes( - self._params.speaker_active_format, self._params.speaker_passive_format - ) - ) - for frame in speech_frames - ] - - # If VAD is enabled, then send a speaking frame - if self._params.enable_vad and self._is_speaking and finalized: - logger.debug("User stopped speaking") - self._is_speaking = False - downstream_frames += [UserStoppedSpeakingFrame()] - - # Send the DOWNSTREAM frames - for frame in downstream_frames: - await self.push_frame(frame, FrameDirection.DOWNSTREAM) - - def _add_speech_fragments(self, message: dict[str, Any], is_final: bool = False) -> bool: - """Takes a new Partial or Final from the STT engine. - - Accumulates it into the _speech_data list. As new final data is added, all - partials are removed from the list. - - Note: If a known speaker is `__[A-Z0-9_]{2,}__`, then the words are skipped, - as this is used to protect against self-interruption by the assistant or to - block out specific known voices. - - Args: - message: The new Partial or Final from the STT engine. - is_final: Whether the data is final or partial. + def can_generate_metrics(self) -> bool: + """Check if this service can generate processing metrics. Returns: - bool: True if the speech data was updated, False otherwise. + True, as Speechmatics STT supports generation of metrics. """ - # Parsed new speech data from the STT engine - fragments: list[SpeechFragment] = [] - - # Current length of the speech data - current_length = len(self._speech_fragments) - - # Iterate over the results in the payload - for result in message.get("results", []): - alt = result.get("alternatives", [{}])[0] - if alt.get("content", None): - # Create the new fragment - fragment = SpeechFragment( - start_time=result.get("start_time", 0), - end_time=result.get("end_time", 0), - language=alt.get("language", Language.EN), - is_eos=alt.get("is_eos", False), - is_final=is_final, - attaches_to=result.get("attaches_to", ""), - content=alt.get("content", ""), - speaker=alt.get("speaker", None), - confidence=alt.get("confidence", 1.0), - result=result, - ) - - # Speaker filtering - if fragment.speaker: - # Drop `__XX__` speakers - if re.match(r"^__[A-Z0-9_]{2,}__$", fragment.speaker): - continue - - # Drop speakers not focussed on - if ( - self._params.focus_mode == DiarizationFocusMode.IGNORE - and self._params.focus_speakers - and fragment.speaker not in self._params.focus_speakers - ): - continue - - # Drop ignored speakers - if ( - self._params.ignore_speakers - and fragment.speaker in self._params.ignore_speakers - ): - continue - - # Add the fragment - fragments.append(fragment) - - # Remove existing partials, as new partials and finals are provided - self._speech_fragments = [frag for frag in self._speech_fragments if frag.is_final] - - # Return if no new fragments and length of the existing data is unchanged - if not fragments and len(self._speech_fragments) == current_length: - return False - - # Add the fragments to the speech data - self._speech_fragments.extend(fragments) - - # Data was updated return True - def _get_frames_from_fragments(self) -> list[SpeakerFragments]: - """Get speech data objects for the current fragment list. + @traced_stt + async def _handle_transcription(self, transcript: str, is_final: bool, language: Language): + """Record transcription event for tracing.""" + pass - Each speech fragments is grouped by contiguous speaker and then - returned as internal SpeakerFragments objects with the `speaker_id` field - set to the current speaker (string). An utterance may contain speech from - more than one speaker (e.g. S1, S2, S1, S3, ...), so they are kept - in strict order for the context of the conversation. + async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: + """Adds audio to the audio buffer and yields None.""" + try: + if self._client: + await self._client.send_audio(audio) + yield None + except Exception as e: + yield ErrorFrame(f"Speechmatics error: {e}") + await self._disconnect() - Returns: - List[SpeakerFragments]: The list of objects. + async def _emit_metrics(self, processing_time: float) -> None: + """Create TTFB metrics. + + The TTFB is the seconds between the person speaking and the STT + engine emitting the first partial. This is only calculated at the + start of an utterance. """ - # Speaker groups - current_speaker: str | None = None - speaker_groups: list[list[SpeechFragment]] = [[]] + # Skip if metrics not available + if not self._metrics or processing_time == 0.0: + return - # Group by speakers - for frag in self._speech_fragments: - if frag.speaker != current_speaker: - current_speaker = frag.speaker - if speaker_groups[-1]: - speaker_groups.append([]) - speaker_groups[-1].append(frag) + # Calculate time as time.time() - ttfb (which is seconds) + start_time = time.time() - processing_time - # Create SpeakerFragments objects - speaker_fragments: list[SpeakerFragments] = [] - for group in speaker_groups: - sd = self._get_speaker_fragments_from_fragment_group(group) - if sd: - speaker_fragments.append(sd) + # Update internal metrics + self._metrics._start_ttfb_time = start_time + self._metrics._start_processing_time = start_time - # Return the grouped SpeakerFragments objects - return speaker_fragments + # Stop TTFB metrics + await self.stop_ttfb_metrics() + await self.stop_processing_metrics() - def _get_speaker_fragments_from_fragment_group( - self, - group: list[SpeechFragment], - ) -> SpeakerFragments | None: - """Take a group of fragments and piece together into SpeakerFragments. + # ============================================================================ + # HELPERS + # ============================================================================ - Each fragment for a given speaker is assembled into a string, - taking into consideration whether words are attached to the - previous or next word (notably punctuation). This ensures that - the text does not have extra spaces. This will also check for - any straggling punctuation from earlier utterances that should - be removed. + def _language_to_speechmatics_language(self, language: Language) -> str: + """Convert a Language enum to a Speechmatics language code. Args: - group: List of SpeechFragment objects. + language: The Language enum to convert. Returns: - SpeakerFragments: The object for the group. + str: The Speechmatics language code, if found. """ - # Check for starting fragments that are attached to previous - if group and group[0].attaches_to == "previous": - group = group[1:] + # List of supported input languages + BASE_LANGUAGES = { + Language.AR: "ar", + Language.BA: "ba", + Language.EU: "eu", + Language.BE: "be", + Language.BG: "bg", + Language.BN: "bn", + Language.YUE: "yue", + Language.CA: "ca", + Language.HR: "hr", + Language.CS: "cs", + Language.DA: "da", + Language.NL: "nl", + Language.EN: "en", + Language.EO: "eo", + Language.ET: "et", + Language.FA: "fa", + Language.FI: "fi", + Language.FR: "fr", + Language.GL: "gl", + Language.DE: "de", + Language.EL: "el", + Language.HE: "he", + Language.HI: "hi", + Language.HU: "hu", + Language.IT: "it", + Language.ID: "id", + Language.GA: "ga", + Language.JA: "ja", + Language.KO: "ko", + Language.LV: "lv", + Language.LT: "lt", + Language.MS: "ms", + Language.MT: "mt", + Language.CMN: "cmn", + Language.MR: "mr", + Language.MN: "mn", + Language.NO: "no", + Language.PL: "pl", + Language.PT: "pt", + Language.RO: "ro", + Language.RU: "ru", + Language.SK: "sk", + Language.SL: "sl", + Language.ES: "es", + Language.SV: "sv", + Language.SW: "sw", + Language.TA: "ta", + Language.TH: "th", + Language.TR: "tr", + Language.UG: "ug", + Language.UK: "uk", + Language.UR: "ur", + Language.VI: "vi", + Language.CY: "cy", + } - # Check for trailing fragments that are attached to next - if group and group[-1].attaches_to == "next": - group = group[:-1] + # Get the language code + result = resolve_language(language, BASE_LANGUAGES, use_base_code=True) - # Check there are results - if not group: + # Fail if language is not supported + if not result: + raise ValueError(f"Unsupported language: {language}") + + # Return the language code + return result + + def _locale_to_speechmatics_locale(self, base_code: str, locale: Language) -> str | None: + """Convert a Language enum to a Speechmatics language / locale code. + + Args: + base_code: The language code. + locale: The Language enum to convert. + + Returns: + str: The Speechmatics language code, if found. + """ + # Languages and output locales + LOCALES = { + "en": { + Language.EN_GB: "en-GB", + Language.EN_US: "en-US", + Language.EN_AU: "en-AU", + }, + } + + # Ensure language code is in the map + if "-" not in str(locale) or base_code not in LOCALES: return None - # Get the timing extremes - start_time = min(frag.start_time for frag in group) + # Get the locale code + result = LOCALES.get(base_code).get(locale, None) - # Timestamp - ts = (self._start_time + datetime.timedelta(seconds=start_time)).isoformat( - timespec="milliseconds" - ) + # Fail if locale is not supported + if not result: + logger.warning(f"{self} Unsupported output locale: {locale}, defaulting to {base_code}") - # Determine if the speaker is considered active - is_active = True - if self._params.enable_diarization and self._params.focus_speakers: - is_active = group[0].speaker in self._params.focus_speakers + # Return the locale code + return result - # Return the SpeakerFragments object - return SpeakerFragments( - speaker_id=group[0].speaker, - timestamp=ts, - language=group[0].language, - fragments=group, - is_active=is_active, - ) + def _check_deprecated_args(self, kwargs: dict, params: InputParams) -> None: + """Check arguments for deprecation and update params if necessary. + This function will show deprecation warnings for deprecated arguments and + migrate them to the new location in the params object. If the new location + is None, the argument is not used. -def _get_endpoint_url(url: str) -> str: - """Format the endpoint URL with the SDK and app versions. + Args: + kwargs: Keyword arguments passed to the constructor. + params: Input parameters for the service. + """ - Args: - url: The base URL for the endpoint. + # Show deprecation warnings + def _deprecation_warning(old: str, new: str | None = None) -> None: + import warnings - Returns: - str: The formatted endpoint URL. - """ - query_params = dict() - query_params["sm-app"] = f"pipecat/{__version__}" - query = urlencode(query_params) + with warnings.catch_warnings(): + warnings.simplefilter("always") + if new: + message = f"`{old}` is deprecated, use `InputParams.{new}`" + else: + message = f"`{old}` is deprecated and not used" + warnings.warn(message, DeprecationWarning) - return f"{url}?{query}" + # List of deprecated arguments and their new location + deprecated_args = [ + ("language", "language"), + ("language_code", "language"), + ("domain", "domain"), + ("output_locale", None), + ("output_locale_code", None), + ("enable_partials", None), + ("max_delay", "max_delay"), + ("chunk_size", None), + ("audio_encoding", "audio_encoding"), + ("end_of_utterance_silence_trigger", "end_of_utterance_silence_trigger"), + {"enable_speaker_diarization", "enable_diarization"}, + ("text_format", "speaker_active_format"), + ("max_speakers", "max_speakers"), + ("transcription_config", None), + ("enable_vad", None), + ("end_of_utterance_mode", None), + ] - -def _language_to_speechmatics_language(language: Language) -> str: - """Convert a Language enum to a Speechmatics language code. - - Args: - language: The Language enum to convert. - - Returns: - str: The Speechmatics language code. - """ - # List of supported input languages - LANGUAGE_MAP = { - Language.AR: "ar", - Language.BA: "ba", - Language.EU: "eu", - Language.BE: "be", - Language.BG: "bg", - Language.BN: "bn", - Language.YUE: "yue", - Language.CA: "ca", - Language.HR: "hr", - Language.CS: "cs", - Language.DA: "da", - Language.NL: "nl", - Language.EN: "en", - Language.EO: "eo", - Language.ET: "et", - Language.FA: "fa", - Language.FI: "fi", - Language.FR: "fr", - Language.GL: "gl", - Language.DE: "de", - Language.EL: "el", - Language.HE: "he", - Language.HI: "hi", - Language.HU: "hu", - Language.IT: "it", - Language.ID: "id", - Language.GA: "ga", - Language.JA: "ja", - Language.KO: "ko", - Language.LV: "lv", - Language.LT: "lt", - Language.MS: "ms", - Language.MT: "mt", - Language.CMN: "cmn", - Language.MR: "mr", - Language.MN: "mn", - Language.NO: "no", - Language.PL: "pl", - Language.PT: "pt", - Language.RO: "ro", - Language.RU: "ru", - Language.SK: "sk", - Language.SL: "sl", - Language.ES: "es", - Language.SV: "sv", - Language.SW: "sw", - Language.TA: "ta", - Language.TH: "th", - Language.TR: "tr", - Language.UG: "ug", - Language.UK: "uk", - Language.UR: "ur", - Language.VI: "vi", - Language.CY: "cy", - } - - return resolve_language(language, LANGUAGE_MAP, use_base_code=True) - - -def _locale_to_speechmatics_locale(language_code: str, locale: Language) -> str | None: - """Convert a Language enum to a Speechmatics language code. - - Args: - language_code: The language code. - locale: The Language enum to convert. - - Returns: - str: The Speechmatics language code, if found. - """ - # Languages and output locales - LOCALES = { - "en": { - Language.EN_GB: "en-GB", - Language.EN_US: "en-US", - Language.EN_AU: "en-AU", - }, - } - - # Get the locale code - result = LOCALES.get(language_code, {}).get(locale) - - # Fail if locale is not supported - if not result: - logger.warning(f"Unsupported output locale: {locale}, defaulting to {language_code}") - - # Return the locale code - return result - - -def _check_deprecated_args(kwargs: dict, params: SpeechmaticsSTTService.InputParams) -> None: - """Check arguments for deprecation and update params if necessary. - - This function will show deprecation warnings for deprecated arguments and - migrate them to the new location in the params object. If the new location - is None, the argument is not used. - - Args: - kwargs: Keyword arguments passed to the constructor. - params: Input parameters for the service. - """ - - # Show deprecation warnings - def _deprecation_warning(old: str, new: str | None = None): - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - if new: - message = f"`{old}` is deprecated, use `InputParams.{new}`" - else: - message = f"`{old}` is deprecated and not used" - warnings.warn(message, DeprecationWarning) - - # List of deprecated arguments and their new location - deprecated_args = [ - ("language", "language"), - ("language_code", "language"), - ("domain", "domain"), - ("output_locale", "output_locale"), - ("output_locale_code", "output_locale"), - ("enable_partials", "enable_partials"), - ("max_delay", "max_delay"), - ("chunk_size", "chunk_size"), - ("audio_encoding", "audio_encoding"), - ("end_of_utterance_silence_trigger", "end_of_utterance_silence_trigger"), - {"enable_speaker_diarization", "enable_diarization"}, - ("text_format", "speaker_active_format"), - ("max_speakers", "max_speakers"), - ("transcription_config", None), - ] - - # Show warnings + migrate the arguments - for old, new in deprecated_args: - if old in kwargs: - _deprecation_warning(old, new) - if kwargs.get(old, None) is not None: - params.__setattr__(new, kwargs[old]) + # Show warnings + migrate the arguments + for old, new in deprecated_args: + if old in kwargs: + _deprecation_warning(old, new) + if kwargs.get(old, None) is not None: + params.__setattr__(new, kwargs[old])