Aggregators for LLM messages

src/dailyai/queue_aggregators.py

@@ -0,0 +1,67 @@
+import asyncio
+
+from dailyai.queue_frame import QueueFrame, FrameType
+from dailyai.services.ai_services import AIService
+
+from typing import AsyncGenerator, List
+
+class QueueTee:
+    async def run_to_queue_and_generate(
+        self,
+        output_queue: asyncio.Queue,
+        generator: AsyncGenerator[QueueFrame, None]
+    ) -> AsyncGenerator[QueueFrame, None]:
+        async for frame in generator:
+            await output_queue.put(frame)
+            yield frame
+
+    async def run_to_queues(
+        self,
+        output_queues: List[asyncio.Queue],
+        generator: AsyncGenerator[QueueFrame, None]
+    ):
+        async for frame in generator:
+            for queue in output_queues:
+                await queue.put(frame)
+
+class TranscriptionToLLMMessageAggregator(AIService):
+    def __init__(self, messages, bot_participant_id):
+        self.messages = messages
+        self.bot_participant_id = bot_participant_id
+        self.sentence = ""
+
+    async def process_frame(self, frame:QueueFrame) -> AsyncGenerator[QueueFrame, None]:
+        if frame.frame_type != FrameType.TRANSCRIPTION:
+            return
+
+        message = frame.frame_data
+        if not isinstance(message, dict):
+            return
+
+        if message["session_id"] == self.bot_participant_id:
+            return
+
+        print("transcription to message", frame)
+
+        # todo: we could differentiate between transcriptions from different participants
+        self.sentence += message["text"]
+        if self.sentence.endswith((".", "?", "!")):
+            self.messages.append({"role": "user", "content": self.sentence})
+            self.sentence = ""
+            yield QueueFrame(FrameType.LLM_MESSAGE, self.messages)
+
+
+class LLMResponseToLLMMessageAggregator(AIService):
+    def __init__(self, messages):
+        self.messages = messages
+        self.sentence = ""
+
+    async def process_frame(self, frame:QueueFrame) -> AsyncGenerator[QueueFrame, None]:
+        if frame.frame_type == FrameType.TEXT and isinstance(frame.frame_data, str):
+            print("llmresponse to message", frame)
+            self.sentence += frame.frame_data
+            if self.sentence.endswith((".", "?", "!")):
+                self.messages.append({"role": "assistant", "content": self.sentence})
+                self.sentence = ""
+
+        yield frame

src/dailyai/queue_frame.py

@@ -8,8 +8,9 @@ class FrameType(Enum):
     AUDIO = 2
     IMAGE = 3
     TEXT = 4
-    LLM_MESSAGE = 5
-    APP_MESSAGE = 6
+    TRANSCRIPTION = 5
+    LLM_MESSAGE = 6
+    APP_MESSAGE = 7
 
 @dataclass(frozen=True)
 class QueueFrame:

src/dailyai/services/ai_services.py

@@ -7,11 +7,9 @@ from httpx import request
 from dailyai.queue_frame import QueueFrame, FrameType
 
 from abc import abstractmethod
-from typing import AsyncGenerator, Iterable
+from typing import AsyncGenerator, AsyncIterable, Iterable
 from dataclasses import dataclass
-from typing import AsyncGenerator
 
-from collections.abc import Iterable, AsyncIterable
 
 class AIService:
 

src/dailyai/services/daily_transport_service.py

@@ -57,6 +57,7 @@ class DailyTransportService(EventHandler):
         self.receive_queue = asyncio.Queue()
 
         self.other_participant_has_joined = False
+        self.my_participant_id = None
 
         self.camera_thread = None
         self.frame_consumer_thread = None
@@ -150,6 +151,7 @@ class DailyTransportService(EventHandler):
 
         self.client.set_user_name(self.bot_name)
         self.client.join(self.room_url, self.token, completion=self.call_joined)
+        self.my_participant_id = self.client.participants()["local"]["id"]
 
         self.client.update_inputs(
             {
@@ -193,8 +195,6 @@ class DailyTransportService(EventHandler):
         if self.token:
             self.client.start_transcription(self.transcription_settings)
 
-        self.my_participant_id = self.client.participants()["local"]["id"]
-
     async def get_receive_frames(self):
         while True:
             frame = await self.receive_queue.get()
@@ -279,7 +279,7 @@ class DailyTransportService(EventHandler):
 
     def on_transcription_message(self, message:dict):
         if self.loop:
-            frame = QueueFrame(FrameType.TEXT, message)
+            frame = QueueFrame(FrameType.TRANSCRIPTION, message)
             asyncio.run_coroutine_threadsafe(self.receive_queue.put(frame), self.loop)
 
     def on_transcription_stopped(self, stopped_by, stopped_by_error):

src/samples/foundational/06-listen-and-respond.py

@@ -6,7 +6,10 @@ import urllib.parse
 
 from dailyai.services.daily_transport_service import DailyTransportService
 from dailyai.services.azure_ai_services import AzureLLMService, AzureTTSService
-from dailyai.queue_frame import QueueFrame, FrameType
+from dailyai.queue_aggregators import (
+    TranscriptionToLLMMessageAggregator,
+    LLMResponseToLLMMessageAggregator,
+)
 
 async def main(room_url:str, token):
     global transport
@@ -17,7 +20,7 @@ async def main(room_url:str, token):
         room_url,
         token,
         "Respond bot",
-        1,
+        5,
     )
     transport.mic_enabled = True
     transport.mic_sample_rate = 16000
@@ -26,33 +29,27 @@ async def main(room_url:str, token):
     llm = AzureLLMService()
     tts = AzureTTSService()
 
+    @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."},
         ]
 
-        sentence = ""
-        async for frame in transport.get_receive_frames():
-            if frame.frame_type != FrameType.TEXT:
-                continue
-
-            message = frame.frame_data
-            if message["session_id"] == transport.my_participant_id:
-                continue
-
-            # todo: we could differentiate between transcriptions from different participants
-            sentence += message["text"]
-            if sentence.endswith((".", "?", "!")):
-                messages.append({"role": "user", "content": sentence})
-                sentence = ''
-
-                full_response = ""
-                async for response in llm.run_llm_async_sentences(messages):
-                    full_response += response
-                    async for audio in tts.run_tts(response):
-                        await transport.send_queue.put(QueueFrame(FrameType.AUDIO, audio))
-
-                messages.append({"role": "assistant", "content": full_response})
+        tma_in = TranscriptionToLLMMessageAggregator(messages, transport.my_participant_id)
+        tma_out = LLMResponseToLLMMessageAggregator(messages)
+        await tts.run_to_queue(
+            transport.send_queue,
+            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())