Introduced a new AggregatedTextFrame Frame type that TTSTextFrame inherits from

This frame introduces an `aggregated_by` field to describe the type of text included in the frame and allows unspoken groupings of text to be pushed through the pipeline and treated similar to TTSTextFrames.
2025-11-17 14:55:04 -05:00
parent 0e820a01b9
commit 7a4372a909
8 changed files with 88 additions and 34 deletions
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -39,6 +39,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

 - Added word-level timestamps support to Hume TTS service

+- Introduced a new `AggregatedTextFrame` type to support passing text along with an
+  `aggregated_by` field to describe the type of text included. `TTSTextFrame`s now
+  inherit from `AggregatedTextFrame`. With this inheritance, an observer can watch for
+  `AggregatedTextFrame`s to accumlate the perceived output and determine whether or not
+  the text was spoken based on if that frame is also a `TTSTextFrame`.
+
+  With this frame, the llm token stream can be transformed into custom composable
+  chunks, allowing for aggregation outside the TTS service. This makes it possible to
+  listen for or handle those aggregations and sets the stage for doing things like
+  composing a best effort of the perceived llm output in a more digestable form and
+  to do so whether or not it is processed by a TTS or if even a TTS exists.
+
 ### Changed

 - ⚠️ Breaking change: `LLMContext.create_image_message()`,
--- a/src/pipecat/frames/frames.py
+++ b/src/pipecat/frames/frames.py
@@ -12,6 +12,7 @@ and LLM processing.
 """

 from dataclasses import dataclass, field
+from enum import Enum
 from typing import (
    TYPE_CHECKING,
    Any,
@@ -361,8 +362,32 @@ class LLMTextFrame(TextFrame):
        self.includes_inter_frame_spaces = True


+class AggregationType(str, Enum):
+    """Built-in aggregation strings."""
+
+    SENTENCE = "sentence"
+    WORD = "word"
+
+    def __str__(self):
+        return self.value
+
+
@dataclass
-class TTSTextFrame(TextFrame):
+class AggregatedTextFrame(TextFrame):
+    """Text frame representing an aggregation of TextFrames.
+
+    This frame contains multiple TextFrames aggregated together for processing
+    or output along with a field to indicate how they are aggregated.
+
+    Parameters:
+        aggregated_by: Method used to aggregate the text frames.
+    """
+
+    aggregated_by: AggregationType | str
+
+
+@dataclass
+class TTSTextFrame(AggregatedTextFrame):
    """Text frame generated by Text-to-Speech services."""

    pass
--- a/src/pipecat/services/aws/nova_sonic/llm.py
+++ b/src/pipecat/services/aws/nova_sonic/llm.py
@@ -27,6 +27,7 @@ from pydantic import BaseModel, Field
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
 from pipecat.adapters.services.aws_nova_sonic_adapter import AWSNovaSonicLLMAdapter, Role
 from pipecat.frames.frames import (
+    AggregationType,
    BotStoppedSpeakingFrame,
    CancelFrame,
    EndFrame,
@@ -1027,7 +1028,7 @@ class AWSNovaSonicLLMService(LLMService):
        logger.debug(f"Assistant response text added: {text}")

        # Report the text of the assistant response.
-        frame = TTSTextFrame(text)
+        frame = TTSTextFrame(text, aggregated_by=AggregationType.SENTENCE)
        frame.includes_inter_frame_spaces = True
        await self.push_frame(frame)

@@ -1062,7 +1063,9 @@ 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(self._assistant_text_buffer)
+                frame = TTSTextFrame(
+                    self._assistant_text_buffer, aggregated_by=AggregationType.SENTENCE
+                )
                frame.includes_inter_frame_spaces = True
                await self.push_frame(frame)
            self._may_need_repush_assistant_text = False
--- a/src/pipecat/services/google/gemini_live/llm.py
+++ b/src/pipecat/services/google/gemini_live/llm.py
@@ -27,6 +27,7 @@ from pydantic import BaseModel, Field
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
 from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter
 from pipecat.frames.frames import (
+    AggregationType,
    BotStartedSpeakingFrame,
    BotStoppedSpeakingFrame,
    CancelFrame,
@@ -1644,7 +1645,7 @@ class GeminiLiveLLMService(LLMService):
            await self.push_frame(TTSStartedFrame())
            await self.push_frame(LLMFullResponseStartFrame())

-        frame = TTSTextFrame(text=text)
+        frame = TTSTextFrame(text=text, aggregated_by=AggregationType.SENTENCE)
        # Gemini Live text already includes any necessary inter-chunk spaces
        frame.includes_inter_frame_spaces = True

--- a/src/pipecat/services/openai/realtime/llm.py
+++ b/src/pipecat/services/openai/realtime/llm.py
@@ -19,6 +19,7 @@ from pipecat.adapters.services.open_ai_realtime_adapter import (
    OpenAIRealtimeLLMAdapter,
 )
 from pipecat.frames.frames import (
+    AggregationType,
    BotStoppedSpeakingFrame,
    CancelFrame,
    EndFrame,
@@ -684,7 +685,7 @@ class OpenAIRealtimeLLMService(LLMService):
        # We receive audio transcript deltas (as opposed to text deltas) when
        # the output modality is "audio" (the default)
        if evt.delta:
-            frame = TTSTextFrame(evt.delta)
+            frame = TTSTextFrame(evt.delta, aggregated_by=AggregationType.SENTENCE)
            # OpenAI Realtime text already includes any necessary inter-chunk spaces
            frame.includes_inter_frame_spaces = True
            await self.push_frame(frame)
--- a/src/pipecat/services/openai_realtime_beta/openai.py
+++ b/src/pipecat/services/openai_realtime_beta/openai.py
@@ -17,6 +17,7 @@ from loguru import logger

 from pipecat.adapters.services.open_ai_realtime_adapter import OpenAIRealtimeLLMAdapter
 from pipecat.frames.frames import (
+    AggregationType,
    BotStoppedSpeakingFrame,
    CancelFrame,
    EndFrame,
@@ -652,7 +653,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))
+            await self.push_frame(TTSTextFrame(evt.delta, aggregated_by=AggregationType.SENTENCE))

    async def _handle_evt_speech_started(self, evt):
        await self._truncate_current_audio_response()
--- a/src/pipecat/services/tts_service.py
+++ b/src/pipecat/services/tts_service.py
@@ -23,6 +23,8 @@ from typing import (
 from loguru import logger

 from pipecat.frames.frames import (
+    AggregatedTextFrame,
+    AggregationType,
    BotStartedSpeakingFrame,
    BotStoppedSpeakingFrame,
    CancelFrame,
@@ -358,7 +360,9 @@ class TTSService(AIService):
            self._processing_text = False

            if pending_aggregation.text:
-                await self._push_tts_frames(pending_aggregation.text)
+                await self._push_tts_frames(
+                    AggregatedTextFrame(pending_aggregation.text, pending_aggregation.type)
+                )
            if isinstance(frame, LLMFullResponseEndFrame):
                if self._push_text_frames:
                    await self.push_frame(frame, direction)
@@ -368,7 +372,7 @@ class TTSService(AIService):
            # Store if we were processing text or not so we can set it back.
            processing_text = self._processing_text
            # Assumption: text in TTSSpeakFrame does not include inter-frame spaces
-            await self._push_tts_frames(frame.text)
+            await self._push_tts_frames(AggregatedTextFrame(frame.text, AggregationType.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()
@@ -462,18 +466,24 @@ class TTSService(AIService):
        if not self._aggregate_sentences:
            text = frame.text
            includes_inter_frame_spaces = frame.includes_inter_frame_spaces
+            aggregated_by = "token"
        else:
-            aggregation = await self._text_aggregator.aggregate(frame.text)
-            text = aggregation.text
+            aggregate = await self._text_aggregator.aggregate(frame.text)
+            if aggregate:
+                text = aggregate.text
+                aggregated_by = aggregate.type

        if text:
-            await self._push_tts_frames(text, includes_inter_frame_spaces)
+            logger.trace(f"Pushing TTS frames for text: {text}, {aggregated_by}")
+            await self._push_tts_frames(
+                AggregatedTextFrame(text, aggregated_by), includes_inter_frame_spaces
+            )

    async def _push_tts_frames(
-        self, text: str, includes_inter_frame_spaces: Optional[bool] = False
+        self, src_frame: AggregatedTextFrame, includes_inter_frame_spaces: Optional[bool] = False
    ):
        # Remove leading newlines only
-        text = text.lstrip("\n")
+        text = src_frame.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.
@@ -500,7 +510,7 @@ class TTSService(AIService):
        if self._push_text_frames:
            # 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)
+            frame = TTSTextFrame(text, aggregated_by=src_frame.aggregated_by)
            frame.includes_inter_frame_spaces = includes_inter_frame_spaces
            await self.push_frame(frame)

@@ -630,7 +640,7 @@ class WordTTSService(TTSService):
            else:
                # Assumption: word-by-word text frames don't include spaces, so
                # we can rely on the default includes_inter_frame_spaces=False
-                frame = TTSTextFrame(word)
+                frame = TTSTextFrame(word, aggregated_by=AggregationType.WORD)
                frame.pts = self._initial_word_timestamp + timestamp
            if frame:
                last_pts = frame.pts
--- a/tests/test_transcript_processor.py
+++ b/tests/test_transcript_processor.py
@@ -11,6 +11,7 @@ from datetime import datetime, timezone
 from typing import List, Tuple, cast

 from pipecat.frames.frames import (
+    AggregationType,
    BotStartedSpeakingFrame,
    BotStoppedSpeakingFrame,
    CancelFrame,
@@ -130,11 +131,11 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
        frames_to_send = [
            BotStartedSpeakingFrame(),
            SleepFrame(),  # Wait for StartedSpeaking to process
-            TTSTextFrame(text="Hello"),
-            TTSTextFrame(text="world!"),
-            TTSTextFrame(text="How"),
-            TTSTextFrame(text="are"),
-            TTSTextFrame(text="you?"),
+            TTSTextFrame(text="Hello", aggregated_by=AggregationType.WORD),
+            TTSTextFrame(text="world!", aggregated_by=AggregationType.WORD),
+            TTSTextFrame(text="How", aggregated_by=AggregationType.WORD),
+            TTSTextFrame(text="are", aggregated_by=AggregationType.WORD),
+            TTSTextFrame(text="you?", aggregated_by=AggregationType.WORD),
            SleepFrame(),  # Wait for text frames to queue
            BotStoppedSpeakingFrame(),
        ]
@@ -195,9 +196,9 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
        frames_to_send = [
            BotStartedSpeakingFrame(),
            SleepFrame(),
-            TTSTextFrame(text=""),  # Empty text
-            TTSTextFrame(text="   "),  # Just whitespace
-            TTSTextFrame(text="\n"),  # Just newline
+            TTSTextFrame(text="", aggregated_by=AggregationType.WORD),  # Empty text
+            TTSTextFrame(text="   ", aggregated_by=AggregationType.WORD),  # Just whitespace
+            TTSTextFrame(text="\n", aggregated_by=AggregationType.WORD),  # Just newline
            BotStoppedSpeakingFrame(),
            # Pipeline ends here; run_test will automatically send EndFrame
        ]
@@ -235,14 +236,14 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
        frames_to_send = [
            BotStartedSpeakingFrame(),
            SleepFrame(),
-            TTSTextFrame(text="Hello"),
-            TTSTextFrame(text="world!"),
+            TTSTextFrame(text="Hello", aggregated_by=AggregationType.WORD),
+            TTSTextFrame(text="world!", aggregated_by=AggregationType.WORD),
            SleepFrame(),
            InterruptionFrame(),  # User interrupts here
            SleepFrame(),
            BotStartedSpeakingFrame(),
-            TTSTextFrame(text="New"),
-            TTSTextFrame(text="response"),
+            TTSTextFrame(text="New", aggregated_by=AggregationType.WORD),
+            TTSTextFrame(text="response", aggregated_by=AggregationType.WORD),
            SleepFrame(),
            BotStoppedSpeakingFrame(),
        ]
@@ -299,8 +300,8 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
        frames_to_send = [
            BotStartedSpeakingFrame(),
            SleepFrame(),
-            TTSTextFrame(text="Hello"),
-            TTSTextFrame(text="world"),
+            TTSTextFrame(text="Hello", aggregated_by=AggregationType.WORD),
+            TTSTextFrame(text="world", aggregated_by=AggregationType.WORD),
            # Pipeline ends here; run_test will automatically send EndFrame
        ]

@@ -338,8 +339,8 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
        frames_to_send = [
            BotStartedSpeakingFrame(),
            SleepFrame(),
-            TTSTextFrame(text="Hello"),
-            TTSTextFrame(text="world"),
+            TTSTextFrame(text="Hello", aggregated_by=AggregationType.WORD),
+            TTSTextFrame(text="world", aggregated_by=AggregationType.WORD),
            SleepFrame(),  # Ensure messages are processed
            CancelFrame(),
        ]
@@ -401,8 +402,8 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
        frames_to_send = [
            BotStartedSpeakingFrame(),
            SleepFrame(),
-            TTSTextFrame(text="Assistant"),
-            TTSTextFrame(text="message"),
+            TTSTextFrame(text="Assistant", aggregated_by=AggregationType.WORD),
+            TTSTextFrame(text="message", aggregated_by=AggregationType.WORD),
            BotStoppedSpeakingFrame(),
        ]

@@ -439,7 +440,7 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):

        # Test the specific pattern shared
        def make_tts_text_frame(text: str) -> TTSTextFrame:
-            frame = TTSTextFrame(text=text)
+            frame = TTSTextFrame(text=text, aggregated_by=AggregationType.WORD)
            frame.includes_inter_frame_spaces = True
            return frame