diff --git a/changelog/3696.added.md b/changelog/3696.added.md new file mode 100644 index 000000000..39726d930 --- /dev/null +++ b/changelog/3696.added.md @@ -0,0 +1 @@ +- Added `TextAggregationMetricsData` metric measuring the time from the first LLM token to the first complete sentence, representing the latency cost of sentence aggregation in the TTS pipeline. diff --git a/src/pipecat/metrics/metrics.py b/src/pipecat/metrics/metrics.py index ccf30227a..2030306e5 100644 --- a/src/pipecat/metrics/metrics.py +++ b/src/pipecat/metrics/metrics.py @@ -87,6 +87,19 @@ class TTSUsageMetricsData(MetricsData): value: int +class TextAggregationMetricsData(MetricsData): + """Text aggregation time metrics data. + + Measures the time from the first LLM token to the first complete sentence, + representing the latency cost of sentence aggregation in the TTS pipeline. + + Parameters: + value: Aggregation time in seconds. + """ + + value: float + + class TurnMetricsData(MetricsData): """Metrics data for turn detection predictions. diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index bcdb2d57b..37e8dc10d 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -485,6 +485,18 @@ class FrameProcessor(BaseObject): if frame: await self.push_frame(frame) + async def start_text_aggregation_metrics(self): + """Start text aggregation time metrics collection.""" + if self.can_generate_metrics() and self.metrics_enabled: + await self._metrics.start_text_aggregation_metrics() + + async def stop_text_aggregation_metrics(self): + """Stop text aggregation time metrics collection and push results.""" + if self.can_generate_metrics() and self.metrics_enabled: + frame = await self._metrics.stop_text_aggregation_metrics() + if frame: + await self.push_frame(frame) + async def stop_all_metrics(self): """Stop all active metrics collection.""" await self.stop_ttfb_metrics() diff --git a/src/pipecat/processors/metrics/frame_processor_metrics.py b/src/pipecat/processors/metrics/frame_processor_metrics.py index c82fd9698..cb5bc8a42 100644 --- a/src/pipecat/processors/metrics/frame_processor_metrics.py +++ b/src/pipecat/processors/metrics/frame_processor_metrics.py @@ -17,6 +17,7 @@ from pipecat.metrics.metrics import ( LLMUsageMetricsData, MetricsData, ProcessingMetricsData, + TextAggregationMetricsData, TTFBMetricsData, TTSUsageMetricsData, ) @@ -211,3 +212,27 @@ class FrameProcessorMetrics(BaseObject): ) logger.debug(f"{self._processor_name()} usage characters: {characters.value}") return MetricsFrame(data=[characters]) + + async def start_text_aggregation_metrics(self): + """Start measuring text aggregation time (first token to first sentence).""" + self._start_text_aggregation_time = time.time() + + async def stop_text_aggregation_metrics(self): + """Stop text aggregation measurement and generate metrics frame. + + Returns: + MetricsFrame containing text aggregation time, or None if not measuring. + """ + if ( + not hasattr(self, "_start_text_aggregation_time") + or self._start_text_aggregation_time == 0 + ): + return None + + value = time.time() - self._start_text_aggregation_time + logger.debug(f"{self._processor_name()} text aggregation time: {value}") + aggregation = TextAggregationMetricsData( + processor=self._processor_name(), value=value, model=self._model_name() + ) + self._start_text_aggregation_time = 0 + return MetricsFrame(data=[aggregation])