diff --git a/examples/foundational/07a-interruptible-speechmatics-vad.py b/examples/foundational/07a-interruptible-speechmatics-vad.py index 55514017f..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 @@ -20,10 +21,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,121 +52,127 @@ 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") - - 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. " + 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 3d1e639b9..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 @@ -24,10 +25,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,100 +62,106 @@ 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") - 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 = 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." + 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 new file mode 100644 index 000000000..23d10c5e1 --- /dev/null +++ b/src/pipecat/services/speechmatics/tts.py @@ -0,0 +1,189 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Speechmatics TTS service integration.""" + +from typing import AsyncGenerator, Optional +from urllib.parse import urlencode + +import aiohttp +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 + +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. + + 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. + """ + + SPEECHMATICS_SAMPLE_RATE = 16000 + + class InputParams(BaseModel): + """Optional input parameters for Speechmatics TTS configuration.""" + + pass + + def __init__( + self, + *, + 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. + 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. + 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 + 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") + + # Default parameters + self._params = params or SpeechmaticsTTSService.InputParams() + + # Set voice from constructor parameter + self.set_voice(voice_id) + + 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 = _get_endpoint_url(self._base_url, self._voice_id, self.sample_rate) + + 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"" + + 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 all complete 2-byte int16 samples from buffer + if len(buffer) >= 2: + 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 + + yield TTSAudioRawFrame( + audio=audio_data, + sample_rate=self.sample_rate, + num_channels=1, + ) + + except Exception as e: + logger.exception(f"Error generating TTS: {e}") + 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}"