diff --git a/src/dailyai/services/ai_services.py b/src/dailyai/services/ai_services.py index 282f8adeb..4733cbd6e 100644 --- a/src/dailyai/services/ai_services.py +++ b/src/dailyai/services/ai_services.py @@ -35,7 +35,6 @@ 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( @@ -94,20 +93,15 @@ class LLMService(AIService): function_name = "" arguments = "" if isinstance(frame, LLMMessagesQueueFrame): - 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: {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}") if (function_name and arguments): - print("made it inside") yield LLMFunctionCallFrame(function_name=function_name, arguments=arguments) function_name = "" arguments = "" @@ -134,7 +128,6 @@ 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 @@ -149,7 +142,6 @@ 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) @@ -223,6 +215,6 @@ class FrameLogger(AIService): if isinstance(frame, (AudioQueueFrame, ImageQueueFrame)): self.logger.info(f"{self.prefix}: {type(frame)}") else: - print(f"{self.prefix}: {frame}") + self.logger.info(f"{self.prefix}: {frame}") yield frame diff --git a/src/dailyai/services/base_transport_service.py b/src/dailyai/services/base_transport_service.py index cc081f41c..1686fadf3 100644 --- a/src/dailyai/services/base_transport_service.py +++ b/src/dailyai/services/base_transport_service.py @@ -76,16 +76,12 @@ 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: - print(f"Exception {e}") + self._logger.error(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() @@ -96,6 +92,7 @@ class BaseTransportService(): def stop(self): self._stop_threads.set() + async def stop_when_done(self): await self._wait_for_send_queue_to_empty() @@ -153,7 +150,6 @@ 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 ) @@ -194,7 +190,6 @@ 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() diff --git a/src/dailyai/services/daily_transport_service.py b/src/dailyai/services/daily_transport_service.py index 575a6560b..5a083aa8e 100644 --- a/src/dailyai/services/daily_transport_service.py +++ b/src/dailyai/services/daily_transport_service.py @@ -45,7 +45,8 @@ class DailyTransportService(BaseTransportService, EventHandler): start_transcription: bool = False, **kwargs, ): - super().__init__(**kwargs) # This will call BaseTransportService.__init__ method, not EventHandler + # This will call BaseTransportService.__init__ method, not EventHandler + super().__init__(**kwargs) self._room_url: str = room_url self._bot_name: str = bot_name @@ -80,7 +81,8 @@ class DailyTransportService(BaseTransportService, EventHandler): for handler in self._event_handlers[event_name]: if inspect.iscoroutinefunction(handler): if self._loop: - asyncio.run_coroutine_threadsafe(handler(*args, **kwargs), self._loop) + asyncio.run_coroutine_threadsafe( + handler(*args, **kwargs), self._loop) else: raise Exception( "No event loop to run coroutine. In order to use async event handlers, you must run the DailyTransportService in an asyncio event loop.") @@ -92,7 +94,8 @@ class DailyTransportService(BaseTransportService, EventHandler): def add_event_handler(self, event_name: str, handler): if not event_name.startswith("on_"): - raise Exception(f"Event handler {event_name} must start with 'on_'") + raise Exception( + f"Event handler {event_name} must start with 'on_'") methods = inspect.getmembers(self, predicate=inspect.ismethod) if event_name not in [method[0] for method in methods]: @@ -105,7 +108,8 @@ class DailyTransportService(BaseTransportService, EventHandler): handler, self)] setattr(self, event_name, partial(self._patch_method, event_name)) else: - self._event_handlers[event_name].append(types.MethodType(handler, self)) + self._event_handlers[event_name].append( + types.MethodType(handler, self)) def event_handler(self, event_name: str): def decorator(handler): @@ -149,7 +153,8 @@ class DailyTransportService(BaseTransportService, EventHandler): Daily.select_speaker_device("speaker") self.client.set_user_name(self._bot_name) - self.client.join(self._room_url, self._token, completion=self.call_joined) + self.client.join(self._room_url, self._token, + completion=self.call_joined) self._my_participant_id = self.client.participants()["local"]["id"] self.client.update_inputs( @@ -217,9 +222,11 @@ class DailyTransportService(BaseTransportService, EventHandler): self._other_participant_has_joined = True self.on_first_other_participant_joined() + """ def on_participant_left(self, participant, reason): if len(self.client.participants()) < self._min_others_count + 1: self._stop_threads.set() + """ def on_app_message(self, message, sender): pass @@ -231,8 +238,10 @@ class DailyTransportService(BaseTransportService, EventHandler): participantId = message["participantId"] elif "session_id" in message: participantId = message["session_id"] - frame = TranscriptionQueueFrame(message["text"], participantId, message["timestamp"]) - asyncio.run_coroutine_threadsafe(self.receive_queue.put(frame), self._loop) + frame = TranscriptionQueueFrame( + message["text"], participantId, message["timestamp"]) + asyncio.run_coroutine_threadsafe( + self.receive_queue.put(frame), self._loop) def on_transcription_stopped(self, stopped_by, stopped_by_error): pass @@ -242,3 +251,7 @@ class DailyTransportService(BaseTransportService, EventHandler): def on_transcription_started(self, status): pass + + def stop(self): + super().stop() + self.client.leave() diff --git a/src/examples/foundational/06a-multi-step.py b/src/examples/foundational/06a-multi-step.py index 0b11d108d..d5664b59e 100644 --- a/src/examples/foundational/06a-multi-step.py +++ b/src/examples/foundational/06a-multi-step.py @@ -49,7 +49,6 @@ class ChecklistProcessor(AIService): yield LLMMessagesQueueFrame(self._messages) print(f"past llmmessagesqueueframe yield") async for frame in llm.process_frame(LLMMessagesQueueFrame(self._messages)): - print(f"yielding frame from llm.process_frame: {frame}") yield frame else: yield frame diff --git a/src/examples/foundational/06b-patient-intake.py b/src/examples/foundational/06b-patient-intake.py index 07c24ed5a..53c8df2c2 100644 --- a/src/examples/foundational/06b-patient-intake.py +++ b/src/examples/foundational/06b-patient-intake.py @@ -150,7 +150,7 @@ class ChecklistProcessor(AIService): self._current_step = 0 self._messages = messages self._llm = llm - 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._id = "You are Jessica, an agent for a company called Tri-County Advanced 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.", @@ -165,19 +165,15 @@ class ChecklistProcessor(AIService): async def process_frame(self, frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]: if isinstance(frame, LLMFunctionCallFrame): - print(f"GOT A FUNCTION CALL: {frame}") + print(f"FUNCTION CALL: {frame}") self._current_step += 1 # yield TextQueueFrame(f"We should move on to Step {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), 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 @@ -247,7 +243,10 @@ async def main(room_url: str, token): ) transport.transcription_settings["extra"]["punctuate"] = True - await asyncio.gather(transport.run(), handle_transcriptions()) + try: + await asyncio.gather(transport.run(), handle_transcriptions()) + except (asyncio.CancelledError, KeyboardInterrupt): + transport.stop() if __name__ == "__main__":