fix: PatternPairAggregator to process patterns only once

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]:
+    async def _process_complete_patterns(
-        """Process all complete pattern pairs in the text.
+        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
+        Searches for pattern pairs that have been completed since last_processed_position,
-        appropriate handlers, and optionally removes the matches.
+        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
+        # Process any newly complete patterns in the buffer
-        patterns, processed_text = await self._process_complete_patterns(self._text)
+        # 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

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 <test>pattern content</test>!"
-        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, "<test>pattern content</test>")
         self.assertEqual(call_args.text, "pattern content")
 
-        # The exclamation point should be treated as a sentence boundary,
+        # No results yet (waiting for lookahead after "!")
-        # but with lookahead, we need more text or flush() to confirm
+        self.assertEqual(len(results), 0)
-        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
+                results.append(result)
 
-        self.assertEqual(result.text, "Hello !")
+        # First result should be "Hello !" triggered by the space lookahead
-        self.assertEqual(result.type, "sentence")
+        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
+        # Feed text character by character, collecting all results
-        result = None
+        results = []
-        for char in "Here is code <code>pattern content</code>":
+        full_text = "Here is code <code>pattern content</code> 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(results[0].text, "Here is code")
-        self.assertEqual(self.aggregator.text.text, "<code>pattern content</code>")
+        self.assertEqual(results[0].type, "sentence")
-        self.assertEqual(self.aggregator.text.type, "code_pattern")
 
-        # Continue feeding to complete the pattern
+        # Second result should be the code pattern content
-        result = None
+        self.assertEqual(results[1].text, "pattern content")
-        for char in "":  # Already fed all chars, just need to check state
+        self.assertEqual(results[1].type, "code_pattern")
-            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()
@@ -107,18 +101,8 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(call_args.type, "code_pattern")
         self.assertEqual(call_args.full_match, "<code>pattern content</code>")
         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")