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])