Code review feedback

examples/foundational/07-interruptible.py

@@ -57,7 +57,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     tts = CartesiaTTSService(
         api_key=os.getenv("CARTESIA_API_KEY"),
         voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
-        text_aggregation_mode=TextAggregationMode.TOKEN,
+        # Alternatively, you can use TextAggregationMode.TOKEN to stream tokens instead of
+        # sentencesfor faster response times.
+        # text_aggregation_mode=TextAggregationMode.TOKEN,
     )
 
     llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))

src/pipecat/processors/frame_processor.py

@@ -501,6 +501,7 @@ class FrameProcessor(BaseObject):
         """Stop all active metrics collection."""
         await self.stop_ttfb_metrics()
         await self.stop_processing_metrics()
+        await self.stop_text_aggregation_metrics()
 
     def create_task(self, coroutine: Coroutine, name: Optional[str] = None) -> asyncio.Task:
         """Create a new task managed by this processor.

src/pipecat/processors/metrics/frame_processor_metrics.py

@@ -44,6 +44,7 @@ class FrameProcessorMetrics(BaseObject):
         self._task_manager = None
         self._start_ttfb_time = 0
         self._start_processing_time = 0
+        self._start_text_aggregation_time = 0
         self._last_ttfb_time = 0
         self._should_report_ttfb = True
 
@@ -223,10 +224,7 @@ class FrameProcessorMetrics(BaseObject):
         Returns:
             MetricsFrame containing text aggregation time, or None if not measuring.
         """
-        if (
+        if self._start_text_aggregation_time == 0:
-            not hasattr(self, "_start_text_aggregation_time")
-            or self._start_text_aggregation_time == 0
-        ):
             return None
 
         value = time.time() - self._start_text_aggregation_time

src/pipecat/services/cartesia/tts.py

@@ -303,9 +303,11 @@ class CartesiaTTSService(AudioContextTTSService):
         """
         # By default, we aggregate sentences before sending to TTS. This adds
         # ~200-300ms of latency per sentence (waiting for the sentence-ending
-        # punctuation token from the LLM). Setting aggregate_sentences=False
+        # punctuation token from the LLM). Setting
-        # streams tokens directly, which reduces latency. Streaming quality
+        # text_aggregation_mode=TextAggregationMode.TOKEN streams tokens
-        # is good but less tested than sentence aggregation.
+        # directly, which reduces latency. Streaming quality is good but less
+        # tested than sentence aggregation.
+        # TODO: Consider making TOKEN the default for Cartesia in 1.0.
         #
         # We also don't want to automatically push LLM response text frames,
         # because the context aggregators will add them to the LLM context even
@@ -667,9 +669,7 @@ class CartesiaTTSService(AudioContextTTSService):
 
             try:
                 await self._get_websocket().send(msg)
-                # Usage metrics are aggregated at flush time when streaming tokens.
+                await self.start_tts_usage_metrics(text)
-                if not self._is_streaming_tokens:
-                    await self.start_tts_usage_metrics(text)
             except Exception as e:
                 yield ErrorFrame(error=f"Unknown error occurred: {e}")
                 yield TTSStoppedFrame(context_id=context_id)

src/pipecat/services/elevenlabs/tts.py

@@ -389,9 +389,10 @@ class ElevenLabsTTSService(AudioContextTTSService):
         """
         # By default, we aggregate sentences before sending to TTS. This adds
         # ~200-300ms of latency per sentence (waiting for the sentence-ending
-        # punctuation token from the LLM). Setting aggregate_sentences=False
+        # punctuation token from the LLM). Setting
-        # streams tokens directly. To use this mode, you must set auto_mode=False.
+        # text_aggregation_mode=TextAggregationMode.TOKEN streams tokens
-        # This eliminates aggregation time, but slows down ElevenLabs.
+        # directly. To use this mode, you must set auto_mode=False. This
+        # eliminates aggregation time, but slows down ElevenLabs.
         #
         # We also don't want to automatically push LLM response text frames,
         # because the context aggregators will add them to the LLM context even

src/pipecat/services/tts_service.py

@@ -247,8 +247,6 @@ class TTSService(AIService):
             text_aggregation_mode = TextAggregationMode.SENTENCE
 
         self._text_aggregation_mode: TextAggregationMode = text_aggregation_mode
-        # Keep for backward compat with subclasses that read self._aggregate_sentences
-        self._aggregate_sentences: bool = text_aggregation_mode != TextAggregationMode.TOKEN
         self._push_text_frames: bool = push_text_frames
         self._push_stop_frames: bool = push_stop_frames
         self._stop_frame_timeout_s: float = stop_frame_timeout_s
@@ -296,8 +294,8 @@ class TTSService(AIService):
 
         self._processing_text: bool = False
         self._tts_contexts: Dict[str, TTSContext] = {}
-        self._streaming_text_log: str = ""
+        self._streamed_text: str = ""
-        self._aggregation_logged: bool = False
+        self._text_aggregation_metrics_started: bool = False
 
         # Word timestamp state (active when supports_word_timestamps=True)
         self._supports_word_timestamps: bool = supports_word_timestamps
@@ -316,6 +314,35 @@ class TTSService(AIService):
         """Whether the service is streaming tokens directly without sentence aggregation."""
         return self._text_aggregation_mode == TextAggregationMode.TOKEN
 
+    async def start_tts_usage_metrics(self, text: str):
+        """Record TTS usage metrics.
+
+        When streaming tokens, usage metrics are aggregated and reported at
+        flush time instead of per token, so individual calls are skipped.
+
+        Args:
+            text: The text being processed by TTS.
+        """
+        if self._is_streaming_tokens:
+            return
+        await super().start_tts_usage_metrics(text)
+
+    async def start_text_aggregation_metrics(self):
+        """Start text aggregation metrics if not already started.
+
+        Only starts the metric once per LLM response. Skipped when streaming
+        tokens since per-token aggregation time is not meaningful.
+        """
+        if self._is_streaming_tokens or self._text_aggregation_metrics_started:
+            return
+        self._text_aggregation_metrics_started = True
+        await super().start_text_aggregation_metrics()
+
+    async def stop_text_aggregation_metrics(self):
+        """Stop text aggregation metrics and reset the started flag."""
+        self._text_aggregation_metrics_started = False
+        await super().stop_text_aggregation_metrics()
+
     @property
     def sample_rate(self) -> int:
         """Get the current sample rate for audio output.
@@ -574,9 +601,7 @@ class TTSService(AIService):
             and not isinstance(frame, InterimTranscriptionFrame)
             and not isinstance(frame, TranscriptionFrame)
         ):
-            if not self._is_streaming_tokens and not self._aggregation_logged:
+            await self.start_text_aggregation_metrics()
-                await self.start_text_aggregation_metrics()
-                self._aggregation_logged = True
             await self._process_text_frame(frame)
         elif isinstance(frame, InterruptionFrame):
             await self._handle_interruption(frame, direction)
@@ -592,18 +617,16 @@ class TTSService(AIService):
 
             # Flush any remaining text (including text waiting for lookahead)
             remaining = await self._text_aggregator.flush()
+            # Stop the aggregation metric (no-op if already stopped on first sentence).
+            await self.stop_text_aggregation_metrics()
             if remaining:
-                # If this is the first (and only) sentence, stop the aggregation metric.
-                await self.stop_text_aggregation_metrics()
                 await self._push_tts_frames(AggregatedTextFrame(remaining.text, remaining.type))
 
-            self._aggregation_logged = False
-
             # Log accumulated streamed text and emit aggregated usage metric.
-            if self._streaming_text_log:
+            if self._streamed_text:
-                logger.debug(f"{self}: Generating TTS [{self._streaming_text_log}]")
+                logger.debug(f"{self}: Generating TTS [{self._streamed_text}]")
-                await self.start_tts_usage_metrics(self._streaming_text_log)
+                await super().start_tts_usage_metrics(self._streamed_text)
-                self._streaming_text_log = ""
+                self._streamed_text = ""
 
             # Reset aggregator state
             self._processing_text = False
@@ -754,6 +777,8 @@ class TTSService(AIService):
             await filter.handle_interruption()
 
         self._llm_response_started = False
+        self._streamed_text = ""
+        self._text_aggregation_metrics_started = False
         if self._supports_word_timestamps:
             await self.reset_word_timestamps()
 
@@ -809,7 +834,7 @@ class TTSService(AIService):
 
         # Accumulate text for a single debug log at flush time when streaming tokens.
         if self._is_streaming_tokens:
-            self._streaming_text_log += text
+            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