explicit MetadataFrame

examples/storytelling-chatbot/src/bot.py

@@ -17,13 +17,25 @@ from prompts import CUE_USER_TURN, LLM_BASE_PROMPT
 from utils.helpers import load_images, load_sounds
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
-from pipecat.frames.frames import EndFrame, Frame, ImageRawFrame, TextFrame
+from pipecat.frames.frames import (
+    AudioRawFrame,
+    EndFrame,
+    Frame,
+    ImageRawFrame,
+    MetadataFrame,
+    SystemFrame,
+    TextFrame,
+    TTSStartedFrame,
+    TTSStoppedFrame,
+)
 from pipecat.observers.base_observer import BaseObserver
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.sync_parallel_pipeline import SyncParallelPipeline
 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.processors.logger import FrameLogger
 from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.fal import FalImageGenService
 from pipecat.services.google import GoogleImageGenService, GoogleLLMService
@@ -84,6 +96,18 @@ class DebugObserver(BaseObserver):
             logger.info(
                 f"⚡ TEXT FRAME: {src} {arrow} {dst} at {time_sec:.2f}s, metadata: {frame.metadata}"
             )
+        elif isinstance(frame, TTSStartedFrame):
+            logger.info(
+                f"⚡ TTS STARTED FRAME: {src} {arrow} {dst} at {time_sec:.2f}s, metadata: {frame.metadata}"
+            )
+        elif isinstance(frame, TTSStoppedFrame):
+            logger.info(
+                f"⚡ TTS STOPPED FRAME: {src} {arrow} {dst} at {time_sec:.2f}s, metadata: {frame.metadata}"
+            )
+        elif isinstance(frame, MetadataFrame):
+            logger.info(
+                f"⚡ METADATA FRAME: {src} {arrow} {dst} at {time_sec:.2f}s, metadata: {frame.metadata}"
+            )
 
 
 async def main(room_url, token=None):
@@ -136,13 +160,18 @@ async def main(room_url, token=None):
         runner = PipelineRunner()
 
         logger.debug("Waiting for participant...")
+        after = FrameLogger("After", "red", ignored_frame_types=[SystemFrame, AudioRawFrame])
+        before = FrameLogger("Before", "cyan", ignored_frame_types=[SystemFrame, AudioRawFrame])
         main_pipeline = Pipeline(
             [
                 transport.input(),
                 context_aggregator.user(),
                 llm_service,
                 story_processor,
+                # SyncParallelPipeline([image_processor], [tts_service]),
+                before,
                 image_processor,
+                after,
                 tts_service,
                 transport.output(),
                 context_aggregator.assistant(),

examples/storytelling-chatbot/src/processors.py

@@ -17,6 +17,7 @@ from pipecat.frames.frames import (
     Frame,
     LLMFullResponseEndFrame,
     LLMMessagesFrame,
+    MetadataFrame,
     TextFrame,
     UserStoppedSpeakingFrame,
 )
@@ -154,6 +155,7 @@ class StoryProcessor(FrameProcessor):
         # Driven by the prompt, the LLM should have asked the user for input
         elif isinstance(frame, LLMFullResponseEndFrame):
             # We use a different frame type, as to avoid image generation ingest
+            await self.push_frame(MetadataFrame())
             await self.push_frame(StoryPromptFrame(self._text))
             self._text = ""
             await self.push_frame(frame)
@@ -193,6 +195,10 @@ class StoryProcessor(FrameProcessor):
 
                 if len(before_break) > 2:
                     self._story.append(before_break)
+                    mf = MetadataFrame()
+                    mf.metadata = {"story_page_id": self._current_page}
+                    await self.push_frame(mf)
+
                     spf = StoryPageFrame(before_break)
                     spf.metadata["story_page_id"] = self._current_page
                     self._current_page += 1

src/pipecat/frames/frames.py

@@ -91,6 +91,13 @@ class ControlFrame(Frame):
     pass
 
 
+@dataclass
+class MetadataFrame(ControlFrame):
+    """Used to set default metadata for downstream processors to apply to newly
+    created frames such as a frame_group_id.
+    """
+
+
 #
 # Mixins
 #

src/pipecat/processors/frame_processor.py

@@ -16,6 +16,7 @@ from pipecat.frames.frames import (
     CancelFrame,
     ErrorFrame,
     Frame,
+    MetadataFrame,
     StartFrame,
     StartInterruptionFrame,
     StopInterruptionFrame,
@@ -228,7 +229,7 @@ class FrameProcessor:
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         # System frames skip the queue and blow up determinism
-        if not isinstance(frame, SystemFrame):
+        if isinstance(frame, MetadataFrame):
             self._input_frame_metadata = frame.metadata
             # print(
             #     f"!!! PROCESS: I am {self._name}, input frame is a {frame.name}, metadata is {self._input_frame_metadata}"