Introduce AggregatedLLMTextFrame to allow a separation of TTSTextFrame, indicating a spoken frame vs other aggregated, non-spoken frames

src/pipecat/frames/frames.py

@@ -356,11 +356,15 @@ class LLMTextFrame(TextFrame):
 
 
 @dataclass
-class TTSTextFrame(TextFrame):
+class AggregatedLLMTextFrame(TextFrame):
     """Text frame generated by Text-to-Speech services."""
 
     aggregated_by: Literal["sentence", "word"] | str
-    spoken: Optional[bool] = True  # Whether this text has been spoken by TTS
+
+
+@dataclass
+class TTSTextFrame(AggregatedLLMTextFrame):
+    """Text frame generated by Text-to-Speech services."""
 
     pass
 

src/pipecat/processors/frameworks/rtvi.py

@@ -32,6 +32,7 @@ from pydantic import BaseModel, Field, PrivateAttr, ValidationError
 
 from pipecat.audio.utils import calculate_audio_volume
 from pipecat.frames.frames import (
+    AggregatedLLMTextFrame,
     BotStartedSpeakingFrame,
     BotStoppedSpeakingFrame,
     CancelFrame,
@@ -1023,6 +1024,16 @@ class RTVIObserver(BaseObserver):
         if self._rtvi:
             await self._rtvi.push_transport_message(model, exclude_none)
 
+    async def send_aggregated_llm_text(self, text: str, aggregated_by: Optional[str] = None):
+        """Send aggregated LLM text as a bot output message.
+
+        Args:
+            text: The aggregated text to send.
+            aggregated_by: The method of aggregation (e.g., "word", "sentence").
+        """
+        if self._rtvi:
+            await self._rtvi.push_aggregated_llm_text(text, aggregated_by)
+
     async def on_push_frame(self, data: FramePushed):
         """Process a frame being pushed through the pipeline.
 
@@ -1073,11 +1084,13 @@ 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, TTSTextFrame) and self._params.bot_tts_enabled:
+        elif isinstance(frame, AggregatedLLMTextFrame):
-            if isinstance(src, BaseOutputTransport):
+            if isinstance(frame, TTSTextFrame) and not isinstance(src, BaseOutputTransport):
-                await self._handle_tts_text_frame(frame)
+                # This check is to make sure we handle the frame when it has gone
-            else:
+                # through the transport and has correct timing.
                 mark_as_seen = False
+            else:
+                await self._handle_aggregated_llm_text(frame)
         elif isinstance(frame, MetricsFrame) and self._params.metrics_enabled:
             await self._handle_metrics(frame)
         elif isinstance(frame, RTVIServerMessageFrame):
@@ -1139,19 +1152,20 @@ class RTVIObserver(BaseObserver):
         if message:
             await self.send_rtvi_message(message)
 
-    async def _handle_tts_text_frame(self, frame: TTSTextFrame):
+    async def _handle_aggregated_llm_text(self, frame: AggregatedLLMTextFrame):
-        """Handle TTS text output frames."""
+        """Handle aggregated LLM text output frames."""
-        # send the tts-text message
+        isTTS = isinstance(frame, TTSTextFrame)
-        message = RTVIBotTTSTextMessage(data=RTVITextMessageData(text=frame.text))
-        await self.send_rtvi_message(message)
-        # send the bot-output message
         message = RTVIBotOutputMessage(
             data=RTVIBotOutputMessageData(
-                text=frame.text, spoken=frame.spoken, aggregated_by=frame.aggregated_by
+                text=frame.text, spoken=isTTS, aggregated_by=frame.aggregated_by
             )
         )
         await self.send_rtvi_message(message)
 
+        if isTTS and self._params.bot_tts_enabled:
+            tts_message = RTVIBotTTSTextMessage(data=RTVITextMessageData(text=frame.text))
+            await self.send_rtvi_message(tts_message)
+
     async def _handle_llm_text_frame(self, frame: LLMTextFrame):
         """Handle LLM text output frames."""
         message = RTVIBotLLMTextMessage(data=RTVITextMessageData(text=frame.text))
@@ -1161,41 +1175,26 @@ class RTVIObserver(BaseObserver):
         if self._skip_tts is None:
             self._skip_tts = frame.skip_tts
 
-        messages = []
+        orig_text = self._bot_transcription
-        should_reset_transcription = False
         self._bot_transcription += frame.text
 
-        if not frame.skip_tts and self._skip_tts:
-            # We just switched from skipping TTS to not skipping TTS.
-            # Send and reset any existing transcription.
-            if len(self._bot_transcription) > 0:
-                message.append(
-                    RTVIBotOutputMessage(
-                        data=RTVIBotOutputMessageData(
-                            text=self._bot_transcription, spoken=False, aggregated_by="sentence"
-                        )
-                    )
-                )
-                should_reset_transcription = True
-
         if match_endofsentence(self._bot_transcription) and len(self._bot_transcription) > 0:
-            messages.append(
+            # TODO: Remove this message when we fully deprecate bot-transcription messages.
+            await self.send_rtvi_message(
                 RTVIBotTranscriptionMessage(data=RTVITextMessageData(text=self._bot_transcription))
             )
             if frame.skip_tts:
-                messages.append(
+                await self.send_aggregated_llm_text(
-                    RTVIBotOutputMessage(
+                    text=self._bot_transcription, aggregated_by="sentence"
-                        data=RTVIBotOutputMessageData(
-                            text=self._bot_transcription, spoken=False, aggregated_by="sentence"
-                        )
-                    )
                 )
-            should_reset_transcription = True
-
-        for msg in messages:
-            await self.send_rtvi_message(msg)
-        if should_reset_transcription:
             self._bot_transcription = ""
+        elif not frame.skip_tts and self._skip_tts:
+            # We just switched from skipping TTS to not skipping TTS.
+            # Send any dangling transcription.
+            if len(orig_text) > 0:
+                await self.send_aggregated_llm_text(text=orig_text, aggregated_by="sentence")
+                self._bot_transcription = frame.text
+        self._skip_tts = frame.skip_tts
 
     async def _handle_user_transcriptions(self, frame: Frame):
         """Handle user transcription frames."""
@@ -1321,7 +1320,7 @@ class RTVIProcessor(FrameProcessor):
         # Default to 0.3.0 which is the last version before actually having a
         # "client-version".
         self._client_version = [0, 3, 0]
-        self._skip_tts: bool = False  # Keep in sync with llm_service.py
+        self._llm_skip_tts: bool = False  # Keep in sync with llm_service.py's configuration.
 
         self._registered_actions: Dict[str, RTVIAction] = {}
         self._registered_services: Dict[str, RTVIService] = {}
@@ -1429,6 +1428,12 @@ class RTVIProcessor(FrameProcessor):
         )
         await self.push_frame(frame)
 
+    async def push_aggregated_llm_text(self, text: str, aggregated_by: Optional[str] = None):
+        """Push an aggregated LLM text frame."""
+        frame = AggregatedLLMTextFrame(text=text, aggregated_by=aggregated_by)
+        frame.skip_tts = True
+        await self.push_frame(frame)
+
     async def handle_message(self, message: RTVIMessage):
         """Handle an incoming RTVI message.
 
@@ -1514,7 +1519,7 @@ class RTVIProcessor(FrameProcessor):
         elif isinstance(frame, RTVIActionFrame):
             await self._action_queue.put(frame)
         elif isinstance(frame, LLMConfigureOutputFrame):
-            self._skip_tts = frame.skip_tts
+            self._llm_skip_tts = frame.skip_tts
             await self.push_frame(frame, direction)
         # Other frames
         else:
@@ -1770,9 +1775,9 @@ class RTVIProcessor(FrameProcessor):
         opts = data.options if data.options is not None else RTVISendTextOptions()
         if opts.run_immediately:
             await self.interrupt_bot()
-        cur_skip_tts = self._skip_tts
+        cur_llm_skip_tts = self._llm_skip_tts
         should_skip_tts = not opts.audio_response
-        toggle_skip_tts = cur_skip_tts != should_skip_tts
+        toggle_skip_tts = cur_llm_skip_tts != should_skip_tts
         if toggle_skip_tts:
             output_frame = LLMConfigureOutputFrame(skip_tts=should_skip_tts)
             await self.push_frame(output_frame)
@@ -1782,7 +1787,7 @@ class RTVIProcessor(FrameProcessor):
         )
         await self.push_frame(text_frame)
         if toggle_skip_tts:
-            output_frame = LLMConfigureOutputFrame(skip_tts=cur_skip_tts)
+            output_frame = LLMConfigureOutputFrame(skip_tts=cur_llm_skip_tts)
             await self.push_frame(output_frame)
 
     async def _handle_update_context(self, data: RTVIAppendToContextData):

src/pipecat/services/aws/nova_sonic/llm.py

@@ -1027,7 +1027,7 @@ class AWSNovaSonicLLMService(LLMService):
         logger.debug(f"Assistant response text added: {text}")
 
         # Report the text of the assistant response.
-        frame = TTSTextFrame(text, aggregated_by="sentence", spoken=True)
+        frame = TTSTextFrame(text, aggregated_by="sentence")
         frame.includes_inter_frame_spaces = True
         await self.push_frame(frame)
 
@@ -1062,9 +1062,7 @@ class AWSNovaSonicLLMService(LLMService):
                 # TTSTextFrame would be ignored otherwise (the interruption frame
                 # would have cleared the assistant aggregator state).
                 await self.push_frame(LLMFullResponseStartFrame())
-                frame = TTSTextFrame(
+                frame = TTSTextFrame(self._assistant_text_buffer, aggregated_by="sentence")
-                    self._assistant_text_buffer, aggregated_by="sentence", spoken=True
-                )
                 frame.includes_inter_frame_spaces = True
                 await self.push_frame(frame)
             self._may_need_repush_assistant_text = False

src/pipecat/services/openai_realtime_beta/openai.py

@@ -652,7 +652,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
     async def _handle_evt_audio_transcript_delta(self, evt):
         if evt.delta:
             await self.push_frame(LLMTextFrame(evt.delta))
-            await self.push_frame(TTSTextFrame(evt.delta, aggregated_by="sentence", spoken=True))
+            await self.push_frame(TTSTextFrame(evt.delta, aggregated_by="sentence"))
 
     async def _handle_evt_speech_started(self, evt):
         await self._truncate_current_audio_response()

src/pipecat/services/tts_service.py

@@ -23,6 +23,7 @@ from typing import (
 from loguru import logger
 
 from pipecat.frames.frames import (
+    AggregatedLLMTextFrame,
     BotStartedSpeakingFrame,
     BotStoppedSpeakingFrame,
     CancelFrame,
@@ -516,15 +517,24 @@ class TTSService(AIService):
                 text = await filter.filter(text)
 
             if text:
-                await self.push_frame(TTSTextFrame(text, spoken=True, aggregated_by=aggregated_by))
+                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.
+                    await self.push_frame(AggregatedLLMTextFrame(text, aggregated_by=aggregated_by))
                 await self.process_generator(self.run_tts(text))
 
             await self.stop_processing_metrics()
 
-        if self._push_text_frames or not should_speak:
+        if not should_speak:
-            # We send the original text after the audio. This way, if we are
+            await self.push_frame(AggregatedLLMTextFrame(text, aggregated_by=aggregated_by))
+        elif 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
             # interrupted, the text is not added to the assistant context.
-            frame = TTSTextFrame(text, spoken=should_speak, aggregated_by=aggregated_by)
+            frame = TTSTextFrame(text, aggregated_by=aggregated_by)
             frame.includes_inter_frame_spaces = self.includes_inter_frame_spaces
             await self.push_frame(frame)
 
@@ -652,7 +662,7 @@ class WordTTSService(TTSService):
                 frame = TTSStoppedFrame()
                 frame.pts = last_pts
             else:
-                frame = TTSTextFrame(word, spoken=True, aggregated_by="word")
+                frame = TTSTextFrame(word, aggregated_by="word")
                 frame.pts = self._initial_word_timestamp + timestamp
             if frame:
                 last_pts = frame.pts

src/pipecat/tests/utils.py

@@ -203,8 +203,16 @@ async def run_test(
             if not isinstance(frame, EndFrame) or not send_end_frame:
                 received_down_frames.append(frame)
 
-        print("received DOWN frames =", received_down_frames)
+        down_frames_printed = "["
-        print("expected DOWN frames =", expected_down_frames)
+        for frame in received_down_frames:
+            down_frames_printed += f"{frame.__class__.__name__}, "
+        down_frames_printed += "]"
+        expected_frames_printed = "["
+        for frame in expected_down_frames:
+            expected_frames_printed += f"{frame.__name__}, "
+        expected_frames_printed += "]"
+        print("received DOWN frames =", down_frames_printed)
+        print("expected DOWN frames =", expected_frames_printed)
 
         assert len(received_down_frames) == len(expected_down_frames)
 

src/pipecat/utils/text/pattern_pair_aggregator.py

@@ -111,7 +111,8 @@ class PatternPairAggregator(BaseTextAggregator):
             action: What to do when a complete pattern is matched:
                     - "remove": Remove the matched pattern from the text.
                     - "keep": Keep the matched pattern in the text and treat it as
-                                normal text.
+                              normal text. This allows you to register handlers for
+                              the pattern without affecting the aggregation logic.
                     - "aggregate": Return the matched pattern as a separate
                                    aggregation object.
 
@@ -259,12 +260,14 @@ class PatternPairAggregator(BaseTextAggregator):
 
         #
         if len(patterns) > 0:
+            print(f"Found patterns: {[str(p) for p in patterns]}")
             if len(patterns) > 1:
                 logger.warning(
                     f"Multiple patterns matched: {[p.pattern_id for p in patterns]}. Only the first pattern will be returned."
                 )
             # If the pattern found is set to be aggregated, return it
             action = self._patterns[patterns[0].pattern_id].get("action", "remove")
+            print(f"Pattern action: {action}")
             if action == "aggregate":
                 self._text = ""
                 print(f"Returning pattern: {patterns[0]}")

tests/test_piper_tts.py

@@ -13,6 +13,7 @@ import pytest
 from aiohttp import web
 
 from pipecat.frames.frames import (
+    AggregatedLLMTextFrame,
     ErrorFrame,
     TTSAudioRawFrame,
     TTSSpeakFrame,
@@ -74,7 +75,6 @@ async def test_run_piper_tts_success(aiohttp_client):
         ]
 
         expected_returned_frames = [
-            TTSTextFrame,
             TTSStartedFrame,
             TTSAudioRawFrame,
             TTSAudioRawFrame,
@@ -122,7 +122,7 @@ async def test_run_piper_tts_error(aiohttp_client):
             TTSSpeakFrame(text="Error case."),
         ]
 
-        expected_down_frames = [TTSTextFrame, TTSStoppedFrame, TTSTextFrame]
+        expected_down_frames = [TTSStoppedFrame, TTSTextFrame]
 
         expected_up_frames = [ErrorFrame]