wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

cleanup and ctrl-c handling

Browse Source
This commit is contained in:
Chad Bailey
2024-02-13 16:28:28 +00:00
parent 33ea1f9925
commit 4548f91fdc
5 changed files with 29 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/dailyai/services/ai_services.py
View File

@@ -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

9
src/dailyai/services/base_transport_service.py
View File

@@ -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()

27
src/dailyai/services/daily_transport_service.py
View File

@@ -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()

1
src/examples/foundational/06a-multi-step.py
View File

@@ -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

13
src/examples/foundational/06b-patient-intake.py
View File

@@ -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__":