From 22e176e329742cd44e519ac5c73ddb56f1a76bab Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 8 Aug 2024 16:46:56 -0700 Subject: [PATCH] processors(base): add start_llm_usage_metrics and start_tts_usage_metrics --- CHANGELOG.md | 5 ++++ examples/foundational/07-interruptible.py | 1 + src/pipecat/frames/frames.py | 1 + src/pipecat/pipeline/task.py | 2 ++ src/pipecat/processors/frame_processor.py | 31 +++++++++++++++++++++++ src/pipecat/services/azure.py | 9 ++----- src/pipecat/services/cartesia.py | 8 +----- src/pipecat/services/deepgram.py | 10 +++----- src/pipecat/services/elevenlabs.py | 10 +++----- src/pipecat/services/openai.py | 28 ++++++++------------ src/pipecat/services/playht.py | 10 +++----- src/pipecat/services/xtts.py | 10 +++----- 12 files changed, 65 insertions(+), 60 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f285bc014..4cfec75d8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `DailyRESTHelper.delete_room_by_name()`. + +- Added LLM and TTS usage metrics. Those will be enabled by when + `enable_usage_metrics` is True. + - `AudioRawFrame`s are not pushed downstream from the base output transport. This allows capturing the exact words the bot says by adding an STT service at the end of the pipeline. diff --git a/examples/foundational/07-interruptible.py b/examples/foundational/07-interruptible.py index 05c0ccd5a..6aab95b31 100644 --- a/examples/foundational/07-interruptible.py +++ b/examples/foundational/07-interruptible.py @@ -79,6 +79,7 @@ async def main(): task = PipelineTask(pipeline, PipelineParams( allow_interruptions=True, enable_metrics=True, + enable_usage_metrics=True, report_only_initial_ttfb=True, )) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 8cd1fe8e7..90127492b 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -294,6 +294,7 @@ class StartFrame(ControlFrame): """This is the first frame that should be pushed down a pipeline.""" allow_interruptions: bool = False enable_metrics: bool = False + enable_usage_metrics: bool = False report_only_initial_ttfb: bool = False diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 9f45f33ea..cdd3eafb2 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -21,6 +21,7 @@ from loguru import logger class PipelineParams(BaseModel): allow_interruptions: bool = False enable_metrics: bool = False + enable_usage_metrics: bool = False send_initial_empty_metrics: bool = True report_only_initial_ttfb: bool = False @@ -104,6 +105,7 @@ class PipelineTask: start_frame = StartFrame( allow_interruptions=self._params.allow_interruptions, enable_metrics=self._params.enable_metrics, + enable_usage_metrics=self._params.enable_metrics, report_only_initial_ttfb=self._params.report_only_initial_ttfb ) await self._source.process_frame(start_frame, FrameDirection.DOWNSTREAM) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 405936e06..5bc47fd03 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -61,6 +61,19 @@ class FrameProcessorMetrics: self._start_processing_time = 0 return MetricsFrame(processing=[processing]) + async def start_llm_usage_metrics(self, tokens: dict): + logger.debug( + f"{self._name} prompt tokens: {tokens['prompt_tokens']}, completion tokens: {tokens['completion_tokens']}") + return MetricsFrame(tokens=[tokens]) + + async def start_tts_usage_metrics(self, text: str): + characters = { + "processor": self._name, + "value": len(text), + } + logger.debug(f"{self._name} usage characters: {characters['value']}") + return MetricsFrame(characters=[characters]) + class FrameProcessor: @@ -80,6 +93,7 @@ class FrameProcessor: # Properties self._allow_interruptions = False self._enable_metrics = False + self._enable_usage_metrics = False self._report_only_initial_ttfb = False # Metrics @@ -93,6 +107,10 @@ class FrameProcessor: def metrics_enabled(self): return self._enable_metrics + @property + def usage_metrics_enabled(self): + return self._enable_usage_metrics + @property def report_only_initial_ttfb(self): return self._report_only_initial_ttfb @@ -120,6 +138,18 @@ class FrameProcessor: if frame: await self.push_frame(frame) + async def start_llm_usage_metrics(self, tokens: dict): + if self.can_generate_metrics() and self.usage_metrics_enabled: + frame = await self._metrics.start_llm_usage_metrics(tokens) + if frame: + await self.push_frame(frame) + + async def start_tts_usage_metrics(self, text: str): + if self.can_generate_metrics() and self.usage_metrics_enabled: + frame = await self._metrics.start_tts_usage_metrics(text) + if frame: + await self.push_frame(frame) + async def stop_all_metrics(self): await self.stop_ttfb_metrics() await self.stop_processing_metrics() @@ -145,6 +175,7 @@ class FrameProcessor: if isinstance(frame, StartFrame): self._allow_interruptions = frame.allow_interruptions self._enable_metrics = frame.enable_metrics + self._enable_usage_metrics = frame.enable_usage_metrics self._report_only_initial_ttfb = frame.report_only_initial_ttfb elif isinstance(frame, StartInterruptionFrame): await self.stop_all_metrics() diff --git a/src/pipecat/services/azure.py b/src/pipecat/services/azure.py index 13a3e3b38..aa3b5831e 100644 --- a/src/pipecat/services/azure.py +++ b/src/pipecat/services/azure.py @@ -88,13 +88,7 @@ class AzureTTSService(TTSService): async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") - if self.can_generate_metrics() and self.metrics_enabled: - characters = { - "processor": self.name, - "value": len(text), - } - logger.debug(f"{self.name} Characters: {characters['value']}") - await self.push_frame(MetricsFrame(characters=[characters])) + await self.start_ttfb_metrics() ssml = ( @@ -110,6 +104,7 @@ class AzureTTSService(TTSService): result = await asyncio.to_thread(self._speech_synthesizer.speak_ssml, (ssml)) if result.reason == ResultReason.SynthesizingAudioCompleted: + await self.start_tts_usage_metrics(text) await self.stop_ttfb_metrics() # Azure always sends a 44-byte header. Strip it off. yield AudioRawFrame(audio=result.audio_data[44:], sample_rate=16000, num_channels=1) diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index b9e7499db..5233366cd 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -201,13 +201,6 @@ class CartesiaTTSService(TTSService): async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") - if self.can_generate_metrics() and self.metrics_enabled: - characters = { - "processor": self.name, - "value": len(text), - } - logger.debug(f"{self.name} Characters: {characters['value']}") - await self.push_frame(MetricsFrame(characters=[characters])) try: if not self._websocket: @@ -232,6 +225,7 @@ class CartesiaTTSService(TTSService): } try: await self._websocket.send(json.dumps(msg)) + await self.start_tts_usage_metrics(text) except Exception as e: logger.exception(f"{self} error sending message: {e}") await self._disconnect() diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py index bbe5d3007..8d58def56 100644 --- a/src/pipecat/services/deepgram.py +++ b/src/pipecat/services/deepgram.py @@ -71,13 +71,7 @@ class DeepgramTTSService(TTSService): async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") - if self.can_generate_metrics() and self.metrics_enabled: - characters = { - "processor": self.name, - "value": len(text), - } - logger.debug(f"{self.name} Characters: {characters['value']}") - await self.push_frame(MetricsFrame(characters=[characters])) + base_url = self._base_url request_url = f"{base_url}?model={self._voice}&encoding={self._encoding}&container=none&sample_rate={self._sample_rate}" headers = {"authorization": f"token {self._api_key}"} @@ -100,6 +94,8 @@ class DeepgramTTSService(TTSService): yield ErrorFrame(f"Error getting audio (status: {r.status}, error: {response_text})") return + await self.start_tts_usage_metrics(text) + async for data in r.content: await self.stop_ttfb_metrics() frame = AudioRawFrame(audio=data, sample_rate=self._sample_rate, num_channels=1) diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index 33590a637..55419f815 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -40,13 +40,7 @@ class ElevenLabsTTSService(TTSService): async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") - if self.can_generate_metrics() and self.metrics_enabled: - characters = { - "processor": self.name, - "value": len(text), - } - logger.debug(f"{self.name} Characters: {characters['value']}") - await self.push_frame(MetricsFrame(characters=[characters])) + url = f"https://api.elevenlabs.io/v1/text-to-speech/{self._voice_id}/stream" payload = {"text": text, "model_id": self._model} @@ -69,6 +63,8 @@ class ElevenLabsTTSService(TTSService): yield ErrorFrame(f"Error getting audio (status: {r.status}, error: {text})") return + await self.start_tts_usage_metrics(text) + async for chunk in r.content: if len(chunk) > 0: await self.stop_ttfb_metrics() diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index 01f5e596f..ad15b9907 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -135,17 +135,13 @@ class BaseOpenAILLMService(LLMService): async for chunk in chunk_stream: if chunk.usage: - if self.can_generate_metrics() and self.metrics_enabled: - tokens = { - "processor": self.name, - "prompt_tokens": chunk.usage.prompt_tokens, - "completion_tokens": chunk.usage.completion_tokens, - "total_tokens": chunk.usage.total_tokens - } - logger.debug( - f"{self.name} prompt tokens: {tokens['prompt_tokens']}, completion tokens: {tokens['completion_tokens']}") - - await self.push_frame(MetricsFrame(tokens=[tokens])) + tokens = { + "processor": self.name, + "prompt_tokens": chunk.usage.prompt_tokens, + "completion_tokens": chunk.usage.completion_tokens, + "total_tokens": chunk.usage.total_tokens + } + await self.start_llm_usage_metrics(tokens) if len(chunk.choices) == 0: continue @@ -338,13 +334,6 @@ class OpenAITTSService(TTSService): async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") - if self.can_generate_metrics() and self.metrics_enabled: - characters = { - "processor": self.name, - "value": len(text), - } - logger.debug(f"{self.name} Characters: {characters['value']}") - await self.push_frame(MetricsFrame(characters=[characters])) try: await self.start_ttfb_metrics() @@ -360,6 +349,9 @@ class OpenAITTSService(TTSService): f"{self} error getting audio (status: {r.status_code}, error: {error})") yield ErrorFrame(f"Error getting audio (status: {r.status_code}, error: {error})") return + + await self.start_tts_usage_metrics(text) + async for chunk in r.iter_bytes(8192): if len(chunk) > 0: await self.stop_ttfb_metrics() diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index c24d8b8b3..b738040b6 100644 --- a/src/pipecat/services/playht.py +++ b/src/pipecat/services/playht.py @@ -48,13 +48,7 @@ class PlayHTTTSService(TTSService): async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") - if self.can_generate_metrics() and self.metrics_enabled: - characters = { - "processor": self.name, - "value": len(text), - } - logger.debug(f"{self.name} Characters: {characters['value']}") - await self.push_frame(MetricsFrame(characters=[characters])) + try: b = bytearray() in_header = True @@ -66,6 +60,8 @@ class PlayHTTTSService(TTSService): voice_engine="PlayHT2.0-turbo", options=self._options) + await self.start_tts_usage_metrics(text) + async for chunk in playht_gen: # skip the RIFF header. if in_header: diff --git a/src/pipecat/services/xtts.py b/src/pipecat/services/xtts.py index d67472891..a4b144b9a 100644 --- a/src/pipecat/services/xtts.py +++ b/src/pipecat/services/xtts.py @@ -70,13 +70,7 @@ class XTTSService(TTSService): async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") - if self.can_generate_metrics() and self.metrics_enabled: - characters = { - "processor": self.name, - "value": len(text), - } - logger.debug(f"{self.name} Characters: {characters['value']}") - await self.push_frame(MetricsFrame(characters=[characters])) + if not self._studio_speakers: logger.error(f"{self} no studio speakers available") return @@ -103,6 +97,8 @@ class XTTSService(TTSService): yield ErrorFrame(f"Error getting audio (status: {r.status}, error: {text})") return + await self.start_tts_usage_metrics(text) + buffer = bytearray() async for chunk in r.content.iter_chunked(1024):