diff --git a/CHANGELOG.md b/CHANGELOG.md index 2891a53e5..52606239d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,22 @@ 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: + + - **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 + 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 interaction (e.g., voice agents, speech-to-text) and operates at a native @@ -41,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`. 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 0d9636fd8..f0de46617 100644 --- a/src/pipecat/processors/aggregators/llm_text_processor.py +++ b/src/pipecat/processors/aggregators/llm_text_processor.py @@ -83,8 +83,7 @@ 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: + async for aggregation in self._text_aggregator.aggregate(in_frame.text): out_frame = AggregatedTextFrame( text=aggregation.text, aggregated_by=aggregation.type, @@ -93,14 +92,12 @@ class LLMTextProcessor(FrameProcessor): 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 - aggregation = self._text_aggregator.text - await self._text_aggregator.reset() - text = aggregation.text.strip() - if text: + # Flush any remaining text + remaining = await self._text_aggregator.flush() + if remaining: out_frame = AggregatedTextFrame( - text=text, - aggregated_by=aggregation.type, + text=remaining.text, + aggregated_by=remaining.type, ) out_frame.skip_tts = skip_tts await self.push_frame(out_frame) diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index 977d6957b..ca15cb2c0 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -425,16 +425,13 @@ class TTSService(AIService): # pause to avoid audio overlapping. await self._maybe_pause_frame_processing() - pending_aggregation = self._text_aggregator.text + # Flush any remaining text (including text waiting for lookahead) + remaining = await self._text_aggregator.flush() + if remaining: + await self._push_tts_frames(AggregatedTextFrame(remaining.text, remaining.type)) # Reset aggregator state - await self._text_aggregator.reset() self._processing_text = False - - if pending_aggregation.text: - await self._push_tts_frames( - AggregatedTextFrame(pending_aggregation.text, pending_aggregation.type) - ) if isinstance(frame, LLMFullResponseEndFrame): if self._push_text_frames: await self.push_frame(frame, direction) @@ -539,17 +536,20 @@ class TTSService(AIService): text = frame.text includes_inter_frame_spaces = frame.includes_inter_frame_spaces aggregated_by = "token" + + 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: - aggregate = await self._text_aggregator.aggregate(frame.text) - if aggregate: + async for aggregate in self._text_aggregator.aggregate(frame.text): 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 - ) + 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/text/base_text_aggregator.py b/src/pipecat/utils/text/base_text_aggregator.py index 5c39ce769..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,33 +80,43 @@ 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. + 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]: + """Flush any pending aggregation. + + This method is called at the end of a stream (e.g., when receiving + LLMFullResponseEndFrame) to return any text that was buffered. + 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'). + An Aggregation object if there is pending text, or None if there + is no pending text. """ pass diff --git a/src/pipecat/utils/text/pattern_pair_aggregator.py b/src/pipecat/utils/text/pattern_pair_aggregator.py index c140e3243..518ec13ae 100644 --- a/src/pipecat/utils/text/pattern_pair_aggregator.py +++ b/src/pipecat/utils/text/pattern_pair_aggregator.py @@ -13,12 +13,12 @@ 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 -from pipecat.utils.string import match_endofsentence -from pipecat.utils.text.base_text_aggregator import Aggregation, AggregationType, BaseTextAggregator +from pipecat.utils.text.base_text_aggregator import Aggregation, AggregationType +from pipecat.utils.text.simple_text_aggregator import SimpleTextAggregator class MatchAction(Enum): @@ -72,7 +72,7 @@ class PatternMatch(Aggregation): return f"PatternMatch(type={self.type}, text={self.text}, full_match={self.full_match})" -class PatternPairAggregator(BaseTextAggregator): +class PatternPairAggregator(SimpleTextAggregator): """Aggregator that identifies and processes content between pattern pairs. This aggregator buffers text until it can identify complete pattern pairs @@ -97,9 +97,10 @@ class PatternPairAggregator(BaseTextAggregator): Creates an empty aggregator with no patterns or handlers registered. Text buffering and pattern detection will begin when text is aggregated. """ - self._text = "" + super().__init__() self._patterns = {} self._handlers = {} + self._last_processed_position = 0 # Track where we last checked for complete patterns @property def text(self) -> Aggregation: @@ -218,14 +219,18 @@ class PatternPairAggregator(BaseTextAggregator): self._handlers[type] = handler return self - async def _process_complete_patterns(self, text: str) -> Tuple[List[PatternMatch], str]: - """Process all complete pattern pairs in the text. + async def _process_complete_patterns( + self, text: str, last_processed_position: int = 0 + ) -> Tuple[List[PatternMatch], str]: + """Process newly complete pattern pairs in the text. - Searches for all complete pattern pairs in the text, calls the - appropriate handlers, and optionally removes the matches. + Searches for pattern pairs that have been completed since last_processed_position, + calls the appropriate handlers, and optionally removes the matches. Args: text: The text to process for pattern matches. + last_processed_position: The position in text that was already processed. + Only patterns that end at or after this position will be processed. Returns: Tuple of (all_matches, processed_text) where: @@ -259,17 +264,23 @@ class PatternPairAggregator(BaseTextAggregator): content=content.strip(), type=type, full_match=full_match ) - # Call the appropriate handler if registered - 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 @@ -305,76 +316,84 @@ class PatternPairAggregator(BaseTextAggregator): 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 returns processed text up to sentence boundaries if possible. - If there are incomplete patterns (start without matching end), it will - continue buffering text. + 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 complete patterns in the buffer - patterns, processed_text = await self._process_complete_patterns(self._text) + # 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 + ) - self._text = processed_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) - 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." + 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 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 - ) - - # Find sentence boundary if no incomplete patterns - 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 PatternMatch( - content=result.strip(), type=AggregationType.SENTENCE, full_match=result - ) - - # 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. + """Handle interruptions by clearing the buffer and pattern state. Called when an interruption occurs in the processing pipeline, to reset the state and discard any partially aggregated text. """ - self._text = "" + await super().handle_interruption() + self._last_processed_position = 0 + # Pattern and handler state persists across interruptions async def reset(self): """Clear the internally aggregated text. @@ -382,4 +401,6 @@ class PatternPairAggregator(BaseTextAggregator): Resets the aggregator to its initial state, discarding any buffered text and clearing pattern tracking state. """ - self._text = "" + await super().reset() + self._last_processed_position = 0 + # Pattern and handler state persists across resets diff --git a/src/pipecat/utils/text/simple_text_aggregator.py b/src/pipecat/utils/text/simple_text_aggregator.py index 56eab7032..1d123e8fb 100644 --- a/src/pipecat/utils/text/simple_text_aggregator.py +++ b/src/pipecat/utils/text/simple_text_aggregator.py @@ -11,9 +11,9 @@ 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 match_endofsentence +from pipecat.utils.string import SENTENCE_ENDING_PUNCTUATION, match_endofsentence from pipecat.utils.text.base_text_aggregator import Aggregation, AggregationType, BaseTextAggregator @@ -31,6 +31,7 @@ class SimpleTextAggregator(BaseTextAggregator): Creates an empty text buffer ready to begin accumulating text tokens. """ self._text = "" + self._needs_lookahead: bool = False @property def text(self) -> Aggregation: @@ -41,30 +42,87 @@ 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 and checks for end-of-sentence markers. - When a sentence boundary is found, returns the completed sentence and - removes it from the buffer. + 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. + + Yields: + Complete sentences as Aggregation objects. + """ + # Process text character by character + for char in text: + self._text += char + + # Check for sentence with lookahead + result = await self._check_sentence_with_lookahead(char) + if result: + yield result + + 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. + When sentence-ending punctuation is detected, it waits for the next + non-whitespace character before calling NLTK. This disambiguates cases + like "$29." (not a sentence) vs "$29. Next" (sentence ends at period). + Whitespace alone is not meaningful lookahead since it appears in both + cases. Instead, the first non-whitespace character after the punctuation + is used to confirm the sentence boundary. + + Subclasses can call this via super() to reuse the lookahead behavior + while adding their own logic (e.g., tag handling, pattern matching). + + Args: + char: The most recently added character (used for lookahead check). Returns: - A complete sentence if an end-of-sentence marker is found, - or None if more text is needed to complete a sentence. + Aggregation if sentence found, None otherwise. """ - result: Optional[str] = None + # If we need lookahead, check if we now have non-whitespace + if self._needs_lookahead: + # Check if the new character is non-whitespace + if char.strip(): + # We have meaningful lookahead, call NLTK + self._needs_lookahead = False + eos_marker = match_endofsentence(self._text) - self._text += text + if eos_marker: + # NLTK confirmed a sentence - return it + result = self._text[:eos_marker] + self._text = self._text[eos_marker:] + return Aggregation(text=result, type=AggregationType.SENTENCE) + # No sentence found - keep accumulating + return None + # Still whitespace, keep waiting + return None - eos_end_marker = match_endofsentence(self._text) - if eos_end_marker: - result = self._text[:eos_end_marker] - self._text = self._text[eos_end_marker:] + # Check if we just added 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 - if result: + return None + + async def flush(self) -> Optional[Aggregation]: + """Flush any remaining text in the buffer. + + Returns any text remaining in the buffer. This is called at the end + of a stream to ensure all text is processed. + + Returns: + Any remaining text as a sentence, or None if buffer is empty. + """ + if self._text: + # Return whatever we have in the buffer + result = self._text + await self.reset() return Aggregation(text=result.strip(), type=AggregationType.SENTENCE) return None @@ -75,6 +133,7 @@ class SimpleTextAggregator(BaseTextAggregator): discarding any partially accumulated text. """ self._text = "" + self._needs_lookahead = False async def reset(self): """Clear the internally aggregated text. @@ -83,3 +142,4 @@ class SimpleTextAggregator(BaseTextAggregator): any accumulated text content. """ self._text = "" + self._needs_lookahead = False diff --git a/src/pipecat/utils/text/skip_tags_aggregator.py b/src/pipecat/utils/text/skip_tags_aggregator.py index 3c8b95aab..e82444ea6 100644 --- a/src/pipecat/utils/text/skip_tags_aggregator.py +++ b/src/pipecat/utils/text/skip_tags_aggregator.py @@ -11,13 +11,14 @@ 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, match_endofsentence, parse_start_end_tags -from pipecat.utils.text.base_text_aggregator import Aggregation, AggregationType, BaseTextAggregator +from pipecat.utils.string import StartEndTags, parse_start_end_tags +from pipecat.utils.text.base_text_aggregator import Aggregation, AggregationType +from pipecat.utils.text.simple_text_aggregator import SimpleTextAggregator -class SkipTagsAggregator(BaseTextAggregator): +class SkipTagsAggregator(SimpleTextAggregator): """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 @@ -37,67 +38,59 @@ class SkipTagsAggregator(BaseTextAggregator): tags: Sequence of StartEndTags objects defining the tag pairs that should prevent sentence boundary detection. """ - self._text = "" + super().__init__() self._tags = tags self._current_tag: Optional[StartEndTags] = None self._current_tag_index: int = 0 - @property - def text(self) -> Aggregation: - """Get the currently buffered text. - - Returns: - The current text buffer content that hasn't been processed yet. - """ - return Aggregation(text=self._text.strip(), type=AggregationType.SENTENCE) - - 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, 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. + 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 - (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 + ) - # 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 Aggregation(text=result.strip(), type=AggregationType.SENTENCE) + # If inside tags, don't check for sentences + if self._current_tag: + continue - # No complete sentence found yet - return None + # 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. + """Handle interruptions by clearing the buffer and tag state. Called when an interruption occurs in the processing pipeline, to reset the state and discard any partially aggregated text. """ - self._text = "" + await super().handle_interruption() + self._current_tag = None + self._current_tag_index = 0 async def reset(self): - """Clear the internally aggregated text. + """Clear the internally aggregated text and tag state. Resets the aggregator to its initial state, discarding any buffered text. """ - self._text = "" + await super().reset() + self._current_tag = None + self._current_tag_index = 0 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 20b44a03c..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): - # First part doesn't complete the pattern - result = await self.aggregator.aggregate("Hello pattern") - self.assertIsNone(result) - self.assertEqual(self.aggregator.text.text, "Hello pattern") - self.assertEqual(self.aggregator.text.type, "test_pattern") - - # Second part completes the pattern and includes an exclamation point - result = await self.aggregator.aggregate(" content!") + 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() @@ -55,28 +49,37 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(call_args.full_match, "pattern content") self.assertEqual(call_args.text, "pattern content") - # The exclamation point should be treated as a sentence boundary, - # so the result should include just text up to and including "!" - self.assertEqual(result.text, "Hello !") - self.assertEqual(result.type, "sentence") + # No results yet (waiting for lookahead after "!") + self.assertEqual(len(results), 0) - # Next sentence should be processed separately. Spaces around the sentence - # should be stripped in the returned Aggregation. - result = await self.aggregator.aggregate(" This is another sentence.") + # Next sentence should provide the lookahead and trigger the previous sentence + 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) + self.assertEqual(results[0].text, "Hello !") + self.assertEqual(results[0].type, "sentence") + + # Now flush to get the remaining sentence + result = await self.aggregator.flush() self.assertEqual(result.text, "This is another sentence.") # Buffer should be empty after returning a complete sentence self.assertEqual(self.aggregator.text.text, "") async def test_pattern_match_and_aggregate(self): - # First part doesn't complete the pattern - result = await self.aggregator.aggregate("Here is code pattern") - self.assertEqual(result.text, "Here is code") - self.assertEqual(self.aggregator.text.text, "pattern") - self.assertEqual(self.aggregator.text.type, "code_pattern") + text = "Here is code pattern content This is another sentence." - # Second part completes the pattern and includes an exclamation point - result = await self.aggregator.aggregate(" content") + 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") + self.assertEqual(results[0].type, "sentence") + + # Second result should be the code pattern content + self.assertEqual(results[1].text, "pattern content") + self.assertEqual(results[1].type, "code_pattern") # Verify the handler was called with correct PatternMatch object self.code_handler.assert_called_once() @@ -85,11 +88,9 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(call_args.type, "code_pattern") self.assertEqual(call_args.full_match, "pattern content") self.assertEqual(call_args.text, "pattern content") - self.assertEqual(result.text, "pattern content") - self.assertEqual(result.type, "code_pattern") - # Next sentence should be processed separately - result = await self.aggregator.aggregate(" This is another sentence.") + # Last sentence needs flush (waiting for lookahead after ".") + result = await self.aggregator.flush() self.assertEqual(result.text, "This is another sentence.") self.assertEqual(result.type, "sentence") @@ -97,11 +98,10 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(self.aggregator.text.text, "") async def test_incomplete_pattern(self): - # Add text with incomplete pattern - result = await self.aggregator.aggregate("Hello pattern content") - + 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() @@ -136,9 +136,8 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.aggregator.on_pattern_match("voice", voice_handler) self.aggregator.on_pattern_match("emphasis", emphasis_handler) - # Test with multiple patterns in one text block text = "Hello female I am very excited to meet you!" - result = await self.aggregator.aggregate(text) + results = [result async for result in self.aggregator.aggregate(text)] # Both handlers should be called with correct data voice_handler.assert_called_once() @@ -151,6 +150,10 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(emphasis_match.type, "emphasis") self.assertEqual(emphasis_match.text, "very") + # With lookahead, we need to flush to get the final sentence + self.assertEqual(len(results), 0) # Waiting for lookahead after "!" + + result = await self.aggregator.flush() # Voice pattern should be removed, emphasis pattern should remain self.assertEqual(result.text, "Hello I am very excited to meet you!") @@ -158,9 +161,9 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(self.aggregator.text.text, "") async def test_handle_interruption(self): - # Start with incomplete pattern - result = await self.aggregator.aggregate("Hello pattern") - 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() @@ -172,20 +175,18 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase): self.test_handler.assert_not_called() async def test_pattern_across_sentences(self): - # Test pattern that spans multiple sentences - result = await self.aggregator.aggregate("Hello This is sentence one.") - - # First sentence contains start of pattern but no end, so no complete pattern yet - self.assertIsNone(result) - - # Add second part with pattern end - result = await self.aggregator.aggregate(" This is sentence two. Final sentence.") + 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() call_args = self.test_handler.call_args[0][0] 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.assertEqual(len(results), 0) # Waiting for lookahead after "." + + result = await self.aggregator.flush() # Pattern should be removed, resulting in text with sentences merged self.assertEqual(result.text, "Hello Final sentence.") diff --git a/tests/test_simple_text_aggregator.py b/tests/test_simple_text_aggregator.py index f8e2ee553..7b87c551c 100644 --- a/tests/test_simple_text_aggregator.py +++ b/tests/test_simple_text_aggregator.py @@ -14,22 +14,112 @@ class TestSimpleTextAggregator(unittest.IsolatedAsyncioTestCase): self.aggregator = SimpleTextAggregator() async def test_reset_aggregations(self): - assert await self.aggregator.aggregate("Hello ") == 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): - assert await self.aggregator.aggregate("Hello ") == None - aggregate = await self.aggregator.aggregate("Pipecat!") + 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!" assert aggregate.type == "sentence" assert self.aggregator.text.text == "" async def test_multiple_sentences(self): - aggregate = await self.aggregator.aggregate("Hello Pipecat! How are ") - assert aggregate.text == "Hello Pipecat!" - # Aggregators should strip leading/trailing spaces when returning text - assert self.aggregator.text.text == "How are" - aggregate = await self.aggregator.aggregate("you?") - assert aggregate.text == "How are you?" + 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.""" + 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.""" + 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.""" + 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.""" + 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 + assert result.text == "Hello world." + # Flush again should return None + assert await self.aggregator.flush() == None + + async def test_flush_with_no_pending(self): + """Test that flush() returns any remaining text in buffer.""" + 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 + assert result.text == "Hello" + # Buffer should be empty after flush + assert self.aggregator.text.text == "" + + async def test_flush_after_lookahead_confirmed(self): + """Test flush after lookahead has already confirmed sentence.""" + 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 702b991ce..ffed966e1 100644 --- a/tests/test_skip_tags_aggregator.py +++ b/tests/test_skip_tags_aggregator.py @@ -17,7 +17,14 @@ class TestSkipTagsAggregator(unittest.IsolatedAsyncioTestCase): await self.aggregator.reset() # No tags involved, aggregate at end of sentence. - result = await self.aggregator.aggregate("Hello Pipecat!") + text = "Hello Pipecat!" + results = [agg async for agg in self.aggregator.aggregate(text)] + + # Should still be waiting for lookahead after "!" + self.assertEqual(len(results), 0) + + # Flush to get the pending sentence + result = await self.aggregator.flush() self.assertEqual(result.text, "Hello Pipecat!") self.assertEqual(result.type, "sentence") self.assertEqual(self.aggregator.text.text, "") @@ -26,7 +33,14 @@ class TestSkipTagsAggregator(unittest.IsolatedAsyncioTestCase): await self.aggregator.reset() # Tags involved, avoid aggregation during tags. - result = await self.aggregator.aggregate("My email is foo@pipecat.ai.") + 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.assertEqual(len(results), 0) + + # Flush to get the pending sentence + result = await self.aggregator.flush() self.assertEqual(result.text, "My email is foo@pipecat.ai.") self.assertEqual(result.type, "sentence") self.assertEqual(self.aggregator.text.text, "") @@ -34,25 +48,17 @@ class TestSkipTagsAggregator(unittest.IsolatedAsyncioTestCase): async def test_streaming_tags(self): await self.aggregator.reset() - # Tags involved, stream small chunk of texts. - result = await self.aggregator.aggregate("My email is foo.") - self.assertIsNone(result) - self.assertEqual(self.aggregator.text.text, "My email is foo.") - - result = await self.aggregator.aggregate("bar@pipecat.") - self.assertIsNone(result) - self.assertEqual(self.aggregator.text.text, "My email is foo.bar@pipecat.") - - result = await self.aggregator.aggregate("aifoo.bar@pipecat.ai.") - self.assertEqual(result.text, "My email is foo.bar@pipecat.ai.") + # Flush to get the pending sentence + result = await self.aggregator.flush() + self.assertEqual(result.text, text) self.assertEqual(self.aggregator.text.text, "") self.assertEqual(self.aggregator.text.type, "sentence") 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 = [