updates and chat hack

src/dailyai/queue_aggregators.py

@@ -61,28 +61,38 @@ class LLMContextAggregator(AIService):
 
         # TODO: split up transcription by participant
         if self.complete_sentences:
-            self.sentence += frame.text # type: ignore -- the linter thinks this isn't a TextQueueFrame, even though we check it above
+            # type: ignore -- the linter thinks this isn't a TextQueueFrame, even though we check it above
+            self.sentence += frame.text
             if self.sentence.endswith((".", "?", "!")):
-                self.messages.append({"role": self.role, "content": self.sentence})
+                self.messages.append(
+                    {"role": self.role, "content": self.sentence})
                 self.sentence = ""
+                for message in self.messages:
+                    print(f"{message['role']}: {message['content']}")
                 yield LLMMessagesQueueFrame(self.messages)
         else:
-            self.messages.append({"role": self.role, "content": frame.text})  # type: ignore -- the linter thinks this isn't a TextQueueFrame, even though we check it above
+            # type: ignore -- the linter thinks this isn't a TextQueueFrame, even though we check it above
+            self.messages.append({"role": self.role, "content": frame.text})
+            for message in self.messages:
+                print(f"{message['role']}: {message['content']}")
             yield LLMMessagesQueueFrame(self.messages)
 
     async def finalize(self) -> AsyncGenerator[QueueFrame, None]:
         # Send any dangling words that weren't finished with punctuation.
         if self.complete_sentences and self.sentence:
             self.messages.append({"role": self.role, "content": self.sentence})
+            for message in self.messages:
+                print(f"{message['role']}: {message['content']}")
             yield LLMMessagesQueueFrame(self.messages)
 
 
 class LLMUserContextAggregator(LLMContextAggregator):
     def __init__(self,
-            messages: list[dict],
+                 messages: list[dict],
-            bot_participant_id=None,
+                 bot_participant_id=None,
-            complete_sentences=True):
+                 complete_sentences=True):
-        super().__init__(messages, "user", bot_participant_id, complete_sentences, pass_through=False)
+        super().__init__(messages, "user", bot_participant_id,
+                         complete_sentences, pass_through=False)
 
 
 class LLMAssistantContextAggregator(LLMContextAggregator):

src/dailyai/queue_frame.py

@@ -18,14 +18,22 @@ class StartStreamQueueFrame(ControlQueueFrame):
 class EndStreamQueueFrame(ControlQueueFrame):
     pass
 
+
 class LLMResponseEndQueueFrame(QueueFrame):
     pass
 
+
+@dataclass()
+class ChatMessageQueueFrame(QueueFrame):
+    message: str
+
+
 @dataclass()
 class LLMFunctionCallFrame(QueueFrame):
     function_name: str
     arguments: str
 
+
 @dataclass()
 class AudioQueueFrame(QueueFrame):
     data: bytes

src/dailyai/services/ai_services.py

@@ -12,6 +12,7 @@ from dailyai.queue_frame import (
     LLMMessagesQueueFrame,
     LLMResponseEndQueueFrame,
     LLMFunctionCallFrame,
+    ChatMessageQueueFrame,
     QueueFrame,
     TextQueueFrame,
     TranscriptionQueueFrame,
@@ -145,6 +146,7 @@ class TTSService(AIService):
                 self.current_sentence = ""
 
         if text:
+            yield ChatMessageQueueFrame(message=text)
             async for audio_chunk in self.run_tts(text):
                 yield AudioQueueFrame(audio_chunk)
 

src/dailyai/services/base_transport_service.py

@@ -9,6 +9,7 @@ from typing import AsyncGenerator
 
 from dailyai.queue_frame import (
     AudioQueueFrame,
+    ChatMessageQueueFrame,
     EndStreamQueueFrame,
     ImageQueueFrame,
     QueueFrame,
@@ -92,7 +93,6 @@ class BaseTransportService():
 
     def stop(self):
         self._stop_threads.set()
-        
 
     async def stop_when_done(self):
         await self._wait_for_send_queue_to_empty()
@@ -215,6 +215,8 @@ class BaseTransportService():
                                 self._set_image(frame.image)
                             elif isinstance(frame, SpriteQueueFrame):
                                 self._set_images(frame.images)
+                            elif isinstance(frame, ChatMessageQueueFrame):
+                                self._send_chat_message(frame)
                         elif len(b):
                             self.write_frame_to_mic(bytes(b))
                             b = bytearray()

src/dailyai/services/daily_transport_service.py

@@ -229,6 +229,12 @@ class DailyTransportService(BaseTransportService, EventHandler):
     """
 
     def on_app_message(self, message, sender):
+        print(f"app message: {message}")
+        if self._loop:
+            frame = TranscriptionQueueFrame(
+                message["message"], message["name"], message["date"])
+            asyncio.run_coroutine_threadsafe(
+                self.receive_queue.put(frame), self._loop)
         pass
 
     def on_transcription_message(self, message: dict):
@@ -252,6 +258,10 @@ class DailyTransportService(BaseTransportService, EventHandler):
     def on_transcription_started(self, status):
         pass
 
+    def _send_chat_message(self, frame):
+        self.client.send_app_message(
+            {'message': frame.message, 'event': 'chat-msg', 'name': self._bot_name, 'date': time.time(), 'room': 'main-room'})
+
     def stop(self):
         super().stop()
         self.client.leave()

src/examples/foundational/06b-patient-intake.py

@@ -12,7 +12,7 @@ from dailyai.queue_aggregators import LLMAssistantContextAggregator, LLMContextA
 from examples.foundational.support.runner import configure
 from dailyai.queue_frame import LLMMessagesQueueFrame, TranscriptionQueueFrame, QueueFrame, TextQueueFrame, LLMFunctionCallFrame, LLMResponseEndQueueFrame, StartStreamQueueFrame
 from dailyai.services.ai_services import FrameLogger, AIService
-
+from dailyai.conversation_wrappers import InterruptibleConversationWrapper
 
 import logging
 logging.basicConfig(level=logging.ERROR)
@@ -272,28 +272,32 @@ async def main(room_url: str, token):
         fl = FrameLogger("got transcript")
         fl2 = FrameLogger("just above the checklist")
 
-        async def handle_transcriptions():
+        async def handle_transcriptions(user_speech, tma_in, tma_out):
             tf = TranscriptFilter(transport._my_participant_id)
             await tts.run_to_queue(
                 transport.send_queue,
-                fl2.run(
+                checklist.run(
-                    checklist.run(
+                    tma_out.run(
-                        tma_out.run(
+                        llm.run(
-                            llm.run(
+                            tma_in.run(
-                                tma_in.run(
+                                [StartStreamQueueFrame(), TextQueueFrame(user_speech)]
-                                    tf.run(
-                                        fl.run(
-                                            transport.get_receive_frames()
-                                        )
-                                    )
-                                )
                             )
                         )
                     )
                 )
-
             )
 
+        async def run_conversation():
+
+            conversation_wrapper = InterruptibleConversationWrapper(
+                frame_generator=transport.get_receive_frames,
+                runner=handle_transcriptions,
+                interrupt=transport.interrupt,
+                my_participant_id=transport._my_participant_id,
+                llm_messages=messages,
+            )
+            await conversation_wrapper.run_conversation()
+
         @transport.event_handler("on_first_other_participant_joined")
         async def on_first_other_participant_joined(transport):
             fl = FrameLogger("first other participant")
@@ -308,7 +312,7 @@ async def main(room_url: str, token):
 
         transport.transcription_settings["extra"]["punctuate"] = True
         try:
-            await asyncio.gather(transport.run(), handle_transcriptions())
+            await asyncio.gather(transport.run(), run_conversation())
         except (asyncio.CancelledError, KeyboardInterrupt):
             transport.stop()