Add frame_order parameter to SyncParallelPipeline

Adds a FrameOrder enum with ARRIVAL (default, existing behavior) and
PIPELINE (pushes frames in pipeline definition order). This lets callers
guarantee output ordering between parallel pipelines — e.g. ensuring
image frames precede audio frames — without needing a separate reordering
processor downstream.

Updates the 05-sync-speech-and-image example to use FrameOrder.PIPELINE,
removing the ImageBeforeAudioReorderer class entirely.

Author: kompfner

changelog/4029.added.2.md

@@ -0,0 +1 @@
+- Added `frame_order` parameter to `SyncParallelPipeline`. Set `frame_order=FrameOrder.PIPELINE` to push synchronized output frames in pipeline definition order (all frames from the first pipeline, then the second, etc.) instead of the default arrival order.

examples/foundational/05-sync-speech-and-image.py

@@ -12,21 +12,16 @@
 from loguru import logger
 
 from pipecat.frames.frames import (
-    AggregatedTextFrame,
     DataFrame,
     Frame,
     LLMContextFrame,
     LLMFullResponseStartFrame,
     OutputImageRawFrame,
     TextFrame,
-    TTSAudioRawFrame,
-    TTSStartedFrame,
-    TTSStoppedFrame,
-    TTSTextFrame,
 )
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.sync_parallel_pipeline import SyncParallelPipeline
+from pipecat.pipeline.sync_parallel_pipeline import FrameOrder, SyncParallelPipeline
 from pipecat.pipeline.task import PipelineTask
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.sentence import SentenceAggregator
@@ -63,61 +58,6 @@ async def process_frame(self, frame: Frame, direction: FrameDirection):
         await self.push_frame(frame, direction)
 
 
-class ImageBeforeAudioReorderer(FrameProcessor):
-    """Ensures each image frame precedes its corresponding TTS audio frames.
-
-    SyncParallelPipeline guarantees that each image is in the same synchronized
-    batch as its audio, but doesn't guarantee which branch's output comes first.
-    This processor detects when TTS frames arrive before their image and holds
-    them until the image arrives.
-
-    All frames pass through immediately unless we detect an ordering problem:
-    TTS frames arrived without a preceding image for the current batch (identified
-    by context_id). In that case, the TTS frames are held until the next image
-    frame, which is pushed first.
-    """
-
-    def __init__(self):
-        super().__init__()
-        self._held_tts_frames = []
-        self._seen_image = False
-        self._current_context_id = None
-
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        await super().process_frame(frame, direction)
-
-        if isinstance(frame, OutputImageRawFrame):
-            self._seen_image = True
-            if self._held_tts_frames:
-                # Image arrived after TTS frames — push image first, then release held frames.
-                logger.debug("ImageBeforeAudioReorderer: reordered — moved image before audio")
-                await self.push_frame(frame, direction)
-                for f in self._held_tts_frames:
-                    await self.push_frame(f, direction)
-                self._held_tts_frames = []
-            else:
-                logger.debug(
-                    "ImageBeforeAudioReorderer: no reorder needed — image was already first"
-                )
-                await self.push_frame(frame, direction)
-        elif isinstance(
-            frame,
-            (AggregatedTextFrame, TTSStartedFrame, TTSAudioRawFrame, TTSStoppedFrame, TTSTextFrame),
-        ):
-            # A new context_id means a new batch — reset image tracking.
-            context_id = frame.context_id
-            if context_id and context_id != self._current_context_id:
-                self._current_context_id = context_id
-                self._seen_image = False
-
-            if self._seen_image:
-                await self.push_frame(frame, direction)
-            else:
-                self._held_tts_frames.append(frame)
-        else:
-            await self.push_frame(frame, direction)
-
-
 class MonthPrepender(FrameProcessor):
     def __init__(self):
         super().__init__()
@@ -197,22 +137,27 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         # that, each pipeline runs concurrently and `SyncParallelPipeline` will
         # wait for the input frame to be processed.
         #
+        # We use `FrameOrder.PIPELINE` so that each synchronized batch of output
+        # frames is pushed in the order the pipelines are listed: image first,
+        # then audio. This ensures the transport receives the image before the
+        # audio frames it should accompany.
+        #
         # Note that `SyncParallelPipeline` requires the last processor in each
         # of the pipelines to be synchronous. In this case, we use
-        # `CartesiaHttpTTSService` and `FalImageGenService` which make HTTP
+        # `FalImageGenService` and `CartesiaHttpTTSService` which make HTTP
         # requests and wait for the response.
         pipeline = Pipeline(
             [
                 llm,  # LLM
                 sentence_aggregator,  # Aggregates LLM output into full sentences
                 SyncParallelPipeline(  # Run pipelines in parallel aggregating the result
-                    [month_prepender, tts],  # Create "Month: sentence" and output audio
                     [
                         imagegen,  # Generate image
                         MarkImageForPlaybackSync(),  # Mark image as needing sync w/audio during playback
                     ],
+                    [month_prepender, tts],  # Create "Month: sentence" and output audio
+                    frame_order=FrameOrder.PIPELINE,
                 ),
-                ImageBeforeAudioReorderer(),  # Ensure each image precedes its audio (important for playback)
                 transport.output(),  # Transport output
             ]
         )

src/pipecat/pipeline/sync_parallel_pipeline.py

@@ -13,6 +13,7 @@
 
 import asyncio
 from dataclasses import dataclass
+from enum import Enum
 from itertools import chain
 from typing import List
 
@@ -24,6 +25,25 @@
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
 
 
+class FrameOrder(Enum):
+    """Controls the order in which synchronized frames are pushed downstream.
+
+    When multiple parallel pipelines produce output for the same input frame,
+    this setting determines the order in which those output frames are pushed.
+
+    Attributes:
+        ARRIVAL: Frames are pushed in the order they arrive from any pipeline.
+            This is the default and matches the behavior of prior versions.
+        PIPELINE: Frames are pushed in pipeline definition order — all frames
+            from the first pipeline are pushed, then all frames from the second
+            pipeline, and so on. Useful when the relative ordering between
+            pipelines matters (e.g. ensuring image frames precede audio frames).
+    """
+
+    ARRIVAL = "arrival"
+    PIPELINE = "pipeline"
+
+
 @dataclass
 class SyncFrame(ControlFrame):
     """Control frame used to synchronize parallel pipeline processing.
@@ -109,20 +129,30 @@ class SyncParallelPipeline(BasePipeline):
 
     The pipeline uses SyncFrame control frames to coordinate between parallel paths
     and ensure all paths have completed processing before moving to the next frame.
+
+    By default, output frames are pushed in the order they arrive from any pipeline
+    (``FrameOrder.ARRIVAL``). Set ``frame_order=FrameOrder.PIPELINE`` to push frames
+    in pipeline definition order instead — all output from the first pipeline, then
+    the second, and so on.
     """
 
-    def __init__(self, *args):
+    def __init__(self, *args, frame_order: FrameOrder = FrameOrder.ARRIVAL):
         """Initialize the synchronous parallel pipeline.
 
         Args:
-            *args: Variable number of processor lists, each representing a parallel pipeline path.
-                   Each argument should be a list of FrameProcessor instances.
+            *args: Variable number of processor lists, each representing a parallel
+                pipeline path. Each argument should be a list of FrameProcessor instances.
+            frame_order: Controls the order in which synchronized output frames are
+                pushed. ``FrameOrder.ARRIVAL`` (default) pushes frames in the order they arrive.
+                ``FrameOrder.PIPELINE`` pushes all frames from the first pipeline
+                before the second, and so on.
 
         Raises:
             Exception: If no arguments are provided.
             TypeError: If any argument is not a list of processors.
         """
         super().__init__()
+        self._frame_order = frame_order
 
         if len(args) == 0:
             raise Exception(f"SyncParallelPipeline needs at least one argument")
@@ -215,6 +245,11 @@ async def process_frame(self, frame: Frame, direction: FrameDirection):
         to maintain proper ordering and prevent duplicate processing. Uses SyncFrame
         control frames to coordinate between parallel paths.
 
+        When ``frame_order`` is ``FrameOrder.ARRIVAL``, output frames are pushed in
+        the order they arrive from any pipeline (via a shared queue). When it is
+        ``FrameOrder.PIPELINE``, each pipeline collects its output into a separate
+        list and the lists are drained in pipeline definition order.
+
         Args:
             frame: The frame to process.
             direction: The direction of frame flow.
@@ -235,60 +270,88 @@ async def process_frame(self, frame: Frame, direction: FrameDirection):
             await self.push_frame(frame, direction)
             return
 
+        use_pipeline_order = self._frame_order == FrameOrder.PIPELINE
+
         # The last processor of each pipeline needs to be synchronous otherwise
-        # this element won't work. Since, we know it should be synchronous we
+        # this element won't work. Since we know it should be synchronous we
         # push a SyncFrame. Since frames are ordered we know this frame will be
         # pushed after the synchronous processor has pushed its data allowing us
         # to synchronize all the internal pipelines by waiting for the
         # SyncFrame in all of them.
+        #
+        # In ARRIVAL mode, output frames are put onto a shared main_queue as
+        # they arrive. In PIPELINE mode, they are accumulated in a per-pipeline
+        # list and returned so the caller can drain them in definition order.
         async def wait_for_sync(
             obj, main_queue: asyncio.Queue, frame: Frame, direction: FrameDirection
-        ):
+        ) -> list[Frame]:
             processor = obj["processor"]
             queue = obj["queue"]
+            output_frames: list[Frame] = []
 
             await processor.process_frame(frame, direction)
 
             if isinstance(frame, EndFrame):
                 new_frame = await queue.get()
                 if isinstance(new_frame, EndFrame):
-                    await main_queue.put(new_frame)
+                    if use_pipeline_order:
+                        output_frames.append(new_frame)
+                    else:
+                        await main_queue.put(new_frame)
                 else:
                     while not isinstance(new_frame, EndFrame):
-                        await main_queue.put(new_frame)
+                        if use_pipeline_order:
+                            output_frames.append(new_frame)
+                        else:
+                            await main_queue.put(new_frame)
                         queue.task_done()
                         new_frame = await queue.get()
             else:
                 await processor.process_frame(SyncFrame(), direction)
                 new_frame = await queue.get()
                 while not isinstance(new_frame, SyncFrame):
-                    await main_queue.put(new_frame)
+                    if use_pipeline_order:
+                        output_frames.append(new_frame)
+                    else:
+                        await main_queue.put(new_frame)
                     queue.task_done()
                     new_frame = await queue.get()
 
+            return output_frames
+
         if direction == FrameDirection.UPSTREAM:
             # If we get an upstream frame we process it in each sink.
-            await asyncio.gather(
+            frames_per_pipeline = await asyncio.gather(
                 *[wait_for_sync(s, self._up_queue, 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(
+            frames_per_pipeline = await asyncio.gather(
                 *[wait_for_sync(s, self._down_queue, frame, direction) for s in self._sources]
             )
 
-        seen_ids = set()
-        while not self._up_queue.empty():
-            frame = await self._up_queue.get()
-            if frame.id not in seen_ids:
-                await self.push_frame(frame, FrameDirection.UPSTREAM)
-                seen_ids.add(frame.id)
-            self._up_queue.task_done()
-
-        seen_ids = set()
-        while not self._down_queue.empty():
-            frame = await self._down_queue.get()
-            if frame.id not in seen_ids:
-                await self.push_frame(frame, FrameDirection.DOWNSTREAM)
-                seen_ids.add(frame.id)
-            self._down_queue.task_done()
+        if use_pipeline_order:
+            # Push frames in pipeline definition order, deduplicating by id.
+            seen_ids = set()
+            for pipeline_frames in frames_per_pipeline:
+                for f in pipeline_frames:
+                    if f.id not in seen_ids:
+                        await self.push_frame(f, direction)
+                        seen_ids.add(f.id)
+        else:
+            # ARRIVAL mode: drain the shared queues in the order frames arrived.
+            seen_ids = set()
+            while not self._up_queue.empty():
+                frame = await self._up_queue.get()
+                if frame.id not in seen_ids:
+                    await self.push_frame(frame, FrameDirection.UPSTREAM)
+                    seen_ids.add(frame.id)
+                self._up_queue.task_done()
+
+            seen_ids = set()
+            while not self._down_queue.empty():
+                frame = await self._down_queue.get()
+                if frame.id not in seen_ids:
+                    await self.push_frame(frame, FrameDirection.DOWNSTREAM)
+                    seen_ids.add(frame.id)
+                self._down_queue.task_done()