Make 08- work again

src/dailyai/services/fal_ai_services.py

@@ -32,8 +32,8 @@ class FalImageGenService(ImageGenService):
     async def run_image_gen(self, sentence) -> tuple[str, bytes]:
         def get_image_url(sentence, size):
             handler = fal.apps.submit(
-                #"110602490-fast-sdxl",
+                "110602490-fast-sdxl",
-                "fal-ai/fast-sdxl",
+                #"fal-ai/fast-sdxl",
                 arguments={"prompt": sentence},
             )
             for event in handler.iter_events():

src/examples/foundational/08-bots-arguing.py

@@ -1,13 +1,16 @@
+from typing import Tuple
 import aiohttp
 import asyncio
 import logging
 import os
+from dailyai.pipeline.aggregators import SentenceAggregator
+from dailyai.pipeline.pipeline import Pipeline
 
 from dailyai.services.daily_transport_service import DailyTransportService
 from dailyai.services.azure_ai_services import AzureLLMService, AzureTTSService
 from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService
 from dailyai.services.fal_ai_services import FalImageGenService
-from dailyai.pipeline.frames import AudioFrame, ImageFrame
+from dailyai.pipeline.frames import AudioFrame, EndFrame, ImageFrame, LLMMessagesQueueFrame, TextFrame
 from examples.support.runner import configure
 
 logging.basicConfig(format=f"%(levelno)s %(asctime)s %(message)s")
@@ -63,33 +66,46 @@ async def main(room_url: str):
             },
         ]
 
-        async def get_bot1_statement():
+        async def get_text_and_audio(messages) -> Tuple[str, bytearray]:
-            # Run the LLMs synchronously for the back-and-forth
+            """This function streams text from the LLM and uses the TTS service to convert
-            bot1_msg = await llm.run_llm(bot1_messages)
+             that text to speech as it's received. """
-            print(f"bot1_msg: {bot1_msg}")
+            source_queue = asyncio.Queue()
-            if bot1_msg:
+            sink_queue = asyncio.Queue()
-                bot1_messages.append({"role": "assistant", "content": bot1_msg})
+            sentence_aggregator = SentenceAggregator()
-                bot2_messages.append({"role": "user", "content": bot1_msg})
+            pipeline = Pipeline(
+                [llm, sentence_aggregator, tts1], source_queue, sink_queue
+            )
 
+            await source_queue.put(LLMMessagesQueueFrame(messages))
+            await source_queue.put(EndFrame())
+            await pipeline.run_pipeline()
+
+            message = ""
             all_audio = bytearray()
-            async for audio in tts1.run_tts(bot1_msg):
+            while sink_queue.qsize():
-                all_audio.extend(audio)
+                frame = sink_queue.get_nowait()
+                if isinstance(frame, TextFrame):
+                    message += frame.text
+                elif isinstance(frame, AudioFrame):
+                    all_audio.extend(frame.data)
 
-            return all_audio
+            return (message, all_audio)
+
+        async def get_bot1_statement():
+            message, audio = await get_text_and_audio(bot1_messages)
+
+            bot1_messages.append({"role": "assistant", "content": message})
+            bot2_messages.append({"role": "user", "content": message})
+
+            return audio
 
         async def get_bot2_statement():
-            # Run the LLMs synchronously for the back-and-forth
+            message, audio = await get_text_and_audio(bot2_messages)
-            bot2_msg = await llm.run_llm(bot2_messages)
-            print(f"bot2_msg: {bot2_msg}")
-            if bot2_msg:
-                bot2_messages.append({"role": "assistant", "content": bot2_msg})
-                bot1_messages.append({"role": "user", "content": bot2_msg})
 
-            all_audio = bytearray()
+            bot2_messages.append({"role": "assistant", "content": message})
-            async for audio in tts2.run_tts(bot2_msg):
+            bot1_messages.append({"role": "user", "content": message})
-                all_audio.extend(audio)
 
-            return all_audio
+            return audio
 
         async def argue():
             for i in range(100):