processors: remove block_on_frames and add pause_processing_frames() instead

CHANGELOG.md

@@ -14,12 +14,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `FrameProcessor.queue_frame()` on the next processor (upstream or downstream)
   and the frame will be internally queued (except system frames). Then, the
   queued frames will get processed. With this input queue it is also possible
-  for FrameProcessors to block processing more frames via a list of blocking
+  for FrameProcessors to block processing more frames by calling
-  frames. For example, some TTS services now block processing more frames if
+  `FrameProcessor.pause_processing_frames()`. The way to resume processing
-  they see a `TTSSpeakFrame` or a `LLMFullResponseEndFrame` until the bot has
+  frames is by calling `FrameProcessor.resume_processing_frames()`.
-  stopped speaking. This makes sure we don't mix audio from different
-  sentences. The way to resume processing frames is by calling
-  `FrameProcessor.resume_processing_frames()`.
 
 - Added audio filter `NoisereduceFilter`.
 

src/pipecat/processors/frame_processor.py

@@ -37,7 +37,6 @@ class FrameProcessor:
         self,
         *,
         name: str | None = None,
-        block_on_frames: tuple = (),
         metrics: FrameProcessorMetrics | None = None,
         loop: asyncio.AbstractEventLoop | None = None,
         **kwargs,
@@ -65,10 +64,10 @@ class FrameProcessor:
         self._metrics.set_processor_name(self.name)
 
         # Processors have an input queue. The input queue will be processed
-        # immediately (default) or it will block if one of the `block_on_frames`
+        # immediately (default) or it will block if `pause_processing_frames()`
-        # is found. To resume processing frames we need to call
+        # is called. To resume processing frames we need to call
         # `resume_processing_frames()`.
-        self._block_on_frames = block_on_frames
+        self.__should_block_frames = False
         self.__create_input_task()
 
         # Every processor in Pipecat should only output frames from a single
@@ -170,8 +169,12 @@ class FrameProcessor:
             # We queue everything else.
             await self.__input_queue.put((frame, direction, callback))
 
+    async def pause_processing_frames(self):
+        self.__should_block_frames = True
+
     async def resume_processing_frames(self):
         self.__input_event.set()
+        self.__should_block_frames = False
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         if isinstance(frame, StartFrame):
@@ -257,13 +260,11 @@ class FrameProcessor:
 
     async def __input_frame_task_handler(self):
         running = True
-        should_block_frames = False
         while running:
             try:
-                if should_block_frames:
+                if self.__should_block_frames:
                     await self.__input_event.wait()
                     self.__input_event.clear()
-                    should_block_frames = False
 
                 (frame, direction, callback) = await self.__input_queue.get()
 
@@ -274,11 +275,6 @@ class FrameProcessor:
                 if callback:
                     await callback(self, frame, direction)
 
-                # Check if we should block incoming frames from now on (until we
-                # resume processing).
-                if isinstance(frame, self._block_on_frames):
-                    should_block_frames = True
-
                 running = not isinstance(frame, EndFrame)
 
                 self.__input_queue.task_done()

src/pipecat/services/ai_services.py

@@ -15,7 +15,6 @@ from loguru import logger
 from pipecat.audio.utils import calculate_audio_volume, exp_smoothing
 from pipecat.frames.frames import (
     AudioRawFrame,
-    BotStoppedSpeakingFrame,
     CancelFrame,
     EndFrame,
     ErrorFrame,
@@ -294,8 +293,6 @@ class TTSService(AIService):
             await self._process_text_frame(frame)
         elif isinstance(frame, StartInterruptionFrame):
             await self._handle_interruption(frame, direction)
-        elif isinstance(frame, BotStoppedSpeakingFrame):
-            await self.resume_processing_frames()
         elif isinstance(frame, (LLMFullResponseEndFrame, EndFrame)):
             sentence = self._current_sentence
             self._current_sentence = ""

src/pipecat/services/cartesia.py

@@ -14,6 +14,7 @@ from loguru import logger
 from pydantic.main import BaseModel
 
 from pipecat.frames.frames import (
+    BotStoppedSpeakingFrame,
     CancelFrame,
     EndFrame,
     ErrorFrame,
@@ -102,7 +103,6 @@ class CartesiaTTSService(WordTTSService):
             aggregate_sentences=True,
             push_text_frames=False,
             sample_rate=sample_rate,
-            block_on_frames=(TTSSpeakFrame, LLMFullResponseEndFrame),
             **kwargs,
         )
 
@@ -261,6 +261,19 @@ class CartesiaTTSService(WordTTSService):
         except Exception as e:
             logger.error(f"{self} exception: {e}")
 
+    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/playht.py

@@ -17,6 +17,7 @@ from loguru import logger
 from pydantic.main import BaseModel
 
 from pipecat.frames.frames import (
+    BotStoppedSpeakingFrame,
     CancelFrame,
     EndFrame,
     ErrorFrame,
@@ -125,7 +126,6 @@ class PlayHTTTSService(TTSService):
     ):
         super().__init__(
             sample_rate=sample_rate,
-            block_on_frames=(TTSSpeakFrame, LLMFullResponseEndFrame),
             **kwargs,
         )
 
@@ -266,6 +266,19 @@ class PlayHTTTSService(TTSService):
         except Exception as e:
             logger.error(f"{self} exception in receive task: {e}")
 
+    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}]")