Merge pull request #5220 from pipecat-ai/pk/tts-processing-metrics-handoff

Stop reporting TTS processing metrics that time just the send instead of the real work

Author: kompfner

changelog/5220.changed.md

@@ -0,0 +1,5 @@
+- ⚠️ `WebsocketTTSService` subclasses and `DeepgramSageMakerTTSService` no longer report processing metrics, which were meaninglessly reporting zero on every turn. The metric is `ProcessingMetricsData` in `MetricsFrame`, surfaced as the `processing` field of RTVI's `metrics` message. TTS services whose processing time was a real number are unaffected.
+
+  Processing time is measured around `run_tts`. For a service that requests audio and waits for it in that call, that covers the real work. `WebsocketTTSService` subclasses instead push the text onto the socket and return, leaving the audio to arrive on a separate receive task, so the measurement only ever covered the send. `DeepgramSageMakerTTSService` does the same over bidirectional HTTP/2. TTFB and TTFA measure the latency that matters for all of them, and are unaffected.
+
+  There's a new `TTSService.supports_processing_metrics` property, which defaults to `True`. Set it to `False` on a custom service whose `run_tts` returns before synthesis finishes, or back to `True` on a `WebsocketTTSService` subclass that waits for the server to signal the end.

src/pipecat/services/deepgram/sagemaker/tts.py

@@ -144,6 +144,17 @@ def can_generate_metrics(self) -> bool:
         """
         return True
 
+    @property
+    def supports_processing_metrics(self) -> bool:
+        """Whether this service has a meaningful processing-time metric.
+
+        False: the SageMaker endpoint is driven over a bidirectional HTTP/2
+        stream, so ``run_tts`` sends the text and returns while audio arrives
+        on the receive task — the same handoff the websocket services make,
+        over a different transport.
+        """
+        return False
+
     async def start(self, frame: StartFrame):
         """Start the Deepgram SageMaker TTS service.
 

src/pipecat/services/tts_service.py

@@ -400,6 +400,19 @@ def _is_streaming_tokens(self) -> bool:
         """Whether the service is streaming tokens directly without sentence aggregation."""
         return self._text_aggregation_mode == TextAggregationMode.TOKEN
 
+    @property
+    def supports_processing_metrics(self) -> bool:
+        """Whether this service has a meaningful processing-time metric.
+
+        Processing time is measured around :meth:`run_tts`, so it only means
+        something when synthesis finishes before ``run_tts`` returns. Services
+        that hand the text off and receive audio elsewhere — anything holding a
+        persistent connection with its own receive task — return False, since
+        the measurement would cover the send and nothing else. TTFB and TTFA
+        carry the latency for those.
+        """
+        return True
+
     async def start_tts_usage_metrics(self, text: str):
         """Record TTS usage metrics.
 
@@ -1162,10 +1175,12 @@ async def _push_tts_frames(
         if self._is_streaming_tokens:
             self._streamed_text += text
 
-        # Skip per-token processing metrics when streaming. The per-token
-        # processing time is just websocket send overhead (~0.1ms) and not
-        # meaningful. TTFB captures the important timing for streaming TTS.
-        if not self._is_streaming_tokens:
+        # Two things disqualify the measurement. A service that hands text off
+        # for synthesis elsewhere would time only the handoff, and a token is
+        # the wrong unit to report a processing time against — one metric per
+        # token, for work that spans a sentence. TTFB and TTFA carry the timing
+        # that matters in both cases.
+        if self.supports_processing_metrics and not self._is_streaming_tokens:
             await self.start_processing_metrics()
 
         # Process all filters.
@@ -1272,7 +1287,7 @@ async def _push_tts_frames(
 
         await self.tts_process_generator(context_id, self.run_tts(prepared_text, context_id))
 
-        if not self._is_streaming_tokens:
+        if self.supports_processing_metrics and not self._is_streaming_tokens:
             await self.stop_processing_metrics()
 
         if self._push_text_frames and not self._is_streaming_tokens:
@@ -1805,6 +1820,17 @@ def __init__(self, *, reconnect_on_error: bool = True, **kwargs):
         TTSService.__init__(self, **kwargs)
         WebsocketService.__init__(self, reconnect_on_error=reconnect_on_error, **kwargs)
 
+    @property
+    def supports_processing_metrics(self) -> bool:
+        """Whether this service has a meaningful processing-time metric.
+
+        False: ``run_tts`` sends the text and returns, and audio arrives later
+        on the receive task, so there is no synthesis inside the measured
+        window. A subclass that instead waits for the server to signal the end
+        of synthesis before returning can override this back to True.
+        """
+        return False
+
     async def stop(self, frame: EndFrame):
         """Stop the websocket TTS service on a graceful end.
 

tests/test_tts_processing_metrics.py

@@ -0,0 +1,50 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Tests for which TTS services report a processing-time metric."""
+
+import pytest
+
+from pipecat.services.cartesia.tts import CartesiaHttpTTSService, CartesiaTTSService
+from pipecat.services.deepgram.tts import DeepgramHttpTTSService, DeepgramTTSService
+from pipecat.services.elevenlabs.tts import ElevenLabsHttpTTSService, ElevenLabsTTSService
+from pipecat.services.openai.tts import OpenAITTSService
+from pipecat.services.tts_service import TTSService, WebsocketTTSService
+
+# Services whose run_tts completes synthesis before returning.
+SYNCHRONOUS = [
+    CartesiaHttpTTSService,
+    DeepgramHttpTTSService,
+    ElevenLabsHttpTTSService,
+    OpenAITTSService,
+]
+
+# Services whose run_tts sends the text and returns, leaving audio to arrive
+# on the receive task.
+HANDS_OFF = [
+    CartesiaTTSService,
+    DeepgramTTSService,
+    ElevenLabsTTSService,
+]
+
+
+@pytest.mark.parametrize("cls", SYNCHRONOUS)
+def test_synchronous_services_report_processing_metrics(cls):
+    service = cls.__new__(cls)
+    assert service.supports_processing_metrics
+
+
+@pytest.mark.parametrize("cls", HANDS_OFF)
+def test_handoff_services_do_not_report_processing_metrics(cls):
+    service = cls.__new__(cls)
+    assert not service.supports_processing_metrics
+
+
+def test_default_is_to_report():
+    # A service that synthesizes inside run_tts is the common case, so the base
+    # class opts in and the handoff services opt out.
+    assert TTSService.supports_processing_metrics.fget(None)
+    assert not WebsocketTTSService.supports_processing_metrics.fget(None)