From 90d928be99869164465c102c77b9cb5a294e0e4f Mon Sep 17 00:00:00 2001 From: Chad Bailey Date: Wed, 21 Feb 2024 18:57:06 +0000 Subject: [PATCH] first commit of transport conversation runner --- .../services/base_transport_service.py | 60 ++++++++++++++++ .../services/daily_transport_service.py | 1 + .../foundational/06-listen-and-respond.py | 20 +++--- ...c-listen-respond-interruptible-refactor.py | 68 +++++++++++++++++++ 4 files changed, 140 insertions(+), 9 deletions(-) create mode 100644 src/examples/foundational/06c-listen-respond-interruptible-refactor.py diff --git a/src/dailyai/services/base_transport_service.py b/src/dailyai/services/base_transport_service.py index ff366c67e..f6f84f09d 100644 --- a/src/dailyai/services/base_transport_service.py +++ b/src/dailyai/services/base_transport_service.py @@ -1,5 +1,7 @@ from abc import abstractmethod import asyncio +import copy +import functools import itertools import logging import queue @@ -13,6 +15,9 @@ import torchaudio from enum import Enum import datetime +from typing import AsyncGenerator, AsyncIterable, BinaryIO, Iterable +from dailyai.queue_aggregators import LLMAssistantContextAggregator, LLMUserContextAggregator + from dailyai.queue_frame import ( AudioQueueFrame, EndStreamQueueFrame, @@ -20,6 +25,7 @@ from dailyai.queue_frame import ( QueueFrame, SpriteQueueFrame, StartStreamQueueFrame, + TranscriptionQueueFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame ) @@ -88,6 +94,7 @@ class BaseTransportService(): self._fps = kwargs.get("fps") or 8 self._vad_start_s = kwargs.get("vad_start_s") or 0.2 self._vad_stop_s = kwargs.get("vad_stop_s") or 1.2 + self._context = kwargs.get("context") or [] self._vad_samples = 1536 vad_frame_s = self._vad_samples / SAMPLE_RATE @@ -107,6 +114,7 @@ class BaseTransportService(): self._images = None self._user_is_speaking = False + self._current_phrase = "" try: self._loop: asyncio.AbstractEventLoop | None = asyncio.get_running_loop() @@ -118,6 +126,58 @@ class BaseTransportService(): self._logger: logging.Logger = logging.getLogger() + def update_messages(self, new_messages: list[dict[str, str]], task: asyncio.Task | None): + if task: + if not task.cancelled(): + self._current_phrase = "" + self._messages = new_messages + + async def speak_after_delay(self, user_speech, context): + print(f"starting to speak_after_delay, {user_speech}") + await asyncio.sleep(0) # self._delay_before_speech_seconds + # TODO-CB: I think this needs to go + print(f"past asyncio sleep, context is {context}") + # TODO-CB: This exception for missing class gets eaten! + tma_in = LLMUserContextAggregator( + context, self._my_participant_id, complete_sentences=False + ) + tma_out = LLMAssistantContextAggregator( + context, self._my_participant_id + ) + print(f"about to call the runner, tma_in is {tma_in}") + await self._runner(user_speech, tma_in, tma_out) + + async def run_conversation(self, runner: Iterable[QueueFrame] + | AsyncIterable[QueueFrame] + | asyncio.Queue[QueueFrame], + ) -> AsyncGenerator[QueueFrame, None]: + current_response_task = None + self._runner = runner + + async for frame in self.get_receive_frames(): + print(f"got frame of type: {type(frame)}") + if isinstance(frame, EndStreamQueueFrame): + break + elif not isinstance(frame, TranscriptionQueueFrame): + continue + + if frame.participantId == self._my_participant_id: + continue + + if current_response_task: + current_response_task.cancel() + self.interrupt() + + self._current_phrase += " " + frame.text + current_llm_context = copy.deepcopy(self._context) + current_response_task = asyncio.create_task( + self.speak_after_delay( + self._current_phrase, current_llm_context) + ) + current_response_task.add_done_callback( + functools.partial(self.update_messages, current_llm_context) + ) + async def run(self): self._prerun() diff --git a/src/dailyai/services/daily_transport_service.py b/src/dailyai/services/daily_transport_service.py index 72a1c3fd9..4146b9d9d 100644 --- a/src/dailyai/services/daily_transport_service.py +++ b/src/dailyai/services/daily_transport_service.py @@ -281,6 +281,7 @@ class DailyTransportService(BaseTransportService, EventHandler): def on_transcription_message(self, message: dict): if self._loop: + print(f"transcription: {message}") participantId = "" if "participantId" in message: participantId = message["participantId"] diff --git a/src/examples/foundational/06-listen-and-respond.py b/src/examples/foundational/06-listen-and-respond.py index c67f696c0..5b57b5b5d 100644 --- a/src/examples/foundational/06-listen-and-respond.py +++ b/src/examples/foundational/06-listen-and-respond.py @@ -9,6 +9,12 @@ from dailyai.services.ai_services import FrameLogger async def main(room_url: str, token): + context = [ + { + "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.", + }, + ] transport = DailyTransportService( room_url, token, @@ -18,7 +24,8 @@ async def main(room_url: str, token): mic_enabled=True, mic_sample_rate=16000, camera_enabled=False, - speaker_enabled=True + speaker_enabled=True, + context=context ) llm = AzureLLMService( @@ -35,17 +42,11 @@ async def main(room_url: str, token): 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.", - }, - ] tma_in = LLMUserContextAggregator( - messages, transport._my_participant_id) + context, transport._my_participant_id) tma_out = LLMAssistantContextAggregator( - messages, transport._my_participant_id) + context, transport._my_participant_id) await tts.run_to_queue( transport.send_queue, tma_out.run( @@ -60,6 +61,7 @@ async def main(room_url: str, token): ) transport.transcription_settings["extra"]["punctuate"] = True + transport.transcription_settings["extra"]["endpointing"] = True await asyncio.gather(transport.run(), handle_transcriptions()) diff --git a/src/examples/foundational/06c-listen-respond-interruptible-refactor.py b/src/examples/foundational/06c-listen-respond-interruptible-refactor.py new file mode 100644 index 000000000..5c6f9e9de --- /dev/null +++ b/src/examples/foundational/06c-listen-respond-interruptible-refactor.py @@ -0,0 +1,68 @@ +import asyncio +import aiohttp +import os +from dailyai.conversation_wrappers import InterruptibleConversationWrapper + +from dailyai.queue_frame import StartStreamQueueFrame, TextQueueFrame +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 +from dailyai.services.ai_services import FrameLogger + +from examples.foundational.support.runner import configure + + +async def main(room_url: str, token): + async with aiohttp.ClientSession() as session: + context = [ + { + "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.", + }, + ] + transport = DailyTransportService( + room_url, + token, + "Respond bot", + duration_minutes=5, + start_transcription=True, + mic_enabled=True, + mic_sample_rate=16000, + camera_enabled=False, + ) + + llm = AzureLLMService( + api_key=os.getenv("AZURE_CHATGPT_API_KEY"), + endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"), + model=os.getenv("AZURE_CHATGPT_MODEL")) + tts = AzureTTSService( + api_key=os.getenv("AZURE_SPEECH_API_KEY"), + region=os.getenv("AZURE_SPEECH_REGION")) + fl = FrameLogger("just outside the innermost layer") + + 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( + fl.run( + [StartStreamQueueFrame(), TextQueueFrame(user_speech)] + ) + ) + ) + ), + ) + + @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) + + transport.transcription_settings["extra"]["endpointing"] = True + transport.transcription_settings["extra"]["punctuate"] = True + await asyncio.gather(transport.run(), transport.run_conversation(run_response)) + + +if __name__ == "__main__": + (url, token) = configure() + asyncio.run(main(url, token))