From 33ea1f99254a59ba5c4f97b5baf0b3f2ab690e6a Mon Sep 17 00:00:00 2001 From: Chad Bailey Date: Mon, 12 Feb 2024 21:15:12 +0000 Subject: [PATCH] got a weird audio bug happening --- src/dailyai/services/ai_services.py | 13 +- .../services/base_transport_service.py | 30 +++-- src/dailyai/services/elevenlabs_ai_service.py | 3 +- src/dailyai/services/open_ai_services.py | 5 +- .../to_be_updated/cloudflare_ai_service.py | 3 +- .../foundational/06b-patient-intake.py | 120 +++++++++++++++--- 6 files changed, 138 insertions(+), 36 deletions(-) diff --git a/src/dailyai/services/ai_services.py b/src/dailyai/services/ai_services.py index 1bccaeaef..282f8adeb 100644 --- a/src/dailyai/services/ai_services.py +++ b/src/dailyai/services/ai_services.py @@ -35,6 +35,7 @@ class AIService: await queue.put(frame) if add_end_of_stream: + print(f"ESQF: adding end of stream from run_to_queue") await queue.put(EndStreamQueueFrame()) async def run( @@ -89,20 +90,22 @@ class LLMService(AIService): async def run_llm(self, messages) -> str: pass - async def process_frame(self, frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]: + async def process_frame(self, frame: QueueFrame, tool_choice: str = None) -> AsyncGenerator[QueueFrame, None]: function_name = "" arguments = "" if isinstance(frame, LLMMessagesQueueFrame): - async for text_chunk in self.run_llm_async(frame.messages): + print(f"llm process_frame, messages: {frame.messages}") + async for text_chunk in self.run_llm_async(frame.messages, tool_choice): if isinstance(text_chunk, str): - print(f"text") + print(f"text: {text_chunk}") yield TextQueueFrame(text_chunk) elif text_chunk.function: if text_chunk.function.name: function_name += text_chunk.function.name if text_chunk.function.arguments: arguments += text_chunk.function.arguments - print(f"out here, function_name is {function_name}, arguments is {arguments}") + print( + f"out here, function_name is {function_name}, arguments is {arguments}") if (function_name and arguments): print("made it inside") yield LLMFunctionCallFrame(function_name=function_name, arguments=arguments) @@ -131,6 +134,7 @@ class TTSService(AIService): yield bytes() async def process_frame(self, frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]: + print(f" TTS GOT FRAME: {type(frame)}") if not isinstance(frame, TextQueueFrame): yield frame return @@ -145,6 +149,7 @@ class TTSService(AIService): self.current_sentence = "" if text: + print(f" IF TEXT TRUE") 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 139edc6ff..cc081f41c 100644 --- a/src/dailyai/services/base_transport_service.py +++ b/src/dailyai/services/base_transport_service.py @@ -16,6 +16,7 @@ from dailyai.queue_frame import ( StartStreamQueueFrame, ) + class BaseTransportService(): def __init__( @@ -54,16 +55,20 @@ class BaseTransportService(): async def run(self): self._prerun() - async_output_queue_marshal_task = asyncio.create_task(self._marshal_frames()) + async_output_queue_marshal_task = asyncio.create_task( + self._marshal_frames()) - self._camera_thread = threading.Thread(target=self._run_camera, daemon=True) + self._camera_thread = threading.Thread( + target=self._run_camera, daemon=True) self._camera_thread.start() - self._frame_consumer_thread = threading.Thread(target=self._frame_consumer, daemon=True) + self._frame_consumer_thread = threading.Thread( + target=self._frame_consumer, daemon=True) self._frame_consumer_thread.start() if self._speaker_enabled: - self._receive_audio_thread = threading.Thread(target=self._receive_audio, daemon=True) + self._receive_audio_thread = threading.Thread( + target=self._receive_audio, daemon=True) self._receive_audio_thread.start() try: @@ -71,13 +76,16 @@ class BaseTransportService(): time.time() < self._expiration and not self._stop_threads.is_set() ): + print( + f"about to sleep; expire time TF: {time.time() < self._expiration}, stop_threads: {self._stop_threads.is_set()}") await asyncio.sleep(1) except Exception as e: - self._logger.error(f"Exception {e}") + print(f"Exception {e}") raise e self._stop_threads.set() + print(f"ESQF: PAST base transport service while loop, about to stop things") await self.send_queue.put(EndStreamQueueFrame()) await async_output_queue_marshal_task await self.send_queue.join() @@ -145,7 +153,7 @@ class BaseTransportService(): asyncio.run_coroutine_threadsafe( self.receive_queue.put(frame), self._loop ) - + print(f"ESQF: this is the only other endstreamqueueframe I can find") asyncio.run_coroutine_threadsafe( self.receive_queue.put(EndStreamQueueFrame()), self._loop ) @@ -186,6 +194,7 @@ class BaseTransportService(): raise Exception("Unknown type in output queue") for frame in frames: + print(f"GOT FRAME OF TYPE: {type(frame)}") if isinstance(frame, EndStreamQueueFrame): self._logger.info("Stopping frame consumer thread") self._threadsafe_send_queue.task_done() @@ -204,7 +213,8 @@ class BaseTransportService(): len(b) % smallest_write_size ) if truncated_length: - self.write_frame_to_mic(bytes(b[:truncated_length])) + self.write_frame_to_mic( + bytes(b[:truncated_length])) b = b[truncated_length:] elif isinstance(frame, ImageQueueFrame): self._set_image(frame.image) @@ -218,7 +228,8 @@ class BaseTransportService(): # can cause static in the audio stream. if len(b): truncated_length = len(b) - (len(b) % 160) - self.write_frame_to_mic(bytes(b[:truncated_length])) + self.write_frame_to_mic( + bytes(b[:truncated_length])) b = bytearray() if isinstance(frame, StartStreamQueueFrame): @@ -231,5 +242,6 @@ class BaseTransportService(): b = bytearray() except Exception as e: - self._logger.error(f"Exception in frame_consumer: {e}, {len(b)}") + print( + f"Exception in frame_consumer: {e}, {len(b)}") raise e diff --git a/src/dailyai/services/elevenlabs_ai_service.py b/src/dailyai/services/elevenlabs_ai_service.py index e1795aab3..e5495dae9 100644 --- a/src/dailyai/services/elevenlabs_ai_service.py +++ b/src/dailyai/services/elevenlabs_ai_service.py @@ -26,7 +26,8 @@ class ElevenLabsTTSService(TTSService): async def run_tts(self, sentence) -> AsyncGenerator[bytes, None]: url = f"https://api.elevenlabs.io/v1/text-to-speech/{self._voice_id}/stream" payload = {"text": sentence, "model_id": "eleven_turbo_v2"} - querystring = {"output_format": "pcm_16000", "optimize_streaming_latency": 2} + querystring = {"output_format": "pcm_16000", + "optimize_streaming_latency": 2} headers = { "xi-api-key": self._api_key, "Content-Type": "application/json", diff --git a/src/dailyai/services/open_ai_services.py b/src/dailyai/services/open_ai_services.py index 481ec9462..64b9efbf9 100644 --- a/src/dailyai/services/open_ai_services.py +++ b/src/dailyai/services/open_ai_services.py @@ -27,11 +27,11 @@ class OpenAILLMService(LLMService): tools=self._tools ) - async def run_llm_async(self, messages) -> AsyncGenerator[str, None]: + async def run_llm_async(self, messages, tool_choice=None) -> AsyncGenerator[str, None]: messages_for_log = json.dumps(messages) self.logger.debug(f"Generating chat via openai: {messages_for_log}") - chunks = await self._client.chat.completions.create(model=self._model, stream=True, messages=messages, tools=self._tools) + chunks = await self._client.chat.completions.create(model=self._model, stream=True, messages=messages, tools=self._tools, tool_choice=tool_choice) async for chunk in chunks: if len(chunk.choices) == 0: continue @@ -39,6 +39,7 @@ class OpenAILLMService(LLMService): yield chunk.choices[0].delta.content elif chunk.choices[0].delta.tool_calls: yield chunk.choices[0].delta.tool_calls[0] + async def run_llm(self, messages) -> str | None: messages_for_log = json.dumps(messages) self.logger.debug(f"Generating chat via openai: {messages_for_log}") diff --git a/src/dailyai/services/to_be_updated/cloudflare_ai_service.py b/src/dailyai/services/to_be_updated/cloudflare_ai_service.py index b4a810bd5..7d4c4096a 100644 --- a/src/dailyai/services/to_be_updated/cloudflare_ai_service.py +++ b/src/dailyai/services/to_be_updated/cloudflare_ai_service.py @@ -17,7 +17,8 @@ class CloudflareAIService(AIService): # base endpoint, used by the others def run(self, model, input): - response = requests.post(f"{self.api_base_url}{model}", headers=self.headers, json=input) + response = requests.post( + f"{self.api_base_url}{model}", headers=self.headers, json=input) return response.json() # https://developers.cloudflare.com/workers-ai/models/llm/ diff --git a/src/examples/foundational/06b-patient-intake.py b/src/examples/foundational/06b-patient-intake.py index 48560aadc..07c24ed5a 100644 --- a/src/examples/foundational/06b-patient-intake.py +++ b/src/examples/foundational/06b-patient-intake.py @@ -56,9 +56,82 @@ tools = [ } } } + }, + { + "type": "function", + "function": { + "name": "list_allergies", + "description": "Once the user has provided a list of their allergies, call this function.", + "parameters": { + "type": "object", + "properties": { + "allergies": { + "type": "array", + "items": { + "type": "object", + "properties": { + "name": { + "type": "string", + "description": "What the user is allergic to" + } + } + } + } + } + } + } + }, + { + "type": "function", + "function": { + "name": "list_conditions", + "description": "Once the user has provided a list of their medical conditions, call this function.", + "parameters": { + "type": "object", + "properties": { + "conditions": { + "type": "array", + "items": { + "type": "object", + "properties": { + "name": { + "type": "string", + "description": "The user's medical condition" + } + } + } + } + } + } + } + }, + { + "type": "function", + "function": { + "name": "list_visit_reasons", + "description": "Once the user has provided a list of the reasons they are visiting a doctor today, call this function.", + "parameters": { + "type": "object", + "properties": { + "reasons": { + "type": "array", + "items": { + "type": "object", + "properties": { + "name": { + "type": "string", + "description": "The user's reason for visiting the doctor" + } + } + } + } + } + } + } } ] + class TranscriptFilter(AIService): def __init__(self, bot_participant_id=None): super().__init__() @@ -70,40 +143,44 @@ class TranscriptFilter(AIService): if frame.participantId != self.bot_participant_id: yield frame + class ChecklistProcessor(AIService): def __init__(self, messages, llm, *args, **kwargs): super().__init__(*args, **kwargs) self._current_step = 0 self._messages = messages self._llm = llm - self._id = "You are Jessica, an agent for a company called Butt Health Specialists. Your job is to collect important information from the user before they visit a doctor. You're talking to Chad Bailey. You should address the user by their first name and be polite and professional. You're not a medical professional, so you shouldn't provide any advice. Your job is to collect information to give to a doctor." + self._id = "You are Jessica, an agent for a company called Optimum Health Solution Specialists. Your job is to collect important information from the user before they visit a doctor. You're talking to Chad Bailey. You should address the user by their first name and be polite and professional. You're not a medical professional, so you shouldn't provide any advice. Keep your responses short. Your job is to collect information to give to a doctor." self._steps = [ "Start by introducing yourself. Then, ask the user to confirm their identity by telling you their birthday. When they answer with their birthday, call the verify_birthday function.", "You've already confirmed the user's birthday, so don't call the verify_birthday function. Ask the user to list their current prescriptions. If the user responds with one or two prescriptions, ask them to confirm it's the complete list. Make sure each medication also includes the dosage. Once the user has provided all their prescriptions, call the list_prescriptions function.", - "Ask the user if they have any allergies. Once they have listed their allergies or confirmed they don't have any , respond only with ABC.", - "Ask the user if they have any medical conditions the doctor should know about. Once they've answered the question, respond only with ABC." - "Ask the user the reason for their doctor visit today. Once they answer, double-check to make sure they don't have any other health concerns. After that, respond only with ABC.", - "Reply with the user's name, prescriptions, and reason for visit in a JSON object.", + "Don't call the verify_birthday or list_prescription functions. Ask the user if they have any allergies. Once they have listed their allergies or confirmed they don't have any, call the list_allergies function.", + "Don't call the verify_birthday, list_allergies, or list_prescriptions functions. Ask the user if they have any medical conditions the doctor should know about. Once they've answered the question, call the list_conditions function." + "Ask the user the reason for their doctor visit today. Once they answer, double-check to make sure they don't have any other health concerns. After that, call the list_visit_reasons function.", + "Now, thank the user and end the conversation.", "" ] - messages.append({"role": "system", "content": f"{self._id} {self._steps[0]}"}) + messages.append( + {"role": "system", "content": f"{self._id} {self._steps[0]}"}) async def process_frame(self, frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]: if isinstance(frame, LLMFunctionCallFrame): print(f"GOT A FUNCTION CALL: {frame}") self._current_step += 1 # yield TextQueueFrame(f"We should move on to Step {self._current_step}.") - self._messages[0] = {"role": "system", "content": f"{self._id} {self._steps[self._current_step]}"} + self._messages.append({ + "role": "system", "content": f"{self._id} {self._steps[self._current_step]}"}) print(f"NEW MESSAGES ARRAY: {self._messages}") yield LLMMessagesQueueFrame(self._messages) print(f"past llmmessagesqueueframe yield") - async for frame in llm.process_frame(LLMMessagesQueueFrame(self._messages)): + async for frame in llm.process_frame(LLMMessagesQueueFrame(self._messages), tool_choice="none"): print(f"yielding frame from llm.process_frame: {frame}") yield frame else: print(f"non LLM function call frame: {type(frame)}") yield frame + async def main(room_url: str, token): async with aiohttp.ClientSession() as session: global transport @@ -122,15 +199,21 @@ async def main(room_url: str, token): ) # llm = AzureLLMService(api_key=os.getenv("AZURE_CHATGPT_API_KEY"), endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"), model=os.getenv("AZURE_CHATGPT_MODEL")) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_CHATGPT_API_KEY"), model="gpt-4", tools=tools) - # 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="EXAVITQu4vr4xnSDxMaL") + llm = OpenAILLMService(api_key=os.getenv( + "OPENAI_CHATGPT_API_KEY"), model="gpt-4", tools=tools) + # 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="EXAVITQu4vr4xnSDxMaL") messages = [ ] - tma_in = LLMUserContextAggregator(messages, transport._my_participant_id) - tma_out = LLMAssistantContextAggregator(messages, transport._my_participant_id) + tma_in = LLMUserContextAggregator( + messages, transport._my_participant_id) + tma_out = LLMAssistantContextAggregator( + messages, transport._my_participant_id) checklist = ChecklistProcessor(messages, llm) fl = FrameLogger("got transcript") + async def handle_transcriptions(): tf = TranscriptFilter(transport._my_participant_id) await tts.run_to_queue( @@ -144,14 +227,13 @@ async def main(room_url: str, token): transport.get_receive_frames() ) ) - ) + ) ) ) ) - + ) - - + @transport.event_handler("on_first_other_participant_joined") async def on_first_other_participant_joined(transport): fl = FrameLogger("first other participant") @@ -161,9 +243,9 @@ async def main(room_url: str, token): tma_out.run( llm.run([LLMMessagesQueueFrame(messages)]), ) - ) + ) ) - + transport.transcription_settings["extra"]["punctuate"] = True await asyncio.gather(transport.run(), handle_transcriptions())