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

Merge pull request #334 from pipecat-ai/aleix/cleanup-examples-remove-requests

Browse Source 
cleanup examples and remove requests
This commit is contained in:
Aleix Conchillo Flaqué
2024-08-01 22:05:01 -07:00
committed by GitHub
parent 3db7f6a284 bee0b238e4
commit 65b136bf15
75 changed files with 7663 additions and 1838 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
CHANGELOG.md
View File

@@ -26,6 +26,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Fixed ### Fixed
- Improved `EndFrame` and `CancelFrame` handling. `EndFrame` should end things
gracefully while a `CancelFrame` should cancel all running tasks as soon as
possible.
- Fixed an issue in `AIService` that would cause a yielded `None` value to be
processed.
- RTVI's `bot-ready` message is now sent when the RTVI pipeline is ready and - RTVI's `bot-ready` message is now sent when the RTVI pipeline is ready and
a first participant joins. a first participant joins.
@@ -35,6 +42,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 - Fixed a `BaseInputTransport` issue that was causing start/stop interruptions
incoming frames to not cancel tasks and be processed properly. 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 ## [0.0.39] - 2024-07-23
### Fixed ### Fixed

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

@@ -1,14 +1,23 @@
import os #
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import aiohttp
import argparse import argparse
import subprocess import subprocess
import requests import os
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomObject, DailyRoomProperties, DailyRoomParams from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, HTTPException from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse from fastapi.responses import JSONResponse
from pipecat.transports.services.helpers.daily_rest import (
DailyRESTHelper, DailyRoomObject, DailyRoomProperties, DailyRoomParams)
from dotenv import load_dotenv from dotenv import load_dotenv
load_dotenv(override=True) load_dotenv(override=True)
@@ -32,14 +41,23 @@ FLY_HEADERS = {
'Content-Type': 'application/json' 'Content-Type': 'application/json'
} }
daily_rest_helper = DailyRESTHelper( daily_helpers = {}
os.getenv("DAILY_API_KEY", ""),
os.getenv("DAILY_API_URL", 'https://api.daily.co/v1'))
# ----------------- API ----------------- # # ----------------- API ----------------- #
app = FastAPI() @asynccontextmanager
async def lifespan(app: FastAPI):
aiohttp_session = aiohttp.ClientSession()
daily_helpers["rest"] = DailyRESTHelper(
daily_api_key=os.getenv("DAILY_API_KEY", ""),
daily_api_url=os.getenv("DAILY_API_URL", 'https://api.daily.co/v1'),
aiohttp_session=aiohttp_session
)
yield
await aiohttp_session.close()
app = FastAPI(lifespan=lifespan)
app.add_middleware( app.add_middleware(
CORSMiddleware, CORSMiddleware,
@@ -52,53 +70,52 @@ app.add_middleware(
# ----------------- Main ----------------- # # ----------------- Main ----------------- #
def spawn_fly_machine(room_url: str, token: str): async def spawn_fly_machine(room_url: str, token: str):
# Use the same image as the bot runner async with aiohttp.ClientSession() as session:
res = requests.get(f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines", headers=FLY_HEADERS) # Use the same image as the bot runner
if res.status_code != 200: async with session.get(f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines", headers=FLY_HEADERS) as r:
raise Exception(f"Unable to get machine info from Fly: {res.text}") if r.status != 200:
image = res.json()[0]['config']['image'] text = await r.text()
raise Exception(f"Unable to get machine info from Fly: {text}")
# Machine configuration data = await r.json()
cmd = f"python3 bot.py -u {room_url} -t {token}" image = data[0]['config']['image']
cmd = cmd.split()
worker_props = { # Machine configuration
"config": { cmd = f"python3 bot.py -u {room_url} -t {token}"
"image": image, cmd = cmd.split()
"auto_destroy": True, worker_props = {
"init": { "config": {
"cmd": cmd "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 data = await r.json()
res = requests.post( # Wait for the machine to enter the started state
f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines", vm_id = data['id']
headers=FLY_HEADERS,
json=worker_props)
if res.status_code != 200: async with session.get(f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines/{vm_id}/wait?state=started", headers=FLY_HEADERS) as r:
raise Exception(f"Problem starting a bot worker: {res.text}") if r.status != 200:
text = await r.text()
# Wait for the machine to enter the started state raise Exception(f"Bot was unable to enter started state: {text}")
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}")
@@ -121,7 +138,7 @@ async def start_bot(request: Request) -> JSONResponse:
properties=DailyRoomProperties() properties=DailyRoomProperties()
) )
try: try:
room: DailyRoomObject = daily_rest_helper.create_room(params=params) room: DailyRoomObject = await daily_helpers["rest"].create_room(params=params)
except Exception as e: except Exception as e:
raise HTTPException( raise HTTPException(
status_code=500, status_code=500,
@@ -129,13 +146,13 @@ async def start_bot(request: Request) -> JSONResponse:
else: else:
# Check passed room URL exists, we should assume that it already has a sip set up # Check passed room URL exists, we should assume that it already has a sip set up
try: try:
room: DailyRoomObject = daily_rest_helper.get_room_from_url(room_url) room: DailyRoomObject = await daily_helpers["rest"].get_room_from_url(room_url)
except Exception: except Exception:
raise HTTPException( raise HTTPException(
status_code=500, detail=f"Room not found: {room_url}") status_code=500, detail=f"Room not found: {room_url}")
# Give the agent a token to join the session # Give the agent a token to join the session
token = daily_rest_helper.get_token(room.url, MAX_SESSION_TIME) token = await daily_helpers["rest"].get_token(room.url, MAX_SESSION_TIME)
if not room or not token: if not room or not token:
raise HTTPException( raise HTTPException(
@@ -156,13 +173,13 @@ async def start_bot(request: Request) -> JSONResponse:
status_code=500, detail=f"Failed to start subprocess: {e}") status_code=500, detail=f"Failed to start subprocess: {e}")
else: else:
try: try:
spawn_fly_machine(room.url, token) await spawn_fly_machine(room.url, token)
except Exception as e: except Exception as e:
raise HTTPException( raise HTTPException(
status_code=500, detail=f"Failed to spawn VM: {e}") status_code=500, detail=f"Failed to spawn VM: {e}")
# Grab a token for the user to join with # 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_helpers["rest"].get_token(room.url, MAX_SESSION_TIME)
return JSONResponse({ return JSONResponse({
"room_url": room.url, "room_url": room.url,
@@ -194,6 +211,5 @@ if __name__ == "__main__":
port=config.port, port=config.port,
reload=config.reload reload=config.reload
) )
except KeyboardInterrupt: except KeyboardInterrupt:
print("Pipecat runner shutting down...") print("Pipecat runner shutting down...")

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

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

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

@@ -6,15 +6,27 @@ provisioning a room and starting a Pipecat bot in response.
Refer to README for more information. Refer to README for more information.
""" """
import aiohttp
import os import os
import argparse import argparse
import subprocess import subprocess
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomObject, DailyRoomProperties, DailyRoomSipParams, DailyRoomParams
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, HTTPException from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, PlainTextResponse from fastapi.responses import JSONResponse, PlainTextResponse
from twilio.twiml.voice_response import VoiceResponse 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 from dotenv import load_dotenv
load_dotenv(override=True) load_dotenv(override=True)
@@ -25,14 +37,23 @@ MAX_SESSION_TIME = 5 * 60 # 5 minutes
REQUIRED_ENV_VARS = ['OPENAI_API_KEY', 'DAILY_API_KEY', REQUIRED_ENV_VARS = ['OPENAI_API_KEY', 'DAILY_API_KEY',
'ELEVENLABS_API_KEY', 'ELEVENLABS_VOICE_ID'] 'ELEVENLABS_API_KEY', 'ELEVENLABS_VOICE_ID']
daily_rest_helper = DailyRESTHelper( daily_helpers = {}
os.getenv("DAILY_API_KEY", ""),
os.getenv("DAILY_API_URL", 'https://api.daily.co/v1'))
# ----------------- API ----------------- # # ----------------- API ----------------- #
app = FastAPI()
@asynccontextmanager
async def lifespan(app: FastAPI):
aiohttp_session = aiohttp.ClientSession()
daily_helpers["rest"] = DailyRESTHelper(
daily_api_key=os.getenv("DAILY_API_KEY", ""),
daily_api_url=os.getenv("DAILY_API_URL", 'https://api.daily.co/v1'),
aiohttp_session=aiohttp_session
)
yield
await aiohttp_session.close()
app = FastAPI(lifespan=lifespan)
app.add_middleware( app.add_middleware(
CORSMiddleware, CORSMiddleware,
@@ -53,7 +74,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: if not room_url:
params = DailyRoomParams( params = DailyRoomParams(
properties=DailyRoomProperties( properties=DailyRoomProperties(
@@ -68,14 +89,13 @@ def _create_daily_room(room_url, callId, callDomain=None, vendor="daily"):
) )
print(f"Creating new room...") print(f"Creating new room...")
room: DailyRoomObject = daily_rest_helper.create_room(params=params) room: DailyRoomObject = await daily_helpers["rest"].create_room(params=params)
else: else:
# Check passed room URL exist (we assume that it already has a sip set up!) # Check passed room URL exist (we assume that it already has a sip set up!)
try: try:
print(f"Joining existing room: {room_url}") print(f"Joining existing room: {room_url}")
room: DailyRoomObject = daily_rest_helper.get_room_from_url( room: DailyRoomObject = await daily_helpers["rest"].get_room_from_url(room_url)
room_url)
except Exception: except Exception:
raise HTTPException( raise HTTPException(
status_code=500, detail=f"Room not found: {room_url}") status_code=500, detail=f"Room not found: {room_url}")
@@ -83,7 +103,7 @@ def _create_daily_room(room_url, callId, callDomain=None, vendor="daily"):
print(f"Daily room: {room.url} {room.config.sip_endpoint}") print(f"Daily room: {room.url} {room.config.sip_endpoint}")
# Give the agent a token to join the session # Give the agent a token to join the session
token = daily_rest_helper.get_token(room.url, MAX_SESSION_TIME) token = await daily_helpers["rest"].get_token(room.url, MAX_SESSION_TIME)
if not room or not token: if not room or not token:
raise HTTPException( raise HTTPException(
@@ -92,11 +112,11 @@ def _create_daily_room(room_url, callId, callDomain=None, vendor="daily"):
# Spawn a new agent, and join the user session # Spawn a new agent, and join the user session
# Note: this is mostly for demonstration purposes (refer to 'deployment' in docs) # Note: this is mostly for demonstration purposes (refer to 'deployment' in docs)
if vendor == "daily": if vendor == "daily":
bot_proc = f"python3 -m bot_daily -u {room.url} -t {token} -i { bot_proc = f"python3 - m bot_daily - u {room.url} - t {token} - i {
callId} -d {callDomain}" callId} - d {callDomain}"
else: else:
bot_proc = f"python3 -m bot_twilio -u {room.url} -t { bot_proc = f"python3 - m bot_twilio - u {room.url} - t {
token} -i {callId} -s {room.config.sip_endpoint}" token} - i {callId} - s {room.config.sip_endpoint}"
try: try:
subprocess.Popen( subprocess.Popen(
@@ -140,8 +160,7 @@ async def twilio_start_bot(request: Request):
# create room and tell the bot to join the created room # create room and tell the bot to join the created room
# note: Twilio does not require a callDomain # note: Twilio does not require a callDomain
room: DailyRoomObject = _create_daily_room( room: DailyRoomObject = await _create_daily_room(room_url, callId, None, "twilio")
room_url, callId, None, "twilio")
print(f"Put Twilio on hold...") print(f"Put Twilio on hold...")
# We have the room and the SIP URI, # We have the room and the SIP URI,
@@ -178,8 +197,7 @@ async def daily_start_bot(request: Request) -> JSONResponse:
detail="Missing properties 'callId' or 'callDomain'") detail="Missing properties 'callId' or 'callDomain'")
print(f"CallId: {callId}, CallDomain: {callDomain}") print(f"CallId: {callId}, CallDomain: {callDomain}")
room: DailyRoomObject = _create_daily_room( room: DailyRoomObject = await _create_daily_room(room_url, callId, callDomain, "daily")
room_url, callId, callDomain, "daily")
# Grab a token for the user to join with # Grab a token for the user to join with
return JSONResponse({ return JSONResponse({

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

@@ -4,6 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License # SPDX-License-Identifier: BSD 2-Clause License
# #
import aiohttp
import asyncio import asyncio
import os import os
import sys import sys
@@ -31,64 +32,66 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG") logger.add(sys.stderr, level="DEBUG")
async def main(room_url: str, token): async def main():
transport = DailyTransport( async with aiohttp.ClientSession() as session:
room_url, (room_url, token) = await configure(session)
token,
"Respond bot", transport = DailyTransport(
DailyParams( room_url,
audio_out_sample_rate=44100, token,
audio_out_enabled=True, "Respond bot",
transcription_enabled=True, DailyParams(
vad_enabled=True, audio_out_sample_rate=44100,
vad_analyzer=SileroVADAnalyzer() audio_out_enabled=True,
transcription_enabled=True,
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer()
)
) )
)
tts = CartesiaTTSService( tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"), api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="a0e99841-438c-4a64-b679-ae501e7d6091", # Barbershop Man voice_id="a0e99841-438c-4a64-b679-ae501e7d6091", # Barbershop Man
sample_rate=44100, sample_rate=44100,
) )
llm = OpenAILLMService( llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"), api_key=os.getenv("OPENAI_API_KEY"),
model="gpt-4o") model="gpt-4o")
messages = [ messages = [
{ {
"role": "system", "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 characters in your answers. Respond to what the user said in a creative and helpful way.", "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.",
}, },
] ]
tma_in = LLMUserResponseAggregator(messages) tma_in = LLMUserResponseAggregator(messages)
tma_out = LLMAssistantResponseAggregator(messages) tma_out = LLMAssistantResponseAggregator(messages)
pipeline = Pipeline([ pipeline = Pipeline([
transport.input(), # Transport user input transport.input(), # Transport user input
tma_in, # User responses tma_in, # User responses
llm, # LLM llm, # LLM
tts, # TTS tts, # TTS
tma_out, # Goes before the transport because cartesia has word-level timestamps! tma_out, # Goes before the transport because cartesia has word-level timestamps!
transport.output(), # Transport bot output transport.output(), # Transport bot output
]) ])
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True, enable_metrics=True)) task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True, enable_metrics=True))
@transport.event_handler("on_first_participant_joined") @transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant): async def on_first_participant_joined(transport, participant):
transport.capture_participant_transcription(participant["id"]) transport.capture_participant_transcription(participant["id"])
# Kick off the conversation. # Kick off the conversation.
messages.append( messages.append(
{"role": "system", "content": "Please introduce yourself to the user."}) {"role": "system", "content": "Please introduce yourself to the user."})
await task.queue_frames([LLMMessagesFrame(messages)]) await task.queue_frames([LLMMessagesFrame(messages)])
runner = PipelineRunner() runner = PipelineRunner()
await runner.run(task) await runner.run(task)
if __name__ == "__main__": if __name__ == "__main__":
(url, token) = configure() asyncio.run(main())
asyncio.run(main(url, token))

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

@@ -4,6 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License # SPDX-License-Identifier: BSD 2-Clause License
# #
import aiohttp
import asyncio import asyncio
import os import os
import sys import sys
@@ -30,64 +31,66 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG") logger.add(sys.stderr, level="DEBUG")
async def main(room_url: str, token): async def main():
transport = DailyTransport( async with aiohttp.ClientSession() as session:
room_url, (room_url, token) = await configure(session)
token,
"Respond bot", transport = DailyTransport(
DailyParams( room_url,
audio_out_enabled=True, token,
audio_out_sample_rate=16000, "Respond bot",
transcription_enabled=True, DailyParams(
vad_enabled=True, audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer() audio_out_sample_rate=16000,
transcription_enabled=True,
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer()
)
) )
)
tts = PlayHTTTSService( tts = PlayHTTTSService(
user_id=os.getenv("PLAYHT_USER_ID"), user_id=os.getenv("PLAYHT_USER_ID"),
api_key=os.getenv("PLAYHT_API_KEY"), api_key=os.getenv("PLAYHT_API_KEY"),
voice_url="s3://voice-cloning-zero-shot/801a663f-efd0-4254-98d0-5c175514c3e8/jennifer/manifest.json", voice_url="s3://voice-cloning-zero-shot/801a663f-efd0-4254-98d0-5c175514c3e8/jennifer/manifest.json",
) )
llm = OpenAILLMService( llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"), api_key=os.getenv("OPENAI_API_KEY"),
model="gpt-4o") model="gpt-4o")
messages = [ messages = [
{ {
"role": "system", "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 characters in your answers. Respond to what the user said in a creative and helpful way.", "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.",
}, },
] ]
tma_in = LLMUserResponseAggregator(messages) tma_in = LLMUserResponseAggregator(messages)
tma_out = LLMAssistantResponseAggregator(messages) tma_out = LLMAssistantResponseAggregator(messages)
pipeline = Pipeline([ pipeline = Pipeline([
transport.input(), # Transport user input transport.input(), # Transport user input
tma_in, # User responses tma_in, # User responses
llm, # LLM llm, # LLM
tts, # TTS tts, # TTS
transport.output(), # Transport bot output transport.output(), # Transport bot output
tma_out # Assistant spoken responses tma_out # Assistant spoken responses
]) ])
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
@transport.event_handler("on_first_participant_joined") @transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant): async def on_first_participant_joined(transport, participant):
transport.capture_participant_transcription(participant["id"]) transport.capture_participant_transcription(participant["id"])
# Kick off the conversation. # Kick off the conversation.
messages.append( messages.append(
{"role": "system", "content": "Please introduce yourself to the user."}) {"role": "system", "content": "Please introduce yourself to the user."})
await task.queue_frames([LLMMessagesFrame(messages)]) await task.queue_frames([LLMMessagesFrame(messages)])
runner = PipelineRunner() runner = PipelineRunner()
await runner.run(task) await runner.run(task)
if __name__ == "__main__": if __name__ == "__main__":
(url, token) = configure() asyncio.run(main())
asyncio.run(main(url, token))

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

@@ -4,6 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License # SPDX-License-Identifier: BSD 2-Clause License
# #
import aiohttp
import asyncio import asyncio
import os import os
import sys import sys
@@ -30,71 +31,73 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG") logger.add(sys.stderr, level="DEBUG")
async def main(room_url: str, token): async def main():
transport = DailyTransport( async with aiohttp.ClientSession() as session:
room_url, (room_url, token) = await configure(session)
token,
"Respond bot", transport = DailyTransport(
DailyParams( room_url,
audio_out_enabled=True, token,
audio_out_sample_rate=16000, "Respond bot",
vad_enabled=True, DailyParams(
vad_analyzer=SileroVADAnalyzer(), audio_out_enabled=True,
vad_audio_passthrough=True, audio_out_sample_rate=16000,
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
vad_audio_passthrough=True,
)
) )
)
stt = AzureSTTService( stt = AzureSTTService(
api_key=os.getenv("AZURE_SPEECH_API_KEY"), api_key=os.getenv("AZURE_SPEECH_API_KEY"),
region=os.getenv("AZURE_SPEECH_REGION"), region=os.getenv("AZURE_SPEECH_REGION"),
) )
tts = AzureTTSService( tts = AzureTTSService(
api_key=os.getenv("AZURE_SPEECH_API_KEY"), api_key=os.getenv("AZURE_SPEECH_API_KEY"),
region=os.getenv("AZURE_SPEECH_REGION"), region=os.getenv("AZURE_SPEECH_REGION"),
) )
llm = AzureLLMService( llm = AzureLLMService(
api_key=os.getenv("AZURE_CHATGPT_API_KEY"), api_key=os.getenv("AZURE_CHATGPT_API_KEY"),
endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"), endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"),
model=os.getenv("AZURE_CHATGPT_MODEL"), model=os.getenv("AZURE_CHATGPT_MODEL"),
) )
messages = [ messages = [
{ {
"role": "system", "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 characters in your answers. Respond to what the user said in a creative and helpful way.", "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.",
}, },
] ]
tma_in = LLMUserResponseAggregator(messages) tma_in = LLMUserResponseAggregator(messages)
tma_out = LLMAssistantResponseAggregator(messages) tma_out = LLMAssistantResponseAggregator(messages)
pipeline = Pipeline([ pipeline = Pipeline([
transport.input(), # Transport user input transport.input(), # Transport user input
stt, # STT stt, # STT
tma_in, # User responses tma_in, # User responses
llm, # LLM llm, # LLM
tts, # TTS tts, # TTS
transport.output(), # Transport bot output transport.output(), # Transport bot output
tma_out # Assistant spoken responses tma_out # Assistant spoken responses
]) ])
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
@transport.event_handler("on_first_participant_joined") @transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant): async def on_first_participant_joined(transport, participant):
transport.capture_participant_transcription(participant["id"]) transport.capture_participant_transcription(participant["id"])
# Kick off the conversation. # Kick off the conversation.
messages.append( messages.append(
{"role": "system", "content": "Please introduce yourself to the user."}) {"role": "system", "content": "Please introduce yourself to the user."})
await task.queue_frames([LLMMessagesFrame(messages)]) await task.queue_frames([LLMMessagesFrame(messages)])
runner = PipelineRunner() runner = PipelineRunner()
await runner.run(task) await runner.run(task)
if __name__ == "__main__": if __name__ == "__main__":
(url, token) = configure() asyncio.run(main())
asyncio.run(main(url, token))

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

@@ -4,6 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License # SPDX-License-Identifier: BSD 2-Clause License
# #
import aiohttp
import asyncio import asyncio
import os import os
import sys import sys
@@ -30,63 +31,65 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG") logger.add(sys.stderr, level="DEBUG")
async def main(room_url: str, token): async def main():
transport = DailyTransport( async with aiohttp.ClientSession() as session:
room_url, (room_url, token) = await configure(session)
token,
"Respond bot", transport = DailyTransport(
DailyParams( room_url,
audio_out_enabled=True, token,
audio_out_sample_rate=24000, "Respond bot",
transcription_enabled=True, DailyParams(
vad_enabled=True, audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer() audio_out_sample_rate=24000,
transcription_enabled=True,
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer()
)
) )
)
tts = OpenAITTSService( tts = OpenAITTSService(
api_key=os.getenv("OPENAI_API_KEY"), api_key=os.getenv("OPENAI_API_KEY"),
voice="alloy" voice="alloy"
) )
llm = OpenAILLMService( llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"), api_key=os.getenv("OPENAI_API_KEY"),
model="gpt-4o") model="gpt-4o")
messages = [ messages = [
{ {
"role": "system", "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 characters in your answers. Respond to what the user said in a creative and helpful way.", "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.",
}, },
] ]
tma_in = LLMUserResponseAggregator(messages) tma_in = LLMUserResponseAggregator(messages)
tma_out = LLMAssistantResponseAggregator(messages) tma_out = LLMAssistantResponseAggregator(messages)
pipeline = Pipeline([ pipeline = Pipeline([
transport.input(), # Transport user input transport.input(), # Transport user input
tma_in, # User responses tma_in, # User responses
llm, # LLM llm, # LLM
tts, # TTS tts, # TTS
transport.output(), # Transport bot output transport.output(), # Transport bot output
tma_out # Assistant spoken responses tma_out # Assistant spoken responses
]) ])
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
@transport.event_handler("on_first_participant_joined") @transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant): async def on_first_participant_joined(transport, participant):
transport.capture_participant_transcription(participant["id"]) transport.capture_participant_transcription(participant["id"])
# Kick off the conversation. # Kick off the conversation.
messages.append( messages.append(
{"role": "system", "content": "Please introduce yourself to the user."}) {"role": "system", "content": "Please introduce yourself to the user."})
await task.queue_frames([LLMMessagesFrame(messages)]) await task.queue_frames([LLMMessagesFrame(messages)])
runner = PipelineRunner() runner = PipelineRunner()
await runner.run(task) await runner.run(task)
if __name__ == "__main__": if __name__ == "__main__":
(url, token) = configure() asyncio.run(main())
asyncio.run(main(url, token))

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

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

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

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

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

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

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

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

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

@@ -4,6 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License # SPDX-License-Identifier: BSD 2-Clause License
# #
import aiohttp
import asyncio import asyncio
import sys import sys
@@ -23,32 +24,34 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG") logger.add(sys.stderr, level="DEBUG")
async def main(room_url, token): async def main():
transport = DailyTransport( async with aiohttp.ClientSession() as session:
room_url, token, "Test", (room_url, token) = await configure(session)
DailyParams(
audio_in_enabled=True, transport = DailyTransport(
audio_out_enabled=True, room_url, token, "Test",
camera_out_enabled=True, DailyParams(
camera_out_is_live=True, audio_in_enabled=True,
camera_out_width=1280, audio_out_enabled=True,
camera_out_height=720 camera_out_enabled=True,
camera_out_is_live=True,
camera_out_width=1280,
camera_out_height=720
)
) )
)
@transport.event_handler("on_first_participant_joined") @transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant): async def on_first_participant_joined(transport, participant):
transport.capture_participant_video(participant["id"]) transport.capture_participant_video(participant["id"])
pipeline = Pipeline([transport.input(), transport.output()]) pipeline = Pipeline([transport.input(), transport.output()])
runner = PipelineRunner() runner = PipelineRunner()
task = PipelineTask(pipeline) task = PipelineTask(pipeline)
await runner.run(task) await runner.run(task)
if __name__ == "__main__": if __name__ == "__main__":
(url, token) = configure() asyncio.run(main())
asyncio.run(main(url, token))

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

@@ -4,6 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License # SPDX-License-Identifier: BSD 2-Clause License
# #
import aiohttp
import asyncio import asyncio
import sys import sys
@@ -27,40 +28,44 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG") logger.add(sys.stderr, level="DEBUG")
async def main(room_url, token): async def main():
tk_root = tk.Tk() async with aiohttp.ClientSession() as session:
tk_root.title("Local Mirror") (room_url, token) = await configure(session)
daily_transport = DailyTransport(room_url, token, "Test", DailyParams(audio_in_enabled=True)) tk_root = tk.Tk()
tk_root.title("Local Mirror")
tk_transport = TkLocalTransport( daily_transport = DailyTransport(
tk_root, room_url, token, "Test", DailyParams(
TransportParams( audio_in_enabled=True))
audio_out_enabled=True,
camera_out_enabled=True,
camera_out_is_live=True,
camera_out_width=1280,
camera_out_height=720))
@daily_transport.event_handler("on_first_participant_joined") tk_transport = TkLocalTransport(
async def on_first_participant_joined(transport, participant): tk_root,
transport.capture_participant_video(participant["id"]) TransportParams(
audio_out_enabled=True,
camera_out_enabled=True,
camera_out_is_live=True,
camera_out_width=1280,
camera_out_height=720))
pipeline = Pipeline([daily_transport.input(), tk_transport.output()]) @daily_transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
transport.capture_participant_video(participant["id"])
task = PipelineTask(pipeline) pipeline = Pipeline([daily_transport.input(), tk_transport.output()])
async def run_tk(): task = PipelineTask(pipeline)
while not task.has_finished():
tk_root.update()
tk_root.update_idletasks()
await asyncio.sleep(0.1)
runner = PipelineRunner() async def run_tk():
while not task.has_finished():
tk_root.update()
tk_root.update_idletasks()
await asyncio.sleep(0.1)
await asyncio.gather(runner.run(task), run_tk()) runner = PipelineRunner()
await asyncio.gather(runner.run(task), run_tk())
if __name__ == "__main__": if __name__ == "__main__":
(url, token) = configure() asyncio.run(main())
asyncio.run(main(url, token))

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

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

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

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

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

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

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

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

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

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

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

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

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

@@ -4,6 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License # SPDX-License-Identifier: BSD 2-Clause License
# #
import aiohttp
import asyncio import asyncio
import sys import sys
@@ -35,23 +36,25 @@ class TranscriptionLogger(FrameProcessor):
print(f"Transcription: {frame.text}") print(f"Transcription: {frame.text}")
async def main(room_url: str): async def main():
transport = DailyTransport(room_url, None, "Transcription bot", async with aiohttp.ClientSession() as session:
DailyParams(audio_in_enabled=True)) (room_url, _) = await configure(session)
stt = WhisperSTTService() transport = DailyTransport(room_url, None, "Transcription bot",
DailyParams(audio_in_enabled=True))
tl = TranscriptionLogger() stt = WhisperSTTService()
pipeline = Pipeline([transport.input(), stt, tl]) tl = TranscriptionLogger()
task = PipelineTask(pipeline) pipeline = Pipeline([transport.input(), stt, tl])
runner = PipelineRunner() task = PipelineTask(pipeline)
await runner.run(task) runner = PipelineRunner()
await runner.run(task)
if __name__ == "__main__": if __name__ == "__main__":
(url, token) = configure() asyncio.run(main())
asyncio.run(main(url))

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

@@ -4,6 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License # SPDX-License-Identifier: BSD 2-Clause License
# #
import aiohttp
import asyncio import asyncio
import os import os
import sys import sys
@@ -36,23 +37,25 @@ class TranscriptionLogger(FrameProcessor):
print(f"Transcription: {frame.text}") print(f"Transcription: {frame.text}")
async def main(room_url: str): async def main():
transport = DailyTransport(room_url, None, "Transcription bot", async with aiohttp.ClientSession() as session:
DailyParams(audio_in_enabled=True)) (room_url, _) = await configure(session)
stt = DeepgramSTTService(os.getenv("DEEPGRAM_API_KEY")) transport = DailyTransport(room_url, None, "Transcription bot",
DailyParams(audio_in_enabled=True))
tl = TranscriptionLogger() stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
pipeline = Pipeline([transport.input(), stt, tl]) tl = TranscriptionLogger()
task = PipelineTask(pipeline) pipeline = Pipeline([transport.input(), stt, tl])
runner = PipelineRunner() task = PipelineTask(pipeline)
await runner.run(task) runner = PipelineRunner()
await runner.run(task)
if __name__ == "__main__": if __name__ == "__main__":
(url, token) = configure() asyncio.run(main())
asyncio.run(main(url))

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

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

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

@@ -4,8 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License # SPDX-License-Identifier: BSD 2-Clause License
# #
import asyncio
import aiohttp import aiohttp
import asyncio
import os import os
import sys import sys
@@ -58,8 +58,10 @@ async def barbershop_man_filter(frame) -> bool:
return current_voice == "Barbershop Man" return current_voice == "Barbershop Man"
async def main(room_url: str, token): async def main():
async with aiohttp.ClientSession() as session: async with aiohttp.ClientSession() as session:
(room_url, token) = await configure(session)
transport = DailyTransport( transport = DailyTransport(
room_url, room_url,
token, token,
@@ -151,5 +153,4 @@ async def main(room_url: str, token):
if __name__ == "__main__": if __name__ == "__main__":
(url, token) = configure() asyncio.run(main())
asyncio.run(main(url, token))

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

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

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

@@ -8,7 +8,6 @@ import asyncio
import aiohttp import aiohttp
import os import os
import sys import sys
import json
from pipecat.frames.frames import LLMMessagesFrame from pipecat.frames.frames import LLMMessagesFrame
from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.pipeline import Pipeline
@@ -32,8 +31,10 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG") logger.add(sys.stderr, level="DEBUG")
async def main(room_url: str, token): async def main():
async with aiohttp.ClientSession() as session: async with aiohttp.ClientSession() as session:
(room_url, token) = await configure(session)
transport = DailyTransport( transport = DailyTransport(
room_url, room_url,
token, token,
@@ -126,5 +127,4 @@ async def main(room_url: str, token):
if __name__ == "__main__": if __name__ == "__main__":
(url, token) = configure() asyncio.run(main())
asyncio.run(main(url, token))

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

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

39
examples/foundational/runner.py
View File

@@ -1,11 +1,17 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import aiohttp
import argparse import argparse
import os import os
import time
import urllib from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
import requests
def configure(): async def configure(aiohttp_session: aiohttp.ClientSession):
parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample") parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
parser.add_argument( parser.add_argument(
"-u", "-u",
@@ -33,26 +39,15 @@ def configure():
if not key: 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.") 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(
daily_api_key=key,
daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
aiohttp_session=aiohttp_session)
# Create a meeting token for the given room with an expiration 1 hour in # Create a meeting token for the given room with an expiration 1 hour in
# the future. # the future.
room_name: str = urllib.parse.urlparse(url).path[1:] expiry_time: float = 60 * 60
expiration: float = time.time() + 60 * 60
res: requests.Response = requests.post( token = await daily_rest_helper.get_token(url, expiry_time)
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"]
return (url, token) return (url, token)

13
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 asyncio
import aiohttp import aiohttp
import os import os
@@ -128,8 +134,10 @@ class ImageFilterProcessor(FrameProcessor):
await self.push_frame(frame) await self.push_frame(frame)
async def main(room_url: str, token): async def main():
async with aiohttp.ClientSession() as session: async with aiohttp.ClientSession() as session:
(room_url, token) = await configure(session)
transport = DailyTransport( transport = DailyTransport(
room_url, room_url,
token, token,
@@ -204,5 +212,4 @@ async def main(room_url: str, token):
if __name__ == "__main__": if __name__ == "__main__":
(url, token) = configure() asyncio.run(main())
asyncio.run(main(url, token))

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

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

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

@@ -1,11 +1,17 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import aiohttp
import argparse import argparse
import os import os
import time
import urllib from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
import requests
def configure(): async def configure(aiohttp_session: aiohttp.ClientSession):
parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample") parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
parser.add_argument( parser.add_argument(
"-u", "-u",
@@ -33,26 +39,16 @@ def configure():
if not key: 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.") 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.")
# Create a meeting token for the given room with an expiration 1 hour in daily_rest_helper = DailyRESTHelper(
# the future. daily_api_key=key,
room_name: str = urllib.parse.urlparse(url).path[1:] daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
expiration: float = time.time() + 60 * 60 aiohttp_session=aiohttp_session
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: # Create a meeting token for the given room with an expiration 1 hour in
raise Exception( # the future.
f"Failed to create meeting token: {res.status_code} {res.text}") expiry_time: float = 60 * 60
token: str = res.json()["token"] token = await daily_rest_helper.get_token(url, expiry_time)
return (url, token) return (url, token)

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

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

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

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

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

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

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

@@ -1,11 +1,17 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import aiohttp
import argparse import argparse
import os import os
import time
import urllib from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
import requests
def configure(): async def configure(aiohttp_session: aiohttp.ClientSession):
parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample") parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
parser.add_argument( parser.add_argument(
"-u", "-u",
@@ -33,26 +39,15 @@ def configure():
if not key: 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.") 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(
daily_api_key=key,
daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
aiohttp_session=aiohttp_session)
# Create a meeting token for the given room with an expiration 1 hour in # Create a meeting token for the given room with an expiration 1 hour in
# the future. # the future.
room_name: str = urllib.parse.urlparse(url).path[1:] expiry_time: float = 60 * 60
expiration: float = time.time() + 60 * 60
res: requests.Response = requests.post( token = await daily_rest_helper.get_token(url, expiry_time)
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"]
return (url, token) return (url, token)

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

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

13
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 asyncio
import aiohttp import aiohttp
import os import os
@@ -77,8 +83,10 @@ class TalkingAnimation(FrameProcessor):
await self.push_frame(frame) await self.push_frame(frame)
async def main(room_url: str, token): async def main():
async with aiohttp.ClientSession() as session: async with aiohttp.ClientSession() as session:
(room_url, token) = await configure(session)
transport = DailyTransport( transport = DailyTransport(
room_url, room_url,
token, token,
@@ -165,5 +173,4 @@ async def main(room_url: str, token):
if __name__ == "__main__": if __name__ == "__main__":
(url, token) = configure() asyncio.run(main())
asyncio.run(main(url, token))

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

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

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

@@ -1,11 +1,17 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import aiohttp
import argparse import argparse
import os import os
import time
import urllib from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
import requests
def configure(): async def configure(aiohttp_session: aiohttp.ClientSession):
parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample") parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
parser.add_argument( parser.add_argument(
"-u", "-u",
@@ -33,26 +39,16 @@ def configure():
if not key: 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.") 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.")
# Create a meeting token for the given room with an expiration 1 hour in daily_rest_helper = DailyRESTHelper(
# the future. daily_api_key=key,
room_name: str = urllib.parse.urlparse(url).path[1:] daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
expiration: float = time.time() + 60 * 60 aiohttp_session=aiohttp_session
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: # Create a meeting token for the given room with an expiration 1 hour in
raise Exception( # the future.
f"Failed to create meeting token: {res.status_code} {res.text}") expiry_time: float = 60 * 60
token: str = res.json()["token"] token = await daily_rest_helper.get_token(url, expiry_time)
return (url, token) return (url, token)

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

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

6240
examples/storytelling-chatbot/frontend/package-lock.json generated Normal file
View File

File diff suppressed because it is too large Load Diff

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 async_timeout
fastapi fastapi
uvicorn uvicorn
requests
python-dotenv python-dotenv
pipecat-ai[daily,openai,fal] pipecat-ai[daily,openai,fal]

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

@@ -1,16 +1,26 @@
import os #
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import aiohttp
import argparse import argparse
import subprocess import subprocess
import requests import os
from pathlib import Path from pathlib import Path
from typing import Optional from typing import Optional
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, HTTPException from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse, JSONResponse 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 from dotenv import load_dotenv
@@ -20,12 +30,21 @@ load_dotenv(override=True)
MAX_SESSION_TIME = 5 * 60 # 5 minutes MAX_SESSION_TIME = 5 * 60 # 5 minutes
daily_rest_helper = DailyRESTHelper( daily_helpers = {}
os.getenv("DAILY_API_KEY", ""),
os.getenv("DAILY_API_URL", 'https://api.daily.co/v1'))
app = FastAPI() @asynccontextmanager
async def lifespan(app: FastAPI):
aiohttp_session = aiohttp.ClientSession()
daily_helpers["rest"] = DailyRESTHelper(
daily_api_key=os.getenv("DAILY_API_KEY", ""),
daily_api_url=os.getenv("DAILY_API_URL", 'https://api.daily.co/v1'),
aiohttp_session=aiohttp_session
)
yield
await aiohttp_session.close()
app = FastAPI(lifespan=lifespan)
app.add_middleware( app.add_middleware(
CORSMiddleware, CORSMiddleware,
@@ -70,7 +89,7 @@ async def start_bot(request: Request) -> JSONResponse:
properties=DailyRoomProperties() properties=DailyRoomProperties()
) )
try: try:
room: DailyRoomObject = daily_rest_helper.create_room(params=params) room: DailyRoomObject = await daily_helpers["rest"].create_room(params=params)
except Exception as e: except Exception as e:
raise HTTPException( raise HTTPException(
status_code=500, status_code=500,
@@ -78,13 +97,13 @@ async def start_bot(request: Request) -> JSONResponse:
else: else:
# Check passed room URL exists, we should assume that it already has a sip set up # Check passed room URL exists, we should assume that it already has a sip set up
try: try:
room: DailyRoomObject = daily_rest_helper.get_room_from_url(room_url) room: DailyRoomObject = await daily_helpers["rest"].get_room_from_url(room_url)
except Exception: except Exception:
raise HTTPException( raise HTTPException(
status_code=500, detail=f"Room not found: {room_url}") status_code=500, detail=f"Room not found: {room_url}")
# Give the agent a token to join the session # Give the agent a token to join the session
token = daily_rest_helper.get_token(room.url, MAX_SESSION_TIME) token = await daily_helpers["rest"].get_token(room.url, MAX_SESSION_TIME)
if not room or not token: if not room or not token:
raise HTTPException( raise HTTPException(
@@ -93,7 +112,7 @@ async def start_bot(request: Request) -> JSONResponse:
# Launch a new VM, or run as a shell process (not recommended) # Launch a new VM, or run as a shell process (not recommended)
if os.getenv("RUN_AS_VM", False): if os.getenv("RUN_AS_VM", False):
try: try:
virtualize_bot(room.url, token) await virtualize_bot(room.url, token)
except Exception as e: except Exception as e:
raise HTTPException( raise HTTPException(
status_code=500, detail=f"Failed to spawn VM: {e}") status_code=500, detail=f"Failed to spawn VM: {e}")
@@ -109,7 +128,7 @@ async def start_bot(request: Request) -> JSONResponse:
status_code=500, detail=f"Failed to start subprocess: {e}") status_code=500, detail=f"Failed to start subprocess: {e}")
# Grab a token for the user to join with # 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_helpers["rest"].get_token(room.url, MAX_SESSION_TIME)
return JSONResponse({ return JSONResponse({
"room_url": room.url, "room_url": room.url,
@@ -136,7 +155,7 @@ async def catch_all(path_name: Optional[str] = ""):
# ------------ Virtualization ------------ # # ------------ 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 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. You can adapt this method to use whichever cloud provider you prefer.
@@ -149,54 +168,53 @@ def virtualize_bot(room_url: str, token: str):
'Content-Type': 'application/json' 'Content-Type': 'application/json'
} }
# Use the same image as the bot runner async with aiohttp.ClientSession() as session:
res = requests.get(f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines", headers=FLY_HEADERS) # Use the same image as the bot runner
if res.status_code != 200: async with session.get(f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines", headers=FLY_HEADERS) as r:
raise Exception(f"Unable to get machine info from Fly: {res.text}") if r.status != 200:
image = res.json()[0]['config']['image'] text = await r.text()
raise Exception(f"Unable to get machine info from Fly: {text}")
# Machine configuration data = await r.json()
cmd = f"python3 src/bot.py -u {room_url} -t {token}" image = data[0]['config']['image']
cmd = cmd.split()
worker_props = { # Machine configuration
"config": { cmd = f"python3 src/bot.py -u {room_url} -t {token}"
"image": image, cmd = cmd.split()
"auto_destroy": True, worker_props = {
"init": { "config": {
"cmd": cmd "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 data = await r.json()
res = requests.post( # Wait for the machine to enter the started state
f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines", vm_id = data['id']
headers=FLY_HEADERS,
json=worker_props)
if res.status_code != 200: async with session.get(f"{FLY_API_HOST}/apps/{FLY_APP_NAME}/machines/{vm_id}/wait?state=started", headers=FLY_HEADERS) as r:
raise Exception(f"Problem starting a bot worker: {res.text}") 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 print(f"Machine joined room: {room_url}")
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}")
# ------------ Main ------------ # # ------------ 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

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

@@ -1,5 +1,11 @@
import asyncio #
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import aiohttp import aiohttp
import asyncio
import os import os
import sys import sys
@@ -12,7 +18,11 @@ from pipecat.processors.aggregators.sentence import SentenceAggregator
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.azure import AzureTTSService from pipecat.services.azure import AzureTTSService
from pipecat.services.openai import OpenAILLMService 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 from runner import configure
@@ -79,8 +89,10 @@ class TranslationSubtitles(FrameProcessor):
await self.push_frame(frame) await self.push_frame(frame)
async def main(room_url: str, token): async def main():
async with aiohttp.ClientSession() as session: async with aiohttp.ClientSession() as session:
(room_url, token) = await configure(session)
transport = DailyTransport( transport = DailyTransport(
room_url, room_url,
token, token,
@@ -133,5 +145,4 @@ async def main(room_url: str, token):
if __name__ == "__main__": if __name__ == "__main__":
(url, token) = configure() asyncio.run(main())
asyncio.run(main(url, token))

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

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

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

@@ -1,11 +1,18 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import argparse import argparse
import os import os
import time
import urllib import aiohttp
import requests
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
def configure(): async def configure(aiohttp_session: aiohttp.ClientSession):
parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample") parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
parser.add_argument( parser.add_argument(
"-u", "-u",
@@ -33,26 +40,16 @@ def configure():
if not key: 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.") 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.")
# Create a meeting token for the given room with an expiration 1 hour in daily_rest_helper = DailyRESTHelper(
# the future. daily_api_key=key,
room_name: str = urllib.parse.urlparse(url).path[1:] daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
expiration: float = time.time() + 60 * 60 aiohttp_session=aiohttp_session
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: # Create a meeting token for the given room with an expiration 1 hour in
raise Exception( # the future.
f"Failed to create meeting token: {res.status_code} {res.text}") expiry_time: float = 60 * 60
token: str = res.json()["token"] token = await daily_rest_helper.get_token(url, expiry_time)
return (url, token) return (url, token)

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

@@ -1,31 +1,51 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import aiohttp
import os import os
import argparse import argparse
import subprocess import subprocess
import atexit
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, HTTPException from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, RedirectResponse 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 MAX_BOTS_PER_ROOM = 1
# Bot sub-process dict for status reporting and concurrency control # Bot sub-process dict for status reporting and concurrency control
bot_procs = {} bot_procs = {}
daily_helpers = {}
def cleanup(): def cleanup():
# Clean up function, just to be extra safe # 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.terminate()
proc.wait() proc.wait()
atexit.register(cleanup) @asynccontextmanager
async def lifespan(app: FastAPI):
aiohttp_session = aiohttp.ClientSession()
daily_helpers["rest"] = DailyRESTHelper(
daily_api_key=os.getenv("DAILY_API_KEY", ""),
daily_api_url=os.getenv("DAILY_API_URL", 'https://api.daily.co/v1'),
aiohttp_session=aiohttp_session
)
yield
await aiohttp_session.close()
cleanup()
app = FastAPI(lifespan=lifespan)
app = FastAPI()
app.add_middleware( app.add_middleware(
CORSMiddleware, CORSMiddleware,
@@ -39,45 +59,45 @@ app.add_middleware(
@app.get("/start") @app.get("/start")
async def start_agent(request: Request): async def start_agent(request: Request):
print(f"!!! Creating room") print(f"!!! Creating room")
room_url, room_name = _create_room() room = await daily_helpers["rest"].create_room(DailyRoomParams())
print(f"!!! Room URL: {room_url}") print(f"!!! Room URL: {room.url}")
# Ensure the room property is present # Ensure the room property is present
if not room_url: if not room.url:
raise HTTPException( raise HTTPException(
status_code=500, status_code=500,
detail="Missing 'room' property in request data. Cannot start agent without a target room!") 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 # Check if there is already an existing process running in this room
num_bots_in_room = sum( 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: if num_bots_in_room >= MAX_BOTS_PER_ROOM:
raise HTTPException( 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 # Get the token for the room
token = get_token(room_url) token = await daily_helpers["rest"].get_token(room.url)
if not token: if not token:
raise HTTPException( 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 # Spawn a new agent, and join the user session
# Note: this is mostly for demonstration purposes (refer to 'deployment' in README) # Note: this is mostly for demonstration purposes (refer to 'deployment' in README)
try: try:
proc = subprocess.Popen( proc = subprocess.Popen(
[ [
f"python3 -m bot -u {room_url} -t {token}" f"python3 -m bot -u {room.url} -t {token}"
], ],
shell=True, shell=True,
bufsize=1, bufsize=1,
cwd=os.path.dirname(os.path.abspath(__file__)) 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: except Exception as e:
raise HTTPException( raise HTTPException(
status_code=500, detail=f"Failed to start subprocess: {e}") status_code=500, detail=f"Failed to start subprocess: {e}")
return RedirectResponse(room_url) return RedirectResponse(room.url)
@app.get("/status/{pid}") @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 ## 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`. 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`. Copy the ngrok URL and update templates/streams.xml with `wss://<ngrok_url>/ws`.
## Running the Application ## Running the Application
@@ -61,11 +67,6 @@ This project is a FastAPI-based chatbot that integrates with Twilio to handle We
python server.py python server.py
``` ```
2. **Start ngrok**:
In a new terminal, start ngrok to tunnel the local server:
```sh
ngrok http 8765
```
### Using Docker ### Using Docker
1. **Build the Docker image**: 1. **Build the Docker image**:

9
src/pipecat/services/ai_services.py
View File

@@ -75,10 +75,11 @@ class AIService(FrameProcessor):
async def process_generator(self, generator: AsyncGenerator[Frame, None]): async def process_generator(self, generator: AsyncGenerator[Frame, None]):
async for f in generator: async for f in generator:
if isinstance(f, ErrorFrame): if f:
await self.push_error(f) if isinstance(f, ErrorFrame):
else: await self.push_error(f)
await self.push_frame(f) else:
await self.push_frame(f)
class AsyncAIService(AsyncFrameProcessor): class AsyncAIService(AsyncFrameProcessor):

10
src/pipecat/services/azure.py
View File

@@ -175,8 +175,8 @@ class AzureImageGenServiceREST(ImageGenService):
api_key: str, api_key: str,
endpoint: str, endpoint: str,
model: str, model: str,
aiohttp_session: aiohttp.ClientSession,
api_version="2023-06-01-preview", api_version="2023-06-01-preview",
aiohttp_session: aiohttp.ClientSession | None = None,
): ):
super().__init__() super().__init__()
@@ -185,13 +185,7 @@ class AzureImageGenServiceREST(ImageGenService):
self._api_version = api_version self._api_version = api_version
self._model = model self._model = model
self._image_size = image_size self._image_size = image_size
self._aiohttp_session = aiohttp_session or aiohttp.ClientSession() self._aiohttp_session = aiohttp_session
self._close_aiohttp_session = aiohttp_session is None
async def cleanup(self):
await super().cleanup()
if self._close_aiohttp_session:
await self._aiohttp_session.close()
async def run_image_gen(self, prompt: str) -> AsyncGenerator[Frame, None]: async def run_image_gen(self, prompt: str) -> AsyncGenerator[Frame, None]:
url = f"{self._azure_endpoint}openai/images/generations:submit?api-version={self._api_version}" url = f"{self._azure_endpoint}openai/images/generations:submit?api-version={self._api_version}"

10
src/pipecat/services/deepgram.py
View File

@@ -46,11 +46,11 @@ class DeepgramTTSService(TTSService):
self, self,
*, *,
api_key: str, api_key: str,
aiohttp_session: aiohttp.ClientSession,
voice: str = "aura-helios-en", voice: str = "aura-helios-en",
base_url: str = "https://api.deepgram.com/v1/speak", base_url: str = "https://api.deepgram.com/v1/speak",
sample_rate: int = 16000, sample_rate: int = 16000,
encoding: str = "linear16", encoding: str = "linear16",
aiohttp_session: aiohttp.ClientSession | None = None,
**kwargs): **kwargs):
super().__init__(**kwargs) super().__init__(**kwargs)
@@ -59,17 +59,11 @@ class DeepgramTTSService(TTSService):
self._base_url = base_url self._base_url = base_url
self._sample_rate = sample_rate self._sample_rate = sample_rate
self._encoding = encoding self._encoding = encoding
self._aiohttp_session = aiohttp_session or aiohttp.ClientSession() self._aiohttp_session = aiohttp_session
self._close_aiohttp_session = aiohttp_session is None
def can_generate_metrics(self) -> bool: def can_generate_metrics(self) -> bool:
return True return True
async def cleanup(self):
await super().cleanup()
if self._close_aiohttp_session:
await self._aiohttp_session.close()
async def set_voice(self, voice: str): async def set_voice(self, voice: str):
logger.debug(f"Switching TTS voice to: [{voice}]") logger.debug(f"Switching TTS voice to: [{voice}]")
self._voice = voice self._voice = voice

10
src/pipecat/services/elevenlabs.py
View File

@@ -21,25 +21,19 @@ class ElevenLabsTTSService(TTSService):
*, *,
api_key: str, api_key: str,
voice_id: str, voice_id: str,
aiohttp_session: aiohttp.ClientSession,
model: str = "eleven_turbo_v2", model: str = "eleven_turbo_v2",
aiohttp_session: aiohttp.ClientSession | None = None,
**kwargs): **kwargs):
super().__init__(**kwargs) super().__init__(**kwargs)
self._api_key = api_key self._api_key = api_key
self._voice_id = voice_id self._voice_id = voice_id
self._model = model self._model = model
self._aiohttp_session = aiohttp_session or aiohttp.ClientSession() self._aiohttp_session = aiohttp_session
self._close_aiohttp_session = aiohttp_session is None
def can_generate_metrics(self) -> bool: def can_generate_metrics(self) -> bool:
return True return True
async def cleanup(self):
await super().cleanup()
if self._close_aiohttp_session:
await self._aiohttp_session.close()
async def set_voice(self, voice: str): async def set_voice(self, voice: str):
logger.debug(f"Switching TTS voice to: [{voice}]") logger.debug(f"Switching TTS voice to: [{voice}]")
self._voice_id = voice self._voice_id = voice

10
src/pipecat/services/fal.py
View File

@@ -40,23 +40,17 @@ class FalImageGenService(ImageGenService):
self, self,
*, *,
params: InputParams, params: InputParams,
aiohttp_session: aiohttp.ClientSession,
model: str = "fal-ai/fast-sdxl", model: str = "fal-ai/fast-sdxl",
key: str | None = None, key: str | None = None,
aiohttp_session: aiohttp.ClientSession | None = None,
): ):
super().__init__() super().__init__()
self._model = model self._model = model
self._params = params self._params = params
self._aiohttp_session = aiohttp_session or aiohttp.ClientSession() self._aiohttp_session = aiohttp_session
self._close_aiohttp_session = aiohttp_session is None
if key: if key:
os.environ["FAL_KEY"] = key os.environ["FAL_KEY"] = key
async def cleanup(self):
await super().cleanup()
if self._close_aiohttp_session:
await self._aiohttp_session.close()
async def run_image_gen(self, prompt: str) -> AsyncGenerator[Frame, None]: async def run_image_gen(self, prompt: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"Generating image from prompt: {prompt}") logger.debug(f"Generating image from prompt: {prompt}")

12
src/pipecat/services/openai.py
View File

@@ -254,21 +254,15 @@ class OpenAIImageGenService(ImageGenService):
self, self,
*, *,
api_key: str, api_key: str,
model: str = "dall-e-3", aiohttp_session: aiohttp.ClientSession,
image_size: Literal["256x256", "512x512", "1024x1024", "1792x1024", "1024x1792"], image_size: Literal["256x256", "512x512", "1024x1024", "1792x1024", "1024x1792"],
aiohttp_session: aiohttp.ClientSession | None = None, model: str = "dall-e-3",
): ):
super().__init__() super().__init__()
self._model = model self._model = model
self._image_size = image_size self._image_size = image_size
self._client = AsyncOpenAI(api_key=api_key) self._client = AsyncOpenAI(api_key=api_key)
self._aiohttp_session = aiohttp_session or aiohttp.ClientSession() self._aiohttp_session = aiohttp_session
self._close_aiohttp_session = aiohttp_session is None
async def cleanup(self):
await super().cleanup()
if self._close_aiohttp_session:
await self._aiohttp_session.close()
async def run_image_gen(self, prompt: str) -> AsyncGenerator[Frame, None]: async def run_image_gen(self, prompt: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"Generating image from prompt: {prompt}") logger.debug(f"Generating image from prompt: {prompt}")

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

@@ -6,15 +6,13 @@
import aiohttp 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 pipecat.services.ai_services import TTSService
from loguru import logger from loguru import logger
import requests
import numpy as np import numpy as np
try: try:
@@ -41,24 +39,30 @@ class XTTSService(TTSService):
voice_id: str, voice_id: str,
language: str, language: str,
base_url: str, base_url: str,
aiohttp_session: aiohttp.ClientSession | None = None, aiohttp_session: aiohttp.ClientSession,
**kwargs): **kwargs):
super().__init__(**kwargs) super().__init__(**kwargs)
self._voice_id = voice_id self._voice_id = voice_id
self._language = language self._language = language
self._base_url = base_url 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._aiohttp_session = aiohttp_session
self._close_aiohttp_session = aiohttp_session is None
def can_generate_metrics(self) -> bool: def can_generate_metrics(self) -> bool:
return True return True
async def cleanup(self): async def start(self, frame: StartFrame):
await super().cleanup() await super().start(frame)
if self._close_aiohttp_session: async with self._aiohttp_session.get(self._base_url + "/studio_speakers") as r:
await self._aiohttp_session.close() 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 set_voice(self, voice: str): async def set_voice(self, voice: str):
logger.debug(f"Switching TTS voice to: [{voice}]") logger.debug(f"Switching TTS voice to: [{voice}]")
@@ -66,6 +70,11 @@ class XTTSService(TTSService):
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"Generating TTS: [{text}]") 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] embeddings = self._studio_speakers[self._voice_id]
url = self._base_url + "/tts_stream" url = self._base_url + "/tts_stream"

2
src/pipecat/transports/network/websocket_server.py
View File

@@ -71,7 +71,7 @@ class WebsocketServerInputTransport(BaseInputTransport):
async def cancel(self, frame: CancelFrame): async def cancel(self, frame: CancelFrame):
await super().cancel(frame) await super().cancel(frame)
self._server_task.cancel() self._stop_server_event.set()
await self._server_task await self._server_task
async def _server_task_handler(self): async def _server_task_handler(self):

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 import time
from urllib.parse import urlparse from urllib.parse import urlparse
@@ -61,24 +61,27 @@ class DailyRoomObject(BaseModel):
class DailyRESTHelper: 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",
aiohttp_session: aiohttp.ClientSession):
self.daily_api_key = daily_api_key self.daily_api_key = daily_api_key
self.daily_api_url = daily_api_url self.daily_api_url = daily_api_url
self.aiohttp_session = aiohttp_session
def _get_name_from_url(self, room_url: str) -> str: def _get_name_from_url(self, room_url: str) -> str:
return urlparse(room_url).path[1:] return urlparse(room_url).path[1:]
def create_room(self, params: DailyRoomParams) -> DailyRoomObject: async def create_room(self, params: DailyRoomParams) -> DailyRoomObject:
res = requests.post( headers = {"Authorization": f"Bearer {self.daily_api_key}"}
f"{self.daily_api_url}/rooms", json = {**params.model_dump(exclude_none=True)}
headers={"Authorization": f"Bearer {self.daily_api_key}"}, async with self.aiohttp_session.post(f"{self.daily_api_url}/rooms", headers=headers, json=json) as r:
json={**params.model_dump(exclude_none=True)} if r.status != 200:
) text = await r.text()
raise Exception(f"Unable to create room: {text}")
if res.status_code != 200: data = await r.json()
raise Exception(f"Unable to create room: {res.text}")
data = res.json()
try: try:
room = DailyRoomObject(**data) room = DailyRoomObject(**data)
@@ -87,16 +90,13 @@ class DailyRESTHelper:
return room return room
def _get_room_from_name(self, room_name: str) -> DailyRoomObject: async def _get_room_from_name(self, room_name: str) -> DailyRoomObject:
res: requests.Response = requests.get( headers = {"Authorization": f"Bearer {self.daily_api_key}"}
f"{self.daily_api_url}/rooms/{room_name}", async with self.aiohttp_session.get(f"{self.daily_api_url}/rooms/{room_name}", headers=headers) as r:
headers={"Authorization": f"Bearer {self.daily_api_key}"} if r.status != 200:
) raise Exception(f"Room not found: {room_name}")
if res.status_code != 200: data = await r.json()
raise Exception(f"Room not found: {room_name}")
data = res.json()
try: try:
room = DailyRoomObject(**data) room = DailyRoomObject(**data)
@@ -105,11 +105,15 @@ class DailyRESTHelper:
return room 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) 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: if not room_url:
raise Exception( raise Exception(
"No Daily room specified. You must specify a Daily room in order a token to be generated.") "No Daily room specified. You must specify a Daily room in order a token to be generated.")
@@ -118,22 +122,19 @@ class DailyRESTHelper:
room_name = self._get_name_from_url(room_url) room_name = self._get_name_from_url(room_url)
res: requests.Response = requests.post( headers = {"Authorization": f"Bearer {self.daily_api_key}"}
f"{self.daily_api_url}/meeting-tokens", json = {
headers={ "properties": {
"Authorization": f"Bearer {self.daily_api_key}"}, "room_name": room_name,
json={ "is_owner": owner,
"properties": { "exp": expiration
"room_name": room_name, }
"is_owner": owner, }
"exp": expiration async with self.aiohttp_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: data = await r.json()
raise Exception(
f"Failed to create meeting token: {res.status_code} {res.text}")
token: str = res.json()["token"] return data["token"]
return token