diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py
index f54ee58c6..aa3dc2487 100644
--- a/src/pipecat/frames/frames.py
+++ b/src/pipecat/frames/frames.py
@@ -359,12 +359,15 @@ class LLMTextFrame(TextFrame):
pass
-class AggregationType(Enum):
+class AggregationType(str, Enum):
"""Built-in aggregation strings."""
SENTENCE = "sentence"
WORD = "word"
+ def __str__(self):
+ return self.value
+
@dataclass
class AggregatedTextFrame(TextFrame):
diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py
index 8b2370967..460b00288 100644
--- a/src/pipecat/processors/aggregators/llm_response_universal.py
+++ b/src/pipecat/processors/aggregators/llm_response_universal.py
@@ -24,7 +24,6 @@ from pipecat.audio.interruptions.base_interruption_strategy import BaseInterrupt
from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams
from pipecat.audio.vad.vad_analyzer import VADParams
from pipecat.frames.frames import (
- AggregatedTextFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
CancelFrame,
@@ -48,7 +47,6 @@ from pipecat.frames.frames import (
LLMRunFrame,
LLMSetToolChoiceFrame,
LLMSetToolsFrame,
- LLMTextFrame,
SpeechControlParamsFrame,
StartFrame,
TextFrame,
@@ -69,8 +67,6 @@ from pipecat.processors.aggregators.llm_response import (
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.utils.string import concatenate_aggregated_text
-from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
-from pipecat.utils.text.simple_text_aggregator import SimpleTextAggregator
from pipecat.utils.time import time_now_iso8601
@@ -599,11 +595,6 @@ class LLMAssistantAggregator(LLMContextAggregator):
self._function_calls_in_progress: Dict[str, Optional[FunctionCallInProgressFrame]] = {}
self._context_updated_tasks: Set[asyncio.Task] = set()
- self._llm_text_aggregator: BaseTextAggregator = (
- self._params.llm_text_aggregator or SimpleTextAggregator()
- )
- self._skip_tts = None
-
@property
def has_function_calls_in_progress(self) -> bool:
"""Check if there are any function calls currently in progress.
@@ -627,9 +618,6 @@ class LLMAssistantAggregator(LLMContextAggregator):
await self.push_frame(frame, direction)
elif isinstance(frame, LLMFullResponseStartFrame):
await self._handle_llm_start(frame)
- # as a subclass of TextFrame, LLMTextFrame must be checked first
- elif isinstance(frame, LLMTextFrame):
- await self._handle_llm_text(frame)
elif isinstance(frame, LLMFullResponseEndFrame):
await self._handle_llm_end(frame)
elif isinstance(frame, TextFrame):
@@ -818,47 +806,12 @@ class LLMAssistantAggregator(LLMContextAggregator):
await self.push_aggregation()
await self.push_context_frame(FrameDirection.UPSTREAM)
- async def _handle_llm_start(self, frame: LLMFullResponseStartFrame):
+ async def _handle_llm_start(self, _: LLMFullResponseStartFrame):
self._started += 1
- if self._skip_tts is None:
- # initialize skip_tts on first start frame
- self._skip_tts = frame.skip_tts
- async def _handle_llm_text(self, frame: LLMTextFrame):
- await self._handle_text(frame)
- await self._maybe_push_llm_aggregation(frame)
-
- async def _handle_llm_end(self, frame: LLMFullResponseEndFrame):
+ async def _handle_llm_end(self, _: LLMFullResponseEndFrame):
self._started -= 1
await self.push_aggregation()
- await self._maybe_push_llm_aggregation(frame)
-
- async def _maybe_push_llm_aggregation(self, frame: LLMTextFrame | LLMFullResponseEndFrame):
- aggregate = None
- should_reset_aggregator = False
- if self._skip_tts and not frame.skip_tts:
- # When skip_tts transitions to False, we need to push any accumulated text.
- # This ensures that any remaining text accumulated while TTS was skipped is
- # sent out when TTS resumes, preventing loss of data and maintaining a smooth
- # transition.
- aggregate = self._llm_text_aggregator.text
- should_reset_aggregator = True
- self._skip_tts = frame.skip_tts
- if self._skip_tts:
- if isinstance(frame, LLMFullResponseEndFrame):
- # on end frame, always push the aggregation
- aggregate = self._llm_text_aggregator.text
- should_reset_aggregator = True
- else: # This is an LLMTextFrame
- aggregate = await self._llm_text_aggregator.aggregate(frame.text)
-
- if not aggregate:
- return
-
- llm_frame = AggregatedTextFrame(text=aggregate.text, aggregated_by=aggregate.type)
- await self.push_frame(llm_frame)
- if should_reset_aggregator:
- await self._llm_text_aggregator.reset()
async def _handle_text(self, frame: TextFrame):
if not self._started or not frame.append_to_context:
diff --git a/src/pipecat/processors/aggregators/llm_text_processor.py b/src/pipecat/processors/aggregators/llm_text_processor.py
new file mode 100644
index 000000000..e8d338310
--- /dev/null
+++ b/src/pipecat/processors/aggregators/llm_text_processor.py
@@ -0,0 +1,102 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""LLM text processor module for processing and aggregating raw LLM output text.
+
+This processor provides functionality to handle or manipulate LLM text frames
+before they are sent to other components such as TTS services or context
+aggregators. It can be used to pre-aggregate, modify, or filter direct output
+tokens from the LLM.
+"""
+
+from typing import Optional
+
+from pipecat.frames.frames import (
+ AggregatedTextFrame,
+ EndFrame,
+ Frame,
+ InterruptionFrame,
+ LLMFullResponseEndFrame,
+ LLMTextFrame,
+)
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
+from pipecat.utils.text.simple_text_aggregator import SimpleTextAggregator
+
+
+class LLMTextProcessor(FrameProcessor):
+ """A processor for handling or manipulating LLM text frames before they are processed further.
+
+ This processor can be used to pre-aggregate, modify, or filter direct output tokens from the LLM
+ before they are sent to other components such as TTS services or context aggregators.
+ """
+
+ def __init__(self, *, text_aggregator: Optional[BaseTextAggregator] = None, **kwargs):
+ """Initialize the LLM text processor.
+
+ Args:
+ text_aggregator: An optional text aggregator to use for processing LLM text frames. By
+ default, a SimpleTextAggregator aggregating by sentence will be used.
+ **kwargs: Additional arguments passed to parent class.
+
+ TODO: Allow transformations per aggregation type or all (and deprecate the TTS filters).
+ """
+ super().__init__(**kwargs)
+ self._text_aggregator: BaseTextAggregator = text_aggregator or SimpleTextAggregator()
+
+ async def process_frame(self, frame: Frame, direction: FrameDirection):
+ """Process an LLMTextFrames using the aggregator to generate AggregatedTextFrames.
+
+ Args:
+ frame: The frame to process.
+ direction: The direction of frame flow in the pipeline.
+ """
+ await super().process_frame(frame, direction)
+
+ if isinstance(frame, InterruptionFrame):
+ await self._handle_interruption(frame)
+ await self.push_frame(frame, direction)
+ elif isinstance(frame, LLMTextFrame):
+ await self._handle_llm_text(frame)
+ elif isinstance(frame, LLMFullResponseEndFrame):
+ await self._handle_llm_end(frame.skip_tts)
+ await self.push_frame(frame, direction)
+ elif isinstance(frame, EndFrame):
+ await self._handle_llm_end()
+ await self.push_frame(frame, direction)
+ else:
+ await self.push_frame(frame, direction)
+
+ async def _handle_interruption(self, _):
+ """Handle interruptions by resetting the text aggregator."""
+ await self._text_aggregator.handle_interruption()
+
+ async def reset(self):
+ """Reset the internal state of the text processor and its aggregator."""
+ 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:
+ 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: bool = False):
+ # 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:
+ out_frame = AggregatedTextFrame(
+ text=text,
+ aggregated_by=aggregation.type,
+ )
+ out_frame.skip_tts = skip_tts
+ await self.push_frame(out_frame)
diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py
index e66aeef1b..90e51e7c4 100644
--- a/src/pipecat/processors/frameworks/rtvi.py
+++ b/src/pipecat/processors/frameworks/rtvi.py
@@ -24,6 +24,7 @@ from typing import (
Literal,
Mapping,
Optional,
+ Tuple,
Union,
)
@@ -933,6 +934,9 @@ class RTVIObserverParams:
metrics_enabled: Indicates if metrics messages should be sent.
system_logs_enabled: Indicates if system logs should be sent.
errors_enabled: [Deprecated] Indicates if errors messages should be sent.
+ skip_aggregator_types: List of aggregation types to skip sending as tts/output messages.
+ Note: if using this to avoid sending secure information, be sure to also disable
+ bot_llm_enabled to avoid leaking through LLM messages.
audio_level_period_secs: How often audio levels should be sent if enabled.
"""
@@ -948,6 +952,7 @@ class RTVIObserverParams:
metrics_enabled: bool = True
system_logs_enabled: bool = False
errors_enabled: Optional[bool] = None
+ skip_aggregator_types: Optional[List[AggregationType | str]] = None
audio_level_period_secs: float = 0.15
@@ -1000,8 +1005,26 @@ class RTVIObserver(BaseObserver):
DeprecationWarning,
)
+ self._aggregation_transforms: List[Tuple[str, Callable[[str, str], Awaitable[str]]]] = []
+
+ def transform_aggregation_type(
+ self, aggregation_type: str, transform_function: Callable[[str, str], Awaitable[str]]
+ ):
+ """Transform text for a specific aggregation type before sending as Bot Output or TTS.
+
+ # TODO: What if someone wanted to remove a registered transform?
+
+ Args:
+ aggregation_type: The type of aggregation to transform. This value can be set to "*" to
+ handle all text before sending to the client.
+ transform_function: The function to apply for transformation. This function should take
+ the text and aggregation type as input and return the transformed text.
+ Ex.: async def my_transform(text: str, aggregation_type: str) -> str:
+ """
+ self._aggregation_transforms.append((aggregation_type, transform_function))
+
async def _logger_sink(self, message):
- """Logger sink so we cna send system logs to RTVI clients."""
+ """Logger sink so we can send system logs to RTVI clients."""
message = RTVISystemLogMessage(data=RTVITextMessageData(text=message))
await self.send_rtvi_message(message)
@@ -1138,17 +1161,28 @@ class RTVIObserver(BaseObserver):
async def _handle_aggregated_llm_text(self, frame: AggregatedTextFrame):
"""Handle aggregated LLM text output frames."""
+ # Skip certain aggregator types if configured to do so.
+ if (
+ self._params.skip_aggregator_types
+ and frame.aggregated_by in self._params.skip_aggregator_types
+ ):
+ return
+
+ text = frame.text
+ type = frame.aggregated_by
+ for aggregation_type, transform in self._aggregation_transforms:
+ if aggregation_type == type or aggregation_type == "*":
+ text = await transform(text, type)
+
isTTS = isinstance(frame, TTSTextFrame)
if self._params.bot_output_enabled:
message = RTVIBotOutputMessage(
- data=RTVIBotOutputMessageData(
- text=frame.text, spoken=isTTS, aggregated_by=frame.aggregated_by
- )
+ data=RTVIBotOutputMessageData(text=text, spoken=isTTS, aggregated_by=type)
)
await self.send_rtvi_message(message)
if isTTS and self._params.bot_tts_enabled:
- tts_message = RTVIBotTTSTextMessage(data=RTVITextMessageData(text=frame.text))
+ tts_message = RTVIBotTTSTextMessage(data=RTVITextMessageData(text=text))
await self.send_rtvi_message(tts_message)
async def _handle_llm_text_frame(self, frame: LLMTextFrame):
@@ -1156,7 +1190,7 @@ class RTVIObserver(BaseObserver):
message = RTVIBotLLMTextMessage(data=RTVITextMessageData(text=frame.text))
await self.send_rtvi_message(message)
- # TODO: Remove all this logic when we fully deprecate bot-transcription messages.
+ # TODO (mrkb): Remove all this logic when we fully deprecate bot-transcription messages.
self._bot_transcription += frame.text
if match_endofsentence(self._bot_transcription) and len(self._bot_transcription) > 0:
diff --git a/src/pipecat/services/cartesia/tts.py b/src/pipecat/services/cartesia/tts.py
index f8881200c..d42802cf2 100644
--- a/src/pipecat/services/cartesia/tts.py
+++ b/src/pipecat/services/cartesia/tts.py
@@ -10,7 +10,8 @@ import base64
import json
import uuid
import warnings
-from typing import AsyncGenerator, List, Literal, Optional, Union
+from enum import Enum
+from typing import AsyncGenerator, List, Literal, Optional
from loguru import logger
from pydantic import BaseModel, Field
@@ -125,6 +126,72 @@ def language_to_cartesia_language(language: Language) -> Optional[str]:
return resolve_language(language, LANGUAGE_MAP, use_base_code=True)
+class CartesiaEmotion(str, Enum):
+ """Predefined Emotions supported by Cartesia."""
+
+ # Primary emotions supported by Cartesia
+ NEUTRAL = "neutral"
+ ANGRY = "angry"
+ EXCITED = "excited"
+ CONTENT = "content"
+ SAD = "sad"
+ SCARED = "scared"
+ # Additional emotions supported by Cartesia
+ HAPPY = "happy"
+ ENTHUSIASTIC = "enthusiastic"
+ ELATED = "elated"
+ EUPHORIC = "euphoric"
+ TRIUMPHANT = "triumphant"
+ AMAZED = "amazed"
+ SURPRISED = "surprised"
+ FLIRTATIOUS = "flirtatious"
+ JOKING_COMEDIC = "joking/comedic"
+ CURIOUS = "curious"
+ PEACEFUL = "peaceful"
+ SERENE = "serene"
+ CALM = "calm"
+ GRATEFUL = "grateful"
+ AFFECTIONATE = "affectionate"
+ TRUST = "trust"
+ SYMPATHETIC = "sympathetic"
+ ANTICIPATION = "anticipation"
+ MYSTERIOUS = "mysterious"
+ MAD = "mad"
+ OUTRAGED = "outraged"
+ FRUSTRATED = "frustrated"
+ AGITATED = "agitated"
+ THREATENED = "threatened"
+ DISGUSTED = "disgusted"
+ CONTEMPT = "contempt"
+ ENVIOUS = "envious"
+ SARCASTIC = "sarcastic"
+ IRONIC = "ironic"
+ DEJECTED = "dejected"
+ MELANCHOLIC = "melancholic"
+ DISAPPOINTED = "disappointed"
+ HURT = "hurt"
+ GUILTY = "guilty"
+ BORED = "bored"
+ TIRED = "tired"
+ REJECTED = "rejected"
+ NOSTALGIC = "nostalgic"
+ WISTFUL = "wistful"
+ APOLOGETIC = "apologetic"
+ HESITANT = "hesitant"
+ INSECURE = "insecure"
+ CONFUSED = "confused"
+ RESIGNED = "resigned"
+ ANXIOUS = "anxious"
+ PANICKED = "panicked"
+ ALARMED = "alarmed"
+ PROUD = "proud"
+ CONFIDENT = "confident"
+ DISTANT = "distant"
+ SKEPTICAL = "skeptical"
+ CONTEMPLATIVE = "contemplative"
+ DETERMINED = "determined"
+
+
class CartesiaTTSService(AudioContextWordTTSService):
"""Cartesia TTS service with WebSocket streaming and word timestamps.
@@ -182,6 +249,10 @@ class CartesiaTTSService(AudioContextWordTTSService):
container: Audio container format.
params: Additional input parameters for voice customization.
text_aggregator: Custom text aggregator for processing input text.
+
+ .. deprecated:: 0.0.95
+ Use an LLMTextProcessor before the TTSService for custom text aggregation.
+
aggregate_sentences: Whether to aggregate sentences within the TTSService.
**kwargs: Additional arguments passed to the parent service.
"""
@@ -200,10 +271,18 @@ class CartesiaTTSService(AudioContextWordTTSService):
push_text_frames=False,
pause_frame_processing=True,
sample_rate=sample_rate,
- text_aggregator=text_aggregator or SkipTagsAggregator([("", "")]),
+ text_aggregator=text_aggregator,
**kwargs,
)
+ if not text_aggregator:
+ # Always skip tags added for spelled-out text
+ # Note: This is primarily to support backwards compatibility.
+ # The preferred way of taking advantage of Cartesia SSML Tags is
+ # to use an LLMTextProcessor and/or a text_transformer to identify
+ # and insert these tags for the purpose of the TTS service alone.
+ self._text_aggregator = SkipTagsAggregator([("", "")])
+
params = params or CartesiaTTSService.InputParams()
self._api_key = api_key
@@ -257,6 +336,27 @@ class CartesiaTTSService(AudioContextWordTTSService):
"""
return language_to_cartesia_language(language)
+ # A set of Cartesia-specific helpers for text transformations
+ def SPELL(text: str) -> str:
+ """Wrap text in Cartesia spell tag."""
+ return f"{text}"
+
+ def EMOTION_TAG(emotion: CartesiaEmotion) -> str:
+ """Convenience method to create an emotion tag."""
+ return f''
+
+ def PAUSE_TAG(seconds: float) -> str:
+ """Convenience method to create a pause tag."""
+ return f''
+
+ def VOLUME_TAG(volume: float) -> str:
+ """Convenience method to create a volume tag."""
+ return f''
+
+ def SPEED_TAG(speed: float) -> str:
+ """Convenience method to create a speed tag."""
+ return f''
+
def _is_cjk_language(self, language: str) -> bool:
"""Check if the given language is CJK (Chinese, Japanese, Korean).
diff --git a/src/pipecat/services/rime/tts.py b/src/pipecat/services/rime/tts.py
index 329aecd7d..52cc72c9b 100644
--- a/src/pipecat/services/rime/tts.py
+++ b/src/pipecat/services/rime/tts.py
@@ -113,6 +113,10 @@ class RimeTTSService(AudioContextWordTTSService):
sample_rate: Audio sample rate in Hz.
params: Additional configuration parameters.
text_aggregator: Custom text aggregator for processing input text.
+
+ .. deprecated:: 0.0.95
+ Use an LLMTextProcessor before the TTSService for custom text aggregation.
+
aggregate_sentences: Whether to aggregate sentences within the TTSService.
**kwargs: Additional arguments passed to parent class.
"""
@@ -123,10 +127,17 @@ class RimeTTSService(AudioContextWordTTSService):
push_stop_frames=True,
pause_frame_processing=True,
sample_rate=sample_rate,
- text_aggregator=text_aggregator or SkipTagsAggregator([("spell(", ")")]),
**kwargs,
)
+ if not text_aggregator:
+ # Always skip tags added for spelled-out text
+ # Note: This is primarily to support backwards compatibility.
+ # The preferred way of taking advantage of Rime spelling is
+ # to use an LLMTextProcessor and/or a text_transformer to identify
+ # and insert these tags for the purpose of the TTS service alone.
+ self._text_aggregator = SkipTagsAggregator([("spell(", ")")])
+
params = params or RimeTTSService.InputParams()
# Store service configuration
@@ -152,6 +163,7 @@ class RimeTTSService(AudioContextWordTTSService):
self._context_id = None # Tracks current turn
self._receive_task = None
self._cumulative_time = 0 # Accumulates time across messages
+ self._extra_msg_fields = {} # Extra fields for next message
def can_generate_metrics(self) -> bool:
"""Check if this service can generate processing metrics.
@@ -181,6 +193,31 @@ class RimeTTSService(AudioContextWordTTSService):
self._model = model
await super().set_model(model)
+ # A set of Rime-specific helpers for text transformations
+ def SPELL(text: str) -> str:
+ """Wrap text in Rime spell function."""
+ return f"spell({text})"
+
+ def PAUSE_TAG(seconds: float) -> str:
+ """Convenience method to create a pause tag."""
+ return f"<{seconds * 1000}>"
+
+ def PRONOUNCE(self, text: str, word: str, phoneme: str) -> str:
+ """Convenience method to support Rime's custom pronunciations feature.
+
+ https://docs.rime.ai/api-reference/custom-pronunciation
+ """
+ self._extra_msg_fields["phonemizeBetweenBrackets"] = True
+ return text.replace(word, f"{phoneme}")
+
+ def INLINE_SPEED(self, text: str, speed: float) -> str:
+ """Convenience method to support inline speeds."""
+ if not self._extra_msg_fields:
+ self._extra_msg_fields = {}
+ speed_vals = self._extra_msg_fields.get("inlineSpeedAlpha", "").split(",")
+ self._extra_msg_fields["inlineSpeedAlpha"] = ",".join(speed_vals + [str(speed)])
+ return f"[{text}]"
+
async def _update_settings(self, settings: Mapping[str, Any]):
"""Update service settings and reconnect if voice changed."""
prev_voice = self._voice_id
@@ -193,7 +230,11 @@ class RimeTTSService(AudioContextWordTTSService):
def _build_msg(self, text: str = "") -> dict:
"""Build JSON message for Rime API."""
- return {"text": text, "contextId": self._context_id}
+ msg = {"text": text, "contextId": self._context_id}
+ if self._extra_msg_fields:
+ msg |= self._extra_msg_fields
+ self._extra_msg_fields = {}
+ return msg
def _build_clear_msg(self) -> dict:
"""Build clear operation message."""
diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py
index 2e1d7d421..147b0a1c1 100644
--- a/src/pipecat/services/tts_service.py
+++ b/src/pipecat/services/tts_service.py
@@ -12,6 +12,8 @@ from typing import (
Any,
AsyncGenerator,
AsyncIterator,
+ Awaitable,
+ Callable,
Dict,
List,
Mapping,
@@ -124,6 +126,10 @@ class TTSService(AIService):
pause_frame_processing: Whether to pause frame processing during audio generation.
sample_rate: Output sample rate for generated audio.
text_aggregator: Custom text aggregator for processing incoming text.
+
+ .. deprecated:: 0.0.95
+ Use an LLMTextProcessor before the TTSService for custom text aggregation.
+
skip_aggregator_types: List of aggregation types that should not be spoken.
text_filters: Sequence of text filters to apply after aggregation.
text_filter: Single text filter (deprecated, use text_filters).
@@ -147,7 +153,19 @@ class TTSService(AIService):
self._voice_id: str = ""
self._settings: Dict[str, Any] = {}
self._text_aggregator: BaseTextAggregator = text_aggregator or SimpleTextAggregator()
+ if text_aggregator:
+ import warnings
+
+ with warnings.catch_warnings():
+ warnings.simplefilter("always")
+ warnings.warn(
+ "Parameter 'text_aggregator' is deprecated. Use an LLMTextProcessor before the TTSService for custom text aggregation.",
+ DeprecationWarning,
+ )
+
self._skip_aggregator_types: List[str] = skip_aggregator_types or []
+ self._text_transforms: List[Tuple[str, Callable[[str, str], Awaitable[str]]]] = []
+ # TODO: Deprecate _text_filters when added to LLMTextProcessor
self._text_filters: Sequence[BaseTextFilter] = text_filters or []
self._transport_destination: Optional[str] = transport_destination
self._tracing_enabled: bool = False
@@ -304,6 +322,22 @@ class TTSService(AIService):
await self.cancel_task(self._stop_frame_task)
self._stop_frame_task = None
+ def transform_aggregation_type(
+ self, aggregation_type: str, transform_function: Callable[[str, str], Awaitable[str]]
+ ):
+ """Transform text for a specific aggregation type.
+
+ # TODO: What if someone wanted to remove a registered transform?
+
+ Args:
+ aggregation_type: The type of aggregation to transform. This value can be set to "*" to
+ handle all text before sending to TTS.
+ transform_function: The function to apply for transformation. This function should take
+ the text and aggregation type as input and return the transformed text.
+ Ex.: async def my_transform(text: str, aggregation_type: str) -> str:
+ """
+ self._text_transforms.append((aggregation_type, transform_function))
+
async def _update_settings(self, settings: Mapping[str, Any]):
for key, value in settings.items():
if key in self._settings:
@@ -359,6 +393,8 @@ class TTSService(AIService):
and frame.skip_tts
):
await self.push_frame(frame, direction)
+ elif isinstance(frame, AggregatedTextFrame):
+ await self._push_tts_frames(frame)
elif (
isinstance(frame, TextFrame)
and not isinstance(frame, InterimTranscriptionFrame)
@@ -377,10 +413,7 @@ class TTSService(AIService):
aggregate = self._text_aggregator.text
await self._text_aggregator.reset()
self._processing_text = False
- await self._push_tts_frames(
- text=aggregate.text,
- aggregated_by=aggregate.type,
- )
+ await self._push_tts_frames(AggregatedTextFrame(aggregate.text, aggregate.type))
if isinstance(frame, LLMFullResponseEndFrame):
if self._push_text_frames:
await self.push_frame(frame, direction)
@@ -389,7 +422,7 @@ class TTSService(AIService):
elif isinstance(frame, TTSSpeakFrame):
# Store if we were processing text or not so we can set it back.
processing_text = self._processing_text
- await self._push_tts_frames(frame.text, aggregated_by=AggregationType.SENTENCE)
+ await self._push_tts_frames(AggregatedTextFrame(frame.text, AggregationType.SENTENCE))
# We pause processing incoming frames because we are sending data to
# the TTS. We pause to avoid audio overlapping.
await self._maybe_pause_frame_processing()
@@ -490,13 +523,13 @@ class TTSService(AIService):
if text:
logger.trace(f"Pushing TTS frames for text: {text}, {aggregated_by}")
- await self._push_tts_frames(text, aggregated_by)
+ await self._push_tts_frames(AggregatedTextFrame(text, aggregated_by))
- async def _push_tts_frames(self, text: str, aggregated_by: str):
- if aggregated_by in self._skip_aggregator_types:
- # If this type of aggregation should be skipped, we just push the text as
- # a basic AggregatedTextFrame without sending it to TTS to speak.
- await self.push_frame(AggregatedTextFrame(text, aggregated_by=aggregated_by))
+ async def _push_tts_frames(self, src_frame: AggregatedTextFrame):
+ type = src_frame.aggregated_by
+ text = src_frame.text
+ if type in self._skip_aggregator_types:
+ await self.push_frame(src_frame)
return
# Remove leading newlines only
@@ -527,13 +560,20 @@ class TTSService(AIService):
# is set to False and these are sent word by word as part of the
# _words_task_handler in the WordTTSService subclass. However, to
# support use cases where an observer may want the full text before
- # the audio is generated, we send an AggregatedTextFrame here, but
- # we set append_to_context to False so it does not cause duplication
+ # the audio is generated, we send along the AggregatedTextFrame here,
+ # but we set append_to_context to False so it does not cause duplication
# in the context. This is primarily used by the RTVIObserver to
# generate a complete bot-output.
- frame = AggregatedTextFrame(text, aggregated_by=aggregated_by)
- frame.append_to_context = False
- await self.push_frame(frame)
+ src_frame.append_to_context = False
+ await self.push_frame(src_frame)
+ # Note: Text transformations only affect the text sent to the TTS. This allows
+ # for explicit TTS-specific modifications (e.g., inserting TTS supported tags
+ # for spelling or emotion or replacing an @ with "at"). For TTS services that
+ # support word-level timestamps, this DOES affect the resulting context as the
+ # the context is built from the TTSTextFrames generated during word timestamping.
+ for aggregation_type, transform in self._text_transforms:
+ if aggregation_type == type or aggregation_type == "*":
+ text = await transform(text, type)
await self.process_generator(self.run_tts(text))
await self.stop_processing_metrics()
@@ -542,7 +582,7 @@ class TTSService(AIService):
# In the case where the TTS service does not support word timestamps,
# we send the full aggregated text after the audio. This way, if we are
# interrupted, the text is not added to the assistant context.
- frame = TTSTextFrame(text, aggregated_by=aggregated_by)
+ frame = TTSTextFrame(text, aggregated_by=type)
frame.includes_inter_frame_spaces = self.includes_inter_frame_spaces
await self.push_frame(frame)