Improve TTSService handling of long LLM token outputs

CHANGELOG.md

@@ -88,6 +88,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- Improved `TTSService` to properly handle buffered sentences at the end of LLM
+  responses. Previously, when an LLM response ended, any complete sentences
+  remaining in the aggregator's buffer would be sent to TTS as one large
+  chunk. Now `TTSService` continues processing aggregation by repeatedly calling
+  `aggregate("")` until all buffered text has been processed, ensuring each
+  sentence is sent to TTS individually for better interruption points.
+
 - Updated `daily-python` to 0.22.0.
 
 - `BaseTextAggregator` changes:

src/pipecat/services/tts_service.py

@@ -425,16 +425,25 @@ class TTSService(AIService):
             # pause to avoid audio overlapping.
             await self._maybe_pause_frame_processing()
 
-            pending_aggregation = self._text_aggregator.text
+            # Drain any remaining complete aggregation units from the aggregator
+            # by repeatedly calling aggregate("") until nothing is left
+            pending_aggregation = await self._text_aggregator.aggregate("")
+            while pending_aggregation:
+                await self._push_tts_frames(
+                    AggregatedTextFrame(pending_aggregation.text, pending_aggregation.type)
+                )
+                pending_aggregation = await self._text_aggregator.aggregate("")
+
+            # After draining all complete units, get any remaining partial text
+            remaining_text = self._text_aggregator.text
+            if remaining_text.text:
+                await self._push_tts_frames(
+                    AggregatedTextFrame(remaining_text.text, remaining_text.type)
+                )
 
             # Reset aggregator state
             await self._text_aggregator.reset()
             self._processing_text = False
-
-            if 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)

tests/test_simple_text_aggregator.py

@@ -33,3 +33,48 @@ class TestSimpleTextAggregator(unittest.IsolatedAsyncioTestCase):
         assert self.aggregator.text.text == "How are"
         aggregate = await self.aggregator.aggregate("you?")
         assert aggregate.text == "How are you?"
+
+    async def test_word_by_word(self):
+        """Test word-by-word token aggregation (e.g., OpenAI)."""
+        assert await self.aggregator.aggregate("Hello") == None
+        aggregate = await self.aggregator.aggregate("!")
+        assert aggregate.text == "Hello!"
+        assert await self.aggregator.aggregate(" I") == None
+        assert await self.aggregator.aggregate(" am") == None
+        aggregate = await self.aggregator.aggregate(" Doug.")
+        assert aggregate.text == "I am Doug."
+        assert self.aggregator.text.text == ""
+
+    async def test_chunks_with_partial_sentences(self):
+        """Test chunks with partial sentences."""
+        aggregate = await self.aggregator.aggregate("Hey!")
+        assert aggregate.text == "Hey!"
+        aggregate = await self.aggregator.aggregate(" Nice to meet you! So")
+        assert aggregate.text == "Nice to meet you!"
+        assert self.aggregator.text.text == "So"
+        assert await self.aggregator.aggregate(" what") == None
+        aggregate = await self.aggregator.aggregate("'d you like?")
+        assert aggregate.text == "So what'd you like?"
+
+    async def test_multi_sentence_chunk(self):
+        """Test chunks with multiple complete sentences."""
+        aggregate = await self.aggregator.aggregate("Hello! I am Doug. Nice to meet you!")
+        assert aggregate.text == "Hello!"
+        # Drain remaining sentences by calling aggregate("")
+        aggregate = await self.aggregator.aggregate("")
+        assert aggregate.text == "I am Doug."
+        aggregate = await self.aggregator.aggregate("")
+        assert aggregate.text == "Nice to meet you!"
+        assert await self.aggregator.aggregate("") == None
+        assert self.aggregator.text.text == ""
+
+    async def test_aggregate_empty_with_incomplete(self):
+        """Test aggregate('') with incomplete sentence in buffer."""
+        aggregate = await self.aggregator.aggregate("Hello! I am")
+        assert aggregate.text == "Hello!"
+        assert await self.aggregator.aggregate("") == None
+        assert self.aggregator.text.text == "I am"
+
+    async def test_aggregate_empty_buffer(self):
+        """Test aggregate('') with empty buffer."""
+        assert await self.aggregator.aggregate("") == None