From 241f241ed931a0e201ad5c7bf72af6c7e77af605 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 30 Jan 2025 13:11:36 -0800 Subject: [PATCH] SyncParallelPipeline: don't add source/sink processors inside pipeline --- src/pipecat/pipeline/sync_parallel_pipeline.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/src/pipecat/pipeline/sync_parallel_pipeline.py b/src/pipecat/pipeline/sync_parallel_pipeline.py index 7fa3577a5..083e992ca 100644 --- a/src/pipecat/pipeline/sync_parallel_pipeline.py +++ b/src/pipecat/pipeline/sync_parallel_pipeline.py @@ -78,16 +78,18 @@ class SyncParallelPipeline(BasePipeline): down_queue = asyncio.Queue() source = SyncParallelPipelineSource(up_queue) sink = SyncParallelPipelineSink(down_queue) - processors: List[FrameProcessor] = [source] + processors + [sink] + + # 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) - self._pipelines.append(pipeline) logger.debug(f"Finished creating {self} pipelines") #