diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 4af13003d..858e1d973 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -356,11 +356,15 @@ class LLMTextFrame(TextFrame): @dataclass -class TTSTextFrame(TextFrame): +class AggregatedLLMTextFrame(TextFrame): """Text frame generated by Text-to-Speech services.""" aggregated_by: Literal["sentence", "word"] | str - spoken: Optional[bool] = True # Whether this text has been spoken by TTS + + +@dataclass +class TTSTextFrame(AggregatedLLMTextFrame): + """Text frame generated by Text-to-Speech services.""" pass diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 18418081f..8b126b5b6 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -32,6 +32,7 @@ from pydantic import BaseModel, Field, PrivateAttr, ValidationError from pipecat.audio.utils import calculate_audio_volume from pipecat.frames.frames import ( + AggregatedLLMTextFrame, BotStartedSpeakingFrame, BotStoppedSpeakingFrame, CancelFrame, @@ -1023,6 +1024,16 @@ class RTVIObserver(BaseObserver): if self._rtvi: await self._rtvi.push_transport_message(model, exclude_none) + async def send_aggregated_llm_text(self, text: str, aggregated_by: Optional[str] = None): + """Send aggregated LLM text as a bot output message. + + Args: + text: The aggregated text to send. + aggregated_by: The method of aggregation (e.g., "word", "sentence"). + """ + if self._rtvi: + await self._rtvi.push_aggregated_llm_text(text, aggregated_by) + async def on_push_frame(self, data: FramePushed): """Process a frame being pushed through the pipeline. @@ -1073,11 +1084,13 @@ 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): - await self._handle_tts_text_frame(frame) - else: + elif isinstance(frame, AggregatedLLMTextFrame): + 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): @@ -1139,19 +1152,20 @@ class RTVIObserver(BaseObserver): if message: await self.send_rtvi_message(message) - async def _handle_tts_text_frame(self, frame: TTSTextFrame): - """Handle TTS text output frames.""" - # send the tts-text message - message = RTVIBotTTSTextMessage(data=RTVITextMessageData(text=frame.text)) - await self.send_rtvi_message(message) - # send the bot-output message + async def _handle_aggregated_llm_text(self, frame: AggregatedLLMTextFrame): + """Handle aggregated LLM text output frames.""" + isTTS = isinstance(frame, TTSTextFrame) message = RTVIBotOutputMessage( data=RTVIBotOutputMessageData( - text=frame.text, spoken=frame.spoken, aggregated_by=frame.aggregated_by + text=frame.text, spoken=isTTS, aggregated_by=frame.aggregated_by ) ) await self.send_rtvi_message(message) + if isTTS and self._params.bot_tts_enabled: + tts_message = RTVIBotTTSTextMessage(data=RTVITextMessageData(text=frame.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)) @@ -1161,41 +1175,26 @@ class RTVIObserver(BaseObserver): if self._skip_tts is None: self._skip_tts = frame.skip_tts - messages = [] - should_reset_transcription = False + orig_text = self._bot_transcription self._bot_transcription += frame.text - if not frame.skip_tts and self._skip_tts: - # We just switched from skipping TTS to not skipping TTS. - # Send and reset any existing transcription. - if len(self._bot_transcription) > 0: - message.append( - RTVIBotOutputMessage( - data=RTVIBotOutputMessageData( - text=self._bot_transcription, spoken=False, aggregated_by="sentence" - ) - ) - ) - should_reset_transcription = True - if match_endofsentence(self._bot_transcription) and len(self._bot_transcription) > 0: - messages.append( + # TODO: Remove this message when we fully deprecate bot-transcription messages. + await self.send_rtvi_message( RTVIBotTranscriptionMessage(data=RTVITextMessageData(text=self._bot_transcription)) ) if frame.skip_tts: - messages.append( - RTVIBotOutputMessage( - data=RTVIBotOutputMessageData( - text=self._bot_transcription, spoken=False, aggregated_by="sentence" - ) - ) + await self.send_aggregated_llm_text( + text=self._bot_transcription, aggregated_by="sentence" ) - should_reset_transcription = True - - for msg in messages: - await self.send_rtvi_message(msg) - if should_reset_transcription: self._bot_transcription = "" + elif not frame.skip_tts and self._skip_tts: + # We just switched from skipping TTS to not skipping TTS. + # Send any dangling transcription. + if len(orig_text) > 0: + await self.send_aggregated_llm_text(text=orig_text, aggregated_by="sentence") + self._bot_transcription = frame.text + self._skip_tts = frame.skip_tts async def _handle_user_transcriptions(self, frame: Frame): """Handle user transcription frames.""" @@ -1321,7 +1320,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] = {} @@ -1429,6 +1428,12 @@ class RTVIProcessor(FrameProcessor): ) await self.push_frame(frame) + async def push_aggregated_llm_text(self, text: str, aggregated_by: Optional[str] = None): + """Push an aggregated LLM text frame.""" + frame = AggregatedLLMTextFrame(text=text, aggregated_by=aggregated_by) + frame.skip_tts = True + await self.push_frame(frame) + async def handle_message(self, message: RTVIMessage): """Handle an incoming RTVI message. @@ -1514,7 +1519,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: @@ -1770,9 +1775,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) @@ -1782,7 +1787,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): diff --git a/src/pipecat/services/aws/nova_sonic/llm.py b/src/pipecat/services/aws/nova_sonic/llm.py index 1a50fc22f..ba537e49b 100644 --- a/src/pipecat/services/aws/nova_sonic/llm.py +++ b/src/pipecat/services/aws/nova_sonic/llm.py @@ -1027,7 +1027,7 @@ class AWSNovaSonicLLMService(LLMService): logger.debug(f"Assistant response text added: {text}") # Report the text of the assistant response. - frame = TTSTextFrame(text, aggregated_by="sentence", spoken=True) + frame = TTSTextFrame(text, aggregated_by="sentence") frame.includes_inter_frame_spaces = True await self.push_frame(frame) @@ -1062,9 +1062,7 @@ class AWSNovaSonicLLMService(LLMService): # TTSTextFrame would be ignored otherwise (the interruption frame # would have cleared the assistant aggregator state). await self.push_frame(LLMFullResponseStartFrame()) - frame = TTSTextFrame( - self._assistant_text_buffer, aggregated_by="sentence", spoken=True - ) + frame = TTSTextFrame(self._assistant_text_buffer, aggregated_by="sentence") frame.includes_inter_frame_spaces = True await self.push_frame(frame) self._may_need_repush_assistant_text = False diff --git a/src/pipecat/services/openai_realtime_beta/openai.py b/src/pipecat/services/openai_realtime_beta/openai.py index d67e58cf8..115019538 100644 --- a/src/pipecat/services/openai_realtime_beta/openai.py +++ b/src/pipecat/services/openai_realtime_beta/openai.py @@ -652,7 +652,7 @@ class OpenAIRealtimeBetaLLMService(LLMService): async def _handle_evt_audio_transcript_delta(self, evt): if evt.delta: await self.push_frame(LLMTextFrame(evt.delta)) - await self.push_frame(TTSTextFrame(evt.delta, aggregated_by="sentence", spoken=True)) + await self.push_frame(TTSTextFrame(evt.delta, aggregated_by="sentence")) async def _handle_evt_speech_started(self, evt): await self._truncate_current_audio_response() diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index fa7956c81..93276254a 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -23,6 +23,7 @@ from typing import ( from loguru import logger from pipecat.frames.frames import ( + AggregatedLLMTextFrame, BotStartedSpeakingFrame, BotStoppedSpeakingFrame, CancelFrame, @@ -516,15 +517,24 @@ class TTSService(AIService): text = await filter.filter(text) if text: - await self.push_frame(TTSTextFrame(text, spoken=True, aggregated_by=aggregated_by)) + if not self._push_text_frames: + # If we are not pushing text frames, we send a TTSTextFrame + # before the audio so downstream processors know what text + # is being spoken. Here, we assume this flag is used when the TTS + # provider supports word timestamps and the TTSTextFrames will be + # generated in the word_task_handler. + await self.push_frame(AggregatedLLMTextFrame(text, aggregated_by=aggregated_by)) await self.process_generator(self.run_tts(text)) await self.stop_processing_metrics() - if self._push_text_frames or not should_speak: - # We send the original text after the audio. This way, if we are + if not should_speak: + await self.push_frame(AggregatedLLMTextFrame(text, aggregated_by=aggregated_by)) + elif self._push_text_frames: + # In the case where the TTS service does not support word timestamps, + # we send the original text after the audio. This way, if we are # interrupted, the text is not added to the assistant context. - frame = TTSTextFrame(text, spoken=should_speak, aggregated_by=aggregated_by) + frame = TTSTextFrame(text, aggregated_by=aggregated_by) frame.includes_inter_frame_spaces = self.includes_inter_frame_spaces await self.push_frame(frame) @@ -652,7 +662,7 @@ class WordTTSService(TTSService): frame = TTSStoppedFrame() frame.pts = last_pts else: - frame = TTSTextFrame(word, spoken=True, aggregated_by="word") + frame = TTSTextFrame(word, aggregated_by="word") frame.pts = self._initial_word_timestamp + timestamp if frame: last_pts = frame.pts diff --git a/src/pipecat/tests/utils.py b/src/pipecat/tests/utils.py index 6ccce4b31..94b8cb1a4 100644 --- a/src/pipecat/tests/utils.py +++ b/src/pipecat/tests/utils.py @@ -203,8 +203,16 @@ async def run_test( if not isinstance(frame, EndFrame) or not send_end_frame: received_down_frames.append(frame) - print("received DOWN frames =", received_down_frames) - print("expected DOWN frames =", expected_down_frames) + down_frames_printed = "[" + for frame in received_down_frames: + down_frames_printed += f"{frame.__class__.__name__}, " + down_frames_printed += "]" + expected_frames_printed = "[" + for frame in expected_down_frames: + expected_frames_printed += f"{frame.__name__}, " + expected_frames_printed += "]" + print("received DOWN frames =", down_frames_printed) + print("expected DOWN frames =", expected_frames_printed) assert len(received_down_frames) == len(expected_down_frames) diff --git a/src/pipecat/utils/text/pattern_pair_aggregator.py b/src/pipecat/utils/text/pattern_pair_aggregator.py index c188c3974..d4712a823 100644 --- a/src/pipecat/utils/text/pattern_pair_aggregator.py +++ b/src/pipecat/utils/text/pattern_pair_aggregator.py @@ -111,7 +111,8 @@ class PatternPairAggregator(BaseTextAggregator): action: What to do when a complete pattern is matched: - "remove": Remove the matched pattern from the text. - "keep": Keep the matched pattern in the text and treat it as - normal text. + normal text. This allows you to register handlers for + the pattern without affecting the aggregation logic. - "aggregate": Return the matched pattern as a separate aggregation object. @@ -259,12 +260,14 @@ class PatternPairAggregator(BaseTextAggregator): # if len(patterns) > 0: + print(f"Found patterns: {[str(p) for p in patterns]}") if len(patterns) > 1: logger.warning( f"Multiple patterns matched: {[p.pattern_id for p in patterns]}. Only the first pattern will be returned." ) # If the pattern found is set to be aggregated, return it action = self._patterns[patterns[0].pattern_id].get("action", "remove") + print(f"Pattern action: {action}") if action == "aggregate": self._text = "" print(f"Returning pattern: {patterns[0]}") diff --git a/tests/test_piper_tts.py b/tests/test_piper_tts.py index 209e6e76c..97b550ed9 100644 --- a/tests/test_piper_tts.py +++ b/tests/test_piper_tts.py @@ -13,6 +13,7 @@ import pytest from aiohttp import web from pipecat.frames.frames import ( + AggregatedLLMTextFrame, ErrorFrame, TTSAudioRawFrame, TTSSpeakFrame, @@ -74,7 +75,6 @@ async def test_run_piper_tts_success(aiohttp_client): ] expected_returned_frames = [ - TTSTextFrame, TTSStartedFrame, TTSAudioRawFrame, TTSAudioRawFrame, @@ -122,7 +122,7 @@ async def test_run_piper_tts_error(aiohttp_client): TTSSpeakFrame(text="Error case."), ] - expected_down_frames = [TTSTextFrame, TTSStoppedFrame, TTSTextFrame] + expected_down_frames = [TTSStoppedFrame, TTSTextFrame] expected_up_frames = [ErrorFrame]