diff --git a/examples/foundational/06a-image-sync.py b/examples/foundational/06a-image-sync.py index ec3c782ad..2cd56ebc9 100644 --- a/examples/foundational/06a-image-sync.py +++ b/examples/foundational/06a-image-sync.py @@ -15,10 +15,17 @@ from PIL import Image from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame, Frame, OutputImageRawFrame, SystemFrame, TextFrame +from pipecat.frames.frames import ( + BotStartedSpeakingFrame, + BotStoppedSpeakingFrame, + EndFrame, + Frame, + OutputImageRawFrame, + TextFrame, +) from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineTask +from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.services.cartesia import CartesiaHttpTTSService @@ -45,7 +52,7 @@ class ImageSyncAggregator(FrameProcessor): async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) - if not isinstance(frame, SystemFrame) and direction == FrameDirection.DOWNSTREAM: + if isinstance(frame, BotStartedSpeakingFrame): await self.push_frame( OutputImageRawFrame( image=self._speaking_image_bytes, @@ -53,7 +60,8 @@ class ImageSyncAggregator(FrameProcessor): format=self._speaking_image_format, ) ) - await self.push_frame(frame) + + elif isinstance(frame, BotStoppedSpeakingFrame): await self.push_frame( OutputImageRawFrame( image=self._waiting_image_bytes, @@ -61,8 +69,8 @@ class ImageSyncAggregator(FrameProcessor): format=self._waiting_image_format, ) ) - else: - await self.push_frame(frame) + + await self.push_frame(frame) async def main(): @@ -109,16 +117,24 @@ async def main(): pipeline = Pipeline( [ transport.input(), - image_sync_aggregator, context_aggregator.user(), llm, tts, + image_sync_aggregator, transport.output(), context_aggregator.assistant(), ] ) - task = PipelineTask(pipeline) + task = PipelineTask( + pipeline, + PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + report_only_initial_ttfb=True, + ), + ) @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant):