From 535b85cf90238dc0b2703e9646c8d607ec9d79ad Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 25 Nov 2025 18:04:07 -0500 Subject: [PATCH] Add split_text_by_spaces string util --- CHANGELOG.md | 5 +++ .../aggregators/llm_text_processor.py | 22 +++++++----- src/pipecat/services/tts_service.py | 30 ++++++++++------ src/pipecat/utils/string.py | 24 +++++++++++++ tests/test_utils_string.py | 34 ++++++++++++++++++- 5 files changed, 96 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2891a53e5..4a5777d5a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- Updated `LLMTextProcessor` and `TTSService` to normalize text input by + splitting into individual characters before aggregation. This ensures proper + sentence boundary detection when LLMs return multiple sentences in a single + chunk (e.g., Google Gemini). + - Updated `AICFilter` to use Quail STT as the default model (`AICModelType.QUAIL_STT`). Quail STT is optimized for human-to-machine interaction (e.g., voice agents, speech-to-text) and operates at a native diff --git a/src/pipecat/processors/aggregators/llm_text_processor.py b/src/pipecat/processors/aggregators/llm_text_processor.py index 0d9636fd8..0a99e08bc 100644 --- a/src/pipecat/processors/aggregators/llm_text_processor.py +++ b/src/pipecat/processors/aggregators/llm_text_processor.py @@ -24,6 +24,7 @@ from pipecat.frames.frames import ( LLMTextFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.utils.string import split_text_by_characters from pipecat.utils.text.base_text_aggregator import BaseTextAggregator from pipecat.utils.text.simple_text_aggregator import SimpleTextAggregator @@ -83,14 +84,19 @@ class LLMTextProcessor(FrameProcessor): await self._text_aggregator.reset() async def _handle_llm_text(self, in_frame: LLMTextFrame): - aggregation = await self._text_aggregator.aggregate(in_frame.text) - if aggregation: - out_frame = AggregatedTextFrame( - text=aggregation.text, - aggregated_by=aggregation.type, - ) - out_frame.skip_tts = in_frame.skip_tts - await self.push_frame(out_frame) + # Split text by characters to normalize LLM output into individual characters + # This ensures consistent aggregation behavior regardless of LLM chunk size + characters = split_text_by_characters(in_frame.text) + + for character in characters: + aggregation = await self._text_aggregator.aggregate(character) + if aggregation: + out_frame = AggregatedTextFrame( + text=aggregation.text, + aggregated_by=aggregation.type, + ) + out_frame.skip_tts = in_frame.skip_tts + await self.push_frame(out_frame) async def _handle_llm_end(self, skip_tts: Optional[bool] = None): # Flush any remaining aggregated text at the end of the LLM response diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index 977d6957b..e8720519d 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -51,6 +51,7 @@ from pipecat.processors.frame_processor import FrameDirection from pipecat.services.ai_service import AIService from pipecat.services.websocket_service import WebsocketService from pipecat.transcriptions.language import Language +from pipecat.utils.string import split_text_by_characters from pipecat.utils.text.base_text_aggregator import BaseTextAggregator from pipecat.utils.text.base_text_filter import BaseTextFilter from pipecat.utils.text.simple_text_aggregator import SimpleTextAggregator @@ -539,17 +540,26 @@ class TTSService(AIService): text = frame.text includes_inter_frame_spaces = frame.includes_inter_frame_spaces aggregated_by = "token" - else: - aggregate = await self._text_aggregator.aggregate(frame.text) - if aggregate: - text = aggregate.text - aggregated_by = aggregate.type - if text: - logger.trace(f"Pushing TTS frames for text: {text}, {aggregated_by}") - await self._push_tts_frames( - AggregatedTextFrame(text, aggregated_by), includes_inter_frame_spaces - ) + if text: + logger.trace(f"Pushing TTS frames for text: {text}, {aggregated_by}") + await self._push_tts_frames( + AggregatedTextFrame(text, aggregated_by), includes_inter_frame_spaces + ) + else: + # Split text by characters to normalize input into individual characters + # This ensures consistent aggregation behavior regardless of input chunk size + characters = split_text_by_characters(frame.text) + + for character in characters: + aggregate = await self._text_aggregator.aggregate(character) + if aggregate: + text = aggregate.text + aggregated_by = aggregate.type + logger.trace(f"Pushing TTS frames for text: {text}, {aggregated_by}") + await self._push_tts_frames( + AggregatedTextFrame(text, aggregated_by), includes_inter_frame_spaces + ) async def _push_tts_frames( self, src_frame: AggregatedTextFrame, includes_inter_frame_spaces: Optional[bool] = False diff --git a/src/pipecat/utils/string.py b/src/pipecat/utils/string.py index 5645d2bcf..a0ed9ca2d 100644 --- a/src/pipecat/utils/string.py +++ b/src/pipecat/utils/string.py @@ -280,3 +280,27 @@ def concatenate_aggregated_text(text_parts: List[TextPartForConcatenation]) -> s result = result.strip() return result + + +def split_text_by_characters(text: str) -> List[str]: + """Split text into individual characters. + + Returns each character as a separate string element, allowing character-by-character + processing while maintaining the ability to reconstruct the original text. + + Args: + text: The text to split into characters. + + Returns: + A list of individual characters. + + Example:: + + >>> split_text_by_characters("Hello world!") + ["H", "e", "l", "l", "o", " ", "w", "o", "r", "l", "d", "!"] + >>> split_text_by_characters("Hi") + ["H", "i"] + >>> split_text_by_characters("") + [] + """ + return list(text) diff --git a/tests/test_utils_string.py b/tests/test_utils_string.py index 607b8d0fa..5140964dd 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, parse_start_end_tags +from pipecat.utils.string import match_endofsentence, parse_start_end_tags, split_text_by_characters class TestUtilsString(unittest.IsolatedAsyncioTestCase): @@ -232,3 +232,35 @@ class TestStartEndTags(unittest.IsolatedAsyncioTestCase): ("", ""), 41, ) + + async def test_split_text_by_characters(self): + """Test splitting text into individual characters.""" + # Basic sentence + assert split_text_by_characters("Hello world!") == [ + "H", + "e", + "l", + "l", + "o", + " ", + "w", + "o", + "r", + "l", + "d", + "!", + ] + + # Single word + assert split_text_by_characters("Hi") == ["H", "i"] + + # Empty string + assert split_text_by_characters("") == [] + + # With spaces + assert split_text_by_characters("A B") == ["A", " ", "B"] + + # Concatenation test - characters should concatenate back to original + characters = split_text_by_characters("Hello world!") + concatenated = "".join(characters) + assert concatenated == "Hello world!"