works, but still needs a parallel pipeline

examples/foundational/14e-function-calling-gemini.py

@@ -63,7 +63,7 @@ async def main():
         )
 
         llm = GoogleLLMService(
-            model="gemini-1.5-flash-latest",
+            model="gemini-2.0-flash-exp",
             # model="gemini-exp-1114",
             api_key=os.getenv("GOOGLE_API_KEY"),
         )

examples/storytelling-chatbot/src/bot.py

@@ -12,19 +12,22 @@ import sys
 import aiohttp
 from dotenv import load_dotenv
 from loguru import logger
-from processors import StoryImageProcessor, StoryProcessor
+from processors import StoryBreakReinsertProcessor, StoryImageProcessor, StoryProcessor
 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
+from pipecat.pipeline.parallel_pipeline import ParallelPipeline
 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.services.elevenlabs import ElevenLabsTTSService
+from pipecat.processors.logger import FrameLogger
+from pipecat.services.elevenlabs import ElevenLabsHttpTTSService, ElevenLabsTTSService
 from pipecat.services.fal import FalImageGenService
-from pipecat.services.google import GoogleLLMService
+from pipecat.services.google import GoogleImageGenService, GoogleLLMService
 from pipecat.transports.services.daily import (
     DailyParams,
     DailyTransport,
@@ -63,13 +66,20 @@ async def main(room_url, token=None):
 
         # -------------- Services --------------- #
 
-        llm_service = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"))
-
-        tts_service = ElevenLabsTTSService(
-            api_key=os.getenv("ELEVENLABS_API_KEY"), voice_id=os.getenv("ELEVENLABS_VOICE_ID")
+        llm_service = GoogleLLMService(
+            api_key=os.getenv("GOOGLE_API_KEY"),
+            model="gemini-2.0-flash-exp",
         )
 
-        image_gen = GoogleImageGenService(api_key=os.getenv("GOOGLE_API_KEY"))
+        tts_service = ElevenLabsHttpTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
+        )
+
+        image_gen = GoogleImageGenService(
+            api_key=os.getenv("GOOGLE_API_KEY"),  # model="imagen-3.0-fast-generate-001"
+        )
 
         # --------------- Setup ----------------- #
 
@@ -90,6 +100,10 @@ async def main(room_url, token=None):
         runner = PipelineRunner()
 
         logger.debug("Waiting for participant...")
+        fl = FrameLogger("Into parallel pipeline", "cyan")
+        fl2 = FrameLogger("Out of parallel pipeline", "red")
+        fl3 = FrameLogger("out of LLM service", "green")
+        fl4 = FrameLogger("out of tts", "magenta")
         main_pipeline = Pipeline(
             [
                 transport.input(),
@@ -97,8 +111,15 @@ async def main(room_url, token=None):
                 llm_service,
                 story_processor,
                 image_processor,
+                # fl,
+                # SyncParallelPipeline(
+                #     [tts_service],  # Audio pipeline
+                #     [image_processor],  # Image pipeline
+                # ),
+                # fl2,
                 tts_service,
                 transport.output(),
+                StoryBreakReinsertProcessor(),
                 context_aggregator.assistant(),
             ]
         )

examples/storytelling-chatbot/src/processors.py

@@ -44,6 +44,15 @@ class StoryPromptFrame(TextFrame):
     pass
 
 
+class StoryBreakFrame(Frame):
+    """Frame for storing story text that needs a [break] tag reinserted.
+    Does not inherit from TextFrame to avoid TTS processing.
+    """
+
+    def __init__(self):
+        super().__init__()
+
+
 # ------------ Frame Processors ----------- #
 
 
@@ -62,7 +71,10 @@ class StoryImageProcessor(FrameProcessor):
         super().__init__()
         self._image_gen_service = image_gen_service
         # Create a new LLM service to use a different system prompt, etc
-        self._llm_service = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"))
+        self._llm_service = GoogleLLMService(
+            api_key=os.getenv("GOOGLE_API_KEY"),
+            model="gemini-2.0-flash-exp",
+        )
 
         self.pages = []
         self.image_descriptions = []
@@ -153,7 +165,7 @@ class StoryProcessor(FrameProcessor):
             await self.push_frame(frame)
             # Send an app message to the UI
             await self.push_frame(DailyTransportMessageFrame(CUE_USER_TURN))
-            await self.push_frame(sounds["listening"])
+            # await self.push_frame(sounds["listening"])
 
         # Anything that is not a TextFrame pass through
         else:
@@ -188,8 +200,25 @@ class StoryProcessor(FrameProcessor):
                 if len(before_break) > 2:
                     self._story.append(before_break)
                     await self.push_frame(StoryPageFrame(before_break))
-                    # await self.push_frame(sounds["ding"])
+                    await self.push_frame(StoryBreakFrame())
                     await self.push_frame(DailyTransportMessageFrame(CUE_ASSISTANT_TURN))
 
                 # Keep the remainder (if any) in the buffer
                 self._text = parts[1].strip() if len(parts) > 1 else ""
+
+
+class StoryBreakReinsertProcessor(FrameProcessor):
+    """Re-inserts [break] tags into story text before it reaches the assistant context aggregator.
+
+    This processor looks for StoryBreakFrames (which aren't processed by TTS) and creates
+    TextFrames with [break] tags for the context aggregator.
+    """
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, StoryBreakFrame):
+            # Create a new TextFrame with [break] tag
+            await self.push_frame(TextFrame(" [break]"))
+        else:
+            await self.push_frame(frame)