diff --git a/examples/storytelling-chatbot/src/bot.py b/examples/storytelling-chatbot/src/bot.py index 26853716b..0439ad029 100644 --- a/examples/storytelling-chatbot/src/bot.py +++ b/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(), diff --git a/examples/storytelling-chatbot/src/processors.py b/examples/storytelling-chatbot/src/processors.py index 7de23b1aa..8e87384fa 100644 --- a/examples/storytelling-chatbot/src/processors.py +++ b/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 diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index e1239a2f8..fff1fd8eb 100644 --- a/src/pipecat/frames/frames.py +++ b/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 # diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 6fbcff712..f12214eeb 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/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}"