From 4699dc2345373e61cb7b9e749340d51718286d77 Mon Sep 17 00:00:00 2001
From: Mark Backman <mark@daily.co>
Date: Mon, 24 Nov 2025 18:59:26 -0500
Subject: [PATCH] Improve TTSService handling of long LLM token outputs

---
 CHANGELOG.md                         |  7 +++++
 src/pipecat/services/tts_service.py  | 21 +++++++++----
 tests/test_simple_text_aggregator.py | 45 ++++++++++++++++++++++++++++
 3 files changed, 67 insertions(+), 6 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index aaf7ec85a..652c0cf31 100644
--- a/CHANGELOG.md
+++ b/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:
diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py
index 8d7c4e6bb..c06423255 100644
--- a/src/pipecat/services/tts_service.py
+++ b/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)
diff --git a/tests/test_simple_text_aggregator.py b/tests/test_simple_text_aggregator.py
index f8e2ee553..6e1aeca11 100644
--- a/tests/test_simple_text_aggregator.py
+++ b/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