diff --git a/changelog/3584.added.2.md b/changelog/3584.added.2.md new file mode 100644 index 000000000..3113154fe --- /dev/null +++ b/changelog/3584.added.2.md @@ -0,0 +1,3 @@ +- Added `append_to_context` parameter to `TTSSpeakFrame` for conditional LLM context addition. + - Allows fine-grained control over whether text should be added to conversation context + - Defaults to `True` to maintain backward compatibility diff --git a/changelog/3584.added.md b/changelog/3584.added.md new file mode 100644 index 000000000..08a81ee85 --- /dev/null +++ b/changelog/3584.added.md @@ -0,0 +1,4 @@ +- Added TTS context tracking system with `context_id` field to trace audio generation through the pipeline. + - `TTSAudioRawFrame`, `TTSStartedFrame`, `TTSStoppedFrame` now include `context_id` + - `AggregatedTextFrame` and `TTSTextFrame` now include `context_id` + - Enables tracking which TTS request generated specific audio chunks diff --git a/changelog/3584.changed.2.md b/changelog/3584.changed.2.md new file mode 100644 index 000000000..f5356b4c9 --- /dev/null +++ b/changelog/3584.changed.2.md @@ -0,0 +1,3 @@ +- Simplified context aggregators to use `frame.append_to_context` flag instead of tracking internal state. + - Cleaner logic in `LLMResponseAggregator` and `LLMResponseUniversalAggregator` + - More consistent behavior across aggregator implementations diff --git a/changelog/3584.changed.3.md b/changelog/3584.changed.3.md new file mode 100644 index 000000000..426cfbde7 --- /dev/null +++ b/changelog/3584.changed.3.md @@ -0,0 +1,4 @@ +- ⚠️ `TTSService.run_tts()` now requires a `context_id` parameter for context tracking. + - Custom TTS service implementations must update their `run_tts()` signature + - Before: `async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:` + - After: `async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:` diff --git a/changelog/3584.changed.md b/changelog/3584.changed.md new file mode 100644 index 000000000..fd535b7d8 --- /dev/null +++ b/changelog/3584.changed.md @@ -0,0 +1,3 @@ +- Updated all 30+ TTS service implementations to support context tracking with `context_id`. + - Services now generate and propagate context IDs through TTS frames + - Enables end-to-end tracing of TTS requests through the pipeline diff --git a/examples/foundational/14d-function-calling-openai-video.py b/examples/foundational/14d-function-calling-openai-video.py index 88c0e8ba3..d948e9caa 100644 --- a/examples/foundational/14d-function-calling-openai-video.py +++ b/examples/foundational/14d-function-calling-openai-video.py @@ -99,7 +99,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): @llm.event_handler("on_function_calls_started") async def on_function_calls_started(service, function_calls): - await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) + await tts.queue_frame(TTSSpeakFrame("Let me check on that.", append_to_context=False)) fetch_image_function = FunctionSchema( name="fetch_user_image", @@ -174,6 +174,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Client disconnected") await task.cancel() + @tts.event_handler("on_tts_request") + async def on_tts_request(tts, context_id: str, text: str): + logger.debug(f"On TTS request: {context_id}: {text}") + runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) await runner.run(task) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index de33c6187..7b821a9bc 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -279,9 +279,12 @@ class TTSAudioRawFrame(OutputAudioRawFrame): """Audio data frame generated by Text-to-Speech services. A chunk of output audio generated by a TTS service, ready for playback. + + Parameters: + context_id: Unique identifier for the TTS context that generated this audio. """ - pass + context_id: Optional[str] = None @dataclass @@ -343,6 +346,11 @@ class TextFrame(DataFrame): Parameters: text: The text content. + skip_tts: Whether this text should be skipped by the TTS service. + includes_inter_frame_spaces: Whether any necessary inter-frame (leading/trailing) spaces are already + included in the text. + append_to_context: Whether this text should be appended to the LLM context. + Defaults to True. """ text: str @@ -397,9 +405,11 @@ class AggregatedTextFrame(TextFrame): Parameters: aggregated_by: Method used to aggregate the text frames. + context_id: Unique identifier for the TTS context that generated this text. """ aggregated_by: AggregationType | str + context_id: Optional[str] = None @dataclass @@ -411,9 +421,13 @@ class VisionTextFrame(LLMTextFrame): @dataclass class TTSTextFrame(AggregatedTextFrame): - """Text frame generated by Text-to-Speech services.""" + """Text frame generated by Text-to-Speech services. - pass + Parameters: + context_id: Unique identifier for the TTS context that generated this text. + """ + + context_id: Optional[str] = None @dataclass @@ -923,9 +937,11 @@ class TTSSpeakFrame(DataFrame): Parameters: text: The text to be spoken. + append_to_context: Whether to append the text to the context. """ text: str + append_to_context: Optional[bool] = None @dataclass @@ -2023,16 +2039,23 @@ class TTSStartedFrame(ControlFrame): TTSStoppedFrame. These frames can be used for aggregating audio frames in a transport to optimize the size of frames sent to the session, without needing to control this in the TTS service. + + Parameters: + context_id: Unique identifier for this TTS context. """ - pass + context_id: Optional[str] = None @dataclass class TTSStoppedFrame(ControlFrame): - """Frame indicating the end of a TTS response.""" + """Frame indicating the end of a TTS response. - pass + Parameters: + context_id: Unique identifier for this TTS context. + """ + + context_id: Optional[str] = None @dataclass diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index e11ededbf..44e5ce252 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -1059,7 +1059,7 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): await self.push_aggregation() async def _handle_text(self, frame: TextFrame): - if not self._started or not frame.append_to_context: + if not frame.append_to_context: return if self._params.expect_stripped_words: diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 3b162d48f..574af7bbf 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -52,6 +52,7 @@ from pipecat.frames.frames import ( StartFrame, TextFrame, TranscriptionFrame, + TranslationFrame, UserImageRawFrame, UserMuteStartedFrame, UserMuteStoppedFrame, @@ -795,7 +796,6 @@ class LLMAssistantAggregator(LLMContextAggregator): DeprecationWarning, ) - self._started = 0 self._function_calls_in_progress: Dict[str, Optional[FunctionCallInProgressFrame]] = {} self._function_calls_image_results: Dict[str, UserImageRawFrame] = {} self._context_updated_tasks: Set[asyncio.Task] = set() @@ -917,12 +917,10 @@ class LLMAssistantAggregator(LLMContextAggregator): async def _handle_interruptions(self, frame: InterruptionFrame): await self._trigger_assistant_turn_stopped() - self._started = 0 await self.reset() async def _handle_end_or_cancel(self, frame: Frame): await self._trigger_assistant_turn_stopped() - self._started = 0 async def _handle_function_calls_started(self, frame: FunctionCallsStartedFrame): function_names = [f"{f.function_name}:{f.tool_call_id}" for f in frame.function_calls] @@ -1067,15 +1065,17 @@ class LLMAssistantAggregator(LLMContextAggregator): ) async def _handle_llm_start(self, _: LLMFullResponseStartFrame): - self._started += 1 await self._trigger_assistant_turn_started() async def _handle_llm_end(self, _: LLMFullResponseEndFrame): - self._started -= 1 await self._trigger_assistant_turn_stopped() async def _handle_text(self, frame: TextFrame): - if not self._started or not frame.append_to_context: + # Skip TextFrame types not intended to build the assistant context + if isinstance(frame, (TranscriptionFrame, TranslationFrame, InterimTranscriptionFrame)): + return + + if not frame.append_to_context: return # Make sure we really have text (spaces count, too!) @@ -1089,18 +1089,12 @@ class LLMAssistantAggregator(LLMContextAggregator): ) async def _handle_thought_start(self, frame: LLMThoughtStartFrame): - if not self._started: - return - await self._reset_thought_aggregation() self._thought_append_to_context = frame.append_to_context self._thought_llm = frame.llm self._thought_start_time = time_now_iso8601() async def _handle_thought_text(self, frame: LLMThoughtTextFrame): - if not self._started: - return - # Make sure we really have text (spaces count, too!) if len(frame.text) == 0: return @@ -1112,9 +1106,6 @@ class LLMAssistantAggregator(LLMContextAggregator): ) async def _handle_thought_end(self, frame: LLMThoughtEndFrame): - if not self._started: - return - thought = concatenate_aggregated_text(self._thought_aggregation) if self._thought_append_to_context: 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/tts_service.py b/src/pipecat/services/tts_service.py index 37aad0372..239d2398b 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -7,7 +7,9 @@ """Base classes for Text-to-speech services.""" import asyncio +import uuid from abc import abstractmethod +from dataclasses import dataclass from typing import ( Any, AsyncGenerator, @@ -58,6 +60,17 @@ from pipecat.utils.text.simple_text_aggregator import SimpleTextAggregator from pipecat.utils.time import seconds_to_nanoseconds +@dataclass +class TTSContext: + """Context information for a TTS request. + + Attributes: + append_to_context: Whether this TTS output should be appended to the conversation context. + """ + + append_to_context: bool = True + + class TTSService(AIService): """Base class for text-to-speech services. @@ -66,9 +79,10 @@ class TTSService(AIService): sentence aggregation, silence insertion, and frame processing control. Event handlers: - on_connected: Called when connected to the STT service. - on_connected: Called when disconnected from the STT service. - on_connection_error: Called when a connection to the STT service error occurs. + on_connected: Called when connected to the TTS service. + on_disconnected: Called when disconnected from the TTS service. + on_connection_error: Called when a connection to the TTS service error occurs. + on_tts_request: Called before a TTS request is made, with the context ID and text. Example:: @@ -81,8 +95,12 @@ class TTSService(AIService): logger.debug(f"TTS disconnected") @tts.event_handler("on_connection_error") - async def on_connection_error(stt: TTSService, error: str): + async def on_connection_error(tts: TTSService, error: str): logger.error(f"TTS connection error: {error}") + + @tts.event_handler("on_tts_request") + async def on_tts_request(tts: TTSService, context_id: str, text: str): + logger.debug(f"TTS request: {context_id} - {text}") """ def __init__( @@ -209,10 +227,12 @@ class TTSService(AIService): self._stop_frame_queue: asyncio.Queue = asyncio.Queue() self._processing_text: bool = False + self._tts_contexts: Dict[str, TTSContext] = {} self._register_event_handler("on_connected") self._register_event_handler("on_disconnected") self._register_event_handler("on_connection_error") + self._register_event_handler("on_tts_request") @property def sample_rate(self) -> int: @@ -256,15 +276,26 @@ class TTSService(AIService): """ self._voice_id = voice + def create_context_id(self) -> str: + """Generate a unique context ID for a TTS request. + + This method can be overridden by subclasses to provide custom context ID generation. + + Returns: + A unique string identifier for the TTS context. + """ + return str(uuid.uuid4()) + # Converts the text to audio. @abstractmethod - 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. This method must be implemented by subclasses to provide actual TTS functionality. Args: text: The text to synthesize into speech. + context_id: Unique identifier for this TTS context. Yields: Frame: Audio frames containing the synthesized speech. @@ -463,7 +494,10 @@ class TTSService(AIService): # Store if we were processing text or not so we can set it back. processing_text = self._processing_text # Assumption: text in TTSSpeakFrame does not include inter-frame spaces - await self._push_tts_frames(AggregatedTextFrame(frame.text, AggregationType.SENTENCE)) + await self._push_tts_frames( + AggregatedTextFrame(frame.text, AggregationType.SENTENCE), + append_tts_text_to_context=frame.append_to_context, + ) # We pause processing incoming frames because we are sending data to # the TTS. We pause to avoid audio overlapping. await self._maybe_pause_frame_processing() @@ -484,6 +518,12 @@ class TTSService(AIService): frame: The frame to push. direction: The direction to push the frame. """ + # Clean up context when we see TTSStoppedFrame + if isinstance(frame, TTSStoppedFrame) and frame.context_id: + if frame.context_id in self._tts_contexts: + logger.debug(f"{self} cleaning up TTS context {frame.context_id}") + del self._tts_contexts[frame.context_id] + if self._push_silence_after_stop and isinstance(frame, TTSStoppedFrame): silence_num_bytes = int(self._silence_time_s * self.sample_rate * 2) # 16-bit silence_frame = TTSAudioRawFrame( @@ -513,6 +553,7 @@ class TTSService(AIService): *, strip_wav_header: bool = False, in_sample_rate: Optional[int] = None, + context_id: Optional[str] = None, ) -> AsyncGenerator[Frame, None]: """Stream audio frames from an async byte iterator with optional resampling. @@ -526,6 +567,7 @@ class TTSService(AIService): strip_wav_header: Strip WAV header and parse source sample rate from it. in_sample_rate: Source sample rate for raw PCM data. Overrides WAV-detected rate if both are provided. + context_id: Unique identifier for this TTS context. """ buffer = bytearray() @@ -555,7 +597,10 @@ class TTSService(AIService): buffer = buffer[aligned_length:] # keep any leftover byte if len(aligned_chunk) > 0: - yield TTSAudioRawFrame(aligned_chunk, self.sample_rate, 1) + frame = TTSAudioRawFrame( + bytes(aligned_chunk), self.sample_rate, 1, context_id=context_id + ) + yield frame if len(buffer) > 0: # Make sure we don't need an extra padding byte. @@ -601,7 +646,10 @@ class TTSService(AIService): ) async def _push_tts_frames( - self, src_frame: AggregatedTextFrame, includes_inter_frame_spaces: Optional[bool] = False + self, + src_frame: AggregatedTextFrame, + includes_inter_frame_spaces: Optional[bool] = False, + append_tts_text_to_context: Optional[bool] = True, ): type = src_frame.aggregated_by text = src_frame.text @@ -636,11 +684,15 @@ class TTSService(AIService): await self.stop_processing_metrics() return + # Create context ID and store metadata + context_id = self.create_context_id() + # To support use cases that may want to know the text before it's spoken, we # push the AggregatedTextFrame version before transforming and sending to TTS. # However, we do not want to add this text to the assistant context until it # is spoken, so we set append_to_context to False. src_frame.append_to_context = False + src_frame.context_id = context_id await self.push_frame(src_frame) # Note: Text transformations are meant to only affect the text sent to the TTS for @@ -653,9 +705,19 @@ class TTSService(AIService): if aggregation_type == type or aggregation_type == "*": transformed_text = await transform(transformed_text, type) + self._tts_contexts[context_id] = TTSContext( + append_to_context=append_tts_text_to_context + if append_tts_text_to_context is not None + else True + ) + # Apply any final text preparation (e.g., trailing space) prepared_text = self._prepare_text_for_tts(transformed_text) - await self.process_generator(self.run_tts(prepared_text)) + + # Trigger event before starting TTS + await self._call_event_handler("on_tts_request", context_id, prepared_text) + + await self.process_generator(self.run_tts(prepared_text, context_id)) await self.stop_processing_metrics() @@ -669,6 +731,10 @@ class TTSService(AIService): # or transformations. frame = TTSTextFrame(text, aggregated_by=type) frame.includes_inter_frame_spaces = includes_inter_frame_spaces + frame.context_id = context_id + # Only override append_to_context if explicitly set + if append_tts_text_to_context is not None: + frame.append_to_context = append_tts_text_to_context await self.push_frame(frame) async def _stop_frame_handler(self): @@ -721,18 +787,24 @@ class WordTTSService(TTSService): """Reset word timestamp tracking.""" self._initial_word_timestamp = -1 - async def add_word_timestamps(self, word_times: List[Tuple[str, float]]): + async def add_word_timestamps( + self, word_times: List[Tuple[str, float]], context_id: Optional[str] = None + ): """Add word timestamps to the processing queue. Args: word_times: List of (word, timestamp) tuples where timestamp is in seconds. + context_id: Unique identifier for the TTS context. """ + # Transform to include context_id in each tuple + word_times_with_context = [(word, timestamp, context_id) for word, timestamp in word_times] + if self._initial_word_timestamp == -1: # Cache word timestamps and don't add them until we have started # (i.e. we have some audio). - self._initial_word_times.extend(word_times) + self._initial_word_times.extend(word_times_with_context) else: - await self._add_word_timestamps(word_times) + await self._add_word_timestamps(word_times_with_context) async def start(self, frame: StartFrame): """Start the word TTS service. @@ -790,15 +862,15 @@ class WordTTSService(TTSService): await self.cancel_task(self._words_task) self._words_task = None - async def _add_word_timestamps(self, word_times: List[Tuple[str, float]]): - for word, timestamp in word_times: - await self._words_queue.put((word, seconds_to_nanoseconds(timestamp))) + async def _add_word_timestamps(self, word_times_with_context: List[Tuple[str, float, str]]): + for word, timestamp, context_id in word_times_with_context: + await self._words_queue.put((word, seconds_to_nanoseconds(timestamp), context_id)) async def _words_task_handler(self): last_pts = 0 while True: frame = None - (word, timestamp) = await self._words_queue.get() + (word, timestamp, context_id) = await self._words_queue.get() if word == "Reset" and timestamp == 0: await self.reset_word_timestamps() if self._llm_response_started: @@ -808,11 +880,16 @@ class WordTTSService(TTSService): elif word == "TTSStoppedFrame" and timestamp == 0: frame = TTSStoppedFrame() frame.pts = last_pts + frame.context_id = context_id else: # Assumption: word-by-word text frames don't include spaces, so # we can rely on the default includes_inter_frame_spaces=False frame = TTSTextFrame(word, aggregated_by=AggregationType.WORD) frame.pts = self._initial_word_timestamp + timestamp + frame.context_id = context_id + # Look up append_to_context from context metadata + if context_id in self._tts_contexts: + frame.append_to_context = self._tts_contexts[context_id].append_to_context if frame: last_pts = frame.pts await self.push_frame(frame) 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)