Merge pull request #2012 from pipecat-ai/aleix/frame-pause-resume-frames

FrameProcessor: handle new FrameProcessorPauseFrame/FrameProcessorResumeFrame

CHANGELOG.md

@@ -9,6 +9,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added new frames `FrameProcessorPauseFrame` and `FrameProcessorResumeFrame`
+  which allow pausing and resuming frame processing for a given frame
+  processor. These are control frames, so they are ordered. Pausing frame
+  processor will keep old frames in the internal queues until resume takes
+  place. Frames being pushed while a frame processor is paused will be pushed to
+  the queues. When frame processing is resumed all queued frames will be
+  processed in order. Also added `FrameProcessorPauseUrgentFrame` and
+  `FrameProcessorResumeUrgentFrame` which are system frames and therefore they
+  have high priority.
+
 - Added a property called `has_function_calls_in_progress` in
   `LLMAssistantContextAggregator` that exposes whether a function call is in
   progress.
@@ -18,8 +28,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Fixed an issue with `GroqTTSService` where it was not properly parsing the
   WAV file header.
 
-- Fixed an issue with `GoogleSTTService` where it was constantly reconnecting
-
 - Fixed an issue with `GoogleSTTService` where it was constantly reconnecting
   before starting to receive audio from the user.
 

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
     #