FrameProcessor: handle new FrameProcessorPauseFrame/FrameProcessorResumeFrame

src/pipecat/frames/frames.py

@@ -527,6 +527,29 @@ class StopTaskFrame(SystemFrame):
     pass
 
 
+@dataclass
+class FrameProcessorPauseUrgentFrame(SystemFrame):
+    """This processor is used to pause frame processing for the given processor
+    as fast as possible. Pausing frame processing will keep frames in the
+    internal queue which will then be processed when frame processing is resumed
+    with `FrameProcessorResumeFrame`.
+
+    """
+
+    processor: str
+
+
+@dataclass
+class FrameProcessorResumeUrgentFrame(SystemFrame):
+    """This processor is used to resume frame processing for the given processor
+    if it was previously paused as fast as possible. After resuming frame
+    processing all queued frames will be processed in the order received.
+
+    """
+
+    processor: str
+
+
 @dataclass
 class StartInterruptionFrame(SystemFrame):
     """Emitted by VAD to indicate that a user has started speaking (i.e. is
@@ -854,6 +877,27 @@ class StopFrame(ControlFrame):
     pass
 
 
+@dataclass
+class FrameProcessorPauseFrame(ControlFrame):
+    """This processor is used to pause frame processing for the given
+    processor. Pausing frame processing will keep frames in the internal queue
+    which will then be processed when frame processing is resumed with
+    `FrameProcessorResumeFrame`."""
+
+    processor: str
+
+
+@dataclass
+class FrameProcessorResumeFrame(ControlFrame):
+    """This processor is used to resume frame processing for the given processor
+    if it was previously paused. After resuming frame processing all queued
+    frames will be processed in the order received.
+
+    """
+
+    processor: str
+
+
 @dataclass
 class LLMFullResponseStartFrame(ControlFrame):
     """Used to indicate the beginning of an LLM response. Following by one or

src/pipecat/processors/frame_processor.py

@@ -17,6 +17,10 @@ from pipecat.frames.frames import (
     CancelFrame,
     ErrorFrame,
     Frame,
+    FrameProcessorPauseFrame,
+    FrameProcessorPauseUrgentFrame,
+    FrameProcessorResumeFrame,
+    FrameProcessorResumeUrgentFrame,
     StartFrame,
     StartInterruptionFrame,
     StopInterruptionFrame,
@@ -259,6 +263,10 @@ class FrameProcessor(BaseObject):
             self._should_report_ttfb = True
         elif isinstance(frame, CancelFrame):
             await self.__cancel(frame)
+        elif isinstance(frame, (FrameProcessorPauseFrame, FrameProcessorPauseUrgentFrame)):
+            await self.__pause(frame)
+        elif isinstance(frame, (FrameProcessorResumeFrame, FrameProcessorResumeUrgentFrame)):
+            await self.__resume(frame)
 
     async def push_error(self, error: ErrorFrame):
         await self.push_frame(error, FrameDirection.UPSTREAM)
@@ -287,6 +295,14 @@ class FrameProcessor(BaseObject):
         await self.__cancel_input_task()
         await self.__cancel_push_task()
 
+    async def __pause(self, frame: FrameProcessorPauseFrame | FrameProcessorPauseUrgentFrame):
+        if frame.name == self.name:
+            await self.pause_processing_frames()
+
+    async def __resume(self, frame: FrameProcessorResumeFrame | FrameProcessorResumeUrgentFrame):
+        if frame.name == self.name:
+            await self.resume_processing_frames()
+
     #
     # Handle interruptions
     #