Merge pull request #267 from pipecat-ai/aleix/processing-metrics

add support for processing metrics

CHANGELOG.md

@@ -5,6 +5,14 @@ All notable changes to **pipecat** will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Added
+
+- The `MetricsFrame` now includes processing metrics if metrics are enabled. The
+  processing metrics indicate the time a processor needs to generate all its
+  output. Note that not all processors generate these kind of metrics.
+
 ## [0.0.35] - 2024-06-28
 
 ### Changed

src/pipecat/frames/frames.py

@@ -244,8 +244,8 @@ class StopInterruptionFrame(SystemFrame):
 class MetricsFrame(SystemFrame):
     """Emitted by processor that can compute metrics like latencies.
     """
-    ttfb: Mapping[str, float]
+    ttfb: List[Mapping[str, Any]] | None = None
-
+    processing: List[Mapping[str, Any]] | None = None
 
 #
 # Control frames

src/pipecat/pipeline/task.py

@@ -95,8 +95,9 @@ class PipelineTask:
 
     def _initial_metrics_frame(self) -> MetricsFrame:
         processors = self._pipeline.processors_with_metrics()
-        ttfb = dict(zip([p.name for p in processors], [0] * len(processors)))
+        ttfb = [{"name": p.name, "time": 0.0} for p in processors]
-        return MetricsFrame(ttfb=ttfb)
+        processing = [{"name": p.name, "time": 0.0} for p in processors]
+        return MetricsFrame(ttfb=ttfb, processing=processing)
 
     async def _process_down_queue(self):
         start_frame = StartFrame(

src/pipecat/processors/frame_processor.py

@@ -9,7 +9,7 @@ import time
 
 from enum import Enum
 
-from pipecat.frames.frames import ErrorFrame, Frame, MetricsFrame, StartFrame, UserStoppedSpeakingFrame
+from pipecat.frames.frames import ErrorFrame, Frame, MetricsFrame, StartFrame, StartInterruptionFrame, UserStoppedSpeakingFrame
 from pipecat.utils.utils import obj_count, obj_id
 
 from loguru import logger
@@ -20,6 +20,48 @@ class FrameDirection(Enum):
     UPSTREAM = 2
 
 
+class FrameProcessorMetrics:
+    def __init__(self, name: str):
+        self._name = name
+        self._start_ttfb_time = 0
+        self._start_processing_time = 0
+        self._should_report_ttfb = True
+
+    async def start_ttfb_metrics(self, report_only_initial_ttfb):
+        if self._should_report_ttfb:
+            self._start_ttfb_time = time.time()
+            self._should_report_ttfb = not report_only_initial_ttfb
+
+    async def stop_ttfb_metrics(self):
+        if self._start_ttfb_time == 0:
+            return None
+
+        value = time.time() - self._start_ttfb_time
+        logger.debug(f"{self._name} TTFB: {value}")
+        ttfb = {
+            "processor": self._name,
+            "value": value
+        }
+        self._start_ttfb_time = 0
+        return MetricsFrame(ttfb=[ttfb])
+
+    async def start_processing_metrics(self):
+        self._start_processing_time = time.time()
+
+    async def stop_processing_metrics(self):
+        if self._start_processing_time == 0:
+            return None
+
+        value = time.time() - self._start_processing_time
+        logger.debug(f"{self._name} processing time: {value}")
+        processing = {
+            "processor": self._name,
+            "value": value
+        }
+        self._start_processing_time = 0
+        return MetricsFrame(processing=[processing])
+
+
 class FrameProcessor:
 
     def __init__(
@@ -39,8 +81,7 @@ class FrameProcessor:
         self._report_only_initial_ttfb = False
 
         # Metrics
-        self._start_ttfb_time = 0
+        self._metrics = FrameProcessorMetrics(name=self.name)
-        self._should_report_ttfb = True
 
     @property
     def interruptions_allowed(self):
@@ -58,16 +99,28 @@ class FrameProcessor:
         return False
 
     async def start_ttfb_metrics(self):
-        if self.metrics_enabled and self._should_report_ttfb:
+        if self.can_generate_metrics() and self.metrics_enabled:
-            self._start_ttfb_time = time.time()
+            await self._metrics.start_ttfb_metrics(self._report_only_initial_ttfb)
-            self._should_report_ttfb = not self._report_only_initial_ttfb
 
     async def stop_ttfb_metrics(self):
-        if self.metrics_enabled and self._start_ttfb_time > 0:
+        if self.can_generate_metrics() and self.metrics_enabled:
-            ttfb = time.time() - self._start_ttfb_time
+            frame = await self._metrics.stop_ttfb_metrics()
-            logger.debug(f"{self.name} TTFB: {ttfb}")
+            if frame:
-            await self.push_frame(MetricsFrame(ttfb={self.name: ttfb}))
+                await self.push_frame(frame)
-            self._start_ttfb_time = 0
+
+    async def start_processing_metrics(self):
+        if self.can_generate_metrics() and self.metrics_enabled:
+            await self._metrics.start_processing_metrics()
+
+    async def stop_processing_metrics(self):
+        if self.can_generate_metrics() and self.metrics_enabled:
+            frame = await self._metrics.stop_processing_metrics()
+            if frame:
+                await self.push_frame(frame)
+
+    async def stop_all_metrics(self):
+        await self.stop_ttfb_metrics()
+        await self.stop_processing_metrics()
 
     async def cleanup(self):
         pass
@@ -85,6 +138,8 @@ class FrameProcessor:
             self._allow_interruptions = frame.allow_interruptions
             self._enable_metrics = frame.enable_metrics
             self._report_only_initial_ttfb = frame.report_only_initial_ttfb
+        elif isinstance(frame, StartInterruptionFrame):
+            await self.stop_all_metrics()
         elif isinstance(frame, UserStoppedSpeakingFrame):
             self._should_report_ttfb = True
 

src/pipecat/services/ai_services.py

@@ -127,7 +127,9 @@ class TTSService(AIService):
             return
 
         await self.push_frame(TTSStartedFrame())
+        await self.start_processing_metrics()
         await self.process_generator(self.run_tts(text))
+        await self.stop_processing_metrics()
         await self.push_frame(TTSStoppedFrame())
         # We send the original text after the audio. This way, if we are
         # interrupted, the text is not added to the assistant context.
@@ -208,7 +210,9 @@ class STTService(AIService):
             self._silence_num_frames = 0
             self._wave.close()
             self._content.seek(0)
+            await self.start_processing_metrics()
             await self.process_generator(self.run_stt(self._content.read()))
+            await self.stop_processing_metrics()
             (self._content, self._wave) = self._new_wave()
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -241,7 +245,9 @@ class ImageGenService(AIService):
 
         if isinstance(frame, TextFrame):
             await self.push_frame(frame, direction)
+            await self.start_processing_metrics()
             await self.process_generator(self.run_image_gen(frame.text))
+            await self.stop_processing_metrics()
         else:
             await self.push_frame(frame, direction)
 
@@ -261,6 +267,8 @@ class VisionService(AIService):
         await super().process_frame(frame, direction)
 
         if isinstance(frame, VisionImageRawFrame):
+            await self.start_processing_metrics()
             await self.process_generator(self.run_vision(frame))
+            await self.stop_processing_metrics()
         else:
             await self.push_frame(frame, direction)

src/pipecat/services/openai.py

@@ -9,7 +9,7 @@ import base64
 import io
 import json
 
-from typing import Any, AsyncGenerator, List, Literal
+from typing import AsyncGenerator, List, Literal
 
 from loguru import logger
 from PIL import Image
@@ -231,7 +231,9 @@ class BaseOpenAILLMService(LLMService):
 
         if context:
             await self.push_frame(LLMFullResponseStartFrame())
+            await self.start_processing_metrics()
             await self._process_context(context)
+            await self.stop_processing_metrics()
             await self.push_frame(LLMFullResponseEndFrame())
 
 

src/pipecat/transports/services/daily.py

@@ -656,11 +656,11 @@ class DailyOutputTransport(BaseOutputTransport):
         await self._client.send_message(frame)
 
     async def send_metrics(self, frame: MetricsFrame):
-        ttfb = [{"name": n, "time": t} for n, t in frame.ttfb.items()]
         message = DailyTransportMessageFrame(message={
             "type": "pipecat-metrics",
             "metrics": {
-                "ttfb": ttfb
+                "ttfb": frame.ttfb or [],
+                "processing": frame.processing or [],
             },
         })
         await self._client.send_message(message)

src/pipecat/vad/silero.py

@@ -37,7 +37,7 @@ class SileroVADAnalyzer(VADAnalyzer):
         super().__init__(sample_rate=sample_rate, num_channels=1, params=params)
 
         if sample_rate != 16000 and sample_rate != 8000:
-            raise Exception("Silero VAD sample rate needs to be 16000 or 8000")
+            raise ValueError("Silero VAD sample rate needs to be 16000 or 8000")
 
         logger.debug("Loading Silero VAD model...")