diff --git a/CHANGELOG.md b/CHANGELOG.md index 4a98700df..3e21b4655 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Merge `RTVIBotLLMProcessor`/`RTVIBotLLMTextProcessor` and + `RTVIBotTTSProcessor`/`RTVIBotTTSTextProcessor` to avoid out of order issues. + - Fixed an issue in Daily transport that would cause tasks to be hanging if urgent transport messages were being sent from a transport event handler. diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 1616b6790..bb9cd17a7 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -5,7 +5,6 @@ # import asyncio -import base64 from dataclasses import dataclass from typing import Any, Awaitable, Callable, Dict, List, Literal, Optional, Union @@ -25,7 +24,6 @@ from pipecat.frames.frames import ( InterimTranscriptionFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, - OutputAudioRawFrame, StartFrame, SystemFrame, TextFrame, @@ -484,6 +482,9 @@ class RTVIBotLLMProcessor(RTVIFrameProcessor): await self._push_transport_message_urgent(RTVIBotLLMStartedMessage()) elif isinstance(frame, LLMFullResponseEndFrame): await self._push_transport_message_urgent(RTVIBotLLMStoppedMessage()) + elif type(frame) is TextFrame: + message = RTVIBotLLMTextMessage(data=RTVITextMessageData(text=frame.text)) + await self._push_transport_message_urgent(message) class RTVIBotTTSProcessor(RTVIFrameProcessor): @@ -499,62 +500,9 @@ class RTVIBotTTSProcessor(RTVIFrameProcessor): await self._push_transport_message_urgent(RTVIBotTTSStartedMessage()) elif isinstance(frame, TTSStoppedFrame): await self._push_transport_message_urgent(RTVIBotTTSStoppedMessage()) - - -class RTVIBotLLMTextProcessor(RTVIFrameProcessor): - def __init__(self, **kwargs): - super().__init__(**kwargs) - - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - await self.push_frame(frame, direction) - - if type(frame) is TextFrame: - await self._handle_text(frame) - - async def _handle_text(self, frame: TextFrame): - message = RTVIBotLLMTextMessage(data=RTVITextMessageData(text=frame.text)) - await self._push_transport_message_urgent(message) - - -class RTVIBotTTSTextProcessor(RTVIFrameProcessor): - def __init__(self, **kwargs): - super().__init__(**kwargs) - - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - await self.push_frame(frame, direction) - - if type(frame) is TextFrame: - await self._handle_text(frame) - - async def _handle_text(self, frame: TextFrame): - message = RTVIBotTTSTextMessage(data=RTVITextMessageData(text=frame.text)) - await self._push_transport_message_urgent(message) - - -class RTVIBotTTSAudioProcessor(RTVIFrameProcessor): - def __init__(self, **kwargs): - super().__init__(**kwargs) - - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - await self.push_frame(frame, direction) - - if isinstance(frame, OutputAudioRawFrame): - await self._handle_audio(frame) - - async def _handle_audio(self, frame: OutputAudioRawFrame): - encoded = base64.b64encode(frame.audio).decode("utf-8") - message = RTVIBotTTSAudioMessage( - data=RTVIAudioMessageData( - audio=encoded, sample_rate=frame.sample_rate, num_channels=frame.num_channels - ) - ) - await self._push_transport_message_urgent(message) + elif type(frame) is TextFrame: + message = RTVIBotTTSTextMessage(data=RTVITextMessageData(text=frame.text)) + await self._push_transport_message_urgent(message) class RTVIProcessor(FrameProcessor): diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 7ca043e97..15b9cb9b7 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -427,15 +427,21 @@ class WordTTSService(TTSService): self._words_task = None async def _words_task_handler(self): + last_pts = 0 while True: try: (word, timestamp) = await self._words_queue.get() if word == "LLMFullResponseEndFrame" and timestamp == 0: - await self.push_frame(LLMFullResponseEndFrame()) + frame = LLMFullResponseEndFrame() + frame.pts = last_pts + elif word == "TTSStoppedFrame" and timestamp == 0: + frame = TTSStoppedFrame() + frame.pts = last_pts else: frame = TextFrame(word) frame.pts = self._initial_word_timestamp + timestamp - await self.push_frame(frame) + last_pts = frame.pts + await self.push_frame(frame) self._words_queue.task_done() except asyncio.CancelledError: break diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index c0e4a89a2..f5657126c 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -210,7 +210,6 @@ class CartesiaTTSService(WordTTSService): async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection): await super()._handle_interruption(frame, direction) await self.stop_all_metrics() - await self.push_frame(LLMFullResponseEndFrame()) self._context_id = None async def flush_audio(self): @@ -228,12 +227,13 @@ class CartesiaTTSService(WordTTSService): continue if msg["type"] == "done": await self.stop_ttfb_metrics() - await self.push_frame(TTSStoppedFrame()) # Unset _context_id but not the _context_id_start_timestamp # because we are likely still playing out audio and need the # timestamp to set send context frames. self._context_id = None - await self.add_word_timestamps([("LLMFullResponseEndFrame", 0)]) + await self.add_word_timestamps( + [("TTSStoppedFrame", 0), ("LLMFullResponseEndFrame", 0)] + ) elif msg["type"] == "timestamps": await self.add_word_timestamps( list(zip(msg["word_timestamps"]["words"], msg["word_timestamps"]["start"])) diff --git a/src/pipecat/services/openai_realtime_beta/llm_and_context.py b/src/pipecat/services/openai_realtime_beta/llm_and_context.py index 173ee5103..68f1be177 100644 --- a/src/pipecat/services/openai_realtime_beta/llm_and_context.py +++ b/src/pipecat/services/openai_realtime_beta/llm_and_context.py @@ -364,7 +364,6 @@ class OpenAILLMServiceRealtimeBeta(LLMService): ) ) - # # frame processing # @@ -643,8 +642,8 @@ class OpenAILLMServiceRealtimeBeta(LLMService): await self._truncate_current_audio_response() # todo: might need to guard sending these when we fully support using either openai # turn detection of Pipecat turn detection - await self._start_interruption() # cancels this processor task - await self.push_frame(StartInterruptionFrame()) # cancels downstream tasks + await self._start_interruption() # cancels this processor task + await self.push_frame(StartInterruptionFrame()) # cancels downstream tasks await self.push_frame(UserStartedSpeakingFrame()) async def _handle_evt_speech_stopped(self, evt): diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 7ec503d24..9bd508f1d 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -1,6 +1,5 @@ # # Copyright (c) 2024, Daily - # # SPDX-License-Identifier: BSD 2-Clause License # @@ -31,7 +30,6 @@ from pipecat.frames.frames import ( SystemFrame, TTSStartedFrame, TTSStoppedFrame, - TextFrame, TransportMessageFrame, TransportMessageUrgentFrame, ) @@ -296,12 +294,6 @@ class BaseOutputTransport(FrameProcessor): except Exception as e: logger.exception(f"{self} error processing sink queue: {e}") - async def _sink_clock_frame_handler(self, frame: Frame): - # TODO(aleix): For now we just process TextFrame. But we should process - # audio and video as well. - if isinstance(frame, TextFrame): - await self.push_frame(frame) - async def _sink_clock_task_handler(self): running = True while running: @@ -316,12 +308,10 @@ class BaseOutputTransport(FrameProcessor): # time to process it. if running: current_time = self.get_clock().get_time() - if timestamp <= current_time: - await self._sink_clock_frame_handler(frame) - else: + if timestamp > current_time: wait_time = nanoseconds_to_seconds(timestamp - current_time) await asyncio.sleep(wait_time) - await self._sink_frame_handler(frame) + await self._sink_frame_handler(frame) self._sink_clock_queue.task_done() except asyncio.CancelledError: