From 82b9c4f0b6ee8b431bf57e2dcc93422a638b729c Mon Sep 17 00:00:00 2001 From: mattie ruth backman Date: Tue, 4 Nov 2025 09:13:45 -0500 Subject: [PATCH] various PR Review fixes: 1. Added support for turning off bot-output messages with the bot_output_enabled flag 2. Cleaned up logic and comments around TTSService:_push_tts_frames to hopefully make it easier to understand 3. Other minor cleanup --- .../aggregators/llm_response_universal.py | 17 ++-- src/pipecat/processors/frameworks/rtvi.py | 26 +++-- src/pipecat/services/tts_service.py | 95 ++++++++++--------- 3 files changed, 70 insertions(+), 68 deletions(-) diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index d8bf12bb7..309a13634 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -602,7 +602,7 @@ class LLMAssistantAggregator(LLMContextAggregator): self._llm_text_aggregator: BaseTextAggregator = ( self._params.llm_text_aggregator or SimpleTextAggregator() ) - self._skip_tts: Optional[bool] = None + self._skip_tts = None @property def has_function_calls_in_progress(self) -> bool: @@ -820,8 +820,8 @@ class LLMAssistantAggregator(LLMContextAggregator): async def _handle_llm_start(self, frame: LLMFullResponseStartFrame): self._started += 1 if self._skip_tts is None: + # initialize skip_tts on first start frame self._skip_tts = frame.skip_tts - await self._maybe_push_llm_aggregation(frame) async def _handle_llm_text(self, frame: LLMTextFrame): await self._handle_text(frame) @@ -832,22 +832,23 @@ class LLMAssistantAggregator(LLMContextAggregator): await self.push_aggregation() await self._maybe_push_llm_aggregation(frame) - async def _maybe_push_llm_aggregation( - self, frame: LLMFullResponseStartFrame | LLMTextFrame | LLMFullResponseEndFrame - ): + async def _maybe_push_llm_aggregation(self, frame: LLMTextFrame | LLMFullResponseEndFrame): aggregate = None should_reset_aggregator = False if self._skip_tts and not frame.skip_tts: - # if the skip_tts flag switches, to false, push the current aggregation + # When skip_tts transitions to False, we need to push any accumulated text. + # This ensures that any remaining text accumulated while TTS was skipped is + # sent out when TTS resumes, preventing loss of data and maintaining a smooth + # transition. aggregate = self._llm_text_aggregator.text should_reset_aggregator = True self._skip_tts = frame.skip_tts if self._skip_tts: - if self._skip_tts and isinstance(frame, LLMFullResponseEndFrame): + if isinstance(frame, LLMFullResponseEndFrame): # on end frame, always push the aggregation aggregate = self._llm_text_aggregator.text should_reset_aggregator = True - elif isinstance(frame, LLMTextFrame): + else: # This is an LLMTextFrame aggregate = await self._llm_text_aggregator.aggregate(frame.text) if not aggregate: diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index e5eff9567..4ee48ee7e 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -920,6 +920,7 @@ class RTVIObserverParams: Parameter `errors_enabled` is deprecated. Error messages are always enabled. Parameters: + bot_output_enabled: Indicates if bot output messages should be sent. bot_llm_enabled: Indicates if the bot's LLM messages should be sent. bot_tts_enabled: Indicates if the bot's TTS messages should be sent. bot_speaking_enabled: Indicates if the bot's started/stopped speaking messages should be sent. @@ -934,6 +935,7 @@ class RTVIObserverParams: audio_level_period_secs: How often audio levels should be sent if enabled. """ + bot_output_enabled: bool = True bot_llm_enabled: bool = True bot_tts_enabled: bool = True bot_speaking_enabled: bool = True @@ -1072,7 +1074,9 @@ class RTVIObserver(BaseObserver): await self.send_rtvi_message(RTVIBotTTSStartedMessage()) elif isinstance(frame, TTSStoppedFrame) and self._params.bot_tts_enabled: await self.send_rtvi_message(RTVIBotTTSStoppedMessage()) - elif isinstance(frame, AggregatedLLMTextFrame): + elif isinstance(frame, AggregatedLLMTextFrame) and ( + self._params.bot_output_enabled or self._params.bot_tts_enabled + ): if isinstance(frame, TTSTextFrame) and not isinstance(src, BaseOutputTransport): # This check is to make sure we handle the frame when it has gone # through the transport and has correct timing. @@ -1109,15 +1113,6 @@ class RTVIObserver(BaseObserver): if mark_as_seen: self._frames_seen.add(frame.id) - async def _push_bot_transcription(self): - """Push accumulated bot transcription as a message.""" - if len(self._bot_transcription) > 0: - message = RTVIBotTranscriptionMessage( - data=RTVITextMessageData(text=self._bot_transcription) - ) - await self.send_rtvi_message(message) - self._bot_transcription = "" - async def _handle_interruptions(self, frame: Frame): """Handle user speaking interruption frames.""" message = None @@ -1143,12 +1138,13 @@ class RTVIObserver(BaseObserver): async def _handle_aggregated_llm_text(self, frame: AggregatedLLMTextFrame): """Handle aggregated LLM text output frames.""" isTTS = isinstance(frame, TTSTextFrame) - message = RTVIBotOutputMessage( - data=RTVIBotOutputMessageData( - text=frame.text, spoken=isTTS, aggregated_by=frame.aggregated_by + if self._params.bot_output_enabled: + message = RTVIBotOutputMessage( + data=RTVIBotOutputMessageData( + text=frame.text, spoken=isTTS, aggregated_by=frame.aggregated_by + ) ) - ) - await self.send_rtvi_message(message) + await self.send_rtvi_message(message) if isTTS and self._params.bot_tts_enabled: tts_message = RTVIBotTTSTextMessage(data=RTVITextMessageData(text=frame.text)) diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index 88d0c8847..0079f7708 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -378,7 +378,6 @@ class TTSService(AIService): self._processing_text = False await self._push_tts_frames( text=aggregate.text, - should_speak=aggregate.type not in self._skip_aggregator_types, aggregated_by=aggregate.type, ) if isinstance(frame, LLMFullResponseEndFrame): @@ -389,7 +388,7 @@ class TTSService(AIService): elif isinstance(frame, TTSSpeakFrame): # Store if we were processing text or not so we can set it back. processing_text = self._processing_text - await self._push_tts_frames(frame.text, should_speak=True, aggregated_by="word") + await self._push_tts_frames(frame.text, aggregated_by="sentence") # We pause processing incoming frames because we are sending data to # the TTS. We pause to avoid audio overlapping. await self._maybe_pause_frame_processing() @@ -481,60 +480,66 @@ class TTSService(AIService): text: Optional[str] = None if not self._aggregate_sentences: text = frame.text - should_speak = True aggregated_by = "token" else: aggregate = await self._text_aggregator.aggregate(frame.text) if aggregate: text = aggregate.text - should_speak = aggregate.type not in self._skip_aggregator_types aggregated_by = aggregate.type if text: - logger.trace(f"Pushing TTS frames for text: {text}, {should_speak}, {aggregated_by}") - await self._push_tts_frames(text, should_speak, aggregated_by) + logger.trace(f"Pushing TTS frames for text: {text}, {aggregated_by}") + await self._push_tts_frames(text, aggregated_by) - async def _push_tts_frames(self, text: str, should_speak: bool, aggregated_by: str): - if should_speak: - # Remove leading newlines only - text = text.lstrip("\n") - - # Don't send only whitespace. This causes problems for some TTS models. But also don't - # strip all whitespace, as whitespace can influence prosody. - if not text.strip(): - return - - # This is just a flag that indicates if we sent something to the TTS - # service. It will be cleared if we sent text because of a TTSSpeakFrame - # or when we received an LLMFullResponseEndFrame - self._processing_text = True - - await self.start_processing_metrics() - - # Process all filter. - for filter in self._text_filters: - await filter.reset_interruption() - text = await filter.filter(text) - - if text: - if not self._push_text_frames: - # If we are not pushing text frames, we send a TTSTextFrame - # before the audio so downstream processors know what text - # is being spoken. Here, we assume this flag is used when the TTS - # provider supports word timestamps and the TTSTextFrames will be - # generated in the word_task_handler. - frame = AggregatedLLMTextFrame(text, aggregated_by=aggregated_by) - frame.append_to_context = False - await self.push_frame(frame) - await self.process_generator(self.run_tts(text)) - - await self.stop_processing_metrics() - - if not should_speak: + async def _push_tts_frames(self, text: str, aggregated_by: str): + if aggregated_by in self._skip_aggregator_types: + # If this type of aggregation should be skipped, we just push the text as + # a basic AggregatedLLMTextFrame without sending it to TTS to speak. await self.push_frame(AggregatedLLMTextFrame(text, aggregated_by=aggregated_by)) - elif self._push_text_frames: + return + + # Remove leading newlines only + text = text.lstrip("\n") + + # Don't send only whitespace. This causes problems for some TTS models. But also don't + # strip all whitespace, as whitespace can influence prosody. + if not text.strip(): + return + + # This is just a flag that indicates if we sent something to the TTS + # service. It will be cleared if we sent text because of a TTSSpeakFrame + # or when we received an LLMFullResponseEndFrame + self._processing_text = True + + await self.start_processing_metrics() + + # Process all filter. + for filter in self._text_filters: + await filter.reset_interruption() + text = await filter.filter(text) + + if text: + if not self._push_text_frames: + # In a typical pipeline, there is an assistant context aggregator + # that listens for TTSTextFrames to add spoken text to the context. + # If the TTS service supports word timestamps, then _push_text_frames + # is set to False and these are sent word by word as part of the + # _words_task_handler in the WordTTSService subclass. However, to + # support use cases where an observer may want the full text before + # the audio is generated, we send an AggregatedLLMTextFrame here, but + # we set append_to_context to False so it does not cause duplication + # in the context. This is primarily used by the RTVIObserver to + # generate a complete bot-output. + frame = AggregatedLLMTextFrame(text, aggregated_by=aggregated_by) + frame.append_to_context = False + await self.push_frame(frame) + await self.process_generator(self.run_tts(text)) + + await self.stop_processing_metrics() + + if self._push_text_frames: # In the case where the TTS service does not support word timestamps, - # we send the original text after the audio. This way, if we are + # we send the full aggregated text after the audio. This way, if we are # interrupted, the text is not added to the assistant context. frame = TTSTextFrame(text, aggregated_by=aggregated_by) frame.includes_inter_frame_spaces = self.includes_inter_frame_spaces