From d79dd94019928a71f7f5c55bdcee19eac5d8663b Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 3 Dec 2025 14:46:30 -0500 Subject: [PATCH] Make aggregate return an AsyncIterator, other clean up --- CHANGELOG.md | 29 ++-- src/pipecat/extensions/ivr/ivr_navigator.py | 13 +- .../aggregators/llm_text_processor.py | 21 +-- src/pipecat/services/tts_service.py | 21 +-- src/pipecat/utils/string.py | 24 --- .../utils/text/base_text_aggregator.py | 33 ++--- .../utils/text/pattern_pair_aggregator.py | 140 +++++++++--------- .../utils/text/simple_text_aggregator.py | 39 ++--- .../utils/text/skip_tags_aggregator.py | 44 +++--- tests/test_ivr_navigation.py | 3 + tests/test_pattern_pair_aggregator.py | 68 +++------ tests/test_simple_text_aggregator.py | 125 +++++++--------- tests/test_skip_tags_aggregator.py | 37 ++--- tests/test_utils_string.py | 34 +---- uv.lock | 6 +- 15 files changed, 259 insertions(+), 378 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c571b374f..52606239d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,18 +15,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- Improved interruption handling to prevent bots from repeating themselves. + LLM services that return multiple sentences in a single response (e.g., + `GoogleLLMService`) are now split into individual sentences before being sent + to TTS. This ensures interruptions occur at sentence boundaries, preventing + the bot from repeating content after being interrupted during long responses. + - Text Aggregation Improvements: - - 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). - - All text aggregators now properly support character-by-character streaming - input. + - **Breaking Change**: `BaseTextAggregator.aggregate()` now returns + `AsyncIterator[Aggregation]` instead of `Optional[Aggregation]`. This + enables the aggregator to return multiple results based on the provided + text. - Refactored text aggregators to use inheritance: `SkipTagsAggregator` and `PatternPairAggregator` now inherit from `SimpleTextAggregator`, reusing - its lookahead-based sentence detection logic via - `_check_sentence_with_lookahead()`. + the base class's sentence detection logic. - Updated `AICFilter` to use Quail STT as the default model (`AICModelType.QUAIL_STT`). Quail STT is optimized for human-to-machine @@ -54,6 +57,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed an issue where `LLMTextFrame.skip_tts` was being overwritten by LLM services. +- Fixed sentence aggregation to correctly handle ambiguous punctuation in + streaming text, such as currency ("$29.95") and abbreviations ("Mr. Smith"). + +- Fixed bug in `PatternPairAggregator` where pattern handlers could be called + multiple times for `KEEP` or `AGGREGATE` patterns. + - Fixed an issue in `SarvamTTSService` where the last sentence was not being spoken. Now, audio is flushed when the TTS services receives the `LLMFullResponseEndFrame` or `EndFrame`. @@ -66,10 +75,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 voice-ui-kit's conversational panel rending of the LLM output after a function call. -- Fixed a bug in `PatternPairAggregator` where pattern handlers could be called - multiple times for patterns with `MatchAction.KEEP` or `MatchAction.AGGREGATE` - actions. - ## [0.0.96] - 2025-11-26 🦃 "Happy Thanksgiving!" 🦃 ### Added diff --git a/src/pipecat/extensions/ivr/ivr_navigator.py b/src/pipecat/extensions/ivr/ivr_navigator.py index 1ddb41ed8..505224cde 100644 --- a/src/pipecat/extensions/ivr/ivr_navigator.py +++ b/src/pipecat/extensions/ivr/ivr_navigator.py @@ -18,8 +18,10 @@ from loguru import logger from pipecat.audio.dtmf.types import KeypadEntry from pipecat.audio.vad.vad_analyzer import VADParams from pipecat.frames.frames import ( + EndFrame, Frame, LLMContextFrame, + LLMFullResponseEndFrame, LLMMessagesUpdateFrame, LLMTextFrame, OutputDTMFUrgentFrame, @@ -149,11 +151,18 @@ class IVRProcessor(FrameProcessor): elif isinstance(frame, LLMTextFrame): # Process text through the pattern aggregator - result = await self._aggregator.aggregate(frame.text) - if result: + async for result in self._aggregator.aggregate(frame.text): # Push aggregated text that doesn't contain XML patterns await self.push_frame(LLMTextFrame(result.text), direction) + elif isinstance(frame, (LLMFullResponseEndFrame, EndFrame)): + # Flush any remaining text from the aggregator + remaining = await self._aggregator.flush() + if remaining: + await self.push_frame(LLMTextFrame(remaining.text), direction) + # Push the end frame + await self.push_frame(frame, direction) + else: await self.push_frame(frame, direction) diff --git a/src/pipecat/processors/aggregators/llm_text_processor.py b/src/pipecat/processors/aggregators/llm_text_processor.py index 8cdb40127..f0de46617 100644 --- a/src/pipecat/processors/aggregators/llm_text_processor.py +++ b/src/pipecat/processors/aggregators/llm_text_processor.py @@ -24,7 +24,6 @@ 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 @@ -84,19 +83,13 @@ class LLMTextProcessor(FrameProcessor): await self._text_aggregator.reset() async def _handle_llm_text(self, in_frame: LLMTextFrame): - # 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 for aggregation in self._text_aggregator.aggregate(in_frame.text): + 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 text diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index 74050550a..ca15cb2c0 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -51,7 +51,6 @@ 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 @@ -544,19 +543,13 @@ class TTSService(AIService): 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 for aggregate in self._text_aggregator.aggregate(frame.text): + 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 a0ed9ca2d..5645d2bcf 100644 --- a/src/pipecat/utils/string.py +++ b/src/pipecat/utils/string.py @@ -280,27 +280,3 @@ 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/src/pipecat/utils/text/base_text_aggregator.py b/src/pipecat/utils/text/base_text_aggregator.py index 1b029833e..e9e7d82c3 100644 --- a/src/pipecat/utils/text/base_text_aggregator.py +++ b/src/pipecat/utils/text/base_text_aggregator.py @@ -14,7 +14,7 @@ aggregated text should be sent for speech synthesis. from abc import ABC, abstractmethod from dataclasses import dataclass from enum import Enum -from typing import Optional +from typing import AsyncIterator, Optional class AggregationType(str, Enum): @@ -80,35 +80,32 @@ class BaseTextAggregator(ABC): pass @abstractmethod - async def aggregate(self, text: str) -> Optional[Aggregation]: - """Aggregate the specified text with the currently accumulated text. + async def aggregate(self, text: str) -> AsyncIterator[Aggregation]: + """Aggregate the specified text and yield completed aggregations. - This method should be implemented to define how the new text contributes - to the aggregation process. It returns the aggregated text and a string - describing how it was aggregated if it's ready to be processed, - or None otherwise. + This method processes the input text character-by-character internally + and yields Aggregation objects as they complete. Subclasses should implement their specific logic for: - - How to combine new text with existing accumulated text + - How to process text character-by-character - When to consider the aggregated text ready for processing - What criteria determine text completion (e.g., sentence boundaries) - - When a completion occurs, the method should return an Aggregation object - containing the aggregated text and its type. The text should be stripped - of leading/trailing whitespace so that consumers can rely on a consistent - format. + - When a completion occurs, yield an Aggregation object containing the + aggregated text (stripped of leading/trailing whitespace) and its type Args: text: The text to be aggregated. - Returns: - An Aggregation object if ready for processing, or None if more - text is needed before the aggregated content is ready. If an Aggregation - object is returned, it should consist of the updated aggregated text, - stripped of leading/trailing whitespace, and a string indicating the - type of aggregation (e.g., 'sentence', 'word', 'token', 'my_custom_aggregation'). + Yields: + Aggregation objects as they complete. Each Aggregation consists of + the aggregated text (stripped of leading/trailing whitespace) and + a string indicating the type of aggregation (e.g., 'sentence', 'word', + 'token', 'my_custom_aggregation'). """ pass + # Make this a generator to satisfy type checker + yield # pragma: no cover @abstractmethod async def flush(self) -> Optional[Aggregation]: diff --git a/src/pipecat/utils/text/pattern_pair_aggregator.py b/src/pipecat/utils/text/pattern_pair_aggregator.py index d57ccd81b..518ec13ae 100644 --- a/src/pipecat/utils/text/pattern_pair_aggregator.py +++ b/src/pipecat/utils/text/pattern_pair_aggregator.py @@ -13,7 +13,7 @@ support for custom handlers and configurable actions for when a pattern is found import re from enum import Enum -from typing import Awaitable, Callable, List, Optional, Tuple +from typing import AsyncIterator, Awaitable, Callable, List, Optional, Tuple from loguru import logger @@ -256,20 +256,6 @@ class PatternPairAggregator(SimpleTextAggregator): matches = list(match_iter) # Convert to list for safe iteration for match in matches: - # Only process patterns that end at or after last_processed_position - # This ensures we only call handlers once when a pattern completes - if match.end() <= last_processed_position: - # This pattern was already processed in a previous call - if action != MatchAction.REMOVE: - # For KEEP/AGGREGATE patterns, we still need to track them - content = match.group(1) - full_match = match.group(0) - pattern_match = PatternMatch( - content=content.strip(), type=type, full_match=full_match - ) - all_matches.append(pattern_match) - continue - content = match.group(1) # Content between patterns full_match = match.group(0) # Full match including patterns @@ -278,17 +264,23 @@ class PatternPairAggregator(SimpleTextAggregator): content=content.strip(), type=type, full_match=full_match ) - # Call the appropriate handler if registered (only for newly complete patterns) - if type in self._handlers: + # Check if this pattern was already processed + already_processed = match.end() <= last_processed_position + + # Only call handler for newly completed patterns + if not already_processed and type in self._handlers: try: await self._handlers[type](pattern_match) except Exception as e: logger.error(f"Error in pattern handler for {type}: {e}") - # Remove the pattern from the text if configured + # Handle pattern based on action if action == MatchAction.REMOVE: - processed_text = processed_text.replace(full_match, "", 1) + # Remove patterns are only removed once (when newly completed) + if not already_processed: + processed_text = processed_text.replace(full_match, "", 1) else: + # KEEP/AGGREGATE patterns stay in all_matches all_matches.append(pattern_match) return all_matches, processed_text @@ -324,72 +316,74 @@ class PatternPairAggregator(SimpleTextAggregator): return None - async def aggregate(self, text: str) -> Optional[PatternMatch]: + async def aggregate(self, text: str) -> AsyncIterator[PatternMatch]: """Aggregate text and process pattern pairs. - This method adds the new text to the buffer, processes any complete pattern - pairs, and uses the parent's lookahead logic for sentence detection when - no patterns are active. + Processes the input text character-by-character, handles pattern pairs, + and uses the parent's lookahead logic for sentence detection when no + patterns are active. Args: - text: New text to add to the buffer. + text: Text to aggregate. - Returns: - Processed text up to a sentence boundary, or None if more - text is needed to form a complete sentence or pattern. + Yields: + PatternMatch objects as patterns complete or sentences are detected. """ - # Add new text to buffer - self._text += text + # Process text character by character + for char in text: + self._text += char - # Process any newly complete patterns in the buffer - # Only patterns that complete after _last_processed_position will trigger handlers - patterns, processed_text = await self._process_complete_patterns( - self._text, self._last_processed_position - ) + # Process any newly complete patterns in the buffer + # Only patterns that complete after _last_processed_position will trigger handlers + patterns, processed_text = await self._process_complete_patterns( + self._text, self._last_processed_position + ) - # Update the last processed position before modifying the text - # For REMOVE patterns, the text will be shorter, so we track the original position - self._last_processed_position = len(self._text) + # Update the last processed position to prevent re-processing patterns + # This tracks where in the buffer we've already called handlers, so we + # only trigger handlers once when a pattern completes + self._last_processed_position = len(self._text) - self._text = processed_text + self._text = processed_text - if len(patterns) > 0: - if len(patterns) > 1: - logger.warning( - f"Multiple patterns matched: {[p.type for p in patterns]}. Only the first pattern will be returned." + if len(patterns) > 0: + if len(patterns) > 1: + logger.warning( + f"Multiple patterns matched: {[p.type for p in patterns]}. Only the first pattern will be returned." + ) + # If the pattern found is set to be aggregated, return it + action = self._patterns[patterns[0].type].get("action", MatchAction.REMOVE) + if action == MatchAction.AGGREGATE: + self._text = "" + yield patterns[0] + continue + + # Check if we have incomplete patterns + pattern_start = self._match_start_of_pattern(self._text) + if pattern_start is not None: + # If the start pattern is at the beginning or should not be separately aggregated, continue + if ( + pattern_start[0] == 0 + or pattern_start[1].get("action", MatchAction.REMOVE) != MatchAction.AGGREGATE + ): + continue + # For AGGREGATE patterns: yield any text before the pattern starts + # This ensures text doesn't get stuck in the buffer waiting for sentence + # boundaries when a pattern begins (e.g., "Here is code ..." yields "Here is code") + result = self._text[: pattern_start[0]] + self._text = self._text[pattern_start[0] :] + yield PatternMatch( + content=result.strip(), type=AggregationType.SENTENCE, full_match=result ) - # If the pattern found is set to be aggregated, return it - action = self._patterns[patterns[0].type].get("action", MatchAction.REMOVE) - if action == MatchAction.AGGREGATE: - self._text = "" - return patterns[0] + continue - # Check if we have incomplete patterns - pattern_start = self._match_start_of_pattern(self._text) - if pattern_start is not None: - # If the start pattern is at the beginning or should not be separately aggregated, return None - if ( - pattern_start[0] == 0 - or pattern_start[1].get("action", MatchAction.REMOVE) != MatchAction.AGGREGATE - ): - return None - # Otherwise, strip the text up to the start pattern and return it - result = self._text[: pattern_start[0]] - self._text = self._text[pattern_start[0] :] - return PatternMatch( - content=result.strip(), type=AggregationType.SENTENCE, full_match=result - ) - - # Use parent's lookahead logic for sentence detection - aggregation = await super()._check_sentence_with_lookahead(text) - if aggregation: - # Convert to PatternMatch for consistency with return type - return PatternMatch( - content=aggregation.text, type=aggregation.type, full_match=aggregation.text - ) - - # No complete sentence found yet - return None + # Use parent's lookahead logic for sentence detection + aggregation = await super()._check_sentence_with_lookahead(char) + if aggregation: + # Convert to PatternMatch for consistency with return type + yield PatternMatch( + content=aggregation.text, type=aggregation.type, full_match=aggregation.text + ) async def handle_interruption(self): """Handle interruptions by clearing the buffer and pattern state. diff --git a/src/pipecat/utils/text/simple_text_aggregator.py b/src/pipecat/utils/text/simple_text_aggregator.py index c3bdb99df..1d123e8fb 100644 --- a/src/pipecat/utils/text/simple_text_aggregator.py +++ b/src/pipecat/utils/text/simple_text_aggregator.py @@ -11,7 +11,7 @@ until it finds an end-of-sentence marker, making it suitable for basic TTS text processing scenarios. """ -from typing import Optional +from typing import AsyncIterator, Optional from pipecat.utils.string import SENTENCE_ENDING_PUNCTUATION, match_endofsentence from pipecat.utils.text.base_text_aggregator import Aggregation, AggregationType, BaseTextAggregator @@ -42,27 +42,30 @@ class SimpleTextAggregator(BaseTextAggregator): """ return Aggregation(text=self._text.strip(), type=AggregationType.SENTENCE) - async def aggregate(self, text: str) -> Optional[Aggregation]: - """Aggregate text and return completed sentences. + async def aggregate(self, text: str) -> AsyncIterator[Aggregation]: + """Aggregate text and yield completed sentences. - Adds the new text to the buffer. When sentence-ending punctuation is - detected, it waits for non-whitespace lookahead before calling NLTK. - This prevents false positives like "$29." being detected as a sentence - when it's actually "$29.95", and avoids unnecessary NLTK calls. + Processes the input text character-by-character. When sentence-ending + punctuation is detected, it waits for non-whitespace lookahead before + calling NLTK. This prevents false positives like "$29." being detected + as a sentence when it's actually "$29.95". Args: - text: New text to add to the aggregation buffer. + text: Text to aggregate. - Returns: - A complete sentence if an end-of-sentence marker is confirmed, - or None if more text is needed. + Yields: + Complete sentences as Aggregation objects. """ - # Add new text to buffer - self._text += text + # Process text character by character + for char in text: + self._text += char - return await self._check_sentence_with_lookahead(text) + # Check for sentence with lookahead + result = await self._check_sentence_with_lookahead(char) + if result: + yield result - async def _check_sentence_with_lookahead(self, text: str) -> Optional[Aggregation]: + async def _check_sentence_with_lookahead(self, char: str) -> Optional[Aggregation]: """Check for sentence boundaries using lookahead logic. This method implements the core sentence detection logic with lookahead. @@ -77,7 +80,7 @@ class SimpleTextAggregator(BaseTextAggregator): while adding their own logic (e.g., tag handling, pattern matching). Args: - text: The most recently added text (used for lookahead check). + char: The most recently added character (used for lookahead check). Returns: Aggregation if sentence found, None otherwise. @@ -85,7 +88,7 @@ class SimpleTextAggregator(BaseTextAggregator): # If we need lookahead, check if we now have non-whitespace if self._needs_lookahead: # Check if the new character is non-whitespace - if text.strip(): + if char.strip(): # We have meaningful lookahead, call NLTK self._needs_lookahead = False eos_marker = match_endofsentence(self._text) @@ -101,7 +104,7 @@ class SimpleTextAggregator(BaseTextAggregator): return None # Check if we just added sentence-ending punctuation - if self._text[-1] in SENTENCE_ENDING_PUNCTUATION: + if self._text and self._text[-1] in SENTENCE_ENDING_PUNCTUATION: # Mark that we need lookahead (don't call NLTK yet) self._needs_lookahead = True diff --git a/src/pipecat/utils/text/skip_tags_aggregator.py b/src/pipecat/utils/text/skip_tags_aggregator.py index d693285eb..e82444ea6 100644 --- a/src/pipecat/utils/text/skip_tags_aggregator.py +++ b/src/pipecat/utils/text/skip_tags_aggregator.py @@ -11,7 +11,7 @@ between specified start/end tag pairs, ensuring that tagged content is processed as a unit regardless of internal punctuation. """ -from typing import Optional, Sequence +from typing import AsyncIterator, Optional, Sequence from pipecat.utils.string import StartEndTags, parse_start_end_tags from pipecat.utils.text.base_text_aggregator import Aggregation, AggregationType @@ -43,35 +43,37 @@ class SkipTagsAggregator(SimpleTextAggregator): self._current_tag: Optional[StartEndTags] = None self._current_tag_index: int = 0 - async def aggregate(self, text: str) -> Optional[Aggregation]: + async def aggregate(self, text: str) -> AsyncIterator[Aggregation]: """Aggregate text while respecting tag boundaries. - This method adds the new text to the buffer, updates tag state, - and uses the parent's lookahead logic for sentence detection when - not inside tags. + Processes the input text character-by-character, updates tag state, and + uses the parent's lookahead logic for sentence detection when not + inside tags. Args: - text: New text to add to the buffer. + text: Text to aggregate. - Returns: - An Aggregation object containing text up to a sentence boundary and - marked as SENTENCE type or None if more text is needed to complete a - sentence or close tags. + Yields: + Aggregation objects containing text up to a sentence boundary, + marked as SENTENCE type. """ - # Add new text to buffer - self._text += text + # Process text character by character + for char in text: + self._text += char - # 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 - ) + # 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 + ) - # If inside tags, don't check for sentences - if self._current_tag: - return None + # If inside tags, don't check for sentences + if self._current_tag: + continue - # Otherwise, use parent's lookahead logic for sentence detection - return await super()._check_sentence_with_lookahead(text) + # Otherwise, use parent's lookahead logic for sentence detection + result = await super()._check_sentence_with_lookahead(char) + if result: + yield result async def handle_interruption(self): """Handle interruptions by clearing the buffer and tag state. diff --git a/tests/test_ivr_navigation.py b/tests/test_ivr_navigation.py index 61464fb32..74bbb6266 100644 --- a/tests/test_ivr_navigation.py +++ b/tests/test_ivr_navigation.py @@ -10,6 +10,7 @@ from unittest.mock import AsyncMock from pipecat.audio.vad.vad_analyzer import VADParams from pipecat.extensions.ivr.ivr_navigator import IVRProcessor from pipecat.frames.frames import ( + LLMFullResponseEndFrame, LLMMessagesUpdateFrame, LLMTextFrame, OutputDTMFUrgentFrame, @@ -334,10 +335,12 @@ class TestIVRNavigation(unittest.IsolatedAsyncioTestCase): frames_to_send = [ LLMTextFrame(text="Hello, I'm trying to reach billing."), + LLMFullResponseEndFrame(), ] expected_down_frames = [ LLMTextFrame, # Should pass through unchanged + LLMFullResponseEndFrame, ] expected_up_frames = [ diff --git a/tests/test_pattern_pair_aggregator.py b/tests/test_pattern_pair_aggregator.py index b1d9d7150..4726e7b81 100644 --- a/tests/test_pattern_pair_aggregator.py +++ b/tests/test_pattern_pair_aggregator.py @@ -38,14 +38,8 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.aggregator.on_pattern_match("code_pattern", self.code_handler) async def test_pattern_match_and_removal(self): - # Feed text character by character - full_text = "Hello pattern content!" - results = [] - - for char in full_text: - result = await self.aggregator.aggregate(char) - if result: - results.append(result) + text = "Hello pattern content!" + results = [result async for result in self.aggregator.aggregate(text)] # Verify the handler was called with correct PatternMatch object self.test_handler.assert_called_once() @@ -59,10 +53,8 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(len(results), 0) # 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: - results.append(result) + async for result in self.aggregator.aggregate(" This is another sentence."): + results.append(result) # First result should be "Hello !" triggered by the space lookahead self.assertEqual(len(results), 1) @@ -77,14 +69,9 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(self.aggregator.text.text, "") async def test_pattern_match_and_aggregate(self): - # Feed text character by character, collecting all results - results = [] - full_text = "Here is code pattern content This is another sentence." + text = "Here is code pattern content This is another sentence." - for char in full_text: - result = await self.aggregator.aggregate(char) - if result: - results.append(result) + results = [result async for result in self.aggregator.aggregate(text)] # First result should be "Here is code" when pattern starts self.assertEqual(results[0].text, "Here is code") @@ -111,15 +98,10 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(self.aggregator.text.text, "") async def test_incomplete_pattern(self): - # Feed text character by character with incomplete pattern - result = None - for char in "Hello pattern content": - result = await self.aggregator.aggregate(char) - if result: - break - + text = "Hello pattern content" + results = [result async for result in self.aggregator.aggregate(text)] # No complete pattern yet, so nothing should be returned - self.assertIsNone(result) + self.assertEqual(len(results), 0) # The handler should not be called yet self.test_handler.assert_not_called() @@ -154,13 +136,8 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.aggregator.on_pattern_match("voice", voice_handler) self.aggregator.on_pattern_match("emphasis", emphasis_handler) - # Feed text character by character text = "Hello female I am very excited to meet you!" - result = None - for char in text: - result = await self.aggregator.aggregate(char) - if result: - break + results = [result async for result in self.aggregator.aggregate(text)] # Both handlers should be called with correct data voice_handler.assert_called_once() @@ -174,7 +151,7 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(emphasis_match.text, "very") # With lookahead, we need to flush to get the final sentence - self.assertIsNone(result) # Waiting for lookahead after "!" + self.assertEqual(len(results), 0) # Waiting for lookahead after "!" result = await self.aggregator.flush() # Voice pattern should be removed, emphasis pattern should remain @@ -184,14 +161,9 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(self.aggregator.text.text, "") async def test_handle_interruption(self): - # Feed text character by character with incomplete pattern - result = None - for char in "Hello pattern": - result = await self.aggregator.aggregate(char) - if result: - break - - self.assertIsNone(result) + text = "Hello pattern" + results = [result async for result in self.aggregator.aggregate(text)] + self.assertEqual(len(results), 0) # Simulate interruption await self.aggregator.handle_interruption() @@ -203,14 +175,8 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.test_handler.assert_not_called() async def test_pattern_across_sentences(self): - # Feed text character by character - pattern spans multiple sentences - full_text = "Hello This is sentence one. This is sentence two. Final sentence." - result = None - - for char in full_text: - result = await self.aggregator.aggregate(char) - if result: - break + text = "Hello This is sentence one. This is sentence two. Final sentence." + results = [result async for result in self.aggregator.aggregate(text)] # Handler should be called with entire content self.test_handler.assert_called_once() @@ -218,7 +184,7 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): 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 "." + self.assertEqual(len(results), 0) # Waiting for lookahead after "." result = await self.aggregator.flush() # Pattern should be removed, resulting in text with sentences merged diff --git a/tests/test_simple_text_aggregator.py b/tests/test_simple_text_aggregator.py index 663700fdf..7b87c551c 100644 --- a/tests/test_simple_text_aggregator.py +++ b/tests/test_simple_text_aggregator.py @@ -14,19 +14,22 @@ class TestSimpleTextAggregator(unittest.IsolatedAsyncioTestCase): self.aggregator = SimpleTextAggregator() async def test_reset_aggregations(self): - # Feed character-by-character - for char in "Hello ": - assert await self.aggregator.aggregate(char) == None + text = "Hello " + results = [agg async for agg in self.aggregator.aggregate(text)] + + # No complete sentences yet + assert len(results) == 0 assert self.aggregator.text.text == "Hello" await self.aggregator.reset() assert self.aggregator.text.text == "" async def test_simple_sentence(self): - # Feed character-by-character: "Hello Pipecat!" - for char in "Hello Pipecat": - assert await self.aggregator.aggregate(char) == None - # After "!", lookahead waits for confirmation - assert await self.aggregator.aggregate("!") == None + text = "Hello Pipecat!" + results = [agg async for agg in self.aggregator.aggregate(text)] + + # No complete sentences yet (waiting for lookahead after "!") + assert len(results) == 0 + # Flush to get the pending sentence aggregate = await self.aggregator.flush() assert aggregate.text == "Hello Pipecat!" @@ -34,81 +37,58 @@ class TestSimpleTextAggregator(unittest.IsolatedAsyncioTestCase): assert self.aggregator.text.text == "" async def test_multiple_sentences(self): - # Feed character-by-character: "Hello Pipecat! How are you?" - for char in "Hello Pipecat": - assert await self.aggregator.aggregate(char) == None - # Hit "!" - wait for lookahead - assert await self.aggregator.aggregate("!") == None - # Space is whitespace - keep waiting - assert await self.aggregator.aggregate(" ") == None - # "H" confirms sentence end - result = await self.aggregator.aggregate("H") - assert result.text == "Hello Pipecat!" - # Continue with second sentence - for char in "ow are you": - assert await self.aggregator.aggregate(char) == None - # Hit "?" - wait for lookahead - assert await self.aggregator.aggregate("?") == None + text = "Hello Pipecat! How are you?" + results = [agg async for agg in self.aggregator.aggregate(text)] + + # First sentence should be complete (lookahead from "H" confirmed it) + assert len(results) == 1 + assert results[0].text == "Hello Pipecat!" + # Flush to get the pending sentence result = await self.aggregator.flush() assert result.text == "How are you?" async def test_lookahead_decimal_number(self): """Test that $29.95 is not split at $29.""" - # Feed character by character: "Ask me for only $29.95/month." - for char in "Ask me for only $29": - assert await self.aggregator.aggregate(char) == None - # When we hit ".", it looks like end of sentence, but should wait for lookahead - assert await self.aggregator.aggregate(".") == None - # Next character "9" confirms it's not end of sentence (NLTK changes boundary) - for char in "95/month": - assert await self.aggregator.aggregate(char) == None - # Now we hit the real end of sentence - wait for lookahead - assert await self.aggregator.aggregate(".") == None + text = "Ask me for only $29.95/month." + results = [agg async for agg in self.aggregator.aggregate(text)] + + # No complete sentences yet (waiting for lookahead after final ".") + assert len(results) == 0 + # Can use flush() to get the pending sentence at end of stream result = await self.aggregator.flush() assert result.text == "Ask me for only $29.95/month." async def test_lookahead_abbreviation(self): """Test that Mr. Smith is not split at Mr.""" - # Feed character by character: "Hello Mr. Smith." - for char in "Hello Mr": - assert await self.aggregator.aggregate(char) == None - # When we hit ".", it looks like end of sentence, but should wait for lookahead - assert await self.aggregator.aggregate(".") == None - # Space alone is not enough - assert await self.aggregator.aggregate(" ") == None - # "S" confirms it's not end of sentence (NLTK changes boundary detection) - for char in "Smith": - assert await self.aggregator.aggregate(char) == None - # Now we hit the real end of sentence - wait for lookahead - assert await self.aggregator.aggregate(".") == None + text = "Hello Mr. Smith." + results = [agg async for agg in self.aggregator.aggregate(text)] + + # No complete sentences yet (waiting for lookahead after final ".") + assert len(results) == 0 + # Can use flush() to get the pending sentence at end of stream result = await self.aggregator.flush() assert result.text == "Hello Mr. Smith." async def test_lookahead_actual_sentence_end(self): """Test that a real sentence end is detected after lookahead.""" - # Feed character by character: "Hello world. Next sentence" - for char in "Hello world": - assert await self.aggregator.aggregate(char) == None - # Hit period - should wait for lookahead - assert await self.aggregator.aggregate(".") == None - # Space alone is not enough - need non-whitespace for meaningful lookahead - assert await self.aggregator.aggregate(" ") == None - # Capital letter confirms sentence end (NLTK detects boundary at same position) - result = await self.aggregator.aggregate("N") - assert result.text == "Hello world." - # Continue with next sentence - assert await self.aggregator.aggregate("e") == None + text = "Hello world. Next sentence" + results = [agg async for agg in self.aggregator.aggregate(text)] + + # First sentence should be complete (lookahead from "N" confirmed it) + assert len(results) == 1 + assert results[0].text == "Hello world." async def test_flush_pending_sentence(self): """Test that flush() returns pending sentence waiting for lookahead.""" - # Feed up to a period - for char in "Hello world": - assert await self.aggregator.aggregate(char) == None - assert await self.aggregator.aggregate(".") == None - # At this point, "Hello world." is pending lookahead + text = "Hello world." + results = [agg async for agg in self.aggregator.aggregate(text)] + + # No complete sentences yet (waiting for lookahead) + assert len(results) == 0 + # Call flush to get it result = await self.aggregator.flush() assert result is not None @@ -118,7 +98,12 @@ class TestSimpleTextAggregator(unittest.IsolatedAsyncioTestCase): async def test_flush_with_no_pending(self): """Test that flush() returns any remaining text in buffer.""" - assert await self.aggregator.aggregate("Hello") == None + text = "Hello" + results = [agg async for agg in self.aggregator.aggregate(text)] + + # No complete sentences + assert len(results) == 0 + result = await self.aggregator.flush() # flush() now returns any remaining text, not just pending lookahead assert result is not None @@ -128,13 +113,13 @@ class TestSimpleTextAggregator(unittest.IsolatedAsyncioTestCase): async def test_flush_after_lookahead_confirmed(self): """Test flush after lookahead has already confirmed sentence.""" - for char in "Hello.": - await self.aggregator.aggregate(char) - # Space alone is not enough - still waiting - assert await self.aggregator.aggregate(" ") == None - # Non-whitespace lookahead confirms it's a sentence - result = await self.aggregator.aggregate("W") - assert result.text == "Hello." + text = "Hello. W" + results = [agg async for agg in self.aggregator.aggregate(text)] + + # First sentence should be complete (lookahead from "W" confirmed it) + assert len(results) == 1 + assert results[0].text == "Hello." + # flush() returns any remaining text (the "W" in this case) result = await self.aggregator.flush() assert result.text == "W" diff --git a/tests/test_skip_tags_aggregator.py b/tests/test_skip_tags_aggregator.py index 13869ca0b..ffed966e1 100644 --- a/tests/test_skip_tags_aggregator.py +++ b/tests/test_skip_tags_aggregator.py @@ -17,15 +17,11 @@ class TestSkipTagsAggregator(unittest.IsolatedAsyncioTestCase): await self.aggregator.reset() # No tags involved, aggregate at end of sentence. - # Feed text character by character - result = None - for char in "Hello Pipecat!": - result = await self.aggregator.aggregate(char) - if result: - break + text = "Hello Pipecat!" + results = [agg async for agg in self.aggregator.aggregate(text)] # Should still be waiting for lookahead after "!" - self.assertIsNone(result) + self.assertEqual(len(results), 0) # Flush to get the pending sentence result = await self.aggregator.flush() @@ -37,15 +33,11 @@ class TestSkipTagsAggregator(unittest.IsolatedAsyncioTestCase): await self.aggregator.reset() # Tags involved, avoid aggregation during tags. - # Feed text character by character - result = None - for char in "My email is foo@pipecat.ai.": - result = await self.aggregator.aggregate(char) - if result: - break + text = "My email is foo@pipecat.ai." + results = [agg async for agg in self.aggregator.aggregate(text)] # Should still be waiting for lookahead after "." - self.assertIsNone(result) + self.assertEqual(len(results), 0) # Flush to get the pending sentence result = await self.aggregator.flush() @@ -56,22 +48,17 @@ class TestSkipTagsAggregator(unittest.IsolatedAsyncioTestCase): async def test_streaming_tags(self): await self.aggregator.reset() - # Tags involved, feed character by character - full_text = "My email is foo.bar@pipecat.ai." - result = None - - for char in full_text: - result = await self.aggregator.aggregate(char) - if result: - break + # Tags involved + text = "My email is foo.bar@pipecat.ai." + results = [agg async for agg in self.aggregator.aggregate(text)] # Should still be waiting for lookahead after "." - self.assertIsNone(result) - self.assertEqual(self.aggregator.text.text, full_text) + self.assertEqual(len(results), 0) + self.assertEqual(self.aggregator.text.text, text) self.assertEqual(self.aggregator.text.type, "sentence") # Flush to get the pending sentence result = await self.aggregator.flush() - self.assertEqual(result.text, full_text) + self.assertEqual(result.text, text) self.assertEqual(self.aggregator.text.text, "") self.assertEqual(self.aggregator.text.type, "sentence") diff --git a/tests/test_utils_string.py b/tests/test_utils_string.py index 5140964dd..607b8d0fa 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, split_text_by_characters +from pipecat.utils.string import match_endofsentence, parse_start_end_tags class TestUtilsString(unittest.IsolatedAsyncioTestCase): @@ -232,35 +232,3 @@ 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!" diff --git a/uv.lock b/uv.lock index 17bdef385..da1b785a1 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 2 +revision = 3 requires-python = ">=3.10" resolution-markers = [ "python_full_version >= '3.13'", @@ -4710,7 +4710,6 @@ requires-dist = [ { name = "numba", specifier = "==0.61.2" }, { name = "numpy", specifier = ">=1.26.4,<3" }, { name = "nvidia-riva-client", marker = "extra == 'nvidia'", specifier = "~=2.21.1" }, - { name = "nvidia-riva-client", marker = "extra == 'riva'", specifier = "~=2.21.1" }, { name = "onnxruntime", marker = "extra == 'local-smart-turn-v3'", specifier = ">=1.20.1,<2" }, { name = "onnxruntime", marker = "extra == 'silero'", specifier = ">=1.20.1,<2" }, { name = "openai", specifier = ">=1.74.0,<3" }, @@ -4721,6 +4720,7 @@ requires-dist = [ { name = "opentelemetry-sdk", marker = "extra == 'tracing'", specifier = ">=1.33.0" }, { name = "ormsgpack", marker = "extra == 'fish'", specifier = "~=1.7.0" }, { name = "pillow", specifier = ">=11.1.0,<12" }, + { name = "pipecat-ai", extras = ["nvidia"], marker = "extra == 'riva'" }, { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'assemblyai'" }, { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'asyncai'" }, { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'aws'" }, @@ -4771,7 +4771,7 @@ requires-dist = [ { name = "wait-for2", marker = "python_full_version < '3.12'", specifier = ">=0.4.1" }, { name = "websockets", marker = "extra == 'websockets-base'", specifier = ">=13.1,<16.0" }, ] -provides-extras = ["aic", "anthropic", "assemblyai", "asyncai", "aws", "aws-nova-sonic", "azure", "cartesia", "cerebras", "daily", "deepgram", "deepseek", "elevenlabs", "fal", "fireworks", "fish", "gladia", "google", "grok", "groq", "gstreamer", "heygen", "hume", "inworld", "koala", "krisp", "langchain", "livekit", "lmnt", "local", "local-smart-turn", "local-smart-turn-v3", "mcp", "mem0", "mistral", "mlx-whisper", "moondream", "neuphonic", "nim", "noisereduce", "openai", "openpipe", "openrouter", "perplexity", "playht", "qwen", "remote-smart-turn", "rime", "riva", "nvidia", "runner", "sagemaker", "sambanova", "sarvam", "sentry", "silero", "simli", "soniox", "soundfile", "speechmatics", "strands", "tavus", "together", "tracing", "ultravox", "webrtc", "websocket", "websockets-base", "whisper"] +provides-extras = ["aic", "anthropic", "assemblyai", "asyncai", "aws", "aws-nova-sonic", "azure", "cartesia", "cerebras", "daily", "deepgram", "deepseek", "elevenlabs", "fal", "fireworks", "fish", "gladia", "google", "grok", "groq", "gstreamer", "heygen", "hume", "inworld", "koala", "krisp", "langchain", "livekit", "lmnt", "local", "local-smart-turn", "local-smart-turn-v3", "mcp", "mem0", "mistral", "mlx-whisper", "moondream", "neuphonic", "noisereduce", "nvidia", "openai", "openpipe", "openrouter", "perplexity", "playht", "qwen", "remote-smart-turn", "rime", "riva", "runner", "sagemaker", "sambanova", "sarvam", "sentry", "silero", "simli", "soniox", "soundfile", "speechmatics", "strands", "tavus", "together", "tracing", "ultravox", "webrtc", "websocket", "websockets-base", "whisper"] [package.metadata.requires-dev] dev = [