Fix example 5

src/dailyai/services/ai_services.py

@@ -29,7 +29,6 @@ class AIService:
 
     async def run_to_queue(self, queue: asyncio.Queue, frames, add_end_of_stream=False) -> None:
         async for frame in self.run(frames):
-            print("got frame", frame.frame_type)
             await queue.put(frame)
 
         if add_end_of_stream:
@@ -48,29 +47,18 @@ class AIService:
         if not requested_frame_types:
             requested_frame_types = self.possible_output_frame_types()
 
-        print("running", self.__class__.__name__, "with frame types", requested_frame_types)
-
         if isinstance(frames, AsyncIterable):
             async for frame in frames:
                 async for output_frame in self.process_frame(requested_frame_types, frame):
-                    print(
-                        "yielding frame", self.__class__.__name__, output_frame.frame_type
-                    )
                     yield output_frame
         elif isinstance(frames, Iterable):
             for frame in frames:
                 async for output_frame in self.process_frame(requested_frame_types, frame):
-                    print(
-                        "yielding frame", self.__class__.__name__, output_frame.frame_type
-                    )
                     yield output_frame
         elif isinstance(frames, asyncio.Queue):
             while True:
                 frame = await frames.get()
                 async for output_frame in self.process_frame(requested_frame_types, frame):
-                    print(
-                        "yielding frame", self.__class__.__name__, output_frame.frame_type
-                    )
                     yield output_frame
                 if frame.frame_type == FrameType.END_STREAM:
                     break

src/dailyai/services/daily_transport_service.py

@@ -200,10 +200,10 @@ class DailyTransportService(EventHandler):
 
     async def marshal_frames(self):
         while True:
-            frame = await self.send_queue.get()
+            frame: QueueFrame | list = await self.send_queue.get()
             self.threadsafe_send_queue.put(frame)
             self.send_queue.task_done()
-            if frame.frame_type == FrameType.END_STREAM:
+            if type(frame) == QueueFrame and frame.frame_type == FrameType.END_STREAM:
                 break
 
     def wait_for_send_queue_to_empty(self):

src/dailyai/services/to_be_updated/huggingface_ai_service.py

@@ -13,7 +13,6 @@ class HuggingFaceAIService(AIService):
     # available models at https://huggingface.co/Helsinki-NLP (**not all models use 2-character language codes**)
     def run_text_translation(self, sentence, source_language, target_language):
         translator = pipeline(f"translation", model=f"Helsinki-NLP/opus-mt-{source_language}-{target_language}")
-        print(translator(sentence))
 
         return translator(sentence)[0]["translation_text"]
 

src/samples/theoretical-to-real/05-sync-speech-and-text.py

@@ -5,6 +5,7 @@ from asyncio.queues import Queue
 import re
 
 from dailyai.queue_frame import QueueFrame, FrameType
+from dailyai.services.ai_services import SentenceAggregator
 from dailyai.services.azure_ai_services import AzureLLMService
 from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService
 from dailyai.services.open_ai_services import OpenAIImageGenService
@@ -31,7 +32,7 @@ async def main(room_url):
     #dalle = OpenAIImageGenService(image_size="1024x1024")
 
     # Get a complete audio chunk from the given text. Splitting this into its own
-    # coroutine lets us ensure proper ordering of the audio chunks on the output queue.
+    # coroutine lets us ensure proper ordering of the audio chunks on the send queue.
     async def get_all_audio(text):
         all_audio = bytearray()
         async for audio in tts.run_tts(text):
@@ -43,14 +44,18 @@ async def main(room_url):
         image_text = ""
         tts_tasks = []
         first_sentence = True
-        async for sentence in llm.run_llm_async_sentences(
+        messages = [
-            [
             {
                 "role": "system",
-                    "content": f"Describe a nature photograph suitable for use in a calendar, for the month of {month}. Include only the image description with no preamble. Limit the description to one sentence, please."
+                "content": f"Describe a nature photograph suitable for use in a calendar, for the month of {month}. Include only the image description with no preamble. Limit the description to one sentence, please.",
             }
         ]
-        ):
+
+        async for frame in SentenceAggregator().run(llm.run([QueueFrame(FrameType.LLM_MESSAGE, messages)])):
+            if type(frame.frame_data) != str:
+                raise Exception("LLM service requires a string for the data field")
+
+            sentence: str = frame.frame_data
             image_text += sentence
 
             if first_sentence:
@@ -100,18 +105,17 @@ async def main(room_url):
         # likely no delay between months, but the months won't display in order.
         for month_data_task in asyncio.as_completed(month_tasks):
             data = await month_data_task
-            print(f"got data, queueing frames...")
+            await transport.send_queue.put(
-            transport.output_queue.put(
                 [
                     QueueFrame(FrameType.IMAGE, data["image"]),
                     QueueFrame(FrameType.AUDIO, data["audio"][0]),
                 ]
             )
             for audio in data["audio"][1:]:
-                transport.output_queue.put(QueueFrame(FrameType.AUDIO, audio))
+                await transport.send_queue.put(QueueFrame(FrameType.AUDIO, audio))
 
         # wait for the output queue to be empty, then leave the meeting
-        transport.output_queue.join()
+        transport.wait_for_send_queue_to_empty()
         transport.stop()
 
     month_tasks = [asyncio.create_task(get_month_data(month)) for month in months]