From 6a24457f0e3f00fc6fa5be181a4bac4fac7334a2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 8 Aug 2025 18:18:12 -0700 Subject: [PATCH] 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. --- CHANGELOG.md | 20 +++++++++++++++++++ src/pipecat/pipeline/parallel_pipeline.py | 4 ++-- src/pipecat/pipeline/pipeline.py | 4 ++-- .../pipeline/sync_parallel_pipeline.py | 4 ++-- src/pipecat/pipeline/task.py | 8 ++++---- src/pipecat/processors/frame_processor.py | 18 +++++++++++++++-- 6 files changed, 46 insertions(+), 12 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ec717db58..9e2fb3781 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `enable_direct_mode` argument to `FrameProcessor`. The direct mode is + for processors which require very little I/O or compute resources, that is + processors that can perform their task almost immediately. These type of + processors don't need any of the internal tasks and queues usually created by + frame processors which means overall application performance might be slightly + increased. Use with care. + - Added TTFB metrics for `HeyGenVideoService` and `TavusVideoService`. - Added `endpoint_id` parameter to `AzureSTTService`. ([Custom EndpointId](https://docs.azure.cn/en-us/ai-services/speech-service/how-to-recognize-speech?pivots=programming-language-python#use-a-custom-endpoint)) @@ -18,6 +25,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Updated `pyproject.toml` to once again pin `numba` to `>=0.61.2` in order to resolve package versioning issues. +### Performance + +- Improved some frame processors performance by using the new frame processor + direct mode. In direct mode a frame processor will process frames right away + avoiding the need for internal queues and tasks. This is useful for some + simple processors. For example, in processors that wrap other processors + (e.g. `Pipeline`, `ParallelPipeline`), we add one processor before and one + after the wrapped processors (internally, you will see them as sources and + sinks). These sources and sinks don't do any special processing and they + basically forward frames. So, for these simple processors we now enable the + new direct mode which avoids creating any internal tasks (and queues) and + therefore improves performance. + ### Other - Improving the latency of the `HeyGenVideoService`. diff --git a/src/pipecat/pipeline/parallel_pipeline.py b/src/pipecat/pipeline/parallel_pipeline.py index 103736928..cb66f62b5 100644 --- a/src/pipecat/pipeline/parallel_pipeline.py +++ b/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 diff --git a/src/pipecat/pipeline/pipeline.py b/src/pipecat/pipeline/pipeline.py index ee96dd7fc..5d9dd91c0 100644 --- a/src/pipecat/pipeline/pipeline.py +++ b/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): diff --git a/src/pipecat/pipeline/sync_parallel_pipeline.py b/src/pipecat/pipeline/sync_parallel_pipeline.py index 6454c3929..366da824a 100644 --- a/src/pipecat/pipeline/sync_parallel_pipeline.py +++ b/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): diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 1d9d642e0..6bc0c3d1e 100644 --- a/src/pipecat/pipeline/task.py +++ b/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): diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 0f18740bf..e780d5a6f 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/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.