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}")