From 3ae173520ee739c6a3e3948b7a16fedbab1d11be Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 26 Feb 2026 08:59:38 -0500 Subject: [PATCH] Code review feedback --- examples/foundational/07-interruptible.py | 4 +- src/pipecat/processors/frame_processor.py | 1 + .../metrics/frame_processor_metrics.py | 6 +- src/pipecat/services/cartesia/tts.py | 12 ++-- src/pipecat/services/elevenlabs/tts.py | 7 ++- src/pipecat/services/tts_service.py | 57 +++++++++++++------ 6 files changed, 57 insertions(+), 30 deletions(-) diff --git a/examples/foundational/07-interruptible.py b/examples/foundational/07-interruptible.py index e47d2c811..074e091ea 100644 --- a/examples/foundational/07-interruptible.py +++ b/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")) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 37e8dc10d..baa52cc70 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/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. diff --git a/src/pipecat/processors/metrics/frame_processor_metrics.py b/src/pipecat/processors/metrics/frame_processor_metrics.py index cb5bc8a42..7a52895a2 100644 --- a/src/pipecat/processors/metrics/frame_processor_metrics.py +++ b/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 ( - not hasattr(self, "_start_text_aggregation_time") - or self._start_text_aggregation_time == 0 - ): + if self._start_text_aggregation_time == 0: return None value = time.time() - self._start_text_aggregation_time diff --git a/src/pipecat/services/cartesia/tts.py b/src/pipecat/services/cartesia/tts.py index 0749af062..2e637c339 100644 --- a/src/pipecat/services/cartesia/tts.py +++ b/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 - # streams tokens directly, which reduces latency. Streaming quality - # is good but less tested than sentence aggregation. + # punctuation token from the LLM). Setting + # text_aggregation_mode=TextAggregationMode.TOKEN streams tokens + # 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. - if not self._is_streaming_tokens: - await self.start_tts_usage_metrics(text) + 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) diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py index dcfdebc2f..1811ed971 100644 --- a/src/pipecat/services/elevenlabs/tts.py +++ b/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 - # streams tokens directly. To use this mode, you must set auto_mode=False. - # This eliminates aggregation time, but slows down ElevenLabs. + # punctuation token from the LLM). Setting + # text_aggregation_mode=TextAggregationMode.TOKEN streams tokens + # 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 diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index 8c61f225d..c6d2672d6 100644 --- a/src/pipecat/services/tts_service.py +++ b/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._aggregation_logged: bool = False + self._streamed_text: str = "" + 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() - self._aggregation_logged = True + await self.start_text_aggregation_metrics() 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: - logger.debug(f"{self}: Generating TTS [{self._streaming_text_log}]") - await self.start_tts_usage_metrics(self._streaming_text_log) - self._streaming_text_log = "" + if self._streamed_text: + logger.debug(f"{self}: Generating TTS [{self._streamed_text}]") + await super().start_tts_usage_metrics(self._streamed_text) + 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