got a weird audio bug happening

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)
 

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

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",

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}")

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/

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.",
+            "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.",
-            "Ask the user if they have any medical conditions the doctor should know about. Once they've answered the question, respond only with ABC."
+            "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, 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, call the list_visit_reasons function.",
-            "Reply with the user's name, prescriptions, and reason for visit in a JSON object.",
+            "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)
+        llm = OpenAILLMService(api_key=os.getenv(
-        # tts = AzureTTSService(api_key=os.getenv("AZURE_SPEECH_API_KEY"), region=os.getenv("AZURE_SPEECH_REGION"))
+            "OPENAI_CHATGPT_API_KEY"), model="gpt-4", tools=tools)
-        tts = ElevenLabsTTSService(aiohttp_session=session, api_key=os.getenv("ELEVENLABS_API_KEY"), voice_id="EXAVITQu4vr4xnSDxMaL")
+        # 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_in = LLMUserContextAggregator(
-        tma_out = LLMAssistantContextAggregator(messages, transport._my_participant_id)
+            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())