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

Make aggregate return an AsyncIterator, other clean up

Browse Source
This commit is contained in:
Mark Backman
2025-12-03 14:46:30 -05:00
parent fa8e7458e1
commit d79dd94019
15 changed files with 259 additions and 378 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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)

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

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

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

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

24
src/pipecat/utils/string.py
View File

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

33
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,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]:

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

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

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

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

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

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