diff --git a/CHANGELOG.md b/CHANGELOG.md index 8a54c5d1d..4f213e1db 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -59,6 +59,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 TTS's internal text_aggregator, but instead, insert this processor between your LLM and TTS in the pipeline. +- New `bot-output` RTVI message to represent what the bot actually "says". + - The `RTVIObserver` now emits `bot-output` messages based off the new `AggregatedTextFrame`s + (`bot-tts-text` and `bot-llm-text` are still supported and generated, but `bot-transcript` is + now deprecated in lieu of this new, more thorough, message). + - The new `RTVIBotOutputMessage` includes the fields: + - `spoken`: A boolean indicating whether the text was spoken by TTS + - `aggregated_by`: A string representing how the text was aggregated ("sentence", "word", + "my custom aggregation") + - Introduced new fields to `RTVIObserver` to support the new `bot-output` messaging: + - `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`) + ### Changed - ⚠️ Breaking change: `LLMContext.create_image_message()`, @@ -194,6 +207,10 @@ use `test_normalization` instead. behavior, but if you want to override the aggregation behavior, you should use the new processor. +- The RTVI `bot-transcription` event is deprecated in favor of the new `bot-output` + message which is the canonical representation of bot output (spoken or not). The code + still emits a transcription message for backwards compatibility while transition occurs. + ### Fixed - Fixed a `SimliVideoService` connection issue. diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index f04cbd395..7bd84fb42 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, ) @@ -32,6 +33,8 @@ from pydantic import BaseModel, Field, PrivateAttr, ValidationError from pipecat.audio.utils import calculate_audio_volume from pipecat.frames.frames import ( + AggregatedTextFrame, + AggregationType, BotStartedSpeakingFrame, BotStoppedSpeakingFrame, CancelFrame, @@ -704,6 +707,29 @@ class RTVITextMessageData(BaseModel): text: str +class RTVIBotOutputMessageData(RTVITextMessageData): + """Data for bot output RTVI messages. + + Extends RTVITextMessageData to include metadata about the output. + """ + + 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.) + + +class RTVIBotOutputMessage(BaseModel): + """Message containing bot output text. + + An event meant to holistically represent what the bot is outputting, + along with metadata about the output and if it has been spoken. + """ + + label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL + type: Literal["bot-output"] = "bot-output" + data: RTVIBotOutputMessageData + + class RTVIBotTranscriptionMessage(BaseModel): """Message containing bot transcription text. @@ -896,6 +922,7 @@ class RTVIObserverParams: Parameter `errors_enabled` is deprecated. Error messages are always enabled. Parameters: + bot_output_enabled: Indicates if bot output messages should be sent. bot_llm_enabled: Indicates if the bot's LLM messages should be sent. bot_tts_enabled: Indicates if the bot's TTS messages should be sent. bot_speaking_enabled: Indicates if the bot's started/stopped speaking messages should be sent. @@ -907,9 +934,17 @@ 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. + 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. """ + bot_output_enabled: bool = True bot_llm_enabled: bool = True bot_tts_enabled: bool = True bot_speaking_enabled: bool = True @@ -921,6 +956,15 @@ class RTVIObserverParams: metrics_enabled: bool = True 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 @@ -973,8 +1017,45 @@ class RTVIObserver(BaseObserver): DeprecationWarning, ) + 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, AggregationType | str], Awaitable[str]], + aggregation_type: AggregationType | str = "*", + ): + """Transform text for a specific aggregation type before sending as Bot Output or TTS. + + 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. + 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, AggregationType | str], Awaitable[str]], + aggregation_type: AggregationType | 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 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) @@ -1048,12 +1129,15 @@ class RTVIObserver(BaseObserver): await self.send_rtvi_message(RTVIBotTTSStartedMessage()) elif isinstance(frame, TTSStoppedFrame) and self._params.bot_tts_enabled: await self.send_rtvi_message(RTVIBotTTSStoppedMessage()) - elif isinstance(frame, TTSTextFrame) and self._params.bot_tts_enabled: - if isinstance(src, BaseOutputTransport): - message = RTVIBotTTSTextMessage(data=RTVITextMessageData(text=frame.text)) - await self.send_rtvi_message(message) - else: + elif isinstance(frame, AggregatedTextFrame) and ( + self._params.bot_output_enabled or self._params.bot_tts_enabled + ): + if isinstance(frame, TTSTextFrame) and not isinstance(src, BaseOutputTransport): + # This check is to make sure we handle the frame when it has gone + # through the transport and has correct timing. mark_as_seen = False + else: + await self._handle_aggregated_llm_text(frame) elif isinstance(frame, MetricsFrame) and self._params.metrics_enabled: await self._handle_metrics(frame) elif isinstance(frame, RTVIServerMessageFrame): @@ -1084,15 +1168,6 @@ class RTVIObserver(BaseObserver): if mark_as_seen: self._frames_seen.add(frame.id) - async def _push_bot_transcription(self): - """Push accumulated bot transcription as a message.""" - if len(self._bot_transcription) > 0: - message = RTVIBotTranscriptionMessage( - data=RTVITextMessageData(text=self._bot_transcription) - ) - await self.send_rtvi_message(message) - self._bot_transcription = "" - async def _handle_interruptions(self, frame: Frame): """Handle user speaking interruption frames.""" message = None @@ -1115,14 +1190,45 @@ class RTVIObserver(BaseObserver): if message: await self.send_rtvi_message(message) + 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=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=text)) + await self.send_rtvi_message(tts_message) + async def _handle_llm_text_frame(self, frame: LLMTextFrame): """Handle LLM text output frames.""" message = RTVIBotLLMTextMessage(data=RTVITextMessageData(text=frame.text)) await self.send_rtvi_message(message) + # TODO (mrkb): Remove all this logic when we fully deprecate bot-transcription messages. self._bot_transcription += frame.text - if match_endofsentence(self._bot_transcription): - await self._push_bot_transcription() + + if match_endofsentence(self._bot_transcription) and len(self._bot_transcription) > 0: + await self.send_rtvi_message( + RTVIBotTranscriptionMessage(data=RTVITextMessageData(text=self._bot_transcription)) + ) + self._bot_transcription = "" async def _handle_user_transcriptions(self, frame: Frame): """Handle user transcription frames.""" @@ -1248,7 +1354,7 @@ class RTVIProcessor(FrameProcessor): # Default to 0.3.0 which is the last version before actually having a # "client-version". self._client_version = [0, 3, 0] - self._skip_tts: bool = False # Keep in sync with llm_service.py + self._llm_skip_tts: bool = False # Keep in sync with llm_service.py's configuration. self._registered_actions: Dict[str, RTVIAction] = {} self._registered_services: Dict[str, RTVIService] = {} @@ -1441,7 +1547,7 @@ class RTVIProcessor(FrameProcessor): elif isinstance(frame, RTVIActionFrame): await self._action_queue.put(frame) elif isinstance(frame, LLMConfigureOutputFrame): - self._skip_tts = frame.skip_tts + self._llm_skip_tts = frame.skip_tts await self.push_frame(frame, direction) # Other frames else: @@ -1697,9 +1803,9 @@ class RTVIProcessor(FrameProcessor): opts = data.options if data.options is not None else RTVISendTextOptions() if opts.run_immediately: await self.interrupt_bot() - cur_skip_tts = self._skip_tts + cur_llm_skip_tts = self._llm_skip_tts should_skip_tts = not opts.audio_response - toggle_skip_tts = cur_skip_tts != should_skip_tts + toggle_skip_tts = cur_llm_skip_tts != should_skip_tts if toggle_skip_tts: output_frame = LLMConfigureOutputFrame(skip_tts=should_skip_tts) await self.push_frame(output_frame) @@ -1709,7 +1815,7 @@ class RTVIProcessor(FrameProcessor): ) await self.push_frame(text_frame) if toggle_skip_tts: - output_frame = LLMConfigureOutputFrame(skip_tts=cur_skip_tts) + output_frame = LLMConfigureOutputFrame(skip_tts=cur_llm_skip_tts) await self.push_frame(output_frame) async def _handle_update_context(self, data: RTVIAppendToContextData):