Speaking / waiting images

src/dailyai/queue_aggregators.py

@@ -1,6 +1,6 @@
 import asyncio
 
-from dailyai.queue_frame import LLMMessagesQueueFrame, QueueFrame, TextQueueFrame
+from dailyai.queue_frame import LLMMessagesQueueFrame, QueueFrame, TextQueueFrame, TranscriptionQueueFrame
 from dailyai.services.ai_services import AIService
 
 from typing import AsyncGenerator, List
@@ -32,16 +32,24 @@ class LLMContextAggregator(AIService):
             messages: list[dict],
             role: str,
             bot_participant_id=None,
-            complete_sentences=True):
+            complete_sentences=True,
+            pass_through=False):
         self.messages = messages
         self.bot_participant_id = bot_participant_id
         self.role = role
         self.sentence = ""
         self.complete_sentences = complete_sentences
+        self.pass_through = pass_through
 
     async def process_frame(self, frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]:
         # TODO: split up transcription by participant
         if isinstance(frame, TextQueueFrame):
+
+            # Ignore transcription frames from the bot
+            if isinstance(frame, TranscriptionQueueFrame):
+                if frame.participantId == self.bot_participant_id:
+                    return
+
             if self.complete_sentences:
                 self.sentence += frame.text
                 if self.sentence.endswith((".", "?", "!")):
@@ -52,4 +60,23 @@ class LLMContextAggregator(AIService):
                 self.messages.append({"role": self.role, "content": frame.text})
                 yield LLMMessagesQueueFrame(self.messages)
 
-        yield frame
+            if self.pass_through:
+                yield frame
+        else:
+            yield frame
+
+class LLMUserContextAggregator(LLMContextAggregator):
+    def __init__(self,
+            messages: list[dict],
+            bot_participant_id=None,
+            complete_sentences=True):
+        super().__init__(messages, "user", bot_participant_id, complete_sentences, pass_through=False)
+
+
+class LLMAssistantContextAggregator(LLMContextAggregator):
+    def __init__(
+        self, messages: list[dict], bot_participant_id=None, complete_sentences=True
+    ):
+        super().__init__(
+            messages, "assistan", bot_participant_id, complete_sentences, pass_through=True
+        )

src/dailyai/services/ai_services.py

@@ -86,9 +86,7 @@ class LLMService(AIService):
         pass
 
     async def process_frame(self, frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]:
-        if isinstance(frame, ControlQueueFrame):
-            yield frame
-        elif isinstance(frame, LLMMessagesQueueFrame):
+        if isinstance(frame, LLMMessagesQueueFrame):
             async for text_chunk in self.run_llm_async(frame.messages):
                 yield TextQueueFrame(text_chunk)
 
@@ -111,11 +109,9 @@ class TTSService(AIService):
         yield bytes()
 
     async def process_frame(self, frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]:
-        if isinstance(frame, ControlQueueFrame):
+        if not isinstance(frame, TextQueueFrame):
             yield frame
             return
-        elif not isinstance(frame, TextQueueFrame):
-            return
 
         text: str | None = None
         if not self.aggregate_sentences:

src/samples/foundational/06a-image-sync.py

@@ -0,0 +1,134 @@
+import argparse
+import asyncio
+from typing import AsyncGenerator
+import requests
+import time
+import urllib.parse
+
+from PIL import Image
+from dailyai.queue_frame import ImageQueueFrame, QueueFrame
+
+from dailyai.services.daily_transport_service import DailyTransportService
+from dailyai.services.azure_ai_services import AzureLLMService, AzureTTSService
+from dailyai.services.ai_services import AIService
+from dailyai.queue_aggregators import LLMAssistantContextAggregator, LLMUserContextAggregator
+from dailyai.services.fal_ai_services import FalImageGenService
+
+
+class ImageSyncAggregator(AIService):
+    def __init__(self, speaking_path:str, waiting_path:str):
+        self._speaking_image = Image.open(speaking_path)
+        self._speaking_image_bytes = self._speaking_image.tobytes()
+
+        self._waiting_image = Image.open(waiting_path)
+        self._waiting_image_bytes = self._waiting_image.tobytes()
+
+    async def process_frame(self, frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]:
+        yield ImageQueueFrame(None, self._speaking_image_bytes)
+        yield frame
+        yield ImageQueueFrame(None, self._waiting_image_bytes)
+
+async def main(room_url: str, token):
+    global transport
+    global llm
+    global tts
+
+    transport = DailyTransportService(
+        room_url,
+        token,
+        "Respond bot",
+        5,
+    )
+    transport.camera_enabled = True
+    transport.camera_width = 1024
+    transport.camera_height = 1024
+    transport.mic_enabled = True
+    transport.mic_sample_rate = 16000
+
+    llm = AzureLLMService()
+    tts = AzureTTSService()
+    img = FalImageGenService(image_size="1024x1024")
+
+    async def get_images():
+        get_speaking_task = asyncio.create_task(
+            img.run_image_gen("An image of a cat speaking")
+        )
+        get_waiting_task = asyncio.create_task(
+            img.run_image_gen("An image of a cat waiting")
+        )
+
+        (speaking_data, waiting_data) = await asyncio.gather(
+            get_speaking_task, get_waiting_task
+        )
+
+        return speaking_data, waiting_data
+
+    @transport.event_handler("on_first_other_participant_joined")
+    async def on_first_other_participant_joined(transport):
+        await tts.say("Hi, I'm listening!", transport.send_queue)
+
+    async def handle_transcriptions():
+        messages = [
+            {"role": "system", "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio. Respond to what the user said in a creative and helpful way."},
+        ]
+
+        tma_in = LLMUserContextAggregator(
+            messages, transport.my_participant_id
+        )
+        tma_out = LLMAssistantContextAggregator(
+            messages, transport.my_participant_id
+        )
+        image_sync_aggregator = ImageSyncAggregator(
+            "/Users/moishe/src/daily-ai-sdk/src/samples/foundational/speaking.png",
+            "/Users/moishe/src/daily-ai-sdk/src/samples/foundational/waiting.png",
+        )
+        await tts.run_to_queue(
+            transport.send_queue,
+            image_sync_aggregator.run(
+                tma_out.run(
+                    llm.run(
+                        tma_in.run(
+                            transport.get_receive_frames()
+                        )
+                    )
+                )
+            )
+        )
+
+    transport.transcription_settings["extra"]["punctuate"] = True
+    await asyncio.gather(transport.run(), handle_transcriptions())
+
+
+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"
+    )
+    parser.add_argument(
+        "-k",
+        "--apikey",
+        type=str,
+        required=True,
+        help="Daily API Key (needed to create token)",
+    )
+
+    args, unknown = parser.parse_known_args()
+
+    # Create a meeting token for the given room with an expiration 1 hour in the future.
+    room_name: str = urllib.parse.urlparse(args.url).path[1:]
+    expiration: float = time.time() + 60 * 60
+
+    res: requests.Response = requests.post(
+        f"https://api.daily.co/v1/meeting-tokens",
+        headers={"Authorization": f"Bearer {args.apikey}"},
+        json={
+            "properties": {"room_name": room_name, "is_owner": True, "exp": expiration}
+        },
+    )
+
+    if res.status_code != 200:
+        raise Exception(f"Failed to create meeting token: {res.status_code} {res.text}")
+
+    token: str = res.json()["token"]
+
+    asyncio.run(main(args.url, token))