Improve pipeline of data gathering for 05- sample (I think it can be better, though)

src/dailyai/services/azure_ai_services.py

@@ -2,7 +2,7 @@ import aiohttp
 import asyncio
 import io
 import json
-from openai import AzureOpenAI
+from openai import AsyncAzureOpenAI
 
 import os
 import requests
@@ -51,29 +51,29 @@ class AzureLLMService(LLMService):
     def __init__(self, api_key=None, azure_endpoint=None, api_version=None, model=None):
         super().__init__()
         api_key = api_key or os.getenv("AZURE_CHATGPT_KEY")
+
         azure_endpoint = azure_endpoint or os.getenv("AZURE_CHATGPT_ENDPOINT")
+        if not azure_endpoint:
+            raise Exception("No azure endpoint specified for Azure LLM, please set AZURE_CHATGPT_ENDPOINT in the environment or pass it to the AzureLLMService constructor")
+
+        model: str | None = model or os.getenv("AZURE_CHATGPT_DEPLOYMENT_ID")
+        if not model:
+            raise Exception("No model specified for Azure LLM, please set AZURE_CHATGPT_DEPLOYMENT_ID in the environment or pass it to the AzureLLMService constructor")
+        self.model: str = model
+
         api_version = api_version or "2023-12-01-preview"
-        self.client = AzureOpenAI(
+        self.client = AsyncAzureOpenAI(
             api_key=api_key,
             azure_endpoint=azure_endpoint,
             api_version=api_version,
         )
-        self.model = model or os.getenv("AZURE_CHATGPT_DEPLOYMENT_ID")
 
-    def get_response(self, messages, stream):
+    async def run_llm_async(self, messages) -> AsyncGenerator[str, None]:
-        return self.client.chat.completions.create(
-            stream=stream,
-            messages=messages,
-            model=self.model,
-        )
-
-    async def run_llm_async(self, messages) -> AsyncGenerator[str, None, None]:
         messages_for_log = json.dumps(messages)
         self.logger.debug(f"Generating chat via azure: {messages_for_log}")
 
-        response = self.get_response(messages, stream=True)
+        chunks = await self.client.chat.completions.create(model=self.model, stream=True, messages=messages)
-
+        async for chunk in chunks:
-        for chunk in response:
             if len(chunk.choices) == 0:
                 continue
 
@@ -84,7 +84,7 @@ class AzureLLMService(LLMService):
         messages_for_log = json.dumps(messages)
         self.logger.debug(f"Generating chat via azure: {messages_for_log}")
 
-        response = await asyncio.to_thread(self.get_response, messages, False)
+        response = await self.client.chat.completions.create(model=self.model, stream=False, messages=messages)
         if response and len(response.choices) > 0:
             return response.choices[0].message.content
         else:

src/samples/theoretical-to-real/05-queued.py

@@ -1,129 +0,0 @@
-import argparse
-import asyncio
-
-from asyncio.queues import Queue
-import re
-
-from dailyai.output_queue import OutputQueueFrame, FrameType
-from dailyai.services.azure_ai_services import AzureLLMService, AzureTTSService
-from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService
-from dailyai.services.open_ai_services import OpenAILLMService, OpenAIImageGenService
-from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService
-from dailyai.services.daily_transport_service import DailyTransportService
-
-async def main(room_url):
-    meeting_duration_minutes = 5
-    transport = DailyTransportService(
-        room_url,
-        None,
-        "Month Narration Bot",
-        meeting_duration_minutes,
-    )
-    transport.mic_enabled = True
-    transport.camera_enabled = True
-    transport.mic_sample_rate = 16000
-    transport.camera_width = 1024
-    transport.camera_height = 1024
-
-    llm = AzureLLMService()
-    tts = ElevenLabsTTSService()
-    dalle = OpenAIImageGenService()
-
-    # 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.
-    async def get_all_audio(text):
-        all_audio = bytearray()
-        async for audio in tts.run_tts(text):
-            all_audio.extend(audio)
-
-        return all_audio
-
-    async def get_month_data(month):
-        print("getting month data", month)
-        image_text = ""
-        current_clause = ""
-        tts_tasks = []
-        async for text in llm.run_llm_async(
-            [
-                {
-                    "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."
-                }
-            ]
-        ):
-            print(f"{month}: got text {text}")
-            image_text += text
-            current_clause += text
-            if re.match(r"^.*[.!?]$", text):
-                tts_tasks.append(get_all_audio(current_clause))
-                current_clause = ""
-
-        tts_tasks.insert(0, dalle.run_image_gen(image_text, "1024x1024"))
-
-        data = await asyncio.gather(
-            *tts_tasks
-        )
-
-        print("done with month", month)
-
-        return {
-            "month": month,
-            "text": image_text,
-            "image": data[0][1],
-            "audio": data[1:],
-        }
-
-    months: list[str] = [
-        "January",
-        "February",
-        "March",
-        "April",
-    ]
-
-    unused_months = [
-        "May",
-        "June",
-        "July",
-        "August",
-        "September",
-        "October",
-        "November",
-        "December",
-    ]
-
-    @transport.event_handler("on_participant_joined")
-    async def on_participant_joined(transport, participant):
-        if participant["id"] == transport.my_participant_id:
-            return
-
-        print("participant joined")
-        for month_data_task in asyncio.as_completed(month_tasks):
-            data = await month_data_task
-            print("rendering month", data["month"])
-            transport.output_queue.put(
-                [
-                    OutputQueueFrame(FrameType.IMAGE_FRAME, data["image"]),
-                    OutputQueueFrame(FrameType.AUDIO_FRAME, data["audio"][0]),
-                ]
-            )
-            for audio in data["audio"][1:]:
-                transport.output_queue.put(OutputQueueFrame(FrameType.AUDIO_FRAME, audio))
-
-        # wait for the output queue to be empty, then leave the meeting
-        transport.output_queue.join()
-        transport.stop()
-
-    month_tasks = [asyncio.create_task(get_month_data(month)) for month in months]
-
-    await transport.run()
-    print("Done")
-
-if __name__=="__main__":
-    parser = argparse.ArgumentParser(description="Simple Daily Bot Sample")
-    parser.add_argument(
-        "-u", "--url", type=str, required=True, help="URL of the Daily room to join"
-    )
-
-    args: argparse.Namespace = parser.parse_args()
-
-    asyncio.run(main(args.url))

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

@@ -1,8 +1,11 @@
 import argparse
 import asyncio
 
+from asyncio.queues import Queue
+import re
+
 from dailyai.output_queue import OutputQueueFrame, FrameType
-from dailyai.services.azure_ai_services import AzureTTSService
+from dailyai.services.azure_ai_services import AzureLLMService, AzureTTSService
 from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService
 from dailyai.services.open_ai_services import OpenAILLMService, OpenAIImageGenService
 from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService
@@ -22,10 +25,12 @@ async def main(room_url):
     transport.camera_width = 1024
     transport.camera_height = 1024
 
-    llm = OpenAILLMService()
+    llm = AzureLLMService()
     tts = ElevenLabsTTSService()
     dalle = OpenAIImageGenService()
 
+    # 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.
     async def get_all_audio(text):
         all_audio = bytearray()
         async for audio in tts.run_tts(text):
@@ -33,58 +38,78 @@ async def main(room_url):
 
         return all_audio
 
-    async def show_month(month):
+    async def get_month_data(month):
-        inference_text = await llm.run_llm(
+        image_text = ""
+        current_clause = ""
+        tts_tasks = []
+        async for text in llm.run_llm_async(
             [
                 {
                     "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 your description to 1 sentence."
+                    "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."
                 }
             ]
+        ):
+            image_text += text
+            current_clause += text
+            if re.match(r"^.*[.!?]$", text):
+                tts_tasks.append(get_all_audio(current_clause))
+                current_clause = ""
+
+        tts_tasks.insert(0, dalle.run_image_gen(image_text, "1024x1024"))
+
+        data = await asyncio.gather(
+            *tts_tasks
         )
 
-        (image, audio) = await asyncio.gather(
+        return {
-            *[dalle.run_image_gen(inference_text, "1024x1024"), get_all_audio(inference_text)]
+            "month": month,
-        )
+            "text": image_text,
-        transport.output_queue.put(
+            "image": data[0][1],
-            [
+            "audio": data[1:],
-                OutputQueueFrame(FrameType.IMAGE_FRAME, image[1]),
+        }
-                OutputQueueFrame(FrameType.AUDIO_FRAME, audio),
-            ]
-        )
 
-    async def show_all_months():
+    months: list[str] = [
-        # for now just two to avoid 429s with Azure
+        "January",
-        months: list[str] = [
+        "February",
-            "January",
+        "March",
-            "February",
+        "April",
-            "March",
+    ]
-            "April",
-            "May",
-            "June",
-            "July",
-            "August",
-            "September",
-            "October",
-            "November",
-            "December",
-        ]
 
-        await asyncio.gather(*[show_month(month) for month in months])
+    unused_months = [
+        "May",
+        "June",
+        "July",
+        "August",
+        "September",
+        "October",
+        "November",
+        "December",
+    ]
 
     @transport.event_handler("on_participant_joined")
     async def on_participant_joined(transport, participant):
         if participant["id"] == transport.my_participant_id:
             return
 
-        await show_all_months()
+        for month_data_task in asyncio.as_completed(month_tasks):
+            data = await month_data_task
+            transport.output_queue.put(
+                [
+                    OutputQueueFrame(FrameType.IMAGE_FRAME, data["image"]),
+                    OutputQueueFrame(FrameType.AUDIO_FRAME, data["audio"][0]),
+                ]
+            )
+            for audio in data["audio"][1:]:
+                transport.output_queue.put(OutputQueueFrame(FrameType.AUDIO_FRAME, audio))
 
         # wait for the output queue to be empty, then leave the meeting
         transport.output_queue.join()
         transport.stop()
 
+    month_tasks = [asyncio.create_task(get_month_data(month)) for month in months]
+
     await transport.run()
-    print("Done")
 
 if __name__=="__main__":
     parser = argparse.ArgumentParser(description="Simple Daily Bot Sample")