From b32448e967266c3dcce8d973c924a3eaaf6743f6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 24 Oct 2024 17:58:57 -0700 Subject: [PATCH] transports(daily): make functions async --- CHANGELOG.md | 5 ++ examples/canonical-metrics/bot.py | 2 +- examples/chatbot-audio-recording/bot.py | 2 +- examples/deployment/flyio-example/bot.py | 2 +- examples/dialin-chatbot/bot_daily.py | 2 +- examples/dialin-chatbot/bot_twilio.py | 2 +- .../foundational/06-listen-and-respond.py | 2 +- examples/foundational/06a-image-sync.py | 2 +- examples/foundational/07-interruptible-vad.py | 2 +- examples/foundational/07-interruptible.py | 2 +- .../07a-interruptible-anthropic.py | 2 +- .../07b-interruptible-langchain.py | 2 +- .../07d-interruptible-elevenlabs.py | 2 +- .../foundational/07e-interruptible-playht.py | 2 +- .../foundational/07f-interruptible-azure.py | 2 +- .../07g-interruptible-openai-tts.py | 2 +- .../07h-interruptible-openpipe.py | 2 +- .../foundational/07i-interruptible-xtts.py | 2 +- .../foundational/07j-interruptible-gladia.py | 2 +- .../foundational/07k-interruptible-lmnt.py | 2 +- .../07l-interruptible-together.py | 2 +- .../foundational/07m-interruptible-aws.py | 2 +- .../foundational/07n-interruptible-google.py | 2 +- .../07o-interruptible-assemblyai.py | 2 +- examples/foundational/09-mirror.py | 2 +- examples/foundational/09a-local-mirror.py | 2 +- examples/foundational/10-wake-phrase.py | 2 +- examples/foundational/11-sound-effects.py | 2 +- examples/foundational/12-describe-video.py | 4 +- .../12a-describe-video-gemini-flash.py | 4 +- .../foundational/12b-describe-video-gpt-4o.py | 4 +- .../12c-describe-video-anthropic.py | 4 +- examples/foundational/14-function-calling.py | 2 +- .../14a-function-calling-anthropic.py | 2 +- .../14b-function-calling-anthropic-video.py | 4 +- .../14c-function-calling-together.py | 2 +- .../14d-function-calling-video.py | 4 +- .../14e-function-calling-gemini.py | 4 +- examples/foundational/15-switch-voices.py | 2 +- examples/foundational/15a-switch-languages.py | 2 +- .../16-gpu-container-local-bot.py | 2 +- examples/foundational/17-detect-user-idle.py | 2 +- .../foundational/19-openai-realtime-beta.py | 2 +- .../20a-persistent-context-openai.py | 2 +- .../20b-persistent-context-openai-realtime.py | 2 +- .../20c-persistent-context-anthropic.py | 2 +- .../20d-persistent-context-gemini.py | 8 +- examples/moondream-chatbot/bot.py | 4 +- examples/patient-intake/bot.py | 2 +- examples/storytelling-chatbot/src/bot.py | 2 +- examples/studypal/studypal.py | 2 +- examples/translation-chatbot/bot.py | 2 +- src/pipecat/transports/services/daily.py | 76 +++++++++++-------- 53 files changed, 112 insertions(+), 93 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5b8771a7b..268ca8504 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- The following `DailyTransport` functions are now `async` which means they need + to be awaited: `start_dialout`, `stop_dialout`, `start_recording`, + `stop_recording`, `capture_participant_transcription` and + `capture_participant_video`. + - Changed default output sample rate to 24000. This changes all TTS service to output to 24000 and also the default output transport sample rate. This improves audio quality at the cost of some extra bandwidth. diff --git a/examples/canonical-metrics/bot.py b/examples/canonical-metrics/bot.py index 746d67dae..efe4823fb 100644 --- a/examples/canonical-metrics/bot.py +++ b/examples/canonical-metrics/bot.py @@ -124,7 +124,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) await task.queue_frames([LLMMessagesFrame(messages)]) @transport.event_handler("on_participant_left") diff --git a/examples/chatbot-audio-recording/bot.py b/examples/chatbot-audio-recording/bot.py index 2542a9e40..708de522d 100644 --- a/examples/chatbot-audio-recording/bot.py +++ b/examples/chatbot-audio-recording/bot.py @@ -123,7 +123,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) await task.queue_frames([LLMMessagesFrame(messages)]) @transport.event_handler("on_participant_left") diff --git a/examples/deployment/flyio-example/bot.py b/examples/deployment/flyio-example/bot.py index 77de3f3ab..7c69f62bd 100644 --- a/examples/deployment/flyio-example/bot.py +++ b/examples/deployment/flyio-example/bot.py @@ -75,7 +75,7 @@ async def main(room_url: str, token: str): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) await task.queue_frames([LLMMessagesFrame(messages)]) @transport.event_handler("on_participant_left") diff --git a/examples/dialin-chatbot/bot_daily.py b/examples/dialin-chatbot/bot_daily.py index dee0589fb..79e8665b3 100644 --- a/examples/dialin-chatbot/bot_daily.py +++ b/examples/dialin-chatbot/bot_daily.py @@ -81,7 +81,7 @@ async def main(room_url: str, token: str, callId: str, callDomain: str): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) await task.queue_frames([LLMMessagesFrame(messages)]) @transport.event_handler("on_participant_left") diff --git a/examples/dialin-chatbot/bot_twilio.py b/examples/dialin-chatbot/bot_twilio.py index 95fef893e..135fcf4ea 100644 --- a/examples/dialin-chatbot/bot_twilio.py +++ b/examples/dialin-chatbot/bot_twilio.py @@ -84,7 +84,7 @@ async def main(room_url: str, token: str, callId: str, sipUri: str): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) await task.queue_frames([LLMMessagesFrame(messages)]) @transport.event_handler("on_participant_left") diff --git a/examples/foundational/06-listen-and-respond.py b/examples/foundational/06-listen-and-respond.py index f46ffc77a..5f9bb0a1f 100644 --- a/examples/foundational/06-listen-and-respond.py +++ b/examples/foundational/06-listen-and-respond.py @@ -110,7 +110,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. messages.append({"role": "system", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMMessagesFrame(messages)]) diff --git a/examples/foundational/06a-image-sync.py b/examples/foundational/06a-image-sync.py index 5cb625729..eda3c61df 100644 --- a/examples/foundational/06a-image-sync.py +++ b/examples/foundational/06a-image-sync.py @@ -127,7 +127,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): participant_name = participant.get("info", {}).get("userName", "") - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) await task.queue_frames([TextFrame(f"Hi there {participant_name}!")]) runner = PipelineRunner() diff --git a/examples/foundational/07-interruptible-vad.py b/examples/foundational/07-interruptible-vad.py index 8de28ff6d..59013bd3a 100644 --- a/examples/foundational/07-interruptible-vad.py +++ b/examples/foundational/07-interruptible-vad.py @@ -89,7 +89,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. messages.append({"role": "system", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMMessagesFrame(messages)]) diff --git a/examples/foundational/07-interruptible.py b/examples/foundational/07-interruptible.py index ed2891d22..3148986a8 100644 --- a/examples/foundational/07-interruptible.py +++ b/examples/foundational/07-interruptible.py @@ -87,7 +87,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. messages.append({"role": "system", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMMessagesFrame(messages)]) diff --git a/examples/foundational/07a-interruptible-anthropic.py b/examples/foundational/07a-interruptible-anthropic.py index 406f5fef1..e7e680eab 100644 --- a/examples/foundational/07a-interruptible-anthropic.py +++ b/examples/foundational/07a-interruptible-anthropic.py @@ -82,7 +82,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. await task.queue_frames([LLMMessagesFrame(messages)]) diff --git a/examples/foundational/07b-interruptible-langchain.py b/examples/foundational/07b-interruptible-langchain.py index ec6cdb0e2..e86c4a8e9 100644 --- a/examples/foundational/07b-interruptible-langchain.py +++ b/examples/foundational/07b-interruptible-langchain.py @@ -109,7 +109,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) lc.set_participant_id(participant["id"]) # Kick off the conversation. # the `LLMMessagesFrame` will be picked up by the LangchainProcessor using diff --git a/examples/foundational/07d-interruptible-elevenlabs.py b/examples/foundational/07d-interruptible-elevenlabs.py index 524b50fd6..24ccf6a24 100644 --- a/examples/foundational/07d-interruptible-elevenlabs.py +++ b/examples/foundational/07d-interruptible-elevenlabs.py @@ -85,7 +85,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. messages.append({"role": "system", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMMessagesFrame(messages)]) diff --git a/examples/foundational/07e-interruptible-playht.py b/examples/foundational/07e-interruptible-playht.py index f93bf6580..8a2af47bb 100644 --- a/examples/foundational/07e-interruptible-playht.py +++ b/examples/foundational/07e-interruptible-playht.py @@ -88,7 +88,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. messages.append({"role": "system", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMMessagesFrame(messages)]) diff --git a/examples/foundational/07f-interruptible-azure.py b/examples/foundational/07f-interruptible-azure.py index c605de431..fb3e8ea0c 100644 --- a/examples/foundational/07f-interruptible-azure.py +++ b/examples/foundational/07f-interruptible-azure.py @@ -89,7 +89,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. messages.append({"role": "system", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMMessagesFrame(messages)]) diff --git a/examples/foundational/07g-interruptible-openai-tts.py b/examples/foundational/07g-interruptible-openai-tts.py index 49ee08595..af414e3e3 100644 --- a/examples/foundational/07g-interruptible-openai-tts.py +++ b/examples/foundational/07g-interruptible-openai-tts.py @@ -74,7 +74,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. messages.append({"role": "system", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMMessagesFrame(messages)]) diff --git a/examples/foundational/07h-interruptible-openpipe.py b/examples/foundational/07h-interruptible-openpipe.py index 186ea2fc1..f01d535df 100644 --- a/examples/foundational/07h-interruptible-openpipe.py +++ b/examples/foundational/07h-interruptible-openpipe.py @@ -86,7 +86,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. messages.append({"role": "system", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMMessagesFrame(messages)]) diff --git a/examples/foundational/07i-interruptible-xtts.py b/examples/foundational/07i-interruptible-xtts.py index 46cc86562..bc14c04a7 100644 --- a/examples/foundational/07i-interruptible-xtts.py +++ b/examples/foundational/07i-interruptible-xtts.py @@ -81,7 +81,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. messages.append({"role": "system", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMMessagesFrame(messages)]) diff --git a/examples/foundational/07j-interruptible-gladia.py b/examples/foundational/07j-interruptible-gladia.py index 4f313fcb2..b6866d8af 100644 --- a/examples/foundational/07j-interruptible-gladia.py +++ b/examples/foundational/07j-interruptible-gladia.py @@ -83,7 +83,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. messages.append({"role": "system", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMMessagesFrame(messages)]) diff --git a/examples/foundational/07k-interruptible-lmnt.py b/examples/foundational/07k-interruptible-lmnt.py index f898c237a..49179e747 100644 --- a/examples/foundational/07k-interruptible-lmnt.py +++ b/examples/foundational/07k-interruptible-lmnt.py @@ -77,7 +77,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. messages.append({"role": "system", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMMessagesFrame(messages)]) diff --git a/examples/foundational/07l-interruptible-together.py b/examples/foundational/07l-interruptible-together.py index cb143dc88..96315623b 100644 --- a/examples/foundational/07l-interruptible-together.py +++ b/examples/foundational/07l-interruptible-together.py @@ -96,7 +96,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. await task.queue_frames([LLMMessagesFrame(messages)]) diff --git a/examples/foundational/07m-interruptible-aws.py b/examples/foundational/07m-interruptible-aws.py index 41d1f0530..175f8864f 100644 --- a/examples/foundational/07m-interruptible-aws.py +++ b/examples/foundational/07m-interruptible-aws.py @@ -84,7 +84,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. messages.append({"role": "system", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMMessagesFrame(messages)]) diff --git a/examples/foundational/07n-interruptible-google.py b/examples/foundational/07n-interruptible-google.py index e5f8f7063..b5524b1fa 100644 --- a/examples/foundational/07n-interruptible-google.py +++ b/examples/foundational/07n-interruptible-google.py @@ -82,7 +82,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. messages.append({"role": "system", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMMessagesFrame(messages)]) diff --git a/examples/foundational/07o-interruptible-assemblyai.py b/examples/foundational/07o-interruptible-assemblyai.py index 76ade04c2..0ae548f9c 100644 --- a/examples/foundational/07o-interruptible-assemblyai.py +++ b/examples/foundational/07o-interruptible-assemblyai.py @@ -83,7 +83,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. messages.append({"role": "system", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMMessagesFrame(messages)]) diff --git a/examples/foundational/09-mirror.py b/examples/foundational/09-mirror.py index 50c3a5b18..a719d54f6 100644 --- a/examples/foundational/09-mirror.py +++ b/examples/foundational/09-mirror.py @@ -74,7 +74,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_video(participant["id"]) + await transport.capture_participant_video(participant["id"]) pipeline = Pipeline([transport.input(), MirrorProcessor(), transport.output()]) diff --git a/examples/foundational/09a-local-mirror.py b/examples/foundational/09a-local-mirror.py index 3d5ce347c..539cca600 100644 --- a/examples/foundational/09a-local-mirror.py +++ b/examples/foundational/09a-local-mirror.py @@ -81,7 +81,7 @@ async def main(): @daily_transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_video(participant["id"]) + await transport.capture_participant_video(participant["id"]) pipeline = Pipeline([daily_transport.input(), MirrorProcessor(), tk_transport.output()]) diff --git a/examples/foundational/10-wake-phrase.py b/examples/foundational/10-wake-phrase.py index eeec2d4f3..982303aee 100644 --- a/examples/foundational/10-wake-phrase.py +++ b/examples/foundational/10-wake-phrase.py @@ -82,7 +82,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) await tts.say("Hi! If you want to talk to me, just say 'Hey Robot'.") runner = PipelineRunner() diff --git a/examples/foundational/11-sound-effects.py b/examples/foundational/11-sound-effects.py index a921fb9c9..c3912ff6d 100644 --- a/examples/foundational/11-sound-effects.py +++ b/examples/foundational/11-sound-effects.py @@ -134,7 +134,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) await tts.say("Hi, I'm listening!") await transport.send_audio(sounds["ding1.wav"]) diff --git a/examples/foundational/12-describe-video.py b/examples/foundational/12-describe-video.py index 3b17ba040..b5bb577aa 100644 --- a/examples/foundational/12-describe-video.py +++ b/examples/foundational/12-describe-video.py @@ -84,8 +84,8 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): await tts.say("Hi there! Feel free to ask me what I see.") - transport.capture_participant_video(participant["id"], framerate=0) - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_video(participant["id"], framerate=0) + await transport.capture_participant_transcription(participant["id"]) image_requester.set_participant_id(participant["id"]) pipeline = Pipeline( diff --git a/examples/foundational/12a-describe-video-gemini-flash.py b/examples/foundational/12a-describe-video-gemini-flash.py index 7f00daa68..bc76afc73 100644 --- a/examples/foundational/12a-describe-video-gemini-flash.py +++ b/examples/foundational/12a-describe-video-gemini-flash.py @@ -86,8 +86,8 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): await tts.say("Hi there! Feel free to ask me what I see.") - transport.capture_participant_video(participant["id"], framerate=0) - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_video(participant["id"], framerate=0) + await transport.capture_participant_transcription(participant["id"]) image_requester.set_participant_id(participant["id"]) pipeline = Pipeline( diff --git a/examples/foundational/12b-describe-video-gpt-4o.py b/examples/foundational/12b-describe-video-gpt-4o.py index 035d1dddd..d8474b568 100644 --- a/examples/foundational/12b-describe-video-gpt-4o.py +++ b/examples/foundational/12b-describe-video-gpt-4o.py @@ -83,8 +83,8 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): await tts.say("Hi there! Feel free to ask me what I see.") - transport.capture_participant_video(participant["id"], framerate=0) - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_video(participant["id"], framerate=0) + await transport.capture_participant_transcription(participant["id"]) image_requester.set_participant_id(participant["id"]) pipeline = Pipeline( diff --git a/examples/foundational/12c-describe-video-anthropic.py b/examples/foundational/12c-describe-video-anthropic.py index ede473b31..bc6f5a4ea 100644 --- a/examples/foundational/12c-describe-video-anthropic.py +++ b/examples/foundational/12c-describe-video-anthropic.py @@ -83,8 +83,8 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): await tts.say("Hi there! Feel free to ask me what I see.") - transport.capture_participant_video(participant["id"], framerate=0) - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_video(participant["id"], framerate=0) + await transport.capture_participant_transcription(participant["id"]) image_requester.set_participant_id(participant["id"]) pipeline = Pipeline( diff --git a/examples/foundational/14-function-calling.py b/examples/foundational/14-function-calling.py index aabc20e86..2479e3f3e 100644 --- a/examples/foundational/14-function-calling.py +++ b/examples/foundational/14-function-calling.py @@ -127,7 +127,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. await task.queue_frames([context_aggregator.user().get_context_frame()]) diff --git a/examples/foundational/14a-function-calling-anthropic.py b/examples/foundational/14a-function-calling-anthropic.py index 7afab6b84..9d3335809 100644 --- a/examples/foundational/14a-function-calling-anthropic.py +++ b/examples/foundational/14a-function-calling-anthropic.py @@ -105,7 +105,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. await task.queue_frames([context_aggregator.user().get_context_frame()]) diff --git a/examples/foundational/14b-function-calling-anthropic-video.py b/examples/foundational/14b-function-calling-anthropic-video.py index ff832f873..8c3821ce4 100644 --- a/examples/foundational/14b-function-calling-anthropic-video.py +++ b/examples/foundational/14b-function-calling-anthropic-video.py @@ -160,8 +160,8 @@ If you need to use a tool, simply use the tool. Do not tell the user the tool yo async def on_first_participant_joined(transport, participant): global video_participant_id video_participant_id = participant["id"] - transport.capture_participant_transcription(video_participant_id) - transport.capture_participant_video(video_participant_id, framerate=0) + await transport.capture_participant_transcription(video_participant_id) + await transport.capture_participant_video(video_participant_id, framerate=0) # Kick off the conversation. await task.queue_frames([context_aggregator.user().get_context_frame()]) diff --git a/examples/foundational/14c-function-calling-together.py b/examples/foundational/14c-function-calling-together.py index 391a96887..283b2fe70 100644 --- a/examples/foundational/14c-function-calling-together.py +++ b/examples/foundational/14c-function-calling-together.py @@ -123,7 +123,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. # await tts.say("Hi! Ask me about the weather in San Francisco.") diff --git a/examples/foundational/14d-function-calling-video.py b/examples/foundational/14d-function-calling-video.py index 5e149c815..e238d91eb 100644 --- a/examples/foundational/14d-function-calling-video.py +++ b/examples/foundational/14d-function-calling-video.py @@ -153,8 +153,8 @@ indicate you should use the get_image tool are: async def on_first_participant_joined(transport, participant): global video_participant_id video_participant_id = participant["id"] - transport.capture_participant_transcription(participant["id"]) - transport.capture_participant_video(video_participant_id, framerate=0) + await transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_video(video_participant_id, framerate=0) # Kick off the conversation. await tts.say("Hi! Ask me about the weather in San Francisco.") diff --git a/examples/foundational/14e-function-calling-gemini.py b/examples/foundational/14e-function-calling-gemini.py index c124e4c2e..087794a11 100644 --- a/examples/foundational/14e-function-calling-gemini.py +++ b/examples/foundational/14e-function-calling-gemini.py @@ -159,8 +159,8 @@ indicate you should use the get_image tool are: async def on_first_participant_joined(transport, participant): global video_participant_id video_participant_id = participant["id"] - transport.capture_participant_transcription(participant["id"]) - transport.capture_participant_video(video_participant_id, framerate=0) + await transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_video(video_participant_id, framerate=0) # Kick off the conversation. await task.queue_frames([context_aggregator.user().get_context_frame()]) diff --git a/examples/foundational/15-switch-voices.py b/examples/foundational/15-switch-voices.py index d874d0fba..5c61dd25f 100644 --- a/examples/foundational/15-switch-voices.py +++ b/examples/foundational/15-switch-voices.py @@ -141,7 +141,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. messages.append( { diff --git a/examples/foundational/15a-switch-languages.py b/examples/foundational/15a-switch-languages.py index c2af63ad8..a6a7edfdf 100644 --- a/examples/foundational/15a-switch-languages.py +++ b/examples/foundational/15a-switch-languages.py @@ -128,7 +128,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. messages.append( { diff --git a/examples/foundational/16-gpu-container-local-bot.py b/examples/foundational/16-gpu-container-local-bot.py index 9d154fd51..85a2ddb0b 100644 --- a/examples/foundational/16-gpu-container-local-bot.py +++ b/examples/foundational/16-gpu-container-local-bot.py @@ -92,7 +92,7 @@ async def main(): # bot can "hear" and respond to them. @transport.event_handler("on_participant_joined") async def on_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # When the first participant joins, the bot should introduce itself. @transport.event_handler("on_first_participant_joined") diff --git a/examples/foundational/17-detect-user-idle.py b/examples/foundational/17-detect-user-idle.py index 6acad8800..7268e2305 100644 --- a/examples/foundational/17-detect-user-idle.py +++ b/examples/foundational/17-detect-user-idle.py @@ -99,7 +99,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. messages.append({"role": "system", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMMessagesFrame(messages)]) diff --git a/examples/foundational/19-openai-realtime-beta.py b/examples/foundational/19-openai-realtime-beta.py index f258f95e4..49ede65be 100644 --- a/examples/foundational/19-openai-realtime-beta.py +++ b/examples/foundational/19-openai-realtime-beta.py @@ -166,7 +166,7 @@ Remember, your responses should be short. Just one or two sentences, usually.""" @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. await task.queue_frames([context_aggregator.user().get_context_frame()]) diff --git a/examples/foundational/20a-persistent-context-openai.py b/examples/foundational/20a-persistent-context-openai.py index 97ef88685..d4d418e91 100644 --- a/examples/foundational/20a-persistent-context-openai.py +++ b/examples/foundational/20a-persistent-context-openai.py @@ -223,7 +223,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. await task.queue_frames([context_aggregator.user().get_context_frame()]) diff --git a/examples/foundational/20b-persistent-context-openai-realtime.py b/examples/foundational/20b-persistent-context-openai-realtime.py index 2dc7bf19e..c22eaf128 100644 --- a/examples/foundational/20b-persistent-context-openai-realtime.py +++ b/examples/foundational/20b-persistent-context-openai-realtime.py @@ -249,7 +249,7 @@ Remember, your responses should be short. Just one or two sentences, usually.""" @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. await task.queue_frames([context_aggregator.user().get_context_frame()]) diff --git a/examples/foundational/20c-persistent-context-anthropic.py b/examples/foundational/20c-persistent-context-anthropic.py index 02bd4febb..8e8e9c3f7 100644 --- a/examples/foundational/20c-persistent-context-anthropic.py +++ b/examples/foundational/20c-persistent-context-anthropic.py @@ -219,7 +219,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. await task.queue_frames([context_aggregator.user().get_context_frame()]) diff --git a/examples/foundational/20d-persistent-context-gemini.py b/examples/foundational/20d-persistent-context-gemini.py index 96abd8a41..25d809021 100644 --- a/examples/foundational/20d-persistent-context-gemini.py +++ b/examples/foundational/20d-persistent-context-gemini.py @@ -111,8 +111,8 @@ async def load_conversation(function_name, tool_call_id, args, llm, context, res messages = [ { "role": "system", - "content": """You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your -capabilities in a succinct way. Your output will be converted to audio so don't include special + "content": """You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your +capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way. You have several tools you can use to help you. @@ -276,8 +276,8 @@ async def main(): async def on_first_participant_joined(transport, participant): global video_participant_id video_participant_id = participant["id"] - transport.capture_participant_transcription(participant["id"]) - transport.capture_participant_video(video_participant_id, framerate=0) + await transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_video(video_participant_id, framerate=0) # Kick off the conversation. await task.queue_frames([context_aggregator.user().get_context_frame()]) diff --git a/examples/moondream-chatbot/bot.py b/examples/moondream-chatbot/bot.py index f481473fe..b095acf6b 100644 --- a/examples/moondream-chatbot/bot.py +++ b/examples/moondream-chatbot/bot.py @@ -203,8 +203,8 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) - transport.capture_participant_video(participant["id"], framerate=0) + await transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_video(participant["id"], framerate=0) ir.set_participant_id(participant["id"]) await task.queue_frames([LLMMessagesFrame(messages)]) diff --git a/examples/patient-intake/bot.py b/examples/patient-intake/bot.py index efdd1caa9..adb3785e4 100644 --- a/examples/patient-intake/bot.py +++ b/examples/patient-intake/bot.py @@ -352,7 +352,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) print(f"Context is: {context}") await task.queue_frames([OpenAILLMContextFrame(context)]) diff --git a/examples/storytelling-chatbot/src/bot.py b/examples/storytelling-chatbot/src/bot.py index 82c55e013..6b5b3b974 100644 --- a/examples/storytelling-chatbot/src/bot.py +++ b/examples/storytelling-chatbot/src/bot.py @@ -102,7 +102,7 @@ async def main(room_url, token=None): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): logger.debug("Participant joined, storytime commence!") - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) await intro_task.queue_frames( [ images["book1"], diff --git a/examples/studypal/studypal.py b/examples/studypal/studypal.py index fb4c7fead..1ec165069 100644 --- a/examples/studypal/studypal.py +++ b/examples/studypal/studypal.py @@ -165,7 +165,7 @@ Your task is to help the user understand and learn from this article in 2 senten @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) messages.append( { "role": "system", diff --git a/examples/translation-chatbot/bot.py b/examples/translation-chatbot/bot.py index 55302b392..e654c0159 100644 --- a/examples/translation-chatbot/bot.py +++ b/examples/translation-chatbot/bot.py @@ -121,7 +121,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) + await transport.capture_participant_transcription(participant["id"]) runner = PipelineRunner() diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index ba0da8754..88f8b0bfa 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -454,27 +454,37 @@ class DailyTransportClient(EventHandler): def participant_counts(self): return self._client.participant_counts() - def start_dialout(self, settings): - self._client.start_dialout(settings) + async def start_dialout(self, settings): + future = self._loop.create_future() + self._client.start_dialout(settings, completion=completion_callback(future)) + await future - def stop_dialout(self, participant_id): - self._client.stop_dialout(participant_id) + async def stop_dialout(self, participant_id): + future = self._loop.create_future() + self._client.stop_dialout(participant_id, completion=completion_callback(future)) + await future - def start_recording(self, streaming_settings, stream_id, force_new): - self._client.start_recording(streaming_settings, stream_id, force_new) + async def start_recording(self, streaming_settings, stream_id, force_new): + future = self._loop.create_future() + self._client.start_recording( + streaming_settings, stream_id, force_new, completion=completion_callback(future) + ) + await future - def stop_recording(self, stream_id): - self._client.stop_recording(stream_id) + async def stop_recording(self, stream_id): + future = self._loop.create_future() + self._client.stop_recording(stream_id, completion=completion_callback(future)) + await future - def capture_participant_transcription(self, participant_id: str): + async def capture_participant_transcription(self, participant_id: str): if not self._params.transcription_enabled: return self._transcription_ids.append(participant_id) if self._joined and self._transcription_status: - self.update_transcription(self._transcription_ids) + await self.update_transcription(self._transcription_ids) - def capture_participant_video( + async def capture_participant_video( self, participant_id: str, callback: Callable, @@ -483,7 +493,7 @@ class DailyTransportClient(EventHandler): color_format: str = "RGB", ): # Only enable camera subscription on this participant - self._client.update_subscriptions( + await self.update_subscriptions( participant_settings={participant_id: {"media": "subscribed"}} ) @@ -496,8 +506,12 @@ class DailyTransportClient(EventHandler): color_format=color_format, ) - def update_transcription(self, participants=None, instance_id=None): - self._client.update_transcription(participants, instance_id) + async def update_transcription(self, participants=None, instance_id=None): + future = self._loop.create_future() + self._client.update_transcription( + participants, instance_id, completion=completion_callback(future) + ) + await future async def update_subscriptions(self, participant_settings=None, profile_settings=None): future = self._loop.create_future() @@ -559,7 +573,7 @@ class DailyTransportClient(EventHandler): def on_transcription_started(self, status): logger.debug(f"Transcription started: {status}") self._transcription_status = status - self.update_transcription(self._transcription_ids) + self._call_async_callback(self.update_transcription, self._transcription_ids) def on_transcription_stopped(self, stopped_by, stopped_by_error): logger.debug("Transcription stopped") @@ -662,7 +676,7 @@ class DailyInputTransport(BaseInputTransport): await super().process_frame(frame, direction) if isinstance(frame, UserImageRequestFrame): - self.request_participant_image(frame.user_id) + await self.request_participant_image(frame.user_id) # # Frames @@ -692,7 +706,7 @@ class DailyInputTransport(BaseInputTransport): # Camera in # - def capture_participant_video( + async def capture_participant_video( self, participant_id: str, framerate: int = 30, @@ -705,11 +719,11 @@ class DailyInputTransport(BaseInputTransport): "render_next_frame": False, } - self._client.capture_participant_video( + await self._client.capture_participant_video( participant_id, self._on_participant_video_frame, framerate, video_source, color_format ) - def request_participant_image(self, participant_id: str): + async def request_participant_image(self, participant_id: str): if participant_id in self._video_renderers: self._video_renderers[participant_id]["render_next_frame"] = True @@ -866,22 +880,22 @@ class DailyTransport(BaseTransport): def participant_counts(self): return self._client.participant_counts() - def start_dialout(self, settings=None): - self._client.start_dialout(settings) + async def start_dialout(self, settings=None): + await self._client.start_dialout(settings) - def stop_dialout(self, participant_id): - self._client.stop_dialout(participant_id) + async def stop_dialout(self, participant_id): + await self._client.stop_dialout(participant_id) - def start_recording(self, streaming_settings=None, stream_id=None, force_new=None): - self._client.start_recording(streaming_settings, stream_id, force_new) + async def start_recording(self, streaming_settings=None, stream_id=None, force_new=None): + await self._client.start_recording(streaming_settings, stream_id, force_new) - def stop_recording(self, stream_id=None): - self._client.stop_recording(stream_id) + async def stop_recording(self, stream_id=None): + await self._client.stop_recording(stream_id) - def capture_participant_transcription(self, participant_id: str): - self._client.capture_participant_transcription(participant_id) + async def capture_participant_transcription(self, participant_id: str): + await self._client.capture_participant_transcription(participant_id) - def capture_participant_video( + async def capture_participant_video( self, participant_id: str, framerate: int = 30, @@ -889,7 +903,7 @@ class DailyTransport(BaseTransport): color_format: str = "RGB", ): if self._input: - self._input.capture_participant_video( + await self._input.capture_participant_video( participant_id, framerate, video_source, color_format )