From 4c698777f3a2459c69a703348b87d2e491fc0d79 Mon Sep 17 00:00:00 2001 From: mattie ruth backman Date: Thu, 13 Nov 2025 12:44:10 -0500 Subject: [PATCH] PR Feedback --- CHANGELOG.md | 8 +++--- .../processors/aggregators/llm_response.py | 2 -- .../aggregators/llm_text_processor.py | 16 ++++++----- src/pipecat/processors/frameworks/rtvi.py | 27 ++++++++++++++----- src/pipecat/services/tts_service.py | 25 ++++++++++++----- 5 files changed, 54 insertions(+), 24 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 840e92427..b09927a49 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -45,7 +45,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `bot_output_enabled`: Defaults to True. Set to false to disable bot-output messages. - `skip_aggregator_types`: Defaults to `None`. Set to a list of strings that match aggregation types that should not be included in bot-output messages. (Ex. `credit_card`) - - Introduced new `transform_aggregation_type` method to `RTVIObserver` to support providing + - Introduced new methods, `add_text_transformer()` and `remove_text_transformer()`, to `RTVIObserver` to support providing (and subsequently removing) callbacks for various types of aggregations (or all aggregations with `*`) that can modify the text before being sent as a `bot-output` or `tts-text` message. (Think obscuring the credit card or inserting extra detail the client might want that the context doesn't need.) @@ -95,7 +95,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 as a separate aggregation. Any text before the start of the pattern will be returned early, whether or not a complete sentence was found. Then the pattern will be returned. Then the aggregation will continue on sentence matching after - the closing delimeter is found. The content between the delimeters is not + the closing delimiter is found. The content between the delimiters is not aggregated by sentence. It is aggregated as one single block of text. - `PatternMatch` now extends `Aggregation` and provides richer info to handlers. @@ -130,8 +130,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 timestamping. In the latter case, the `TTSService` preliminarily generates an `AggregatedTextFrame`, aggregated by sentence to generate the full sentence content as early as possible. - - Introduced a new method, `transform_aggregation_type()`: - This function provides the ability to provide callbacks to the TTS to transform text based on + - Introduced a new methods, `add_text_transformer()` and `remove_text_transformer()`: + These functions introduce the ability to provide (and subsequently remove) 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. diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index ad64b0894..ec13b643f 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -59,7 +59,6 @@ from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContextFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.utils.text.base_text_aggregator import BaseTextAggregator from pipecat.utils.time import time_now_iso8601 @@ -96,7 +95,6 @@ class LLMAssistantAggregatorParams: """ expect_stripped_words: bool = True - llm_text_aggregator: Optional[BaseTextAggregator] = None class LLMFullResponseAggregator(FrameProcessor): diff --git a/src/pipecat/processors/aggregators/llm_text_processor.py b/src/pipecat/processors/aggregators/llm_text_processor.py index e8d338310..44a8dc24e 100644 --- a/src/pipecat/processors/aggregators/llm_text_processor.py +++ b/src/pipecat/processors/aggregators/llm_text_processor.py @@ -6,10 +6,11 @@ """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. +This processor will convert LLMTextFrames into AggregatedTextFrames based on the +configured text aggregator. Using the customizable aggregator, it 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 +and categorize, modify, or filter direct output tokens from the LLM. """ from typing import Optional @@ -30,8 +31,11 @@ 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. + This processor will convert LLMTextFrames into AggregatedTextFrames based on the configured + text aggregator. Using the customizable aggregator, it 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 and categorize, modify, or filter direct + output tokens from the LLM. """ def __init__(self, *, text_aggregator: Optional[BaseTextAggregator] = None, **kwargs): diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 90e51e7c4..31229bb3a 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -713,8 +713,8 @@ class RTVIBotOutputMessageData(RTVITextMessageData): Extends RTVITextMessageData to include metadata about the output. """ - spoken: bool = True # Indicates if the text has been spoken by TTS - aggregated_by: Optional[AggregationType | str] = None + spoken: bool = False # Indicates if the text has been spoken by TTS + aggregated_by: AggregationType | str # Indicates what form the text is in (e.g., by word, sentence, etc.) @@ -1007,22 +1007,37 @@ class RTVIObserver(BaseObserver): 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]] + def add_bot_output_transformer( + self, transform_function: Callable[[str, str], Awaitable[str]], aggregation_type: 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: + aggregation_type: The type of aggregation to transform. This value defaults to "*" to + handle all text before sending to the client. """ self._aggregation_transforms.append((aggregation_type, transform_function)) + def remove_bot_output_transformer( + self, transform_function: Callable[[str, str], Awaitable[str]], aggregation_type: str = "*" + ): + """Remove a text transformer for a specific aggregation type. + + Args: + transform_function: The function to remove. + aggregation_type: The type of aggregation to remove the transformer for. + """ + self._aggregation_transforms = [ + (agg_type, func) + for agg_type, func in self._aggregation_transforms + if not (agg_type == aggregation_type and func == transform_function) + ] + async def _logger_sink(self, message): """Logger sink so we can send system logs to RTVI clients.""" message = RTVISystemLogMessage(data=RTVITextMessageData(text=message)) diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index 147b0a1c1..09304a581 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -322,22 +322,35 @@ 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]] + def add_text_transformer( + self, transform_function: Callable[[str, str], Awaitable[str]], aggregation_type: 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: + aggregation_type: The type of aggregation to transform. This value defaults to "*" indicating + the function should handle all text before sending to TTS. """ self._text_transforms.append((aggregation_type, transform_function)) + def remove_text_transformer( + self, transform_function: Callable[[str, str], Awaitable[str]], aggregation_type: str = "*" + ): + """Remove a text transformer for a specific aggregation type. + + Args: + transform_function: The function to remove. + aggregation_type: The type of aggregation to remove the transformer for. + """ + self._text_transforms = [ + (agg_type, func) + for agg_type, func in self._text_transforms + if not (agg_type == aggregation_type and func == transform_function) + ] + async def _update_settings(self, settings: Mapping[str, Any]): for key, value in settings.items(): if key in self._settings: