Use more flexibile aggregator

src/dailyai/queue_aggregators.py

@@ -24,44 +24,36 @@ class QueueTee:
             for queue in output_queues:
                 await queue.put(frame)
 
-class TranscriptionToLLMMessageAggregator(AIService):
+class LLMContextAggregator(AIService):
-    def __init__(self, messages, bot_participant_id):
+    def __init__(self, messages: list[dict], role:str, bot_participant_id=None):
         self.messages = messages
         self.bot_participant_id = bot_participant_id
+        self.role = role
         self.sentence = ""
 
     async def process_frame(self, frame:QueueFrame) -> AsyncGenerator[QueueFrame, None]:
-        if frame.frame_type != FrameType.TRANSCRIPTION:
+        content: str = ""
-            return
 
-        message = frame.frame_data
+        if frame.frame_type == FrameType.TRANSCRIPTION:
-        if not isinstance(message, dict):
+            message = frame.frame_data
-            return
+            if not isinstance(message, dict):
+                return
 
-        if message["session_id"] == self.bot_participant_id:
+            if message["session_id"] == self.bot_participant_id:
-            return
+                return
 
-        print("transcription to message", frame)
+            content = message["text"]
+        elif frame.frame_type == FrameType.TEXT:
+            if not isinstance(frame.frame_data, str):
+                return
 
-        # todo: we could differentiate between transcriptions from different participants
+            content = frame.frame_data
-        self.sentence += message["text"]
+
+        # todo: we should differentiate between transcriptions from different participants
+        self.sentence += content
         if self.sentence.endswith((".", "?", "!")):
-            self.messages.append({"role": "user", "content": self.sentence})
+            self.messages.append({"role": self.role, "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/services/ai_services.py

@@ -112,8 +112,12 @@ class TTSService(AIService):
         yield bytes()
 
     async def process_frame(self, frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]:
-        if frame.frame_type != FrameType.TEXT or type(frame.frame_data) != str:
+        if frame.frame_type != FrameType.TEXT:
-            raise Exception(f"TTS service requires a string for the data field, got {frame.frame_type} and frame_data type {type(frame.frame_data)}")
+            yield frame
+            return
+
+        if not isinstance(frame.frame_data, str):
+            raise(Exception(f"Invalid data type in frame type: {frame.frame_type}, type: {type(frame.frame_data)}"))
 
         text: str | None = None
         if not self.aggregate_sentences:

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

@@ -6,10 +6,7 @@ import urllib.parse
 
 from dailyai.services.daily_transport_service import DailyTransportService
 from dailyai.services.azure_ai_services import AzureLLMService, AzureTTSService
-from dailyai.queue_aggregators import (
+from dailyai.queue_aggregators import LLMContextAggregator
-    TranscriptionToLLMMessageAggregator,
-    LLMResponseToLLMMessageAggregator,
-)
 
 async def main(room_url:str, token):
     global transport
@@ -38,8 +35,12 @@ async def main(room_url:str, token):
             {"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 = TranscriptionToLLMMessageAggregator(messages, transport.my_participant_id)
+        tma_in = LLMContextAggregator(
-        tma_out = LLMResponseToLLMMessageAggregator(messages)
+            messages, "user", transport.my_participant_id
+        )
+        tma_out = LLMContextAggregator(
+            messages, "assistant", transport.my_participant_id
+        )
         await tts.run_to_queue(
             transport.send_queue,
             tma_out.run(