diff --git a/src/dailyai/queue_aggregators.py b/src/dailyai/queue_aggregators.py new file mode 100644 index 000000000..a7d815d85 --- /dev/null +++ b/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 diff --git a/src/dailyai/queue_frame.py b/src/dailyai/queue_frame.py index 7e77ff89d..9d6ce6407 100644 --- a/src/dailyai/queue_frame.py +++ b/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: diff --git a/src/dailyai/services/ai_services.py b/src/dailyai/services/ai_services.py index 9ad8e0399..fc7002d92 100644 --- a/src/dailyai/services/ai_services.py +++ b/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: diff --git a/src/dailyai/services/daily_transport_service.py b/src/dailyai/services/daily_transport_service.py index 5d300cad3..f71e19268 100644 --- a/src/dailyai/services/daily_transport_service.py +++ b/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): diff --git a/src/samples/foundational/06-listen-and-respond.py b/src/samples/foundational/06-listen-and-respond.py index ef34d1832..50e8c1cc3 100644 --- a/src/samples/foundational/06-listen-and-respond.py +++ b/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())