Add text aggregation time metric for TTS sentence aggregation

Add TextAggregationMetricsData measuring the time from the first LLM
token to the first complete sentence, representing the latency cost of
sentence aggregation in the TTS pipeline.

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.

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.
 

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

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