diff --git a/src/dailyai/queue_frame.py b/src/dailyai/queue_frame.py index 570c66447..cb0078a89 100644 --- a/src/dailyai/queue_frame.py +++ b/src/dailyai/queue_frame.py @@ -51,6 +51,9 @@ class SpriteQueueFrame(QueueFrame): class TextQueueFrame(QueueFrame): text: str +@dataclass() +class TTSCompletedFrame(QueueFrame): + text: str @dataclass() class TranscriptionQueueFrame(TextQueueFrame): diff --git a/src/dailyai/services/ai_services.py b/src/dailyai/services/ai_services.py index e4b1fe598..9a5dabd58 100644 --- a/src/dailyai/services/ai_services.py +++ b/src/dailyai/services/ai_services.py @@ -14,7 +14,9 @@ from dailyai.queue_frame import ( LLMResponseEndQueueFrame, QueueFrame, TextQueueFrame, + TTSCompletedFrame, TranscriptionQueueFrame, + UserStoppedSpeakingFrame ) from abc import abstractmethod @@ -81,6 +83,11 @@ class AIService: class LLMService(AIService): + + def __init__(self, context): + super().__init__() + self._context = context + @abstractmethod async def run_llm_async(self, messages) -> AsyncGenerator[str, None]: yield "" @@ -90,8 +97,11 @@ class LLMService(AIService): pass async def process_frame(self, frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]: - if isinstance(frame, LLMMessagesQueueFrame): - async for text_chunk in self.run_llm_async(frame.messages): + print(f"##### process frame got a frame, {type(frame)}") + if isinstance(frame, UserStoppedSpeakingFrame): + print( + f"### Got a user stopped speaking frame, context is {self._context}") + async for text_chunk in self.run_llm_async(self._context): yield TextQueueFrame(text_chunk) yield LLMResponseEndQueueFrame() else: @@ -117,6 +127,12 @@ class TTSService(AIService): async def process_frame(self, frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]: if not isinstance(frame, TextQueueFrame): + # We don't want transcription frames, which are a subclass + yield frame + return + + # TODO-CB: Clean this up + if isinstance(frame, TranscriptionQueueFrame): yield frame return @@ -132,6 +148,7 @@ class TTSService(AIService): if text: async for audio_chunk in self.run_tts(text): yield AudioQueueFrame(audio_chunk) + yield TTSCompletedFrame(text) async def finalize(self): if self.current_sentence: diff --git a/src/dailyai/services/azure_ai_services.py b/src/dailyai/services/azure_ai_services.py index 1f3e67cd2..d59acfc47 100644 --- a/src/dailyai/services/azure_ai_services.py +++ b/src/dailyai/services/azure_ai_services.py @@ -42,14 +42,16 @@ class AzureTTSService(TTSService): yield result.audio_data[44:] elif result.reason == ResultReason.Canceled: cancellation_details = result.cancellation_details - self.logger.info("Speech synthesis canceled: {}".format(cancellation_details.reason)) + self.logger.info("Speech synthesis canceled: {}".format( + cancellation_details.reason)) if cancellation_details.reason == CancellationReason.Error: - self.logger.info("Error details: {}".format(cancellation_details.error_details)) + self.logger.info("Error details: {}".format( + cancellation_details.error_details)) class AzureLLMService(LLMService): - def __init__(self, *, api_key, endpoint, api_version="2023-12-01-preview", model): - super().__init__() + def __init__(self, *, api_key, endpoint, api_version="2023-12-01-preview", model, context): + super().__init__(context) self._model: str = model self._client = AsyncAzureOpenAI( @@ -102,7 +104,8 @@ class AzureImageGenServiceREST(ImageGenService): async def run_image_gen(self, sentence) -> tuple[str, bytes]: url = f"{self._azure_endpoint}openai/images/generations:submit?api-version={self._api_version}" - headers = {"api-key": self._api_key, "Content-Type": "application/json"} + headers = {"api-key": self._api_key, + "Content-Type": "application/json"} body = { # Enter your prompt text here "prompt": sentence, diff --git a/src/dailyai/services/base_transport_service.py b/src/dailyai/services/base_transport_service.py index f6f84f09d..ef548b5ad 100644 --- a/src/dailyai/services/base_transport_service.py +++ b/src/dailyai/services/base_transport_service.py @@ -26,6 +26,7 @@ from dailyai.queue_frame import ( SpriteQueueFrame, StartStreamQueueFrame, TranscriptionQueueFrame, + TTSCompletedFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame ) @@ -126,16 +127,21 @@ class BaseTransportService(): self._logger: logging.Logger = logging.getLogger() - def update_messages(self, new_messages: list[dict[str, str]], task: asyncio.Task | None): + def update_messages(self, new_context: list[dict[str, str]], task: asyncio.Task | None): if task: if not task.cancelled(): self._current_phrase = "" - self._messages = new_messages + self._context = new_context - 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 + def append_to_context(self, role, text): + last_context_item = self._context[-1] + if last_context_item and last_context_item['role'] == role: + last_context_item['content'] += f" {text}" + else: + self._context.append({"role": role, "content": text}) + + async def run_pipeline(self, frame, context): + print(f"starting to speak_after_delay, {frame}") print(f"past asyncio sleep, context is {context}") # TODO-CB: This exception for missing class gets eaten! tma_in = LLMUserContextAggregator( @@ -145,7 +151,7 @@ class BaseTransportService(): 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) + await self._runner(frame, tma_in, tma_out) async def run_conversation(self, runner: Iterable[QueueFrame] | AsyncIterable[QueueFrame] @@ -158,21 +164,22 @@ class BaseTransportService(): print(f"got frame of type: {type(frame)}") if isinstance(frame, EndStreamQueueFrame): break - elif not isinstance(frame, TranscriptionQueueFrame): - continue + # elif not isinstance(frame, TranscriptionQueueFrame): + # continue - if frame.participantId == self._my_participant_id: + if hasattr(frame, 'participantId') and frame.participantId == self._my_participant_id: continue if current_response_task: + # TODO-CB: Maybe not always interrupt? Are there frame types we can pass through? current_response_task.cancel() self.interrupt() - self._current_phrase += " " + frame.text + # 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) + self.run_pipeline( + frame, current_llm_context) ) current_response_task.add_done_callback( functools.partial(self.update_messages, current_llm_context) @@ -292,16 +299,18 @@ class BaseTransportService(): if self._vad_state == VADState.STARTING and self._vad_starting_count >= self._vad_start_frames: print( - f'{datetime.datetime.utcnow().isoformat()} queueing start frame') + f'!!! {datetime.datetime.utcnow().isoformat()} queueing start frame') asyncio.run_coroutine_threadsafe( self.receive_queue.put( UserStartedSpeakingFrame()), self._loop ) + print(f"!!! VAD started, calling interrupt") + self.interrupt() self._vad_state = VADState.SPEAKING self._vad_starting_count = 0 if self._vad_state == VADState.STOPPING and self._vad_stopping_count >= self._vad_stop_frames: print( - f'{datetime.datetime.utcnow().isoformat()} queueing stop frame') + f'!!! {datetime.datetime.utcnow().isoformat()} queueing stop frame') asyncio.run_coroutine_threadsafe( self.receive_queue.put( UserStoppedSpeakingFrame()), self._loop @@ -318,6 +327,7 @@ class BaseTransportService(): break def interrupt(self): + print(f"!!! setting interrupt") self._is_interrupted.set() async def get_receive_frames(self) -> AsyncGenerator[QueueFrame, None]: @@ -389,10 +399,13 @@ class BaseTransportService(): # if interrupted, we just pull frames off the queue and discard them if not self._is_interrupted.is_set(): + print( + f"~~~ not interrupted so popping frame of type {type(frame)}") if frame: if isinstance(frame, AudioQueueFrame): chunk = frame.data - + print( + f"~~~ length of this chunk: {len(chunk)}") all_audio_frames.extend(chunk) b.extend(chunk) @@ -407,12 +420,16 @@ class BaseTransportService(): self._set_image(frame.image) elif isinstance(frame, SpriteQueueFrame): self._set_images(frame.images) + elif isinstance(frame, TTSCompletedFrame): + self.append_to_context( + "assistant", frame.text) elif len(b): self.write_frame_to_mic(bytes(b)) b = bytearray() else: # if there are leftover audio bytes, write them now; failing to do so # can cause static in the audio stream. + print(f"!!! interrupted, flushing audio") if len(b): truncated_length = len(b) - (len(b) % 160) self.write_frame_to_mic( diff --git a/src/dailyai/services/daily_transport_service.py b/src/dailyai/services/daily_transport_service.py index 4146b9d9d..9c613d7ab 100644 --- a/src/dailyai/services/daily_transport_service.py +++ b/src/dailyai/services/daily_transport_service.py @@ -289,6 +289,8 @@ class DailyTransportService(BaseTransportService, EventHandler): participantId = message["session_id"] frame = TranscriptionQueueFrame( message["text"], participantId, message["timestamp"]) + if self._my_participant_id and participantId != self._my_participant_id: + self.append_to_context("user", message["text"]) asyncio.run_coroutine_threadsafe( self.receive_queue.put(frame), self._loop) diff --git a/src/dailyai/services/open_ai_services.py b/src/dailyai/services/open_ai_services.py index 2aea60fa9..7cb89c91c 100644 --- a/src/dailyai/services/open_ai_services.py +++ b/src/dailyai/services/open_ai_services.py @@ -10,8 +10,8 @@ from dailyai.services.ai_services import LLMService, ImageGenService class OpenAILLMService(LLMService): - def __init__(self, *, api_key, model="gpt-4"): - super().__init__() + def __init__(self, *, api_key, model="gpt-4", context): + super().__init__(context) self._model = model self._client = AsyncOpenAI(api_key=api_key) diff --git a/src/examples/foundational/06c-listen-respond-interruptible-refactor.py b/src/examples/foundational/06c-listen-respond-interruptible-refactor.py index 5c6f9e9de..ed6b9d1d7 100644 --- a/src/examples/foundational/06c-listen-respond-interruptible-refactor.py +++ b/src/examples/foundational/06c-listen-respond-interruptible-refactor.py @@ -7,6 +7,7 @@ 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.open_ai_services import OpenAILLMService from dailyai.services.ai_services import FrameLogger from examples.foundational.support.runner import configure @@ -29,29 +30,40 @@ async def main(room_url: str, token): mic_enabled=True, mic_sample_rate=16000, camera_enabled=False, + # TODO-CB: Should this be VAD enabled or something? + speaker_enabled=True, + context=context ) - llm = AzureLLMService( - api_key=os.getenv("AZURE_CHATGPT_API_KEY"), - endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"), - model=os.getenv("AZURE_CHATGPT_MODEL")) + # llm = AzureLLMService( + # api_key=os.getenv("AZURE_CHATGPT_API_KEY"), + # endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"), + # model=os.getenv("AZURE_CHATGPT_MODEL"), + # context=context) + llm = OpenAILLMService( + context=context, api_key=os.getenv("OPENAI_CHATGPT_API_KEY")) + tts = AzureTTSService( api_key=os.getenv("AZURE_SPEECH_API_KEY"), region=os.getenv("AZURE_SPEECH_REGION")) + # tts = ElevenLabsTTSService( + # aiohttp_session=session, + # api_key=os.getenv("ELEVENLABS_API_KEY"), + # voice_id=os.getenv("ELEVENLABS_VOICE_ID")) fl = FrameLogger("just outside the innermost layer") - async def run_response(user_speech, tma_in, tma_out): + async def run_response(in_frame, 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)] - ) - ) + # tma_out.run( + llm.run( + # tma_in.run( + fl.run( + [StartStreamQueueFrame(), in_frame] ) - ), + # ) + ) + # ), ) @transport.event_handler("on_first_other_participant_joined")