FrameProcessor: introduce direct mode

Direct mode avoids creating internal queues and tasks and processes frames right
away. This might be useful for some very simple processors.

src/pipecat/pipeline/parallel_pipeline.py

@@ -49,7 +49,7 @@ class ParallelPipelineSource(FrameProcessor):
             upstream_queue: Queue for collecting upstream frames from this branch.
             push_frame_func: Function to push frames to the parent parallel pipeline.
         """
-        super().__init__()
+        super().__init__(enable_direct_mode=True)
         self._up_queue = upstream_queue
         self._push_frame_func = push_frame_func
 
@@ -90,7 +90,7 @@ class ParallelPipelineSink(FrameProcessor):
             downstream_queue: Queue for collecting downstream frames from this branch.
             push_frame_func: Function to push frames to the parent parallel pipeline.
         """
-        super().__init__()
+        super().__init__(enable_direct_mode=True)
         self._down_queue = downstream_queue
         self._push_frame_func = push_frame_func
 

src/pipecat/pipeline/pipeline.py

@@ -32,7 +32,7 @@ class PipelineSource(FrameProcessor):
         Args:
             upstream_push_frame: Coroutine function to handle upstream frames.
         """
-        super().__init__()
+        super().__init__(enable_direct_mode=True)
         self._upstream_push_frame = upstream_push_frame
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -65,7 +65,7 @@ class PipelineSink(FrameProcessor):
         Args:
             downstream_push_frame: Coroutine function to handle downstream frames.
         """
-        super().__init__()
+        super().__init__(enable_direct_mode=True)
         self._downstream_push_frame = downstream_push_frame
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):

src/pipecat/pipeline/sync_parallel_pipeline.py

@@ -49,7 +49,7 @@ class SyncParallelPipelineSource(FrameProcessor):
         Args:
             upstream_queue: Queue for collecting upstream frames from the pipeline.
         """
-        super().__init__()
+        super().__init__(enable_direct_mode=True)
         self._up_queue = upstream_queue
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -81,7 +81,7 @@ class SyncParallelPipelineSink(FrameProcessor):
         Args:
             downstream_queue: Queue for collecting downstream frames from the pipeline.
         """
-        super().__init__()
+        super().__init__(enable_direct_mode=True)
         self._down_queue = downstream_queue
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):

src/pipecat/pipeline/task.py

@@ -110,14 +110,14 @@ class PipelineTaskSource(FrameProcessor):
     pipeline.
     """
 
-    def __init__(self, up_queue: asyncio.Queue, **kwargs):
+    def __init__(self, up_queue: asyncio.Queue):
         """Initialize the pipeline task source.
 
         Args:
             up_queue: Queue for upstream frame processing.
             **kwargs: Additional arguments passed to the parent class.
         """
-        super().__init__(**kwargs)
+        super().__init__(enable_direct_mode=True)
         self._up_queue = up_queue
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -144,14 +144,14 @@ class PipelineTaskSink(FrameProcessor):
     act on them, for example, waiting to receive an EndFrame.
     """
 
-    def __init__(self, down_queue: asyncio.Queue, **kwargs):
+    def __init__(self, down_queue: asyncio.Queue):
         """Initialize the pipeline task sink.
 
         Args:
             down_queue: Queue for downstream frame processing.
             **kwargs: Additional arguments passed to the parent class.
         """
-        super().__init__(**kwargs)
+        super().__init__(enable_direct_mode=True)
         self._down_queue = down_queue
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):

src/pipecat/processors/frame_processor.py

@@ -175,6 +175,7 @@ class FrameProcessor(BaseObject):
         self,
         *,
         name: Optional[str] = None,
+        enable_direct_mode: bool = False,
         enable_watchdog_logging: Optional[bool] = None,
         enable_watchdog_timers: Optional[bool] = None,
         metrics: Optional[FrameProcessorMetrics] = None,
@@ -185,6 +186,7 @@ class FrameProcessor(BaseObject):
 
         Args:
             name: Optional name for this processor instance.
+            enable_direct_mode: Whether to process frames immediately or use internal queues.
             enable_watchdog_logging: Whether to enable watchdog logging for tasks.
             enable_watchdog_timers: Whether to enable watchdog timers for tasks.
             metrics: Optional metrics collector for this processor.
@@ -196,6 +198,9 @@ class FrameProcessor(BaseObject):
         self._prev: Optional["FrameProcessor"] = None
         self._next: Optional["FrameProcessor"] = None
 
+        # Enable direct mode to skip queues and process frames right away.
+        self._enable_direct_mode = enable_direct_mode
+
         # Enable watchdog timers for all tasks created by this frame processor.
         self._enable_watchdog_timers = enable_watchdog_timers
 
@@ -558,7 +563,10 @@ class FrameProcessor(BaseObject):
         if self._cancelling:
             return
 
-        await self.__input_queue.put((frame, direction, callback))
+        if self._enable_direct_mode:
+            await self.__process_frame(frame, direction, callback)
+        else:
+            await self.__input_queue.put((frame, direction, callback))
 
     async def pause_processing_frames(self):
         """Pause processing of queued frames."""
@@ -730,6 +738,9 @@ class FrameProcessor(BaseObject):
 
     def __create_input_task(self):
         """Create the frame input processing task."""
+        if self._enable_direct_mode:
+            return
+
         if not self.__input_frame_task:
             self.__input_queue = FrameProcessorQueue(self.task_manager)
             self.__input_frame_task = self.create_task(self.__input_frame_task_handler())
@@ -743,6 +754,9 @@ class FrameProcessor(BaseObject):
 
     def __create_process_task(self):
         """Create the non-system frame processing task."""
+        if self._enable_direct_mode:
+            return
+
         if not self.__process_frame_task:
             self.__should_block_frames = False
             if not self.__process_event:
@@ -759,7 +773,7 @@ class FrameProcessor(BaseObject):
             self.__process_frame_task = None
 
     async def __process_frame(
-        self, frame: Frame, direction: FrameDirection, callback: FrameCallback
+        self, frame: Frame, direction: FrameDirection, callback: Optional[FrameCallback]
     ):
         try:
             # Process the frame.