From 4050e8b7dc65f39f03d1a7e13c211157f923d4ba Mon Sep 17 00:00:00 2001 From: Aaron Ng Date: Wed, 29 Oct 2025 14:53:20 +0000 Subject: [PATCH 1/3] add speechmatics tts --- .../07a-interruptible-speechmatics-vad.py | 35 ++-- .../07a-interruptible-speechmatics.py | 38 ++-- src/pipecat/services/speechmatics/tts.py | 174 ++++++++++++++++++ 3 files changed, 217 insertions(+), 30 deletions(-) create mode 100644 src/pipecat/services/speechmatics/tts.py diff --git a/examples/foundational/07a-interruptible-speechmatics-vad.py b/examples/foundational/07a-interruptible-speechmatics-vad.py index 55514017f..666f580f8 100644 --- a/examples/foundational/07a-interruptible-speechmatics-vad.py +++ b/examples/foundational/07a-interruptible-speechmatics-vad.py @@ -20,10 +20,10 @@ from pipecat.processors.aggregators.llm_response import ( from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport -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.services.speechmatics.tts import SpeechmaticsTTSService from pipecat.transcriptions.language import Language from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams @@ -51,35 +51,41 @@ transport_params = { async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): - """Speechmatics STT Service Example + """Speechmatics STT and TTS Service Example - This example demonstrates using Speechmatics Speech-to-Text service with speaker diarization and intelligent speaker management. Key features: + This example demonstrates using Speechmatics Speech-to-Text and Text-to-Speech services + with speaker diarization and intelligent speaker management. Key features: - 1. Speaker Diarization + 1. Speaker Diarization (STT) - 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 + 2. Smart Speaker Control (STT) - `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 + 3. Voice Activity Detection (STT) - 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 + 4. Text-to-Speech (TTS) + - Low latency streaming audio synthesis + - Multiple voice options available including `sarah`, `theo`, and `megan` + + 5. 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 + For detailed information: + - STT: https://docs.speechmatics.com/rt-api-ref + - TTS: https://docs.speechmatics.com/text-to-speech/quickstart """ logger.info(f"Starting bot") @@ -97,10 +103,11 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ), ) - tts = ElevenLabsTTSService( - api_key=os.getenv("ELEVENLABS_API_KEY"), - voice_id=os.getenv("ELEVENLABS_VOICE_ID"), - model="eleven_turbo_v2_5", + tts = SpeechmaticsTTSService( + api_key=os.getenv("SPEECHMATICS_API_KEY"), + params=SpeechmaticsTTSService.InputParams( + voice="sarah", + ), ) llm = OpenAILLMService( @@ -112,7 +119,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): { "role": "system", "content": ( - "You are a helpful British assistant called Alfred. " + "You are a helpful British assistant called Sarah. " "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. " diff --git a/examples/foundational/07a-interruptible-speechmatics.py b/examples/foundational/07a-interruptible-speechmatics.py index 3d1e639b9..196b6bf68 100644 --- a/examples/foundational/07a-interruptible-speechmatics.py +++ b/examples/foundational/07a-interruptible-speechmatics.py @@ -24,10 +24,10 @@ from pipecat.processors.aggregators.llm_response import ( from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport -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.services.speechmatics.tts import SpeechmaticsTTSService from pipecat.transcriptions.language import Language from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams @@ -61,19 +61,24 @@ transport_params = { async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): - """Run example using Speechmatics STT. + """Run example using Speechmatics STT and TTS. - This example will use diarization within our STT service and output the words spoken by - each individual speaker and wrap them with XML tags for the LLM to process. Note the - 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. + This example demonstrates a complete Speechmatics integration with both Speech-to-Text + and Text-to-Speech services: - 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. + STT Features: + - Diarization to identify and distinguish between different speakers + - Words spoken by each speaker are wrapped with XML tags for LLM processing + - System context instructions help the LLM understand multi-party conversations + - ENHANCED operating point by default for optimal accuracy - For more information on operating points, see the Speechmatics documentation: - https://docs.speechmatics.com/rt-api-ref + TTS Features: + - Low latency streaming audio synthesis + - Multiple voice options available including `sarah`, `theo`, and `megan` + + For more information: + - STT: https://docs.speechmatics.com/rt-api-ref + - TTS: https://docs.speechmatics.com/text-to-speech/quickstart """ logger.info(f"Starting bot") @@ -87,10 +92,11 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ), ) - tts = ElevenLabsTTSService( - api_key=os.getenv("ELEVENLABS_API_KEY"), - voice_id=os.getenv("ELEVENLABS_VOICE_ID"), - model="eleven_turbo_v2_5", + tts = SpeechmaticsTTSService( + api_key=os.getenv("SPEECHMATICS_API_KEY"), + params=SpeechmaticsTTSService.InputParams( + voice="sarah", + ), ) llm = OpenAILLMService( @@ -102,7 +108,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): { "role": "system", "content": ( - "You are a helpful British assistant called Alfred. " + "You are a helpful British assistant called Sarah. " "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. " diff --git a/src/pipecat/services/speechmatics/tts.py b/src/pipecat/services/speechmatics/tts.py new file mode 100644 index 000000000..46421360f --- /dev/null +++ b/src/pipecat/services/speechmatics/tts.py @@ -0,0 +1,174 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Speechmatics TTS service integration.""" + +import os +from typing import AsyncGenerator, Optional + +import aiohttp +import numpy as np +from loguru import logger +from pydantic import BaseModel + +from pipecat.frames.frames import ( + ErrorFrame, + Frame, + TTSAudioRawFrame, + TTSStartedFrame, + TTSStoppedFrame, +) +from pipecat.services.tts_service import TTSService +from pipecat.utils.tracing.service_decorators import traced_tts + + +class SpeechmaticsTTSService(TTSService): + """Speechmatics TTS service implementation. + + This service provides text-to-speech synthesis using the Speechmatics HTTP API. + It converts text to speech and returns raw PCM audio data for real-time playback. + """ + + class InputParams(BaseModel): + """Configuration parameters for Speechmatics TTS service. + + Parameters: + voice: Voice model to use for synthesis. Defaults to "sarah". + """ + + voice: str = "sarah" + + def __init__( + self, + *, + api_key: str | None = None, + base_url: str | None = None, + aiohttp_session: aiohttp.ClientSession | None = None, + sample_rate: Optional[int] = 16000, + params: InputParams | None = None, + **kwargs, + ): + """Initialize the Speechmatics TTS service. + + Args: + api_key: Speechmatics API key for authentication. Uses environment variable + `SPEECHMATICS_API_KEY` if not provided. + base_url: Base URL for Speechmatics TTS API. Defaults to + `https://preview.tts.speechmatics.com`. + aiohttp_session: Shared aiohttp session for HTTP requests. + sample_rate: Audio sample rate in Hz. Defaults to 16000. + params: Optional[InputParams]: Input parameters for the service. + **kwargs: Additional arguments passed to TTSService. + """ + super().__init__(sample_rate=sample_rate, **kwargs) + + # Service parameters + self._api_key: str = api_key or os.getenv("SPEECHMATICS_API_KEY") + self._base_url: str = base_url or "https://preview.tts.speechmatics.com" + self._session = aiohttp_session or aiohttp.ClientSession() + + # Check we have required attributes + if not self._api_key: + raise ValueError("Missing Speechmatics API key") + if not self._base_url: + raise ValueError("Missing Speechmatics base URL") + + # Default parameters + self._params = params or SpeechmaticsTTSService.InputParams() + + # Set voice from parameters + self.set_voice(self._params.voice) + + def can_generate_metrics(self) -> bool: + """Check if this service can generate processing metrics. + + Returns: + True, as Speechmatics service supports metrics generation. + """ + return True + + @traced_tts + async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + """Generate speech from text using Speechmatics' HTTP API. + + Args: + text: The text to synthesize into speech. + + Yields: + Frame: Audio frames containing the synthesized speech. + """ + logger.debug(f"{self}: Generating TTS [{text}]") + + headers = { + "Authorization": f"Bearer {self._api_key}", + "Content-Type": "application/json", + } + + payload = { + "text": text, + } + + url = f"{self._base_url}/generate/{self._voice_id}?output_format=pcm_16000" + + try: + await self.start_ttfb_metrics() + + async with self._session.post(url, json=payload, headers=headers) as response: + if response.status != 200: + error_message = f"Speechmatics TTS error: HTTP {response.status}" + logger.error(error_message) + yield ErrorFrame(error=error_message) + return + + await self.start_tts_usage_metrics(text) + + yield TTSStartedFrame() + + # Process the response in streaming chunks + first_chunk = True + buffer = b"" + + # Helper to move all complete 2-byte int16 samples from buffer into a frame + def _emit_complete_samples(): + nonlocal buffer + if len(buffer) < 2: + return None + complete_samples = len(buffer) // 2 + complete_bytes = complete_samples * 2 + + audio_data = buffer[:complete_bytes] + buffer = buffer[complete_bytes:] # Keep remaining bytes for next iteration + + return TTSAudioRawFrame( + audio=audio_data, + sample_rate=self.sample_rate, + num_channels=1, + ) + + async for chunk in response.content.iter_any(): + if not chunk: + continue + if first_chunk: + await self.stop_ttfb_metrics() + first_chunk = False + + buffer += chunk + + # Emit a frame for all complete samples currently in buffer + frame = _emit_complete_samples() + if frame: + yield frame + + # Process any remaining bytes in buffer after streaming ends + frame = _emit_complete_samples() + if frame: + yield frame + + except Exception as e: + logger.exception(f"Error generating TTS: {e}") + yield ErrorFrame(error=f"Speechmatics TTS error: {str(e)}") + finally: + yield TTSStoppedFrame() From b0acbeffb9e0591fbcf41703df9a723835cb1299 Mon Sep 17 00:00:00 2001 From: Aaron Ng Date: Wed, 29 Oct 2025 16:33:18 +0000 Subject: [PATCH 2/3] add sm-app param --- src/pipecat/services/speechmatics/tts.py | 32 ++++++++++++++++++++++-- 1 file changed, 30 insertions(+), 2 deletions(-) diff --git a/src/pipecat/services/speechmatics/tts.py b/src/pipecat/services/speechmatics/tts.py index 46421360f..207f898aa 100644 --- a/src/pipecat/services/speechmatics/tts.py +++ b/src/pipecat/services/speechmatics/tts.py @@ -8,9 +8,9 @@ import os from typing import AsyncGenerator, Optional +from urllib.parse import urlencode import aiohttp -import numpy as np from loguru import logger from pydantic import BaseModel @@ -24,6 +24,15 @@ from pipecat.frames.frames import ( from pipecat.services.tts_service import TTSService from pipecat.utils.tracing.service_decorators import traced_tts +try: + from speechmatics.rt import __version__ +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error( + "In order to use Speechmatics, you need to `pip install pipecat-ai[speechmatics]`." + ) + raise Exception(f"Missing module: {e}") + class SpeechmaticsTTSService(TTSService): """Speechmatics TTS service implementation. @@ -111,7 +120,7 @@ class SpeechmaticsTTSService(TTSService): "text": text, } - url = f"{self._base_url}/generate/{self._voice_id}?output_format=pcm_16000" + url = _get_endpoint_url(self._base_url, self._voice_id, self.sample_rate) try: await self.start_ttfb_metrics() @@ -172,3 +181,22 @@ class SpeechmaticsTTSService(TTSService): yield ErrorFrame(error=f"Speechmatics TTS error: {str(e)}") finally: yield TTSStoppedFrame() + + +def _get_endpoint_url(base_url: str, voice: str, sample_rate: int) -> str: + """Format the TTS endpoint URL with voice, output format, and version params. + + Args: + base_url: The base URL for the TTS endpoint. + voice: The voice model to use. + sample_rate: The audio sample rate. + + Returns: + str: The formatted TTS endpoint URL. + """ + query_params = {} + query_params["output_format"] = f"pcm_{sample_rate}" + query_params["sm-app"] = f"pipecat/{__version__}" + query = urlencode(query_params) + + return f"{base_url}/generate/{voice}?{query}" From 9d509bb40956c7ebcbf8a3d478a7a19038c047fc Mon Sep 17 00:00:00 2001 From: Aaron Ng Date: Thu, 30 Oct 2025 16:25:10 +0000 Subject: [PATCH 3/3] address changes --- .../07a-interruptible-speechmatics-vad.py | 150 +++++++++--------- .../07a-interruptible-speechmatics.py | 141 ++++++++-------- src/pipecat/services/speechmatics/tts.py | 85 +++++----- 3 files changed, 182 insertions(+), 194 deletions(-) diff --git a/examples/foundational/07a-interruptible-speechmatics-vad.py b/examples/foundational/07a-interruptible-speechmatics-vad.py index 666f580f8..6e78a5147 100644 --- a/examples/foundational/07a-interruptible-speechmatics-vad.py +++ b/examples/foundational/07a-interruptible-speechmatics-vad.py @@ -6,6 +6,7 @@ import os +import aiohttp from dotenv import load_dotenv from loguru import logger @@ -89,90 +90,89 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): """ 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 = SpeechmaticsTTSService( - api_key=os.getenv("SPEECHMATICS_API_KEY"), - params=SpeechmaticsTTSService.InputParams( - voice="sarah", - ), - ) - - 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 Sarah. " - "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. " + async with aiohttp.ClientSession() as session: + 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}", ), - }, - ] + ) - context = LLMContext(messages) - context_aggregator = LLMContextAggregatorPair( - context, - user_params=LLMUserAggregatorParams(aggregation_timeout=0.005), - ) + tts = SpeechmaticsTTSService( + api_key=os.getenv("SPEECHMATICS_API_KEY"), + voice_id="sarah", + aiohttp_session=session, + ) - 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 + 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 Sarah. " + "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. " + ), + }, ] - ) - task = PipelineTask( - pipeline, - params=PipelineParams( - enable_metrics=True, - enable_usage_metrics=True, - ), - idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, - ) + context = LLMContext(messages) + context_aggregator = LLMContextAggregatorPair( + context, + user_params=LLMUserAggregatorParams(aggregation_timeout=0.005), + ) - @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([LLMRunFrame()]) + 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 + ] + ) - @transport.event_handler("on_client_disconnected") - async def on_client_disconnected(transport, client): - logger.info(f"Client disconnected") - await task.cancel() + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_metrics=True, + enable_usage_metrics=True, + ), + idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, + ) - runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + @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([LLMRunFrame()]) - await runner.run(task) + @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=runner_args.handle_sigint) + + await runner.run(task) async def bot(runner_args: RunnerArguments): diff --git a/examples/foundational/07a-interruptible-speechmatics.py b/examples/foundational/07a-interruptible-speechmatics.py index 196b6bf68..36ac39b82 100644 --- a/examples/foundational/07a-interruptible-speechmatics.py +++ b/examples/foundational/07a-interruptible-speechmatics.py @@ -6,6 +6,7 @@ import os +import aiohttp from dotenv import load_dotenv from loguru import logger @@ -82,85 +83,85 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): """ logger.info(f"Starting bot") - stt = SpeechmaticsSTTService( - api_key=os.getenv("SPEECHMATICS_API_KEY"), - params=SpeechmaticsSTTService.InputParams( - language=Language.EN, - enable_diarization=True, - end_of_utterance_silence_trigger=0.5, - speaker_active_format="<{speaker_id}>{text}", - ), - ) - - tts = SpeechmaticsTTSService( - api_key=os.getenv("SPEECHMATICS_API_KEY"), - params=SpeechmaticsTTSService.InputParams( - voice="sarah", - ), - ) - - 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 Sarah. " - "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." + async with aiohttp.ClientSession() as session: + stt = SpeechmaticsSTTService( + api_key=os.getenv("SPEECHMATICS_API_KEY"), + params=SpeechmaticsSTTService.InputParams( + language=Language.EN, + enable_diarization=True, + end_of_utterance_silence_trigger=0.5, + speaker_active_format="<{speaker_id}>{text}", ), - }, - ] + ) - context = LLMContext(messages) - context_aggregator = LLMContextAggregatorPair( - context, - user_params=LLMUserAggregatorParams(aggregation_timeout=0.005), - ) + tts = SpeechmaticsTTSService( + api_key=os.getenv("SPEECHMATICS_API_KEY"), + voice_id="sarah", + aiohttp_session=session, + ) - pipeline = Pipeline( - [ - transport.input(), # Transport user input - stt, # STT - context_aggregator.user(), # User responses - llm, # LLM - tts, # TTS - transport.output(), # Transport bot output - context_aggregator.assistant(), # Assistant spoken responses + 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 Sarah. " + "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." + ), + }, ] - ) - task = PipelineTask( - pipeline, - params=PipelineParams( - enable_metrics=True, - enable_usage_metrics=True, - ), - idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, - ) + context = LLMContext(messages) + context_aggregator = LLMContextAggregatorPair( + context, + user_params=LLMUserAggregatorParams(aggregation_timeout=0.005), + ) - @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([LLMRunFrame()]) + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, # STT + context_aggregator.user(), # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses + ] + ) - @transport.event_handler("on_client_disconnected") - async def on_client_disconnected(transport, client): - logger.info(f"Client disconnected") - await task.cancel() + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_metrics=True, + enable_usage_metrics=True, + ), + idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, + ) - runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + @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([LLMRunFrame()]) - await runner.run(task) + @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=runner_args.handle_sigint) + + await runner.run(task) async def bot(runner_args: RunnerArguments): diff --git a/src/pipecat/services/speechmatics/tts.py b/src/pipecat/services/speechmatics/tts.py index 207f898aa..23d10c5e1 100644 --- a/src/pipecat/services/speechmatics/tts.py +++ b/src/pipecat/services/speechmatics/tts.py @@ -6,7 +6,6 @@ """Speechmatics TTS service integration.""" -import os from typing import AsyncGenerator, Optional from urllib.parse import urlencode @@ -41,55 +40,56 @@ class SpeechmaticsTTSService(TTSService): It converts text to speech and returns raw PCM audio data for real-time playback. """ + SPEECHMATICS_SAMPLE_RATE = 16000 + class InputParams(BaseModel): - """Configuration parameters for Speechmatics TTS service. + """Optional input parameters for Speechmatics TTS configuration.""" - Parameters: - voice: Voice model to use for synthesis. Defaults to "sarah". - """ - - voice: str = "sarah" + pass def __init__( self, *, - api_key: str | None = None, - base_url: str | None = None, - aiohttp_session: aiohttp.ClientSession | None = None, - sample_rate: Optional[int] = 16000, - params: InputParams | None = None, + api_key: str, + base_url: str = "https://preview.tts.speechmatics.com", + voice_id: str = "sarah", + aiohttp_session: aiohttp.ClientSession, + sample_rate: Optional[int] = SPEECHMATICS_SAMPLE_RATE, + params: Optional[InputParams] = None, **kwargs, ): """Initialize the Speechmatics TTS service. Args: - api_key: Speechmatics API key for authentication. Uses environment variable - `SPEECHMATICS_API_KEY` if not provided. - base_url: Base URL for Speechmatics TTS API. Defaults to - `https://preview.tts.speechmatics.com`. + api_key: Speechmatics API key for authentication. + base_url: Base URL for Speechmatics TTS API. + voice_id: Voice model to use for synthesis. aiohttp_session: Shared aiohttp session for HTTP requests. - sample_rate: Audio sample rate in Hz. Defaults to 16000. + sample_rate: Audio sample rate in Hz. params: Optional[InputParams]: Input parameters for the service. **kwargs: Additional arguments passed to TTSService. """ + if sample_rate and sample_rate != self.SPEECHMATICS_SAMPLE_RATE: + logger.warning( + f"Speechmatics TTS only supports {self.SPEECHMATICS_SAMPLE_RATE}Hz sample rate. " + f"Current rate of {sample_rate}Hz may cause issues." + ) super().__init__(sample_rate=sample_rate, **kwargs) # Service parameters - self._api_key: str = api_key or os.getenv("SPEECHMATICS_API_KEY") - self._base_url: str = base_url or "https://preview.tts.speechmatics.com" - self._session = aiohttp_session or aiohttp.ClientSession() + self._api_key: str = api_key + self._base_url: str = base_url + self._session = aiohttp_session # Check we have required attributes if not self._api_key: raise ValueError("Missing Speechmatics API key") - if not self._base_url: - raise ValueError("Missing Speechmatics base URL") # Default parameters self._params = params or SpeechmaticsTTSService.InputParams() - # Set voice from parameters - self.set_voice(self._params.voice) + # Set voice from constructor parameter + self.set_voice(voice_id) def can_generate_metrics(self) -> bool: """Check if this service can generate processing metrics. @@ -140,23 +140,6 @@ class SpeechmaticsTTSService(TTSService): first_chunk = True buffer = b"" - # Helper to move all complete 2-byte int16 samples from buffer into a frame - def _emit_complete_samples(): - nonlocal buffer - if len(buffer) < 2: - return None - complete_samples = len(buffer) // 2 - complete_bytes = complete_samples * 2 - - audio_data = buffer[:complete_bytes] - buffer = buffer[complete_bytes:] # Keep remaining bytes for next iteration - - return TTSAudioRawFrame( - audio=audio_data, - sample_rate=self.sample_rate, - num_channels=1, - ) - async for chunk in response.content.iter_any(): if not chunk: continue @@ -166,15 +149,19 @@ class SpeechmaticsTTSService(TTSService): buffer += chunk - # Emit a frame for all complete samples currently in buffer - frame = _emit_complete_samples() - if frame: - yield frame + # Emit all complete 2-byte int16 samples from buffer + if len(buffer) >= 2: + complete_samples = len(buffer) // 2 + complete_bytes = complete_samples * 2 - # Process any remaining bytes in buffer after streaming ends - frame = _emit_complete_samples() - if frame: - yield frame + audio_data = buffer[:complete_bytes] + buffer = buffer[complete_bytes:] # Keep remaining bytes for next iteration + + yield TTSAudioRawFrame( + audio=audio_data, + sample_rate=self.sample_rate, + num_channels=1, + ) except Exception as e: logger.exception(f"Error generating TTS: {e}")