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

cleanup examples and remove requests library

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2024-07-31 23:39:51 -07:00
parent 3db7f6a284
commit 27a09c0b2c
67 changed files with 1114 additions and 1632 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
CHANGELOG.md
View File

@@ -35,6 +35,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Fixed a `BaseInputTransport` issue that was causing start/stop interruptions
incoming frames to not cancel tasks and be processed properly.
### Other
- Remove `requests` library usage.
- Cleanup examples and use `DailyRESTHelper`.
## [0.0.39] - 2024-07-23
### Fixed

108
examples/deployment/flyio-example/bot_runner.py
View File

@@ -1,14 +1,21 @@
import os
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import aiohttp
import argparse
import subprocess
import requests
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomObject, DailyRoomProperties, DailyRoomParams
import os
from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from pipecat.transports.services.helpers.daily_rest import (
DailyRESTHelper, DailyRoomObject, DailyRoomProperties, DailyRoomParams)
from dotenv import load_dotenv
load_dotenv(override=True)
@@ -52,53 +59,52 @@ app.add_middleware(
# ----------------- Main ----------------- #
def spawn_fly_machine(room_url: str, token: str):
# Use the same image as the bot runner
res = requests.get(f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines", headers=FLY_HEADERS)
if res.status_code != 200:
raise Exception(f"Unable to get machine info from Fly: {res.text}")
image = res.json()[0]['config']['image']
async def spawn_fly_machine(room_url: str, token: str):
async with aiohttp.ClientSession() as session:
# Use the same image as the bot runner
async with session.get(f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines", headers=FLY_HEADERS) as r:
if r.status != 200:
text = await r.text()
raise Exception(f"Unable to get machine info from Fly: {text}")
# Machine configuration
cmd = f"python3 bot.py -u {room_url} -t {token}"
cmd = cmd.split()
worker_props = {
"config": {
"image": image,
"auto_destroy": True,
"init": {
"cmd": cmd
data = await r.json()
image = data[0]['config']['image']
# Machine configuration
cmd = f"python3 bot.py -u {room_url} -t {token}"
cmd = cmd.split()
worker_props = {
"config": {
"image": image,
"auto_destroy": True,
"init": {
"cmd": cmd
},
"restart": {
"policy": "no"
},
"guest": {
"cpu_kind": "shared",
"cpus": 1,
"memory_mb": 1024
}
},
"restart": {
"policy": "no"
},
"guest": {
"cpu_kind": "shared",
"cpus": 1,
"memory_mb": 1024
}
},
}
}
# Spawn a new machine instance
async with session.post(f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines", headers=FLY_HEADERS, json=worker_props) as r:
if r.status != 200:
text = await r.text()
raise Exception(f"Problem starting a bot worker: {text}")
# Spawn a new machine instance
res = requests.post(
f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines",
headers=FLY_HEADERS,
json=worker_props)
data = await r.json()
# Wait for the machine to enter the started state
vm_id = data['id']
if res.status_code != 200:
raise Exception(f"Problem starting a bot worker: {res.text}")
# Wait for the machine to enter the started state
vm_id = res.json()['id']
res = requests.get(
f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines/{vm_id}/wait?state=started",
headers=FLY_HEADERS)
if res.status_code != 200:
raise Exception(f"Bot was unable to enter started state: {res.text}")
async with session.get(f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines/{vm_id}/wait?state=started", headers=FLY_HEADERS) as r:
if r.status != 200:
text = await r.text()
raise Exception(f"Bot was unable to enter started state: {text}")
print(f"Machine joined room: {room_url}")
@@ -121,7 +127,7 @@ async def start_bot(request: Request) -> JSONResponse:
properties=DailyRoomProperties()
)
try:
room: DailyRoomObject = daily_rest_helper.create_room(params=params)
room: DailyRoomObject = await daily_rest_helper.create_room(params=params)
except Exception as e:
raise HTTPException(
status_code=500,
@@ -129,13 +135,13 @@ async def start_bot(request: Request) -> JSONResponse:
else:
# Check passed room URL exists, we should assume that it already has a sip set up
try:
room: DailyRoomObject = daily_rest_helper.get_room_from_url(room_url)
room: DailyRoomObject = await daily_rest_helper.get_room_from_url(room_url)
except Exception:
raise HTTPException(
status_code=500, detail=f"Room not found: {room_url}")
# Give the agent a token to join the session
token = daily_rest_helper.get_token(room.url, MAX_SESSION_TIME)
token = await daily_rest_helper.get_token(room.url, MAX_SESSION_TIME)
if not room or not token:
raise HTTPException(
@@ -156,13 +162,13 @@ async def start_bot(request: Request) -> JSONResponse:
status_code=500, detail=f"Failed to start subprocess: {e}")
else:
try:
spawn_fly_machine(room.url, token)
await spawn_fly_machine(room.url, token)
except Exception as e:
raise HTTPException(
status_code=500, detail=f"Failed to spawn VM: {e}")
# Grab a token for the user to join with
user_token = daily_rest_helper.get_token(room.url, MAX_SESSION_TIME)
user_token = await daily_rest_helper.get_token(room.url, MAX_SESSION_TIME)
return JSONResponse({
"room_url": room.url,

3
examples/deployment/flyio-example/requirements.txt
View File

@@ -1,6 +1,5 @@
pipecat-ai[daily,openai,silero]
fastapi
uvicorn
requests
python-dotenv
loguru
loguru

33
examples/dialin-chatbot/bot_runner.py
View File

@@ -6,15 +6,23 @@ provisioning a room and starting a Pipecat bot in response.
Refer to README for more information.
"""
import os
import argparse
import subprocess
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomObject, DailyRoomProperties, DailyRoomSipParams, DailyRoomParams
from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, PlainTextResponse
from twilio.twiml.voice_response import VoiceResponse
from pipecat.transports.services.helpers.daily_rest import (
DailyRESTHelper,
DailyRoomObject,
DailyRoomProperties,
DailyRoomSipParams,
DailyRoomParams)
from dotenv import load_dotenv
load_dotenv(override=True)
@@ -53,7 +61,7 @@ action using the Twilio Client library.
"""
def _create_daily_room(room_url, callId, callDomain=None, vendor="daily"):
async def _create_daily_room(room_url, callId, callDomain=None, vendor="daily"):
if not room_url:
params = DailyRoomParams(
properties=DailyRoomProperties(
@@ -68,14 +76,13 @@ def _create_daily_room(room_url, callId, callDomain=None, vendor="daily"):
)
print(f"Creating new room...")
room: DailyRoomObject = daily_rest_helper.create_room(params=params)
room: DailyRoomObject = await daily_rest_helper.create_room(params=params)
else:
# Check passed room URL exist (we assume that it already has a sip set up!)
try:
print(f"Joining existing room: {room_url}")
room: DailyRoomObject = daily_rest_helper.get_room_from_url(
room_url)
room: DailyRoomObject = await daily_rest_helper.get_room_from_url(room_url)
except Exception:
raise HTTPException(
status_code=500, detail=f"Room not found: {room_url}")
@@ -83,7 +90,7 @@ def _create_daily_room(room_url, callId, callDomain=None, vendor="daily"):
print(f"Daily room: {room.url} {room.config.sip_endpoint}")
# Give the agent a token to join the session
token = daily_rest_helper.get_token(room.url, MAX_SESSION_TIME)
token = await daily_rest_helper.get_token(room.url, MAX_SESSION_TIME)
if not room or not token:
raise HTTPException(
@@ -92,11 +99,11 @@ def _create_daily_room(room_url, callId, callDomain=None, vendor="daily"):
# Spawn a new agent, and join the user session
# Note: this is mostly for demonstration purposes (refer to 'deployment' in docs)
if vendor == "daily":
bot_proc = f"python3 -m bot_daily -u {room.url} -t {token} -i {
callId} -d {callDomain}"
bot_proc = f"python3 - m bot_daily - u {room.url} - t {token} - i {
callId} - d {callDomain}"
else:
bot_proc = f"python3 -m bot_twilio -u {room.url} -t {
token} -i {callId} -s {room.config.sip_endpoint}"
bot_proc = f"python3 - m bot_twilio - u {room.url} - t {
token} - i {callId} - s {room.config.sip_endpoint}"
try:
subprocess.Popen(
@@ -140,8 +147,7 @@ async def twilio_start_bot(request: Request):
# create room and tell the bot to join the created room
# note: Twilio does not require a callDomain
room: DailyRoomObject = _create_daily_room(
room_url, callId, None, "twilio")
room: DailyRoomObject = await _create_daily_room(room_url, callId, None, "twilio")
print(f"Put Twilio on hold...")
# We have the room and the SIP URI,
@@ -178,8 +184,7 @@ async def daily_start_bot(request: Request) -> JSONResponse:
detail="Missing properties 'callId' or 'callDomain'")
print(f"CallId: {callId}, CallDomain: {callDomain}")
room: DailyRoomObject = _create_daily_room(
room_url, callId, callDomain, "daily")
room: DailyRoomObject = await _create_daily_room(room_url, callId, callDomain, "daily")
# Grab a token for the user to join with
return JSONResponse({

4
examples/dialin-chatbot/requirements.txt
View File

@@ -1,7 +1,5 @@
pipecat-ai[daily,openai,silero]
fastapi
uvicorn
requests
python-dotenv
loguru
twilio
twilio

6
examples/foundational/01-say-one-thing.py
View File

@@ -27,7 +27,8 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
async def main(room_url):
async def main():
(room_url, _) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url, None, "Say One Thing", DailyParams(audio_out_enabled=True))
@@ -52,5 +53,4 @@ async def main(room_url):
await runner.run(task)
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url))
asyncio.run(main())

6
examples/foundational/02-llm-say-one-thing.py
View File

@@ -28,7 +28,8 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
async def main(room_url):
async def main():
(room_url, _) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -64,5 +65,4 @@ async def main(room_url):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url))
asyncio.run(main())

6
examples/foundational/03-still-frame.py
View File

@@ -27,7 +27,8 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
async def main(room_url):
async def main():
(room_url, _) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -64,5 +65,4 @@ async def main(room_url):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url))
asyncio.run(main())

6
examples/foundational/04-utterance-and-speech.py
View File

@@ -30,7 +30,8 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
async def main(room_url: str):
async def main():
(room_url, _) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(room_url, None, "Static And Dynamic Speech")
@@ -82,5 +83,4 @@ async def main(room_url: str):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url))
asyncio.run(main())

6
examples/foundational/05-sync-speech-and-image.py
View File

@@ -73,7 +73,8 @@ class MonthPrepender(FrameProcessor):
await self.push_frame(frame, direction)
async def main(room_url):
async def main():
(room_url, _) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -162,5 +163,4 @@ async def main(room_url):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url))
asyncio.run(main())

6
examples/foundational/06-listen-and-respond.py
View File

@@ -34,7 +34,8 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -99,5 +100,4 @@ async def main(room_url: str, token):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

6
examples/foundational/06a-image-sync.py
View File

@@ -59,7 +59,8 @@ class ImageSyncAggregator(FrameProcessor):
await self.push_frame(frame)
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -125,5 +126,4 @@ async def main(room_url: str, token):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

7
examples/foundational/07-interruptible.py
View File

@@ -31,7 +31,9 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -94,5 +96,4 @@ async def main(room_url: str, token):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

6
examples/foundational/07a-interruptible-anthropic.py
View File

@@ -31,7 +31,8 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -91,5 +92,4 @@ async def main(room_url: str, token):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

6
examples/foundational/07b-interruptible-langchain.py
View File

@@ -47,7 +47,8 @@ def get_session_history(session_id: str) -> BaseChatMessageHistory:
return message_store[session_id]
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -121,5 +122,4 @@ async def main(room_url: str, token):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

6
examples/foundational/07c-interruptible-deepgram.py
View File

@@ -31,7 +31,8 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -93,5 +94,4 @@ async def main(room_url: str, token):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

6
examples/foundational/07d-interruptible-cartesia.py
View File

@@ -31,7 +31,8 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
transport = DailyTransport(
room_url,
token,
@@ -90,5 +91,4 @@ async def main(room_url: str, token):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

6
examples/foundational/07e-interruptible-playht.py
View File

@@ -30,7 +30,8 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
transport = DailyTransport(
room_url,
token,
@@ -89,5 +90,4 @@ async def main(room_url: str, token):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

6
examples/foundational/07f-interruptible-azure.py
View File

@@ -30,7 +30,8 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
transport = DailyTransport(
room_url,
token,
@@ -96,5 +97,4 @@ async def main(room_url: str, token):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

7
examples/foundational/07g-interruptible-openai-tts.py
View File

@@ -30,7 +30,9 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
transport = DailyTransport(
room_url,
token,
@@ -88,5 +90,4 @@ async def main(room_url: str, token):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

7
examples/foundational/07h-interruptible-openpipe.py
View File

@@ -34,7 +34,9 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -98,5 +100,4 @@ async def main(room_url: str, token):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

7
examples/foundational/07i-interruptible-xtts.py
View File

@@ -32,7 +32,9 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -92,5 +94,4 @@ async def main(room_url: str, token):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

7
examples/foundational/07j-interruptible-gladia.py
View File

@@ -33,7 +33,9 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -97,5 +99,4 @@ async def main(room_url: str, token):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

7
examples/foundational/08-bots-arguing.py
View File

@@ -22,7 +22,9 @@ logger = logging.getLogger("pipecat")
logger.setLevel(logging.DEBUG)
async def main(room_url: str):
async def main():
(room_url, _) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -144,5 +146,4 @@ async def main(room_url: str):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url))
asyncio.run(main())

7
examples/foundational/09-mirror.py
View File

@@ -23,7 +23,9 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
async def main(room_url, token):
async def main():
(room_url, token) = await configure()
transport = DailyTransport(
room_url, token, "Test",
DailyParams(
@@ -50,5 +52,4 @@ async def main(room_url, token):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

7
examples/foundational/09a-local-mirror.py
View File

@@ -27,7 +27,9 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
async def main(room_url, token):
async def main():
(room_url, token) = await configure()
tk_root = tk.Tk()
tk_root.title("Local Mirror")
@@ -62,5 +64,4 @@ async def main(room_url, token):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

6
examples/foundational/10-wake-phrase.py
View File

@@ -31,7 +31,8 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
@@ -90,5 +91,4 @@ async def main(room_url: str, token):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

7
examples/foundational/11-sound-effects.py
View File

@@ -83,7 +83,9 @@ class InboundSoundEffectWrapper(FrameProcessor):
await self.push_frame(frame, direction)
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -148,5 +150,4 @@ async def main(room_url: str, token):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

7
examples/foundational/12-describe-video.py
View File

@@ -49,7 +49,9 @@ class UserImageRequester(FrameProcessor):
await self.push_frame(frame, direction)
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -108,5 +110,4 @@ async def main(room_url: str, token):
await runner.run(task)
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

7
examples/foundational/12a-describe-video-gemini-flash.py
View File

@@ -49,7 +49,9 @@ class UserImageRequester(FrameProcessor):
await self.push_frame(frame, direction)
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -104,5 +106,4 @@ async def main(room_url: str, token):
await runner.run(task)
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

7
examples/foundational/12b-describe-video-gpt-4o.py
View File

@@ -49,7 +49,9 @@ class UserImageRequester(FrameProcessor):
await self.push_frame(frame, direction)
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -104,5 +106,4 @@ async def main(room_url: str, token):
await runner.run(task)
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

7
examples/foundational/12c-describe-video-anthropic.py
View File

@@ -49,7 +49,9 @@ class UserImageRequester(FrameProcessor):
await self.push_frame(frame, direction)
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -104,5 +106,4 @@ async def main(room_url: str, token):
await runner.run(task)
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

7
examples/foundational/13-whisper-transcription.py
View File

@@ -35,7 +35,9 @@ class TranscriptionLogger(FrameProcessor):
print(f"Transcription: {frame.text}")
async def main(room_url: str):
async def main():
(room_url, _) = await configure()
transport = DailyTransport(room_url, None, "Transcription bot",
DailyParams(audio_in_enabled=True))
@@ -53,5 +55,4 @@ async def main(room_url: str):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url))
asyncio.run(main())

9
examples/foundational/13b-deepgram-transcription.py
View File

@@ -36,11 +36,13 @@ class TranscriptionLogger(FrameProcessor):
print(f"Transcription: {frame.text}")
async def main(room_url: str):
async def main():
(room_url, _) = await configure()
transport = DailyTransport(room_url, None, "Transcription bot",
DailyParams(audio_in_enabled=True))
stt = DeepgramSTTService(os.getenv("DEEPGRAM_API_KEY"))
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tl = TranscriptionLogger()
@@ -54,5 +56,4 @@ async def main(room_url: str):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url))
asyncio.run(main())

10
examples/foundational/14-function-calling.py
View File

@@ -44,7 +44,9 @@ async def fetch_weather_from_api(llm, args):
return {"conditions": "nice", "temperature": "75"}
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -124,7 +126,7 @@ async def main(room_url: str, token):
task = PipelineTask(pipeline)
@ transport.event_handler("on_first_participant_joined")
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
transport.capture_participant_transcription(participant["id"])
# Kick off the conversation.
@@ -134,7 +136,5 @@ async def main(room_url: str, token):
await runner.run(task)
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

161
examples/foundational/15-switch-voices.py
View File

@@ -5,7 +5,6 @@
#
import asyncio
import aiohttp
import os
import sys
@@ -58,98 +57,98 @@ async def barbershop_man_filter(frame) -> bool:
return current_voice == "Barbershop Man"
async def main(room_url: str, token):
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
token,
"Pipecat",
DailyParams(
audio_out_enabled=True,
transcription_enabled=True,
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer()
)
async def main():
(room_url, token) = await configure()
transport = DailyTransport(
room_url,
token,
"Pipecat",
DailyParams(
audio_out_enabled=True,
transcription_enabled=True,
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer()
)
)
news_lady = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="bf991597-6c13-47e4-8411-91ec2de5c466", # Newslady
)
news_lady = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="bf991597-6c13-47e4-8411-91ec2de5c466", # Newslady
)
british_lady = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady
)
british_lady = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady
)
barbershop_man = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="a0e99841-438c-4a64-b679-ae501e7d6091", # Barbershop Man
)
barbershop_man = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="a0e99841-438c-4a64-b679-ae501e7d6091", # Barbershop Man
)
llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"),
model="gpt-4o")
llm.register_function("switch_voice", switch_voice)
llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"),
model="gpt-4o")
llm.register_function("switch_voice", switch_voice)
tools = [
ChatCompletionToolParam(
type="function",
function={
"name": "switch_voice",
"description": "Switch your voice only when the user asks you to",
"parameters": {
"type": "object",
"properties": {
"voice": {
"type": "string",
"description": "The voice the user wants you to use",
},
tools = [
ChatCompletionToolParam(
type="function",
function={
"name": "switch_voice",
"description": "Switch your voice only when the user asks you to",
"parameters": {
"type": "object",
"properties": {
"voice": {
"type": "string",
"description": "The voice the user wants you to use",
},
"required": ["voice"],
},
})]
messages = [
"required": ["voice"],
},
})]
messages = [
{
"role": "system",
"content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities. Respond to what the user said in a creative and helpful way. Your output should not include non-alphanumeric characters. You can do the following voices: 'News Lady', 'British Lady' and 'Barbershop Man'.",
},
]
context = OpenAILLMContext(messages, tools)
tma_in = LLMUserContextAggregator(context)
tma_out = LLMAssistantContextAggregator(context)
pipeline = Pipeline([
transport.input(), # Transport user input
tma_in, # User responses
llm, # LLM
ParallelPipeline( # TTS (one of the following vocies)
[FunctionFilter(news_lady_filter), news_lady], # News Lady voice
[FunctionFilter(british_lady_filter), british_lady], # British Lady voice
[FunctionFilter(barbershop_man_filter), barbershop_man], # Barbershop Man voice
),
transport.output(), # Transport bot output
tma_out # Assistant spoken responses
])
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
transport.capture_participant_transcription(participant["id"])
# Kick off the conversation.
messages.append(
{
"role": "system",
"content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities. Respond to what the user said in a creative and helpful way. Your output should not include non-alphanumeric characters. You can do the following voices: 'News Lady', 'British Lady' and 'Barbershop Man'.",
},
]
"content": f"Please introduce yourself to the user and let them know the voices you can do. Your initial responses should be as if you were a {current_voice}."})
await task.queue_frames([LLMMessagesFrame(messages)])
context = OpenAILLMContext(messages, tools)
tma_in = LLMUserContextAggregator(context)
tma_out = LLMAssistantContextAggregator(context)
runner = PipelineRunner()
pipeline = Pipeline([
transport.input(), # Transport user input
tma_in, # User responses
llm, # LLM
ParallelPipeline( # TTS (one of the following vocies)
[FunctionFilter(news_lady_filter), news_lady], # News Lady voice
[FunctionFilter(british_lady_filter), british_lady], # British Lady voice
[FunctionFilter(barbershop_man_filter), barbershop_man], # Barbershop Man voice
),
transport.output(), # Transport bot output
tma_out # Assistant spoken responses
])
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
transport.capture_participant_transcription(participant["id"])
# Kick off the conversation.
messages.append(
{
"role": "system",
"content": f"Please introduce yourself to the user and let them know the voices you can do. Your initial responses should be as if you were a {current_voice}."})
await task.queue_frames([LLMMessagesFrame(messages)])
runner = PipelineRunner()
await runner.run(task)
await runner.run(task)
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

7
examples/foundational/15a-switch-languages.py
View File

@@ -55,7 +55,9 @@ async def spanish_filter(frame) -> bool:
return current_language == "Spanish"
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -149,5 +151,4 @@ async def main(room_url: str, token):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

7
examples/foundational/16-gpu-container-local-bot.py
View File

@@ -32,7 +32,9 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -126,5 +128,4 @@ async def main(room_url: str, token):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

7
examples/foundational/17-detect-user-idle.py
View File

@@ -33,7 +33,9 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -104,5 +106,4 @@ async def main(room_url: str, token):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

35
examples/foundational/runner.py
View File

@@ -1,11 +1,16 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import argparse
import os
import time
import urllib
import requests
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
def configure():
async def configure():
parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
parser.add_argument(
"-u",
@@ -33,26 +38,12 @@ def configure():
if not key:
raise Exception("No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.")
daily_rest_helper = DailyRESTHelper(key, os.getenv("DAILY_API_URL", "https://api.daily.co/v1"))
# Create a meeting token for the given room with an expiration 1 hour in
# the future.
room_name: str = urllib.parse.urlparse(url).path[1:]
expiration: float = time.time() + 60 * 60
expiry_time: float = 60 * 60
res: requests.Response = requests.post(
f"https://api.daily.co/v1/meeting-tokens",
headers={
"Authorization": f"Bearer {key}"},
json={
"properties": {
"room_name": room_name,
"is_owner": True,
"exp": expiration}},
)
if res.status_code != 200:
raise Exception(
f"Failed to create meeting token: {res.status_code} {res.text}")
token: str = res.json()["token"]
token = await daily_rest_helper.get_token(url, expiry_time)
return (url, token)

12
examples/moondream-chatbot/bot.py
View File

@@ -1,3 +1,9 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import aiohttp
import os
@@ -128,7 +134,8 @@ class ImageFilterProcessor(FrameProcessor):
await self.push_frame(frame)
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -204,5 +211,4 @@ async def main(room_url: str, token):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

1
examples/moondream-chatbot/requirements.txt
View File

@@ -1,5 +1,4 @@
python-dotenv
requests
fastapi[all]
uvicorn
pipecat-ai[daily,moondream,openai,silero]

35
examples/moondream-chatbot/runner.py
View File

@@ -1,11 +1,16 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import argparse
import os
import time
import urllib
import requests
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
def configure():
async def configure():
parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
parser.add_argument(
"-u",
@@ -33,26 +38,12 @@ def configure():
if not key:
raise Exception("No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.")
daily_rest_helper = DailyRESTHelper(key, os.getenv("DAILY_API_URL", "https://api.daily.co/v1"))
# Create a meeting token for the given room with an expiration 1 hour in
# the future.
room_name: str = urllib.parse.urlparse(url).path[1:]
expiration: float = time.time() + 60 * 60
expiry_time: float = 60 * 60
res: requests.Response = requests.post(
f"https://api.daily.co/v1/meeting-tokens",
headers={
"Authorization": f"Bearer {key}"},
json={
"properties": {
"room_name": room_name,
"is_owner": True,
"exp": expiration}},
)
if res.status_code != 200:
raise Exception(
f"Failed to create meeting token: {res.status_code} {res.text}")
token: str = res.json()["token"]
token = await daily_rest_helper.get_token(url, expiry_time)
return (url, token)

35
examples/moondream-chatbot/server.py
View File

@@ -1,3 +1,9 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import os
import argparse
import subprocess
@@ -7,17 +13,22 @@ from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, RedirectResponse
from utils.daily_helpers import create_room as _create_room, get_token
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomParams
MAX_BOTS_PER_ROOM = 1
# Bot sub-process dict for status reporting and concurrency control
bot_procs = {}
daily_rest_helper = DailyRESTHelper(
os.getenv("DAILY_API_KEY", ""),
os.getenv("DAILY_API_URL", 'https://api.daily.co/v1'))
def cleanup():
# Clean up function, just to be extra safe
for proc in bot_procs.values():
for entry in bot_procs.values():
proc = entry[0]
proc.terminate()
proc.wait()
@@ -39,45 +50,45 @@ app.add_middleware(
@app.get("/start")
async def start_agent(request: Request):
print(f"!!! Creating room")
room_url, room_name = _create_room()
print(f"!!! Room URL: {room_url}")
room = await daily_rest_helper.create_room(DailyRoomParams())
print(f"!!! Room URL: {room.url}")
# Ensure the room property is present
if not room_url:
if not room.url:
raise HTTPException(
status_code=500,
detail="Missing 'room' property in request data. Cannot start agent without a target room!")
# Check if there is already an existing process running in this room
num_bots_in_room = sum(
1 for proc in bot_procs.values() if proc[1] == room_url and proc[0].poll() is None)
1 for proc in bot_procs.values() if proc[1] == room.url and proc[0].poll() is None)
if num_bots_in_room >= MAX_BOTS_PER_ROOM:
raise HTTPException(
status_code=500, detail=f"Max bot limited reach for room: {room_url}")
status_code=500, detail=f"Max bot limited reach for room: {room.url}")
# Get the token for the room
token = get_token(room_url)
token = await daily_rest_helper.get_token(room.url)
if not token:
raise HTTPException(
status_code=500, detail=f"Failed to get token for room: {room_url}")
status_code=500, detail=f"Failed to get token for room: {room.url}")
# Spawn a new agent, and join the user session
# Note: this is mostly for demonstration purposes (refer to 'deployment' in README)
try:
proc = subprocess.Popen(
[
f"python3 -m bot -u {room_url} -t {token}"
f"python3 -m bot -u {room.url} -t {token}"
],
shell=True,
bufsize=1,
cwd=os.path.dirname(os.path.abspath(__file__))
)
bot_procs[proc.pid] = (proc, room_url)
bot_procs[proc.pid] = (proc, room.url)
except Exception as e:
raise HTTPException(
status_code=500, detail=f"Failed to start subprocess: {e}")
return RedirectResponse(room_url)
return RedirectResponse(room.url)
@app.get("/status/{pid}")

109
examples/moondream-chatbot/utils/daily_helpers.py
View File

@@ -1,109 +0,0 @@
import urllib.parse
import os
import time
import urllib
import requests
from dotenv import load_dotenv
load_dotenv()
daily_api_path = os.getenv("DAILY_API_URL") or "api.daily.co/v1"
daily_api_key = os.getenv("DAILY_API_KEY")
def create_room() -> tuple[str, str]:
"""
Helper function to create a Daily room.
# See: https://docs.daily.co/reference/rest-api/rooms
Returns:
tuple: A tuple containing the room URL and room name.
Raises:
Exception: If the request to create the room fails or if the response does not contain the room URL or room name.
"""
room_props = {
"exp": time.time() + 60 * 60, # 1 hour
"enable_chat": True,
"enable_emoji_reactions": True,
"eject_at_room_exp": True,
"enable_prejoin_ui": False, # Important for the bot to be able to join headlessly
}
res = requests.post(
f"https://{daily_api_path}/rooms",
headers={"Authorization": f"Bearer {daily_api_key}"},
json={
"properties": room_props
},
)
if res.status_code != 200:
raise Exception(f"Unable to create room: {res.text}")
data = res.json()
room_url: str = data.get("url")
room_name: str = data.get("name")
if room_url is None or room_name is None:
raise Exception("Missing room URL or room name in response")
return room_url, room_name
def get_name_from_url(room_url: str) -> str:
"""
Extracts the name from a given room URL.
Args:
room_url (str): The URL of the room.
Returns:
str: The extracted name from the room URL.
"""
return urllib.parse.urlparse(room_url).path[1:]
def get_token(room_url: str) -> str:
"""
Retrieves a meeting token for the specified Daily room URL.
# See: https://docs.daily.co/reference/rest-api/meeting-tokens
Args:
room_url (str): The URL of the Daily room.
Returns:
str: The meeting token.
Raises:
Exception: If no room URL is specified or if no Daily API key is specified.
Exception: If there is an error creating the meeting token.
"""
if not room_url:
raise Exception(
"No Daily room specified. You must specify a Daily room in order a token to be generated.")
if not daily_api_key:
raise Exception(
"No Daily API key specified. set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.")
expiration: float = time.time() + 60 * 60
room_name = get_name_from_url(room_url)
res: requests.Response = requests.post(
f"https://{daily_api_path}/meeting-tokens",
headers={
"Authorization": f"Bearer {daily_api_key}"},
json={
"properties": {
"room_name": room_name,
"is_owner": True, # Owner tokens required for transcription
"exp": expiration}},
)
if res.status_code != 200:
raise Exception(
f"Failed to create meeting token: {res.status_code} {res.text}")
token: str = res.json()["token"]
return token

9
examples/patient-intake/bot.py
View File

@@ -257,7 +257,8 @@ class IntakeProcessor:
return None
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -265,9 +266,6 @@ async def main(room_url: str, token):
"Chatbot",
DailyParams(
audio_out_enabled=True,
camera_out_enabled=True,
camera_out_width=1024,
camera_out_height=576,
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
transcription_enabled=True,
@@ -351,5 +349,4 @@ async def main(room_url: str, token):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

1
examples/patient-intake/requirements.txt
View File

@@ -1,5 +1,4 @@
python-dotenv
requests
fastapi[all]
uvicorn
pipecat-ai[daily,openai,silero]

35
examples/patient-intake/runner.py
View File

@@ -1,11 +1,16 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import argparse
import os
import time
import urllib
import requests
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
def configure():
async def configure():
parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
parser.add_argument(
"-u",
@@ -33,26 +38,12 @@ def configure():
if not key:
raise Exception("No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.")
daily_rest_helper = DailyRESTHelper(key, os.getenv("DAILY_API_URL", "https://api.daily.co/v1"))
# Create a meeting token for the given room with an expiration 1 hour in
# the future.
room_name: str = urllib.parse.urlparse(url).path[1:]
expiration: float = time.time() + 60 * 60
expiry_time: float = 60 * 60
res: requests.Response = requests.post(
f"https://api.daily.co/v1/meeting-tokens",
headers={
"Authorization": f"Bearer {key}"},
json={
"properties": {
"room_name": room_name,
"is_owner": True,
"exp": expiration}},
)
if res.status_code != 200:
raise Exception(
f"Failed to create meeting token: {res.status_code} {res.text}")
token: str = res.json()["token"]
token = await daily_rest_helper.get_token(url, expiry_time)
return (url, token)

35
examples/patient-intake/server.py
View File

@@ -1,3 +1,9 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import os
import argparse
import subprocess
@@ -7,17 +13,22 @@ from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, RedirectResponse
from utils.daily_helpers import create_room as _create_room, get_token
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomParams
MAX_BOTS_PER_ROOM = 1
# Bot sub-process dict for status reporting and concurrency control
bot_procs = {}
daily_rest_helper = DailyRESTHelper(
os.getenv("DAILY_API_KEY", ""),
os.getenv("DAILY_API_URL", 'https://api.daily.co/v1'))
def cleanup():
# Clean up function, just to be extra safe
for proc in bot_procs.values():
for entry in bot_procs.values():
proc = entry[0]
proc.terminate()
proc.wait()
@@ -39,45 +50,45 @@ app.add_middleware(
@app.get("/start")
async def start_agent(request: Request):
print(f"!!! Creating room")
room_url, room_name = _create_room()
print(f"!!! Room URL: {room_url}")
room = await daily_rest_helper.create_room(DailyRoomParams())
print(f"!!! Room URL: {room.url}")
# Ensure the room property is present
if not room_url:
if not room.url:
raise HTTPException(
status_code=500,
detail="Missing 'room' property in request data. Cannot start agent without a target room!")
# Check if there is already an existing process running in this room
num_bots_in_room = sum(
1 for proc in bot_procs.values() if proc[1] == room_url and proc[0].poll() is None)
1 for proc in bot_procs.values() if proc[1] == room.url and proc[0].poll() is None)
if num_bots_in_room >= MAX_BOTS_PER_ROOM:
raise HTTPException(
status_code=500, detail=f"Max bot limited reach for room: {room_url}")
status_code=500, detail=f"Max bot limited reach for room: {room.url}")
# Get the token for the room
token = get_token(room_url)
token = await daily_rest_helper.get_token(room.url)
if not token:
raise HTTPException(
status_code=500, detail=f"Failed to get token for room: {room_url}")
status_code=500, detail=f"Failed to get token for room: {room.url}")
# Spawn a new agent, and join the user session
# Note: this is mostly for demonstration purposes (refer to 'deployment' in README)
try:
proc = subprocess.Popen(
[
f"python3 -m bot -u {room_url} -t {token}"
f"python3 -m bot -u {room.url} -t {token}"
],
shell=True,
bufsize=1,
cwd=os.path.dirname(os.path.abspath(__file__))
)
bot_procs[proc.pid] = (proc, room_url)
bot_procs[proc.pid] = (proc, room.url)
except Exception as e:
raise HTTPException(
status_code=500, detail=f"Failed to start subprocess: {e}")
return RedirectResponse(room_url)
return RedirectResponse(room.url)
@app.get("/status/{pid}")

109
examples/patient-intake/utils/daily_helpers.py
View File

@@ -1,109 +0,0 @@
import urllib.parse
import os
import time
import urllib
import requests
from dotenv import load_dotenv
load_dotenv()
daily_api_path = os.getenv("DAILY_API_URL") or "api.daily.co/v1"
daily_api_key = os.getenv("DAILY_API_KEY")
def create_room() -> tuple[str, str]:
"""
Helper function to create a Daily room.
# See: https://docs.daily.co/reference/rest-api/rooms
Returns:
tuple: A tuple containing the room URL and room name.
Raises:
Exception: If the request to create the room fails or if the response does not contain the room URL or room name.
"""
room_props = {
"exp": time.time() + 60 * 60, # 1 hour
"enable_chat": True,
"enable_emoji_reactions": True,
"eject_at_room_exp": True,
"enable_prejoin_ui": False, # Important for the bot to be able to join headlessly
}
res = requests.post(
f"https://{daily_api_path}/rooms",
headers={"Authorization": f"Bearer {daily_api_key}"},
json={
"properties": room_props
},
)
if res.status_code != 200:
raise Exception(f"Unable to create room: {res.text}")
data = res.json()
room_url: str = data.get("url")
room_name: str = data.get("name")
if room_url is None or room_name is None:
raise Exception("Missing room URL or room name in response")
return room_url, room_name
def get_name_from_url(room_url: str) -> str:
"""
Extracts the name from a given room URL.
Args:
room_url (str): The URL of the room.
Returns:
str: The extracted name from the room URL.
"""
return urllib.parse.urlparse(room_url).path[1:]
def get_token(room_url: str) -> str:
"""
Retrieves a meeting token for the specified Daily room URL.
# See: https://docs.daily.co/reference/rest-api/meeting-tokens
Args:
room_url (str): The URL of the Daily room.
Returns:
str: The meeting token.
Raises:
Exception: If no room URL is specified or if no Daily API key is specified.
Exception: If there is an error creating the meeting token.
"""
if not room_url:
raise Exception(
"No Daily room specified. You must specify a Daily room in order a token to be generated.")
if not daily_api_key:
raise Exception(
"No Daily API key specified. set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.")
expiration: float = time.time() + 60 * 60
room_name = get_name_from_url(room_url)
res: requests.Response = requests.post(
f"https://{daily_api_path}/meeting-tokens",
headers={
"Authorization": f"Bearer {daily_api_key}"},
json={
"properties": {
"room_name": room_name,
"is_owner": True, # Owner tokens required for transcription
"exp": expiration}},
)
if res.status_code != 200:
raise Exception(
f"Failed to create meeting token: {res.status_code} {res.text}")
token: str = res.json()["token"]
return token

12
examples/simple-chatbot/bot.py
View File

@@ -1,3 +1,9 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import aiohttp
import os
@@ -77,7 +83,8 @@ class TalkingAnimation(FrameProcessor):
await self.push_frame(frame)
async def main(room_url: str, token):
async def main():
(room_url, token) = await configure()
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
@@ -165,5 +172,4 @@ async def main(room_url: str, token):
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

1
examples/simple-chatbot/requirements.txt
View File

@@ -1,5 +1,4 @@
python-dotenv
requests
fastapi[all]
uvicorn
pipecat-ai[daily,openai,silero]

35
examples/simple-chatbot/runner.py
View File

@@ -1,11 +1,16 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import argparse
import os
import time
import urllib
import requests
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
def configure():
async def configure():
parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
parser.add_argument(
"-u",
@@ -33,26 +38,12 @@ def configure():
if not key:
raise Exception("No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.")
daily_rest_helper = DailyRESTHelper(key, os.getenv("DAILY_API_URL", "https://api.daily.co/v1"))
# Create a meeting token for the given room with an expiration 1 hour in
# the future.
room_name: str = urllib.parse.urlparse(url).path[1:]
expiration: float = time.time() + 60 * 60
expiry_time: float = 60 * 60
res: requests.Response = requests.post(
f"https://api.daily.co/v1/meeting-tokens",
headers={
"Authorization": f"Bearer {key}"},
json={
"properties": {
"room_name": room_name,
"is_owner": True,
"exp": expiration}},
)
if res.status_code != 200:
raise Exception(
f"Failed to create meeting token: {res.status_code} {res.text}")
token: str = res.json()["token"]
token = await daily_rest_helper.get_token(url, expiry_time)
return (url, token)

35
examples/simple-chatbot/server.py
View File

@@ -1,3 +1,9 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import os
import argparse
import subprocess
@@ -7,17 +13,22 @@ from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, RedirectResponse
from utils.daily_helpers import create_room as _create_room, get_token
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomParams
MAX_BOTS_PER_ROOM = 1
# Bot sub-process dict for status reporting and concurrency control
bot_procs = {}
daily_rest_helper = DailyRESTHelper(
os.getenv("DAILY_API_KEY", ""),
os.getenv("DAILY_API_URL", 'https://api.daily.co/v1'))
def cleanup():
# Clean up function, just to be extra safe
for proc in bot_procs.values():
for entry in bot_procs.values():
proc = entry[0]
proc.terminate()
proc.wait()
@@ -39,45 +50,45 @@ app.add_middleware(
@app.get("/start")
async def start_agent(request: Request):
print(f"!!! Creating room")
room_url, room_name = _create_room()
print(f"!!! Room URL: {room_url}")
room = await daily_rest_helper.create_room(DailyRoomParams())
print(f"!!! Room URL: {room.url}")
# Ensure the room property is present
if not room_url:
if not room.url:
raise HTTPException(
status_code=500,
detail="Missing 'room' property in request data. Cannot start agent without a target room!")
# Check if there is already an existing process running in this room
num_bots_in_room = sum(
1 for proc in bot_procs.values() if proc[1] == room_url and proc[0].poll() is None)
1 for proc in bot_procs.values() if proc[1] == room.url and proc[0].poll() is None)
if num_bots_in_room >= MAX_BOTS_PER_ROOM:
raise HTTPException(
status_code=500, detail=f"Max bot limited reach for room: {room_url}")
status_code=500, detail=f"Max bot limited reach for room: {room.url}")
# Get the token for the room
token = get_token(room_url)
token = await daily_rest_helper.get_token(room.url)
if not token:
raise HTTPException(
status_code=500, detail=f"Failed to get token for room: {room_url}")
status_code=500, detail=f"Failed to get token for room: {room.url}")
# Spawn a new agent, and join the user session
# Note: this is mostly for demonstration purposes (refer to 'deployment' in README)
try:
proc = subprocess.Popen(
[
f"python3 -m bot -u {room_url} -t {token}"
f"python3 -m bot -u {room.url} -t {token}"
],
shell=True,
bufsize=1,
cwd=os.path.dirname(os.path.abspath(__file__))
)
bot_procs[proc.pid] = (proc, room_url)
bot_procs[proc.pid] = (proc, room.url)
except Exception as e:
raise HTTPException(
status_code=500, detail=f"Failed to start subprocess: {e}")
return RedirectResponse(room_url)
return RedirectResponse(room.url)
@app.get("/status/{pid}")

109
examples/simple-chatbot/utils/daily_helpers.py
View File

@@ -1,109 +0,0 @@
import urllib.parse
import os
import time
import urllib
import requests
from dotenv import load_dotenv
load_dotenv()
daily_api_path = os.getenv("DAILY_API_URL") or "api.daily.co/v1"
daily_api_key = os.getenv("DAILY_API_KEY")
def create_room() -> tuple[str, str]:
"""
Helper function to create a Daily room.
# See: https://docs.daily.co/reference/rest-api/rooms
Returns:
tuple: A tuple containing the room URL and room name.
Raises:
Exception: If the request to create the room fails or if the response does not contain the room URL or room name.
"""
room_props = {
"exp": time.time() + 60 * 60, # 1 hour
"enable_chat": True,
"enable_emoji_reactions": True,
"eject_at_room_exp": True,
"enable_prejoin_ui": False, # Important for the bot to be able to join headlessly
}
res = requests.post(
f"https://{daily_api_path}/rooms",
headers={"Authorization": f"Bearer {daily_api_key}"},
json={
"properties": room_props
},
)
if res.status_code != 200:
raise Exception(f"Unable to create room: {res.text}")
data = res.json()
room_url: str = data.get("url")
room_name: str = data.get("name")
if room_url is None or room_name is None:
raise Exception("Missing room URL or room name in response")
return room_url, room_name
def get_name_from_url(room_url: str) -> str:
"""
Extracts the name from a given room URL.
Args:
room_url (str): The URL of the room.
Returns:
str: The extracted name from the room URL.
"""
return urllib.parse.urlparse(room_url).path[1:]
def get_token(room_url: str) -> str:
"""
Retrieves a meeting token for the specified Daily room URL.
# See: https://docs.daily.co/reference/rest-api/meeting-tokens
Args:
room_url (str): The URL of the Daily room.
Returns:
str: The meeting token.
Raises:
Exception: If no room URL is specified or if no Daily API key is specified.
Exception: If there is an error creating the meeting token.
"""
if not room_url:
raise Exception(
"No Daily room specified. You must specify a Daily room in order a token to be generated.")
if not daily_api_key:
raise Exception(
"No Daily API key specified. set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.")
expiration: float = time.time() + 60 * 60
room_name = get_name_from_url(room_url)
res: requests.Response = requests.post(
f"https://{daily_api_path}/meeting-tokens",
headers={
"Authorization": f"Bearer {daily_api_key}"},
json={
"properties": {
"room_name": room_name,
"is_owner": True, # Owner tokens required for transcription
"exp": expiration}},
)
if res.status_code != 200:
raise Exception(
f"Failed to create meeting token: {res.status_code} {res.text}")
token: str = res.json()["token"]
return token

978
examples/storytelling-chatbot/frontend/yarn.lock
View File

File diff suppressed because it is too large Load Diff

1
examples/storytelling-chatbot/requirements.txt
View File

@@ -1,6 +1,5 @@
async_timeout
fastapi
uvicorn
requests
python-dotenv
pipecat-ai[daily,openai,fal]

109
examples/storytelling-chatbot/src/bot_runner.py
View File

@@ -1,7 +1,14 @@
import os
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import aiohttp
import argparse
import subprocess
import requests
import os
from pathlib import Path
from typing import Optional
@@ -10,7 +17,8 @@ from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse, JSONResponse
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomObject, DailyRoomProperties, DailyRoomParams
from pipecat.transports.services.helpers.daily_rest import (
DailyRESTHelper, DailyRoomObject, DailyRoomProperties, DailyRoomParams)
from dotenv import load_dotenv
@@ -70,7 +78,7 @@ async def start_bot(request: Request) -> JSONResponse:
properties=DailyRoomProperties()
)
try:
room: DailyRoomObject = daily_rest_helper.create_room(params=params)
room: DailyRoomObject = await daily_rest_helper.create_room(params=params)
except Exception as e:
raise HTTPException(
status_code=500,
@@ -78,13 +86,13 @@ async def start_bot(request: Request) -> JSONResponse:
else:
# Check passed room URL exists, we should assume that it already has a sip set up
try:
room: DailyRoomObject = daily_rest_helper.get_room_from_url(room_url)
room: DailyRoomObject = await daily_rest_helper.get_room_from_url(room_url)
except Exception:
raise HTTPException(
status_code=500, detail=f"Room not found: {room_url}")
# Give the agent a token to join the session
token = daily_rest_helper.get_token(room.url, MAX_SESSION_TIME)
token = await daily_rest_helper.get_token(room.url, MAX_SESSION_TIME)
if not room or not token:
raise HTTPException(
@@ -93,7 +101,7 @@ async def start_bot(request: Request) -> JSONResponse:
# Launch a new VM, or run as a shell process (not recommended)
if os.getenv("RUN_AS_VM", False):
try:
virtualize_bot(room.url, token)
await virtualize_bot(room.url, token)
except Exception as e:
raise HTTPException(
status_code=500, detail=f"Failed to spawn VM: {e}")
@@ -109,7 +117,7 @@ async def start_bot(request: Request) -> JSONResponse:
status_code=500, detail=f"Failed to start subprocess: {e}")
# Grab a token for the user to join with
user_token = daily_rest_helper.get_token(room.url, MAX_SESSION_TIME)
user_token = await daily_rest_helper.get_token(room.url, MAX_SESSION_TIME)
return JSONResponse({
"room_url": room.url,
@@ -136,7 +144,7 @@ async def catch_all(path_name: Optional[str] = ""):
# ------------ Virtualization ------------ #
def virtualize_bot(room_url: str, token: str):
async def virtualize_bot(room_url: str, token: str):
"""
This is an example of how to virtualize the bot using Fly.io
You can adapt this method to use whichever cloud provider you prefer.
@@ -149,54 +157,53 @@ def virtualize_bot(room_url: str, token: str):
'Content-Type': 'application/json'
}
# Use the same image as the bot runner
res = requests.get(f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines", headers=FLY_HEADERS)
if res.status_code != 200:
raise Exception(f"Unable to get machine info from Fly: {res.text}")
image = res.json()[0]['config']['image']
async with aiohttp.ClientSession() as session:
# Use the same image as the bot runner
async with session.get(f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines", headers=FLY_HEADERS) as r:
if r.status != 200:
text = await r.text()
raise Exception(f"Unable to get machine info from Fly: {text}")
# Machine configuration
cmd = f"python3 src/bot.py -u {room_url} -t {token}"
cmd = cmd.split()
worker_props = {
"config": {
"image": image,
"auto_destroy": True,
"init": {
"cmd": cmd
data = await r.json()
image = data[0]['config']['image']
# Machine configuration
cmd = f"python3 src/bot.py -u {room_url} -t {token}"
cmd = cmd.split()
worker_props = {
"config": {
"image": image,
"auto_destroy": True,
"init": {
"cmd": cmd
},
"restart": {
"policy": "no"
},
"guest": {
"cpu_kind": "shared",
"cpus": 1,
"memory_mb": 512
}
},
"restart": {
"policy": "no"
},
"guest": {
"cpu_kind": "shared",
"cpus": 1,
"memory_mb": 512
}
},
}
}
# Spawn a new machine instance
async with session.post(f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines", headers=FLY_HEADERS, json=worker_props) as r:
if r.status != 200:
text = await r.text()
raise Exception(f"Problem starting a bot worker: {text}")
# Spawn a new machine instance
res = requests.post(
f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines",
headers=FLY_HEADERS,
json=worker_props)
data = await r.json()
# Wait for the machine to enter the started state
vm_id = data['id']
if res.status_code != 200:
raise Exception(f"Problem starting a bot worker: {res.text}")
async with session.get(f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines/{vm_id}/wait?state=started", headers=FLY_HEADERS) as r:
if r.status != 200:
text = await r.text()
raise Exception(f"Bot was unable to enter started state: {text}")
# Wait for the machine to enter the started state
vm_id = res.json()['id']
res = requests.get(
f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines/{vm_id}/wait?state=started",
headers=FLY_HEADERS)
if res.status_code != 200:
raise Exception(f"Bot was unable to enter started state: {res.text}")
print(f"Machine joined room: {room_url}")
print(f"Machine joined room: {room_url}")
# ------------ Main ------------ #

109
examples/storytelling-chatbot/src/utils/daily_helpers.py
View File

@@ -1,109 +0,0 @@
import urllib.parse
import os
import time
import urllib
import requests
from dotenv import load_dotenv
load_dotenv()
daily_api_path = os.getenv("DAILY_API_URL") or "api.daily.co/v1"
daily_api_key = os.getenv("DAILY_API_KEY")
def create_room() -> tuple[str, str]:
"""
Helper function to create a Daily room.
# See: https://docs.daily.co/reference/rest-api/rooms
Returns:
tuple: A tuple containing the room URL and room name.
Raises:
Exception: If the request to create the room fails or if the response does not contain the room URL or room name.
"""
room_props = {
"exp": time.time() + 60 * 60, # 1 hour
"enable_chat": True,
"enable_emoji_reactions": True,
"eject_at_room_exp": True,
"enable_prejoin_ui": False, # Important for the bot to be able to join headlessly
}
res = requests.post(
f"https://{daily_api_path}/rooms",
headers={"Authorization": f"Bearer {daily_api_key}"},
json={
"properties": room_props
},
)
if res.status_code != 200:
raise Exception(f"Unable to create room: {res.text}")
data = res.json()
room_url: str = data.get("url")
room_name: str = data.get("name")
if room_url is None or room_name is None:
raise Exception("Missing room URL or room name in response")
return room_url, room_name
def get_name_from_url(room_url: str) -> str:
"""
Extracts the name from a given room URL.
Args:
room_url (str): The URL of the room.
Returns:
str: The extracted name from the room URL.
"""
return urllib.parse.urlparse(room_url).path[1:]
def get_token(room_url: str) -> str:
"""
Retrieves a meeting token for the specified Daily room URL.
# See: https://docs.daily.co/reference/rest-api/meeting-tokens
Args:
room_url (str): The URL of the Daily room.
Returns:
str: The meeting token.
Raises:
Exception: If no room URL is specified or if no Daily API key is specified.
Exception: If there is an error creating the meeting token.
"""
if not room_url:
raise Exception(
"No Daily room specified. You must specify a Daily room in order a token to be generated.")
if not daily_api_key:
raise Exception(
"No Daily API key specified. set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.")
expiration: float = time.time() + 60 * 60
room_name = get_name_from_url(room_url)
res: requests.Response = requests.post(
f"https://{daily_api_path}/meeting-tokens",
headers={
"Authorization": f"Bearer {daily_api_key}"},
json={
"properties": {
"room_name": room_name,
"is_owner": True, # Owner tokens required for transcription
"exp": expiration}},
)
if res.status_code != 200:
raise Exception(
f"Failed to create meeting token: {res.status_code} {res.text}")
token: str = res.json()["token"]
return token

101
examples/translation-chatbot/bot.py
View File

@@ -1,5 +1,10 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import aiohttp
import os
import sys
@@ -12,7 +17,11 @@ from pipecat.processors.aggregators.sentence import SentenceAggregator
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.azure import AzureTTSService
from pipecat.services.openai import OpenAILLMService
from pipecat.transports.services.daily import DailyParams, DailyTranscriptionSettings, DailyTransport, DailyTransportMessageFrame
from pipecat.transports.services.daily import (
DailyParams,
DailyTranscriptionSettings,
DailyTransport,
DailyTransportMessageFrame)
from runner import configure
@@ -79,59 +88,59 @@ class TranslationSubtitles(FrameProcessor):
await self.push_frame(frame)
async def main(room_url: str, token):
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
token,
"Translator",
DailyParams(
audio_out_enabled=True,
transcription_enabled=True,
transcription_settings=DailyTranscriptionSettings(extra={
"interim_results": False
})
)
async def main():
(room_url, token) = await configure()
transport = DailyTransport(
room_url,
token,
"Translator",
DailyParams(
audio_out_enabled=True,
transcription_enabled=True,
transcription_settings=DailyTranscriptionSettings(extra={
"interim_results": False
})
)
)
tts = AzureTTSService(
api_key=os.getenv("AZURE_SPEECH_API_KEY"),
region=os.getenv("AZURE_SPEECH_REGION"),
voice="es-ES-AlvaroNeural",
)
tts = AzureTTSService(
api_key=os.getenv("AZURE_SPEECH_API_KEY"),
region=os.getenv("AZURE_SPEECH_REGION"),
voice="es-ES-AlvaroNeural",
)
llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"),
model="gpt-4o"
)
llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"),
model="gpt-4o"
)
sa = SentenceAggregator()
tp = TranslationProcessor("Spanish")
lfra = LLMFullResponseAggregator()
ts = TranslationSubtitles("spanish")
sa = SentenceAggregator()
tp = TranslationProcessor("Spanish")
lfra = LLMFullResponseAggregator()
ts = TranslationSubtitles("spanish")
pipeline = Pipeline([
transport.input(),
sa,
tp,
llm,
lfra,
ts,
tts,
transport.output()
])
pipeline = Pipeline([
transport.input(),
sa,
tp,
llm,
lfra,
ts,
tts,
transport.output()
])
task = PipelineTask(pipeline)
task = PipelineTask(pipeline)
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
transport.capture_participant_transcription(participant["id"])
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
transport.capture_participant_transcription(participant["id"])
runner = PipelineRunner()
runner = PipelineRunner()
await runner.run(task)
await runner.run(task)
if __name__ == "__main__":
(url, token) = configure()
asyncio.run(main(url, token))
asyncio.run(main())

1
examples/translation-chatbot/requirements.txt
View File

@@ -1,4 +1,3 @@
python-dotenv
requests
fastapi[all]
pipecat-ai[daily,openai,azure]

35
examples/translation-chatbot/runner.py
View File

@@ -1,11 +1,16 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import argparse
import os
import time
import urllib
import requests
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
def configure():
async def configure():
parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
parser.add_argument(
"-u",
@@ -33,26 +38,12 @@ def configure():
if not key:
raise Exception("No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.")
daily_rest_helper = DailyRESTHelper(key, os.getenv("DAILY_API_URL", "https://api.daily.co/v1"))
# Create a meeting token for the given room with an expiration 1 hour in
# the future.
room_name: str = urllib.parse.urlparse(url).path[1:]
expiration: float = time.time() + 60 * 60
expiry_time: float = 60 * 60
res: requests.Response = requests.post(
f"https://api.daily.co/v1/meeting-tokens",
headers={
"Authorization": f"Bearer {key}"},
json={
"properties": {
"room_name": room_name,
"is_owner": True,
"exp": expiration}},
)
if res.status_code != 200:
raise Exception(
f"Failed to create meeting token: {res.status_code} {res.text}")
token: str = res.json()["token"]
token = await daily_rest_helper.get_token(url, expiry_time)
return (url, token)

35
examples/translation-chatbot/server.py
View File

@@ -1,3 +1,9 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import os
import argparse
import subprocess
@@ -7,17 +13,22 @@ from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, RedirectResponse
from utils.daily_helpers import create_room as _create_room, get_token
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomParams
MAX_BOTS_PER_ROOM = 1
# Bot sub-process dict for status reporting and concurrency control
bot_procs = {}
daily_rest_helper = DailyRESTHelper(
os.getenv("DAILY_API_KEY", ""),
os.getenv("DAILY_API_URL", 'https://api.daily.co/v1'))
def cleanup():
# Clean up function, just to be extra safe
for proc in bot_procs.values():
for entry in bot_procs.values():
proc = entry[0]
proc.terminate()
proc.wait()
@@ -39,45 +50,45 @@ app.add_middleware(
@app.get("/start")
async def start_agent(request: Request):
print(f"!!! Creating room")
room_url, room_name = _create_room()
print(f"!!! Room URL: {room_url}")
room = await daily_rest_helper.create_room(DailyRoomParams())
print(f"!!! Room URL: {room.url}")
# Ensure the room property is present
if not room_url:
if not room.url:
raise HTTPException(
status_code=500,
detail="Missing 'room' property in request data. Cannot start agent without a target room!")
# Check if there is already an existing process running in this room
num_bots_in_room = sum(
1 for proc in bot_procs.values() if proc[1] == room_url and proc[0].poll() is None)
1 for proc in bot_procs.values() if proc[1] == room.url and proc[0].poll() is None)
if num_bots_in_room >= MAX_BOTS_PER_ROOM:
raise HTTPException(
status_code=500, detail=f"Max bot limited reach for room: {room_url}")
status_code=500, detail=f"Max bot limited reach for room: {room.url}")
# Get the token for the room
token = get_token(room_url)
token = await daily_rest_helper.get_token(room.url)
if not token:
raise HTTPException(
status_code=500, detail=f"Failed to get token for room: {room_url}")
status_code=500, detail=f"Failed to get token for room: {room.url}")
# Spawn a new agent, and join the user session
# Note: this is mostly for demonstration purposes (refer to 'deployment' in README)
try:
proc = subprocess.Popen(
[
f"python3 -m bot -u {room_url} -t {token}"
f"python3 -m bot -u {room.url} -t {token}"
],
shell=True,
bufsize=1,
cwd=os.path.dirname(os.path.abspath(__file__))
)
bot_procs[proc.pid] = (proc, room_url)
bot_procs[proc.pid] = (proc, room.url)
except Exception as e:
raise HTTPException(
status_code=500, detail=f"Failed to start subprocess: {e}")
return RedirectResponse(room_url)
return RedirectResponse(room.url)
@app.get("/status/{pid}")

109
examples/translation-chatbot/utils/daily_helpers.py
View File

@@ -1,109 +0,0 @@
import urllib.parse
import os
import time
import urllib
import requests
from dotenv import load_dotenv
load_dotenv()
daily_api_path = os.getenv("DAILY_API_URL") or "api.daily.co/v1"
daily_api_key = os.getenv("DAILY_API_KEY")
def create_room() -> tuple[str, str]:
"""
Helper function to create a Daily room.
# See: https://docs.daily.co/reference/rest-api/rooms
Returns:
tuple: A tuple containing the room URL and room name.
Raises:
Exception: If the request to create the room fails or if the response does not contain the room URL or room name.
"""
room_props = {
"exp": time.time() + 60 * 60, # 1 hour
"enable_chat": True,
"enable_emoji_reactions": True,
"eject_at_room_exp": True,
"enable_prejoin_ui": False, # Important for the bot to be able to join headlessly
}
res = requests.post(
f"https://{daily_api_path}/rooms",
headers={"Authorization": f"Bearer {daily_api_key}"},
json={
"properties": room_props
},
)
if res.status_code != 200:
raise Exception(f"Unable to create room: {res.text}")
data = res.json()
room_url: str = data.get("url")
room_name: str = data.get("name")
if room_url is None or room_name is None:
raise Exception("Missing room URL or room name in response")
return room_url, room_name
def get_name_from_url(room_url: str) -> str:
"""
Extracts the name from a given room URL.
Args:
room_url (str): The URL of the room.
Returns:
str: The extracted name from the room URL.
"""
return urllib.parse.urlparse(room_url).path[1:]
def get_token(room_url: str) -> str:
"""
Retrieves a meeting token for the specified Daily room URL.
# See: https://docs.daily.co/reference/rest-api/meeting-tokens
Args:
room_url (str): The URL of the Daily room.
Returns:
str: The meeting token.
Raises:
Exception: If no room URL is specified or if no Daily API key is specified.
Exception: If there is an error creating the meeting token.
"""
if not room_url:
raise Exception(
"No Daily room specified. You must specify a Daily room in order a token to be generated.")
if not daily_api_key:
raise Exception(
"No Daily API key specified. set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.")
expiration: float = time.time() + 60 * 60
room_name = get_name_from_url(room_url)
res: requests.Response = requests.post(
f"https://{daily_api_path}/meeting-tokens",
headers={
"Authorization": f"Bearer {daily_api_key}"},
json={
"properties": {
"room_name": room_name,
"is_owner": True, # Owner tokens required for transcription
"exp": expiration}},
)
if res.status_code != 200:
raise Exception(
f"Failed to create meeting token: {res.status_code} {res.text}")
token: str = res.json()["token"]
return token

15
examples/twilio-chatbot/README.md
View File

@@ -46,10 +46,16 @@ This project is a FastAPI-based chatbot that integrates with Twilio to handle We
## Configure Twilio URLs
1. **Update the Twilio Webhook**:
1. **Start ngrok**:
In a new terminal, start ngrok to tunnel the local server:
```sh
ngrok http 8765
```
2. **Update the Twilio Webhook**:
Copy the ngrok URL and update your Twilio phone number webhook URL to `http://<ngrok_url>/start_call`.
2. **Update the streams.xml**:
3. **Update the streams.xml**:
Copy the ngrok URL and update templates/streams.xml with `wss://<ngrok_url>/ws`.
## Running the Application
@@ -61,11 +67,6 @@ This project is a FastAPI-based chatbot that integrates with Twilio to handle We
python server.py
```
2. **Start ngrok**:
In a new terminal, start ngrok to tunnel the local server:
```sh
ngrok http 8765
```
### Using Docker
1. **Build the Docker image**:

25
src/pipecat/services/xtts.py
View File

@@ -6,15 +6,13 @@
import aiohttp
from typing import AsyncGenerator
from typing import Any, AsyncGenerator, Dict
from pipecat.frames.frames import AudioRawFrame, ErrorFrame, Frame
from pipecat.frames.frames import AudioRawFrame, ErrorFrame, Frame, StartFrame
from pipecat.services.ai_services import TTSService
from loguru import logger
import requests
import numpy as np
try:
@@ -48,13 +46,25 @@ class XTTSService(TTSService):
self._voice_id = voice_id
self._language = language
self._base_url = base_url
self._studio_speakers = requests.get(self._base_url + "/studio_speakers").json()
self._studio_speakers: Dict[str, Any] | None = None
self._aiohttp_session = aiohttp_session or aiohttp.ClientSession()
self._close_aiohttp_session = aiohttp_session is None
def can_generate_metrics(self) -> bool:
return True
async def start(self, frame: StartFrame):
await super().start(frame)
async with self._aiohttp_session.get(self._base_url + "/studio_speakers") as r:
if r.status != 200:
text = await r.text()
logger.error(
f"{self} error getting studio speakers (status: {r.status}, error: {text})")
await self.push_error(
ErrorFrame(f"Error error getting studio speakers (status: {r.status}, error: {text})"))
return
self._studio_speakers = await r.json()
async def cleanup(self):
await super().cleanup()
if self._close_aiohttp_session:
@@ -66,6 +76,11 @@ class XTTSService(TTSService):
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"Generating TTS: [{text}]")
if not self._studio_speakers:
logger.error(f"{self} no studio speakers available")
return
embeddings = self._studio_speakers[self._voice_id]
url = self._base_url + "/tts_stream"

83
src/pipecat/transports/services/helpers/daily_rest.py
View File

@@ -11,7 +11,7 @@ Methods that wrap the Daily API to create rooms, check room URLs, and get meetin
"""
import requests
import aiohttp
import time
from urllib.parse import urlparse
@@ -61,24 +61,25 @@ class DailyRoomObject(BaseModel):
class DailyRESTHelper:
def __init__(self, daily_api_key: str, daily_api_url: str = "https://api.daily.co/v1"):
def __init__(self,
daily_api_key: str,
daily_api_url: str = "https://api.daily.co/v1"):
self.daily_api_key = daily_api_key
self.daily_api_url = daily_api_url
def _get_name_from_url(self, room_url: str) -> str:
return urlparse(room_url).path[1:]
def create_room(self, params: DailyRoomParams) -> DailyRoomObject:
res = requests.post(
f"{self.daily_api_url}/rooms",
headers={"Authorization": f"Bearer {self.daily_api_key}"},
json={**params.model_dump(exclude_none=True)}
)
async def create_room(self, params: DailyRoomParams) -> DailyRoomObject:
async with aiohttp.ClientSession() as session:
headers = {"Authorization": f"Bearer {self.daily_api_key}"}
json = {**params.model_dump(exclude_none=True)}
async with session.post(f"{self.daily_api_url}/rooms", headers=headers, json=json) as r:
if r.status != 200:
text = await r.text()
raise Exception(f"Unable to create room: {text}")
if res.status_code != 200:
raise Exception(f"Unable to create room: {res.text}")
data = res.json()
data = await r.json()
try:
room = DailyRoomObject(**data)
@@ -87,29 +88,30 @@ class DailyRESTHelper:
return room
def _get_room_from_name(self, room_name: str) -> DailyRoomObject:
res: requests.Response = requests.get(
f"{self.daily_api_url}/rooms/{room_name}",
headers={"Authorization": f"Bearer {self.daily_api_key}"}
)
async def _get_room_from_name(self, room_name: str) -> DailyRoomObject:
async with aiohttp.ClientSession() as session:
headers = {"Authorization": f"Bearer {self.daily_api_key}"}
async with session.get(f"{self.daily_api_url}/rooms/{room_name}", headers=headers) as r:
if r.status != 200:
raise Exception(f"Room not found: {room_name}")
if res.status_code != 200:
raise Exception(f"Room not found: {room_name}")
data = res.json()
try:
room = DailyRoomObject(**data)
except ValidationError as e:
raise Exception(f"Invalid response: {e}")
data = await r.json()
try:
room = DailyRoomObject(**data)
except ValidationError as e:
raise Exception(f"Invalid response: {e}")
return room
def get_room_from_url(self, room_url: str,) -> DailyRoomObject:
async def get_room_from_url(self, room_url: str,) -> DailyRoomObject:
room_name = self._get_name_from_url(room_url)
return self._get_room_from_name(room_name)
return await self._get_room_from_name(room_name)
def get_token(self, room_url: str, expiry_time: float = 60 * 60, owner: bool = True) -> str:
async def get_token(
self,
room_url: str,
expiry_time: float = 60 * 60,
owner: bool = True) -> str:
if not room_url:
raise Exception(
"No Daily room specified. You must specify a Daily room in order a token to be generated.")
@@ -118,22 +120,21 @@ class DailyRESTHelper:
room_name = self._get_name_from_url(room_url)
res: requests.Response = requests.post(
f"{self.daily_api_url}/meeting-tokens",
headers={
"Authorization": f"Bearer {self.daily_api_key}"},
json={
async with aiohttp.ClientSession() as session:
headers = {"Authorization": f"Bearer {self.daily_api_key}"}
json = {
"properties": {
"room_name": room_name,
"is_owner": owner,
"exp": expiration
}},
)
}
}
async with session.post(f"{self.daily_api_url}/meeting-tokens", headers=headers, json=json) as r:
if r.status != 200:
text = await r.text()
raise Exception(f"Failed to create meeting token: {r.status} {text}")
if res.status_code != 200:
raise Exception(
f"Failed to create meeting token: {res.status_code} {res.text}")
token: str = res.json()["token"]
data = await r.json()
token: str = data["token"]
return token