Add ImageBeforeAudioReorderer to sync-speech-and-image example

Add a processor after SyncParallelPipeline that ensures each image frame
precedes its corresponding TTS audio frames. SyncParallelPipeline batches
them together but doesn't guarantee branch ordering. The reorderer detects
when TTS frames arrive before their image (via context_id tracking) and
holds them until the image arrives.

Also rename ImageAudioSync to MarkImageForPlaybackSync for clarity.
This commit is contained in:
Paul Kompfner
2026-03-13 22:09:44 -04:00
committed by Mark Backman
parent 4b66dd444b
commit d2341e0199

View File

@@ -12,12 +12,17 @@ from dotenv import load_dotenv
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
@@ -46,7 +51,7 @@ class MonthFrame(DataFrame):
return f"{self.name}(month: {self.month})"
class ImageAudioSync(FrameProcessor):
class MarkImageForPlaybackSync(FrameProcessor):
"""Marks output image frames to be synchronized with audio playback."""
async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -58,6 +63,61 @@ class ImageAudioSync(FrameProcessor):
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__()
@@ -149,9 +209,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
[month_prepender, tts], # Create "Month: sentence" and output audio
[
imagegen, # Generate image
ImageAudioSync(), # Mark image as needing sync output w/audio
MarkImageForPlaybackSync(), # Mark image as needing sync w/audio during playback
],
),
ImageBeforeAudioReorderer(), # Ensure each image precedes its audio (important for playback)
transport.output(), # Transport output
]
)