From 1dbad2326aeeb2fccda94e392be6cdd8393dba4f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 26 Feb 2025 17:23:06 -0800 Subject: [PATCH 1/7] utils(string): support email addresses in end of sentence matching --- CHANGELOG.md | 3 +++ src/pipecat/utils/string.py | 19 ++++++++++++++++++- tests/test_utils_string.py | 4 ++++ 3 files changed, 25 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 249474080..6e626a7b1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -138,6 +138,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- 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. diff --git a/src/pipecat/utils/string.py b/src/pipecat/utils/string.py index 154d03174..06f2fc175 100644 --- a/src/pipecat/utils/string.py +++ b/src/pipecat/utils/string.py @@ -17,9 +17,26 @@ 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,}") + def match_endofsentence(text: str) -> int: - match = ENDOFSENTENCE_PATTERN.search(text.rstrip()) + text = text.rstrip() + + # Find all emails. + emails = list(EMAIL_PATTERN.finditer(text)) + + # Replace email dots by ampersands so we can find the end of sentence. + for email_match in emails: + start = email_match.start() + end = email_match.end() + new_email = text[start:end].replace(".", "&") + text = text[:start] + new_email + text[end:] + + # Match against the new text. + match = ENDOFSENTENCE_PATTERN.search(text) + return match.end() if match else 0 diff --git a/tests/test_utils_string.py b/tests/test_utils_string.py index e4cdb4cb4..24519f724 100644 --- a/tests/test_utils_string.py +++ b/tests/test_utils_string.py @@ -18,6 +18,9 @@ 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 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 +31,7 @@ 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") async def test_endofsentence_zh(self): chinese_sentences = [ From 11984b89b79661d6d7ae2e0987446434dfe8021b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 26 Feb 2025 18:22:37 -0800 Subject: [PATCH 2/7] utils(string): add support for floating point numbers --- CHANGELOG.md | 3 +++ src/pipecat/utils/string.py | 28 ++++++++++++++++++++-------- tests/test_utils_string.py | 6 ++++++ 3 files changed, 29 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6e626a7b1..c530b0986 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -138,6 +138,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- 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. diff --git a/src/pipecat/utils/string.py b/src/pipecat/utils/string.py index 06f2fc175..a89127de9 100644 --- a/src/pipecat/utils/string.py +++ b/src/pipecat/utils/string.py @@ -8,7 +8,8 @@ import re ENDOFSENTENCE_PATTERN_STR = r""" (? str: + 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: text = text.rstrip() - # Find all emails. + # 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)) - - # Replace email dots by ampersands so we can find the end of sentence. for email_match in emails: - start = email_match.start() - end = email_match.end() - new_email = text[start:end].replace(".", "&") - text = text[:start] + new_email + text[end:] + 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) diff --git a/tests/test_utils_string.py b/tests/test_utils_string.py index 24519f724..ee0946d69 100644 --- a/tests/test_utils_string.py +++ b/tests/test_utils_string.py @@ -21,6 +21,11 @@ class TestUtilsString(unittest.IsolatedAsyncioTestCase): 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("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, ") @@ -32,6 +37,7 @@ class TestUtilsString(unittest.IsolatedAsyncioTestCase): 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 = [ From 1a3a268c9de7d3a0d59ea43da1cc7718c9c430bd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 28 Feb 2025 17:23:54 -0800 Subject: [PATCH 3/7] utils(string): add new function parse_start_end_tags() --- src/pipecat/utils/string.py | 76 +++++++++++++++++++++++++++++++++++++ tests/test_utils_string.py | 50 +++++++++++++++++++++++- 2 files changed, 125 insertions(+), 1 deletion(-) diff --git a/src/pipecat/utils/string.py b/src/pipecat/utils/string.py index a89127de9..69036a665 100644 --- a/src/pipecat/utils/string.py +++ b/src/pipecat/utils/string.py @@ -5,6 +5,7 @@ # import re +from typing import Optional, Sequence, Tuple ENDOFSENTENCE_PATTERN_STR = r""" (? 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) @@ -35,6 +51,20 @@ def replace_match(text: str, match: re.Match, old: str, new: str) -> str: def match_endofsentence(text: str) -> int: + """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 @@ -52,3 +82,49 @@ def match_endofsentence(text: str) -> int: 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) diff --git a/tests/test_utils_string.py b/tests/test_utils_string.py index ee0946d69..cabd88a36 100644 --- a/tests/test_utils_string.py +++ b/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): @@ -23,6 +23,7 @@ class TestUtilsString(unittest.IsolatedAsyncioTestCase): 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 foo.bar@pipecat.ai.") == 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 @@ -60,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): + # (, ) + assert parse_start_end_tags("Hello from Pipecat!", [("", "")], None, 0) == ( + None, + 26, + ) + assert parse_start_end_tags("Hello from Pipecat", [("", "")], None, 0) == ( + ("", ""), + 21, + ) + assert parse_start_end_tags("Hello from Pipecat", [("", "")], None, 6) == ( + ("", ""), + 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): + # (, ) + assert parse_start_end_tags( + "Hello from Pipecat! Hello World!", [("", "")], None, 0 + ) == ( + None, + 46, + ) + + assert parse_start_end_tags( + "Hello from Pipecat! Hello World", [("", "")], None, 0 + ) == ( + ("", ""), + 41, + ) 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 4/7] 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, "") From 54620133d4473e3f8d10d26112fbf002ece6b8ea Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 18 Mar 2025 23:08:34 -0700 Subject: [PATCH 5/7] services: add spelling out support to CartesiaTTSService and RimeTTSService --- CHANGELOG.md | 3 +++ src/pipecat/services/cartesia.py | 6 +++++- src/pipecat/services/rime.py | 6 +++++- 3 files changed, 13 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5abdf4ef5..3006ed5ca 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -142,6 +142,9 @@ 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. diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 8b7f57c63..5a795d750 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -7,7 +7,7 @@ import base64 import json import uuid -from typing import AsyncGenerator, List, Optional, Union +from typing import AsyncGenerator, List, Optional, Sequence, Union from loguru import logger from pydantic import BaseModel @@ -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_aggregators: Sequence[BaseTextAggregator] = [], **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_aggregators=text_aggregators or [SkipTagsAggregator([("", "")])], **kwargs, ) diff --git a/src/pipecat/services/rime.py b/src/pipecat/services/rime.py index b2610b06c..471f82d66 100644 --- a/src/pipecat/services/rime.py +++ b/src/pipecat/services/rime.py @@ -7,7 +7,7 @@ import base64 import json import uuid -from typing import AsyncGenerator, Optional +from typing import AsyncGenerator, Optional, Sequence import aiohttp from loguru import logger @@ -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_aggregators: Sequence[BaseTextAggregator] = [], **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_aggregators=text_aggregators or [SkipTagsAggregator([("spell(", ")")])], **kwargs, ) From fc0f404d26ddb8f7cb589c2f84c2c3ef96885f6d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 26 Feb 2025 17:23:58 -0800 Subject: [PATCH 6/7] examples: add new 36-user-email-gathering.py --- CHANGELOG.md | 4 + .../foundational/36-user-email-gathering.py | 141 ++++++++++++++++++ 2 files changed, 145 insertions(+) create mode 100644 examples/foundational/36-user-email-gathering.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 3006ed5ca..64f342dc5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -166,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 `` + 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 diff --git a/examples/foundational/36-user-email-gathering.py b/examples/foundational/36-user-email-gathering.py new file mode 100644 index 000000000..0e76826e3 --- /dev/null +++ b/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 `` 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 + "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 tags, for example a@a.com.", + # 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()) From 336e2f1579d53fe8d6c0fa336fc91546cc74618c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 19 Mar 2025 11:02:29 -0700 Subject: [PATCH 7/7] TTSServices: for now just specify a single text aggregator --- CHANGELOG.md | 4 +-- .../35-pattern-pair-voice-switching.py | 2 +- src/pipecat/services/ai_services.py | 29 ++++--------------- src/pipecat/services/cartesia.py | 6 ++-- src/pipecat/services/rime.py | 6 ++-- 5 files changed, 15 insertions(+), 32 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 64f342dc5..125ca20d8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,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) diff --git a/examples/foundational/35-pattern-pair-voice-switching.py b/examples/foundational/35-pattern-pair-voice-switching.py index 7d0094132..bb9587706 100644 --- a/examples/foundational/35-pattern-pair-voice-switching.py +++ b/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 diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 904e5cf90..eae030b27 100644 --- a/src/pipecat/services/ai_services.py +++ b/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) diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 5a795d750..3b491d26b 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -7,7 +7,7 @@ import base64 import json import uuid -from typing import AsyncGenerator, List, Optional, Sequence, Union +from typing import AsyncGenerator, List, Optional, Union from loguru import logger from pydantic import BaseModel @@ -91,7 +91,7 @@ class CartesiaTTSService(AudioContextWordTTSService): encoding: str = "pcm_s16le", container: str = "raw", params: InputParams = InputParams(), - text_aggregators: Sequence[BaseTextAggregator] = [], + text_aggregator: Optional[BaseTextAggregator] = None, **kwargs, ): # Aggregating sentences still gives cleaner-sounding results and fewer @@ -109,7 +109,7 @@ class CartesiaTTSService(AudioContextWordTTSService): push_text_frames=False, pause_frame_processing=True, sample_rate=sample_rate, - text_aggregators=text_aggregators or [SkipTagsAggregator([("", "")])], + text_aggregator=text_aggregator or SkipTagsAggregator([("", "")]), **kwargs, ) diff --git a/src/pipecat/services/rime.py b/src/pipecat/services/rime.py index 471f82d66..c6fc50001 100644 --- a/src/pipecat/services/rime.py +++ b/src/pipecat/services/rime.py @@ -7,7 +7,7 @@ import base64 import json import uuid -from typing import AsyncGenerator, Optional, Sequence +from typing import AsyncGenerator, Optional import aiohttp from loguru import logger @@ -80,7 +80,7 @@ class RimeTTSService(AudioContextWordTTSService): model: str = "mistv2", sample_rate: Optional[int] = None, params: InputParams = InputParams(), - text_aggregators: Sequence[BaseTextAggregator] = [], + text_aggregator: Optional[BaseTextAggregator] = None, **kwargs, ): """Initialize Rime TTS service. @@ -100,7 +100,7 @@ class RimeTTSService(AudioContextWordTTSService): push_stop_frames=True, pause_frame_processing=True, sample_rate=sample_rate, - text_aggregators=text_aggregators or [SkipTagsAggregator([("spell(", ")")])], + text_aggregator=text_aggregator or SkipTagsAggregator([("spell(", ")")]), **kwargs, )