PR Feedback

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()`:
+  - Introduced a new methods, `add_text_transformer()` and `remove_text_transformer()`:
-    This function provides the ability to provide callbacks to the TTS to transform text based on
+    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.

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

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
+This processor will convert LLMTextFrames into AggregatedTextFrames based on the
-before they are sent to other components such as TTS services or context
+configured text aggregator. Using the customizable aggregator, it provides
-aggregators. It can be used to pre-aggregate, modify, or filter direct output
+functionality to handle or manipulate LLM text frames before they are sent to other
-tokens from the LLM.
+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
+    This processor will convert LLMTextFrames into AggregatedTextFrames based on the configured
-    before they are sent to other components such as TTS services or context aggregators.
+    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):

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
+    spoken: bool = False  # Indicates if the text has been spoken by TTS
-    aggregated_by: Optional[AggregationType | str] = None
+    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(
+    def add_bot_output_transformer(
-        self, aggregation_type: str, transform_function: Callable[[str, str], Awaitable[str]]
+        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))

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(
+    def add_text_transformer(
-        self, aggregation_type: str, transform_function: Callable[[str, str], Awaitable[str]]
+        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: