diff --git a/src/dailyai/queue_aggregators.py b/src/dailyai/queue_aggregators.py index 80c99bc19..3d9d93c39 100644 --- a/src/dailyai/queue_aggregators.py +++ b/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], - bot_participant_id=None, - complete_sentences=True): - super().__init__(messages, "user", bot_participant_id, complete_sentences, pass_through=False) + 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): diff --git a/src/dailyai/queue_frame.py b/src/dailyai/queue_frame.py index b8602d1b2..48be056c7 100644 --- a/src/dailyai/queue_frame.py +++ b/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 diff --git a/src/dailyai/services/ai_services.py b/src/dailyai/services/ai_services.py index a899c4d64..4df4c429b 100644 --- a/src/dailyai/services/ai_services.py +++ b/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) diff --git a/src/dailyai/services/base_transport_service.py b/src/dailyai/services/base_transport_service.py index 1686fadf3..113fe93dc 100644 --- a/src/dailyai/services/base_transport_service.py +++ b/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() diff --git a/src/dailyai/services/daily_transport_service.py b/src/dailyai/services/daily_transport_service.py index 5a083aa8e..fed6fd661 100644 --- a/src/dailyai/services/daily_transport_service.py +++ b/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() diff --git a/src/examples/foundational/06b-patient-intake.py b/src/examples/foundational/06b-patient-intake.py index 22f596237..de1043324 100644 --- a/src/examples/foundational/06b-patient-intake.py +++ b/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( - tma_out.run( - llm.run( - tma_in.run( - tf.run( - fl.run( - transport.get_receive_frames() - ) - ) - ) + checklist.run( + tma_out.run( + llm.run( + tma_in.run( + [StartStreamQueueFrame(), TextQueueFrame(user_speech)] ) ) ) ) - ) + 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()