Merge pull request #1304 from pipecat-ai/aleix/handle-emails-user-email-gathering

add skip tags aggregator to support TTS service spelling out tags

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
@@ -17,8 +21,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Added new `BaseTextAggregator`. Text aggregators are used by the TTS service
   to aggregate LLM tokens and decide when the aggregated text should be pushed
   to the TTS service. They also allow for the text to be manipulated while it's
-  being aggregated. Multiple text aggregators can be passed with
-  `text_aggregators` to the TTS service.
+  being aggregated. A text aggregator can be passed via `text_aggregator` to the
+  TTS service.
 
 - Added new `UltravoxSTTService`.
   (see https://github.com/fixie-ai/ultravox)
@@ -138,6 +142,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed a `CartesiaTTSService` and `RimeTTSService` issue that would consider
+  text between spelling out tags end of sentence.
+
+- Fixed a `match_endofsentence` issue that would result in floating point
+  numbers to be considered an end of sentence.
+
+- Fixed a `match_endofsentence` issue that would result in emails to be
+  considered an end of sentence.
+
 - Fixed an issue where the RTVI message `disconnect-bot` was pushing an
   `EndFrame`, resulting in the pipeline not shutting down. It now pushes an
   `EndTaskFrame` upstream to shutdown the pipeline.
@@ -153,6 +166,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Other
 
+- Added a new example `examples/foundational/36-user-email-gathering.py` to show
+  how to gather user emails. The example uses's Cartesia's `<spell></spell>`
+  tags and Rime `spell()` function to spell out the emails for confirmation.
+
 - Update the `34-audio-recording.py` example to include an STT processor.
 
 - Added foundational example `35-voice-switching.py` showing how to use the new

examples/foundational/35-pattern-pair-voice-switching.py

@@ -119,7 +119,7 @@ async def main():
         tts = CartesiaTTSService(
             api_key=os.getenv("CARTESIA_API_KEY"),
             voice_id=VOICE_IDS["narrator"],
-            text_aggregators=[pattern_aggregator],
+            text_aggregator=pattern_aggregator,
         )
 
         # Initialize LLM

examples/foundational/36-user-email-gathering.py

@@ -0,0 +1,141 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+import sys
+
+import aiohttp
+from dotenv import load_dotenv
+from loguru import logger
+from openai.types.chat import ChatCompletionToolParam
+from runner import configure
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.openai import OpenAILLMContext, OpenAILLMService
+from pipecat.services.rime import RimeHttpTTSService
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+async def store_user_emails(function_name, tool_call_id, args, llm, context, result_callback):
+    print(f"User emails: {args}")
+
+
+async def main():
+    async with aiohttp.ClientSession() as session:
+        (room_url, token) = await configure(session)
+
+        transport = DailyTransport(
+            room_url,
+            token,
+            "Respond bot",
+            DailyParams(
+                audio_out_enabled=True,
+                transcription_enabled=True,
+                vad_enabled=True,
+                vad_analyzer=SileroVADAnalyzer(),
+            ),
+        )
+
+        # Cartesia offers a `<spell></spell>` tags that we can use to ask the user
+        # to confirm the emails.
+        # (see https://docs.cartesia.ai/build-with-sonic/formatting-text-for-sonic/spelling-out-input-text)
+        tts = CartesiaTTSService(
+            api_key=os.getenv("CARTESIA_API_KEY"),
+            voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+            aiohttp_session=session,
+        )
+
+        # Rime offers a function `spell()` that we can use to ask the user
+        # to confirm the emails.
+        # (see https://docs.rime.ai/api-reference/spell)
+        # tts = RimeHttpTTSService(
+        #     api_key=os.getenv("RIME_API_KEY", ""),
+        #     voice_id="eva",
+        #     aiohttp_session=session,
+        # )
+
+        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
+        # You can aslo register a function_name of None to get all functions
+        # sent to the same callback with an additional function_name parameter.
+        llm.register_function("store_user_emails", store_user_emails)
+
+        tools = [
+            ChatCompletionToolParam(
+                type="function",
+                function={
+                    "name": "store_user_emails",
+                    "description": "Store user emails when confirmed",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {
+                            "emails": {
+                                "type": "array",
+                                "description": "The list of user emails",
+                                "items": {"type": "string"},
+                            },
+                        },
+                        "required": ["emails"],
+                    },
+                },
+            )
+        ]
+        messages = [
+            {
+                "role": "system",
+                # Cartesia <spell></spell>
+                "content": "You need to gather a valid email or emails from the user. Your output will be converted to audio so don't include special characters in your answers. If the user provides one or more email addresses confirm them with the user. Enclose all emails with <spell> tags, for example <spell>a@a.com</spell>.",
+                # Rime spell()
+                # "content": "You need to gather a valid email or emails from the user. Your output will be converted to audio so don't include special characters in your answers. If the user provides one or more email addresses confirm them with the user. Enclose all emails with spell(), for example spell(a@a.com).",
+            },
+        ]
+
+        context = OpenAILLMContext(messages, tools)
+        context_aggregator = llm.create_context_aggregator(context)
+
+        pipeline = Pipeline(
+            [
+                transport.input(),
+                context_aggregator.user(),
+                llm,
+                tts,
+                transport.output(),
+                context_aggregator.assistant(),
+            ]
+        )
+
+        task = PipelineTask(
+            pipeline,
+            params=PipelineParams(
+                allow_interruptions=True,
+                enable_metrics=True,
+                enable_usage_metrics=True,
+                report_only_initial_ttfb=True,
+            ),
+        )
+
+        @transport.event_handler("on_first_participant_joined")
+        async def on_first_participant_joined(transport, participant):
+            await transport.capture_participant_transcription(participant["id"])
+            # Kick off the conversation.
+            await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+        runner = PipelineRunner()
+
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/pipecat/services/ai_services.py

@@ -239,7 +239,7 @@ class TTSService(AIService):
         # TTS output sample rate
         sample_rate: Optional[int] = None,
         # Text aggregator to aggregate incoming tokens and decide when to push to the TTS.
-        text_aggregators: Sequence[BaseTextAggregator] = [],
+        text_aggregator: Optional[BaseTextAggregator] = None,
         # Text filter executed after text has been aggregated.
         text_filters: Sequence[BaseTextFilter] = [],
         text_filter: Optional[BaseTextFilter] = None,
@@ -257,10 +257,7 @@ class TTSService(AIService):
         self._sample_rate = 0
         self._voice_id: str = ""
         self._settings: Dict[str, Any] = {}
-        # Ensure there's at least one text aggregator.
-        self._text_aggregators: Sequence[BaseTextAggregator] = text_aggregators or [
-            SimpleTextAggregator()
-        ]
+        self._text_aggregator: BaseTextAggregator = text_aggregator or SimpleTextAggregator()
         self._text_filters: Sequence[BaseTextFilter] = text_filters
         if text_filter:
             import warnings
@@ -358,8 +355,8 @@ class TTSService(AIService):
             # pause to avoid audio overlapping.
             await self._maybe_pause_frame_processing()
 
-            sentence = self._text_aggregators[-1].text
-            self._reset_aggregators()
+            sentence = self._text_aggregator.text
+            self._text_aggregator.reset()
             self._processing_text = False
             await self._push_tts_frames(sentence)
             if isinstance(frame, LLMFullResponseEndFrame):
@@ -405,8 +402,7 @@ class TTSService(AIService):
 
     async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection):
         self._processing_text = False
-        for aggregator in self._text_aggregators:
-            aggregator.handle_interruption()
+        self._text_aggregator.handle_interruption()
         for filter in self._text_filters:
             filter.handle_interruption()
 
@@ -418,25 +414,12 @@ class TTSService(AIService):
         if self._pause_frame_processing:
             await self.resume_processing_frames()
 
-    def _reset_aggregators(self):
-        for aggregator in self._text_aggregators:
-            aggregator.reset()
-
     async def _process_text_frame(self, frame: TextFrame):
         text: Optional[str] = None
         if not self._aggregate_sentences:
             text = frame.text
         else:
-            current_text = frame.text
-
-            # Process all aggregators except the last one.
-            for aggregator in self._text_aggregators[:-1]:
-                aggregator.aggregate(current_text)
-                current_text = aggregator.text
-
-            # The last aggregator decides whether we are sending text to the
-            # TTS or not.
-            text = self._text_aggregators[-1].aggregate(current_text)
+            text = self._text_aggregator.aggregate(frame.text)
 
         if text:
             await self._push_tts_frames(text)

src/pipecat/services/cartesia.py

@@ -26,6 +26,8 @@ from pipecat.frames.frames import (
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_services import AudioContextWordTTSService, TTSService
 from pipecat.transcriptions.language import Language
+from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
+from pipecat.utils.text.skip_tags_aggregator import SkipTagsAggregator
 
 # See .env.example for Cartesia configuration needed
 try:
@@ -89,6 +91,7 @@ class CartesiaTTSService(AudioContextWordTTSService):
         encoding: str = "pcm_s16le",
         container: str = "raw",
         params: InputParams = InputParams(),
+        text_aggregator: Optional[BaseTextAggregator] = None,
         **kwargs,
     ):
         # Aggregating sentences still gives cleaner-sounding results and fewer
@@ -106,6 +109,7 @@ class CartesiaTTSService(AudioContextWordTTSService):
             push_text_frames=False,
             pause_frame_processing=True,
             sample_rate=sample_rate,
+            text_aggregator=text_aggregator or SkipTagsAggregator([("<spell>", "</spell>")]),
             **kwargs,
         )
 

src/pipecat/services/rime.py

@@ -27,6 +27,8 @@ from pipecat.frames.frames import (
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_services import AudioContextWordTTSService, TTSService
 from pipecat.transcriptions.language import Language
+from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
+from pipecat.utils.text.skip_tags_aggregator import SkipTagsAggregator
 
 try:
     import websockets
@@ -78,6 +80,7 @@ class RimeTTSService(AudioContextWordTTSService):
         model: str = "mistv2",
         sample_rate: Optional[int] = None,
         params: InputParams = InputParams(),
+        text_aggregator: Optional[BaseTextAggregator] = None,
         **kwargs,
     ):
         """Initialize Rime TTS service.
@@ -97,6 +100,7 @@ class RimeTTSService(AudioContextWordTTSService):
             push_stop_frames=True,
             pause_frame_processing=True,
             sample_rate=sample_rate,
+            text_aggregator=text_aggregator or SkipTagsAggregator([("spell(", ")")]),
             **kwargs,
         )
 

src/pipecat/utils/string.py

@@ -5,10 +5,12 @@
 #
 
 import re
+from typing import Optional, Sequence, Tuple
 
 ENDOFSENTENCE_PATTERN_STR = r"""
     (?<![A-Z])       # Negative lookbehind: not preceded by an uppercase letter (e.g., "U.S.A.")
-    (?<!\d)          # Negative lookbehind: not preceded by a digit (e.g., "1. Let's start")
+    (?<!\d\.\d)      # Not preceded by a decimal number (e.g., "3.14159")
+    (?<!^\d\.)       # Not preceded by a numbered list item (e.g., "1. Let's start")
     (?<!\d\s[ap])    # Negative lookbehind: not preceded by time (e.g., "3:00 a.m.")
     (?<!Mr|Ms|Dr)    # Negative lookbehind: not preceded by Mr, Ms, Dr (combined bc. length is the same)
     (?<!Mrs)         # Negative lookbehind: not preceded by "Mrs"
@@ -17,9 +19,112 @@ ENDOFSENTENCE_PATTERN_STR = r"""
     (\。\s*\。\s*\。|[。？！；।])  # the full-width version (mainly used in East Asian languages such as Chinese, Hindi)
     $                # End of string
 """
+
 ENDOFSENTENCE_PATTERN = re.compile(ENDOFSENTENCE_PATTERN_STR, re.VERBOSE)
 
+EMAIL_PATTERN = re.compile(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}")
+
+NUMBER_PATTERN = re.compile(r"[+-]?(\d+(\.\d*)?|\.\d+)([eE][+-]?\d+)?")
+
+StartEndTags = Tuple[str, str]
+
+
+def replace_match(text: str, match: re.Match, old: str, new: str) -> str:
+    """Replace occurrences of a substring within a matched section of a given
+    text.
+
+    Args:
+        text (str): The input text in which replacements will be made.
+        match (re.Match): A regex match object representing the section of text to modify.
+        old (str): The substring to be replaced.
+        new (str): The substring to replace `old` with.
+
+    Returns:
+        str: The modified text with the specified replacements made within the matched section.
+
+    """
+    start = match.start()
+    end = match.end()
+    replacement = text[start:end].replace(old, new)
+    text = text[:start] + replacement + text[end:]
+    return text
+
 
 def match_endofsentence(text: str) -> int:
-    match = ENDOFSENTENCE_PATTERN.search(text.rstrip())
+    """Finds the position of the end of a sentence in the provided text string.
+
+    This function processes the input text by replacing periods in email
+    addresses and numbers with ampersands to prevent them from being
+    misidentified as sentence terminals. It then searches for the end of a
+    sentence using a specified regex pattern.
+
+    Args:
+        text (str): The input text in which to find the end of the sentence.
+
+    Returns:
+        int: The position of the end of the sentence if found, otherwise 0.
+
+    """
+    text = text.rstrip()
+
+    # Replace email dots by ampersands so we can find the end of sentence. For
+    # example, first.last@email.com becomes first&last@email&com.
+    emails = list(EMAIL_PATTERN.finditer(text))
+    for email_match in emails:
+        text = replace_match(text, email_match, ".", "&")
+
+    # Replace number dots by ampersands so we can find the end of sentence.
+    numbers = list(NUMBER_PATTERN.finditer(text))
+    for number_match in numbers:
+        text = replace_match(text, number_match, ".", "&")
+
+    # Match against the new text.
+    match = ENDOFSENTENCE_PATTERN.search(text)
+
     return match.end() if match else 0
+
+
+def parse_start_end_tags(
+    text: str,
+    tags: Sequence[StartEndTags],
+    current_tag: Optional[StartEndTags],
+    current_tag_index: int,
+) -> Tuple[Optional[StartEndTags], int]:
+    """Parses the given text to identify a pair of start/end tags.
+
+    If a start tag was previously found (i.e. current_tags is valid), wait for
+    the corresponding end tag.  Otherwise, wait for a start tag.
+
+    This function will return the index in the text that we should start parsing
+    in the next call and the current or new tags.
+
+    Parameters:
+    - text (str): The text to be parsed.
+    - tags (Sequence[StartEndTags]): List of tuples containing start and end tags.
+    - current_tags (Optional[StartEndTags]): The currently active tags, if any.
+    - current_tags_index (int): The current index in the text.
+
+    Returns:
+    Tuple[Optional[StartEndTags], int]: A tuple containing None or the current
+    tag and the index of the text.
+
+    """
+    # If we are already inside a tag, check if the end tag is in the text.
+    if current_tag:
+        _, end_tag = current_tag
+        if end_tag in text[current_tag_index:]:
+            return (None, len(text))
+        return (current_tag, current_tag_index)
+
+    # Check if any start tag appears in the text
+    for start_tag, end_tag in tags:
+        start_tag_count = text[current_tag_index:].count(start_tag)
+        end_tag_count = text[current_tag_index:].count(end_tag)
+        if start_tag_count == 0 and end_tag_count == 0:
+            return (None, current_tag_index)
+        elif start_tag_count > end_tag_count:
+            return ((start_tag, end_tag), len(text))
+        elif start_tag_count == end_tag_count:
+            return (None, len(text))
+
+    return (None, current_tag_index)

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

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([("<spell>", "</spell>")])
+
+    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 <spell>foo@pipecat.ai</spell>.")
+        self.assertEqual(result, "My email is <spell>foo@pipecat.ai</spell>.")
+        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 <sp")
+        self.assertIsNone(result)
+        self.assertEqual(self.aggregator.text, "My email is <sp")
+
+        result = self.aggregator.aggregate("ell>foo.")
+        self.assertIsNone(result)
+        self.assertEqual(self.aggregator.text, "My email is <spell>foo.")
+
+        result = self.aggregator.aggregate("bar@pipecat.")
+        self.assertIsNone(result)
+        self.assertEqual(self.aggregator.text, "My email is <spell>foo.bar@pipecat.")
+
+        result = self.aggregator.aggregate("ai</spe")
+        self.assertIsNone(result)
+        self.assertEqual(self.aggregator.text, "My email is <spell>foo.bar@pipecat.ai</spe")
+
+        result = self.aggregator.aggregate("ll>.")
+        self.assertEqual(result, "My email is <spell>foo.bar@pipecat.ai</spell>.")
+        self.assertEqual(self.aggregator.text, "")

tests/test_utils_string.py

@@ -6,7 +6,7 @@
 
 import unittest
 
-from pipecat.utils.string import match_endofsentence
+from pipecat.utils.string import match_endofsentence, parse_start_end_tags
 
 
 class TestUtilsString(unittest.IsolatedAsyncioTestCase):
@@ -18,6 +18,15 @@ class TestUtilsString(unittest.IsolatedAsyncioTestCase):
         assert match_endofsentence("This is a sentence...") == 21
         assert match_endofsentence("This is a sentence . . .") == 24
         assert match_endofsentence("This is a sentence. ..") == 22
+        assert match_endofsentence("This is for Mr. and Mrs. Jones.") == 31
+        assert match_endofsentence("U.S.A and U.S.A..") == 17
+        assert match_endofsentence("My emails are foo@pipecat.ai and bar@pipecat.ai.") == 48
+        assert match_endofsentence("My email is foo.bar@pipecat.ai.") == 31
+        assert match_endofsentence("My email is spell(foo.bar@pipecat.ai).") == 38
+        assert match_endofsentence("My email is <spell>foo.bar@pipecat.ai</spell>.") == 46
+        assert match_endofsentence("The number pi is 3.14159.") == 25
+        assert match_endofsentence("Valid scientific notation 1.23e4.") == 33
+        assert match_endofsentence("Valid scientific notation 0.e4.") == 31
         assert not match_endofsentence("This is not a sentence")
         assert not match_endofsentence("This is not a sentence,")
         assert not match_endofsentence("This is not a sentence, ")
@@ -28,6 +37,8 @@ class TestUtilsString(unittest.IsolatedAsyncioTestCase):
         assert not match_endofsentence("Heute ist Dienstag, der 3.")  # 3. Juli 2024
         assert not match_endofsentence("America, or the U.")  # U.S.A.
         assert not match_endofsentence("It still early, it's 3:00 a.")  # 3:00 a.m.
+        assert not match_endofsentence("My emails are foo@pipecat.ai and bar@pipecat.ai")
+        assert not match_endofsentence("The number pi is 3.14159")
 
     async def test_endofsentence_zh(self):
         chinese_sentences = [
@@ -50,3 +61,50 @@ class TestUtilsString(unittest.IsolatedAsyncioTestCase):
         for i in hindi_sentences:
             assert match_endofsentence(i)
         assert not match_endofsentence("हैलो，")
+
+
+class TestStartEndTags(unittest.IsolatedAsyncioTestCase):
+    async def test_empty(self):
+        assert parse_start_end_tags("", [], None, 0) == (None, 0)
+        assert parse_start_end_tags("Hello from Pipecat!", [], None, 0) == (None, 0)
+
+    async def test_simple(self):
+        # (<a>, </a>)
+        assert parse_start_end_tags("Hello from <a>Pipecat</a>!", [("<a>", "</a>")], None, 0) == (
+            None,
+            26,
+        )
+        assert parse_start_end_tags("Hello from <a>Pipecat", [("<a>", "</a>")], None, 0) == (
+            ("<a>", "</a>"),
+            21,
+        )
+        assert parse_start_end_tags("Hello from <a>Pipecat", [("<a>", "</a>")], None, 6) == (
+            ("<a>", "</a>"),
+            21,
+        )
+
+        # (spell(, ))
+        assert parse_start_end_tags("Hello from spell(Pipecat)!", [("spell(", ")")], None, 0) == (
+            None,
+            26,
+        )
+        assert parse_start_end_tags("Hello from spell(Pipecat", [("spell(", ")")], None, 0) == (
+            ("spell(", ")"),
+            24,
+        )
+
+    async def test_multiple(self):
+        # (<a>, </a>)
+        assert parse_start_end_tags(
+            "Hello from <a>Pipecat</a>! Hello <a>World</a>!", [("<a>", "</a>")], None, 0
+        ) == (
+            None,
+            46,
+        )
+
+        assert parse_start_end_tags(
+            "Hello from <a>Pipecat</a>! Hello <a>World", [("<a>", "</a>")], None, 0
+        ) == (
+            ("<a>", "</a>"),
+            41,
+        )