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

Introducing a new processor: LLMTextProcessor

Browse Source 
This new processor wraps an aggregator that can be overridden for the purposes
of customizing how the llm output gets categorized and handled in the pipeline.

Along with this, we are deprecating the ability to override the default
aggregator in the TTS to encourage use of the LLMTextProcessor in cases where
custome aggregation is needed.

This PR also:
- Introduces TTSService.transform_aggregation_type():
  This function provides the ability to provide callbacks to the TTS to
  transform text based on its aggregated type prior to sending the text to the
  underlying TTS service. This makes it possible to do things like introduce
  TTS-specific tags for spelling or emotion or change the pronunciation of
  something on the fly.
- Introduces to the RTVIObserver:
  - new init field skip_aggregator_types: A way to provide a list of aggregation
    types that should not be included in bot-output (or tts-text) messages
  - transform_aggregation_type(): Same as with TTSService, this allows you
    to provide a callback to transform text being sent as bot-output before
    it gets sent.
This commit is contained in:
mattie ruth backman
2025-11-12 10:54:04 -05:00
parent 8ab0c92681
commit 9a3902a82c
7 changed files with 350 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/pipecat/frames/frames.py
View File

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

51
src/pipecat/processors/aggregators/llm_response_universal.py
View File

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

102
src/pipecat/processors/aggregators/llm_text_processor.py Normal file
View File

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

46
src/pipecat/processors/frameworks/rtvi.py
View File

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

104
src/pipecat/services/cartesia/tts.py
View File

@@ -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([("<spell>", "</spell>")]),
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([("<spell>", "</spell>")])
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"<spell>{text}</spell>"
def EMOTION_TAG(emotion: CartesiaEmotion) -> str:
"""Convenience method to create an emotion tag."""
return f'<emotion value="{emotion}" />'
def PAUSE_TAG(seconds: float) -> str:
"""Convenience method to create a pause tag."""
return f'<break time="{seconds}s" />'
def VOLUME_TAG(volume: float) -> str:
"""Convenience method to create a volume tag."""
return f'<volume ratio="{volume}" />'
def SPEED_TAG(speed: float) -> str:
"""Convenience method to create a speed tag."""
return f'<speed ratio="{speed}" />'
def _is_cjk_language(self, language: str) -> bool:
"""Check if the given language is CJK (Chinese, Japanese, Korean).

45
src/pipecat/services/rime/tts.py
View File

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

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

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