add transformers to initialization args

Author: mattieruth

src/pipecat/processors/frameworks/rtvi.py

@@ -937,6 +937,10 @@ class RTVIObserverParams:
         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.
+        bot_output_transforms: A list of callables to transform text before just before sending it
+            to TTS. Each callable takes the aggregated text and its type, and returns the
+            transformed text. To register, provide a list of tuples of
+            (aggregation_type | '*', transform_function).
         audio_level_period_secs: How often audio levels should be sent if enabled.
     """
 
@@ -953,6 +957,14 @@ class RTVIObserverParams:
     system_logs_enabled: bool = False
     errors_enabled: Optional[bool] = None
     skip_aggregator_types: Optional[List[AggregationType | str]] = None
+    bot_output_transforms: Optional[
+        List[
+            Tuple[
+                AggregationType | str,
+                Callable[[str, AggregationType | str], Awaitable[str]],
+            ]
+        ]
+    ] = None
     audio_level_period_secs: float = 0.15
 
 
@@ -1005,15 +1017,17 @@ def __init__(
                     DeprecationWarning,
                 )
 
-        self._aggregation_transforms: List[Tuple[str, Callable[[str, str], Awaitable[str]]]] = []
+        self._aggregation_transforms: List[
+            Tuple[AggregationType | str, Callable[[str, AggregationType | str], Awaitable[str]]]
+        ] = self._params.bot_output_transforms or []
 
     def add_bot_output_transformer(
-        self, transform_function: Callable[[str, str], Awaitable[str]], aggregation_type: str = "*"
+        self,
+        transform_function: Callable[[str, AggregationType | str], Awaitable[str]],
+        aggregation_type: AggregationType | 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:
             transform_function: The function to apply for transformation. This function should take
                 the text and aggregation type as input and return the transformed text.
@@ -1024,7 +1038,9 @@ def add_bot_output_transformer(
         self._aggregation_transforms.append((aggregation_type, transform_function))
 
     def remove_bot_output_transformer(
-        self, transform_function: Callable[[str, str], Awaitable[str]], aggregation_type: str = "*"
+        self,
+        transform_function: Callable[[str, AggregationType | str], Awaitable[str]],
+        aggregation_type: AggregationType | str = "*",
     ):
         """Remove a text transformer for a specific aggregation type.
 

src/pipecat/services/tts_service.py

@@ -107,6 +107,14 @@ def __init__(
         text_aggregator: Optional[BaseTextAggregator] = None,
         # Types of text aggregations that should not be spoken.
         skip_aggregator_types: Optional[List[str]] = [],
+        # A list of callables to transform text before just before sending it to TTS.
+        # Each callable takes the aggregated text and its type, and returns the transformed text.
+        # To register, provide a list of tuples of (aggregation_type | '*', transform_function).
+        text_transforms: Optional[
+            List[
+                Tuple[AggregationType | str, Callable[[str, str | AggregationType], Awaitable[str]]]
+            ]
+        ] = None,
         # Text filter executed after text has been aggregated.
         text_filters: Optional[Sequence[BaseTextFilter]] = None,
         text_filter: Optional[BaseTextFilter] = None,
@@ -131,6 +139,11 @@ def __init__(
                     Use an LLMTextProcessor before the TTSService for custom text aggregation.
 
             skip_aggregator_types: List of aggregation types that should not be spoken.
+            text_transforms: A list of callables to transform text before just before sending it
+                to TTS. Each callable takes the aggregated text and its type, and returns the
+                transformed text. To register, provide a list of tuples of
+                (aggregation_type | '*', transform_function).
+
             text_filters: Sequence of text filters to apply after aggregation.
             text_filter: Single text filter (deprecated, use text_filters).
 
@@ -164,7 +177,9 @@ def __init__(
                 )
 
         self._skip_aggregator_types: List[str] = skip_aggregator_types or []
-        self._text_transforms: List[Tuple[str, Callable[[str, str], Awaitable[str]]]] = []
+        self._text_transforms: List[
+            Tuple[AggregationType | str, Callable[[str, AggregationType | str], Awaitable[str]]]
+        ] = text_transforms or []
         # TODO: Deprecate _text_filters when added to LLMTextProcessor
         self._text_filters: Sequence[BaseTextFilter] = text_filters or []
         self._transport_destination: Optional[str] = transport_destination
@@ -323,7 +338,9 @@ async def cancel(self, frame: CancelFrame):
             self._stop_frame_task = None
 
     def add_text_transformer(
-        self, transform_function: Callable[[str, str], Awaitable[str]], aggregation_type: str = "*"
+        self,
+        transform_function: Callable[[str, AggregationType | str], Awaitable[str]],
+        aggregation_type: AggregationType | str = "*",
     ):
         """Transform text for a specific aggregation type.
 
@@ -337,7 +354,9 @@ def add_text_transformer(
         self._text_transforms.append((aggregation_type, transform_function))
 
     def remove_text_transformer(
-        self, transform_function: Callable[[str, str], Awaitable[str]], aggregation_type: str = "*"
+        self,
+        transform_function: Callable[[str, AggregationType | str], Awaitable[str]],
+        aggregation_type: AggregationType | str = "*",
     ):
         """Remove a text transformer for a specific aggregation type.