From e7224473f2728e37c95aee52d48b9099a16bbdc6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 18 Mar 2025 23:04:14 -0700 Subject: [PATCH] utils(text): add new SkipTagsAggregator --- CHANGELOG.md | 4 + .../utils/text/skip_tags_aggregator.py | 94 +++++++++++++++++++ tests/test_skip_tags_aggregator.py | 54 +++++++++++ 3 files changed, 152 insertions(+) create mode 100644 src/pipecat/utils/text/skip_tags_aggregator.py create mode 100644 tests/test_skip_tags_aggregator.py diff --git a/CHANGELOG.md b/CHANGELOG.md index c530b0986..5abdf4ef5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added new `SkipTagsAggregator` that extends `BaseTextAggregator` to aggregate + text and skips end of sentence matching if aggregated text is between + start/end tags. + - Added new `PatternPairAggregator` that extends `BaseTextAggregator` to identify content between matching pattern pairs in streamed text. This allows for detection and processing of structured content like XML-style tags that diff --git a/src/pipecat/utils/text/skip_tags_aggregator.py b/src/pipecat/utils/text/skip_tags_aggregator.py new file mode 100644 index 000000000..00129028e --- /dev/null +++ b/src/pipecat/utils/text/skip_tags_aggregator.py @@ -0,0 +1,94 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +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 BaseTextAggregator + + +class SkipTagsAggregator(BaseTextAggregator): + """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 + tag. If a start tag is found the aggregator will keep aggregating text + unconditionally until the corresponding end tag is found. It's particularly + useful for processing content with custom delimiters that should prevent + text from being considered for end of sentence matching.. + + The aggregator ensures that tags spanning multiple text chunks are correctly + identified. + + """ + + def __init__(self, tags: Sequence[StartEndTags]): + """Initialize the pattern pair aggregator. + + Creates an empty aggregator with no patterns or handlers registered. + """ + self._text = "" + self._tags = tags + self._current_tag: Optional[StartEndTags] = None + self._current_tag_index: int = 0 + + @property + def text(self) -> str: + """Get the currently buffered text. + + Returns: + The current text buffer content. + """ + return self._text + + def aggregate(self, text: str) -> Optional[str]: + """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. + + Args: + text: New text to add to the buffer. + + Returns: + Processed text up to a sentence boundary, or None if more + text is needed to form a complete sentence or pattern. + """ + # Add new text to buffer + self._text += text + + (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 result + + # No complete sentence found yet + return None + + def handle_interruption(self): + """Handle interruptions by clearing the buffer. + + Called when an interruption occurs in the processing pipeline, + to reset the state and discard any partially aggregated text. + """ + self._text = "" + + def reset(self): + """Clear the internally aggregated text. + + Resets the aggregator to its initial state, discarding any + buffered text. + """ + self._text = "" diff --git a/tests/test_skip_tags_aggregator.py b/tests/test_skip_tags_aggregator.py new file mode 100644 index 000000000..8f36c4c05 --- /dev/null +++ b/tests/test_skip_tags_aggregator.py @@ -0,0 +1,54 @@ +# +# Copyright (c) 2024-2025 Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import unittest + +from pipecat.utils.text.skip_tags_aggregator import SkipTagsAggregator + + +class TestSkipTagsAggregator(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.aggregator = SkipTagsAggregator([("", "")]) + + async def test_no_tags(self): + self.aggregator.reset() + + # No tags involved, aggregate at end of sentence. + result = self.aggregator.aggregate("Hello Pipecat!") + self.assertEqual(result, "Hello Pipecat!") + self.assertEqual(self.aggregator.text, "") + + async def test_basic_tags(self): + self.aggregator.reset() + + # Tags involved, avoid aggregation during tags. + result = self.aggregator.aggregate("My email is foo@pipecat.ai.") + self.assertEqual(result, "My email is foo@pipecat.ai.") + self.assertEqual(self.aggregator.text, "") + + async def test_streaming_tags(self): + self.aggregator.reset() + + # Tags involved, stream small chunk of texts. + result = self.aggregator.aggregate("My email is foo.") + self.assertIsNone(result) + self.assertEqual(self.aggregator.text, "My email is foo.") + + result = self.aggregator.aggregate("bar@pipecat.") + self.assertIsNone(result) + self.assertEqual(self.aggregator.text, "My email is foo.bar@pipecat.") + + result = self.aggregator.aggregate("aifoo.bar@pipecat.ai.") + self.assertEqual(result, "My email is foo.bar@pipecat.ai.") + self.assertEqual(self.aggregator.text, "")