diff --git a/examples/canonical-metrics/bot.py b/examples/canonical-metrics/bot.py index b6da3fbd4..111c1666e 100644 --- a/examples/canonical-metrics/bot.py +++ b/examples/canonical-metrics/bot.py @@ -20,8 +20,10 @@ from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.llm_response import ( LLMAssistantResponseAggregator, LLMUserResponseAggregator) -from pipecat.processors.canonical_metrics_processor import CanonicalMetrics +from pipecat.processors.audio.audio_buffer_processor import \ + AudioBufferProcessor from pipecat.processors.user_marker_processor import UserMarkerProcessor +from pipecat.services.canonical import CanonicalMetricsService from pipecat.services.elevenlabs import ElevenLabsTTSService from pipecat.services.openai import OpenAILLMService from pipecat.transports.services.daily import DailyParams, DailyTransport @@ -103,7 +105,12 @@ async def main(): call completion, CanonicalMetrics will send the audio buffer to Canonical for analysis. Visit https://voice.canonical.chat to learn more. """ - canonical = CanonicalMetrics( + audio_buffer_processor = AudioBufferProcessor() + canonical = CanonicalMetricsService( + audio_buffer_processor=audio_buffer_processor, + aiohttp_session=session, + api_key=os.getenv("CANONICAL_API_KEY"), + api_url=os.getenv("CANONICAL_API_URL"), call_id=str(uuid.uuid4()), assistant="pipecat-chatbot", assistant_speaks_first=True, @@ -111,11 +118,12 @@ async def main(): usermarker = UserMarkerProcessor() pipeline = Pipeline([ transport.input(), # microphone - usermarker, # used to mark the user's audio in the pipeline + usermarker, user_response, llm, tts, - canonical, # captures audio and uploads to Canonical AI for metrics + audio_buffer_processor, # captures audio into a buffer + canonical, # uploads audio buffer to Canonical AI for metrics transport.output(), assistant_response, ]) diff --git a/examples/chatbot-audio-recording/bot.py b/examples/chatbot-audio-recording/bot.py index 7297de215..b6da9b54d 100644 --- a/examples/chatbot-audio-recording/bot.py +++ b/examples/chatbot-audio-recording/bot.py @@ -19,7 +19,7 @@ from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.llm_response import ( LLMAssistantResponseAggregator, LLMUserResponseAggregator) -from pipecat.processors.audio_buffer_processor import AudioBufferProcessor +from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor from pipecat.processors.user_marker_processor import UserMarkerProcessor from pipecat.services.elevenlabs import ElevenLabsTTSService from pipecat.services.openai import OpenAILLMService diff --git a/src/pipecat/processors/audio_buffer_processor.py b/src/pipecat/processors/audio/audio_buffer_processor.py similarity index 60% rename from src/pipecat/processors/audio_buffer_processor.py rename to src/pipecat/processors/audio/audio_buffer_processor.py index f724e7c5e..8edf197e4 100644 --- a/src/pipecat/processors/audio_buffer_processor.py +++ b/src/pipecat/processors/audio/audio_buffer_processor.py @@ -1,6 +1,7 @@ from pipecat.frames.frames import (AudioRawFrame, BotStartedSpeakingFrame, BotStoppedSpeakingFrame, Frame, - UserAudioFrame, UserStoppedSpeakingFrame) + UserAudioFrame, UserStartedSpeakingFrame, + UserStoppedSpeakingFrame) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -21,54 +22,61 @@ class AudioBufferProcessor(FrameProcessor): populated when the first audio frame is processed. """ super().__init__() - self.audio_buffer = bytearray() - self.num_channels = None - self.sample_rate = None - self.assistant_audio = False - self.user_audio = False + self._audio_buffer = bytearray() + self._num_channels = None + self._sample_rate = None + self._assistant_audio = False + self._user_audio = False + print(f"ctor::AudioBufferProcessor object memory address: {id(self)}") - def has_audio(self): + def _has_audio(self): return ( - self.audio_buffer and - len(self.audio_buffer) > 0 and - self.num_channels and - self.sample_rate + self._audio_buffer is not None and + len(self._audio_buffer) > 0 and + self._num_channels is not None and + self._sample_rate is not None ) + def _reset_audio_buffer(self): + self._audio_buffer = bytearray() + async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) - if isinstance(frame, AudioRawFrame) or isinstance(frame, UserAudioFrame): - if self.num_channels is None: - self.num_channels = frame.num_channels - if self.sample_rate is None: - self.sample_rate = frame.sample_rate + if isinstance(frame, AudioRawFrame): + if self._num_channels is None: + self._num_channels = frame.num_channels + if self._sample_rate is None: + self._sample_rate = frame.sample_rate - elif isinstance(frame, UserStoppedSpeakingFrame): - self.user_audio = False + if isinstance(frame, UserStoppedSpeakingFrame): + self._user_audio = False if isinstance(frame, BotStartedSpeakingFrame): - self.assistant_audio = True - self.user_audio = False # do not capture user audio if assistant is speaking + self._assistant_audio = True + self._user_audio = False # do not capture user audio if assistant is speaking + if isinstance(frame, BotStoppedSpeakingFrame): - self.assistant_audio = False + self._assistant_audio = False # Capture user audio if assistant is not speaking, even if it's silence, the point # here is to capture so that the conversation is as close to reality as possible. # This is important for evaluation and metrics capture. - self.user_audio = True + self._user_audio = True # only include audio from the user if the user is speaking, this is because audio from the user's # mic is always coming in. if we include all the user's audio there will be a long latency before # the user starts speaking because all of the user's silence during the assistant's speech will have been # added to the buffer. - if isinstance(frame, UserAudioFrame) and self.user_audio: - self.audio_buffer.extend(frame.audio) + # + # and include all audio from the assistant + if isinstance(frame, UserAudioFrame) and self._user_audio: + self._audio_buffer.extend(frame.audio) # include all audio from the assistant if ( isinstance(frame, AudioRawFrame) and not isinstance(frame, UserAudioFrame) ): - self.audio_buffer.extend(frame.audio) + self._audio_buffer.extend(frame.audio) # do not push the user's audio frame, doing so will result in echo if not isinstance(frame, UserAudioFrame): diff --git a/src/pipecat/processors/canonical_metrics_processor.py b/src/pipecat/processors/canonical_metrics_processor.py deleted file mode 100644 index e7c0afb61..000000000 --- a/src/pipecat/processors/canonical_metrics_processor.py +++ /dev/null @@ -1,202 +0,0 @@ -import os -import uuid -import wave -from datetime import datetime -from io import BytesIO -from typing import Dict, List, Tuple - -import aiohttp -from loguru import logger - -try: - import aiofiles - import aiofiles.os -except ModuleNotFoundError as e: - logger.error(f"Exception: {e}") - logger.error( - "In order to use Canonical Metrics, you need to `pip install pipecat-ai[canonical]`. " + - "Also, set the `CANONICAL_API_KEY` environment variable.") - raise Exception(f"Missing module: {e}") - - -from pipecat.frames.frames import CancelFrame, EndFrame, Frame -from pipecat.processors.audio_buffer_processor import AudioBufferProcessor -from pipecat.processors.frame_processor import FrameDirection - -""" -This class extends AudioBufferProcessor to handle audio processing and uploading -for the Canonical Voice API. -""" - - -class CanonicalMetrics(AudioBufferProcessor): - """ - Initialize a CanonicalAudioProcessor instance. - - This class extends AudioBufferProcessor to handle audio processing and uploading - for the Canonical Voice API. - - Args: - call_id (str): Your unique identifier for the call. This is used to match the call in the Canonical Voice system to the call in your system. - assistant (str): Identifier for the AI assistant. This can be whatever you want, it's intended for you convenience so you can distinguish - between different assistants and a grouping mechanism for calls. - assistant_speaks_first (bool, optional): Indicates if the assistant speaks first in the conversation. Defaults to True. - output_dir (str, optional): Directory to save temporary audio files. Defaults to "recordings". - default_part_size (int, optional): Default size for multipart upload parts in bytes. Defaults to 1MB (1024 * 1024 * 1). - - Attributes: - call_id (str): Stores the unique call identifier. - assistant (str): Stores the assistant identifier. - assistant_speaks_first (bool): Indicates whether the assistant speaks first. - output_dir (str): Directory path for saving temporary audio files. - partsize (int): Size of each part for multipart uploads. - - The constructor also ensures that the output directory exists. - This class requires a Canonical API key to be set in the CANONICAL_API_KEY environment variable. - """ - - def __init__( - self, - call_id: str, - assistant: str, - assistant_speaks_first: bool = True, - output_dir: str = "recordings", - default_part_size: int = 1024 * 1024 * 1): - super().__init__() - if not os.environ.get("CANONICAL_API_KEY"): - raise ValueError( - "CANONICAL_API_KEY is not set, a Canonical API key is required to use this class") - self.call_id = call_id - self.assistant = assistant - self.assistant_speaks_first = assistant_speaks_first - self.output_dir = output_dir - self.partsize = default_part_size - self.end_of_call = False - - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - if self.end_of_call: - return - - if (isinstance(frame, EndFrame) or isinstance(frame, CancelFrame)): - self.end_of_call = True - if self.has_audio(): - os.makedirs(self.output_dir, exist_ok=True) - filename = self.get_output_filename() - with BytesIO() as buffer: - with wave.open(buffer, 'wb') as wf: - wf.setnchannels(self.num_channels) - wf.setsampwidth(self.sample_rate // 8000) - wf.setframerate(self.sample_rate) - wf.writeframes(self.audio_buffer) - wave_data = buffer.getvalue() - - async with aiofiles.open(filename, 'wb') as file: - await file.write(wave_data) - - try: - await self.multipart_upload(filename) - await aiofiles.os.remove(filename) - except FileNotFoundError: - pass - except Exception as e: - raise e - self.audio_buffer = bytearray() - - def get_output_filename(self): - timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") - return f"{self.output_dir}/{timestamp}-{uuid.uuid4().hex}.wav" - - def canonical_api_url(self): - return os.environ.get("CANONICAL_API_URL", "https://voiceapp.canonical.chat/api/v1") - - def request_headers(self): - return { - "Content-Type": "application/json", - "X-Canonical-Api-Key": os.environ.get("CANONICAL_API_KEY") - } - - async def multipart_upload(self, file_path: str): - upload_request, upload_response = await self.request_upload(file_path) - parts = await self.upload_parts(file_path, upload_request, upload_response) - await self.upload_complete(parts, upload_request, upload_response) - - async def request_upload(self, file_path: str) -> Tuple[Dict, Dict]: - filename = os.path.basename(file_path) - filename = f"{str(uuid.uuid4())}-{filename}" - filesize = os.path.getsize(file_path) - numparts = int((filesize + self.partsize - 1) / self.partsize) - - params = { - 'filename': filename, - 'parts': numparts, - 'assistant': self.assistant, - 'assistantSpeaksFirst': self.assistant_speaks_first - } - print(f"Requesting presigned URLs for {numparts} parts") - async with aiohttp.ClientSession() as session: - async with session.post( - f"{self.canonical_api_url()}/recording/uploadRequest", - headers=self.request_headers(), - json=params - ) as response: - if not response.ok: - raise Exception(f"Failed to get presigned URLs: {await response.text()}") - response_json = await response.json() - return params, response_json - - async def upload_parts( - self, - file_path: str, - upload_request: Dict, - upload_response: Dict) -> List[Dict]: - - urls = upload_response['urls'] - parts = [] - try: - async with aiofiles.open(file_path, 'rb') as file: - async with aiohttp.ClientSession() as session: - for partnum, upload_url in enumerate(urls, start=1): - data = await file.read(self.partsize) - if not data: - break - - async with session.put(upload_url, data=data) as response: - if not response.ok: - logger.error(f"Failed to upload part {partnum}: {await response.text()}") - raise Exception(f"Failed to upload part {partnum}: {await response.text()}") - - etag = response.headers['ETag'] - parts.append({'partnum': str(partnum), 'etag': etag}) - - except Exception as e: - logger.error(f"Multipart upload aborted, an error occurred: {str(e)}") - return parts - - async def upload_complete( - self, - parts: List[Dict], - upload_request: Dict, - upload_response: Dict): - - params = { - 'filename': upload_request['filename'], - 'parts': parts, - 'slug': upload_response['slug'], - 'callId': self.call_id, - 'assistant': { - 'id': self.assistant, - 'speaksFirst': self.assistant_speaks_first - } - } - print(f"Completing upload for {params['filename']}") - print(f"Slug: {params['slug']}") - async with aiohttp.ClientSession() as session: - async with session.post( - f"{self.canonical_api_url()}/recording/uploadComplete", - headers=self.request_headers(), - json=params - ) as response: - if not response.ok: - logger.error(f"Failed to complete upload: {await response.text()}") - raise Exception(f"Failed to complete upload: {await response.text()}") diff --git a/src/pipecat/services/canonical.py b/src/pipecat/services/canonical.py new file mode 100644 index 000000000..a973d5481 --- /dev/null +++ b/src/pipecat/services/canonical.py @@ -0,0 +1,212 @@ +import os +import uuid +import wave +from datetime import datetime +from io import BytesIO +from typing import Dict, List, Tuple + +import aiohttp +from loguru import logger + +try: + import aiofiles + import aiofiles.os +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error( + "In order to use Canonical Metrics, you need to `pip install pipecat-ai[canonical]`. " + + "Also, set the `CANONICAL_API_KEY` environment variable.") + raise Exception(f"Missing module: {e}") + + +from pipecat.frames.frames import CancelFrame, EndFrame, Frame +from pipecat.processors.audio.audio_buffer_processor import \ + AudioBufferProcessor +from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.ai_services import AIService + +# Multipart upload part size in bytes, cannot be smaller than 5MB +PART_SIZE = 1024 * 1024 * 5 +""" +This class extends AudioBufferProcessor to handle audio processing and uploading +for the Canonical Voice API. +""" + + +class CanonicalMetricsService(AIService): + """ + Initialize a CanonicalAudioProcessor instance. + + This class extends AudioBufferProcessor to handle audio processing and uploading + for the Canonical Voice API. + + Args: + call_id (str): Your unique identifier for the call. This is used to match the call in the Canonical Voice system to the call in your system. + assistant (str): Identifier for the AI assistant. This can be whatever you want, it's intended for you convenience so you can distinguish + between different assistants and a grouping mechanism for calls. + assistant_speaks_first (bool, optional): Indicates if the assistant speaks first in the conversation. Defaults to True. + output_dir (str, optional): Directory to save temporary audio files. Defaults to "recordings". + + Attributes: + call_id (str): Stores the unique call identifier. + assistant (str): Stores the assistant identifier. + assistant_speaks_first (bool): Indicates whether the assistant speaks first. + output_dir (str): Directory path for saving temporary audio files. + + The constructor also ensures that the output directory exists. + This class requires a Canonical API key to be set in the CANONICAL_API_KEY environment variable. + """ + + def __init__( + self, + aiohttp_session: aiohttp.ClientSession, + audio_buffer_processor: AudioBufferProcessor, + call_id: str, + assistant: str, + api_key: str, + api_url: str = "https://voiceapp.canonical.chat/api/v1", + assistant_speaks_first: bool = True, + output_dir: str = "recordings"): + super().__init__() + self._aiohttp_session = aiohttp_session + self._audio_buffer_processor = audio_buffer_processor + self._api_key = api_key + self._api_url = api_url + self._call_id = call_id + self._assistant = assistant + self._assistant_speaks_first = assistant_speaks_first + self._output_dir = output_dir + + async def stop(self, frame: EndFrame): + await self._process_audio() + + async def cancel(self, frame: CancelFrame): + await self._process_audio() + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + await self.push_frame(frame, direction) + + async def _process_audio(self): + pipeline = self._audio_buffer_processor + if pipeline._has_audio(): + os.makedirs(self._output_dir, exist_ok=True) + filename = self._get_output_filename() + with BytesIO() as buffer: + with wave.open(buffer, 'wb') as wf: + wf.setnchannels(pipeline._num_channels) + wf.setsampwidth(pipeline._sample_rate // 8000) + wf.setframerate(pipeline._sample_rate) + wf.writeframes(pipeline._audio_buffer) + wave_data = buffer.getvalue() + + async with aiofiles.open(filename, 'wb') as file: + await file.write(wave_data) + + try: + await self._multipart_upload(filename) + pipeline._reset_audio_buffer() + # await aiofiles.os.remove(filename) + except FileNotFoundError: + pass + except Exception as e: + logger.error(f"Failed to upload recording: {e}") + + def _get_output_filename(self): + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + return f"{self._output_dir}/{timestamp}-{uuid.uuid4().hex}.wav" + + def _request_headers(self): + return { + "Content-Type": "application/json", + "X-Canonical-Api-Key": self._api_key + } + + async def _multipart_upload(self, file_path: str): + upload_request, upload_response = await self._request_upload(file_path) + if upload_request is None or upload_response is None: + return + parts = await self._upload_parts(file_path, upload_response) + if parts is None: + return + await self._upload_complete(parts, upload_request, upload_response) + + async def _request_upload(self, file_path: str) -> Tuple[Dict, Dict]: + filename = os.path.basename(file_path) + filesize = os.path.getsize(file_path) + numparts = int((filesize + PART_SIZE - 1) / PART_SIZE) + + params = { + 'filename': filename, + 'parts': numparts, + 'callId': self._call_id, + 'assistant': { + 'id': self._assistant, + 'speaksFirst': self._assistant_speaks_first + } + } + logger.debug(f"Requesting presigned URLs for {numparts} parts") + response = await self._aiohttp_session.post( + f"{self._api_url}/recording/uploadRequest", + headers=self._request_headers(), + json=params + ) + if not response.ok: + logger.error(f"Failed to get presigned URLs: {await response.text()}") + return None, None + response_json = await response.json() + return params, response_json + + async def _upload_parts( + self, + file_path: str, + upload_response: Dict) -> List[Dict]: + + urls = upload_response['urls'] + parts = [] + try: + async with aiofiles.open(file_path, 'rb') as file: + for partnum, upload_url in enumerate(urls, start=1): + data = await file.read(PART_SIZE) + if not data: + break + + response = await self._aiohttp_session.put(upload_url, data=data) + if not response.ok: + logger.error(f"Failed to upload part {partnum}: {await response.text()}") + return None + + etag = response.headers['ETag'] + parts.append({'partnum': str(partnum), 'etag': etag}) + + except Exception as e: + logger.error(f"Multipart upload aborted, an error occurred: {str(e)}") + return parts + + async def _upload_complete( + self, + parts: List[Dict], + upload_request: Dict, + upload_response: Dict): + + params = { + 'filename': upload_request['filename'], + 'parts': parts, + 'slug': upload_response['slug'], + 'callId': self._call_id, + 'assistant': { + 'id': self._assistant, + 'speaksFirst': self._assistant_speaks_first + } + } + logger.debug(f"Completing upload for {params['filename']}") + logger.debug(f"Slug: {params['slug']}") + response = await self._aiohttp_session.post( + f"{self._api_url}/recording/uploadComplete", + headers=self._request_headers(), + json=params + ) + if not response.ok: + logger.error(f"Failed to complete upload: {await response.text()}") + return +