From 5126d4de92212e4f3addb5bcb205f5a1f5b8559b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 14 Feb 2025 14:34:38 -0800 Subject: [PATCH] tts: handle incoming frames pausing/resuming from base TTSService class --- src/pipecat/services/ai_services.py | 21 +++++++++++++++++++++ src/pipecat/services/cartesia.py | 13 ------------- src/pipecat/services/elevenlabs.py | 13 ------------- src/pipecat/services/fish.py | 10 ---------- src/pipecat/services/playht.py | 13 ------------- src/pipecat/services/rime.py | 18 +----------------- 6 files changed, 22 insertions(+), 66 deletions(-) diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 0fce2482a..7d49c5676 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -15,6 +15,7 @@ from loguru import logger from pipecat.audio.utils import calculate_audio_volume, exp_smoothing from pipecat.frames.frames import ( AudioRawFrame, + BotStoppedSpeakingFrame, CancelFrame, EndFrame, ErrorFrame, @@ -234,6 +235,7 @@ class TTSService(AIService): self._stop_frame_queue: asyncio.Queue = asyncio.Queue() self._current_sentence: str = "" + self._processing_text: bool = False @property def sample_rate(self) -> int: @@ -299,6 +301,7 @@ class TTSService(AIService): async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) + if ( isinstance(frame, TextFrame) and not isinstance(frame, InterimTranscriptionFrame) @@ -308,8 +311,15 @@ class TTSService(AIService): elif isinstance(frame, StartInterruptionFrame): await self._handle_interruption(frame, direction) elif isinstance(frame, (LLMFullResponseEndFrame, EndFrame)): + # We pause processing incoming frames if the LLM response included + # text (it might be that it's only a function calling response). We + # pause to avoid audio overlapping. + if self._processing_text: + await self.pause_processing_frames() + sentence = self._current_sentence self._current_sentence = "" + self._processing_text = False await self._push_tts_frames(sentence) if isinstance(frame, LLMFullResponseEndFrame): if self._push_text_frames: @@ -317,10 +327,16 @@ class TTSService(AIService): else: await self.push_frame(frame, direction) elif isinstance(frame, TTSSpeakFrame): + # We pause processing incoming frames because we are sending data to + # the TTS. We pause to avoid audio overlapping. + await self.pause_processing_frames() await self._push_tts_frames(frame.text) await self.flush_audio() + self._processing_text = False elif isinstance(frame, TTSUpdateSettingsFrame): await self._update_settings(frame.settings) + elif isinstance(frame, BotStoppedSpeakingFrame): + await self.resume_processing_frames() else: await self.push_frame(frame, direction) @@ -371,6 +387,11 @@ class TTSService(AIService): if not text.strip(): return + # This is just a flag that indicates if we sent something to the TTS + # service. It will be cleared if we sent text because of a TTSSpeakFrame + # or when we received an LLMFullResponseEndFrame + self._processing_text = True + await self.start_processing_metrics() if self._text_filter: self._text_filter.reset_interruption() diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 49bcac2f1..becd39f79 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -274,19 +274,6 @@ class CartesiaTTSService(AudioContextWordTTSService, WebsocketService): else: logger.error(f"{self} error, unknown message type: {msg}") - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - # If we received a TTSSpeakFrame and the LLM response included text (it - # might be that it's only a function calling response) we pause - # processing more frames until we receive a BotStoppedSpeakingFrame. - if isinstance(frame, TTSSpeakFrame): - await self.pause_processing_frames() - elif isinstance(frame, LLMFullResponseEndFrame) and self._context_id: - await self.pause_processing_frames() - elif isinstance(frame, BotStoppedSpeakingFrame): - await self.resume_processing_frames() - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index b2047d34e..d8acedd6c 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -289,19 +289,6 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService): if isinstance(frame, TTSStoppedFrame): await self.add_word_timestamps([("LLMFullResponseEndFrame", 0), ("Reset", 0)]) - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - # If we received a TTSSpeakFrame and the LLM response included text (it - # might be that it's only a function calling response) we pause - # processing more frames until we receive a BotStoppedSpeakingFrame. - if isinstance(frame, TTSSpeakFrame): - await self.pause_processing_frames() - elif isinstance(frame, LLMFullResponseEndFrame) and self._started: - await self.pause_processing_frames() - elif isinstance(frame, BotStoppedSpeakingFrame): - await self.resume_processing_frames() - async def _connect(self): await self._connect_websocket() diff --git a/src/pipecat/services/fish.py b/src/pipecat/services/fish.py index d61514eb2..e36eaf497 100644 --- a/src/pipecat/services/fish.py +++ b/src/pipecat/services/fish.py @@ -166,16 +166,6 @@ class FishAudioTTSService(TTSService, WebsocketService): except Exception as e: logger.error(f"Error processing message: {e}") - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - if isinstance(frame, TTSSpeakFrame): - await self.pause_processing_frames() - elif isinstance(frame, LLMFullResponseEndFrame) and self._request_id: - await self.pause_processing_frames() - elif isinstance(frame, BotStoppedSpeakingFrame): - await self.resume_processing_frames() - async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection): await super()._handle_interruption(frame, direction) await self.stop_all_metrics() diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index bd9463f91..3196c306d 100644 --- a/src/pipecat/services/playht.py +++ b/src/pipecat/services/playht.py @@ -269,19 +269,6 @@ class PlayHTTTSService(TTSService, WebsocketService): except json.JSONDecodeError: logger.error(f"Invalid JSON message: {message}") - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - # If we received a TTSSpeakFrame and the LLM response included text (it - # might be that it's only a function calling response) we pause - # processing more frames until we receive a BotStoppedSpeakingFrame. - if isinstance(frame, TTSSpeakFrame): - await self.pause_processing_frames() - elif isinstance(frame, LLMFullResponseEndFrame) and self._request_id: - await self.pause_processing_frames() - elif isinstance(frame, BotStoppedSpeakingFrame): - await self.resume_processing_frames() - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") diff --git a/src/pipecat/services/rime.py b/src/pipecat/services/rime.py index 1634578cd..79e557f48 100644 --- a/src/pipecat/services/rime.py +++ b/src/pipecat/services/rime.py @@ -126,7 +126,6 @@ class RimeTTSService(AudioContextWordTTSService, WebsocketService): # State tracking self._context_id = None # Tracks current turn self._receive_task = None - self._started = False self._cumulative_time = 0 # Accumulates time across messages def can_generate_metrics(self) -> bool: @@ -200,7 +199,6 @@ class RimeTTSService(AudioContextWordTTSService, WebsocketService): await self._websocket.send(json.dumps(self._build_eos_msg())) await self._websocket.close() self._websocket = None - self._started = False self._context_id = None except Exception as e: logger.error(f"{self} error closing websocket: {e}") @@ -217,7 +215,6 @@ class RimeTTSService(AudioContextWordTTSService, WebsocketService): await self.stop_all_metrics() if self._context_id: await self._get_websocket().send(json.dumps(self._build_clear_msg())) - self._started = False self._context_id = None def _calculate_word_times(self, words: list, starts: list, ends: list) -> list: @@ -300,21 +297,9 @@ class RimeTTSService(AudioContextWordTTSService, WebsocketService): """Push frame and handle end-of-turn conditions.""" await super().push_frame(frame, direction) if isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)): - self._started = False if isinstance(frame, TTSStoppedFrame): await self.add_word_timestamps([("LLMFullResponseEndFrame", 0), ("Reset", 0)]) - async def process_frame(self, frame: Frame, direction: FrameDirection): - """Process frames and manage turn state.""" - await super().process_frame(frame, direction) - - if isinstance(frame, TTSSpeakFrame): - await self.pause_processing_frames() - elif isinstance(frame, LLMFullResponseEndFrame) and self._started: - await self.pause_processing_frames() - elif isinstance(frame, BotStoppedSpeakingFrame): - await self.resume_processing_frames() - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: """Generate speech from text. @@ -330,10 +315,9 @@ class RimeTTSService(AudioContextWordTTSService, WebsocketService): await self._connect() try: - if not self._started: + if not self._context_id: await self.start_ttfb_metrics() yield TTSStartedFrame() - self._started = True self._cumulative_time = 0 self._context_id = str(uuid.uuid4()) await self.create_audio_context(self._context_id)