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

ruff formatting

Browse Source
This commit is contained in:
Adrian Cowham
2024-10-11 11:35:02 -07:00
parent 083d221dd2
commit 79c8aa2c4a
8 changed files with 147 additions and 180 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
examples/canonical-metrics/bot.py
View File

@@ -19,9 +19,10 @@ from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_response import ( from pipecat.processors.aggregators.llm_response import (
LLMAssistantResponseAggregator, LLMUserResponseAggregator) LLMAssistantResponseAggregator,
from pipecat.processors.audio.audio_buffer_processor import \ LLMUserResponseAggregator,
AudioBufferProcessor )
from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor
from pipecat.services.canonical import CanonicalMetricsService from pipecat.services.canonical import CanonicalMetricsService
from pipecat.services.elevenlabs import ElevenLabsTTSService from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.services.openai import OpenAILLMService from pipecat.services.openai import OpenAILLMService
@@ -58,18 +59,16 @@ async def main():
# tier="nova", # tier="nova",
# model="2-general" # model="2-general"
# ) # )
) ),
) )
tts = ElevenLabsTTSService( tts = ElevenLabsTTSService(
api_key=os.getenv("ELEVENLABS_API_KEY"), api_key=os.getenv("ELEVENLABS_API_KEY"),
# #
# English # English
# #
voice_id="cgSgspJ2msm6clMCkdW9", voice_id="cgSgspJ2msm6clMCkdW9",
aiohttp_session=session, aiohttp_session=session,
# #
# Spanish # Spanish
# #
@@ -77,9 +76,7 @@ async def main():
# voice_id="gD1IexrzCvsXPHUuT0s3", # voice_id="gD1IexrzCvsXPHUuT0s3",
) )
llm = OpenAILLMService( llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
api_key=os.getenv("OPENAI_API_KEY"),
model="gpt-4o")
messages = [ messages = [
{ {
@@ -88,7 +85,6 @@ async def main():
# English # English
# #
"content": "You are Chatbot, a friendly, helpful robot. 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, but keep your responses brief. Start by introducing yourself. Keep all your responses to 12 words or fewer.", "content": "You are Chatbot, a friendly, helpful robot. 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, but keep your responses brief. Start by introducing yourself. Keep all your responses to 12 words or fewer.",
# #
# Spanish # Spanish
# #
@@ -114,16 +110,18 @@ async def main():
assistant="pipecat-chatbot", assistant="pipecat-chatbot",
assistant_speaks_first=True, assistant_speaks_first=True,
) )
pipeline = Pipeline([ pipeline = Pipeline(
transport.input(), # microphone [
user_response, transport.input(), # microphone
llm, user_response,
tts, llm,
transport.output(), tts,
audio_buffer_processor, # captures audio into a buffer transport.output(),
canonical, # uploads audio buffer to Canonical AI for metrics audio_buffer_processor, # captures audio into a buffer
assistant_response, canonical, # uploads audio buffer to Canonical AI for metrics
]) assistant_response,
]
)
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))

20
examples/canonical-metrics/runner.py
View File

@@ -4,21 +4,19 @@
# SPDX-License-Identifier: BSD 2-Clause License # SPDX-License-Identifier: BSD 2-Clause License
# #
import aiohttp
import argparse import argparse
import os import os
import aiohttp
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
async def configure(aiohttp_session: aiohttp.ClientSession): 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", "--url", type=str, required=False, help="URL of the Daily room to join"
"--url", )
type=str,
required=False,
help="URL of the Daily room to join")
parser.add_argument( parser.add_argument(
"-k", "-k",
"--apikey", "--apikey",
@@ -34,15 +32,18 @@ async def configure(aiohttp_session: aiohttp.ClientSession):
if not url: if not url:
raise Exception( raise Exception(
"No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL.") "No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL."
)
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_rest_helper = DailyRESTHelper(
daily_api_key=key, daily_api_key=key,
daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"), daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
aiohttp_session=aiohttp_session 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
@@ -52,3 +53,4 @@ async def configure(aiohttp_session: aiohttp.ClientSession):
token = await daily_rest_helper.get_token(url, expiry_time) token = await daily_rest_helper.get_token(url, expiry_time)
return (url, token) return (url, token)
return (url, token)

44
examples/canonical-metrics/server.py
View File

@@ -15,8 +15,7 @@ from fastapi import FastAPI, HTTPException, Request
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 pipecat.transports.services.helpers.daily_rest import (DailyRESTHelper, from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomParams
DailyRoomParams)
MAX_BOTS_PER_ROOM = 1 MAX_BOTS_PER_ROOM = 1
@@ -41,13 +40,14 @@ async def lifespan(app: FastAPI):
aiohttp_session = aiohttp.ClientSession() aiohttp_session = aiohttp.ClientSession()
daily_helpers["rest"] = DailyRESTHelper( daily_helpers["rest"] = DailyRESTHelper(
daily_api_key=os.getenv("DAILY_API_KEY", ""), daily_api_key=os.getenv("DAILY_API_KEY", ""),
daily_api_url=os.getenv("DAILY_API_URL", 'https://api.daily.co/v1'), daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
aiohttp_session=aiohttp_session aiohttp_session=aiohttp_session,
) )
yield yield
await aiohttp_session.close() await aiohttp_session.close()
cleanup() cleanup()
app = FastAPI(lifespan=lifespan) app = FastAPI(lifespan=lifespan)
app.add_middleware( app.add_middleware(
@@ -68,37 +68,34 @@ async def start_agent(request: Request):
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 = await daily_helpers["rest"].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)
@@ -110,8 +107,7 @@ def get_status(pid: int):
# If the subprocess doesn't exist, return an error # If the subprocess doesn't exist, return an error
if not proc: if not proc:
raise HTTPException( raise HTTPException(status_code=404, detail=f"Bot with process id: {pid} not found")
status_code=404, detail=f"Bot with process id: {pid} not found")
# Check the status of the subprocess # Check the status of the subprocess
if proc[0].poll() is None: if proc[0].poll() is None:
@@ -128,14 +124,10 @@ if __name__ == "__main__":
default_host = os.getenv("HOST", "0.0.0.0") default_host = os.getenv("HOST", "0.0.0.0")
default_port = int(os.getenv("FAST_API_PORT", "7860")) default_port = int(os.getenv("FAST_API_PORT", "7860"))
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(description="Daily Storyteller FastAPI server")
description="Daily Storyteller FastAPI server") parser.add_argument("--host", type=str, default=default_host, help="Host address")
parser.add_argument("--host", type=str, parser.add_argument("--port", type=int, default=default_port, help="Port number")
default=default_host, help="Host address") parser.add_argument("--reload", action="store_true", help="Reload code on change")
parser.add_argument("--port", type=int,
default=default_port, help="Port number")
parser.add_argument("--reload", action="store_true",
help="Reload code on change")
config = parser.parse_args() config = parser.parse_args()

36
examples/chatbot-audio-recording/bot.py
View File

@@ -18,9 +18,10 @@ from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_response import ( from pipecat.processors.aggregators.llm_response import (
LLMAssistantResponseAggregator, LLMUserResponseAggregator) LLMAssistantResponseAggregator,
from pipecat.processors.audio.audio_buffer_processor import \ LLMUserResponseAggregator,
AudioBufferProcessor )
from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor
from pipecat.services.elevenlabs import ElevenLabsTTSService from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.services.openai import OpenAILLMService from pipecat.services.openai import OpenAILLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -56,18 +57,16 @@ async def main():
# tier="nova", # tier="nova",
# model="2-general" # model="2-general"
# ) # )
) ),
) )
tts = ElevenLabsTTSService( tts = ElevenLabsTTSService(
api_key=os.getenv("ELEVENLABS_API_KEY"), api_key=os.getenv("ELEVENLABS_API_KEY"),
# #
# English # English
# #
voice_id="cgSgspJ2msm6clMCkdW9", voice_id="cgSgspJ2msm6clMCkdW9",
aiohttp_session=session, aiohttp_session=session,
# #
# Spanish # Spanish
# #
@@ -75,9 +74,7 @@ async def main():
# voice_id="gD1IexrzCvsXPHUuT0s3", # voice_id="gD1IexrzCvsXPHUuT0s3",
) )
llm = OpenAILLMService( llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
api_key=os.getenv("OPENAI_API_KEY"),
model="gpt-4o")
messages = [ messages = [
{ {
@@ -86,7 +83,6 @@ async def main():
# English # English
# #
"content": "You are Chatbot, a friendly, helpful robot. 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, but keep your responses brief. Start by introducing yourself. Keep all your response to 12 words or fewer.", "content": "You are Chatbot, a friendly, helpful robot. 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, but keep your responses brief. Start by introducing yourself. Keep all your response to 12 words or fewer.",
# #
# Spanish # Spanish
# #
@@ -98,15 +94,17 @@ async def main():
assistant_response = LLMAssistantResponseAggregator() assistant_response = LLMAssistantResponseAggregator()
audiobuffer = AudioBufferProcessor() audiobuffer = AudioBufferProcessor()
pipeline = Pipeline([ pipeline = Pipeline(
transport.input(), # microphone [
user_response, transport.input(), # microphone
llm, user_response,
tts, llm,
transport.output(), tts,
audiobuffer, # used to buffer the audio in the pipeline transport.output(),
assistant_response, audiobuffer, # used to buffer the audio in the pipeline
]) assistant_response,
]
)
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))

20
examples/chatbot-audio-recording/runner.py
View File

@@ -4,21 +4,19 @@
# SPDX-License-Identifier: BSD 2-Clause License # SPDX-License-Identifier: BSD 2-Clause License
# #
import aiohttp
import argparse import argparse
import os import os
import aiohttp
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
async def configure(aiohttp_session: aiohttp.ClientSession): 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", "--url", type=str, required=False, help="URL of the Daily room to join"
"--url", )
type=str,
required=False,
help="URL of the Daily room to join")
parser.add_argument( parser.add_argument(
"-k", "-k",
"--apikey", "--apikey",
@@ -34,15 +32,18 @@ async def configure(aiohttp_session: aiohttp.ClientSession):
if not url: if not url:
raise Exception( raise Exception(
"No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL.") "No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL."
)
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_rest_helper = DailyRESTHelper(
daily_api_key=key, daily_api_key=key,
daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"), daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
aiohttp_session=aiohttp_session 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
@@ -52,3 +53,4 @@ async def configure(aiohttp_session: aiohttp.ClientSession):
token = await daily_rest_helper.get_token(url, expiry_time) token = await daily_rest_helper.get_token(url, expiry_time)
return (url, token) return (url, token)
return (url, token)

44
examples/chatbot-audio-recording/server.py
View File

@@ -15,8 +15,7 @@ from fastapi import FastAPI, HTTPException, Request
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 pipecat.transports.services.helpers.daily_rest import (DailyRESTHelper, from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomParams
DailyRoomParams)
MAX_BOTS_PER_ROOM = 1 MAX_BOTS_PER_ROOM = 1
@@ -41,13 +40,14 @@ async def lifespan(app: FastAPI):
aiohttp_session = aiohttp.ClientSession() aiohttp_session = aiohttp.ClientSession()
daily_helpers["rest"] = DailyRESTHelper( daily_helpers["rest"] = DailyRESTHelper(
daily_api_key=os.getenv("DAILY_API_KEY", ""), daily_api_key=os.getenv("DAILY_API_KEY", ""),
daily_api_url=os.getenv("DAILY_API_URL", 'https://api.daily.co/v1'), daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
aiohttp_session=aiohttp_session aiohttp_session=aiohttp_session,
) )
yield yield
await aiohttp_session.close() await aiohttp_session.close()
cleanup() cleanup()
app = FastAPI(lifespan=lifespan) app = FastAPI(lifespan=lifespan)
app.add_middleware( app.add_middleware(
@@ -68,37 +68,34 @@ async def start_agent(request: Request):
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 = await daily_helpers["rest"].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)
@@ -110,8 +107,7 @@ def get_status(pid: int):
# If the subprocess doesn't exist, return an error # If the subprocess doesn't exist, return an error
if not proc: if not proc:
raise HTTPException( raise HTTPException(status_code=404, detail=f"Bot with process id: {pid} not found")
status_code=404, detail=f"Bot with process id: {pid} not found")
# Check the status of the subprocess # Check the status of the subprocess
if proc[0].poll() is None: if proc[0].poll() is None:
@@ -128,14 +124,10 @@ if __name__ == "__main__":
default_host = os.getenv("HOST", "0.0.0.0") default_host = os.getenv("HOST", "0.0.0.0")
default_port = int(os.getenv("FAST_API_PORT", "7860")) default_port = int(os.getenv("FAST_API_PORT", "7860"))
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(description="Daily Storyteller FastAPI server")
description="Daily Storyteller FastAPI server") parser.add_argument("--host", type=str, default=default_host, help="Host address")
parser.add_argument("--host", type=str, parser.add_argument("--port", type=int, default=default_port, help="Port number")
default=default_host, help="Host address") parser.add_argument("--reload", action="store_true", help="Reload code on change")
parser.add_argument("--port", type=int,
default=default_port, help="Port number")
parser.add_argument("--reload", action="store_true",
help="Reload code on change")
config = parser.parse_args() config = parser.parse_args()

40
src/pipecat/processors/audio/audio_buffer_processor.py
View File

@@ -1,19 +1,23 @@
import wave import wave
from io import BytesIO from io import BytesIO
from pipecat.frames.frames import (AudioRawFrame, BotInterruptionFrame, from pipecat.frames.frames import (
BotStartedSpeakingFrame, AudioRawFrame,
BotStoppedSpeakingFrame, Frame, BotInterruptionFrame,
InputAudioRawFrame, OutputAudioRawFrame, BotStartedSpeakingFrame,
StartInterruptionFrame, BotStoppedSpeakingFrame,
StopInterruptionFrame, Frame,
UserStartedSpeakingFrame, InputAudioRawFrame,
UserStoppedSpeakingFrame) OutputAudioRawFrame,
StartInterruptionFrame,
StopInterruptionFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
class AudioBufferProcessor(FrameProcessor): class AudioBufferProcessor(FrameProcessor):
def __init__(self, **kwargs): def __init__(self, **kwargs):
""" """
Initialize the AudioBufferProcessor. Initialize the AudioBufferProcessor.
@@ -33,16 +37,13 @@ class AudioBufferProcessor(FrameProcessor):
self._sample_rate = None self._sample_rate = None
def _buffer_has_audio(self, buffer: bytearray): def _buffer_has_audio(self, buffer: bytearray):
return ( return buffer is not None and len(buffer) > 0
buffer is not None and
len(buffer) > 0
)
def _has_audio(self): def _has_audio(self):
return ( return (
self._buffer_has_audio(self._user_audio_buffer) and self._buffer_has_audio(self._user_audio_buffer)
self._buffer_has_audio(self._assistant_audio_buffer) and and self._buffer_has_audio(self._assistant_audio_buffer)
self._sample_rate is not None and self._sample_rate is not None
) )
def _reset_audio_buffer(self): def _reset_audio_buffer(self):
@@ -51,13 +52,12 @@ class AudioBufferProcessor(FrameProcessor):
def _merge_audio_buffers(self): def _merge_audio_buffers(self):
with BytesIO() as buffer: with BytesIO() as buffer:
with wave.open(buffer, 'wb') as wf: with wave.open(buffer, "wb") as wf:
wf.setnchannels(2) wf.setnchannels(2)
wf.setsampwidth(2) wf.setsampwidth(2)
wf.setframerate(self._sample_rate) wf.setframerate(self._sample_rate)
# Interleave the two audio streams # Interleave the two audio streams
max_length = max(len(self._user_audio_buffer), max_length = max(len(self._user_audio_buffer), len(self._assistant_audio_buffer))
len(self._assistant_audio_buffer))
interleaved = bytearray(max_length * 2) interleaved = bytearray(max_length * 2)
for i in range(0, max_length, 2): for i in range(0, max_length, 2):
@@ -89,7 +89,7 @@ class AudioBufferProcessor(FrameProcessor):
# Sync the assistant's buffer to the user's buffer by adding silence if needed # Sync the assistant's buffer to the user's buffer by adding silence if needed
if len(self._user_audio_buffer) > len(self._assistant_audio_buffer): if len(self._user_audio_buffer) > len(self._assistant_audio_buffer):
silence_length = len(self._user_audio_buffer) - len(self._assistant_audio_buffer) silence_length = len(self._user_audio_buffer) - len(self._assistant_audio_buffer)
silence = b'\x00' * silence_length silence = b"\x00" * silence_length
self._assistant_audio_buffer.extend(silence) self._assistant_audio_buffer.extend(silence)
# if the assistant is speaking, include all audio from the assistant, # if the assistant is speaking, include all audio from the assistant,

85
src/pipecat/services/canonical.py
View File

@@ -12,14 +12,14 @@ try:
except ModuleNotFoundError as e: except ModuleNotFoundError as e:
logger.error(f"Exception: {e}") logger.error(f"Exception: {e}")
logger.error( logger.error(
"In order to use Canonical Metrics, you need to `pip install pipecat-ai[canonical]`. " + "In order to use Canonical Metrics, you need to `pip install pipecat-ai[canonical]`. "
"Also, set the `CANONICAL_API_KEY` environment variable.") + "Also, set the `CANONICAL_API_KEY` environment variable."
)
raise Exception(f"Missing module: {e}") raise Exception(f"Missing module: {e}")
from pipecat.frames.frames import CancelFrame, EndFrame, Frame from pipecat.frames.frames import CancelFrame, EndFrame, Frame
from pipecat.processors.audio.audio_buffer_processor import \ from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor
AudioBufferProcessor
from pipecat.processors.frame_processor import FrameDirection from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_services import AIService from pipecat.services.ai_services import AIService
@@ -56,15 +56,16 @@ class CanonicalMetricsService(AIService):
""" """
def __init__( def __init__(
self, self,
aiohttp_session: aiohttp.ClientSession, aiohttp_session: aiohttp.ClientSession,
audio_buffer_processor: AudioBufferProcessor, audio_buffer_processor: AudioBufferProcessor,
call_id: str, call_id: str,
assistant: str, assistant: str,
api_key: str, api_key: str,
api_url: str = "https://voiceapp.canonical.chat/api/v1", api_url: str = "https://voiceapp.canonical.chat/api/v1",
assistant_speaks_first: bool = True, assistant_speaks_first: bool = True,
output_dir: str = "recordings"): output_dir: str = "recordings",
):
super().__init__() super().__init__()
self._aiohttp_session = aiohttp_session self._aiohttp_session = aiohttp_session
self._audio_buffer_processor = audio_buffer_processor self._audio_buffer_processor = audio_buffer_processor
@@ -92,7 +93,7 @@ class CanonicalMetricsService(AIService):
filename = self._get_output_filename() filename = self._get_output_filename()
wave_data = pipeline._merge_audio_buffers() wave_data = pipeline._merge_audio_buffers()
async with aiofiles.open(filename, 'wb') as file: async with aiofiles.open(filename, "wb") as file:
await file.write(wave_data) await file.write(wave_data)
try: try:
@@ -109,10 +110,7 @@ class CanonicalMetricsService(AIService):
return f"{self._output_dir}/{timestamp}-{uuid.uuid4().hex}.wav" return f"{self._output_dir}/{timestamp}-{uuid.uuid4().hex}.wav"
def _request_headers(self): def _request_headers(self):
return { return {"Content-Type": "application/json", "X-Canonical-Api-Key": self._api_key}
"Content-Type": "application/json",
"X-Canonical-Api-Key": self._api_key
}
async def _multipart_upload(self, file_path: str): async def _multipart_upload(self, file_path: str):
upload_request, upload_response = await self._request_upload(file_path) upload_request, upload_response = await self._request_upload(file_path)
@@ -129,19 +127,14 @@ class CanonicalMetricsService(AIService):
numparts = int((filesize + PART_SIZE - 1) / PART_SIZE) numparts = int((filesize + PART_SIZE - 1) / PART_SIZE)
params = { params = {
'filename': filename, "filename": filename,
'parts': numparts, "parts": numparts,
'callId': self._call_id, "callId": self._call_id,
'assistant': { "assistant": {"id": self._assistant, "speaksFirst": self._assistant_speaks_first},
'id': self._assistant,
'speaksFirst': self._assistant_speaks_first
}
} }
logger.debug(f"Requesting presigned URLs for {numparts} parts") logger.debug(f"Requesting presigned URLs for {numparts} parts")
response = await self._aiohttp_session.post( response = await self._aiohttp_session.post(
f"{self._api_url}/recording/uploadRequest", f"{self._api_url}/recording/uploadRequest", headers=self._request_headers(), json=params
headers=self._request_headers(),
json=params
) )
if not response.ok: if not response.ok:
logger.error(f"Failed to get presigned URLs: {await response.text()}") logger.error(f"Failed to get presigned URLs: {await response.text()}")
@@ -149,15 +142,11 @@ class CanonicalMetricsService(AIService):
response_json = await response.json() response_json = await response.json()
return params, response_json return params, response_json
async def _upload_parts( async def _upload_parts(self, file_path: str, upload_response: Dict) -> List[Dict]:
self, urls = upload_response["urls"]
file_path: str,
upload_response: Dict) -> List[Dict]:
urls = upload_response['urls']
parts = [] parts = []
try: try:
async with aiofiles.open(file_path, 'rb') as file: async with aiofiles.open(file_path, "rb") as file:
for partnum, upload_url in enumerate(urls, start=1): for partnum, upload_url in enumerate(urls, start=1):
data = await file.read(PART_SIZE) data = await file.read(PART_SIZE)
if not data: if not data:
@@ -168,35 +157,29 @@ class CanonicalMetricsService(AIService):
logger.error(f"Failed to upload part {partnum}: {await response.text()}") logger.error(f"Failed to upload part {partnum}: {await response.text()}")
return None return None
etag = response.headers['ETag'] etag = response.headers["ETag"]
parts.append({'partnum': str(partnum), 'etag': etag}) parts.append({"partnum": str(partnum), "etag": etag})
except Exception as e: except Exception as e:
logger.error(f"Multipart upload aborted, an error occurred: {str(e)}") logger.error(f"Multipart upload aborted, an error occurred: {str(e)}")
return parts return parts
async def _upload_complete( async def _upload_complete(
self, self, parts: List[Dict], upload_request: Dict, upload_response: Dict
parts: List[Dict], ):
upload_request: Dict,
upload_response: Dict):
params = { params = {
'filename': upload_request['filename'], "filename": upload_request["filename"],
'parts': parts, "parts": parts,
'slug': upload_response['slug'], "slug": upload_response["slug"],
'callId': self._call_id, "callId": self._call_id,
'assistant': { "assistant": {"id": self._assistant, "speaksFirst": self._assistant_speaks_first},
'id': self._assistant,
'speaksFirst': self._assistant_speaks_first
}
} }
logger.debug(f"Completing upload for {params['filename']}") logger.debug(f"Completing upload for {params['filename']}")
logger.debug(f"Slug: {params['slug']}") logger.debug(f"Slug: {params['slug']}")
response = await self._aiohttp_session.post( response = await self._aiohttp_session.post(
f"{self._api_url}/recording/uploadComplete", f"{self._api_url}/recording/uploadComplete",
headers=self._request_headers(), headers=self._request_headers(),
json=params json=params,
) )
if not response.ok: if not response.ok:
logger.error(f"Failed to complete upload: {await response.text()}") logger.error(f"Failed to complete upload: {await response.text()}")