Make transportservice.run async, and add 02- sample

This commit is contained in:
Moishe Lettvin
2024-01-08 15:38:32 -05:00
parent 712ab97f88
commit 0451ae498f
2 changed files with 53 additions and 12 deletions

View File

@@ -55,17 +55,15 @@ class DailyTransportService(EventHandler):
except RuntimeError: except RuntimeError:
self.loop = None self.loop = None
def patch_method(self, event_name, *args): def patch_method(self, event_name, *args, **kwargs):
for handler in self.event_handlers[event_name]: for handler in self.event_handlers[event_name]:
if inspect.iscoroutinefunction(handler): if inspect.iscoroutinefunction(handler):
if self.loop: if self.loop:
future = asyncio.run_coroutine_threadsafe(handler(*args), self.loop) asyncio.run_coroutine_threadsafe(handler(*args, **kwargs), self.loop)
#concurrent.futures.wait(future)
else: 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.") 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.")
asyncio.run(handler(*args))
else: else:
handler(*args) handler(*args, **kwargs)
def add_event_handler(self, event_name: str, handler): def add_event_handler(self, event_name: str, handler):
if not event_name.startswith("on_"): if not event_name.startswith("on_"):
@@ -152,14 +150,9 @@ class DailyTransportService(EventHandler):
self.my_participant_id = self.client.participants()["local"]["id"] self.my_participant_id = self.client.participants()["local"]["id"]
def run(self) -> None: async def run(self) -> None:
self.configure_daily() self.configure_daily()
self.running_thread = Thread(target=self.run_daily, daemon=True)
self.running_thread.start()
self.running_thread.join()
def run_daily(self):
# TODO: this loop could, I think, be replaced with a timer and an event
self.participant_left = False self.participant_left = False
try: try:
@@ -167,7 +160,7 @@ class DailyTransportService(EventHandler):
self.logger.info(f"{participant_count} participants in room") self.logger.info(f"{participant_count} participants in room")
while time.time() < self.expiration and not self.participant_left and not self.stop_threads.is_set(): while time.time() < self.expiration and not self.participant_left and not self.stop_threads.is_set():
# all handling of incoming transcriptions happens in on_transcription_message # all handling of incoming transcriptions happens in on_transcription_message
time.sleep(1) await asyncio.sleep(1)
except Exception as e: except Exception as e:
self.logger.error(f"Exception {e}") self.logger.error(f"Exception {e}")
finally: finally:

View File

@@ -0,0 +1,48 @@
import asyncio
import re
from typing import AsyncGenerator
from dailyai.output_queue import OutputQueueFrame, FrameType
from dailyai.services.daily_transport_service import DailyTransportService
from dailyai.services.azure_ai_services import AzureLLMService, AzureTTSService
local_joined = False
participant_joined = False
async def main(room_url):
meeting_duration_minutes = 1
transport = DailyTransportService(
room_url,
None,
"Say One Thing From an LLM",
meeting_duration_minutes,
)
transport.mic_enabled = True
tts = AzureTTSService()
llm = AzureLLMService()
messages = [{
"role": "system",
"content": "You are an LLM in a WebRTC session, and your text will be converted to audio. Introduce yourself."
}]
llm_generator: AsyncGenerator[str, None] = llm.run_llm_async(messages)
@transport.event_handler("on_participant_joined")
async def on_participant_joined(transport, participant):
print(local_joined, participant_joined)
current_text = ""
async for text in llm_generator:
print("text", text)
current_text += text
if re.match(r"^.*[.!?]$", text):
async for audio in tts.run_tts(current_text):
transport.output_queue.put(OutputQueueFrame(FrameType.AUDIO_FRAME, audio))
current_text = ""
await transport.run()
if __name__ == "__main__":
asyncio.run(main("https://moishe.daily.co/Lettvins"))