diff --git a/src/pipecat/services/asyncai/tts.py b/src/pipecat/services/asyncai/tts.py index f27e6bfbf..4ff6c928d 100644 --- a/src/pipecat/services/asyncai/tts.py +++ b/src/pipecat/services/asyncai/tts.py @@ -9,7 +9,6 @@ import asyncio import base64 import json -import uuid from typing import AsyncGenerator, Optional import aiohttp @@ -148,7 +147,6 @@ class AsyncAITTSService(AudioContextTTSService): self._receive_task = None self._keepalive_task = None - self._started = False self._context_id = None def can_generate_metrics(self) -> bool: @@ -265,7 +263,6 @@ class AsyncAITTSService(AudioContextTTSService): finally: self._websocket = None self._context_id = None - self._started = False await self._call_event_handler("on_disconnected") def _get_websocket(self): @@ -289,8 +286,6 @@ class AsyncAITTSService(AudioContextTTSService): direction: The direction to push the frame. """ await super().push_frame(frame, direction) - if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)): - self._started = False async def _receive_messages(self): async for message in self._get_websocket(): @@ -323,7 +318,7 @@ class AsyncAITTSService(AudioContextTTSService): if msg.get("audio"): await self.stop_ttfb_metrics() audio = base64.b64decode(msg["audio"]) - frame = TTSAudioRawFrame(audio, self.sample_rate, 1) + frame = TTSAudioRawFrame(audio, self.sample_rate, 1, context_id=received_ctx_id) await self.append_to_audio_context(received_ctx_id, frame) async def _keepalive_task_handler(self): @@ -365,14 +360,14 @@ class AsyncAITTSService(AudioContextTTSService): except Exception as e: logger.error(f"Error closing context on interruption: {e}") self._context_id = None - self._started = False @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Async API websocket endpoint. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech. @@ -384,28 +379,21 @@ class AsyncAITTSService(AudioContextTTSService): await self._connect() try: - if not self._started: - await self.start_ttfb_metrics() - yield TTSStartedFrame() - self._started = True + await self.start_ttfb_metrics() + yield TTSStartedFrame(context_id=context_id) - if not self._context_id: - self._context_id = str(uuid.uuid4()) - if not self.audio_context_available(self._context_id): - await self.create_audio_context(self._context_id) + if not self._context_id: + self._context_id = context_id + if not self.audio_context_available(self._context_id): + await self.create_audio_context(self._context_id) - msg = self._build_msg(text=text, force=True, context_id=self._context_id) - await self._get_websocket().send(msg) - await self.start_tts_usage_metrics(text) - else: - if self._websocket and self._context_id: - msg = self._build_msg(text=text, force=True, context_id=self._context_id) - await self._get_websocket().send(msg) + msg = self._build_msg(text=text, force=True, context_id=self._context_id) + await self._get_websocket().send(msg) + await self.start_tts_usage_metrics(text) except Exception as e: yield ErrorFrame(error=f"Unknown error occurred: {e}") - yield TTSStoppedFrame() - self._started = False + yield TTSStoppedFrame(context_id=context_id) return yield None except Exception as e: @@ -510,11 +498,12 @@ class AsyncAIHttpTTSService(TTSService): self._settings["output_format"]["sample_rate"] = self.sample_rate @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Async's HTTP streaming API. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech. @@ -531,7 +520,7 @@ class AsyncAIHttpTTSService(TTSService): "output_format": self._settings["output_format"], "language": self._settings["language"], } - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) headers = { "version": self._api_version, "x-api-key": self._api_key, @@ -560,6 +549,7 @@ class AsyncAIHttpTTSService(TTSService): audio=audio_data, sample_rate=self.sample_rate, num_channels=1, + context_id=context_id, ) yield frame @@ -568,4 +558,4 @@ class AsyncAIHttpTTSService(TTSService): await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: await self.stop_ttfb_metrics() - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) diff --git a/src/pipecat/services/aws/tts.py b/src/pipecat/services/aws/tts.py index 358a894c7..b902564d2 100644 --- a/src/pipecat/services/aws/tts.py +++ b/src/pipecat/services/aws/tts.py @@ -253,11 +253,12 @@ class AWSPollyTTSService(TTSService): return ssml @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using AWS Polly. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech. @@ -298,7 +299,7 @@ class AWSPollyTTSService(TTSService): await self.start_tts_usage_metrics(text) - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) CHUNK_SIZE = self.chunk_size @@ -306,16 +307,16 @@ class AWSPollyTTSService(TTSService): chunk = audio_data[i : i + CHUNK_SIZE] if len(chunk) > 0: await self.stop_ttfb_metrics() - frame = TTSAudioRawFrame(chunk, self.sample_rate, 1) + frame = TTSAudioRawFrame(chunk, self.sample_rate, 1, context_id=context_id) yield frame - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) except (BotoCoreError, ClientError) as error: error_message = f"AWS Polly TTS error: {str(error)}" yield ErrorFrame(error=error_message) finally: - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) class PollyTTSService(AWSPollyTTSService): diff --git a/src/pipecat/services/azure/tts.py b/src/pipecat/services/azure/tts.py index dc0c56e7b..7d4aa0253 100644 --- a/src/pipecat/services/azure/tts.py +++ b/src/pipecat/services/azure/tts.py @@ -277,7 +277,6 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService): self._audio_queue = asyncio.Queue() self._word_boundary_queue = asyncio.Queue() self._word_processor_task = None - self._started = False self._first_chunk = True self._cumulative_audio_offset: float = 0.0 # Cumulative audio duration in seconds self._current_sentence_base_offset: float = 0.0 # Base offset for current sentence @@ -287,6 +286,9 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService): ) self._last_word: Optional[str] = None # Track last word for punctuation merging self._last_timestamp: Optional[float] = None # Track last timestamp + self._current_context_id: Optional[str] = ( + None # Track current context_id for word timestamps + ) def can_generate_metrics(self) -> bool: """Check if this service can generate processing metrics. @@ -478,7 +480,10 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService): while True: try: word, timestamp_seconds = await self._word_boundary_queue.get() - await self.add_word_timestamps([(word, timestamp_seconds)]) + if self._current_context_id: + await self.add_word_timestamps( + [(word, timestamp_seconds)], self._current_context_id + ) self._word_boundary_queue.task_done() except asyncio.CancelledError: break @@ -536,12 +541,11 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService): await super().push_frame(frame, direction) if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)): self._reset_state() - if isinstance(frame, TTSStoppedFrame): - await self.add_word_timestamps([("Reset", 0)]) + if isinstance(frame, TTSStoppedFrame) and self._current_context_id: + await self.add_word_timestamps([("Reset", 0)], self._current_context_id) def _reset_state(self): """Reset TTS state between turns.""" - self._started = False self._first_chunk = True self._cumulative_audio_offset = 0.0 self._current_sentence_base_offset = 0.0 @@ -549,6 +553,7 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService): self._current_sentence_max_word_offset = 0.0 self._last_word = None self._last_timestamp = None + self._current_context_id = None async def flush_audio(self): """Flush any pending audio data.""" @@ -592,11 +597,12 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService): break @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Azure's streaming synthesis. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing synthesized speech data. @@ -615,11 +621,10 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService): return try: - if not self._started: - await self.start_ttfb_metrics() - yield TTSStartedFrame() - self._started = True - self._first_chunk = True + await self.start_ttfb_metrics() + yield TTSStartedFrame(context_id=context_id) + self._first_chunk = True + self._current_context_id = context_id # Capture base offset BEFORE starting synthesis to avoid race conditions # Word boundary callbacks will use this value @@ -647,6 +652,7 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService): audio=chunk, sample_rate=self.sample_rate, num_channels=1, + context_id=context_id, ) yield frame @@ -662,7 +668,7 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService): except Exception as e: yield ErrorFrame(error=f"Unknown error occurred: {e}") - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) self._reset_state() return @@ -738,11 +744,12 @@ class AzureHttpTTSService(TTSService, AzureBaseTTSService): ) @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Azure's HTTP synthesis API. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the complete synthesized speech. @@ -758,14 +765,15 @@ class AzureHttpTTSService(TTSService, AzureBaseTTSService): if result.reason == ResultReason.SynthesizingAudioCompleted: await self.start_tts_usage_metrics(text) await self.stop_ttfb_metrics() - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) # Azure always sends a 44-byte header. Strip it off. yield TTSAudioRawFrame( audio=result.audio_data[44:], sample_rate=self.sample_rate, num_channels=1, + context_id=context_id, ) - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) elif result.reason == ResultReason.Canceled: cancellation_details = result.cancellation_details logger.warning(f"Speech synthesis canceled: {cancellation_details.reason}") diff --git a/src/pipecat/services/camb/tts.py b/src/pipecat/services/camb/tts.py index 00b4eaf9b..def57d3a0 100644 --- a/src/pipecat/services/camb/tts.py +++ b/src/pipecat/services/camb/tts.py @@ -259,11 +259,12 @@ class CambTTSService(TTSService): self._sample_rate = MODEL_SAMPLE_RATES.get(self.model_name, 22050) @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Camb.ai's TTS API. Args: text: The text to synthesize into speech (max 3000 characters). + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech. @@ -292,7 +293,7 @@ class CambTTSService(TTSService): tts_kwargs["user_instructions"] = self._settings["user_instructions"] await self.start_tts_usage_metrics(text) - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) assert self._client is not None, "Camb.ai TTS service not initialized" @@ -312,6 +313,7 @@ class CambTTSService(TTSService): audio=aligned_audio, sample_rate=self.sample_rate, num_channels=1, + context_id=context_id, ) # Yield any remaining complete samples @@ -322,9 +324,10 @@ class CambTTSService(TTSService): audio=aligned_audio, sample_rate=self.sample_rate, num_channels=1, + context_id=context_id, ) except Exception as e: yield ErrorFrame(error=f"Camb.ai TTS error: {e}") finally: - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) diff --git a/src/pipecat/services/cartesia/tts.py b/src/pipecat/services/cartesia/tts.py index 6bfb8703d..791c60a18 100644 --- a/src/pipecat/services/cartesia/tts.py +++ b/src/pipecat/services/cartesia/tts.py @@ -8,7 +8,6 @@ import base64 import json -import uuid import warnings from enum import Enum from typing import AsyncGenerator, List, Literal, Optional @@ -554,16 +553,17 @@ class CartesiaTTSService(AudioContextWordTTSService): msg = json.loads(message) if not msg or not self.audio_context_available(msg["context_id"]): continue + ctx_id = msg["context_id"] if msg["type"] == "done": await self.stop_ttfb_metrics() - await self.add_word_timestamps([("TTSStoppedFrame", 0), ("Reset", 0)]) - await self.remove_audio_context(msg["context_id"]) + await self.add_word_timestamps([("TTSStoppedFrame", 0), ("Reset", 0)], ctx_id) + await self.remove_audio_context(ctx_id) elif msg["type"] == "timestamps": # Process the timestamps based on language before adding them processed_timestamps = self._process_word_timestamps_for_language( msg["word_timestamps"]["words"], msg["word_timestamps"]["start"] ) - await self.add_word_timestamps(processed_timestamps) + await self.add_word_timestamps(processed_timestamps, ctx_id) elif msg["type"] == "chunk": await self.stop_ttfb_metrics() await self.start_word_timestamps() @@ -571,10 +571,11 @@ class CartesiaTTSService(AudioContextWordTTSService): audio=base64.b64decode(msg["data"]), sample_rate=self.sample_rate, num_channels=1, + context_id=ctx_id, ) - await self.append_to_audio_context(msg["context_id"], frame) + await self.append_to_audio_context(ctx_id, frame) elif msg["type"] == "error": - await self.push_frame(TTSStoppedFrame()) + await self.push_frame(TTSStoppedFrame(context_id=ctx_id)) await self.stop_all_metrics() await self.push_error(error_msg=f"Error: {msg}") self._context_id = None @@ -590,11 +591,12 @@ class CartesiaTTSService(AudioContextWordTTSService): await self._connect_websocket() @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Cartesia's streaming API. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech. @@ -607,8 +609,8 @@ class CartesiaTTSService(AudioContextWordTTSService): if not self._context_id: await self.start_ttfb_metrics() - yield TTSStartedFrame() - self._context_id = str(uuid.uuid4()) + yield TTSStartedFrame(context_id=context_id) + self._context_id = context_id await self.create_audio_context(self._context_id) msg = self._build_msg(text=text) @@ -618,7 +620,7 @@ class CartesiaTTSService(AudioContextWordTTSService): await self.start_tts_usage_metrics(text) except Exception as e: yield ErrorFrame(error=f"Unknown error occurred: {e}") - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) await self._disconnect() await self._connect() return @@ -761,11 +763,12 @@ class CartesiaHttpTTSService(TTSService): await self._client.close() @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Cartesia's HTTP API. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech. @@ -808,7 +811,7 @@ class CartesiaHttpTTSService(TTSService): if self._settings["pronunciation_dict_id"]: payload["pronunciation_dict_id"] = self._settings["pronunciation_dict_id"] - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) session = await self._client._get_session() @@ -834,6 +837,7 @@ class CartesiaHttpTTSService(TTSService): audio=audio_data, sample_rate=self.sample_rate, num_channels=1, + context_id=context_id, ) yield frame @@ -842,4 +846,4 @@ class CartesiaHttpTTSService(TTSService): yield ErrorFrame(error=f"Unknown error occurred: {e}") finally: await self.stop_ttfb_metrics() - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) diff --git a/src/pipecat/services/deepgram/tts.py b/src/pipecat/services/deepgram/tts.py index a53ec56d2..12aba4905 100644 --- a/src/pipecat/services/deepgram/tts.py +++ b/src/pipecat/services/deepgram/tts.py @@ -97,6 +97,7 @@ class DeepgramTTSService(WebsocketTTSService): self.set_voice(voice) self._receive_task = None + self._context_id: Optional[str] = None def can_generate_metrics(self) -> bool: """Check if the service can generate metrics. @@ -211,6 +212,7 @@ class DeepgramTTSService(WebsocketTTSService): logger.error(f"{self} exception: {e}") await self.push_error(ErrorFrame(error=f"{self} error: {e}")) finally: + self._context_id = None self._websocket = None await self._call_event_handler("on_disconnected") @@ -242,7 +244,7 @@ class DeepgramTTSService(WebsocketTTSService): if isinstance(message, bytes): # Binary message contains audio data await self.stop_ttfb_metrics() - frame = TTSAudioRawFrame(message, self.sample_rate, 1) + frame = TTSAudioRawFrame(message, self.sample_rate, 1, context_id=self._context_id) await self.push_frame(frame) elif isinstance(message, str): # Text message contains metadata or control messages @@ -283,11 +285,12 @@ class DeepgramTTSService(WebsocketTTSService): logger.error(f"{self} error sending Flush message: {e}") @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Deepgram's WebSocket TTS API. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech, plus start/stop frames. @@ -302,7 +305,9 @@ class DeepgramTTSService(WebsocketTTSService): await self.start_ttfb_metrics() await self.start_tts_usage_metrics(text) - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) + # Store context_id for use in _receive_messages + self._context_id = context_id # Send text message to Deepgram # Note: We don't send Flush here - that should only be sent when the @@ -366,11 +371,12 @@ class DeepgramHttpTTSService(TTSService): return True @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Deepgram's TTS API. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech, plus start/stop frames. @@ -404,7 +410,7 @@ class DeepgramHttpTTSService(TTSService): raise Exception(f"HTTP {response.status}: {error_text}") await self.start_tts_usage_metrics(text) - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) CHUNK_SIZE = self.chunk_size @@ -419,9 +425,10 @@ class DeepgramHttpTTSService(TTSService): audio=chunk, sample_rate=self.sample_rate, num_channels=1, + context_id=context_id, ) - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) except Exception as e: yield ErrorFrame(f"Error getting audio: {str(e)}") diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py index 02ccd0ab3..4dab0c01a 100644 --- a/src/pipecat/services/elevenlabs/tts.py +++ b/src/pipecat/services/elevenlabs/tts.py @@ -13,7 +13,6 @@ with support for streaming audio, word timestamps, and voice customization. import asyncio import base64 import json -import uuid from typing import Any, AsyncGenerator, Dict, List, Literal, Mapping, Optional, Tuple, Union import aiohttp @@ -337,9 +336,6 @@ class ElevenLabsTTSService(AudioContextWordTTSService): self._voice_settings = self._set_voice_settings() self._pronunciation_dictionary_locators = params.pronunciation_dictionary_locators - # Indicates if we have sent TTSStartedFrame. It will reset to False when - # there's an interruption or TTSStoppedFrame. - self._started = False self._cumulative_time = 0 # Track partial words that span across alignment chunks self._partial_word = "" @@ -426,7 +422,6 @@ class ElevenLabsTTSService(AudioContextWordTTSService): except Exception as e: await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) self._context_id = None - self._started = False async def start(self, frame: StartFrame): """Start the ElevenLabs TTS service. @@ -473,9 +468,8 @@ class ElevenLabsTTSService(AudioContextWordTTSService): """ await super().push_frame(frame, direction) if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)): - self._started = False if isinstance(frame, TTSStoppedFrame): - await self.add_word_timestamps([("Reset", 0)]) + await self.add_word_timestamps([("Reset", 0)], self._context_id) async def _connect(self): await super()._connect() @@ -557,7 +551,6 @@ class ElevenLabsTTSService(AudioContextWordTTSService): except Exception as e: await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: - self._started = False self._context_id = None self._websocket = None await self._call_event_handler("on_disconnected") @@ -587,7 +580,6 @@ class ElevenLabsTTSService(AudioContextWordTTSService): except Exception as e: await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) self._context_id = None - self._started = False self._partial_word = "" self._partial_word_start_time = 0.0 @@ -624,7 +616,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService): await self.start_word_timestamps() audio = base64.b64decode(msg["audio"]) - frame = TTSAudioRawFrame(audio, self.sample_rate, 1) + frame = TTSAudioRawFrame(audio, self.sample_rate, 1, context_id=received_ctx_id) await self.append_to_audio_context(received_ctx_id, frame) if msg.get("alignment"): @@ -639,7 +631,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService): ) if word_times: - await self.add_word_timestamps(word_times) + await self.add_word_timestamps(word_times, received_ctx_id) # Calculate the actual end time of this audio chunk char_start_times_ms = alignment.get("charStartTimesMs", []) @@ -689,11 +681,12 @@ class ElevenLabsTTSService(AudioContextWordTTSService): await self._websocket.send(json.dumps(msg)) @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using ElevenLabs' streaming WebSocket API. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech. @@ -705,41 +698,37 @@ class ElevenLabsTTSService(AudioContextWordTTSService): await self._connect() try: - if not self._started: - await self.start_ttfb_metrics() - yield TTSStartedFrame() - self._started = True - self._cumulative_time = 0 - self._partial_word = "" - self._partial_word_start_time = 0.0 - # If a context ID does not exist, create a new one and - # register it. If an ID exists, that means the Pipeline - # doesn't allow user interruptions, so continue using the - # current ID. When interruptions are allowed, user speech - # results in an interruption, which resets the context ID. - if not self._context_id: - self._context_id = str(uuid.uuid4()) - if not self.audio_context_available(self._context_id): - await self.create_audio_context(self._context_id) + await self.start_ttfb_metrics() + yield TTSStartedFrame(context_id=context_id) + self._cumulative_time = 0 + self._partial_word = "" + self._partial_word_start_time = 0.0 + # If a context ID does not exist, use the provided one. + # If an ID exists, that means the Pipeline doesn't allow + # user interruptions, so continue using the current ID. + # When interruptions are allowed, user speech results in + # an interruption, which resets the context ID. + if not self._context_id: + self._context_id = context_id + if not self.audio_context_available(self._context_id): + await self.create_audio_context(self._context_id) - # Initialize context with voice settings and pronunciation dictionaries - msg = {"text": " ", "context_id": self._context_id} - if self._voice_settings: - msg["voice_settings"] = self._voice_settings - if self._pronunciation_dictionary_locators: - msg["pronunciation_dictionary_locators"] = [ - locator.model_dump() - for locator in self._pronunciation_dictionary_locators - ] - await self._websocket.send(json.dumps(msg)) - logger.trace(f"Created new context {self._context_id}") + # Initialize context with voice settings and pronunciation dictionaries + msg = {"text": " ", "context_id": self._context_id} + if self._voice_settings: + msg["voice_settings"] = self._voice_settings + if self._pronunciation_dictionary_locators: + msg["pronunciation_dictionary_locators"] = [ + locator.model_dump() for locator in self._pronunciation_dictionary_locators + ] + await self._websocket.send(json.dumps(msg)) + logger.trace(f"Created new context {self._context_id}") await self._send_text(text) await self.start_tts_usage_metrics(text) except Exception as e: - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) yield ErrorFrame(error=f"Unknown error occurred: {e}") - self._started = False return yield None except Exception as e: @@ -840,7 +829,6 @@ class ElevenLabsHttpTTSService(WordTTSService): # Track cumulative time to properly sequence word timestamps across utterances self._cumulative_time = 0 - self._started = False # Store previous text for context within a turn self._previous_text = "" @@ -879,7 +867,6 @@ class ElevenLabsHttpTTSService(WordTTSService): def _reset_state(self): """Reset internal state variables.""" self._cumulative_time = 0 - self._started = False self._previous_text = "" self._partial_word = "" self._partial_word_start_time = 0.0 @@ -976,7 +963,7 @@ class ElevenLabsHttpTTSService(WordTTSService): return word_times @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using ElevenLabs streaming API with timestamps. Makes a request to the ElevenLabs API to generate audio and timing data. @@ -985,6 +972,7 @@ class ElevenLabsHttpTTSService(WordTTSService): Args: text: Text to convert to speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio and control frames containing the synthesized speech. @@ -1048,11 +1036,9 @@ class ElevenLabsHttpTTSService(WordTTSService): await self.start_tts_usage_metrics(text) - # Start TTS sequence if not already started - if not self._started: - await self.start_word_timestamps() - yield TTSStartedFrame() - self._started = True + # Start TTS sequence + await self.start_word_timestamps() + yield TTSStartedFrame(context_id=context_id) # Track the duration of this utterance based on the last character's end time utterance_duration = 0 @@ -1069,7 +1055,9 @@ class ElevenLabsHttpTTSService(WordTTSService): if data and "audio_base64" in data: await self.stop_ttfb_metrics() audio = base64.b64decode(data["audio_base64"]) - yield TTSAudioRawFrame(audio, self.sample_rate, 1) + yield TTSAudioRawFrame( + audio, self.sample_rate, 1, context_id=context_id + ) # Process alignment if present if data and "alignment" in data: @@ -1085,7 +1073,7 @@ class ElevenLabsHttpTTSService(WordTTSService): # Calculate word timestamps word_times = self.calculate_word_times(alignment) if word_times: - await self.add_word_timestamps(word_times) + await self.add_word_timestamps(word_times, context_id) except json.JSONDecodeError as e: logger.warning(f"Failed to parse JSON from stream: {e}") continue @@ -1097,7 +1085,7 @@ class ElevenLabsHttpTTSService(WordTTSService): # since this is the end of the utterance if self._partial_word: final_word_time = [(self._partial_word, self._partial_word_start_time)] - await self.add_word_timestamps(final_word_time) + await self.add_word_timestamps(final_word_time, context_id) self._partial_word = "" self._partial_word_start_time = 0.0 diff --git a/src/pipecat/services/fish/tts.py b/src/pipecat/services/fish/tts.py index 357b82346..93a718429 100644 --- a/src/pipecat/services/fish/tts.py +++ b/src/pipecat/services/fish/tts.py @@ -135,7 +135,6 @@ class FishAudioTTSService(InterruptibleTTSService): self._websocket = None self._receive_task = None self._request_id = None - self._started = False self._settings = { "sample_rate": 0, @@ -249,7 +248,6 @@ class FishAudioTTSService(InterruptibleTTSService): await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: self._request_id = None - self._started = False self._websocket = None await self._call_event_handler("on_disconnected") @@ -291,11 +289,12 @@ class FishAudioTTSService(InterruptibleTTSService): await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Fish Audio's streaming API. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames and control frames for the synthesized speech. @@ -308,7 +307,7 @@ class FishAudioTTSService(InterruptibleTTSService): if not self._request_id: await self.start_ttfb_metrics() await self.start_tts_usage_metrics(text) - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) self._request_id = str(uuid.uuid4()) # Send the text @@ -325,7 +324,7 @@ class FishAudioTTSService(InterruptibleTTSService): await self._get_websocket().send(ormsgpack.packb(flush_message)) except Exception as e: yield ErrorFrame(error=f"Unknown error occurred: {e}") - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) await self._disconnect() await self._connect() diff --git a/src/pipecat/services/google/tts.py b/src/pipecat/services/google/tts.py index 2c22fb9c5..4016286df 100644 --- a/src/pipecat/services/google/tts.py +++ b/src/pipecat/services/google/tts.py @@ -682,11 +682,12 @@ class GoogleHttpTTSService(TTSService): return ssml @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Google's HTTP TTS API. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech. @@ -731,7 +732,7 @@ class GoogleHttpTTSService(TTSService): await self.start_tts_usage_metrics(text) - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) # Skip the first 44 bytes to remove the WAV header audio_content = response.audio_content[44:] @@ -743,10 +744,10 @@ class GoogleHttpTTSService(TTSService): if not chunk: break await self.stop_ttfb_metrics() - frame = TTSAudioRawFrame(chunk, self.sample_rate, 1) + frame = TTSAudioRawFrame(chunk, self.sample_rate, 1, context_id=context_id) yield frame - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) except Exception as e: error_message = f"TTS generation error: {str(e)}" @@ -828,6 +829,7 @@ class GoogleBaseTTSService(TTSService): self, streaming_config: texttospeech_v1.StreamingSynthesizeConfig, text: str, + context_id: str, prompt: Optional[str] = None, ) -> AsyncGenerator[Frame, None]: """Shared streaming synthesis logic. @@ -835,6 +837,7 @@ class GoogleBaseTTSService(TTSService): Args: streaming_config: The streaming configuration. text: The text to synthesize. + context_id: Unique identifier for this TTS context. prompt: Optional prompt for style instructions (Gemini only). Yields: @@ -856,7 +859,7 @@ class GoogleBaseTTSService(TTSService): streaming_responses = await self._client.streaming_synthesize(request_generator()) await self.start_tts_usage_metrics(text) - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) audio_buffer = b"" first_chunk_for_ttfb = False @@ -876,12 +879,12 @@ class GoogleBaseTTSService(TTSService): while len(audio_buffer) >= CHUNK_SIZE: piece = audio_buffer[:CHUNK_SIZE] audio_buffer = audio_buffer[CHUNK_SIZE:] - yield TTSAudioRawFrame(piece, self.sample_rate, 1) + yield TTSAudioRawFrame(piece, self.sample_rate, 1, context_id=context_id) if audio_buffer: - yield TTSAudioRawFrame(audio_buffer, self.sample_rate, 1) + yield TTSAudioRawFrame(audio_buffer, self.sample_rate, 1, context_id=context_id) - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) class GoogleTTSService(GoogleBaseTTSService): @@ -976,11 +979,12 @@ class GoogleTTSService(GoogleBaseTTSService): await super()._update_settings(settings) @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate streaming speech from text using Google's streaming API. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech as it's generated. @@ -1014,7 +1018,7 @@ class GoogleTTSService(GoogleBaseTTSService): ) # Use base class streaming logic - async for frame in self._stream_tts(streaming_config, text): + async for frame in self._stream_tts(streaming_config, text, context_id): yield frame except Exception as e: @@ -1213,11 +1217,12 @@ class GeminiTTSService(GoogleBaseTTSService): await super()._update_settings(settings) @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate streaming speech from text using Gemini TTS models. Args: - text: The text to synthesize into speech. Can include markup tags + text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Can include markup tags like [sigh], [laughing], [whispering] for expressive control. Yields: @@ -1267,7 +1272,9 @@ class GeminiTTSService(GoogleBaseTTSService): ) # Use base class streaming logic with prompt support - async for frame in self._stream_tts(streaming_config, text, self._settings["prompt"]): + async for frame in self._stream_tts( + streaming_config, text, context_id, self._settings["prompt"] + ): yield frame except Exception as e: diff --git a/src/pipecat/services/gradium/tts.py b/src/pipecat/services/gradium/tts.py index df33753ce..0e9865cf0 100644 --- a/src/pipecat/services/gradium/tts.py +++ b/src/pipecat/services/gradium/tts.py @@ -95,6 +95,7 @@ class GradiumTTSService(InterruptibleWordTTSService): # State tracking self._receive_task = None + self._current_context_id: Optional[str] = None def can_generate_metrics(self) -> bool: """Check if this service can generate processing metrics. @@ -261,11 +262,15 @@ class GradiumTTSService(InterruptibleWordTTSService): audio=base64.b64decode(msg["audio"]), sample_rate=self.sample_rate, num_channels=1, + context_id=self._current_context_id, ) await self.push_frame(frame) elif msg["type"] == "text": - await self.add_word_timestamps([(msg["text"], msg["start_s"])]) + if self._current_context_id: + await self.add_word_timestamps( + [(msg["text"], msg["start_s"])], self._current_context_id + ) elif msg["type"] == "end_of_stream": await self.push_frame(TTSStoppedFrame()) await self.stop_all_metrics() @@ -285,11 +290,12 @@ class GradiumTTSService(InterruptibleWordTTSService): await super().push_frame(frame, direction) @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Gradium's streaming API. Args: text: The text to convert to speech. + context_id: Unique identifier for this TTS context. Yields: Frame: Audio frames containing the synthesized speech. @@ -302,14 +308,15 @@ class GradiumTTSService(InterruptibleWordTTSService): await self._connect() try: - yield TTSStartedFrame() + self._current_context_id = context_id + yield TTSStartedFrame(context_id=context_id) msg = self._build_msg(text=text) await self._get_websocket().send(json.dumps(msg)) await self.start_tts_usage_metrics(text) except Exception as e: yield ErrorFrame(error=f"Unknown error occurred: {e}") - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) await self._disconnect() await self._connect() return diff --git a/src/pipecat/services/groq/tts.py b/src/pipecat/services/groq/tts.py index e66dc07e7..331af8eb7 100644 --- a/src/pipecat/services/groq/tts.py +++ b/src/pipecat/services/groq/tts.py @@ -112,11 +112,12 @@ class GroqTTSService(TTSService): return True @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Groq's TTS API. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech data. @@ -124,7 +125,7 @@ class GroqTTSService(TTSService): logger.debug(f"{self}: Generating TTS [{text}]") measuring_ttfb = True await self.start_ttfb_metrics() - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) try: response = await self._client.audio.speech.create( @@ -144,8 +145,8 @@ class GroqTTSService(TTSService): frame_rate = w.getframerate() num_frames = w.getnframes() bytes = w.readframes(num_frames) - yield TTSAudioRawFrame(bytes, frame_rate, channels) + yield TTSAudioRawFrame(bytes, frame_rate, channels, context_id=context_id) except Exception as e: yield ErrorFrame(error=f"Unknown error occurred: {e}") - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) diff --git a/src/pipecat/services/hathora/tts.py b/src/pipecat/services/hathora/tts.py index e59c4ad46..80cbd4fe8 100644 --- a/src/pipecat/services/hathora/tts.py +++ b/src/pipecat/services/hathora/tts.py @@ -115,11 +115,12 @@ class HathoraTTSService(TTSService): return True @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Run text-to-speech synthesis on the provided text. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech. @@ -142,7 +143,7 @@ class HathoraTTSService(TTSService): for option in self._settings["config"] ] - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) async with aiohttp.ClientSession() as session: async with session.post( @@ -161,6 +162,7 @@ class HathoraTTSService(TTSService): audio=pcm_audio, sample_rate=self.sample_rate, num_channels=num_channels, + context_id=context_id, ) yield frame @@ -170,4 +172,4 @@ class HathoraTTSService(TTSService): finally: await self.stop_ttfb_metrics() await self.stop_processing_metrics() - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) diff --git a/src/pipecat/services/hume/tts.py b/src/pipecat/services/hume/tts.py index 7f8943816..2d98e1f8c 100644 --- a/src/pipecat/services/hume/tts.py +++ b/src/pipecat/services/hume/tts.py @@ -209,11 +209,12 @@ class HumeTTSService(WordTTSService): await super().update_setting(key, value) @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Hume TTS with word timestamps. Args: text: The text to be synthesized. + context_id: Unique identifier for this TTS context. Returns: An async generator that yields `Frame` objects, including @@ -245,7 +246,7 @@ class HumeTTSService(WordTTSService): # Start TTS sequence if not already started if not self._started: await self.start_word_timestamps() - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) self._started = True try: @@ -281,6 +282,7 @@ class HumeTTSService(WordTTSService): audio=self._audio_bytes, sample_rate=self.sample_rate, num_channels=1, + context_id=context_id, ) yield frame self._audio_bytes = b"" @@ -297,7 +299,9 @@ class HumeTTSService(WordTTSService): utterance_duration = max(utterance_duration, word_end_time) # Add word timestamp - await self.add_word_timestamps([(timestamp.text, word_start_time)]) + await self.add_word_timestamps( + [(timestamp.text, word_start_time)], context_id + ) # Flush any remaining audio bytes if self._audio_bytes: @@ -305,6 +309,7 @@ class HumeTTSService(WordTTSService): audio=self._audio_bytes, sample_rate=self.sample_rate, num_channels=1, + context_id=context_id, ) yield frame diff --git a/src/pipecat/services/inworld/tts.py b/src/pipecat/services/inworld/tts.py index d8798b75e..2ea94399b 100644 --- a/src/pipecat/services/inworld/tts.py +++ b/src/pipecat/services/inworld/tts.py @@ -16,7 +16,6 @@ Inworld’s text-to-speech (TTS) models offer ultra-realistic, context-aware spe import asyncio import base64 import json -import uuid from typing import Any, AsyncGenerator, Dict, List, Optional, Tuple import aiohttp @@ -125,7 +124,6 @@ class InworldHttpTTSService(WordTTSService): if params.speaking_rate is not None: self._settings["audioConfig"]["speakingRate"] = params.speaking_rate - self._started = False self._cumulative_time = 0.0 self.set_voice(voice_id) @@ -173,7 +171,6 @@ class InworldHttpTTSService(WordTTSService): """ await super().push_frame(frame, direction) if isinstance(frame, (InterruptionFrame, TTSStoppedFrame)): - self._started = False self._cumulative_time = 0.0 if isinstance(frame, TTSStoppedFrame): await self.add_word_timestamps([("Reset", 0)]) @@ -214,11 +211,12 @@ class InworldHttpTTSService(WordTTSService): return (word_times, chunk_end_time) @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate TTS audio for the given text. Args: text: The text to generate TTS audio for. + context_id: Unique identifier for this TTS context. Returns: An asynchronous generator of frames. @@ -246,10 +244,8 @@ class InworldHttpTTSService(WordTTSService): try: await self.start_ttfb_metrics() - if not self._started: - await self.start_word_timestamps() - yield TTSStartedFrame() - self._started = True + await self.start_word_timestamps() + yield TTSStartedFrame(context_id=context_id) async with self._session.post( self._base_url, json=payload, headers=headers @@ -261,10 +257,10 @@ class InworldHttpTTSService(WordTTSService): return if self._streaming: - async for frame in self._process_streaming_response(response): + async for frame in self._process_streaming_response(response, context_id): yield frame else: - async for frame in self._process_non_streaming_response(response): + async for frame in self._process_non_streaming_response(response, context_id): yield frame await self.start_tts_usage_metrics(text) @@ -276,12 +272,13 @@ class InworldHttpTTSService(WordTTSService): await self.stop_all_metrics() async def _process_streaming_response( - self, response: aiohttp.ClientResponse + self, response: aiohttp.ClientResponse, context_id: str ) -> AsyncGenerator[Frame, None]: """Process a streaming response from the Inworld API. Args: response: The response from the Inworld API. + context_id: Unique identifier for this TTS context. Yields: An asynchronous generator of frames. @@ -306,7 +303,7 @@ class InworldHttpTTSService(WordTTSService): if "result" in chunk_data and "audioContent" in chunk_data["result"]: await self.stop_ttfb_metrics() async for frame in self._process_audio_chunk( - base64.b64decode(chunk_data["result"]["audioContent"]) + base64.b64decode(chunk_data["result"]["audioContent"]), context_id ): yield frame @@ -314,7 +311,7 @@ class InworldHttpTTSService(WordTTSService): timestamp_info = chunk_data["result"]["timestampInfo"] word_times, chunk_end_time = self._calculate_word_times(timestamp_info) if word_times: - await self.add_word_timestamps(word_times) + await self.add_word_timestamps(word_times, context_id) # Track the maximum end time across all chunks utterance_duration = max(utterance_duration, chunk_end_time) @@ -327,12 +324,13 @@ class InworldHttpTTSService(WordTTSService): self._cumulative_time += utterance_duration async def _process_non_streaming_response( - self, response: aiohttp.ClientResponse + self, response: aiohttp.ClientResponse, context_id: str ) -> AsyncGenerator[Frame, None]: """Process a non-streaming response from the Inworld API. Args: response: The response from the Inworld API. + context_id: Unique identifier for this TTS context. Returns: An asynchronous generator of frames. @@ -349,7 +347,7 @@ class InworldHttpTTSService(WordTTSService): timestamp_info = response_data["timestampInfo"] word_times, chunk_end_time = self._calculate_word_times(timestamp_info) if word_times: - await self.add_word_timestamps(word_times) + await self.add_word_timestamps(word_times, context_id) utterance_duration = chunk_end_time audio_data = base64.b64decode(response_data["audioContent"]) @@ -363,20 +361,21 @@ class InworldHttpTTSService(WordTTSService): if chunk: await self.stop_ttfb_metrics() yield TTSAudioRawFrame( - audio=chunk, - sample_rate=self.sample_rate, - num_channels=1, + audio=chunk, sample_rate=self.sample_rate, num_channels=1, context_id=context_id ) # After processing all audio, add the utterance duration to cumulative time if utterance_duration > 0: self._cumulative_time += utterance_duration - async def _process_audio_chunk(self, audio_chunk: bytes) -> AsyncGenerator[Frame, None]: + async def _process_audio_chunk( + self, audio_chunk: bytes, context_id: str + ) -> AsyncGenerator[Frame, None]: """Process an audio chunk from the Inworld API. Args: audio_chunk: The audio chunk to process. + context_id: Unique identifier for this TTS context. Returns: An asynchronous generator of frames. @@ -394,6 +393,7 @@ class InworldHttpTTSService(WordTTSService): audio=audio_data, sample_rate=self.sample_rate, num_channels=1, + context_id=context_id, ) @@ -500,7 +500,6 @@ class InworldTTSService(AudioContextWordTTSService): self._receive_task = None self._keepalive_task = None self._context_id = None - self._started = False # Track cumulative time across generations for monotonic timestamps within a turn. # When auto_mode is enabled, the server controls generations and timestamps reset @@ -569,7 +568,6 @@ class InworldTTSService(AudioContextWordTTSService): """ await super().push_frame(frame, direction) if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)): - self._started = False logger.trace( f"{self}: Resetting timestamp tracking due to {type(frame).__name__} - " f"cumulative_time was {self._cumulative_time}" @@ -637,7 +635,6 @@ class InworldTTSService(AudioContextWordTTSService): await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) self._context_id = None - self._started = False self._cumulative_time = 0.0 self._generation_end_time = 0.0 logger.trace(f"{self}: Interruption handled, context reset to None") @@ -726,7 +723,6 @@ class InworldTTSService(AudioContextWordTTSService): except Exception as e: await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: - self._started = False self._context_id = None self._websocket = None self._cumulative_time = 0.0 @@ -801,7 +797,7 @@ class InworldTTSService(AudioContextWordTTSService): audio = base64.b64decode(audio_b64) if len(audio) > 44 and audio.startswith(b"RIFF"): audio = audio[44:] - frame = TTSAudioRawFrame(audio, self.sample_rate, 1) + frame = TTSAudioRawFrame(audio, self.sample_rate, 1, context_id=ctx_id) if ctx_id: await self.append_to_audio_context(ctx_id, frame) @@ -811,7 +807,7 @@ class InworldTTSService(AudioContextWordTTSService): if timestamp_info: word_times = self._calculate_word_times(timestamp_info) if word_times: - await self.add_word_timestamps(word_times) + await self.add_word_timestamps(word_times, ctx_id) # Handle context created confirmation if "contextCreated" in result: @@ -832,10 +828,9 @@ class InworldTTSService(AudioContextWordTTSService): # Only reset if this is our current context if ctx_id == self._context_id: self._context_id = None - self._started = False if ctx_id and self.audio_context_available(ctx_id): await self.remove_audio_context(ctx_id) - await self.add_word_timestamps([("TTSStoppedFrame", 0), ("Reset", 0)]) + await self.add_word_timestamps([("TTSStoppedFrame", 0), ("Reset", 0)], ctx_id) async def _keepalive_task_handler(self): """Send periodic keepalive messages to maintain WebSocket connection.""" @@ -917,11 +912,12 @@ class InworldTTSService(AudioContextWordTTSService): await self.send_with_retry(json.dumps(msg), self._report_error) @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate TTS audio for the given text using the Inworld WebSocket TTS service. Args: text: The text to generate TTS audio for. + context_id: Unique identifier for this TTS context. Returns: An asynchronous generator of frames. @@ -933,28 +929,25 @@ class InworldTTSService(AudioContextWordTTSService): await self._connect() try: - if not self._started: - await self.start_ttfb_metrics() - yield TTSStartedFrame() - self._started = True + await self.start_ttfb_metrics() + yield TTSStartedFrame(context_id=context_id) - if not self._context_id: - self._context_id = str(uuid.uuid4()) - logger.trace(f"{self}: Creating new context {self._context_id}") - await self.create_audio_context(self._context_id) - await self._send_context(self._context_id) - elif not self.audio_context_available(self._context_id): - # Context exists on server but local tracking was removed - logger.trace(f"{self}: Recreating local audio context {self._context_id}") - await self.create_audio_context(self._context_id) + if not self._context_id: + self._context_id = context_id + logger.trace(f"{self}: Creating new context {self._context_id}") + await self.create_audio_context(self._context_id) + await self._send_context(self._context_id) + elif not self.audio_context_available(self._context_id): + # Context exists on server but local tracking was removed + logger.trace(f"{self}: Recreating local audio context {self._context_id}") + await self.create_audio_context(self._context_id) await self._send_text(self._context_id, text) await self.start_tts_usage_metrics(text) except Exception as e: yield ErrorFrame(error=f"Unknown error occurred: {e}") - yield TTSStoppedFrame() - self._started = False + yield TTSStoppedFrame(context_id=context_id) return yield None except Exception as e: diff --git a/src/pipecat/services/kokoro/tts.py b/src/pipecat/services/kokoro/tts.py index b4d309a3f..49ede2409 100644 --- a/src/pipecat/services/kokoro/tts.py +++ b/src/pipecat/services/kokoro/tts.py @@ -143,13 +143,14 @@ class KokoroTTSService(TTSService): return True @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Synthesize speech from text using kokoro-onnx. Uses the async streaming API to generate audio frames. Args: text: The text to synthesize. + context_id: Unique identifier for this TTS context. """ logger.debug(f"{self}: Generating TTS [{text}]") @@ -157,7 +158,7 @@ class KokoroTTSService(TTSService): try: await self.start_ttfb_metrics() await self.start_tts_usage_metrics(text) - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) stream = self._kokoro.create_stream( text, voice=self._voice_id, lang=self._lang_code, speed=1.0 @@ -172,10 +173,13 @@ class KokoroTTSService(TTSService): ) yield TTSAudioRawFrame( - audio=audio_data, sample_rate=self.sample_rate, num_channels=1 + audio=audio_data, + sample_rate=self.sample_rate, + num_channels=1, + context_id=context_id, ) except Exception as e: yield ErrorFrame(error=f"Unknown error occurred: {e}") finally: await self.stop_ttfb_metrics() - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) diff --git a/src/pipecat/services/lmnt/tts.py b/src/pipecat/services/lmnt/tts.py index 911d13923..4c34e28d5 100644 --- a/src/pipecat/services/lmnt/tts.py +++ b/src/pipecat/services/lmnt/tts.py @@ -113,8 +113,8 @@ class LmntTTSService(InterruptibleTTSService): "language": self.language_to_service_language(language), "format": "raw", # Use raw format for direct PCM data } - self._started = False self._receive_task = None + self._context_id: Optional[str] = None def can_generate_metrics(self) -> bool: """Check if this service can generate processing metrics. @@ -170,8 +170,6 @@ class LmntTTSService(InterruptibleTTSService): direction: The direction to push the frame. """ await super().push_frame(frame, direction) - if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)): - self._started = False async def _connect(self): """Connect to LMNT WebSocket and start receive task.""" @@ -236,7 +234,7 @@ class LmntTTSService(InterruptibleTTSService): except Exception as e: await self.push_error(error_msg=f"Error disconnecting from LMNT: {e}", exception=e) finally: - self._started = False + self._context_id = None self._websocket = None await self._call_event_handler("on_disconnected") @@ -262,6 +260,7 @@ class LmntTTSService(InterruptibleTTSService): audio=message, sample_rate=self.sample_rate, num_channels=1, + context_id=self._context_id, ) await self.push_frame(frame) else: @@ -276,11 +275,12 @@ class LmntTTSService(InterruptibleTTSService): logger.error(f"Invalid JSON message: {message}") @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate TTS audio from text using LMNT's streaming API. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech. @@ -292,10 +292,10 @@ class LmntTTSService(InterruptibleTTSService): await self._connect() try: - if not self._started: - await self.start_ttfb_metrics() - yield TTSStartedFrame() - self._started = True + await self.start_ttfb_metrics() + # Store context_id for use in _receive_messages + self._context_id = context_id + yield TTSStartedFrame(context_id=context_id) # Send text to LMNT await self._get_websocket().send(json.dumps({"text": text})) @@ -304,7 +304,7 @@ class LmntTTSService(InterruptibleTTSService): await self.start_tts_usage_metrics(text) except Exception as e: yield ErrorFrame(error=f"Unknown error occurred: {e}") - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) await self._disconnect() await self._connect() return diff --git a/src/pipecat/services/minimax/tts.py b/src/pipecat/services/minimax/tts.py index c3e434361..7284d9630 100644 --- a/src/pipecat/services/minimax/tts.py +++ b/src/pipecat/services/minimax/tts.py @@ -284,11 +284,12 @@ class MiniMaxHttpTTSService(TTSService): logger.debug(f"MiniMax TTS initialized with sample_rate: {self.sample_rate}") @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate TTS audio from text using MiniMax's streaming API. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech. @@ -318,7 +319,7 @@ class MiniMaxHttpTTSService(TTSService): return await self.start_tts_usage_metrics(text) - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) # Process the streaming response buffer = bytearray() @@ -377,6 +378,7 @@ class MiniMaxHttpTTSService(TTSService): audio=audio_chunk, sample_rate=self.sample_rate, num_channels=1, + context_id=context_id, ) except ValueError as e: logger.error( @@ -394,4 +396,4 @@ class MiniMaxHttpTTSService(TTSService): yield ErrorFrame(error=f"Unknown error occurred: {e}", exception=e) finally: await self.stop_ttfb_metrics() - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) diff --git a/src/pipecat/services/neuphonic/tts.py b/src/pipecat/services/neuphonic/tts.py index 2666c0cfc..24eb05bd3 100644 --- a/src/pipecat/services/neuphonic/tts.py +++ b/src/pipecat/services/neuphonic/tts.py @@ -135,13 +135,11 @@ class NeuphonicTTSService(InterruptibleTTSService): } self.set_voice(voice_id) - # Indicates if we have sent TTSStartedFrame. It will reset to False when - # there's an interruption or TTSStoppedFrame. - self._started = False self._cumulative_time = 0 self._receive_task = None self._keepalive_task = None + self._context_id: Optional[str] = None def can_generate_metrics(self) -> bool: """Check if this service can generate processing metrics. @@ -213,8 +211,6 @@ class NeuphonicTTSService(InterruptibleTTSService): direction: The direction to push the frame. """ await super().push_frame(frame, direction) - if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)): - self._started = False async def process_frame(self, frame: Frame, direction: FrameDirection): """Process frames with special handling for speech control. @@ -230,7 +226,7 @@ class NeuphonicTTSService(InterruptibleTTSService): # processing more frames until we receive a BotStoppedSpeakingFrame. if isinstance(frame, TTSSpeakFrame): await self.pause_processing_frames() - elif isinstance(frame, LLMFullResponseEndFrame) and self._started: + elif isinstance(frame, LLMFullResponseEndFrame): await self.pause_processing_frames() elif isinstance(frame, BotStoppedSpeakingFrame): await self.resume_processing_frames() @@ -304,7 +300,7 @@ class NeuphonicTTSService(InterruptibleTTSService): except Exception as e: await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: - self._started = False + self._context_id = None self._websocket = None await self._call_event_handler("on_disconnected") @@ -317,7 +313,9 @@ class NeuphonicTTSService(InterruptibleTTSService): await self.stop_ttfb_metrics() audio = base64.b64decode(msg["data"]["audio"]) - frame = TTSAudioRawFrame(audio, self.sample_rate, 1) + frame = TTSAudioRawFrame( + audio, self.sample_rate, 1, context_id=self._context_id + ) await self.push_frame(frame) async def _keepalive_task_handler(self): @@ -342,11 +340,12 @@ class NeuphonicTTSService(InterruptibleTTSService): await self._websocket.send(json.dumps(msg)) @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Neuphonic's streaming API. Args: text: The text to synthesize into speech. + context_id: Unique identifier for this TTS context. Yields: Frame: Audio frames containing the synthesized speech. @@ -358,17 +357,17 @@ class NeuphonicTTSService(InterruptibleTTSService): await self._connect() try: - if not self._started: - await self.start_ttfb_metrics() - yield TTSStartedFrame() - self._started = True - self._cumulative_time = 0 + await self.start_ttfb_metrics() + # Store context_id for use in _receive_messages + self._context_id = context_id + yield TTSStartedFrame(context_id=context_id) + self._cumulative_time = 0 await self._send_text(text) await self.start_tts_usage_metrics(text) except Exception as e: yield ErrorFrame(error=f"Unknown error occurred: {e}") - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) await self._disconnect() await self._connect() return @@ -502,11 +501,12 @@ class NeuphonicHttpTTSService(TTSService): return None @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Neuphonic streaming API. Args: text: The text to convert to speech. + context_id: Unique identifier for this TTS context. Yields: Frame: Audio frames containing the synthesized speech and status information. @@ -542,7 +542,7 @@ class NeuphonicHttpTTSService(TTSService): return await self.start_tts_usage_metrics(text) - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) # Process SSE stream line by line async for line in response.content: @@ -564,7 +564,9 @@ class NeuphonicHttpTTSService(TTSService): audio_bytes = base64.b64decode(audio_b64) await self.stop_ttfb_metrics() - yield TTSAudioRawFrame(audio_bytes, self.sample_rate, 1) + yield TTSAudioRawFrame( + audio_bytes, self.sample_rate, 1, context_id=context_id + ) except Exception as e: yield ErrorFrame(error=f"Unknown error occurred: {e}") @@ -578,4 +580,4 @@ class NeuphonicHttpTTSService(TTSService): yield ErrorFrame(error=f"Unknown error occurred: {e}") finally: await self.stop_ttfb_metrics() - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) diff --git a/src/pipecat/services/nvidia/tts.py b/src/pipecat/services/nvidia/tts.py index e90462ad9..6bac54e3a 100644 --- a/src/pipecat/services/nvidia/tts.py +++ b/src/pipecat/services/nvidia/tts.py @@ -153,11 +153,12 @@ class NvidiaTTSService(TTSService): logger.debug(f"Initialized NvidiaTTSService with model: {self.model_name}") @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using NVIDIA Riva TTS. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech data. @@ -193,7 +194,7 @@ class NvidiaTTSService(TTSService): assert self._config is not None, "Synthesis configuration not created" await self.start_ttfb_metrics() - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) logger.debug(f"{self}: Generating TTS [{text}]") @@ -205,12 +206,13 @@ class NvidiaTTSService(TTSService): audio=resp.audio, sample_rate=self.sample_rate, num_channels=1, + context_id=context_id, ) yield frame await self.start_tts_usage_metrics(text) - yield TTSStoppedFrame() - except asyncio.TimeoutError: + yield TTSStoppedFrame(context_id=context_id) + except asyncio.TimeoutError as e: logger.error(f"{self} timeout waiting for audio response") yield ErrorFrame(error=f"{self} error: {e}") except Exception as e: diff --git a/src/pipecat/services/openai/tts.py b/src/pipecat/services/openai/tts.py index 444ae883f..f59f0b31b 100644 --- a/src/pipecat/services/openai/tts.py +++ b/src/pipecat/services/openai/tts.py @@ -168,11 +168,12 @@ class OpenAITTSService(TTSService): ) @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using OpenAI's TTS API. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech data. @@ -212,12 +213,12 @@ class OpenAITTSService(TTSService): CHUNK_SIZE = self.chunk_size - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) async for chunk in r.iter_bytes(CHUNK_SIZE): if len(chunk) > 0: await self.stop_ttfb_metrics() - frame = TTSAudioRawFrame(chunk, self.sample_rate, 1) + frame = TTSAudioRawFrame(chunk, self.sample_rate, 1, context_id=context_id) yield frame - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) except BadRequestError as e: yield ErrorFrame(error=f"Unknown error occurred: {e}") diff --git a/src/pipecat/services/piper/tts.py b/src/pipecat/services/piper/tts.py index e921447ff..a1a038826 100644 --- a/src/pipecat/services/piper/tts.py +++ b/src/pipecat/services/piper/tts.py @@ -86,11 +86,12 @@ class PiperTTSService(TTSService): return True @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Piper. Args: text: The text to convert to speech. + context_id: Unique identifier for this TTS context. Yields: Frame: Audio frames containing the synthesized speech and status frames. @@ -116,11 +117,12 @@ class PiperTTSService(TTSService): await self.start_tts_usage_metrics(text) - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) async for frame in self._stream_audio_frames_from_iterator( async_iterator(self._voice.synthesize(text)), in_sample_rate=self._voice.config.sample_rate, + context_id=context_id, ): await self.stop_ttfb_metrics() yield frame @@ -130,7 +132,7 @@ class PiperTTSService(TTSService): finally: logger.debug(f"{self}: Finished TTS [{text}]") await self.stop_ttfb_metrics() - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) # This assumes a running TTS service running: @@ -184,11 +186,12 @@ class PiperHttpTTSService(TTSService): return True @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Piper's HTTP API. Args: text: The text to convert to speech. + context_id: Unique identifier for this TTS context. Yields: Frame: Audio frames containing the synthesized speech and status frames. @@ -215,12 +218,14 @@ class PiperHttpTTSService(TTSService): await self.start_tts_usage_metrics(text) - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) CHUNK_SIZE = self.chunk_size async for frame in self._stream_audio_frames_from_iterator( - response.content.iter_chunked(CHUNK_SIZE), strip_wav_header=True + response.content.iter_chunked(CHUNK_SIZE), + strip_wav_header=True, + context_id=context_id, ): await self.stop_ttfb_metrics() yield frame @@ -228,4 +233,4 @@ class PiperHttpTTSService(TTSService): yield ErrorFrame(error=f"Unknown error occurred: {e}") finally: await self.stop_ttfb_metrics() - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) diff --git a/src/pipecat/services/playht/tts.py b/src/pipecat/services/playht/tts.py index bc9dd4859..2d4cd0427 100644 --- a/src/pipecat/services/playht/tts.py +++ b/src/pipecat/services/playht/tts.py @@ -13,7 +13,6 @@ supporting both WebSocket streaming and HTTP-based synthesis. import io import json import struct -import uuid import warnings from typing import AsyncGenerator, Optional @@ -169,7 +168,7 @@ class PlayHTTTSService(InterruptibleTTSService): self._user_id = user_id self._websocket_url = None self._receive_task = None - self._request_id = None + self._context_id = None self._settings = { "language": self.language_to_service_language(params.language) @@ -285,7 +284,7 @@ class PlayHTTTSService(InterruptibleTTSService): except Exception as e: await self.push_error(error_msg=f"Error disconnecting: {e}", exception=e) finally: - self._request_id = None + self._context_id = None self._websocket = None await self._call_event_handler("on_disconnected") @@ -328,7 +327,7 @@ class PlayHTTTSService(InterruptibleTTSService): """Handle interruption by stopping metrics and clearing request ID.""" await super()._handle_interruption(frame, direction) await self.stop_all_metrics() - self._request_id = None + self._context_id = None async def _receive_messages(self): """Receive messages from PlayHT websocket.""" @@ -338,7 +337,7 @@ class PlayHTTTSService(InterruptibleTTSService): if message.startswith(b"RIFF"): continue await self.stop_ttfb_metrics() - frame = TTSAudioRawFrame(message, self.sample_rate, 1) + frame = TTSAudioRawFrame(message, self.sample_rate, 1, context_id=self._context_id) await self.push_frame(frame) else: logger.debug(f"Received text message: {message}") @@ -349,20 +348,21 @@ class PlayHTTTSService(InterruptibleTTSService): logger.debug(f"Started processing request: {msg.get('request_id')}") elif msg.get("type") == "end": # Handle end of stream - if "request_id" in msg and msg["request_id"] == self._request_id: - await self.push_frame(TTSStoppedFrame()) - self._request_id = None + if "request_id" in msg and msg["request_id"] == self._context_id: + await self.push_frame(TTSStoppedFrame(context_id=self._context_id)) + self._context_id = None elif "error" in msg: await self.push_error(error_msg=f"Error: {msg['error']}") except json.JSONDecodeError: logger.error(f"Invalid JSON message: {message}") @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate TTS audio from text using PlayHT's WebSocket API. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech. @@ -374,10 +374,10 @@ class PlayHTTTSService(InterruptibleTTSService): if not self._websocket or self._websocket.state is State.CLOSED: await self._connect() - if not self._request_id: + if not self._context_id: await self.start_ttfb_metrics() - yield TTSStartedFrame() - self._request_id = str(uuid.uuid4()) + yield TTSStartedFrame(context_id=context_id) + self._context_id = context_id tts_command = { "text": text, @@ -388,7 +388,7 @@ class PlayHTTTSService(InterruptibleTTSService): "language": self._settings["language"], "speed": self._settings["speed"], "seed": self._settings["seed"], - "request_id": self._request_id, + "request_id": self._context_id, } try: @@ -396,7 +396,7 @@ class PlayHTTTSService(InterruptibleTTSService): await self.start_tts_usage_metrics(text) except Exception as e: yield ErrorFrame(error=f"Unknown error occurred: {e}") - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) await self._disconnect() await self._connect() return @@ -540,11 +540,12 @@ class PlayHTHttpTTSService(TTSService): return language_to_playht_language(language) @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate TTS audio from text using PlayHT's HTTP API. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech. @@ -579,7 +580,7 @@ class PlayHTHttpTTSService(TTSService): await self.start_tts_usage_metrics(text) - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) async with aiohttp.ClientSession() as session: async with session.post( @@ -616,16 +617,20 @@ class PlayHTHttpTTSService(TTSService): audio_data = buffer[fh.tell() :] if len(audio_data) > 0: await self.stop_ttfb_metrics() - frame = TTSAudioRawFrame(audio_data, self.sample_rate, 1) + frame = TTSAudioRawFrame( + audio_data, self.sample_rate, 1, context_id=context_id + ) yield frame in_header = False elif len(chunk) > 0: await self.stop_ttfb_metrics() - frame = TTSAudioRawFrame(chunk, self.sample_rate, 1) + frame = TTSAudioRawFrame( + chunk, self.sample_rate, 1, context_id=context_id + ) yield frame except Exception as e: yield ErrorFrame(error=f"Unknown error occurred: {e}") finally: await self.stop_ttfb_metrics() - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) diff --git a/src/pipecat/services/resembleai/tts.py b/src/pipecat/services/resembleai/tts.py index 21a8c7a78..964b9fa18 100644 --- a/src/pipecat/services/resembleai/tts.py +++ b/src/pipecat/services/resembleai/tts.py @@ -84,9 +84,11 @@ class ResembleAITTSService(AudioContextWordTTSService): self._websocket = None self._request_id_counter = 0 - self._current_request_id = None self._receive_task = None + # Map request_id to context_id for tracking TTS requests + self._request_id_to_context: dict[int, str] = {} + # Per-request audio buffers to handle concurrent TTS requests # ResembleAI may send odd-length data even for PCM_16, so buffering helps us # create properly aligned frames while maintaining smooth audio output @@ -122,7 +124,7 @@ class ResembleAITTSService(AudioContextWordTTSService): "voice_uuid": self._voice_id, "data": text, "binary_response": False, # Use JSON frames to get timestamps - "request_id": self._request_id_counter, + "request_id": self._request_id_counter, # ResembleAI only accepts number "output_format": self._settings["output_format"], "sample_rate": self._settings["sample_rate"], "precision": self._settings["precision"], @@ -202,10 +204,10 @@ class ResembleAITTSService(AudioContextWordTTSService): except Exception as e: await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: - self._current_request_id = None self._websocket = None self._audio_buffers.clear() self._playback_started.clear() + self._request_id_to_context.clear() await self._call_event_handler("on_disconnected") def _get_websocket(self): @@ -230,18 +232,12 @@ class ResembleAITTSService(AudioContextWordTTSService): """ await super()._handle_interruption(frame, direction) await self.stop_all_metrics() - # Note: Resemble AI doesn't have an explicit cancel mechanism, - # but we can stop processing by resetting our current request_id - self._current_request_id = None async def flush_audio(self): """Flush any pending audio and finalize the current context.""" - if not self._current_request_id: - return logger.trace(f"{self}: flushing audio") # For Resemble AI, we just wait for the audio_end message # which is handled in _process_messages - self._current_request_id = None async def _process_messages(self): """Process incoming WebSocket messages from Resemble AI.""" @@ -259,10 +255,10 @@ class ResembleAITTSService(AudioContextWordTTSService): request_id = msg.get("request_id") # Convert request_id to string for audio context tracking - request_id_str = str(request_id) + context_id = self._request_id_to_context.get(request_id, str(request_id)) # Check if this message belongs to a valid audio context - if not self.audio_context_available(request_id_str): + if not self.audio_context_available(context_id): continue if msg_type == "audio": @@ -279,20 +275,20 @@ class ResembleAITTSService(AudioContextWordTTSService): continue # Get or create buffer for this request - if request_id_str not in self._audio_buffers: - self._audio_buffers[request_id_str] = bytearray() - self._playback_started[request_id_str] = False - buffer = self._audio_buffers[request_id_str] + if context_id not in self._audio_buffers: + self._audio_buffers[context_id] = bytearray() + self._playback_started[context_id] = False + buffer = self._audio_buffers[context_id] # Add to buffer buffer.extend(audio_bytes) # Wait for jitter buffer to fill before starting playback # This absorbs network latency gaps (ResembleAI sends in bursts) - if not self._playback_started.get(request_id_str, False): + if not self._playback_started.get(context_id, False): if len(buffer) < self._jitter_buffer_bytes: continue - self._playback_started[request_id_str] = True + self._playback_started[context_id] = True # Send complete (even-byte) chunks for PCM_16 alignment while len(buffer) >= self._buffer_threshold_bytes: @@ -301,8 +297,8 @@ class ResembleAITTSService(AudioContextWordTTSService): chunk_size -= 1 chunk_to_send = bytes(buffer[:chunk_size]) - self._audio_buffers[request_id_str] = buffer[chunk_size:] - buffer = self._audio_buffers[request_id_str] + self._audio_buffers[context_id] = buffer[chunk_size:] + buffer = self._audio_buffers[context_id] if len(chunk_to_send) == 0: continue @@ -311,8 +307,9 @@ class ResembleAITTSService(AudioContextWordTTSService): audio=chunk_to_send, sample_rate=self.sample_rate, num_channels=1, + context_id=context_id, ) - await self.append_to_audio_context(request_id_str, frame) + await self.append_to_audio_context(context_id, frame) # Process timestamps if available timestamps = msg.get("audio_timestamps", {}) @@ -328,13 +325,13 @@ class ResembleAITTSService(AudioContextWordTTSService): word_times.append((char, start_time)) if word_times: - await self.add_word_timestamps(word_times) + await self.add_word_timestamps(word_times, context_id) elif msg_type == "audio_end": await self.stop_ttfb_metrics() # Flush remaining buffer, ensuring even length for PCM_16 - buffer = self._audio_buffers.get(request_id_str, bytearray()) + buffer = self._audio_buffers.get(context_id, bytearray()) if buffer: remaining = bytes(buffer) # PCM_16 requires even number of bytes @@ -345,20 +342,21 @@ class ResembleAITTSService(AudioContextWordTTSService): audio=remaining, sample_rate=self.sample_rate, num_channels=1, + context_id=context_id, ) - await self.append_to_audio_context(request_id_str, frame) + await self.append_to_audio_context(context_id, frame) # Clean up buffer and playback tracking for this request - if request_id_str in self._audio_buffers: - del self._audio_buffers[request_id_str] - if request_id_str in self._playback_started: - del self._playback_started[request_id_str] + if context_id in self._audio_buffers: + del self._audio_buffers[context_id] + if context_id in self._playback_started: + del self._playback_started[context_id] + # Clean up request_id mapping + if request_id in self._request_id_to_context: + del self._request_id_to_context[request_id] - await self.add_word_timestamps([("TTSStoppedFrame", 0), ("Reset", 0)]) - await self.remove_audio_context(request_id_str) - # Clear current request if this was it - if self._current_request_id == request_id: - self._current_request_id = None + await self.add_word_timestamps([("TTSStoppedFrame", 0), ("Reset", 0)], context_id) + await self.remove_audio_context(context_id) elif msg_type == "error": error_name = msg.get("error_name", "Unknown") @@ -369,19 +367,15 @@ class ResembleAITTSService(AudioContextWordTTSService): ) # Clean up buffer and playback tracking for this request - if request_id_str in self._audio_buffers: - del self._audio_buffers[request_id_str] - if request_id_str in self._playback_started: - del self._playback_started[request_id_str] + if context_id in self._audio_buffers: + del self._audio_buffers[context_id] + if context_id in self._playback_started: + del self._playback_started[context_id] - await self.push_frame(TTSStoppedFrame()) + await self.push_frame(TTSStoppedFrame(context_id=context_id)) await self.stop_all_metrics() await self.push_error(ErrorFrame(error=f"{self} error: {error_name} - {error_msg}")) - # Clear current request if this was it - if self._current_request_id == request_id: - self._current_request_id = None - # Check if this is an unrecoverable error (connection-level failure) if status_code in [401, 403]: # Close and reconnect for auth errors @@ -402,11 +396,12 @@ class ResembleAITTSService(AudioContextWordTTSService): await self._connect_websocket() @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Resemble AI's streaming API. Args: text: The text to synthesize into speech. + context_id: Unique identifier for this TTS context. Yields: Frame: Audio frames containing the synthesized speech. @@ -417,15 +412,13 @@ class ResembleAITTSService(AudioContextWordTTSService): if not self._websocket or self._websocket.state is State.CLOSED: await self._connect() - if not self._current_request_id: - await self.start_ttfb_metrics() - yield TTSStartedFrame() - # Track the current request_id we're processing - self._current_request_id = self._request_id_counter + await self.start_ttfb_metrics() + yield TTSStartedFrame(context_id=context_id) - # Create audio context using request_id (converted to string) - request_id_str = str(self._request_id_counter) - await self.create_audio_context(request_id_str) + # Map request_id to context_id for tracking + self._request_id_to_context[self._request_id_counter] = context_id + + await self.create_audio_context(context_id) msg = self._build_msg(text=text) @@ -434,7 +427,7 @@ class ResembleAITTSService(AudioContextWordTTSService): await self.start_tts_usage_metrics(text) except Exception as e: yield ErrorFrame(error=f"Unknown error occurred: {e}") - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) await self._disconnect() await self._connect() return diff --git a/src/pipecat/services/rime/tts.py b/src/pipecat/services/rime/tts.py index 39f1a626c..e38e840e6 100644 --- a/src/pipecat/services/rime/tts.py +++ b/src/pipecat/services/rime/tts.py @@ -12,7 +12,6 @@ using Rime's API for streaming and batch audio synthesis. import base64 import json -import uuid from typing import Any, AsyncGenerator, Mapping, Optional import aiohttp @@ -387,6 +386,7 @@ class RimeTTSService(AudioContextWordTTSService): if not msg or not self.audio_context_available(msg["contextId"]): continue + context_id = msg["contextId"] if msg["type"] == "chunk": # Process audio chunk await self.stop_ttfb_metrics() @@ -395,8 +395,9 @@ class RimeTTSService(AudioContextWordTTSService): audio=base64.b64decode(msg["data"]), sample_rate=self.sample_rate, num_channels=1, + context_id=context_id, ) - await self.append_to_audio_context(msg["contextId"], frame) + await self.append_to_audio_context(context_id, frame) elif msg["type"] == "timestamps": # Process word timing information @@ -409,7 +410,7 @@ class RimeTTSService(AudioContextWordTTSService): # Calculate word timing pairs word_pairs = self._calculate_word_times(words, starts, ends) if word_pairs: - await self.add_word_timestamps(word_pairs) + await self.add_word_timestamps(word_pairs, context_id=context_id) self._cumulative_time = ends[-1] + self._cumulative_time logger.debug(f"Updated cumulative time to: {self._cumulative_time}") @@ -432,11 +433,12 @@ class RimeTTSService(AudioContextWordTTSService): await self.add_word_timestamps([("Reset", 0)]) @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Rime's streaming API. Args: text: The text to convert to speech. + context_id: Unique identifier for this TTS context. Yields: Frame: Audio frames containing the synthesized speech. @@ -449,9 +451,9 @@ class RimeTTSService(AudioContextWordTTSService): try: if not self._context_id: await self.start_ttfb_metrics() - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) self._cumulative_time = 0 - self._context_id = str(uuid.uuid4()) + self._context_id = context_id await self.create_audio_context(self._context_id) msg = self._build_msg(text=text) @@ -459,7 +461,7 @@ class RimeTTSService(AudioContextWordTTSService): await self.start_tts_usage_metrics(text) except Exception as e: yield ErrorFrame(error=f"Unknown error occurred: {e}") - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) await self._disconnect() await self._connect() return @@ -558,11 +560,12 @@ class RimeHttpTTSService(TTSService): return language_to_rime_language(language) @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Rime's HTTP API. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech. @@ -601,13 +604,14 @@ class RimeHttpTTSService(TTSService): await self.start_tts_usage_metrics(text) - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) CHUNK_SIZE = self.chunk_size async for frame in self._stream_audio_frames_from_iterator( response.content.iter_chunked(CHUNK_SIZE), strip_wav_header=need_to_strip_wav_header, + context_id=context_id, ): await self.stop_ttfb_metrics() yield frame @@ -616,7 +620,7 @@ class RimeHttpTTSService(TTSService): yield ErrorFrame(error=f"Unknown error occurred: {e}") finally: await self.stop_ttfb_metrics() - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) class RimeNonJsonTTSService(InterruptibleTTSService): @@ -716,8 +720,8 @@ class RimeNonJsonTTSService(InterruptibleTTSService): if params.extra: self._settings.update(params.extra) - self._started = False self._receive_task = None + self._context_id: Optional[str] = None def can_generate_metrics(self) -> bool: """Check if this service can generate processing metrics. @@ -767,8 +771,6 @@ class RimeNonJsonTTSService(InterruptibleTTSService): direction: The direction to push the frame. """ await super().push_frame(frame, direction) - if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)): - self._started = False async def _connect(self): """Establish WebSocket connection and start receive task.""" @@ -817,7 +819,7 @@ class RimeNonJsonTTSService(InterruptibleTTSService): except Exception as e: await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: - self._started = False + self._context_id = None self._websocket = None await self._call_event_handler("on_disconnected") @@ -847,17 +849,19 @@ class RimeNonJsonTTSService(InterruptibleTTSService): audio=message, sample_rate=self.sample_rate, num_channels=1, + context_id=self._context_id, ) await self.push_frame(frame) except Exception as e: await self.push_error(error_msg=f"Error: {e}", exception=e) @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Rime's streaming API. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech. @@ -867,17 +871,17 @@ class RimeNonJsonTTSService(InterruptibleTTSService): if not self._websocket or self._websocket.state is State.CLOSED: await self._connect() try: - if not self._started: - await self.start_ttfb_metrics() - yield TTSStartedFrame() - self._started = True + await self.start_ttfb_metrics() + # Store context_id for use in _receive_messages + self._context_id = context_id + yield TTSStartedFrame(context_id=context_id) # Send bare text (not JSON) await self._get_websocket().send(text) await self.start_tts_usage_metrics(text) except Exception as e: yield ErrorFrame(error=f"Unknown error occurred: {e}") - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) await self._disconnect() await self._connect() return diff --git a/src/pipecat/services/sarvam/tts.py b/src/pipecat/services/sarvam/tts.py index 778a29005..753293c75 100644 --- a/src/pipecat/services/sarvam/tts.py +++ b/src/pipecat/services/sarvam/tts.py @@ -462,11 +462,12 @@ class SarvamHttpTTSService(TTSService): self._settings["sample_rate"] = self.sample_rate @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Sarvam AI's API. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech. @@ -503,7 +504,7 @@ class SarvamHttpTTSService(TTSService): url = f"{self._base_url}/text-to-speech" - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) async with self._session.post(url, json=payload, headers=headers) as response: if response.status != 200: @@ -533,6 +534,7 @@ class SarvamHttpTTSService(TTSService): audio=audio_data, sample_rate=self.sample_rate, num_channels=1, + context_id=context_id, ) yield frame @@ -541,7 +543,7 @@ class SarvamHttpTTSService(TTSService): yield ErrorFrame(error=f"Error generating TTS: {e}", exception=e) finally: await self.stop_ttfb_metrics() - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) class SarvamTTSService(InterruptibleTTSService): @@ -789,9 +791,9 @@ class SarvamTTSService(InterruptibleTTSService): elif params.temperature != 0.6: logger.warning(f"temperature parameter is ignored for {model}") - self._started = False self._receive_task = None self._keepalive_task = None + self._context_id: Optional[str] = None def can_generate_metrics(self) -> bool: """Check if this service can generate processing metrics. @@ -859,8 +861,6 @@ class SarvamTTSService(InterruptibleTTSService): direction: The direction to push the frame. """ await super().push_frame(frame, direction) - if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)): - self._started = False async def process_frame(self, frame: Frame, direction: FrameDirection): """Process a frame and flush audio if it's the end of a full response.""" @@ -956,7 +956,7 @@ class SarvamTTSService(InterruptibleTTSService): except Exception as e: await self.push_error(error_msg=f"Error closing websocket: {e}", exception=e) finally: - self._started = False + self._context_id = None self._websocket = None await self._call_event_handler("on_disconnected") @@ -974,7 +974,9 @@ class SarvamTTSService(InterruptibleTTSService): # Check for interruption before processing audio await self.stop_ttfb_metrics() audio = base64.b64decode(msg["data"]["audio"]) - frame = TTSAudioRawFrame(audio, self.sample_rate, 1) + frame = TTSAudioRawFrame( + audio, self.sample_rate, 1, context_id=self._context_id + ) await self.push_frame(frame) elif msg.get("type") == "error": error_msg = msg["data"]["message"] @@ -984,7 +986,6 @@ class SarvamTTSService(InterruptibleTTSService): if "too long" in error_msg.lower() or "timeout" in error_msg.lower(): logger.warning("Connection timeout detected, service may need restart") - self._started = False await self.push_frame(ErrorFrame(error=f"TTS Error: {error_msg}")) async def _keepalive_task_handler(self): @@ -1009,16 +1010,17 @@ class SarvamTTSService(InterruptibleTTSService): logger.warning("WebSocket not ready, cannot send text") @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech audio frames from input text using Sarvam TTS. Sends text over WebSocket for synthesis and yields corresponding audio or status frames. Args: text: The text input to synthesize. + context_id: The context ID for tracking audio frames. Yields: - Frame objects including TTSStartedFrame, TTSAudioRawFrame(s), or TTSStoppedFrame. + Frame objects including TTSStartedFrame, TTSAudioRawFrame(s, context_id=context_id), or TTSStoppedFrame. """ logger.debug(f"Generating TTS: [{text}]") @@ -1027,15 +1029,15 @@ class SarvamTTSService(InterruptibleTTSService): await self._connect() try: - if not self._started: - await self.start_ttfb_metrics() - yield TTSStartedFrame() - self._started = True + await self.start_ttfb_metrics() + # Store context_id for use in _receive_messages + self._context_id = context_id + yield TTSStartedFrame(context_id=context_id) await self._send_text(text) await self.start_tts_usage_metrics(text) except Exception as e: yield ErrorFrame(error=f"Unknown error occurred: {e}") - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) await self._disconnect() await self._connect() return diff --git a/src/pipecat/services/speechmatics/tts.py b/src/pipecat/services/speechmatics/tts.py index d7b2f4e2a..0f3ff0cb6 100644 --- a/src/pipecat/services/speechmatics/tts.py +++ b/src/pipecat/services/speechmatics/tts.py @@ -106,11 +106,12 @@ class SpeechmaticsTTSService(TTSService): return True @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Speechmatics' HTTP API. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech. @@ -187,7 +188,7 @@ class SpeechmaticsTTSService(TTSService): await self.start_tts_usage_metrics(text) # Emit the TTS started frame - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) # Process the response in streaming chunks first_chunk = True @@ -216,6 +217,7 @@ class SpeechmaticsTTSService(TTSService): audio=audio_data, sample_rate=self.sample_rate, num_channels=1, + context_id=context_id, ) # Successfully processed the response, break out of retry loop @@ -225,7 +227,7 @@ class SpeechmaticsTTSService(TTSService): yield ErrorFrame(error=f"Error generating TTS: {e}") finally: # Emit the TTS stopped frame - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id) def _get_endpoint_url(base_url: str, voice: str, sample_rate: int) -> str: diff --git a/src/pipecat/services/xtts/tts.py b/src/pipecat/services/xtts/tts.py index 2e43d828c..bf4eb4f03 100644 --- a/src/pipecat/services/xtts/tts.py +++ b/src/pipecat/services/xtts/tts.py @@ -148,11 +148,12 @@ class XTTSService(TTSService): self._studio_speakers = await r.json() @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using XTTS streaming server. Args: text: The text to synthesize into speech. + context_id: The context ID for tracking audio frames. Yields: Frame: Audio frames containing the synthesized speech. @@ -186,7 +187,7 @@ class XTTSService(TTSService): await self.start_tts_usage_metrics(text) - yield TTSStartedFrame() + yield TTSStartedFrame(context_id=context_id) CHUNK_SIZE = self.chunk_size @@ -211,7 +212,9 @@ class XTTSService(TTSService): bytes(process_data), 24000, self.sample_rate ) # Create the frame with the resampled audio - frame = TTSAudioRawFrame(resampled_audio, self.sample_rate, 1) + frame = TTSAudioRawFrame( + resampled_audio, self.sample_rate, 1, context_id=context_id + ) yield frame # Process any remaining data in the buffer. @@ -219,7 +222,9 @@ class XTTSService(TTSService): resampled_audio = await self._resampler.resample( bytes(buffer), 24000, self.sample_rate ) - frame = TTSAudioRawFrame(resampled_audio, self.sample_rate, 1) + frame = TTSAudioRawFrame( + resampled_audio, self.sample_rate, 1, context_id=context_id + ) yield frame - yield TTSStoppedFrame() + yield TTSStoppedFrame(context_id=context_id)