From 4d661919632d456601b570ff07e59bed00bf56be Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 2 Dec 2025 18:35:35 -0500 Subject: [PATCH] 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")