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

Merge pull request #1387 from pipecat-ai/mb/pattern-aggregator

Browse Source 
Add PatternPairAggregator
This commit is contained in:
Mark Backman
2025-03-18 08:46:42 -04:00
committed by GitHub
parent 4918eab4e8 b28276446d
commit 4677c34663
4 changed files with 650 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
CHANGELOG.md
View File

@@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Added
- Added new `PatternPairAggregator` that extends `BaseTextAggregator` to
identify content between matching pattern pairs in streamed text. This allows
for detection and processing of structured content like XML-style tags that
may span across multiple text chunks or sentence boundaries.
- Added new `BaseTextAggregator`. Text aggregators are used by the TTS service
to aggregate LLM tokens and decide when the aggregated text should be pushed
to the TTS service. It also allows for the text to be manipulated while it's
@@ -124,6 +129,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Update the `34-audio-recording.py` example to include an STT processor.
- Added foundational example `35-voice-switching.py` showing how to use the new
`PatternPairAggregator`. This example shows how to encode information for the
LLM to instruct TTS voice changes, but this can be used to encode any
information into the LLM response, which you want to parse and use in other
parts of your application.
- Added a Pipecat Cloud deployment example to the `examples` directory.
- Removed foundational examples 28b and 28c as the TranscriptProcessor no

230
examples/foundational/35-pattern-pair-voice-switching.py Normal file
View File

@@ -0,0 +1,230 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Pattern Pair Voice Switching Example with Pipecat.
This example demonstrates how to use the PatternPairAggregator to dynamically switch
between different voices in a storytelling application. It showcases how pattern matching
can be used to control TTS behavior in streaming text from an LLM.
The example:
1. Sets up a storytelling bot with three distinct voices (narrator, male, female)
2. Uses pattern pairs (<voice>name</voice>) to trigger voice switching
3. Processes the patterns in real-time as text streams from the LLM
4. Removes the pattern tags before sending text to TTS
The PatternPairAggregator:
- Buffers text until complete patterns are detected
- Identifies content between start/end pattern pairs
- Triggers callbacks when patterns are matched
- Processes patterns that may span across multiple text chunks
- Returns processed text at sentence boundaries
Example usage (run from pipecat root directory):
$ pip install "pipecat-ai[daily,openai,cartesia,silero]"
$ pip install -r dev-requirements.txt
$ python examples/foundational/35-pattern-pair-voice-switching.py
Requirements:
- OpenAI API key (for GPT-4o)
- Cartesia API key (for text-to-speech)
- Daily API key (for video/audio transport)
Environment variables (.env file):
OPENAI_API_KEY=your_openai_key
CARTESIA_API_KEY=your_cartesia_key
DAILY_API_KEY=your_daily_key
Note:
This example shows one application of PatternPairAggregator (voice switching),
but the same approach can be used for various pattern-based text processing needs,
such as formatting instructions, command recognition, or structured data extraction.
"""
import asyncio
import os
import sys
import aiohttp
from dotenv import load_dotenv
from loguru import logger
from runner import configure
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.cartesia import CartesiaTTSService
from pipecat.services.openai import OpenAILLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport
from pipecat.utils.text.pattern_pair_aggregator import PatternMatch, PatternPairAggregator
load_dotenv(override=True)
logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
# Define voice IDs
VOICE_IDS = {
"narrator": "c45bc5ec-dc68-4feb-8829-6e6b2748095d", # Narrator voice
"female": "71a7ad14-091c-4e8e-a314-022ece01c121", # Female character voice
"male": "7cf0e2b1-8daf-4fe4-89ad-f6039398f359", # Male character voice
}
async def main():
async with aiohttp.ClientSession() as session:
(room_url, token) = await configure(session)
transport = DailyTransport(
room_url,
token,
"Multi-voice storyteller",
DailyParams(
audio_out_enabled=True,
transcription_enabled=True,
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
)
# Create pattern pair aggregator for voice switching
pattern_aggregator = PatternPairAggregator()
# Add pattern for voice switching
pattern_aggregator.add_pattern_pair(
pattern_id="voice_tag",
start_pattern="<voice>",
end_pattern="</voice>",
remove_match=True,
)
# Register handler for voice switching
def on_voice_tag(match: PatternMatch):
voice_name = match.content.strip().lower()
if voice_name in VOICE_IDS:
voice_id = VOICE_IDS[voice_name]
tts.set_voice(voice_id)
logger.info(f"Switched to {voice_name} voice")
else:
logger.warning(f"Unknown voice: {voice_name}")
pattern_aggregator.on_pattern_match("voice_tag", on_voice_tag)
# Initialize TTS with narrator voice as default
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id=VOICE_IDS["narrator"],
text_aggregator=pattern_aggregator,
)
# Initialize LLM
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
# System prompt for storytelling with voice switching
system_prompt = """You are an engaging storyteller that uses different voices to bring stories to life.
You have three voices to use, but each has a specific purpose:
<voice>narrator</voice>
This is the default narrator voice. Use this for all narration, descriptions, and non-dialogue text.
<voice>female</voice>
Use this ONLY for direct speech by female characters (just the quoted text).
<voice>male</voice>
Use this ONLY for direct speech by male characters (just the quoted text).
IMPORTANT: Switch back to narrator voice immediately after character dialogue.
Here's an EXAMPLE of correct voice usage:
<voice>narrator</voice>
Sarah spotted her old friend across the café. She couldn't believe her eyes.
<voice>female</voice>
"Jacob! It's been so long!"
<voice>narrator</voice>
Sarah exclaimed, jumping up from her seat with a radiant smile.
<voice>male</voice>
"Sarah, is it really you? I can't believe it!"
<voice>narrator</voice>
Jacob replied, grinning widely as he walked over to her. The two friends embraced warmly, as if trying to make up for all the years spent apart.
<voice>female</voice>
"What are you doing in town? Last I heard you were in Seattle."
<voice>narrator</voice>
She asked, gesturing for him to join her at the table.
FOLLOW THESE RULES:
1. Always begin with the narrator voice
2. Only use character voices for the EXACT words they speak (in quotes)
3. SWITCH BACK to narrator voice for speech tags and all other text
4. Begin by asking what kind of story the user would like to hear
5. Create engaging dialogue with distinct characters
Remember: Use narrator voice for EVERYTHING except the actual quoted dialogue."""
# Set up LLM context
messages = [
{
"role": "system",
"content": system_prompt,
},
]
context = OpenAILLMContext(messages)
context_aggregator = llm.create_context_aggregator(context)
# Create pipeline
pipeline = Pipeline(
[
transport.input(),
context_aggregator.user(),
llm,
tts, # TTS with pattern aggregator
transport.output(),
context_aggregator.assistant(),
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
allow_interruptions=True,
enable_metrics=True,
enable_usage_metrics=True,
report_only_initial_ttfb=True,
),
)
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
logger.info(f"First participant joined: {participant['id']}")
await transport.capture_participant_transcription(participant["id"])
# Start conversation - empty prompt to let LLM follow system instructions
await task.queue_frames([context_aggregator.user().get_context_frame()])
@transport.event_handler("on_participant_left")
async def on_participant_left(transport, participant, reason):
logger.info(f"Participant left: {participant['id']}")
await task.cancel()
logger.info(f"Starting storytelling bot at: {room_url}")
logger.info("Join the room to interact with the bot!")
runner = PipelineRunner()
await runner.run(task)
if __name__ == "__main__":
asyncio.run(main())

262
src/pipecat/utils/text/pattern_pair_aggregator.py Normal file
View File

@@ -0,0 +1,262 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import re
from typing import Callable, Optional, Tuple
from loguru import logger
from pipecat.utils.string import match_endofsentence
from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
class PatternMatch:
"""Represents a matched pattern pair with its content.
A PatternMatch object is created when a complete pattern pair is found
in the text. It contains information about which pattern was matched,
the full matched text (including start and end patterns), and the
content between the patterns.
Attributes:
pattern_id: The identifier of the matched pattern pair.
full_match: The complete text including start and end patterns.
content: The text content between the start and end patterns.
"""
def __init__(self, pattern_id: str, full_match: str, content: str):
"""Initialize a pattern match.
Args:
pattern_id: ID of the pattern pair.
full_match: Complete matched text including start and end patterns.
content: Content between the start and end patterns.
"""
self.pattern_id = pattern_id
self.full_match = full_match
self.content = content
def __str__(self) -> str:
"""Return a string representation of the pattern match.
Returns:
A string describing the pattern match.
"""
return f"PatternMatch(id={self.pattern_id}, content={self.content})"
class PatternPairAggregator(BaseTextAggregator):
"""Aggregator that identifies and processes content between pattern pairs.
This aggregator buffers text until it can identify complete pattern pairs
(defined by start and end patterns), processes the content between these
patterns using registered handlers, and returns text at sentence boundaries.
It's particularly useful for processing structured content in streaming text,
such as XML tags, markdown formatting, or custom delimiters.
The aggregator ensures that patterns spanning multiple text chunks are
correctly identified and handles cases where patterns contain sentence
boundaries.
"""
def __init__(self):
"""Initialize the pattern pair aggregator.
Creates an empty aggregator with no patterns or handlers registered.
"""
self._text = ""
self._patterns = {}
self._handlers = {}
@property
def text(self) -> str:
"""Get the currently buffered text.
Returns:
The current text buffer content.
"""
return self._text
def add_pattern_pair(
self, pattern_id: str, start_pattern: str, end_pattern: str, remove_match: bool = True
) -> "PatternPairAggregator":
"""Add a pattern pair to detect in the text.
Registers a new pattern pair with a unique identifier. The aggregator
will look for text that starts with the start pattern and ends with
the end pattern, and treat the content between them as a match.
Args:
pattern_id: Unique identifier for this pattern pair.
start_pattern: Pattern that marks the beginning of content.
end_pattern: Pattern that marks the end of content.
remove_match: Whether to remove the matched content from the text.
Returns:
Self for method chaining.
"""
self._patterns[pattern_id] = {
"start": start_pattern,
"end": end_pattern,
"remove_match": remove_match,
}
return self
def on_pattern_match(
self, pattern_id: str, handler: Callable[[PatternMatch], None]
) -> "PatternPairAggregator":
"""Register a handler for when a pattern pair is matched.
The handler will be called whenever a complete match for the
specified pattern ID is found in the text.
Args:
pattern_id: ID of the pattern pair to match.
handler: Function to call when pattern is matched.
The function should accept a PatternMatch object.
Returns:
Self for method chaining.
"""
self._handlers[pattern_id] = handler
return self
def _process_complete_patterns(self, text: str) -> Tuple[str, bool]:
"""Process all complete pattern pairs in the text.
Searches for all complete pattern pairs in the text, calls the
appropriate handlers, and optionally removes the matches.
Args:
text: The text to process.
Returns:
Tuple of (processed_text, was_modified) where:
- processed_text is the text after processing patterns
- was_modified indicates whether any changes were made
"""
processed_text = text
modified = False
for pattern_id, pattern_info in self._patterns.items():
# Escape special regex characters in the patterns
start = re.escape(pattern_info["start"])
end = re.escape(pattern_info["end"])
remove_match = pattern_info["remove_match"]
# Create regex to match from start pattern to end pattern
# The .*? is non-greedy to handle nested patterns
regex = f"{start}(.*?){end}"
# Find all matches
match_iter = re.finditer(regex, processed_text, re.DOTALL)
matches = list(match_iter) # Convert to list for safe iteration
for match in matches:
content = match.group(1) # Content between patterns
full_match = match.group(0) # Full match including patterns
# Create pattern match object
pattern_match = PatternMatch(
pattern_id=pattern_id, full_match=full_match, content=content
)
# Call the appropriate handler if registered
if pattern_id in self._handlers:
try:
self._handlers[pattern_id](pattern_match)
except Exception as e:
logger.error(f"Error in pattern handler for {pattern_id}: {e}")
# Remove the pattern from the text if configured
if remove_match:
processed_text = processed_text.replace(full_match, "", 1)
modified = True
return processed_text, modified
def _has_incomplete_patterns(self, text: str) -> bool:
"""Check if text contains incomplete pattern pairs.
Determines whether the text contains any start patterns without
matching end patterns, which would indicate incomplete content.
Args:
text: The text to check.
Returns:
True if there are incomplete patterns, False otherwise.
"""
for pattern_id, pattern_info in self._patterns.items():
start = pattern_info["start"]
end = pattern_info["end"]
# Count occurrences
start_count = text.count(start)
end_count = text.count(end)
# If there are more starts than ends, we have incomplete patterns
if start_count > end_count:
return True
return False
def aggregate(self, text: str) -> Optional[str]:
"""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.
Args:
text: New text to add to the buffer.
Returns:
Processed text up to a sentence boundary, or None if more
text is needed to form a complete sentence or pattern.
"""
# Add new text to buffer
self._text += text
# Process any complete patterns in the buffer
processed_text, modified = self._process_complete_patterns(self._text)
# Only update the buffer if modifications were made
if modified:
self._text = processed_text
# Check if we have incomplete patterns
if self._has_incomplete_patterns(self._text):
# Still waiting for complete patterns
return None
# 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 result
# No complete sentence found yet
return None
def handle_interruption(self):
"""Handle interruptions by clearing the buffer.
Called when an interruption occurs in the processing pipeline,
to reset the state and discard any partially aggregated text.
"""
self._text = ""
def reset(self):
"""Clear the internally aggregated text.
Resets the aggregator to its initial state, discarding any
buffered text.
"""
self._text = ""

147
tests/test_pattern_pair_aggregator.py Normal file
View File

@@ -0,0 +1,147 @@
#
# Copyright (c) 2024-2025 Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import unittest
from unittest.mock import Mock
from pipecat.utils.text.pattern_pair_aggregator import PatternMatch, PatternPairAggregator
class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
def setUp(self):
self.aggregator = PatternPairAggregator()
self.test_handler = Mock()
# Add a test pattern
self.aggregator.add_pattern_pair(
pattern_id="test_pattern",
start_pattern="<test>",
end_pattern="</test>",
remove_match=True,
)
# Register the mock handler
self.aggregator.on_pattern_match("test_pattern", self.test_handler)
async def test_pattern_match_and_removal(self):
# First part doesn't complete the pattern
result = self.aggregator.aggregate("Hello <test>pattern")
self.assertIsNone(result)
self.assertEqual(self.aggregator.text, "Hello <test>pattern")
# Second part completes the pattern and includes an exclamation point
result = self.aggregator.aggregate(" content</test>!")
# Verify the handler was called with correct PatternMatch object
self.test_handler.assert_called_once()
call_args = self.test_handler.call_args[0][0]
self.assertIsInstance(call_args, PatternMatch)
self.assertEqual(call_args.pattern_id, "test_pattern")
self.assertEqual(call_args.full_match, "<test>pattern content</test>")
self.assertEqual(call_args.content, "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, "Hello !")
# Next sentence should be processed separately
result = self.aggregator.aggregate(" This is another sentence.")
self.assertEqual(result, " This is another sentence.")
# Buffer should be empty after returning a complete sentence
self.assertEqual(self.aggregator.text, "")
async def test_incomplete_pattern(self):
# Add text with incomplete pattern
result = self.aggregator.aggregate("Hello <test>pattern content")
# No complete pattern yet, so nothing should be returned
self.assertIsNone(result)
# The handler should not be called yet
self.test_handler.assert_not_called()
# Buffer should contain the incomplete text
self.assertEqual(self.aggregator.text, "Hello <test>pattern content")
# Reset and confirm buffer is cleared
self.aggregator.reset()
self.assertEqual(self.aggregator.text, "")
async def test_multiple_patterns(self):
# Set up multiple patterns and handlers
voice_handler = Mock()
emphasis_handler = Mock()
self.aggregator.add_pattern_pair(
pattern_id="voice", start_pattern="<voice>", end_pattern="</voice>", remove_match=True
)
self.aggregator.add_pattern_pair(
pattern_id="emphasis",
start_pattern="<em>",
end_pattern="</em>",
remove_match=False, # Keep emphasis tags
)
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 = self.aggregator.aggregate(text)
# Both handlers should be called with correct data
voice_handler.assert_called_once()
voice_match = voice_handler.call_args[0][0]
self.assertEqual(voice_match.pattern_id, "voice")
self.assertEqual(voice_match.content, "female")
emphasis_handler.assert_called_once()
emphasis_match = emphasis_handler.call_args[0][0]
self.assertEqual(emphasis_match.pattern_id, "emphasis")
self.assertEqual(emphasis_match.content, "very")
# Voice pattern should be removed, emphasis pattern should remain
self.assertEqual(result, "Hello I am <em>very</em> excited to meet you!")
# Buffer should be empty
self.assertEqual(self.aggregator.text, "")
async def test_handle_interruption(self):
# Start with incomplete pattern
result = self.aggregator.aggregate("Hello <test>pattern")
self.assertIsNone(result)
# Simulate interruption
self.aggregator.handle_interruption()
# Buffer should be cleared
self.assertEqual(self.aggregator.text, "")
# Handler should not have been called
self.test_handler.assert_not_called()
async def test_pattern_across_sentences(self):
# Test pattern that spans multiple sentences
result = 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 = self.aggregator.aggregate(" This is sentence two.</test> Final sentence.")
# 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.content, "This is sentence one. This is sentence two.")
# Pattern should be removed, resulting in text with sentences merged
self.assertEqual(result, "Hello Final sentence.")
# Buffer should be empty
self.assertEqual(self.aggregator.text, "")