SkipTagsAggregator and PatternPairAggregator now subclass SimpleTextAggregator

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
-from pipecat.utils.text.base_text_aggregator import Aggregation, AggregationType, BaseTextAggregator
+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.
+        pairs, and uses the parent's lookahead logic for sentence detection when
-        If there are incomplete patterns (start without matching end), it will
+        no patterns are active.
-        continue buffering text.
 
         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
+        # Use parent's lookahead logic for sentence detection
-        eos_marker = match_endofsentence(self._text)
+        aggregation = await super()._check_sentence_with_lookahead(text)
-        if eos_marker:
+        if aggregation:
-            # Extract text up to the sentence boundary
+            # Convert to PatternMatch for consistency with return type
-            result = self._text[:eos_marker]
-            self._text = self._text[eos_marker:]
             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

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

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.string import StartEndTags, parse_start_end_tags
-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 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
+        This method adds the new text to the buffer, updates tag state,
-        pattern pairs, and returns processed text up to sentence boundaries if
+        and uses the parent's lookahead logic for sentence detection when
-        possible. If there are incomplete patterns (start without matching
+        not inside tags.
-        end), it will continue buffering text.
 
         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 inside tags, don't check for sentences
-        if not self._current_tag:
+        if self._current_tag:
-            eos_marker = match_endofsentence(self._text)
+            return None
-            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)
 
-        # No complete sentence found yet
+        # Otherwise, use parent's lookahead logic for sentence detection
-        return None
+        return await super()._check_sentence_with_lookahead(text)
-
-    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.
+        """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

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
+        # Feed text character by character
-        result = await self.aggregator.aggregate("Hello <test>pattern")
+        full_text = "Hello <test>pattern content</test>!"
-        self.assertIsNone(result)
+        result = None
-        self.assertEqual(self.aggregator.text.text, "Hello <test>pattern")
-        self.assertEqual(self.aggregator.text.type, "test_pattern")
 
-        # Second part completes the pattern and includes an exclamation point
+        for char in full_text:
-        result = await self.aggregator.aggregate(" content</test>!")
+            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
+        # Now flush to get the remaining sentence
-        # should be stripped in the returned Aggregation.
+        result = await self.aggregator.flush()
-        result = await self.aggregator.aggregate(" This is another sentence.")
         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
+        # Feed text character by character until we get the first aggregation
-        result = await self.aggregator.aggregate("Here is code <code>pattern")
+        result = None
+        for char in "Here is code <code>pattern content</code>":
+            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, "<code>pattern")
+        self.assertEqual(self.aggregator.text.text, "<code>pattern content</code>")
         self.assertEqual(self.aggregator.text.type, "code_pattern")
 
-        # Second part completes the pattern and includes an exclamation point
+        # Continue feeding to complete the pattern
-        result = await self.aggregator.aggregate(" content</code>")
+        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
+        # Feed text character by character with incomplete pattern
-        result = await self.aggregator.aggregate("Hello <test>pattern content")
+        result = None
+        for char in "Hello <test>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 <voice>female</voice> I am <em>very</em> 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 <em>very</em> 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
+        # Feed text character by character with incomplete pattern
-        result = await self.aggregator.aggregate("Hello <test>pattern")
+        result = None
+        for char in "Hello <test>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
+        # Feed text character by character - pattern spans multiple sentences
-        result = await self.aggregator.aggregate("Hello <test>This is sentence one.")
+        full_text = "Hello <test>This is sentence one. This is sentence two.</test> Final sentence."
+        result = None
 
-        # First sentence contains start of pattern but no end, so no complete pattern yet
+        for char in full_text:
-        self.assertIsNone(result)
+            result = await self.aggregator.aggregate(char)
-
+            if result:
-        # Add second part with pattern end
+                break
-        result = await self.aggregator.aggregate(" This is sentence two.</test> Final sentence.")
 
         # 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.")
 

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 <spell>foo@pipecat.ai</spell>.")
+        # Feed text character by character
+        result = None
+        for char in "My email is <spell>foo@pipecat.ai</spell>.":
+            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 <spell>foo@pipecat.ai</spell>.")
         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.
+        # Tags involved, feed character by character
-        result = await self.aggregator.aggregate("My email is <sp")
+        full_text = "My email is <spell>foo.bar@pipecat.ai</spell>."
-        self.assertIsNone(result)
+        result = None
-        self.assertEqual(self.aggregator.text.text, "My email is <sp")
 
-        result = await self.aggregator.aggregate("ell>foo.")
+        for char in full_text:
-        self.assertIsNone(result)
+            result = await self.aggregator.aggregate(char)
-        self.assertEqual(self.aggregator.text.text, "My email is <spell>foo.")
+            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 <spell>foo.bar@pipecat.")
+        self.assertEqual(self.aggregator.text.text, full_text)
-
-        result = await self.aggregator.aggregate("ai</spe")
-        self.assertIsNone(result)
-        self.assertEqual(self.aggregator.text.text, "My email is <spell>foo.bar@pipecat.ai</spe")
         self.assertEqual(self.aggregator.text.type, "sentence")
 
-        result = await self.aggregator.aggregate("ll>.")
+        # Flush to get the pending sentence
-        self.assertEqual(result.text, "My email is <spell>foo.bar@pipecat.ai</spell>.")
+        result = await self.aggregator.flush()
+        self.assertEqual(result.text, full_text)
         self.assertEqual(self.aggregator.text.text, "")
         self.assertEqual(self.aggregator.text.type, "sentence")