wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #4029 from pipecat-ai/pk/sync-parallel-pipeline-fixes

Browse Source 
`SyncParallelPipeline` and related fixes
This commit is contained in:
kompfner
2026-03-19 14:41:16 -04:00
committed by GitHub
parent a9256ebc35 57fd29f0c4
commit f6e7599e49
9 changed files with 301 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog/4029.added.2.md Normal file
View File

@@ -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.

1
changelog/4029.added.md Normal file
View File

@@ -0,0 +1 @@
- Added `sync_with_audio` field to `OutputImageRawFrame`. When set to `True`, the output transport queues image frames with audio so they are displayed only after all preceding audio has been sent, enabling synchronized audio/image playback.

1
changelog/4029.fixed.3.md Normal file
View File

@@ -0,0 +1 @@
- Fixed `SyncParallelPipeline` breaking the Whisker debugger.

1
changelog/4029.fixed.md Normal file
View File

@@ -0,0 +1 @@
- Fixed `SyncParallelPipeline` race condition where concurrent SystemFrame processing (e.g. from RTVI) could corrupt sink queues and cause deadlocks. SystemFrames now take a fast path that passes them through without draining queued output.

33
examples/foundational/05-sync-speech-and-image.py
View File

@@ -16,11 +16,12 @@ from pipecat.frames.frames import (
Frame,
LLMContextFrame,
LLMFullResponseStartFrame,
OutputImageRawFrame,
TextFrame,
)
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
@@ -30,6 +31,7 @@ from pipecat.runner.utils import create_transport
from pipecat.services.cartesia.tts import CartesiaHttpTTSService
from pipecat.services.fal.image import FalImageGenService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.services.tts_service import TextAggregationMode
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
@@ -44,6 +46,18 @@ class MonthFrame(DataFrame):
return f"{self.name}(month: {self.month})"
class MarkImageForPlaybackSync(FrameProcessor):
"""Marks output image frames to be synchronized with audio playback."""
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
if isinstance(frame, OutputImageRawFrame):
frame.sync_with_audio = True
await self.push_frame(frame, direction)
class MonthPrepender(FrameProcessor):
def __init__(self):
super().__init__()
@@ -101,6 +115,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
settings=CartesiaHttpTTSService.Settings(
voice="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
),
# No need to aggregate by sentences (the default), as we already know we're getting full sentences
# (Otherwise the service will unnecessarily wait for follow-up input to confirm the sentence is complete,
# which, sadly, actually breaks the synchronization mechanism)
text_aggregation_mode=TextAggregationMode.TOKEN,
)
imagegen = FalImageGenService(
@@ -119,17 +137,26 @@ 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
[
imagegen, # Generate image
MarkImageForPlaybackSync(), # Mark image as needing sync w/audio during playback
],
[month_prepender, tts], # Create "Month: sentence" and output audio
[imagegen], # Generate image
frame_order=FrameOrder.PIPELINE,
),
transport.output(), # Transport output
]

8
src/pipecat/frames/frames.py
View File

@@ -274,8 +274,16 @@ class OutputImageRawFrame(DataFrame, ImageRawFrame):
An image that will be shown by the transport. If the transport supports
multiple video destinations (e.g. multiple video tracks) the destination
name can be specified in transport_destination.
Parameters:
sync_with_audio: If True, the image is queued with audio frames so
it is only displayed after all preceding audio has been sent.
Defaults to False (image is displayed immediately when the output
transport receives it).
"""
sync_with_audio: bool = field(default=False, init=False)
def __str__(self):
pts = format_pts(self.pts)
return f"{self.name}(pts: {pts}, destination: {self.transport_destination}, size: {self.size}, format: {self.format})"

190
src/pipecat/pipeline/sync_parallel_pipeline.py
View File

@@ -4,15 +4,21 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Synchronous parallel pipeline implementation for concurrent frame processing.
"""Synchronized parallel pipeline that holds output until all branches finish.
This module provides a pipeline that processes frames through multiple parallel
pipelines simultaneously, synchronizing their output to maintain frame ordering
and prevent duplicate processing.
A SyncParallelPipeline fans each inbound frame out to multiple parallel pipelines
and waits for every pipeline to finish processing before releasing any of the
resulting output frames. This ensures that all frames produced in response to a
single input frame are emitted together.
System frames (except EndFrame) are exempt from this synchronization — they pass
straight through without waiting, since they are expected to race ahead of
regular data frames.
"""
import asyncio
from dataclasses import dataclass
from enum import Enum
from itertools import chain
from typing import List
@@ -24,22 +30,42 @@ from pipecat.pipeline.pipeline import Pipeline
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.
"""Sentinel frame used to detect when a parallel pipeline has finished processing.
This frame is sent through parallel pipelines to determine when the
internal pipelines have finished processing a batch of frames.
After sending a real frame into a parallel pipeline, a SyncFrame is sent
behind it. When the SyncFrame emerges from the pipeline's output, we know
all output frames for the preceding input have been produced.
"""
pass
class SyncParallelPipelineSource(FrameProcessor):
"""Source processor for synchronous parallel pipeline processing.
"""Bookend processor placed at the start of each parallel pipeline.
Routes frames to parallel pipelines and collects upstream responses
for synchronization purposes.
Forwards downstream frames into the pipeline and captures upstream frames
into a queue so the parent SyncParallelPipeline can release them later.
"""
def __init__(self, upstream_queue: asyncio.Queue):
@@ -68,10 +94,11 @@ class SyncParallelPipelineSource(FrameProcessor):
class SyncParallelPipelineSink(FrameProcessor):
"""Sink processor for synchronous parallel pipeline processing.
"""Bookend processor placed at the end of each parallel pipeline.
Collects downstream frames from parallel pipelines and routes
upstream frames back through the pipeline.
Captures downstream output frames into a queue so the parent
SyncParallelPipeline can release them later, and forwards upstream
frames back through the pipeline.
"""
def __init__(self, downstream_queue: asyncio.Queue):
@@ -100,29 +127,44 @@ class SyncParallelPipelineSink(FrameProcessor):
class SyncParallelPipeline(BasePipeline):
"""Pipeline that processes frames through multiple parallel pipelines synchronously.
"""Fans each input frame to parallel pipelines then holds output until every pipeline finishes.
Creates multiple parallel processing paths that all receive the same input frames
and produces synchronized output. Each parallel path is a separate pipeline that
processes frames independently, with synchronization points to ensure consistent
ordering and prevent duplicate frame processing.
For each inbound frame the pipeline:
The pipeline uses SyncFrame control frames to coordinate between parallel paths
and ensure all paths have completed processing before moving to the next frame.
1. Sends the frame into every parallel pipeline.
2. Sends a ``SyncFrame`` sentinel behind it in each pipeline.
3. Waits until every pipeline has produced its ``SyncFrame``, meaning all
output for that input is ready.
4. Releases the collected output frames (deduplicating by frame id, since
the same frame may emerge from more than one branch).
System frames (except ``EndFrame``) bypass this mechanism entirely — they are
forwarded through each pipeline and pushed immediately, since system frames
are expected to race ahead of regular data frames.
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")
@@ -184,7 +226,7 @@ class SyncParallelPipeline(BasePipeline):
Returns:
The list of entry processors.
"""
return self._sources
return [s["processor"] for s in self._sources]
def processors_with_metrics(self) -> List[FrameProcessor]:
"""Collect processors that can generate metrics from all parallel pipelines.
@@ -209,11 +251,11 @@ class SyncParallelPipeline(BasePipeline):
await asyncio.gather(*[p.cleanup() for p in self._pipelines])
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames through all parallel pipelines with synchronization.
"""Send a frame through all parallel pipelines and release output once all finish.
Distributes frames to all parallel pipelines and synchronizes their output
to maintain proper ordering and prevent duplicate processing. Uses SyncFrame
control frames to coordinate between parallel paths.
System frames (except EndFrame) skip synchronization and pass straight
through. All other frames are fanned out to every pipeline, and output is
held until every pipeline signals completion (via SyncFrame).
Args:
frame: The frame to process.
@@ -221,60 +263,102 @@ class SyncParallelPipeline(BasePipeline):
"""
await super().process_frame(frame, direction)
# SystemFrames (but not EndFrame) are simply passed through all
# internal pipelines without draining queued output. This avoids
# the race condition where a SystemFrame's wait_for_sync steals
# frames from a concurrent non-SystemFrame's wait_for_sync.
if isinstance(frame, SystemFrame) and not isinstance(frame, EndFrame):
if direction == FrameDirection.UPSTREAM:
for s in self._sinks:
await s["processor"].process_frame(frame, direction)
elif direction == FrameDirection.DOWNSTREAM:
for s in self._sources:
await s["processor"].process_frame(frame, direction)
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 synchrnonize all the internal pipelines by waiting for the
# 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, (SystemFrame, EndFrame)):
if isinstance(frame, EndFrame):
new_frame = await queue.get()
if isinstance(new_frame, (SystemFrame, EndFrame)):
await main_queue.put(new_frame)
else:
while not isinstance(new_frame, (SystemFrame, EndFrame)):
if isinstance(new_frame, EndFrame):
if use_pipeline_order:
output_frames.append(new_frame)
else:
await main_queue.put(new_frame)
else:
while not isinstance(new_frame, EndFrame):
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()
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()
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()

6
src/pipecat/transports/base_output.py
View File

@@ -569,7 +569,11 @@ class BaseOutputTransport(FrameProcessor):
if not self._params.video_out_enabled:
return
if self._params.video_out_is_live and isinstance(frame, OutputImageRawFrame):
if isinstance(frame, OutputImageRawFrame) and frame.sync_with_audio:
# Route through the audio queue so the image is only
# displayed after all preceding audio has been sent.
await self._audio_queue.put(frame)
elif self._params.video_out_is_live and isinstance(frame, OutputImageRawFrame):
await self._video_queue.put(frame)
elif isinstance(frame, OutputImageRawFrame):
await self._set_video_image(frame)

117
tests/test_sync_parallel_pipeline.py Normal file
View File

@@ -0,0 +1,117 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import unittest
from dataclasses import dataclass
from pipecat.frames.frames import Frame, TextFrame
from pipecat.pipeline.sync_parallel_pipeline import FrameOrder, SyncParallelPipeline
from pipecat.processors.filters.identity_filter import IdentityFilter
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.tests.utils import run_test
@dataclass
class TaggedFrame(Frame):
"""A simple tagged frame for testing pipeline ordering."""
tag: str = ""
def __str__(self):
return f"{self.name}(tag: {self.tag})"
class EmitTaggedFrameProcessor(FrameProcessor):
"""Emits a TaggedFrame for every TextFrame it receives.
Used to produce distinguishable output from different pipelines so tests
can verify ordering.
"""
def __init__(self, tag: str, *, delay: float = 0, **kwargs):
super().__init__(**kwargs)
self._tag = tag
self._delay = delay
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
if isinstance(frame, TextFrame):
if self._delay > 0:
await asyncio.sleep(self._delay)
await self.push_frame(TaggedFrame(tag=self._tag))
else:
await self.push_frame(frame, direction)
class TestSyncParallelPipeline(unittest.IsolatedAsyncioTestCase):
async def test_dedup_multiple_frames(self):
"""Identical frames from multiple paths should be deduplicated."""
pipeline = SyncParallelPipeline([IdentityFilter()], [IdentityFilter()])
frames_to_send = [TextFrame(text="one"), TextFrame(text="two")]
expected_down_frames = [TextFrame, TextFrame]
await run_test(
pipeline,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
)
async def test_arrival_order(self):
"""With FrameOrder.ARRIVAL, a slow first pipeline's frames should
arrive after a fast second pipeline's frames."""
pipeline = SyncParallelPipeline(
[EmitTaggedFrameProcessor("slow", delay=0.05)],
[EmitTaggedFrameProcessor("fast")],
frame_order=FrameOrder.ARRIVAL,
)
frames_to_send = [TextFrame(text="one"), TextFrame(text="two")]
(down_frames, _) = await run_test(
pipeline,
frames_to_send=frames_to_send,
)
tags = [f.tag for f in down_frames if isinstance(f, TaggedFrame)]
assert tags == [
"fast",
"slow",
"fast",
"slow",
], f"Expected fast before slow in each batch, got {tags}"
async def test_pipeline_order(self):
"""With FrameOrder.PIPELINE and multiple input frames, each batch
should follow pipeline definition order regardless of processing speed."""
pipeline = SyncParallelPipeline(
[EmitTaggedFrameProcessor("slow", delay=0.05)],
[EmitTaggedFrameProcessor("fast")],
frame_order=FrameOrder.PIPELINE,
)
frames_to_send = [TextFrame(text="one"), TextFrame(text="two")]
(down_frames, _) = await run_test(
pipeline,
frames_to_send=frames_to_send,
)
tags = [f.tag for f in down_frames if isinstance(f, TaggedFrame)]
assert tags == [
"slow",
"fast",
"slow",
"fast",
], f"Expected pipeline definition order (slow, fast) in each batch, got {tags}"
async def test_default_is_arrival(self):
"""The default frame_order should be ARRIVAL."""
pipeline = SyncParallelPipeline([IdentityFilter()])
assert pipeline._frame_order == FrameOrder.ARRIVAL
if __name__ == "__main__":
unittest.main()