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")