From 80510178950cec53806d211f46fd26e5a2d34394 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 15 Aug 2025 10:18:14 -0700 Subject: [PATCH] pipeline: wrap with pipelines, use direct mode and reduce tasks --- CHANGELOG.md | 3 ++ src/pipecat/pipeline/parallel_pipeline.py | 26 ++++---------- src/pipecat/pipeline/pipeline.py | 4 ++- .../pipeline/sync_parallel_pipeline.py | 14 +++----- src/pipecat/pipeline/task.py | 34 +++++++------------ 5 files changed, 30 insertions(+), 51 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 754469167..2986e15d1 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/pipecat/pipeline/parallel_pipeline.py b/src/pipecat/pipeline/parallel_pipeline.py index cb66f62b5..b2c20e74e 100644 --- a/src/pipecat/pipeline/parallel_pipeline.py +++ b/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) - source.link(pipeline) - pipeline.link(sink) + pipeline = Pipeline([source] + processors + [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: - # If we get an upstream frame we process it in each sink. - await asyncio.gather(*[s.queue_frame(frame, direction) for s in self._sinks]) - 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]) + # Process frames in each of the sub-pipelines. + for p in self._pipelines: + await p.queue_frame(frame, direction) # Handle interruptions after everything has been cancelled. if isinstance(frame, StartInterruptionFrame): diff --git a/src/pipecat/pipeline/pipeline.py b/src/pipecat/pipeline/pipeline.py index 9a88c8426..227fc7139 100644 --- a/src/pipecat/pipeline/pipeline.py +++ b/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) diff --git a/src/pipecat/pipeline/sync_parallel_pipeline.py b/src/pipecat/pipeline/sync_parallel_pipeline.py index 366da824a..c4cffe112 100644 --- a/src/pipecat/pipeline/sync_parallel_pipeline.py +++ b/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. diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 6bc0c3d1e..5ad22077e 100644 --- a/src/pipecat/pipeline/task.py +++ b/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 - # pipeline. This source processor allows up to receive and react to - # upstream frames. - self._source = PipelineTaskSource(self._up_queue) - self._source.link(pipeline) - - # This is a sink processor that we connect to the provided - # pipeline. This sink processor allows up to receive and react to - # downstream frames. - self._sink = PipelineTaskSink(self._down_queue) - pipeline.link(self._sink) + # This is the final pipeline. It is composed of a source processor, + # followed by the user pipeline, and ending with a sink processor. The + # source allows us to receive and react to upstream frames, and the sink + # allows us to receive and react to downstream frames. + source = PipelineTaskSource(self._up_queue) + sink = PipelineTaskSink(self._down_queue) + self._pipeline = Pipeline([source, pipeline, 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):