Merge pull request #1379 from pipecat-ai/aleix/introduce-text-aggregators

introduce text aggregators

CHANGELOG.md

@@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added new `BaseTextAggregator`. Text aggregators are used by the TTS service
+  to aggregate LLM tokens and decide when the aggregated text should be pushed
+  to the TTS service. It also allows for the text to be manipulated while it's
+  being aggregated.
+
 - Added new `UltravoxSTTService`.
   (see https://github.com/fixie-ai/ultravox)
 

src/pipecat/services/ai_services.py

@@ -45,8 +45,9 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.services.websocket_service import WebsocketService
 from pipecat.transcriptions.language import Language
-from pipecat.utils.string import match_endofsentence
+from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
 from pipecat.utils.text.base_text_filter import BaseTextFilter
+from pipecat.utils.text.simple_text_aggregator import SimpleTextAggregator
 from pipecat.utils.time import seconds_to_nanoseconds
 
 
@@ -237,6 +238,9 @@ class TTSService(AIService):
         pause_frame_processing: bool = False,
         # TTS output sample rate
         sample_rate: Optional[int] = None,
+        # Text aggregator to aggregate incoming tokens and decide when to push to the TTS.
+        text_aggregator: Optional[BaseTextAggregator] = None,
+        # Text filter executed after text has been aggregated.
         text_filter: Optional[BaseTextFilter] = None,
         **kwargs,
     ):
@@ -252,12 +256,12 @@ class TTSService(AIService):
         self._sample_rate = 0
         self._voice_id: str = ""
         self._settings: Dict[str, Any] = {}
+        self._text_aggregator: BaseTextAggregator = text_aggregator or SimpleTextAggregator()
         self._text_filter: Optional[BaseTextFilter] = text_filter
 
         self._stop_frame_task: Optional[asyncio.Task] = None
         self._stop_frame_queue: asyncio.Queue = asyncio.Queue()
 
-        self._current_sentence: str = ""
         self._processing_text: bool = False
 
     @property
@@ -281,6 +285,9 @@ class TTSService(AIService):
     async def update_setting(self, key: str, value: Any):
         pass
 
+    async def flush_audio(self):
+        pass
+
     async def start(self, frame: StartFrame):
         await super().start(frame)
         self._sample_rate = self._init_sample_rate or frame.audio_out_sample_rate
@@ -336,8 +343,8 @@ class TTSService(AIService):
             # pause to avoid audio overlapping.
             await self._maybe_pause_frame_processing()
 
-            sentence = self._current_sentence
-            self._current_sentence = ""
+            sentence = self._text_aggregator.text
+            self._text_aggregator.reset()
             self._processing_text = False
             await self._push_tts_frames(sentence)
             if isinstance(frame, LLMFullResponseEndFrame):
@@ -382,8 +389,8 @@ class TTSService(AIService):
             await self._stop_frame_queue.put(frame)
 
     async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection):
-        self._current_sentence = ""
         self._processing_text = False
+        self._text_aggregator.handle_interruption()
         if self._text_filter:
             self._text_filter.handle_interruption()
 
@@ -400,11 +407,7 @@ class TTSService(AIService):
         if not self._aggregate_sentences:
             text = frame.text
         else:
-            self._current_sentence += frame.text
-            eos_end_marker = match_endofsentence(self._current_sentence)
-            if eos_end_marker:
-                text = self._current_sentence[:eos_end_marker]
-                self._current_sentence = self._current_sentence[eos_end_marker:]
+            text = self._text_aggregator.aggregate(frame.text)
 
         if text:
             await self._push_tts_frames(text)
@@ -535,9 +538,6 @@ class WebsocketTTSService(TTSService, WebsocketService):
         TTSService.__init__(self, **kwargs)
         WebsocketService.__init__(self)
 
-    async def flush_audio(self):
-        pass
-
 
 class InterruptibleTTSService(WebsocketTTSService):
     """This is a base class for websocket-based TTS services that don't support

src/pipecat/utils/text/base_text_aggregator.py

@@ -0,0 +1,57 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from abc import ABC, abstractmethod
+from typing import Optional
+
+
+class BaseTextAggregator(ABC):
+    """This is the base class for text aggregators. Text aggregators are usually
+    used by the TTS service to aggregate LLM tokens and decide when the
+    aggregated text should be pushed to the TTS service.
+
+    Text aggregators can also be used to manipulate text while it's being
+    aggregated (e.g. reasoning blocks can be removed).
+
+    """
+
+    @property
+    @abstractmethod
+    def text(self) -> str:
+        """Returns the currently aggregated text."""
+        pass
+
+    @abstractmethod
+    def aggregate(self, text: str) -> Optional[str]:
+        """Aggregates the specified text with the currently accumulated text.
+
+        This method should be implemented to define how the new text contributes
+        to the aggregation process. It returns the updated aggregated text if
+        it's ready to be processed, or None otherwise.
+
+        Args:
+            text (str): The text to be aggregated.
+
+        Returns:
+            Optional[str]: The updated aggregated text or None if aggregated
+            text is not ready.
+
+        """
+        pass
+
+    @abstractmethod
+    def handle_interruption(self):
+        """Handles interruptions. When an interruption occurs it is possible
+        that we might want to discard the aggregated text or do some internal
+        modifications to the aggregated text.
+
+        """
+        pass
+
+    @abstractmethod
+    def reset(self):
+        """Clears the internally aggregated text."""
+        pass

src/pipecat/utils/text/simple_text_aggregator.py

@@ -0,0 +1,42 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from typing import Optional
+
+from pipecat.utils.string import match_endofsentence
+from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
+
+
+class SimpleTextAggregator(BaseTextAggregator):
+    """This is a simple text aggregator. It aggregates text until an end of
+    sentence is found.
+
+    """
+
+    def __init__(self):
+        self._text = ""
+
+    @property
+    def text(self) -> str:
+        return self._text
+
+    def aggregate(self, text: str) -> Optional[str]:
+        result: Optional[str] = None
+
+        self._text += text
+
+        eos_end_marker = match_endofsentence(self._text)
+        if eos_end_marker:
+            result = self._text[:eos_end_marker]
+            self._text = self._text[eos_end_marker:]
+
+        return result
+
+    def handle_interruption(self):
+        self._text = ""
+
+    def reset(self):
+        self._text = ""

tests/test_simple_text_aggregator.py

@@ -0,0 +1,29 @@
+#
+# Copyright (c) 2024-2025 Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import unittest
+
+from pipecat.utils.text.simple_text_aggregator import SimpleTextAggregator
+
+
+class TestSimpleTextAggregator(unittest.IsolatedAsyncioTestCase):
+    def setUp(self):
+        self.aggregator = SimpleTextAggregator()
+
+    async def test_reset_aggregations(self):
+        assert self.aggregator.aggregate("Hello ") == None
+        assert self.aggregator.text == "Hello "
+        self.aggregator.reset()
+        assert self.aggregator.text == ""
+
+    async def test_simple_sentence(self):
+        assert self.aggregator.aggregate("Hello ") == None
+        assert self.aggregator.aggregate("Pipecat!") == "Hello Pipecat!"
+        assert self.aggregator.text == ""
+
+    async def test_multiple_sentences(self):
+        assert self.aggregator.aggregate("Hello Pipecat! How are ") == "Hello Pipecat!"
+        assert self.aggregator.aggregate("you?") == " How are you?"