Merge pull request #355 from pipecat-ai/aleix/usage-metrics-update

processors(base): add start_llm_usage_metrics and start_tts_usage_met…

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.

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,
         ))
 

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
 
 

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)

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()

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)

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()

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)

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()

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 = {
-                    tokens = {
+                    "processor": self.name,
-                        "processor": self.name,
+                    "prompt_tokens": chunk.usage.prompt_tokens,
-                        "prompt_tokens": chunk.usage.prompt_tokens,
+                    "completion_tokens": chunk.usage.completion_tokens,
-                        "completion_tokens": chunk.usage.completion_tokens,
+                    "total_tokens": chunk.usage.total_tokens
-                        "total_tokens": chunk.usage.total_tokens
+                }
-                    }
+                await self.start_llm_usage_metrics(tokens)
-                    logger.debug(
-                        f"{self.name} prompt tokens: {tokens['prompt_tokens']}, completion tokens: {tokens['completion_tokens']}")
-
-                    await self.push_frame(MetricsFrame(tokens=[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()

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:

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):