From f546bb32da2f6802e3518698967df52692c72b9c Mon Sep 17 00:00:00 2001
From: Moishe Lettvin <moishel@gmail.com>
Date: Tue, 12 Mar 2024 10:34:52 -0400
Subject: [PATCH] Make 08- work again

---
 src/dailyai/services/fal_ai_services.py      |  4 +-
 src/examples/foundational/08-bots-arguing.py | 58 +++++++++++++-------
 2 files changed, 39 insertions(+), 23 deletions(-)

diff --git a/src/dailyai/services/fal_ai_services.py b/src/dailyai/services/fal_ai_services.py
index 0ba6436b6..60226dd52 100644
--- a/src/dailyai/services/fal_ai_services.py
+++ b/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",
-                "fal-ai/fast-sdxl",
+                "110602490-fast-sdxl",
+                #"fal-ai/fast-sdxl",
                 arguments={"prompt": sentence},
             )
             for event in handler.iter_events():
diff --git a/src/examples/foundational/08-bots-arguing.py b/src/examples/foundational/08-bots-arguing.py
index c625430d7..35cc062e7 100644
--- a/src/examples/foundational/08-bots-arguing.py
+++ b/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():
-            # Run the LLMs synchronously for the back-and-forth
-            bot1_msg = await llm.run_llm(bot1_messages)
-            print(f"bot1_msg: {bot1_msg}")
-            if bot1_msg:
-                bot1_messages.append({"role": "assistant", "content": bot1_msg})
-                bot2_messages.append({"role": "user", "content": bot1_msg})
+        async def get_text_and_audio(messages) -> Tuple[str, bytearray]:
+            """This function streams text from the LLM and uses the TTS service to convert
+             that text to speech as it's received. """
+            source_queue = asyncio.Queue()
+            sink_queue = asyncio.Queue()
+            sentence_aggregator = SentenceAggregator()
+            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):
-                all_audio.extend(audio)
+            while sink_queue.qsize():
+                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
-            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})
+            message, audio = await get_text_and_audio(bot2_messages)
 
-            all_audio = bytearray()
-            async for audio in tts2.run_tts(bot2_msg):
-                all_audio.extend(audio)
+            bot2_messages.append({"role": "assistant", "content": message})
+            bot1_messages.append({"role": "user", "content": message})
 
-            return all_audio
+            return audio
 
         async def argue():
             for i in range(100):