diff --git a/src/dailyai/services/base_transport_service.py b/src/dailyai/services/base_transport_service.py index 490489b87..6ad51261d 100644 --- a/src/dailyai/services/base_transport_service.py +++ b/src/dailyai/services/base_transport_service.py @@ -18,6 +18,7 @@ from dailyai.pipeline.frames import ( QueueFrame, SpriteQueueFrame, StartStreamQueueFrame, + TranscriptionQueueFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame ) @@ -108,6 +109,8 @@ class BaseTransportService(): self.send_queue = asyncio.Queue() self.receive_queue = asyncio.Queue() + self.completed_queue = asyncio.Queue() + self._threadsafe_send_queue = queue.Queue() self._images = None @@ -167,21 +170,54 @@ class BaseTransportService(): if self._vad_enabled: self._vad_thread.join() - async def run_pipeline(self, pipeline:Pipeline, allow_interruptions=True): + async def run_uninterruptible_pipeline(self, pipeline:Pipeline): pipeline.set_sink(self.send_queue) - if not allow_interruptions: - pipeline.set_source(self.receive_queue) - await pipeline.run_pipeline() - else: - source_queue = asyncio.Queue() - pipeline.set_source(source_queue) - pipeline.set_sink(self.send_queue) - pipeline_task = asyncio.create_task(pipeline.run_pipeline()) + pipeline.set_source(self.receive_queue) + await pipeline.run_pipeline() - async for frame in self.get_receive_frames(): + async def run_interruptible_pipeline(self, pipeline:Pipeline, allow_interruptions=True, pre_processor=None, post_processor=None): + pipeline.set_sink(self.send_queue) + source_queue = asyncio.Queue() + pipeline.set_source(source_queue) + pipeline.set_sink(self.send_queue) + pipeline_task = asyncio.create_task(pipeline.run_pipeline()) + + async def yield_frame(frame:QueueFrame) -> AsyncGenerator[QueueFrame, None]: + yield frame + + async def post_process(post_processor): + if not post_processor: + return + + while True: + frame = await self.completed_queue.get() + print("post-processing frame: ", frame.__class__.__name__) + await post_processor.process_frame(frame) + + if isinstance(frame, EndStreamQueueFrame): + break + + post_process_task = asyncio.create_task(post_process(post_processor)) + + async for frame in self.get_receive_frames(): + print("Got frame: ", frame.__class__.__name__) + if isinstance(frame, UserStartedSpeakingFrame): + pipeline_task.cancel() + self.interrupt() + pipeline_task = asyncio.create_task(pipeline.run_pipeline()) + + if pre_processor: + frame_generator = pre_processor.process_frame(frame) + else: + frame_generator = yield_frame(frame) + + async for frame in frame_generator: await source_queue.put(frame) + if isinstance(frame, EndStreamQueueFrame): + break + await asyncio.gather(pipeline_task, post_process_task) def _post_run(self): # Note that this function must be idempotent! It can be called multiple times @@ -341,6 +377,10 @@ class BaseTransportService(): if isinstance(frame, EndStreamQueueFrame): self._logger.info("Stopping frame consumer thread") self._threadsafe_send_queue.task_done() + if self._loop: + asyncio.run_coroutine_threadsafe( + self.completed_queue.put(frame), self._loop + ) return # if interrupted, we just pull frames off the queue and discard them @@ -365,6 +405,11 @@ class BaseTransportService(): elif len(b): self.write_frame_to_mic(bytes(b)) b = bytearray() + + if self._loop: + asyncio.run_coroutine_threadsafe( + self.completed_queue.put(frame), self._loop + ) else: # if there are leftover audio bytes, write them now; failing to do so # can cause static in the audio stream. diff --git a/src/examples/foundational/06-listen-and-respond.py b/src/examples/foundational/06-listen-and-respond.py index 044105f33..c70bc2acb 100644 --- a/src/examples/foundational/06-listen-and-respond.py +++ b/src/examples/foundational/06-listen-and-respond.py @@ -5,8 +5,8 @@ from dailyai.pipeline.pipeline import Pipeline from dailyai.services.daily_transport_service import DailyTransportService from dailyai.services.azure_ai_services import AzureLLMService, AzureTTSService from dailyai.services.ai_services import FrameLogger -from dailyai.pipeline.aggregators import LLMAssistantContextAggregator, LLMContextAggregator, LLMUserContextAggregator -from support.runner import configure +from dailyai.pipeline.aggregators import LLMAssistantContextAggregator, LLMUserContextAggregator +from examples.foundational.support.runner import configure async def main(room_url: str, token): @@ -46,8 +46,6 @@ async def main(room_url: str, token): tma_in = LLMUserContextAggregator(messages, transport._my_participant_id) tma_out = LLMAssistantContextAggregator(messages, transport._my_participant_id) pipeline = Pipeline( - source=transport.receive_queue, - sink=transport.send_queue, processors=[ fl, tma_in, @@ -57,7 +55,7 @@ async def main(room_url: str, token): tts ], ) - await pipeline.run_pipeline() + await transport.run_pipeline(pipeline) transport.transcription_settings["extra"]["endpointing"] = True transport.transcription_settings["extra"]["punctuate"] = True diff --git a/src/examples/foundational/07-interruptible.py b/src/examples/foundational/07-interruptible.py index b49532cbc..59c5c752c 100644 --- a/src/examples/foundational/07-interruptible.py +++ b/src/examples/foundational/07-interruptible.py @@ -2,8 +2,11 @@ import asyncio import aiohttp import os from dailyai.conversation_wrappers import InterruptibleConversationWrapper +from dailyai.pipeline.aggregators import LLMAssistantContextAggregator, LLMUserContextAggregator from dailyai.pipeline.frames import StartStreamQueueFrame, TextQueueFrame +from dailyai.pipeline.pipeline import Pipeline +from dailyai.services.ai_services import FrameLogger from dailyai.services.daily_transport_service import DailyTransportService from dailyai.services.azure_ai_services import AzureLLMService, AzureTTSService from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService @@ -32,17 +35,7 @@ async def main(room_url: str, token): api_key=os.getenv("AZURE_SPEECH_API_KEY"), region=os.getenv("AZURE_SPEECH_REGION")) - async def run_response(user_speech, tma_in, tma_out): - await tts.run_to_queue( - transport.send_queue, - tma_out.run( - llm.run( - tma_in.run( - [StartStreamQueueFrame(), TextQueueFrame(user_speech)] - ) - ) - ), - ) + pipeline = Pipeline([FrameLogger(), llm, FrameLogger(), tts]) @transport.event_handler("on_first_other_participant_joined") async def on_first_other_participant_joined(transport): @@ -53,14 +46,15 @@ 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."}, ] - conversation_wrapper = InterruptibleConversationWrapper( - frame_generator=transport.get_receive_frames, - runner=run_response, - interrupt=transport.interrupt, - my_participant_id=transport._my_participant_id, - llm_messages=messages, + await transport.run_interruptible_pipeline( + pipeline, + post_processor=LLMAssistantContextAggregator( + messages, transport._my_participant_id + ), + pre_processor=LLMUserContextAggregator( + messages, transport._my_participant_id, complete_sentences=False + ), ) - await conversation_wrapper.run_conversation() transport.transcription_settings["extra"]["punctuate"] = False await asyncio.gather(transport.run(), run_conversation())