processors: introduce input queues

Frame processors can now decide if they should continue processing frames or
not, and if so also decide when to continue processing frames. For example,
asynchronous TTS services will stop processing frames until they have generated
all the audio for an LLM response.

src/pipecat/processors/frame_processor.py

@@ -36,6 +36,7 @@ class FrameProcessor:
         self,
         *,
         name: str | None = None,
+        block_on_frames: tuple = (),
         metrics: FrameProcessorMetrics | None = None,
         loop: asyncio.AbstractEventLoop | None = None,
         **kwargs,
@@ -62,6 +63,13 @@ class FrameProcessor:
         self._metrics = metrics or FrameProcessorMetrics()
         self._metrics.set_processor_name(self.name)
 
+        # Processors have an input queue. The input queue can be processed
+        # immediately (default) or it can be processed when the processor asks
+        # to process a new frame (via process_next_frame()).
+        self._block_on_frames = block_on_frames
+        self._should_queue_input_frames = False
+        self.__create_input_task()
+
         # Every processor in Pipecat should only output frames from a single
         # task. This avoid problems like audio overlapping. System frames are
         # the exception to this rule. This create this task.
@@ -145,6 +153,19 @@ class FrameProcessor:
     def get_clock(self) -> BaseClock:
         return self._clock
 
+    async def queue_frame(
+        self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM
+    ):
+        if isinstance(frame, SystemFrame):
+            # We don't want to queue system frames.
+            await self.process_frame(frame, direction)
+        else:
+            # We queue everything else.
+            await self.__input_queue.put((frame, direction))
+
+    async def resume_processing_frames(self):
+        self.__input_event.set()
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         if isinstance(frame, StartFrame):
             self._clock = frame.clock
@@ -204,13 +225,46 @@ class FrameProcessor:
         try:
             if direction == FrameDirection.DOWNSTREAM and self._next:
                 logger.trace(f"Pushing {frame} from {self} to {self._next}")
-                await self._next.process_frame(frame, direction)
+                await self._next.queue_frame(frame, direction)
             elif direction == FrameDirection.UPSTREAM and self._prev:
                 logger.trace(f"Pushing {frame} upstream from {self} to {self._prev}")
-                await self._prev.process_frame(frame, direction)
+                await self._prev.queue_frame(frame, direction)
         except Exception as e:
             logger.exception(f"Uncaught exception in {self}: {e}")
 
+    def __create_input_task(self):
+        self.__input_queue = asyncio.Queue()
+        self.__input_frame_task = self.get_event_loop().create_task(
+            self.__input_frame_task_handler()
+        )
+        self.__input_event = asyncio.Event()
+
+    async def __input_frame_task_handler(self):
+        running = True
+        should_block_frames = False
+        while running:
+            try:
+                if should_block_frames:
+                    await self.__input_event.wait()
+                    self.__input_event.clear()
+                    should_block_frames = False
+
+                (frame, direction) = await self.__input_queue.get()
+
+                # Process the frame.
+                await self.process_frame(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()
+            except asyncio.CancelledError:
+                break
+
     def __create_push_task(self):
         self.__push_queue = asyncio.Queue()
         self.__push_frame_task = self.get_event_loop().create_task(self.__push_frame_task_handler())