diff --git a/examples/foundational/05-sync-speech-and-image.py b/examples/foundational/05-sync-speech-and-image.py index b896261ae..646235e7c 100644 --- a/examples/foundational/05-sync-speech-and-image.py +++ b/examples/foundational/05-sync-speech-and-image.py @@ -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 ] )