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

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(
+    async def _maybe_push_llm_aggregation(self, frame: LLMTextFrame | LLMFullResponseEndFrame):
-        self, frame: LLMFullResponseStartFrame | 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:

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(
+        if self._params.bot_output_enabled:
-            data=RTVIBotOutputMessageData(
+            message = RTVIBotOutputMessage(
-                text=frame.text, spoken=isTTS, aggregated_by=frame.aggregated_by
+                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))

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}")
+            logger.trace(f"Pushing TTS frames for text: {text}, {aggregated_by}")
-            await self._push_tts_frames(text, should_speak, aggregated_by)
+            await self._push_tts_frames(text, aggregated_by)
 
-    async def _push_tts_frames(self, text: str, should_speak: bool, aggregated_by: str):
+    async def _push_tts_frames(self, text: str, aggregated_by: str):
-        if should_speak:
+        if aggregated_by in self._skip_aggregator_types:
-            # Remove leading newlines only
+            # If this type of aggregation should be skipped, we just push the text as
-            text = text.lstrip("\n")
+            # a basic AggregatedLLMTextFrame without sending it to TTS to speak.
-
-            # 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:
             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