From 535b85cf90238dc0b2703e9646c8d607ec9d79ad Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 25 Nov 2025 18:04:07 -0500 Subject: [PATCH 1/6] Add split_text_by_spaces string util --- CHANGELOG.md | 5 +++ .../aggregators/llm_text_processor.py | 22 +++++++----- src/pipecat/services/tts_service.py | 30 ++++++++++------ src/pipecat/utils/string.py | 24 +++++++++++++ tests/test_utils_string.py | 34 ++++++++++++++++++- 5 files changed, 96 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2891a53e5..4a5777d5a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- Updated `LLMTextProcessor` and `TTSService` to normalize text input by + splitting into individual characters before aggregation. This ensures proper + sentence boundary detection when LLMs return multiple sentences in a single + chunk (e.g., Google Gemini). + - Updated `AICFilter` to use Quail STT as the default model (`AICModelType.QUAIL_STT`). Quail STT is optimized for human-to-machine interaction (e.g., voice agents, speech-to-text) and operates at a native diff --git a/src/pipecat/processors/aggregators/llm_text_processor.py b/src/pipecat/processors/aggregators/llm_text_processor.py index 0d9636fd8..0a99e08bc 100644 --- a/src/pipecat/processors/aggregators/llm_text_processor.py +++ b/src/pipecat/processors/aggregators/llm_text_processor.py @@ -24,6 +24,7 @@ from pipecat.frames.frames import ( LLMTextFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.utils.string import split_text_by_characters from pipecat.utils.text.base_text_aggregator import BaseTextAggregator from pipecat.utils.text.simple_text_aggregator import SimpleTextAggregator @@ -83,14 +84,19 @@ class LLMTextProcessor(FrameProcessor): await self._text_aggregator.reset() async def _handle_llm_text(self, in_frame: LLMTextFrame): - aggregation = await self._text_aggregator.aggregate(in_frame.text) - if aggregation: - out_frame = AggregatedTextFrame( - text=aggregation.text, - aggregated_by=aggregation.type, - ) - out_frame.skip_tts = in_frame.skip_tts - await self.push_frame(out_frame) + # Split text by characters to normalize LLM output into individual characters + # This ensures consistent aggregation behavior regardless of LLM chunk size + characters = split_text_by_characters(in_frame.text) + + for character in characters: + aggregation = await self._text_aggregator.aggregate(character) + if aggregation: + out_frame = AggregatedTextFrame( + text=aggregation.text, + aggregated_by=aggregation.type, + ) + out_frame.skip_tts = in_frame.skip_tts + 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 diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index 977d6957b..e8720519d 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -51,6 +51,7 @@ from pipecat.processors.frame_processor import FrameDirection from pipecat.services.ai_service import AIService from pipecat.services.websocket_service import WebsocketService from pipecat.transcriptions.language import Language +from pipecat.utils.string import split_text_by_characters 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 @@ -539,17 +540,26 @@ class TTSService(AIService): text = frame.text includes_inter_frame_spaces = frame.includes_inter_frame_spaces aggregated_by = "token" - else: - aggregate = await self._text_aggregator.aggregate(frame.text) - if aggregate: - text = aggregate.text - aggregated_by = aggregate.type - if text: - logger.trace(f"Pushing TTS frames for text: {text}, {aggregated_by}") - await self._push_tts_frames( - AggregatedTextFrame(text, aggregated_by), includes_inter_frame_spaces - ) + if text: + logger.trace(f"Pushing TTS frames for text: {text}, {aggregated_by}") + await self._push_tts_frames( + AggregatedTextFrame(text, aggregated_by), includes_inter_frame_spaces + ) + else: + # Split text by characters to normalize input into individual characters + # This ensures consistent aggregation behavior regardless of input chunk size + characters = split_text_by_characters(frame.text) + + for character in characters: + aggregate = await self._text_aggregator.aggregate(character) + if aggregate: + text = aggregate.text + aggregated_by = aggregate.type + 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, src_frame: AggregatedTextFrame, includes_inter_frame_spaces: Optional[bool] = False diff --git a/src/pipecat/utils/string.py b/src/pipecat/utils/string.py index 5645d2bcf..a0ed9ca2d 100644 --- a/src/pipecat/utils/string.py +++ b/src/pipecat/utils/string.py @@ -280,3 +280,27 @@ def concatenate_aggregated_text(text_parts: List[TextPartForConcatenation]) -> s result = result.strip() return result + + +def split_text_by_characters(text: str) -> List[str]: + """Split text into individual characters. + + Returns each character as a separate string element, allowing character-by-character + processing while maintaining the ability to reconstruct the original text. + + Args: + text: The text to split into characters. + + Returns: + A list of individual characters. + + Example:: + + >>> split_text_by_characters("Hello world!") + ["H", "e", "l", "l", "o", " ", "w", "o", "r", "l", "d", "!"] + >>> split_text_by_characters("Hi") + ["H", "i"] + >>> split_text_by_characters("") + [] + """ + return list(text) diff --git a/tests/test_utils_string.py b/tests/test_utils_string.py index 607b8d0fa..5140964dd 100644 --- a/tests/test_utils_string.py +++ b/tests/test_utils_string.py @@ -6,7 +6,7 @@ import unittest -from pipecat.utils.string import match_endofsentence, parse_start_end_tags +from pipecat.utils.string import match_endofsentence, parse_start_end_tags, split_text_by_characters class TestUtilsString(unittest.IsolatedAsyncioTestCase): @@ -232,3 +232,35 @@ class TestStartEndTags(unittest.IsolatedAsyncioTestCase): ("", ""), 41, ) + + async def test_split_text_by_characters(self): + """Test splitting text into individual characters.""" + # Basic sentence + assert split_text_by_characters("Hello world!") == [ + "H", + "e", + "l", + "l", + "o", + " ", + "w", + "o", + "r", + "l", + "d", + "!", + ] + + # Single word + assert split_text_by_characters("Hi") == ["H", "i"] + + # Empty string + assert split_text_by_characters("") == [] + + # With spaces + assert split_text_by_characters("A B") == ["A", " ", "B"] + + # Concatenation test - characters should concatenate back to original + characters = split_text_by_characters("Hello world!") + concatenated = "".join(characters) + assert concatenated == "Hello world!" From ffbb6e593707255b6b8e0c9cbad38957775bf145 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 2 Dec 2025 17:41:59 -0500 Subject: [PATCH 2/6] Update SimpleTextAggregator to handle character by character input, use a buffer to handle ambiguous EOS scenarios, and add a flush method to all aggregators --- .../aggregators/llm_text_processor.py | 12 +- src/pipecat/services/tts_service.py | 11 +- .../utils/text/base_text_aggregator.py | 13 ++ .../utils/text/pattern_pair_aggregator.py | 12 ++ .../utils/text/simple_text_aggregator.py | 61 +++++++-- .../utils/text/skip_tags_aggregator.py | 12 ++ tests/test_simple_text_aggregator.py | 123 ++++++++++++++++-- 7 files changed, 208 insertions(+), 36 deletions(-) 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" From 7e9d67002e025bf0e7c929504dd1b5d499010a01 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 2 Dec 2025 18:24:35 -0500 Subject: [PATCH 3/6] SkipTagsAggregator and PatternPairAggregator now subclass SimpleTextAggregator --- .../utils/text/pattern_pair_aggregator.py | 45 +++---- .../utils/text/simple_text_aggregator.py | 16 +++ .../utils/text/skip_tags_aggregator.py | 65 ++++------ tests/test_pattern_pair_aggregator.py | 111 +++++++++++++----- tests/test_skip_tags_aggregator.py | 53 ++++++--- 5 files changed, 171 insertions(+), 119 deletions(-) diff --git a/src/pipecat/utils/text/pattern_pair_aggregator.py b/src/pipecat/utils/text/pattern_pair_aggregator.py index 67bebd0c9..ceca1ce1b 100644 --- a/src/pipecat/utils/text/pattern_pair_aggregator.py +++ b/src/pipecat/utils/text/pattern_pair_aggregator.py @@ -17,8 +17,8 @@ from typing import Awaitable, Callable, List, Optional, Tuple from loguru import logger -from pipecat.utils.string import match_endofsentence -from pipecat.utils.text.base_text_aggregator import Aggregation, AggregationType, BaseTextAggregator +from pipecat.utils.text.base_text_aggregator import Aggregation, AggregationType +from pipecat.utils.text.simple_text_aggregator import SimpleTextAggregator class MatchAction(Enum): @@ -72,7 +72,7 @@ class PatternMatch(Aggregation): return f"PatternMatch(type={self.type}, text={self.text}, full_match={self.full_match})" -class PatternPairAggregator(BaseTextAggregator): +class PatternPairAggregator(SimpleTextAggregator): """Aggregator that identifies and processes content between pattern pairs. This aggregator buffers text until it can identify complete pattern pairs @@ -97,7 +97,7 @@ class PatternPairAggregator(BaseTextAggregator): Creates an empty aggregator with no patterns or handlers registered. Text buffering and pattern detection will begin when text is aggregated. """ - self._text = "" + super().__init__() self._patterns = {} self._handlers = {} @@ -309,9 +309,8 @@ class PatternPairAggregator(BaseTextAggregator): """Aggregate text and process pattern pairs. This method adds the new text to the buffer, processes any complete pattern - pairs, and returns processed text up to sentence boundaries if possible. - If there are incomplete patterns (start without matching end), it will - continue buffering text. + pairs, and uses the parent's lookahead logic for sentence detection when + no patterns are active. Args: text: New text to add to the buffer. @@ -355,38 +354,25 @@ class PatternPairAggregator(BaseTextAggregator): content=result.strip(), type=AggregationType.SENTENCE, full_match=result ) - # Find sentence boundary if no incomplete patterns - eos_marker = match_endofsentence(self._text) - if eos_marker: - # Extract text up to the sentence boundary - result = self._text[:eos_marker] - self._text = self._text[eos_marker:] + # Use parent's lookahead logic for sentence detection + aggregation = await super()._check_sentence_with_lookahead(text) + if aggregation: + # Convert to PatternMatch for consistency with return type return PatternMatch( - content=result.strip(), type=AggregationType.SENTENCE, full_match=result + content=aggregation.text, type=aggregation.type, full_match=aggregation.text ) # 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. + """Handle interruptions by clearing the buffer and pattern state. Called when an interruption occurs in the processing pipeline, to reset the state and discard any partially aggregated text. """ - self._text = "" + await super().handle_interruption() + # Pattern and handler state persists across interruptions async def reset(self): """Clear the internally aggregated text. @@ -394,4 +380,5 @@ class PatternPairAggregator(BaseTextAggregator): Resets the aggregator to its initial state, discarding any buffered text and clearing pattern tracking state. """ - self._text = "" + await super().reset() + # Pattern and handler state persists across resets diff --git a/src/pipecat/utils/text/simple_text_aggregator.py b/src/pipecat/utils/text/simple_text_aggregator.py index a59ac52b2..14e776502 100644 --- a/src/pipecat/utils/text/simple_text_aggregator.py +++ b/src/pipecat/utils/text/simple_text_aggregator.py @@ -60,6 +60,22 @@ class SimpleTextAggregator(BaseTextAggregator): # Add new text to buffer self._text += text + # Delegate to sentence detection logic + return await self._check_sentence_with_lookahead(text) + + async def _check_sentence_with_lookahead(self, text: str) -> Optional[Aggregation]: + """Check for sentence boundaries using lookahead logic. + + This method implements the core sentence detection logic with lookahead. + Subclasses can call this via super() to reuse the lookahead behavior + while adding their own logic (e.g., tag handling, pattern matching). + + Args: + text: The most recently added text (used for lookahead check). + + Returns: + Aggregation if sentence found, None otherwise. + """ # If we need lookahead, check if we now have non-whitespace if self._needs_lookahead: # Check if the new character is non-whitespace diff --git a/src/pipecat/utils/text/skip_tags_aggregator.py b/src/pipecat/utils/text/skip_tags_aggregator.py index ad8f7d55a..d693285eb 100644 --- a/src/pipecat/utils/text/skip_tags_aggregator.py +++ b/src/pipecat/utils/text/skip_tags_aggregator.py @@ -13,11 +13,12 @@ as a unit regardless of internal punctuation. from typing import Optional, Sequence -from pipecat.utils.string import StartEndTags, match_endofsentence, parse_start_end_tags -from pipecat.utils.text.base_text_aggregator import Aggregation, AggregationType, BaseTextAggregator +from pipecat.utils.string import StartEndTags, parse_start_end_tags +from pipecat.utils.text.base_text_aggregator import Aggregation, AggregationType +from pipecat.utils.text.simple_text_aggregator import SimpleTextAggregator -class SkipTagsAggregator(BaseTextAggregator): +class SkipTagsAggregator(SimpleTextAggregator): """Aggregator that prevents end of sentence matching between start/end tags. This aggregator buffers text until it finds an end of sentence or a start @@ -37,27 +38,17 @@ class SkipTagsAggregator(BaseTextAggregator): tags: Sequence of StartEndTags objects defining the tag pairs that should prevent sentence boundary detection. """ - self._text = "" + super().__init__() self._tags = tags self._current_tag: Optional[StartEndTags] = None self._current_tag_index: int = 0 - @property - def text(self) -> Aggregation: - """Get the currently buffered text. - - Returns: - The current text buffer content that hasn't been processed yet. - """ - return Aggregation(text=self._text.strip(), type=AggregationType.SENTENCE) - async def aggregate(self, text: str) -> Optional[Aggregation]: """Aggregate text while respecting tag boundaries. - This method adds the new text to the buffer, processes any complete - pattern pairs, and returns processed text up to sentence boundaries if - possible. If there are incomplete patterns (start without matching - end), it will continue buffering text. + This method adds the new text to the buffer, updates tag state, + and uses the parent's lookahead logic for sentence detection when + not inside tags. Args: text: New text to add to the buffer. @@ -70,46 +61,34 @@ class SkipTagsAggregator(BaseTextAggregator): # Add new text to buffer self._text += text + # Update tag state (self._current_tag, self._current_tag_index) = parse_start_end_tags( self._text, self._tags, self._current_tag, self._current_tag_index ) - # Find sentence boundary if no incomplete patterns - if not self._current_tag: - eos_marker = match_endofsentence(self._text) - if eos_marker: - # Extract text up to the sentence boundary - result = self._text[:eos_marker] - self._text = self._text[eos_marker:] - return Aggregation(text=result.strip(), type=AggregationType.SENTENCE) + # If inside tags, don't check for sentences + if self._current_tag: + return None - # 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 + # Otherwise, use parent's lookahead logic for sentence detection + return await super()._check_sentence_with_lookahead(text) async def handle_interruption(self): - """Handle interruptions by clearing the buffer. + """Handle interruptions by clearing the buffer and tag state. Called when an interruption occurs in the processing pipeline, to reset the state and discard any partially aggregated text. """ - self._text = "" + await super().handle_interruption() + self._current_tag = None + self._current_tag_index = 0 async def reset(self): - """Clear the internally aggregated text. + """Clear the internally aggregated text and tag state. Resets the aggregator to its initial state, discarding any buffered text. """ - self._text = "" + await super().reset() + self._current_tag = None + self._current_tag_index = 0 diff --git a/tests/test_pattern_pair_aggregator.py b/tests/test_pattern_pair_aggregator.py index 20b44a03c..7be96bbdc 100644 --- a/tests/test_pattern_pair_aggregator.py +++ b/tests/test_pattern_pair_aggregator.py @@ -38,14 +38,14 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.aggregator.on_pattern_match("code_pattern", self.code_handler) async def test_pattern_match_and_removal(self): - # First part doesn't complete the pattern - result = await self.aggregator.aggregate("Hello pattern") - self.assertIsNone(result) - self.assertEqual(self.aggregator.text.text, "Hello pattern") - self.assertEqual(self.aggregator.text.type, "test_pattern") + # Feed text character by character + full_text = "Hello pattern content!" + result = None - # Second part completes the pattern and includes an exclamation point - result = await self.aggregator.aggregate(" content!") + for char in full_text: + result = await self.aggregator.aggregate(char) + if result: + break # Verify the handler was called with correct PatternMatch object self.test_handler.assert_called_once() @@ -56,27 +56,49 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(call_args.text, "pattern content") # The exclamation point should be treated as a sentence boundary, - # so the result should include just text up to and including "!" + # but with lookahead, we need more text or flush() to confirm + self.assertIsNone(result) # Waiting for lookahead after "!" + + # Next sentence should provide the lookahead and trigger the previous sentence + for char in " This is another sentence.": + result = await self.aggregator.aggregate(char) + if result: + break + self.assertEqual(result.text, "Hello !") self.assertEqual(result.type, "sentence") - # Next sentence should be processed separately. Spaces around the sentence - # should be stripped in the returned Aggregation. - result = await self.aggregator.aggregate(" This is another sentence.") + # Now flush to get the remaining sentence + result = await self.aggregator.flush() self.assertEqual(result.text, "This is another sentence.") # Buffer should be empty after returning a complete sentence self.assertEqual(self.aggregator.text.text, "") async def test_pattern_match_and_aggregate(self): - # First part doesn't complete the pattern - result = await self.aggregator.aggregate("Here is code pattern") + # Feed text character by character until we get the first aggregation + result = None + for char in "Here is code pattern content": + result = await self.aggregator.aggregate(char) + if result: + break + + # First result should be "Here is code" when pattern starts self.assertEqual(result.text, "Here is code") - self.assertEqual(self.aggregator.text.text, "pattern") + self.assertEqual(self.aggregator.text.text, "pattern content") self.assertEqual(self.aggregator.text.type, "code_pattern") - # Second part completes the pattern and includes an exclamation point - result = await self.aggregator.aggregate(" content") + # Continue feeding to complete the pattern + result = None + for char in "": # Already fed all chars, just need to check state + result = await self.aggregator.aggregate(char) + if result: + break + + # Since we already fed all characters, we need to trigger pattern completion + # by checking what's in the buffer - the pattern should have been processed + # Let's continue with a space to trigger the next check + result = await self.aggregator.aggregate(" ") # Verify the handler was called with correct PatternMatch object self.code_handler.assert_called_once() @@ -89,7 +111,15 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(result.type, "code_pattern") # Next sentence should be processed separately - result = await self.aggregator.aggregate(" This is another sentence.") + # With lookahead, we need to flush to get the final sentence + for char in "This is another sentence.": + result = await self.aggregator.aggregate(char) + if result: + break + + self.assertIsNone(result) # Waiting for lookahead after "." + + result = await self.aggregator.flush() self.assertEqual(result.text, "This is another sentence.") self.assertEqual(result.type, "sentence") @@ -97,8 +127,12 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(self.aggregator.text.text, "") async def test_incomplete_pattern(self): - # Add text with incomplete pattern - result = await self.aggregator.aggregate("Hello pattern content") + # Feed text character by character with incomplete pattern + result = None + for char in "Hello pattern content": + result = await self.aggregator.aggregate(char) + if result: + break # No complete pattern yet, so nothing should be returned self.assertIsNone(result) @@ -136,9 +170,13 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.aggregator.on_pattern_match("voice", voice_handler) self.aggregator.on_pattern_match("emphasis", emphasis_handler) - # Test with multiple patterns in one text block + # Feed text character by character text = "Hello female I am very excited to meet you!" - result = await self.aggregator.aggregate(text) + result = None + for char in text: + result = await self.aggregator.aggregate(char) + if result: + break # Both handlers should be called with correct data voice_handler.assert_called_once() @@ -151,6 +189,10 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(emphasis_match.type, "emphasis") self.assertEqual(emphasis_match.text, "very") + # With lookahead, we need to flush to get the final sentence + self.assertIsNone(result) # Waiting for lookahead after "!" + + result = await self.aggregator.flush() # Voice pattern should be removed, emphasis pattern should remain self.assertEqual(result.text, "Hello I am very excited to meet you!") @@ -158,8 +200,13 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(self.aggregator.text.text, "") async def test_handle_interruption(self): - # Start with incomplete pattern - result = await self.aggregator.aggregate("Hello pattern") + # Feed text character by character with incomplete pattern + result = None + for char in "Hello pattern": + result = await self.aggregator.aggregate(char) + if result: + break + self.assertIsNone(result) # Simulate interruption @@ -172,20 +219,24 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.test_handler.assert_not_called() async def test_pattern_across_sentences(self): - # Test pattern that spans multiple sentences - result = await self.aggregator.aggregate("Hello This is sentence one.") + # Feed text character by character - pattern spans multiple sentences + full_text = "Hello This is sentence one. This is sentence two. Final sentence." + result = None - # First sentence contains start of pattern but no end, so no complete pattern yet - self.assertIsNone(result) - - # Add second part with pattern end - result = await self.aggregator.aggregate(" This is sentence two. Final sentence.") + for char in full_text: + result = await self.aggregator.aggregate(char) + if result: + break # Handler should be called with entire content self.test_handler.assert_called_once() call_args = self.test_handler.call_args[0][0] self.assertEqual(call_args.text, "This is sentence one. This is sentence two.") + # With lookahead, we need to flush to get the final sentence + self.assertIsNone(result) # Waiting for lookahead after "." + + result = await self.aggregator.flush() # Pattern should be removed, resulting in text with sentences merged self.assertEqual(result.text, "Hello Final sentence.") diff --git a/tests/test_skip_tags_aggregator.py b/tests/test_skip_tags_aggregator.py index 702b991ce..13869ca0b 100644 --- a/tests/test_skip_tags_aggregator.py +++ b/tests/test_skip_tags_aggregator.py @@ -17,7 +17,18 @@ class TestSkipTagsAggregator(unittest.IsolatedAsyncioTestCase): await self.aggregator.reset() # No tags involved, aggregate at end of sentence. - result = await self.aggregator.aggregate("Hello Pipecat!") + # Feed text character by character + result = None + for char in "Hello Pipecat!": + result = await self.aggregator.aggregate(char) + if result: + break + + # Should still be waiting for lookahead after "!" + self.assertIsNone(result) + + # Flush to get the pending sentence + result = await self.aggregator.flush() self.assertEqual(result.text, "Hello Pipecat!") self.assertEqual(result.type, "sentence") self.assertEqual(self.aggregator.text.text, "") @@ -26,7 +37,18 @@ class TestSkipTagsAggregator(unittest.IsolatedAsyncioTestCase): await self.aggregator.reset() # Tags involved, avoid aggregation during tags. - result = await self.aggregator.aggregate("My email is foo@pipecat.ai.") + # Feed text character by character + result = None + for char in "My email is foo@pipecat.ai.": + result = await self.aggregator.aggregate(char) + if result: + break + + # Should still be waiting for lookahead after "." + self.assertIsNone(result) + + # Flush to get the pending sentence + result = await self.aggregator.flush() self.assertEqual(result.text, "My email is foo@pipecat.ai.") self.assertEqual(result.type, "sentence") self.assertEqual(self.aggregator.text.text, "") @@ -34,25 +56,22 @@ class TestSkipTagsAggregator(unittest.IsolatedAsyncioTestCase): async def test_streaming_tags(self): await self.aggregator.reset() - # Tags involved, stream small chunk of texts. - result = await self.aggregator.aggregate("My email is foo.") - self.assertIsNone(result) - self.assertEqual(self.aggregator.text.text, "My email is foo.") + for char in full_text: + result = await self.aggregator.aggregate(char) + if result: + break - result = await self.aggregator.aggregate("bar@pipecat.") + # Should still be waiting for lookahead after "." self.assertIsNone(result) - self.assertEqual(self.aggregator.text.text, "My email is foo.bar@pipecat.") - - result = await self.aggregator.aggregate("aifoo.bar@pipecat.ai.") - self.assertEqual(result.text, "My email is foo.bar@pipecat.ai.") + # Flush to get the pending sentence + result = await self.aggregator.flush() + self.assertEqual(result.text, full_text) self.assertEqual(self.aggregator.text.text, "") self.assertEqual(self.aggregator.text.type, "sentence") From 4d661919632d456601b570ff07e59bed00bf56be Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 2 Dec 2025 18:35:35 -0500 Subject: [PATCH 4/6] fix: PatternPairAggregator to process patterns only once --- .../utils/text/pattern_pair_aggregator.py | 42 +++++++++++--- tests/test_pattern_pair_aggregator.py | 58 +++++++------------ 2 files changed, 56 insertions(+), 44 deletions(-) diff --git a/src/pipecat/utils/text/pattern_pair_aggregator.py b/src/pipecat/utils/text/pattern_pair_aggregator.py index ceca1ce1b..d57ccd81b 100644 --- a/src/pipecat/utils/text/pattern_pair_aggregator.py +++ b/src/pipecat/utils/text/pattern_pair_aggregator.py @@ -100,6 +100,7 @@ class PatternPairAggregator(SimpleTextAggregator): super().__init__() self._patterns = {} self._handlers = {} + self._last_processed_position = 0 # Track where we last checked for complete patterns @property def text(self) -> Aggregation: @@ -218,14 +219,18 @@ class PatternPairAggregator(SimpleTextAggregator): self._handlers[type] = handler return self - async def _process_complete_patterns(self, text: str) -> Tuple[List[PatternMatch], str]: - """Process all complete pattern pairs in the text. + async def _process_complete_patterns( + self, text: str, last_processed_position: int = 0 + ) -> Tuple[List[PatternMatch], str]: + """Process newly complete pattern pairs in the text. - Searches for all complete pattern pairs in the text, calls the - appropriate handlers, and optionally removes the matches. + Searches for pattern pairs that have been completed since last_processed_position, + calls the appropriate handlers, and optionally removes the matches. Args: text: The text to process for pattern matches. + last_processed_position: The position in text that was already processed. + Only patterns that end at or after this position will be processed. Returns: Tuple of (all_matches, processed_text) where: @@ -251,6 +256,20 @@ class PatternPairAggregator(SimpleTextAggregator): matches = list(match_iter) # Convert to list for safe iteration for match in matches: + # Only process patterns that end at or after last_processed_position + # This ensures we only call handlers once when a pattern completes + if match.end() <= last_processed_position: + # This pattern was already processed in a previous call + if action != MatchAction.REMOVE: + # For KEEP/AGGREGATE patterns, we still need to track them + content = match.group(1) + full_match = match.group(0) + pattern_match = PatternMatch( + content=content.strip(), type=type, full_match=full_match + ) + all_matches.append(pattern_match) + continue + content = match.group(1) # Content between patterns full_match = match.group(0) # Full match including patterns @@ -259,7 +278,7 @@ class PatternPairAggregator(SimpleTextAggregator): content=content.strip(), type=type, full_match=full_match ) - # Call the appropriate handler if registered + # Call the appropriate handler if registered (only for newly complete patterns) if type in self._handlers: try: await self._handlers[type](pattern_match) @@ -322,8 +341,15 @@ class PatternPairAggregator(SimpleTextAggregator): # Add new text to buffer self._text += text - # Process any complete patterns in the buffer - patterns, processed_text = await self._process_complete_patterns(self._text) + # Process any newly complete patterns in the buffer + # Only patterns that complete after _last_processed_position will trigger handlers + patterns, processed_text = await self._process_complete_patterns( + self._text, self._last_processed_position + ) + + # Update the last processed position before modifying the text + # For REMOVE patterns, the text will be shorter, so we track the original position + self._last_processed_position = len(self._text) self._text = processed_text @@ -372,6 +398,7 @@ class PatternPairAggregator(SimpleTextAggregator): to reset the state and discard any partially aggregated text. """ await super().handle_interruption() + self._last_processed_position = 0 # Pattern and handler state persists across interruptions async def reset(self): @@ -381,4 +408,5 @@ class PatternPairAggregator(SimpleTextAggregator): buffered text and clearing pattern tracking state. """ await super().reset() + self._last_processed_position = 0 # Pattern and handler state persists across resets diff --git a/tests/test_pattern_pair_aggregator.py b/tests/test_pattern_pair_aggregator.py index 7be96bbdc..b1d9d7150 100644 --- a/tests/test_pattern_pair_aggregator.py +++ b/tests/test_pattern_pair_aggregator.py @@ -40,12 +40,12 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): async def test_pattern_match_and_removal(self): # Feed text character by character full_text = "Hello pattern content!" - result = None + results = [] for char in full_text: result = await self.aggregator.aggregate(char) if result: - break + results.append(result) # Verify the handler was called with correct PatternMatch object self.test_handler.assert_called_once() @@ -55,18 +55,19 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(call_args.full_match, "pattern content") self.assertEqual(call_args.text, "pattern content") - # The exclamation point should be treated as a sentence boundary, - # but with lookahead, we need more text or flush() to confirm - self.assertIsNone(result) # Waiting for lookahead after "!" + # No results yet (waiting for lookahead after "!") + self.assertEqual(len(results), 0) # Next sentence should provide the lookahead and trigger the previous sentence for char in " This is another sentence.": result = await self.aggregator.aggregate(char) if result: - break + results.append(result) - self.assertEqual(result.text, "Hello !") - self.assertEqual(result.type, "sentence") + # First result should be "Hello !" triggered by the space lookahead + self.assertEqual(len(results), 1) + self.assertEqual(results[0].text, "Hello !") + self.assertEqual(results[0].type, "sentence") # Now flush to get the remaining sentence result = await self.aggregator.flush() @@ -76,29 +77,22 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(self.aggregator.text.text, "") async def test_pattern_match_and_aggregate(self): - # Feed text character by character until we get the first aggregation - result = None - for char in "Here is code pattern content": + # Feed text character by character, collecting all results + results = [] + full_text = "Here is code pattern content This is another sentence." + + for char in full_text: result = await self.aggregator.aggregate(char) if result: - break + results.append(result) # First result should be "Here is code" when pattern starts - self.assertEqual(result.text, "Here is code") - self.assertEqual(self.aggregator.text.text, "pattern content") - self.assertEqual(self.aggregator.text.type, "code_pattern") + self.assertEqual(results[0].text, "Here is code") + self.assertEqual(results[0].type, "sentence") - # Continue feeding to complete the pattern - result = None - for char in "": # Already fed all chars, just need to check state - result = await self.aggregator.aggregate(char) - if result: - break - - # Since we already fed all characters, we need to trigger pattern completion - # by checking what's in the buffer - the pattern should have been processed - # Let's continue with a space to trigger the next check - result = await self.aggregator.aggregate(" ") + # Second result should be the code pattern content + self.assertEqual(results[1].text, "pattern content") + self.assertEqual(results[1].type, "code_pattern") # Verify the handler was called with correct PatternMatch object self.code_handler.assert_called_once() @@ -107,18 +101,8 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(call_args.type, "code_pattern") self.assertEqual(call_args.full_match, "pattern content") self.assertEqual(call_args.text, "pattern content") - self.assertEqual(result.text, "pattern content") - self.assertEqual(result.type, "code_pattern") - - # Next sentence should be processed separately - # With lookahead, we need to flush to get the final sentence - for char in "This is another sentence.": - result = await self.aggregator.aggregate(char) - if result: - break - - self.assertIsNone(result) # Waiting for lookahead after "." + # Last sentence needs flush (waiting for lookahead after ".") result = await self.aggregator.flush() self.assertEqual(result.text, "This is another sentence.") self.assertEqual(result.type, "sentence") From fa8e7458e1a29412138da71b9fe9abbeea92d91d Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 2 Dec 2025 20:11:32 -0500 Subject: [PATCH 5/6] Clean up --- CHANGELOG.md | 20 +++++++++++++++---- .../utils/text/simple_text_aggregator.py | 8 +++++++- 2 files changed, 23 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4a5777d5a..c571b374f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,10 +15,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed -- Updated `LLMTextProcessor` and `TTSService` to normalize text input by - splitting into individual characters before aggregation. This ensures proper - sentence boundary detection when LLMs return multiple sentences in a single - chunk (e.g., Google Gemini). +- Text Aggregation Improvements: + + - Updated `LLMTextProcessor` and `TTSService` to normalize text input by + splitting into individual characters before aggregation. This ensures proper + sentence boundary detection when LLMs return multiple sentences in a single + chunk (e.g., Google Gemini). + - All text aggregators now properly support character-by-character streaming + input. + - Refactored text aggregators to use inheritance: `SkipTagsAggregator` and + `PatternPairAggregator` now inherit from `SimpleTextAggregator`, reusing + its lookahead-based sentence detection logic via + `_check_sentence_with_lookahead()`. - Updated `AICFilter` to use Quail STT as the default model (`AICModelType.QUAIL_STT`). Quail STT is optimized for human-to-machine @@ -58,6 +66,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 voice-ui-kit's conversational panel rending of the LLM output after a function call. +- Fixed a bug in `PatternPairAggregator` where pattern handlers could be called + multiple times for patterns with `MatchAction.KEEP` or `MatchAction.AGGREGATE` + actions. + ## [0.0.96] - 2025-11-26 🦃 "Happy Thanksgiving!" 🦃 ### Added diff --git a/src/pipecat/utils/text/simple_text_aggregator.py b/src/pipecat/utils/text/simple_text_aggregator.py index 14e776502..c3bdb99df 100644 --- a/src/pipecat/utils/text/simple_text_aggregator.py +++ b/src/pipecat/utils/text/simple_text_aggregator.py @@ -60,13 +60,19 @@ class SimpleTextAggregator(BaseTextAggregator): # Add new text to buffer self._text += text - # Delegate to sentence detection logic return await self._check_sentence_with_lookahead(text) async def _check_sentence_with_lookahead(self, text: str) -> Optional[Aggregation]: """Check for sentence boundaries using lookahead logic. This method implements the core sentence detection logic with lookahead. + When sentence-ending punctuation is detected, it waits for the next + non-whitespace character before calling NLTK. This disambiguates cases + like "$29." (not a sentence) vs "$29. Next" (sentence ends at period). + Whitespace alone is not meaningful lookahead since it appears in both + cases. Instead, the first non-whitespace character after the punctuation + is used to confirm the sentence boundary. + Subclasses can call this via super() to reuse the lookahead behavior while adding their own logic (e.g., tag handling, pattern matching). From d79dd94019928a71f7f5c55bdcee19eac5d8663b Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 3 Dec 2025 14:46:30 -0500 Subject: [PATCH 6/6] Make aggregate return an AsyncIterator, other clean up --- CHANGELOG.md | 29 ++-- src/pipecat/extensions/ivr/ivr_navigator.py | 13 +- .../aggregators/llm_text_processor.py | 21 +-- src/pipecat/services/tts_service.py | 21 +-- src/pipecat/utils/string.py | 24 --- .../utils/text/base_text_aggregator.py | 33 ++--- .../utils/text/pattern_pair_aggregator.py | 140 +++++++++--------- .../utils/text/simple_text_aggregator.py | 39 ++--- .../utils/text/skip_tags_aggregator.py | 44 +++--- tests/test_ivr_navigation.py | 3 + tests/test_pattern_pair_aggregator.py | 68 +++------ tests/test_simple_text_aggregator.py | 125 +++++++--------- tests/test_skip_tags_aggregator.py | 37 ++--- tests/test_utils_string.py | 34 +---- uv.lock | 6 +- 15 files changed, 259 insertions(+), 378 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c571b374f..52606239d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,18 +15,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- Improved interruption handling to prevent bots from repeating themselves. + LLM services that return multiple sentences in a single response (e.g., + `GoogleLLMService`) are now split into individual sentences before being sent + to TTS. This ensures interruptions occur at sentence boundaries, preventing + the bot from repeating content after being interrupted during long responses. + - Text Aggregation Improvements: - - Updated `LLMTextProcessor` and `TTSService` to normalize text input by - splitting into individual characters before aggregation. This ensures proper - sentence boundary detection when LLMs return multiple sentences in a single - chunk (e.g., Google Gemini). - - All text aggregators now properly support character-by-character streaming - input. + - **Breaking Change**: `BaseTextAggregator.aggregate()` now returns + `AsyncIterator[Aggregation]` instead of `Optional[Aggregation]`. This + enables the aggregator to return multiple results based on the provided + text. - Refactored text aggregators to use inheritance: `SkipTagsAggregator` and `PatternPairAggregator` now inherit from `SimpleTextAggregator`, reusing - its lookahead-based sentence detection logic via - `_check_sentence_with_lookahead()`. + the base class's sentence detection logic. - Updated `AICFilter` to use Quail STT as the default model (`AICModelType.QUAIL_STT`). Quail STT is optimized for human-to-machine @@ -54,6 +57,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed an issue where `LLMTextFrame.skip_tts` was being overwritten by LLM services. +- Fixed sentence aggregation to correctly handle ambiguous punctuation in + streaming text, such as currency ("$29.95") and abbreviations ("Mr. Smith"). + +- Fixed bug in `PatternPairAggregator` where pattern handlers could be called + multiple times for `KEEP` or `AGGREGATE` patterns. + - Fixed an issue in `SarvamTTSService` where the last sentence was not being spoken. Now, audio is flushed when the TTS services receives the `LLMFullResponseEndFrame` or `EndFrame`. @@ -66,10 +75,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 voice-ui-kit's conversational panel rending of the LLM output after a function call. -- Fixed a bug in `PatternPairAggregator` where pattern handlers could be called - multiple times for patterns with `MatchAction.KEEP` or `MatchAction.AGGREGATE` - actions. - ## [0.0.96] - 2025-11-26 🦃 "Happy Thanksgiving!" 🦃 ### Added diff --git a/src/pipecat/extensions/ivr/ivr_navigator.py b/src/pipecat/extensions/ivr/ivr_navigator.py index 1ddb41ed8..505224cde 100644 --- a/src/pipecat/extensions/ivr/ivr_navigator.py +++ b/src/pipecat/extensions/ivr/ivr_navigator.py @@ -18,8 +18,10 @@ from loguru import logger from pipecat.audio.dtmf.types import KeypadEntry from pipecat.audio.vad.vad_analyzer import VADParams from pipecat.frames.frames import ( + EndFrame, Frame, LLMContextFrame, + LLMFullResponseEndFrame, LLMMessagesUpdateFrame, LLMTextFrame, OutputDTMFUrgentFrame, @@ -149,11 +151,18 @@ class IVRProcessor(FrameProcessor): elif isinstance(frame, LLMTextFrame): # Process text through the pattern aggregator - result = await self._aggregator.aggregate(frame.text) - if result: + async for result in self._aggregator.aggregate(frame.text): # Push aggregated text that doesn't contain XML patterns await self.push_frame(LLMTextFrame(result.text), direction) + elif isinstance(frame, (LLMFullResponseEndFrame, EndFrame)): + # Flush any remaining text from the aggregator + remaining = await self._aggregator.flush() + if remaining: + await self.push_frame(LLMTextFrame(remaining.text), direction) + # Push the end frame + await self.push_frame(frame, direction) + else: await self.push_frame(frame, direction) diff --git a/src/pipecat/processors/aggregators/llm_text_processor.py b/src/pipecat/processors/aggregators/llm_text_processor.py index 8cdb40127..f0de46617 100644 --- a/src/pipecat/processors/aggregators/llm_text_processor.py +++ b/src/pipecat/processors/aggregators/llm_text_processor.py @@ -24,7 +24,6 @@ from pipecat.frames.frames import ( LLMTextFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.utils.string import split_text_by_characters from pipecat.utils.text.base_text_aggregator import BaseTextAggregator from pipecat.utils.text.simple_text_aggregator import SimpleTextAggregator @@ -84,19 +83,13 @@ class LLMTextProcessor(FrameProcessor): await self._text_aggregator.reset() async def _handle_llm_text(self, in_frame: LLMTextFrame): - # Split text by characters to normalize LLM output into individual characters - # This ensures consistent aggregation behavior regardless of LLM chunk size - characters = split_text_by_characters(in_frame.text) - - for character in characters: - aggregation = await self._text_aggregator.aggregate(character) - if aggregation: - out_frame = AggregatedTextFrame( - text=aggregation.text, - aggregated_by=aggregation.type, - ) - out_frame.skip_tts = in_frame.skip_tts - await self.push_frame(out_frame) + async for aggregation in self._text_aggregator.aggregate(in_frame.text): + out_frame = AggregatedTextFrame( + text=aggregation.text, + aggregated_by=aggregation.type, + ) + out_frame.skip_tts = in_frame.skip_tts + await self.push_frame(out_frame) async def _handle_llm_end(self, skip_tts: Optional[bool] = None): # Flush any remaining text diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index 74050550a..ca15cb2c0 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -51,7 +51,6 @@ from pipecat.processors.frame_processor import FrameDirection from pipecat.services.ai_service import AIService from pipecat.services.websocket_service import WebsocketService from pipecat.transcriptions.language import Language -from pipecat.utils.string import split_text_by_characters 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 @@ -544,19 +543,13 @@ class TTSService(AIService): AggregatedTextFrame(text, aggregated_by), includes_inter_frame_spaces ) else: - # Split text by characters to normalize input into individual characters - # This ensures consistent aggregation behavior regardless of input chunk size - characters = split_text_by_characters(frame.text) - - for character in characters: - aggregate = await self._text_aggregator.aggregate(character) - if aggregate: - text = aggregate.text - aggregated_by = aggregate.type - 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 for aggregate in self._text_aggregator.aggregate(frame.text): + text = aggregate.text + aggregated_by = aggregate.type + 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, src_frame: AggregatedTextFrame, includes_inter_frame_spaces: Optional[bool] = False diff --git a/src/pipecat/utils/string.py b/src/pipecat/utils/string.py index a0ed9ca2d..5645d2bcf 100644 --- a/src/pipecat/utils/string.py +++ b/src/pipecat/utils/string.py @@ -280,27 +280,3 @@ def concatenate_aggregated_text(text_parts: List[TextPartForConcatenation]) -> s result = result.strip() return result - - -def split_text_by_characters(text: str) -> List[str]: - """Split text into individual characters. - - Returns each character as a separate string element, allowing character-by-character - processing while maintaining the ability to reconstruct the original text. - - Args: - text: The text to split into characters. - - Returns: - A list of individual characters. - - Example:: - - >>> split_text_by_characters("Hello world!") - ["H", "e", "l", "l", "o", " ", "w", "o", "r", "l", "d", "!"] - >>> split_text_by_characters("Hi") - ["H", "i"] - >>> split_text_by_characters("") - [] - """ - return list(text) diff --git a/src/pipecat/utils/text/base_text_aggregator.py b/src/pipecat/utils/text/base_text_aggregator.py index 1b029833e..e9e7d82c3 100644 --- a/src/pipecat/utils/text/base_text_aggregator.py +++ b/src/pipecat/utils/text/base_text_aggregator.py @@ -14,7 +14,7 @@ aggregated text should be sent for speech synthesis. from abc import ABC, abstractmethod from dataclasses import dataclass from enum import Enum -from typing import Optional +from typing import AsyncIterator, Optional class AggregationType(str, Enum): @@ -80,35 +80,32 @@ class BaseTextAggregator(ABC): pass @abstractmethod - async def aggregate(self, text: str) -> Optional[Aggregation]: - """Aggregate the specified text with the currently accumulated text. + async def aggregate(self, text: str) -> AsyncIterator[Aggregation]: + """Aggregate the specified text and yield completed aggregations. - This method should be implemented to define how the new text contributes - to the aggregation process. It returns the aggregated text and a string - describing how it was aggregated if it's ready to be processed, - or None otherwise. + This method processes the input text character-by-character internally + and yields Aggregation objects as they complete. Subclasses should implement their specific logic for: - - How to combine new text with existing accumulated text + - How to process text character-by-character - When to consider the aggregated text ready for processing - What criteria determine text completion (e.g., sentence boundaries) - - When a completion occurs, the method should return an Aggregation object - containing the aggregated text and its type. The text should be stripped - of leading/trailing whitespace so that consumers can rely on a consistent - format. + - When a completion occurs, yield an Aggregation object containing the + aggregated text (stripped of leading/trailing whitespace) and its type Args: text: The text to be aggregated. - Returns: - An Aggregation object if ready for processing, or None if more - text is needed before the aggregated content is ready. If an Aggregation - object is returned, it should consist of the updated aggregated text, - stripped of leading/trailing whitespace, and a string indicating the - type of aggregation (e.g., 'sentence', 'word', 'token', 'my_custom_aggregation'). + Yields: + Aggregation objects as they complete. Each Aggregation consists of + the aggregated text (stripped of leading/trailing whitespace) and + a string indicating the type of aggregation (e.g., 'sentence', 'word', + 'token', 'my_custom_aggregation'). """ pass + # Make this a generator to satisfy type checker + yield # pragma: no cover @abstractmethod async def flush(self) -> Optional[Aggregation]: diff --git a/src/pipecat/utils/text/pattern_pair_aggregator.py b/src/pipecat/utils/text/pattern_pair_aggregator.py index d57ccd81b..518ec13ae 100644 --- a/src/pipecat/utils/text/pattern_pair_aggregator.py +++ b/src/pipecat/utils/text/pattern_pair_aggregator.py @@ -13,7 +13,7 @@ support for custom handlers and configurable actions for when a pattern is found import re from enum import Enum -from typing import Awaitable, Callable, List, Optional, Tuple +from typing import AsyncIterator, Awaitable, Callable, List, Optional, Tuple from loguru import logger @@ -256,20 +256,6 @@ class PatternPairAggregator(SimpleTextAggregator): matches = list(match_iter) # Convert to list for safe iteration for match in matches: - # Only process patterns that end at or after last_processed_position - # This ensures we only call handlers once when a pattern completes - if match.end() <= last_processed_position: - # This pattern was already processed in a previous call - if action != MatchAction.REMOVE: - # For KEEP/AGGREGATE patterns, we still need to track them - content = match.group(1) - full_match = match.group(0) - pattern_match = PatternMatch( - content=content.strip(), type=type, full_match=full_match - ) - all_matches.append(pattern_match) - continue - content = match.group(1) # Content between patterns full_match = match.group(0) # Full match including patterns @@ -278,17 +264,23 @@ class PatternPairAggregator(SimpleTextAggregator): content=content.strip(), type=type, full_match=full_match ) - # Call the appropriate handler if registered (only for newly complete patterns) - if type in self._handlers: + # Check if this pattern was already processed + already_processed = match.end() <= last_processed_position + + # Only call handler for newly completed patterns + if not already_processed and type in self._handlers: try: await self._handlers[type](pattern_match) except Exception as e: logger.error(f"Error in pattern handler for {type}: {e}") - # Remove the pattern from the text if configured + # Handle pattern based on action if action == MatchAction.REMOVE: - processed_text = processed_text.replace(full_match, "", 1) + # Remove patterns are only removed once (when newly completed) + if not already_processed: + processed_text = processed_text.replace(full_match, "", 1) else: + # KEEP/AGGREGATE patterns stay in all_matches all_matches.append(pattern_match) return all_matches, processed_text @@ -324,72 +316,74 @@ class PatternPairAggregator(SimpleTextAggregator): return None - async def aggregate(self, text: str) -> Optional[PatternMatch]: + async def aggregate(self, text: str) -> AsyncIterator[PatternMatch]: """Aggregate text and process pattern pairs. - This method adds the new text to the buffer, processes any complete pattern - pairs, and uses the parent's lookahead logic for sentence detection when - no patterns are active. + Processes the input text character-by-character, handles pattern pairs, + and uses the parent's lookahead logic for sentence detection when no + patterns are active. Args: - text: New text to add to the buffer. + text: Text to aggregate. - Returns: - Processed text up to a sentence boundary, or None if more - text is needed to form a complete sentence or pattern. + Yields: + PatternMatch objects as patterns complete or sentences are detected. """ - # Add new text to buffer - self._text += text + # Process text character by character + for char in text: + self._text += char - # Process any newly complete patterns in the buffer - # Only patterns that complete after _last_processed_position will trigger handlers - patterns, processed_text = await self._process_complete_patterns( - self._text, self._last_processed_position - ) + # Process any newly complete patterns in the buffer + # Only patterns that complete after _last_processed_position will trigger handlers + patterns, processed_text = await self._process_complete_patterns( + self._text, self._last_processed_position + ) - # Update the last processed position before modifying the text - # For REMOVE patterns, the text will be shorter, so we track the original position - self._last_processed_position = len(self._text) + # Update the last processed position to prevent re-processing patterns + # This tracks where in the buffer we've already called handlers, so we + # only trigger handlers once when a pattern completes + self._last_processed_position = len(self._text) - self._text = processed_text + self._text = processed_text - if len(patterns) > 0: - if len(patterns) > 1: - logger.warning( - f"Multiple patterns matched: {[p.type for p in patterns]}. Only the first pattern will be returned." + if len(patterns) > 0: + if len(patterns) > 1: + logger.warning( + f"Multiple patterns matched: {[p.type for p in patterns]}. Only the first pattern will be returned." + ) + # If the pattern found is set to be aggregated, return it + action = self._patterns[patterns[0].type].get("action", MatchAction.REMOVE) + if action == MatchAction.AGGREGATE: + self._text = "" + yield patterns[0] + continue + + # Check if we have incomplete patterns + pattern_start = self._match_start_of_pattern(self._text) + if pattern_start is not None: + # If the start pattern is at the beginning or should not be separately aggregated, continue + if ( + pattern_start[0] == 0 + or pattern_start[1].get("action", MatchAction.REMOVE) != MatchAction.AGGREGATE + ): + continue + # For AGGREGATE patterns: yield any text before the pattern starts + # This ensures text doesn't get stuck in the buffer waiting for sentence + # boundaries when a pattern begins (e.g., "Here is code ..." yields "Here is code") + result = self._text[: pattern_start[0]] + self._text = self._text[pattern_start[0] :] + yield PatternMatch( + content=result.strip(), type=AggregationType.SENTENCE, full_match=result ) - # If the pattern found is set to be aggregated, return it - action = self._patterns[patterns[0].type].get("action", MatchAction.REMOVE) - if action == MatchAction.AGGREGATE: - self._text = "" - return patterns[0] + continue - # Check if we have incomplete patterns - pattern_start = self._match_start_of_pattern(self._text) - if pattern_start is not None: - # If the start pattern is at the beginning or should not be separately aggregated, return None - if ( - pattern_start[0] == 0 - or pattern_start[1].get("action", MatchAction.REMOVE) != MatchAction.AGGREGATE - ): - return None - # Otherwise, strip the text up to the start pattern and return it - result = self._text[: pattern_start[0]] - self._text = self._text[pattern_start[0] :] - return PatternMatch( - content=result.strip(), type=AggregationType.SENTENCE, full_match=result - ) - - # Use parent's lookahead logic for sentence detection - aggregation = await super()._check_sentence_with_lookahead(text) - if aggregation: - # Convert to PatternMatch for consistency with return type - return PatternMatch( - content=aggregation.text, type=aggregation.type, full_match=aggregation.text - ) - - # No complete sentence found yet - return None + # Use parent's lookahead logic for sentence detection + aggregation = await super()._check_sentence_with_lookahead(char) + if aggregation: + # Convert to PatternMatch for consistency with return type + yield PatternMatch( + content=aggregation.text, type=aggregation.type, full_match=aggregation.text + ) async def handle_interruption(self): """Handle interruptions by clearing the buffer and pattern state. diff --git a/src/pipecat/utils/text/simple_text_aggregator.py b/src/pipecat/utils/text/simple_text_aggregator.py index c3bdb99df..1d123e8fb 100644 --- a/src/pipecat/utils/text/simple_text_aggregator.py +++ b/src/pipecat/utils/text/simple_text_aggregator.py @@ -11,7 +11,7 @@ until it finds an end-of-sentence marker, making it suitable for basic TTS text processing scenarios. """ -from typing import Optional +from typing import AsyncIterator, Optional from pipecat.utils.string import SENTENCE_ENDING_PUNCTUATION, match_endofsentence from pipecat.utils.text.base_text_aggregator import Aggregation, AggregationType, BaseTextAggregator @@ -42,27 +42,30 @@ class SimpleTextAggregator(BaseTextAggregator): """ return Aggregation(text=self._text.strip(), type=AggregationType.SENTENCE) - async def aggregate(self, text: str) -> Optional[Aggregation]: - """Aggregate text and return completed sentences. + async def aggregate(self, text: str) -> AsyncIterator[Aggregation]: + """Aggregate text and yield completed sentences. - 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. + Processes the input text character-by-character. 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". Args: - text: New text to add to the aggregation buffer. + text: Text to aggregate. - Returns: - A complete sentence if an end-of-sentence marker is confirmed, - or None if more text is needed. + Yields: + Complete sentences as Aggregation objects. """ - # Add new text to buffer - self._text += text + # Process text character by character + for char in text: + self._text += char - return await self._check_sentence_with_lookahead(text) + # Check for sentence with lookahead + result = await self._check_sentence_with_lookahead(char) + if result: + yield result - async def _check_sentence_with_lookahead(self, text: str) -> Optional[Aggregation]: + async def _check_sentence_with_lookahead(self, char: str) -> Optional[Aggregation]: """Check for sentence boundaries using lookahead logic. This method implements the core sentence detection logic with lookahead. @@ -77,7 +80,7 @@ class SimpleTextAggregator(BaseTextAggregator): while adding their own logic (e.g., tag handling, pattern matching). Args: - text: The most recently added text (used for lookahead check). + char: The most recently added character (used for lookahead check). Returns: Aggregation if sentence found, None otherwise. @@ -85,7 +88,7 @@ class SimpleTextAggregator(BaseTextAggregator): # 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(): + if char.strip(): # We have meaningful lookahead, call NLTK self._needs_lookahead = False eos_marker = match_endofsentence(self._text) @@ -101,7 +104,7 @@ class SimpleTextAggregator(BaseTextAggregator): return None # Check if we just added sentence-ending punctuation - if self._text[-1] in SENTENCE_ENDING_PUNCTUATION: + if self._text and self._text[-1] in SENTENCE_ENDING_PUNCTUATION: # Mark that we need lookahead (don't call NLTK yet) self._needs_lookahead = True diff --git a/src/pipecat/utils/text/skip_tags_aggregator.py b/src/pipecat/utils/text/skip_tags_aggregator.py index d693285eb..e82444ea6 100644 --- a/src/pipecat/utils/text/skip_tags_aggregator.py +++ b/src/pipecat/utils/text/skip_tags_aggregator.py @@ -11,7 +11,7 @@ between specified start/end tag pairs, ensuring that tagged content is processed as a unit regardless of internal punctuation. """ -from typing import Optional, Sequence +from typing import AsyncIterator, Optional, Sequence from pipecat.utils.string import StartEndTags, parse_start_end_tags from pipecat.utils.text.base_text_aggregator import Aggregation, AggregationType @@ -43,35 +43,37 @@ class SkipTagsAggregator(SimpleTextAggregator): self._current_tag: Optional[StartEndTags] = None self._current_tag_index: int = 0 - async def aggregate(self, text: str) -> Optional[Aggregation]: + async def aggregate(self, text: str) -> AsyncIterator[Aggregation]: """Aggregate text while respecting tag boundaries. - This method adds the new text to the buffer, updates tag state, - and uses the parent's lookahead logic for sentence detection when - not inside tags. + Processes the input text character-by-character, updates tag state, and + uses the parent's lookahead logic for sentence detection when not + inside tags. Args: - text: New text to add to the buffer. + text: Text to aggregate. - Returns: - An Aggregation object containing text up to a sentence boundary and - marked as SENTENCE type or None if more text is needed to complete a - sentence or close tags. + Yields: + Aggregation objects containing text up to a sentence boundary, + marked as SENTENCE type. """ - # Add new text to buffer - self._text += text + # Process text character by character + for char in text: + self._text += char - # Update tag state - (self._current_tag, self._current_tag_index) = parse_start_end_tags( - self._text, self._tags, self._current_tag, self._current_tag_index - ) + # Update tag state + (self._current_tag, self._current_tag_index) = parse_start_end_tags( + self._text, self._tags, self._current_tag, self._current_tag_index + ) - # If inside tags, don't check for sentences - if self._current_tag: - return None + # If inside tags, don't check for sentences + if self._current_tag: + continue - # Otherwise, use parent's lookahead logic for sentence detection - return await super()._check_sentence_with_lookahead(text) + # Otherwise, use parent's lookahead logic for sentence detection + result = await super()._check_sentence_with_lookahead(char) + if result: + yield result async def handle_interruption(self): """Handle interruptions by clearing the buffer and tag state. diff --git a/tests/test_ivr_navigation.py b/tests/test_ivr_navigation.py index 61464fb32..74bbb6266 100644 --- a/tests/test_ivr_navigation.py +++ b/tests/test_ivr_navigation.py @@ -10,6 +10,7 @@ from unittest.mock import AsyncMock from pipecat.audio.vad.vad_analyzer import VADParams from pipecat.extensions.ivr.ivr_navigator import IVRProcessor from pipecat.frames.frames import ( + LLMFullResponseEndFrame, LLMMessagesUpdateFrame, LLMTextFrame, OutputDTMFUrgentFrame, @@ -334,10 +335,12 @@ class TestIVRNavigation(unittest.IsolatedAsyncioTestCase): frames_to_send = [ LLMTextFrame(text="Hello, I'm trying to reach billing."), + LLMFullResponseEndFrame(), ] expected_down_frames = [ LLMTextFrame, # Should pass through unchanged + LLMFullResponseEndFrame, ] expected_up_frames = [ diff --git a/tests/test_pattern_pair_aggregator.py b/tests/test_pattern_pair_aggregator.py index b1d9d7150..4726e7b81 100644 --- a/tests/test_pattern_pair_aggregator.py +++ b/tests/test_pattern_pair_aggregator.py @@ -38,14 +38,8 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.aggregator.on_pattern_match("code_pattern", self.code_handler) async def test_pattern_match_and_removal(self): - # Feed text character by character - full_text = "Hello pattern content!" - results = [] - - for char in full_text: - result = await self.aggregator.aggregate(char) - if result: - results.append(result) + text = "Hello pattern content!" + results = [result async for result in self.aggregator.aggregate(text)] # Verify the handler was called with correct PatternMatch object self.test_handler.assert_called_once() @@ -59,10 +53,8 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(len(results), 0) # Next sentence should provide the lookahead and trigger the previous sentence - for char in " This is another sentence.": - result = await self.aggregator.aggregate(char) - if result: - results.append(result) + async for result in self.aggregator.aggregate(" This is another sentence."): + results.append(result) # First result should be "Hello !" triggered by the space lookahead self.assertEqual(len(results), 1) @@ -77,14 +69,9 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(self.aggregator.text.text, "") async def test_pattern_match_and_aggregate(self): - # Feed text character by character, collecting all results - results = [] - full_text = "Here is code pattern content This is another sentence." + text = "Here is code pattern content This is another sentence." - for char in full_text: - result = await self.aggregator.aggregate(char) - if result: - results.append(result) + results = [result async for result in self.aggregator.aggregate(text)] # First result should be "Here is code" when pattern starts self.assertEqual(results[0].text, "Here is code") @@ -111,15 +98,10 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(self.aggregator.text.text, "") async def test_incomplete_pattern(self): - # Feed text character by character with incomplete pattern - result = None - for char in "Hello pattern content": - result = await self.aggregator.aggregate(char) - if result: - break - + text = "Hello pattern content" + results = [result async for result in self.aggregator.aggregate(text)] # No complete pattern yet, so nothing should be returned - self.assertIsNone(result) + self.assertEqual(len(results), 0) # The handler should not be called yet self.test_handler.assert_not_called() @@ -154,13 +136,8 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.aggregator.on_pattern_match("voice", voice_handler) self.aggregator.on_pattern_match("emphasis", emphasis_handler) - # Feed text character by character text = "Hello female I am very excited to meet you!" - result = None - for char in text: - result = await self.aggregator.aggregate(char) - if result: - break + results = [result async for result in self.aggregator.aggregate(text)] # Both handlers should be called with correct data voice_handler.assert_called_once() @@ -174,7 +151,7 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(emphasis_match.text, "very") # With lookahead, we need to flush to get the final sentence - self.assertIsNone(result) # Waiting for lookahead after "!" + self.assertEqual(len(results), 0) # Waiting for lookahead after "!" result = await self.aggregator.flush() # Voice pattern should be removed, emphasis pattern should remain @@ -184,14 +161,9 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(self.aggregator.text.text, "") async def test_handle_interruption(self): - # Feed text character by character with incomplete pattern - result = None - for char in "Hello pattern": - result = await self.aggregator.aggregate(char) - if result: - break - - self.assertIsNone(result) + text = "Hello pattern" + results = [result async for result in self.aggregator.aggregate(text)] + self.assertEqual(len(results), 0) # Simulate interruption await self.aggregator.handle_interruption() @@ -203,14 +175,8 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.test_handler.assert_not_called() async def test_pattern_across_sentences(self): - # Feed text character by character - pattern spans multiple sentences - full_text = "Hello This is sentence one. This is sentence two. Final sentence." - result = None - - for char in full_text: - result = await self.aggregator.aggregate(char) - if result: - break + text = "Hello This is sentence one. This is sentence two. Final sentence." + results = [result async for result in self.aggregator.aggregate(text)] # Handler should be called with entire content self.test_handler.assert_called_once() @@ -218,7 +184,7 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(call_args.text, "This is sentence one. This is sentence two.") # With lookahead, we need to flush to get the final sentence - self.assertIsNone(result) # Waiting for lookahead after "." + self.assertEqual(len(results), 0) # Waiting for lookahead after "." result = await self.aggregator.flush() # Pattern should be removed, resulting in text with sentences merged diff --git a/tests/test_simple_text_aggregator.py b/tests/test_simple_text_aggregator.py index 663700fdf..7b87c551c 100644 --- a/tests/test_simple_text_aggregator.py +++ b/tests/test_simple_text_aggregator.py @@ -14,19 +14,22 @@ class TestSimpleTextAggregator(unittest.IsolatedAsyncioTestCase): self.aggregator = SimpleTextAggregator() async def test_reset_aggregations(self): - # Feed character-by-character - for char in "Hello ": - assert await self.aggregator.aggregate(char) == None + text = "Hello " + results = [agg async for agg in self.aggregator.aggregate(text)] + + # No complete sentences yet + assert len(results) == 0 assert self.aggregator.text.text == "Hello" await self.aggregator.reset() assert self.aggregator.text.text == "" async def test_simple_sentence(self): - # 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 + text = "Hello Pipecat!" + results = [agg async for agg in self.aggregator.aggregate(text)] + + # No complete sentences yet (waiting for lookahead after "!") + assert len(results) == 0 + # Flush to get the pending sentence aggregate = await self.aggregator.flush() assert aggregate.text == "Hello Pipecat!" @@ -34,81 +37,58 @@ class TestSimpleTextAggregator(unittest.IsolatedAsyncioTestCase): assert self.aggregator.text.text == "" async def test_multiple_sentences(self): - # 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 + text = "Hello Pipecat! How are you?" + results = [agg async for agg in self.aggregator.aggregate(text)] + + # First sentence should be complete (lookahead from "H" confirmed it) + assert len(results) == 1 + assert results[0].text == "Hello Pipecat!" + # 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 + text = "Ask me for only $29.95/month." + results = [agg async for agg in self.aggregator.aggregate(text)] + + # No complete sentences yet (waiting for lookahead after final ".") + assert len(results) == 0 + # 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 + text = "Hello Mr. Smith." + results = [agg async for agg in self.aggregator.aggregate(text)] + + # No complete sentences yet (waiting for lookahead after final ".") + assert len(results) == 0 + # 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 + text = "Hello world. Next sentence" + results = [agg async for agg in self.aggregator.aggregate(text)] + + # First sentence should be complete (lookahead from "N" confirmed it) + assert len(results) == 1 + assert results[0].text == "Hello world." 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 + text = "Hello world." + results = [agg async for agg in self.aggregator.aggregate(text)] + + # No complete sentences yet (waiting for lookahead) + assert len(results) == 0 + # Call flush to get it result = await self.aggregator.flush() assert result is not None @@ -118,7 +98,12 @@ class TestSimpleTextAggregator(unittest.IsolatedAsyncioTestCase): async def test_flush_with_no_pending(self): """Test that flush() returns any remaining text in buffer.""" - assert await self.aggregator.aggregate("Hello") == None + text = "Hello" + results = [agg async for agg in self.aggregator.aggregate(text)] + + # No complete sentences + assert len(results) == 0 + result = await self.aggregator.flush() # flush() now returns any remaining text, not just pending lookahead assert result is not None @@ -128,13 +113,13 @@ class TestSimpleTextAggregator(unittest.IsolatedAsyncioTestCase): 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." + text = "Hello. W" + results = [agg async for agg in self.aggregator.aggregate(text)] + + # First sentence should be complete (lookahead from "W" confirmed it) + assert len(results) == 1 + assert results[0].text == "Hello." + # flush() returns any remaining text (the "W" in this case) result = await self.aggregator.flush() assert result.text == "W" diff --git a/tests/test_skip_tags_aggregator.py b/tests/test_skip_tags_aggregator.py index 13869ca0b..ffed966e1 100644 --- a/tests/test_skip_tags_aggregator.py +++ b/tests/test_skip_tags_aggregator.py @@ -17,15 +17,11 @@ class TestSkipTagsAggregator(unittest.IsolatedAsyncioTestCase): await self.aggregator.reset() # No tags involved, aggregate at end of sentence. - # Feed text character by character - result = None - for char in "Hello Pipecat!": - result = await self.aggregator.aggregate(char) - if result: - break + text = "Hello Pipecat!" + results = [agg async for agg in self.aggregator.aggregate(text)] # Should still be waiting for lookahead after "!" - self.assertIsNone(result) + self.assertEqual(len(results), 0) # Flush to get the pending sentence result = await self.aggregator.flush() @@ -37,15 +33,11 @@ class TestSkipTagsAggregator(unittest.IsolatedAsyncioTestCase): await self.aggregator.reset() # Tags involved, avoid aggregation during tags. - # Feed text character by character - result = None - for char in "My email is foo@pipecat.ai.": - result = await self.aggregator.aggregate(char) - if result: - break + text = "My email is foo@pipecat.ai." + results = [agg async for agg in self.aggregator.aggregate(text)] # Should still be waiting for lookahead after "." - self.assertIsNone(result) + self.assertEqual(len(results), 0) # Flush to get the pending sentence result = await self.aggregator.flush() @@ -56,22 +48,17 @@ class TestSkipTagsAggregator(unittest.IsolatedAsyncioTestCase): async def test_streaming_tags(self): await self.aggregator.reset() - # Tags involved, feed character by character - full_text = "My email is foo.bar@pipecat.ai." - result = None - - for char in full_text: - result = await self.aggregator.aggregate(char) - if result: - break + # Tags involved + text = "My email is foo.bar@pipecat.ai." + results = [agg async for agg in self.aggregator.aggregate(text)] # Should still be waiting for lookahead after "." - self.assertIsNone(result) - self.assertEqual(self.aggregator.text.text, full_text) + self.assertEqual(len(results), 0) + self.assertEqual(self.aggregator.text.text, text) self.assertEqual(self.aggregator.text.type, "sentence") # Flush to get the pending sentence result = await self.aggregator.flush() - self.assertEqual(result.text, full_text) + self.assertEqual(result.text, text) self.assertEqual(self.aggregator.text.text, "") self.assertEqual(self.aggregator.text.type, "sentence") diff --git a/tests/test_utils_string.py b/tests/test_utils_string.py index 5140964dd..607b8d0fa 100644 --- a/tests/test_utils_string.py +++ b/tests/test_utils_string.py @@ -6,7 +6,7 @@ import unittest -from pipecat.utils.string import match_endofsentence, parse_start_end_tags, split_text_by_characters +from pipecat.utils.string import match_endofsentence, parse_start_end_tags class TestUtilsString(unittest.IsolatedAsyncioTestCase): @@ -232,35 +232,3 @@ class TestStartEndTags(unittest.IsolatedAsyncioTestCase): ("", ""), 41, ) - - async def test_split_text_by_characters(self): - """Test splitting text into individual characters.""" - # Basic sentence - assert split_text_by_characters("Hello world!") == [ - "H", - "e", - "l", - "l", - "o", - " ", - "w", - "o", - "r", - "l", - "d", - "!", - ] - - # Single word - assert split_text_by_characters("Hi") == ["H", "i"] - - # Empty string - assert split_text_by_characters("") == [] - - # With spaces - assert split_text_by_characters("A B") == ["A", " ", "B"] - - # Concatenation test - characters should concatenate back to original - characters = split_text_by_characters("Hello world!") - concatenated = "".join(characters) - assert concatenated == "Hello world!" diff --git a/uv.lock b/uv.lock index 17bdef385..da1b785a1 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 2 +revision = 3 requires-python = ">=3.10" resolution-markers = [ "python_full_version >= '3.13'", @@ -4710,7 +4710,6 @@ requires-dist = [ { name = "numba", specifier = "==0.61.2" }, { name = "numpy", specifier = ">=1.26.4,<3" }, { name = "nvidia-riva-client", marker = "extra == 'nvidia'", specifier = "~=2.21.1" }, - { name = "nvidia-riva-client", marker = "extra == 'riva'", specifier = "~=2.21.1" }, { name = "onnxruntime", marker = "extra == 'local-smart-turn-v3'", specifier = ">=1.20.1,<2" }, { name = "onnxruntime", marker = "extra == 'silero'", specifier = ">=1.20.1,<2" }, { name = "openai", specifier = ">=1.74.0,<3" }, @@ -4721,6 +4720,7 @@ requires-dist = [ { name = "opentelemetry-sdk", marker = "extra == 'tracing'", specifier = ">=1.33.0" }, { name = "ormsgpack", marker = "extra == 'fish'", specifier = "~=1.7.0" }, { name = "pillow", specifier = ">=11.1.0,<12" }, + { name = "pipecat-ai", extras = ["nvidia"], marker = "extra == 'riva'" }, { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'assemblyai'" }, { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'asyncai'" }, { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'aws'" }, @@ -4771,7 +4771,7 @@ requires-dist = [ { name = "wait-for2", marker = "python_full_version < '3.12'", specifier = ">=0.4.1" }, { name = "websockets", marker = "extra == 'websockets-base'", specifier = ">=13.1,<16.0" }, ] -provides-extras = ["aic", "anthropic", "assemblyai", "asyncai", "aws", "aws-nova-sonic", "azure", "cartesia", "cerebras", "daily", "deepgram", "deepseek", "elevenlabs", "fal", "fireworks", "fish", "gladia", "google", "grok", "groq", "gstreamer", "heygen", "hume", "inworld", "koala", "krisp", "langchain", "livekit", "lmnt", "local", "local-smart-turn", "local-smart-turn-v3", "mcp", "mem0", "mistral", "mlx-whisper", "moondream", "neuphonic", "nim", "noisereduce", "openai", "openpipe", "openrouter", "perplexity", "playht", "qwen", "remote-smart-turn", "rime", "riva", "nvidia", "runner", "sagemaker", "sambanova", "sarvam", "sentry", "silero", "simli", "soniox", "soundfile", "speechmatics", "strands", "tavus", "together", "tracing", "ultravox", "webrtc", "websocket", "websockets-base", "whisper"] +provides-extras = ["aic", "anthropic", "assemblyai", "asyncai", "aws", "aws-nova-sonic", "azure", "cartesia", "cerebras", "daily", "deepgram", "deepseek", "elevenlabs", "fal", "fireworks", "fish", "gladia", "google", "grok", "groq", "gstreamer", "heygen", "hume", "inworld", "koala", "krisp", "langchain", "livekit", "lmnt", "local", "local-smart-turn", "local-smart-turn-v3", "mcp", "mem0", "mistral", "mlx-whisper", "moondream", "neuphonic", "noisereduce", "nvidia", "openai", "openpipe", "openrouter", "perplexity", "playht", "qwen", "remote-smart-turn", "rime", "riva", "runner", "sagemaker", "sambanova", "sarvam", "sentry", "silero", "simli", "soniox", "soundfile", "speechmatics", "strands", "tavus", "together", "tracing", "ultravox", "webrtc", "websocket", "websockets-base", "whisper"] [package.metadata.requires-dev] dev = [