tts: handle incoming frames pausing/resuming from base TTSService class

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()

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

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()
 

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()

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

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)