wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #3132 from pipecat-ai/mb/normalize-llm-text-frame-output

Browse Source 
Add split_text_by_spaces string util, normalize aggregator input
This commit is contained in:
Mark Backman
2025-12-03 22:05:14 -05:00
committed by GitHub
parent 3ca94363ec d79dd94019
commit 57c24670ea
13 changed files with 456 additions and 244 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
CHANGELOG.md
View File

@@ -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`.

13
src/pipecat/extensions/ivr/ivr_navigator.py
View File

@@ -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)

15
src/pipecat/processors/aggregators/llm_text_processor.py
View File

@@ -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)

30
src/pipecat/services/tts_service.py
View File

@@ -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

44
src/pipecat/utils/text/base_text_aggregator.py
View File

@@ -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

155
src/pipecat/utils/text/pattern_pair_aggregator.py
View File

@@ -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 <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

94
src/pipecat/utils/text/simple_text_aggregator.py
View File

@@ -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

79
src/pipecat/utils/text/skip_tags_aggregator.py
View File

@@ -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

3
tests/test_ivr_navigation.py
View File

@@ -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 = [

87
tests/test_pattern_pair_aggregator.py
View File

@@ -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 <test>pattern")
self.assertIsNone(result)
self.assertEqual(self.aggregator.text.text, "Hello <test>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</test>!")
text = "Hello <test>pattern content</test>!"
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, "<test>pattern content</test>")
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 <code>pattern")
self.assertEqual(result.text, "Here is code")
self.assertEqual(self.aggregator.text.text, "<code>pattern")
self.assertEqual(self.aggregator.text.type, "code_pattern")
text = "Here is code <code>pattern content</code> This is another sentence."
# Second part completes the pattern and includes an exclamation point
result = await self.aggregator.aggregate(" content</code>")
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, "<code>pattern content</code>")
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 <test>pattern content")
text = "Hello <test>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 <voice>female</voice> I am <em>very</em> 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 <em>very</em> 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 <test>pattern")
self.assertIsNone(result)
text = "Hello <test>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 <test>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.</test> Final sentence.")
text = "Hello <test>This is sentence one. This is sentence two.</test> 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.")

108
tests/test_simple_text_aggregator.py
View File

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

44
tests/test_skip_tags_aggregator.py
View File

@@ -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 <spell>foo@pipecat.ai</spell>.")
text = "My email is <spell>foo@pipecat.ai</spell>."
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 <spell>foo@pipecat.ai</spell>.")
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 <sp")
self.assertIsNone(result)
self.assertEqual(self.aggregator.text.text, "My email is <sp")
# Tags involved
text = "My email is <spell>foo.bar@pipecat.ai</spell>."
results = [agg async for agg in self.aggregator.aggregate(text)]
result = await self.aggregator.aggregate("ell>foo.")
self.assertIsNone(result)
self.assertEqual(self.aggregator.text.text, "My email is <spell>foo.")
result = await self.aggregator.aggregate("bar@pipecat.")
self.assertIsNone(result)
self.assertEqual(self.aggregator.text.text, "My email is <spell>foo.bar@pipecat.")
result = await self.aggregator.aggregate("ai</spe")
self.assertIsNone(result)
self.assertEqual(self.aggregator.text.text, "My email is <spell>foo.bar@pipecat.ai</spe")
# Should still be waiting for lookahead after "."
self.assertEqual(len(results), 0)
self.assertEqual(self.aggregator.text.text, text)
self.assertEqual(self.aggregator.text.type, "sentence")
result = await self.aggregator.aggregate("ll>.")
self.assertEqual(result.text, "My email is <spell>foo.bar@pipecat.ai</spell>.")
# 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")

6
uv.lock generated
View File

@@ -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 = [