pipeline: wrap with pipelines, use direct mode and reduce tasks

CHANGELOG.md

@@ -40,6 +40,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Performance
 
+- Improve `ParallelPipeline` performance by using direct mode and by not
+  creating a task for each frame and every sub-pipeline.
+
 - `Pipeline` performance improvements by using direct mode.
 
 ### Other

src/pipecat/pipeline/parallel_pipeline.py

@@ -132,14 +132,12 @@ class ParallelPipeline(BasePipeline):
             Exception: If no processor lists are provided.
             TypeError: If any argument is not a list of processors.
         """
-        super().__init__()
+        super().__init__(enable_direct_mode=True)
 
         if len(args) == 0:
             raise Exception(f"ParallelPipeline needs at least one argument")
 
         self._args = args
-        self._sources = []
-        self._sinks = []
         self._pipelines = []
 
         self._seen_ids = set()
@@ -185,30 +183,23 @@ class ParallelPipeline(BasePipeline):
             if not isinstance(processors, list):
                 raise TypeError(f"ParallelPipeline argument {processors} is not a list")
 
-            # We will add a source before the pipeline and a sink after.
+            # We add a source before the pipeline and a sink after so we control
+            # the frames that are pushed upstream and downstream.
             source = ParallelPipelineSource(self._up_queue, self._parallel_push_frame)
             sink = ParallelPipelineSink(self._down_queue, self._pipeline_sink_push_frame)
-            self._sources.append(source)
-            self._sinks.append(sink)
 
             # Create pipeline
-            pipeline = Pipeline(processors)
+            pipeline = Pipeline([source] + processors + [sink])
-            source.link(pipeline)
-            pipeline.link(sink)
             self._pipelines.append(pipeline)
 
         logger.debug(f"Finished creating {self} pipelines")
 
-        await asyncio.gather(*[s.setup(setup) for s in self._sources])
         await asyncio.gather(*[p.setup(setup) for p in self._pipelines])
-        await asyncio.gather(*[s.setup(setup) for s in self._sinks])
 
     async def cleanup(self):
         """Clean up the parallel pipeline and all its branches."""
         await super().cleanup()
-        await asyncio.gather(*[s.cleanup() for s in self._sources])
         await asyncio.gather(*[p.cleanup() for p in self._pipelines])
-        await asyncio.gather(*[s.cleanup() for s in self._sinks])
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         """Process frames through all parallel branches with lifecycle coordination.
@@ -226,12 +217,9 @@ class ParallelPipeline(BasePipeline):
         elif isinstance(frame, CancelFrame):
             await self._cancel()
 
-        if direction == FrameDirection.UPSTREAM:
+        # Process frames in each of the sub-pipelines.
-            # If we get an upstream frame we process it in each sink.
+        for p in self._pipelines:
-            await asyncio.gather(*[s.queue_frame(frame, direction) for s in self._sinks])
+            await p.queue_frame(frame, direction)
-        elif direction == FrameDirection.DOWNSTREAM:
-            # If we get a downstream frame we process it in each source.
-            await asyncio.gather(*[s.queue_frame(frame, direction) for s in self._sources])
 
         # Handle interruptions after everything has been cancelled.
         if isinstance(frame, StartInterruptionFrame):

src/pipecat/pipeline/pipeline.py

@@ -11,7 +11,7 @@ in sequence and manages frame flow between them, along with helper classes
 for pipeline source and sink operations.
 """
 
-from typing import Callable, Coroutine, List
+from typing import Callable, Coroutine, List, Optional
 
 from pipecat.frames.frames import Frame
 from pipecat.pipeline.base_pipeline import BasePipeline
@@ -97,6 +97,8 @@ class Pipeline(BasePipeline):
 
         Args:
             processors: List of frame processors to connect in sequence.
+            source: An optional custom source processor.
+            sink: An optional custom sink processor.
         """
         super().__init__(enable_direct_mode=True)
 

src/pipecat/pipeline/sync_parallel_pipeline.py

@@ -171,29 +171,23 @@ class SyncParallelPipeline(BasePipeline):
             source = SyncParallelPipelineSource(up_queue)
             sink = SyncParallelPipelineSink(down_queue)
 
-            # Create pipeline
-            pipeline = Pipeline(processors)
-            source.link(pipeline)
-            pipeline.link(sink)
-            self._pipelines.append(pipeline)
-
             # Keep track of sources and sinks. We also keep the output queue of
             # the source and the sinks so we can use it later.
             self._sources.append({"processor": source, "queue": down_queue})
             self._sinks.append({"processor": sink, "queue": up_queue})
 
+            # Create pipeline
+            pipeline = Pipeline(processors, source=source, sink=sink)
+            self._pipelines.append(pipeline)
+
         logger.debug(f"Finished creating {self} pipelines")
 
-        await asyncio.gather(*[s["processor"].setup(setup) for s in self._sources])
         await asyncio.gather(*[p.setup(setup) for p in self._pipelines])
-        await asyncio.gather(*[s["processor"].setup(setup) for s in self._sinks])
 
     async def cleanup(self):
         """Clean up the parallel pipeline and all contained processors."""
         await super().cleanup()
-        await asyncio.gather(*[s["processor"].cleanup() for s in self._sources])
         await asyncio.gather(*[p.cleanup() for p in self._pipelines])
-        await asyncio.gather(*[s["processor"].cleanup() for s in self._sinks])
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         """Process frames through all parallel pipelines with synchronization.

src/pipecat/pipeline/task.py

@@ -43,6 +43,7 @@ from pipecat.observers.base_observer import BaseObserver
 from pipecat.observers.turn_tracking_observer import TurnTrackingObserver
 from pipecat.pipeline.base_pipeline import BasePipeline
 from pipecat.pipeline.base_task import BasePipelineTask, PipelineTaskParams
+from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.task_observer import TaskObserver
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
 from pipecat.utils.asyncio.task_manager import (
@@ -244,7 +245,6 @@ class PipelineTask(BasePipelineTask):
                 will be logged if the watchdog timer is not reset before this timeout.
         """
         super().__init__()
-        self._pipeline = pipeline
         self._params = params or PipelineParams()
         self._additional_span_attributes = additional_span_attributes or {}
         self._cancel_on_idle_timeout = cancel_on_idle_timeout
@@ -311,17 +311,13 @@ class PipelineTask(BasePipelineTask):
         # StopFrame) has been received in the down queue.
         self._pipeline_end_event = asyncio.Event()
 
-        # This is a source processor that we connect to the provided
+        # This is the final pipeline. It is composed of a source processor,
-        # pipeline. This source processor allows up to receive and react to
+        # followed by the user pipeline, and ending with a sink processor. The
-        # upstream frames.
+        # source allows us to receive and react to upstream frames, and the sink
-        self._source = PipelineTaskSource(self._up_queue)
+        # allows us to receive and react to downstream frames.
-        self._source.link(pipeline)
+        source = PipelineTaskSource(self._up_queue)
-
+        sink = PipelineTaskSink(self._down_queue)
-        # This is a sink processor that we connect to the provided
+        self._pipeline = Pipeline([source, pipeline, sink])
-        # pipeline. This sink processor allows up to receive and react to
-        # downstream frames.
-        self._sink = PipelineTaskSink(self._down_queue)
-        pipeline.link(self._sink)
 
         # The task observer acts as a proxy to the provided observers. This way,
         # we only need to pass a single observer (using the StartFrame) which
@@ -494,7 +490,7 @@ class PipelineTask(BasePipelineTask):
             # Make sure everything is cleaned up downstream. This is sent
             # out-of-band from the main streaming task which is what we want since
             # we want to cancel right away.
-            await self._source.push_frame(CancelFrame())
+            await self._pipeline.queue_frame(CancelFrame())
             # Wait for CancelFrame to make it throught the pipeline.
             await self._wait_for_pipeline_end()
             # Only cancel the push task, we don't want to be able to process any
@@ -606,9 +602,7 @@ class PipelineTask(BasePipelineTask):
             observer=self._observer,
             watchdog_timers_enabled=self._enable_watchdog_timers,
         )
-        await self._source.setup(setup)
         await self._pipeline.setup(setup)
-        await self._sink.setup(setup)
 
     async def _cleanup(self, cleanup_pipeline: bool):
         """Clean up the pipeline task and processors."""
@@ -620,10 +614,8 @@ class PipelineTask(BasePipelineTask):
             self._turn_trace_observer.end_conversation_tracing()
 
         # Cleanup pipeline processors.
-        await self._source.cleanup()
         if cleanup_pipeline:
             await self._pipeline.cleanup()
-        await self._sink.cleanup()
 
     async def _process_push_queue(self):
         """Process frames from the push queue and send them through the pipeline.
@@ -647,16 +639,16 @@ class PipelineTask(BasePipelineTask):
             interruption_strategies=self._params.interruption_strategies,
         )
         start_frame.metadata = self._params.start_metadata
-        await self._source.queue_frame(start_frame, FrameDirection.DOWNSTREAM)
+        await self._pipeline.queue_frame(start_frame)
 
         if self._params.enable_metrics and self._params.send_initial_empty_metrics:
-            await self._source.queue_frame(self._initial_metrics_frame(), FrameDirection.DOWNSTREAM)
+            await self._pipeline.queue_frame(self._initial_metrics_frame())
 
         running = True
         cleanup_pipeline = True
         while running:
             frame = await self._push_queue.get()
-            await self._source.queue_frame(frame, FrameDirection.DOWNSTREAM)
+            await self._pipeline.queue_frame(frame)
             if isinstance(frame, (CancelFrame, EndFrame, StopFrame)):
                 await self._wait_for_pipeline_end()
             running = not isinstance(frame, (CancelFrame, EndFrame, StopFrame))
@@ -741,7 +733,7 @@ class PipelineTask(BasePipelineTask):
             # Don't use `queue_frame()` because if an EndFrame is queued the
             # task will just stop waiting for the pipeline to finish not
             # allowing more frames to be pushed.
-            await self._source.queue_frame(HeartbeatFrame(timestamp=self._clock.get_time()))
+            await self._pipeline.queue_frame(HeartbeatFrame(timestamp=self._clock.get_time()))
             await asyncio.sleep(self._params.heartbeats_period_secs)
 
     async def _heartbeat_monitor_handler(self):