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