diff --git a/CHANGELOG.md b/CHANGELOG.md index 135bc8db7..9e7e7f87b 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 49c3eaaaf..0b013dbfb 100644 --- a/src/pipecat/frames/frames.py +++ b/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 diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index ac3e1f618..3424aac86 100644 --- a/src/pipecat/pipeline/task.py +++ b/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))) - return MetricsFrame(ttfb=ttfb) + ttfb = [{"name": p.name, "time": 0.0} for p in processors] + 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( diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index db873ca26..1fa805c1e 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/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._should_report_ttfb = True + self._metrics = FrameProcessorMetrics(name=self.name) @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: - self._start_ttfb_time = time.time() - self._should_report_ttfb = not self._report_only_initial_ttfb + if self.can_generate_metrics() and self.metrics_enabled: + await self._metrics.start_ttfb_metrics(self._report_only_initial_ttfb) async def stop_ttfb_metrics(self): - if self.metrics_enabled and self._start_ttfb_time > 0: - ttfb = time.time() - self._start_ttfb_time - logger.debug(f"{self.name} TTFB: {ttfb}") - await self.push_frame(MetricsFrame(ttfb={self.name: ttfb})) - self._start_ttfb_time = 0 + if self.can_generate_metrics() and self.metrics_enabled: + frame = await self._metrics.stop_ttfb_metrics() + if frame: + await self.push_frame(frame) + + 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 diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 308ffc7dc..0fd7cb9d1 100644 --- a/src/pipecat/services/ai_services.py +++ b/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) diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index ce480e783..adf4f597a 100644 --- a/src/pipecat/services/openai.py +++ b/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()) diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 13c8dbce6..929ac5b42 100644 --- a/src/pipecat/transports/services/daily.py +++ b/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) diff --git a/src/pipecat/vad/silero.py b/src/pipecat/vad/silero.py index 99bc71ee6..362a9eae9 100644 --- a/src/pipecat/vad/silero.py +++ b/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...")