From 2d3f61aa074ca738a56bdc374b9f421d485a613b Mon Sep 17 00:00:00 2001 From: Sam Sykes Date: Thu, 31 Jul 2025 21:51:38 +0100 Subject: [PATCH] Updated Speechmatics Plugin (#2225) Changes Split out module attributes to make engine settings clearer Removed internal audio buffer to use latest Speechmatics python SDK (0.4.0) Use diarization for improved VAD in multi-speaker situations Support custom dictionary / vocabulary with attributes Deprecated attributes superseded by re-organised attributes Diarization Enhancements Focus on specific speakers (using speaker labels) Ignore specific speakers (using speaker labels) Separate transcription formats for active and inactive speakers Support for known speakers --- CHANGELOG.md | 4 + .../07a-interruptible-speechmatics-vad.py | 170 ++++ .../07a-interruptible-speechmatics.py | 16 +- .../13h-speechmatics-transcription.py | 8 +- pyproject.toml | 2 +- src/pipecat/services/speechmatics/stt.py | 806 ++++++++++++------ 6 files changed, 749 insertions(+), 257 deletions(-) create mode 100644 examples/foundational/07a-interruptible-speechmatics-vad.py diff --git a/CHANGELOG.md b/CHANGELOG.md index ff322a7d7..1305c54ba 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -122,6 +122,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - For `LmntTTSService`, changed the default `model` to `blizzard`, LMNT's recommended model. +- Updated `SpeechmaticsSTTService`: + - Added support for additional diarization options. + - Added foundational example `07a-interruptible-speechmatics-vad.py`, which uses VAD detection provided by `SpeechmaticsSTTService`. + ### Fixed - Fixed a `LLMUserResponseAggregator` issue where interruptions were not being diff --git a/examples/foundational/07a-interruptible-speechmatics-vad.py b/examples/foundational/07a-interruptible-speechmatics-vad.py new file mode 100644 index 000000000..f2138c591 --- /dev/null +++ b/examples/foundational/07a-interruptible-speechmatics-vad.py @@ -0,0 +1,170 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import argparse +import os + +from dotenv import load_dotenv +from loguru import logger + +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.llm_response import ( + LLMUserAggregatorParams, +) +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.services.elevenlabs.tts import ElevenLabsTTSService +from pipecat.services.openai.base_llm import BaseOpenAILLMService +from pipecat.services.openai.llm import OpenAILLMService +from pipecat.services.speechmatics.stt import SpeechmaticsSTTService +from pipecat.transcriptions.language import Language +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams +from pipecat.transports.services.daily import DailyParams + +load_dotenv(override=True) + +# We store functions so objects (e.g. SileroVADAnalyzer) don't get +# instantiated. The function will be called when the desired transport gets +# selected. +transport_params = { + "daily": lambda: DailyParams( + audio_in_enabled=True, + audio_out_enabled=True, + ), + "twilio": lambda: FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + ), + "webrtc": lambda: TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + ), +} + + +async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_sigint: bool): + """Speechmatics STT Service Example + + This example demonstrates using Speechmatics Speech-to-Text service with speaker diarization and intelligent speaker management. Key features: + + 1. Speaker Diarization + - Automatically identifies and distinguishes between different speakers + - First speaker is identified as 'S1', others get subsequent IDs + - Uses `enable_diarization` parameter to manage speaker detection + + 2. Smart Speaker Control + - `focus_speakers` parameter lets you target specific speakers (e.g. ["S1"]) + - Other speakers will be wrapped in PASSIVE tags + - Only processes speech from focused speakers + - Words from all speakers are wrapped with XML tags for clear speaker identification + - Other speakers' speech only sent when focused speaker is active + + 3. Voice Activity Detection + - Built-in VAD using `enable_vad` parameter + - Remove `vad_analyzer` from `transport` config to use module's VAD + - Emits speaker started/stopped events + + 4. Configuration Options + - `operating_point` parameter defaults to `ENHANCED` for optimal accuracy + - Configurable `end_of_utterance_silence_trigger` (default 0.5s) + - Customizable speaker formatting + - Additional diarization settings available + + For detailed information about operating points and configuration: + https://docs.speechmatics.com/rt-api-ref + """ + + logger.info(f"Starting bot") + + stt = SpeechmaticsSTTService( + 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, + speaker_active_format="<{speaker_id}>{text}", + speaker_passive_format="<{speaker_id}>{text}", + ), + ) + + tts = ElevenLabsTTSService( + api_key=os.getenv("ELEVENLABS_API_KEY"), + voice_id=os.getenv("ELEVENLABS_VOICE_ID"), + model="eleven_turbo_v2_5", + ) + + llm = OpenAILLMService( + api_key=os.getenv("OPENAI_API_KEY"), + params=BaseOpenAILLMService.InputParams(temperature=0.75), + ) + + messages = [ + { + "role": "system", + "content": ( + "You are a helpful British assistant called Alfred. " + "Your goal is to demonstrate your capabilities in a succinct way. " + "Your output will be converted to audio so don't include special characters in your answers. " + "Always include punctuation in your responses. " + "Give very short replies - do not give longer replies unless strictly necessary. " + "Respond to what the user said in a concise, funny, creative and helpful way. " + "Use `` tags to identify different speakers - do not use tags in your replies. " + "Do not respond to speakers within `` tags unless explicitly asked to. " + ), + }, + ] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator( + context, + user_params=LLMUserAggregatorParams(aggregation_timeout=0.005), + ) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, + context_aggregator.user(), # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_metrics=True, + enable_usage_metrics=True, + ), + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + # Kick off the conversation. + messages.append({"role": "system", "content": "Say a short hello to the user."}) + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + await task.cancel() + + runner = PipelineRunner(handle_sigint=handle_sigint) + + await runner.run(task) + + +if __name__ == "__main__": + from pipecat.examples.run import main + + main(run_example, transport_params=transport_params) diff --git a/examples/foundational/07a-interruptible-speechmatics.py b/examples/foundational/07a-interruptible-speechmatics.py index 1582e79ba..55003f964 100644 --- a/examples/foundational/07a-interruptible-speechmatics.py +++ b/examples/foundational/07a-interruptible-speechmatics.py @@ -59,9 +59,6 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si instructions in the system context for the LLM. This greatly improves the conversation experience by allowing the LLM to understand who is speaking in a multi-party call. - If you do not wish to use diarization, then set the `enable_speaker_diarization` parameter - to `False` or omit it altogether. The `text_format` will only be used if diarization is enabled. - By default, this example will use our ENHANCED operating point, which is optimized for high accuracy. You can change this by setting the `operating_point` parameter to a different value. @@ -73,14 +70,17 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si stt = SpeechmaticsSTTService( api_key=os.getenv("SPEECHMATICS_API_KEY"), - language=Language.EN, - enable_speaker_diarization=True, - text_format="<{speaker_id}>{text}", + params=SpeechmaticsSTTService.InputParams( + language=Language.EN, + enable_diarization=True, + end_of_utterance_silence_trigger=0.5, + speaker_active_format="<{speaker_id}>{text}", + ), ) tts = ElevenLabsTTSService( - api_key=os.getenv("ELEVENLABS_API_KEY", ""), - voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""), + api_key=os.getenv("ELEVENLABS_API_KEY"), + voice_id=os.getenv("ELEVENLABS_VOICE_ID"), model="eleven_turbo_v2_5", ) diff --git a/examples/foundational/13h-speechmatics-transcription.py b/examples/foundational/13h-speechmatics-transcription.py index ec3197e19..ea75702e6 100644 --- a/examples/foundational/13h-speechmatics-transcription.py +++ b/examples/foundational/13h-speechmatics-transcription.py @@ -62,9 +62,11 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si stt = SpeechmaticsSTTService( api_key=os.getenv("SPEECHMATICS_API_KEY"), - language=Language.EN, - enable_speaker_diarization=True, - text_format="<{speaker_id}>{text}", + params=SpeechmaticsSTTService.InputParams( + language=Language.EN, + enable_diarization=True, + speaker_active_format="<{speaker_id}>{text}", + ), ) tl = TranscriptionLogger() diff --git a/pyproject.toml b/pyproject.toml index a5275881e..2086b24af 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -95,7 +95,7 @@ silero = [ "onnxruntime~=1.20.1" ] simli = [ "simli-ai~=0.1.10"] soniox = [ "websockets>=13.1,<15.0" ] soundfile = [ "soundfile~=0.13.0" ] -speechmatics = [ "speechmatics-rt>=0.3.1" ] +speechmatics = [ "speechmatics-rt>=0.4.0" ] tavus=[] together = [] tracing = [ "opentelemetry-sdk>=1.33.0", "opentelemetry-api>=1.33.0", "opentelemetry-instrumentation>=0.54b0" ] diff --git a/src/pipecat/services/speechmatics/stt.py b/src/pipecat/services/speechmatics/stt.py index 5cba8d931..303c214bb 100644 --- a/src/pipecat/services/speechmatics/stt.py +++ b/src/pipecat/services/speechmatics/stt.py @@ -8,21 +8,29 @@ import asyncio import datetime +import os import re +import warnings from dataclasses import dataclass, field -from typing import Any, AsyncGenerator, Optional +from enum import Enum +from typing import Any, AsyncGenerator from urllib.parse import urlencode from loguru import logger +from pydantic import BaseModel from pipecat.frames.frames import ( + BotInterruptionFrame, CancelFrame, EndFrame, Frame, InterimTranscriptionFrame, StartFrame, TranscriptionFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, ) +from pipecat.processors.frame_processor import FrameDirection from pipecat.services.stt_service import STTService from pipecat.transcriptions.language import Language from pipecat.utils.tracing.service_decorators import traced_stt @@ -32,10 +40,10 @@ try: AsyncClient, AudioEncoding, AudioFormat, + ClientMessageType, ConversationConfig, OperatingPoint, ServerMessageType, - SpeakerDiarizationConfig, TranscriptionConfig, __version__, ) @@ -47,84 +55,45 @@ except ModuleNotFoundError as e: raise Exception(f"Missing module: {e}") -class AudioBuffer: - """Audio buffer for STT clients. +class EndOfUtteranceMode(str, Enum): + """End of turn delay options for transcription.""" - The Python SDK expects audio in a pre-defined number of frames. This - buffer will accumulate the data from the pipeline and provide it to the - STT client in the correct lengths, waiting for the number of frames to - be available. + NONE = "none" + FIXED = "fixed" + ADAPTIVE = "adaptive" + + +class DiarizationFocusMode(str, Enum): + """Speaker focus mode for diarization.""" + + RETAIN = "retain" + IGNORE = "ignore" + + +@dataclass +class AdditionalVocabEntry: + """Additional vocabulary entry. + + Attributes: + content: The word to add to the dictionary. + sounds_like: Similar words to the word. """ - def __init__(self, maxsize: int = 0): - """Initialize the audio buffer. + content: str + sounds_like: list[str] = field(default_factory=list) - Args: - maxsize: Maximum size of the buffer. - """ - self._queue = asyncio.Queue(maxsize=maxsize) - self._current_chunk = b"" - self._position = 0 - self._closed = False - def write_audio(self, data: bytes) -> None: - """Write audio data to the buffer (thread-safe). +@dataclass +class DiarizationKnownSpeaker: + """Known speakers for speaker diarization. - Args: - data: Audio data to write. - """ - if data: - try: - self._queue.put_nowait(data) - except asyncio.QueueFull: - pass + Attributes: + label: The label of the speaker. + speaker_identifiers: One or more data strings for the speaker. + """ - async def read(self, size: int) -> bytes: - """Read exactly `size` bytes from the buffer (thread-safe). - - This process will block until the required number of bytes are available - in the buffer. Audio is received from the pipeline in varying sizes, so - this buffer will accumulate the data and provide it to the STT client in - the correct lengths, waiting for the number of frames to be available. - - Calling stop() will close the buffer and release the blocking read - process. - - Args: - size: Number of bytes to read. - - Returns: - bytes: Audio data read from the buffer. - """ - result = b"" - bytes_needed = size - - while bytes_needed > 0 and not self._closed: - # Use data from current chunk if available - if self._position < len(self._current_chunk): - available = len(self._current_chunk) - self._position - take = min(bytes_needed, available) - result += self._current_chunk[self._position : self._position + take] - self._position += take - bytes_needed -= take - continue - - # Get next chunk - try: - chunk = await asyncio.wait_for(self._queue.get(), timeout=0.1) - if chunk is None: - continue - self._current_chunk = chunk - self._position = 0 - except asyncio.TimeoutError: - await asyncio.sleep(0) - continue - - return result - - def stop(self) -> None: - """Close the audio buffer.""" - self._closed = True + label: str + speaker_identifiers: list[str] @dataclass @@ -137,6 +106,8 @@ class SpeechFragment: 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`. @@ -149,11 +120,13 @@ class SpeechFragment: 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: Optional[str] = None + speaker: str | None = None confidence: float = 1.0 - result: Optional[Any] = None + result: Any | None = None @dataclass @@ -162,21 +135,23 @@ class SpeakerFragments: 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: Optional[str] = None - timestamp: Optional[str] = None - language: Optional[Language] = None + 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: Optional[str] = None) -> str: + def _format_text(self, format: str | None = None) -> str: """Wrap text with speaker ID in an optional f-string format. Args: @@ -200,17 +175,22 @@ class SpeakerFragments: return content return format.format(**{"speaker_id": self.speaker_id, "text": content}) - def _as_frame_attributes(self, format: Optional[str] = None) -> dict[str, Any]: + 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: - format: Format to wrap the text with. + 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(format), + "text": self._format_text(active_format if self.is_active else passive_format), "user_id": self.speaker_id, "timestamp": self.timestamp, "language": self.language, @@ -226,72 +206,209 @@ class SpeechmaticsSTTService(STTService): and speaker diarization. """ + 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. + + 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. + + 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. + + 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 + other speakers are still processed, but only emitted when a focussed speaker has + also said new words. A list of labels (e.g. `S1`, `S2`) or identifiers of known + speakers (e.g. `speaker_1`, `speaker_2`) can be used. + Defaults to []. + + ignore_speakers: List of speaker IDs to ignore. When enabled, these speakers are + excluded from the transcription and their words are not processed. Their speech + will not trigger any VAD or end of utterance detection. By default, any speaker + with a label starting and ending with double underscores will be excluded (e.g. + `__ASSISTANT__`). + Defaults to []. + + focus_mode: Speaker focus mode for diarization. When set to `DiarizationFocusMode.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`, + the STT engine will ignore words spoken by other speakers and they will not be processed. + Defaults to `DiarizationFocusMode.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 + any spoken words to that speaker. This is useful when you want to attribute words + to a specific speaker, such as the assistant or a specific user. Labels and identifiers + can be obtained from a running STT session and then used in subsequent sessions. + Identifiers are unique to each Speechmatics account and cannot be used across accounts. + Refer to our examples on the format of the known_speakers parameter. + Defaults to []. + + chunk_size: Audio chunk size for streaming. Defaults to 160. + audio_encoding: Audio encoding format. Defaults to AudioEncoding.PCM_S16LE. + """ + + # 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 + + # 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 + focus_speakers: list[str] = [] + ignore_speakers: list[str] = [] + focus_mode: DiarizationFocusMode = DiarizationFocusMode.RETAIN + known_speakers: list[DiarizationKnownSpeaker] = [] + + # Audio + chunk_size: int = 160 + audio_encoding: AudioEncoding = AudioEncoding.PCM_S16LE + + class UpdateParams(BaseModel): + """Update parameters for Speechmatics STT service. + + These are the only parameters that can be changed once a session has started. If you need to + change the language, etc., then you must create a new instance of the service. + + Parameters: + 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 + other speakers are still processed, but only emitted when a focussed speaker has + also said new words. A list of labels (e.g. `S1`, `S2`) or identifiers of known + speakers (e.g. `speaker_1`, `speaker_2`) can be used. + Defaults to []. + + ignore_speakers: List of speaker IDs to ignore. When enabled, these speakers are + excluded from the transcription and their words are not processed. Their speech + will not trigger any VAD or end of utterance detection. By default, any speaker + with a label starting and ending with double underscores will be excluded (e.g. + `__ASSISTANT__`). + Defaults to []. + + focus_mode: Speaker focus mode for diarization. When set to `DiarizationFocusMode.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`, + the STT engine will ignore words spoken by other speakers and they will not be processed. + Defaults to `DiarizationFocusMode.RETAIN`. + """ + + focus_speakers: list[str] = [] + ignore_speakers: list[str] = [] + focus_mode: DiarizationFocusMode = DiarizationFocusMode.RETAIN + def __init__( self, *, - api_key: str, - language: Optional[Language] = None, - language_code: Optional[str] = None, - base_url: str = "wss://eu2.rt.speechmatics.com/v2", - domain: Optional[str] = None, - output_locale: Optional[Language] = None, - output_locale_code: Optional[str] = None, - enable_partials: bool = True, - max_delay: float = 1.5, - sample_rate: Optional[int] = 16000, - chunk_size: int = 256, - audio_encoding: AudioEncoding = AudioEncoding.PCM_S16LE, - end_of_utterance_silence_trigger: float = 0.5, - operating_point: OperatingPoint = OperatingPoint.ENHANCED, - enable_speaker_diarization: bool = False, - text_format: str = "<{speaker_id}>{text}", - max_speakers: Optional[int] = None, - transcription_config: Optional[TranscriptionConfig] = None, + api_key: str | None = None, + base_url: str | None = None, + sample_rate: int = 16000, + params: InputParams | None = None, **kwargs, ): """Initialize the Speechmatics STT service. Args: - api_key: Speechmatics API key for authentication. - language: Language code for transcription. Defaults to `None`. - language_code: Language code string for transcription. Defaults to `None`. - base_url: Base URL for Speechmatics API. Defaults to `wss://eu2.rt.speechmatics.com/v2`. - domain: Domain for Speechmatics API. Defaults to `None`. - output_locale: Output locale for transcription, e.g. `Language.EN_GB`. Defaults to `None`. - output_locale_code: Output locale code for transcription. Defaults to `None`. - enable_partials: Enable partial transcription results. Defaults to `True`. - max_delay: Maximum delay for transcription in seconds. Defaults to `1.5`. - sample_rate: Audio sample rate in Hz. Defaults to `16000`. - chunk_size: Audio chunk size for streaming. Defaults to `256`. - audio_encoding: Audio encoding format. Defaults to `pcm_s16le`. - end_of_utterance_silence_trigger: Silence duration in seconds to trigger end of utterance detection. Defaults to `0.5`. - operating_point: Operating point for transcription accuracy vs. latency tradeoff. Defaults to `enhanced`. - enable_speaker_diarization: Enable speaker diarization to identify different speakers. Defaults to `False`. - text_format: Wrapper for speaker ID. Defaults to `<{speaker_id}>{text}`. - max_speakers: Maximum number of speakers to detect. Defaults to `None` (auto-detect). - transcription_config: Custom transcription configuration (other set parameters are merged). Defaults to `None`. + api_key: Speechmatics API key for authentication. Uses environment variable + `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. + params: Optional[InputParams]: Input parameters for the service. **kwargs: Additional arguments passed to STTService. """ super().__init__(sample_rate=sample_rate, **kwargs) - # Client configuration - self._api_key: str = api_key - self._language: Optional[Language] = language - self._language_code: Optional[str] = language_code - self._base_url: str = base_url - self._domain: Optional[str] = domain - self._output_locale: Optional[Language] = output_locale - self._output_locale_code: Optional[str] = output_locale_code - self._enable_partials: bool = enable_partials - self._max_delay: float = max_delay - self._sample_rate: int = sample_rate - self._chunk_size: int = chunk_size - self._audio_encoding: AudioEncoding = audio_encoding - self._end_of_utterance_silence_trigger: Optional[float] = end_of_utterance_silence_trigger - self._operating_point: OperatingPoint = operating_point - self._enable_speaker_diarization: bool = enable_speaker_diarization - self._text_format: str = text_format - self._max_speakers: Optional[int] = max_speakers + # Service parameters + self._api_key: str = api_key or os.getenv("SPEECHMATICS_API_KEY") + self._base_url: str = ( + base_url or os.getenv("SPEECHMATICS_RT_URL") or "wss://eu2.rt.speechmatics.com/v2" + ) # Check we have required attributes if not self._api_key: @@ -299,33 +416,36 @@ class SpeechmaticsSTTService(STTService): if not self._base_url: raise ValueError("Missing Speechmatics base URL") - # Validate the language code - if self._language and self._language_code: - raise ValueError("Language and language code cannot both be specified") - elif self._language: - self._language_code = _language_to_speechmatics_language(self._language) + # Default parameters + self._params = params or SpeechmaticsSTTService.InputParams() - # Validate the output locale code - if self._output_locale and self._output_locale_code: - raise ValueError("Output locale and output locale code cannot both be specified") - elif self._output_locale: - self._output_locale_code = _locale_to_speechmatics_locale( - self._language_code, self._output_locale - ) + # Deprecation check + _check_deprecated_args(kwargs, self._params) # Complete configuration objects self._transcription_config: TranscriptionConfig = None - self._process_config(transcription_config) + self._process_config() # STT client - self._client: Optional[AsyncClient] = None - self._client_task: Optional[asyncio.Task] = None - self._audio_buffer: AudioBuffer = AudioBuffer(maxsize=10) - self._start_time: Optional[datetime.datetime] = None + self._client: AsyncClient | None = None # Current utterance speech data self._speech_fragments: list[SpeechFragment] = [] + # Speaking states + self._is_speaking: bool = False + + # Timing info + self._start_time: datetime.datetime | None = None + self._total_time: datetime.timedelta | None = None + + # Event handlers + if self._params.enable_diarization: + self._register_event_handler("on_speakers_result") + + # EndOfUtterance fallback timer + self._end_of_utterance_timer: asyncio.Task | None = None + async def start(self, frame: StartFrame): """Called when the new session starts.""" await super().start(frame) @@ -343,20 +463,53 @@ class SpeechmaticsSTTService(STTService): async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: """Adds audio to the audio buffer and yields None.""" - self._audio_buffer.write_audio(audio) + await self._client.send_audio(audio) yield None - async def _run_client(self) -> None: - """Runs the Speechmatics client in a thread.""" - await self._client.transcribe( - self._audio_buffer, - transcription_config=self._transcription_config, - audio_format=AudioFormat( - encoding=self._audio_encoding, - sample_rate=self.sample_rate, - chunk_size=self._chunk_size, - ), - ) + def update_params( + self, + params: UpdateParams, + ) -> None: + """Updates the speaker configuration. + + This can update the speakers to listen to or ignore during an in-flight + transcription. Only available if diarization is enabled. + + Args: + params: Update parameters for the service. + """ + # Check possible + if not self._params.enable_diarization: + raise ValueError("Diarization is not enabled") + + # Update the diarization configuration + if params.focus_speakers is not None: + self._params.focus_speakers = params.focus_speakers + if params.ignore_speakers is not None: + self._params.ignore_speakers = params.ignore_speakers + if params.focus_mode is not None: + self._params.focus_mode = params.focus_mode + + async def send_message(self, message: ClientMessageType | 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 + is not running, this will raise an exception. Messages in the wrong format will also + cause an error. + + Args: + message: Message to send to the STT service. + **kwargs: Additional arguments passed to the underlying transport. + """ + 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() + ) + except Exception as e: + raise RuntimeError(f"error sending message to STT: {e}") async def _connect(self) -> None: """Connect to the STT service.""" @@ -376,9 +529,11 @@ class SpeechmaticsSTTService(STTService): self._start_time = datetime.datetime.now(datetime.timezone.utc) # Partial transcript event - @self._client.on(ServerMessageType.ADD_PARTIAL_TRANSCRIPT) - def _evt_on_partial_transcript(message: dict[str, Any]): - self._handle_transcript(message, is_final=False) + 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) @@ -386,21 +541,38 @@ class SpeechmaticsSTTService(STTService): self._handle_transcript(message, is_final=True) # End of Utterance - @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._send_frames(finalized=True), self.get_event_loop() - ) + if self._params.end_of_utterance_mode == EndOfUtteranceMode.FIXED: - # Start the client in a thread - self._client_task = self.create_task(self._run_client()) + @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 + 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, + ), + ) async def _disconnect(self) -> None: """Disconnect from the STT service.""" - # Stop the audio buffer - self._audio_buffer.stop() - # Disconnect the client try: if self._client: @@ -412,62 +584,64 @@ class SpeechmaticsSTTService(STTService): finally: self._client = None - # Cancel the client task - if self._client_task: - await self.cancel_task(self._client_task) - self._client_task = None - # Log the event logger.debug("Disconnected from Speechmatics STT service") - def _process_config(self, transcription_config: Optional[TranscriptionConfig] = None) -> None: + def _process_config(self) -> None: """Create a formatted STT transcription config. - This takes an optional TranscriptionConfig object and populates it with the - values from the STT service. Individual parameters take priority over those - within the config object. - - Args: - transcription_config: Optional transcription config to use. + Creates a transcription config object based on the service parameters. Aligns + with the Speechmatics RT API transcription config. """ # Transcription config - if not transcription_config: - transcription_config = TranscriptionConfig( - language=self._language_code or "en", - domain=self._domain, - output_locale=self._output_locale_code, - operating_point=self._operating_point, - diarization="speaker" if self._enable_speaker_diarization else None, - enable_partials=self._enable_partials, - max_delay=self._max_delay or 2.0, - ) - else: - if self._language_code: - transcription_config.language = self._language_code - if self._domain: - transcription_config.domain = self._domain - if self._output_locale_code: - transcription_config.output_locale = self._output_locale_code - if self._operating_point: - transcription_config.operating_point = self._operating_point - if self._enable_speaker_diarization: - transcription_config.diarization = "speaker" - if self._enable_partials: - transcription_config.enable_partials = self._enable_partials - if self._max_delay: - transcription_config.max_delay = self._max_delay + transcription_config = TranscriptionConfig( + language=self._params.language, + domain=self._params.domain, + output_locale=self._params.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, + } + for e in self._params.additional_vocab + ] # Diarization - if self._enable_speaker_diarization and self._max_speakers: - transcription_config.speaker_diarization_config = SpeakerDiarizationConfig( - max_speakers=self._max_speakers, + 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, ) - # End of Utterance - if self._end_of_utterance_silence_trigger: - transcription_config.conversation_config = ConversationConfig( - end_of_utterance_silence_trigger=self._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 @@ -489,16 +663,59 @@ class SpeechmaticsSTTService(STTService): 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: Optional[Language] = None + 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. @@ -517,34 +734,63 @@ class SpeechmaticsSTTService(STTService): if not speech_frames: return - # If final, then re=parse into TranscriptionFrame + # Check at least one frame is active + if not any(frame.is_active for frame in speech_frames): + return + + # Frames to send + upstream_frames: list[Frame] = [] + 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 + upstream_frames += [BotInterruptionFrame()] + downstream_frames += [UserStartedSpeakingFrame()] + + # If final, then re-parse into TranscriptionFrame if finalized: # Reset the speech fragments self._speech_fragments.clear() # Transform frames - frames = [ - TranscriptionFrame(**frame._as_frame_attributes(self._text_format)) + 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 frames]}") + logger.debug(f"Finalized transcript: {[f.text for f in downstream_frames]}") - # Return as interim results + # Return as interim results (unformatted) else: - frames = [ - InterimTranscriptionFrame(**frame._as_frame_attributes()) for frame in speech_frames + downstream_frames += [ + InterimTranscriptionFrame( + **frame._as_frame_attributes( + self._params.speaker_active_format, self._params.speaker_passive_format + ) + ) + for frame in speech_frames ] - # Send the frames back to pipecat - for frame in frames: - await self._handle_transcription( - transcript=frame.text, - is_final=finalized, - language=frame.language, - ) - await self.push_frame(frame) + # 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 UPSTREAM frames + for frame in upstream_frames: + await self.push_frame(frame, FrameDirection.UPSTREAM) + + # 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. @@ -587,9 +833,26 @@ class SpeechmaticsSTTService(STTService): result=result, ) - # Drop `__XX__` speakers - if fragment.speaker and re.match(r"^__[A-Z0-9_]{2,}__$", fragment.speaker): - continue + # 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) @@ -617,7 +880,7 @@ class SpeechmaticsSTTService(STTService): in strict order for the context of the conversation. Returns: - list[SpeakerFragments]: The list of objects. + List[SpeakerFragments]: The list of objects. """ # Speaker groups current_speaker: str | None = None @@ -680,12 +943,18 @@ class SpeechmaticsSTTService(STTService): timespec="milliseconds" ) + # 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 SpeakerFragments object return SpeakerFragments( speaker_id=group[0].speaker, timestamp=ts, language=group[0].language, fragments=group, + is_active=is_active, ) @@ -783,7 +1052,7 @@ def _language_to_speechmatics_language(language: Language) -> str: return result -def _locale_to_speechmatics_locale(language_code: str, locale: Language) -> Optional[str]: +def _locale_to_speechmatics_locale(language_code: str, locale: Language) -> str | None: """Convert a Language enum to a Speechmatics language code. Args: @@ -811,3 +1080,50 @@ def _locale_to_speechmatics_locale(language_code: str, locale: Language) -> Opti # 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): + 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"), + ("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])