diff --git a/src/pipecat/processors/aggregators/llm_text_processor.py b/src/pipecat/processors/aggregators/llm_text_processor.py index 0a99e08bc..8cdb40127 100644 --- a/src/pipecat/processors/aggregators/llm_text_processor.py +++ b/src/pipecat/processors/aggregators/llm_text_processor.py @@ -99,14 +99,12 @@ class LLMTextProcessor(FrameProcessor): await self.push_frame(out_frame) async def _handle_llm_end(self, skip_tts: Optional[bool] = None): - # Flush any remaining aggregated text at the end of the LLM response - aggregation = self._text_aggregator.text - await self._text_aggregator.reset() - text = aggregation.text.strip() - if text: + # Flush any remaining text + remaining = await self._text_aggregator.flush() + if remaining: out_frame = AggregatedTextFrame( - text=text, - aggregated_by=aggregation.type, + text=remaining.text, + aggregated_by=remaining.type, ) out_frame.skip_tts = skip_tts await self.push_frame(out_frame) diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index e8720519d..74050550a 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -426,16 +426,13 @@ class TTSService(AIService): # pause to avoid audio overlapping. await self._maybe_pause_frame_processing() - pending_aggregation = self._text_aggregator.text + # Flush any remaining text (including text waiting for lookahead) + remaining = await self._text_aggregator.flush() + if remaining: + await self._push_tts_frames(AggregatedTextFrame(remaining.text, remaining.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/src/pipecat/utils/text/base_text_aggregator.py b/src/pipecat/utils/text/base_text_aggregator.py index 5c39ce769..1b029833e 100644 --- a/src/pipecat/utils/text/base_text_aggregator.py +++ b/src/pipecat/utils/text/base_text_aggregator.py @@ -110,6 +110,19 @@ class BaseTextAggregator(ABC): """ pass + @abstractmethod + async def flush(self) -> Optional[Aggregation]: + """Flush any pending aggregation. + + This method is called at the end of a stream (e.g., when receiving + LLMFullResponseEndFrame) to return any text that was buffered. + + Returns: + An Aggregation object if there is pending text, or None if there + is no pending text. + """ + pass + @abstractmethod async def handle_interruption(self): """Handle interruptions in the text aggregation process. diff --git a/src/pipecat/utils/text/pattern_pair_aggregator.py b/src/pipecat/utils/text/pattern_pair_aggregator.py index c140e3243..67bebd0c9 100644 --- a/src/pipecat/utils/text/pattern_pair_aggregator.py +++ b/src/pipecat/utils/text/pattern_pair_aggregator.py @@ -368,6 +368,18 @@ class PatternPairAggregator(BaseTextAggregator): # No complete sentence found yet return None + async def flush(self) -> Optional[Aggregation]: + """Flush any pending aggregation. + + For PatternPairAggregator, there is no lookahead buffering, so this + simply returns None. Any remaining text can be retrieved via the + .text property. + + Returns: + None, as this aggregator does not buffer pending sentences. + """ + return None + async def handle_interruption(self): """Handle interruptions by clearing the buffer. diff --git a/src/pipecat/utils/text/simple_text_aggregator.py b/src/pipecat/utils/text/simple_text_aggregator.py index 56eab7032..a59ac52b2 100644 --- a/src/pipecat/utils/text/simple_text_aggregator.py +++ b/src/pipecat/utils/text/simple_text_aggregator.py @@ -13,7 +13,7 @@ text processing scenarios. from typing import Optional -from pipecat.utils.string import match_endofsentence +from pipecat.utils.string import SENTENCE_ENDING_PUNCTUATION, match_endofsentence from pipecat.utils.text.base_text_aggregator import Aggregation, AggregationType, BaseTextAggregator @@ -31,6 +31,7 @@ class SimpleTextAggregator(BaseTextAggregator): Creates an empty text buffer ready to begin accumulating text tokens. """ self._text = "" + self._needs_lookahead: bool = False @property def text(self) -> Aggregation: @@ -44,27 +45,59 @@ class SimpleTextAggregator(BaseTextAggregator): async def aggregate(self, text: str) -> Optional[Aggregation]: """Aggregate text and return completed sentences. - Adds the new text to the buffer and checks for end-of-sentence markers. - When a sentence boundary is found, returns the completed sentence and - removes it from the buffer. + Adds the new text to the buffer. When sentence-ending punctuation is + detected, it waits for non-whitespace lookahead before calling NLTK. + This prevents false positives like "$29." being detected as a sentence + when it's actually "$29.95", and avoids unnecessary NLTK calls. Args: text: New text to add to the aggregation buffer. Returns: - A complete sentence if an end-of-sentence marker is found, - or None if more text is needed to complete a sentence. + A complete sentence if an end-of-sentence marker is confirmed, + or None if more text is needed. """ - result: Optional[str] = None - + # Add new text to buffer 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:] + # If we need lookahead, check if we now have non-whitespace + if self._needs_lookahead: + # Check if the new character is non-whitespace + if text.strip(): + # We have meaningful lookahead, call NLTK + self._needs_lookahead = False + eos_marker = match_endofsentence(self._text) - if result: + if eos_marker: + # NLTK confirmed a sentence - return it + result = self._text[:eos_marker] + self._text = self._text[eos_marker:] + return Aggregation(text=result, type=AggregationType.SENTENCE) + # No sentence found - keep accumulating + return None + # Still whitespace, keep waiting + return None + + # Check if we just added sentence-ending punctuation + if self._text[-1] in SENTENCE_ENDING_PUNCTUATION: + # Mark that we need lookahead (don't call NLTK yet) + self._needs_lookahead = True + + return None + + async def flush(self) -> Optional[Aggregation]: + """Flush any remaining text in the buffer. + + Returns any text remaining in the buffer. This is called at the end + of a stream to ensure all text is processed. + + Returns: + Any remaining text as a sentence, or None if buffer is empty. + """ + if self._text: + # Return whatever we have in the buffer + result = self._text + await self.reset() return Aggregation(text=result.strip(), type=AggregationType.SENTENCE) return None @@ -75,6 +108,7 @@ class SimpleTextAggregator(BaseTextAggregator): discarding any partially accumulated text. """ self._text = "" + self._needs_lookahead = False async def reset(self): """Clear the internally aggregated text. @@ -83,3 +117,4 @@ class SimpleTextAggregator(BaseTextAggregator): any accumulated text content. """ self._text = "" + self._needs_lookahead = False diff --git a/src/pipecat/utils/text/skip_tags_aggregator.py b/src/pipecat/utils/text/skip_tags_aggregator.py index 3c8b95aab..ad8f7d55a 100644 --- a/src/pipecat/utils/text/skip_tags_aggregator.py +++ b/src/pipecat/utils/text/skip_tags_aggregator.py @@ -86,6 +86,18 @@ class SkipTagsAggregator(BaseTextAggregator): # No complete sentence found yet return None + async def flush(self) -> Optional[Aggregation]: + """Flush any pending aggregation. + + For SkipTagsAggregator, there is no lookahead buffering, so this + simply returns None. Any remaining text can be retrieved via the + .text property. + + Returns: + None, as this aggregator does not buffer pending sentences. + """ + return None + async def handle_interruption(self): """Handle interruptions by clearing the buffer. diff --git a/tests/test_simple_text_aggregator.py b/tests/test_simple_text_aggregator.py index f8e2ee553..663700fdf 100644 --- a/tests/test_simple_text_aggregator.py +++ b/tests/test_simple_text_aggregator.py @@ -14,22 +14,127 @@ class TestSimpleTextAggregator(unittest.IsolatedAsyncioTestCase): self.aggregator = SimpleTextAggregator() async def test_reset_aggregations(self): - assert await self.aggregator.aggregate("Hello ") == None + # Feed character-by-character + for char in "Hello ": + assert await self.aggregator.aggregate(char) == None assert self.aggregator.text.text == "Hello" await self.aggregator.reset() assert self.aggregator.text.text == "" async def test_simple_sentence(self): - assert await self.aggregator.aggregate("Hello ") == None - aggregate = await self.aggregator.aggregate("Pipecat!") + # Feed character-by-character: "Hello Pipecat!" + for char in "Hello Pipecat": + assert await self.aggregator.aggregate(char) == None + # After "!", lookahead waits for confirmation + assert await self.aggregator.aggregate("!") == None + # Flush to get the pending sentence + aggregate = await self.aggregator.flush() assert aggregate.text == "Hello Pipecat!" assert aggregate.type == "sentence" assert self.aggregator.text.text == "" async def test_multiple_sentences(self): - aggregate = await self.aggregator.aggregate("Hello Pipecat! How are ") - assert aggregate.text == "Hello Pipecat!" - # Aggregators should strip leading/trailing spaces when returning text - assert self.aggregator.text.text == "How are" - aggregate = await self.aggregator.aggregate("you?") - assert aggregate.text == "How are you?" + # Feed character-by-character: "Hello Pipecat! How are you?" + for char in "Hello Pipecat": + assert await self.aggregator.aggregate(char) == None + # Hit "!" - wait for lookahead + assert await self.aggregator.aggregate("!") == None + # Space is whitespace - keep waiting + assert await self.aggregator.aggregate(" ") == None + # "H" confirms sentence end + result = await self.aggregator.aggregate("H") + assert result.text == "Hello Pipecat!" + # Continue with second sentence + for char in "ow are you": + assert await self.aggregator.aggregate(char) == None + # Hit "?" - wait for lookahead + assert await self.aggregator.aggregate("?") == None + # Flush to get the pending sentence + result = await self.aggregator.flush() + assert result.text == "How are you?" + + async def test_lookahead_decimal_number(self): + """Test that $29.95 is not split at $29.""" + # Feed character by character: "Ask me for only $29.95/month." + for char in "Ask me for only $29": + assert await self.aggregator.aggregate(char) == None + # When we hit ".", it looks like end of sentence, but should wait for lookahead + assert await self.aggregator.aggregate(".") == None + # Next character "9" confirms it's not end of sentence (NLTK changes boundary) + for char in "95/month": + assert await self.aggregator.aggregate(char) == None + # Now we hit the real end of sentence - wait for lookahead + assert await self.aggregator.aggregate(".") == None + # Can use flush() to get the pending sentence at end of stream + result = await self.aggregator.flush() + assert result.text == "Ask me for only $29.95/month." + + async def test_lookahead_abbreviation(self): + """Test that Mr. Smith is not split at Mr.""" + # Feed character by character: "Hello Mr. Smith." + for char in "Hello Mr": + assert await self.aggregator.aggregate(char) == None + # When we hit ".", it looks like end of sentence, but should wait for lookahead + assert await self.aggregator.aggregate(".") == None + # Space alone is not enough + assert await self.aggregator.aggregate(" ") == None + # "S" confirms it's not end of sentence (NLTK changes boundary detection) + for char in "Smith": + assert await self.aggregator.aggregate(char) == None + # Now we hit the real end of sentence - wait for lookahead + assert await self.aggregator.aggregate(".") == None + # Can use flush() to get the pending sentence at end of stream + result = await self.aggregator.flush() + assert result.text == "Hello Mr. Smith." + + async def test_lookahead_actual_sentence_end(self): + """Test that a real sentence end is detected after lookahead.""" + # Feed character by character: "Hello world. Next sentence" + for char in "Hello world": + assert await self.aggregator.aggregate(char) == None + # Hit period - should wait for lookahead + assert await self.aggregator.aggregate(".") == None + # Space alone is not enough - need non-whitespace for meaningful lookahead + assert await self.aggregator.aggregate(" ") == None + # Capital letter confirms sentence end (NLTK detects boundary at same position) + result = await self.aggregator.aggregate("N") + assert result.text == "Hello world." + # Continue with next sentence + assert await self.aggregator.aggregate("e") == None + + async def test_flush_pending_sentence(self): + """Test that flush() returns pending sentence waiting for lookahead.""" + # Feed up to a period + for char in "Hello world": + assert await self.aggregator.aggregate(char) == None + assert await self.aggregator.aggregate(".") == None + # At this point, "Hello world." is pending lookahead + # Call flush to get it + result = await self.aggregator.flush() + assert result is not None + assert result.text == "Hello world." + # Flush again should return None + assert await self.aggregator.flush() == None + + async def test_flush_with_no_pending(self): + """Test that flush() returns any remaining text in buffer.""" + assert await self.aggregator.aggregate("Hello") == None + result = await self.aggregator.flush() + # flush() now returns any remaining text, not just pending lookahead + assert result is not None + assert result.text == "Hello" + # Buffer should be empty after flush + assert self.aggregator.text.text == "" + + async def test_flush_after_lookahead_confirmed(self): + """Test flush after lookahead has already confirmed sentence.""" + for char in "Hello.": + await self.aggregator.aggregate(char) + # Space alone is not enough - still waiting + assert await self.aggregator.aggregate(" ") == None + # Non-whitespace lookahead confirms it's a sentence + result = await self.aggregator.aggregate("W") + assert result.text == "Hello." + # flush() returns any remaining text (the "W" in this case) + result = await self.aggregator.flush() + assert result.text == "W"