From 6d778e04915cc93851bcee69d334b4183ae9dd01 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 14 Oct 2024 10:40:41 -0700 Subject: [PATCH 1/6] services: add pts to LLMFullResponseEndFrame in WordTTSService --- src/pipecat/services/ai_services.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 7ca043e97..f1e01a18e 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -427,15 +427,18 @@ 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 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 From 51bc4839d186cfd9f81a7051443c4cb2270009fb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 14 Oct 2024 10:41:09 -0700 Subject: [PATCH 2/6] transport(base_output): simplify code --- src/pipecat/transports/base_output.py | 14 ++------------ 1 file changed, 2 insertions(+), 12 deletions(-) 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: From 164f06415cb6efb6fd591cb31814dbf13abfa60b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 14 Oct 2024 10:41:38 -0700 Subject: [PATCH 3/6] servcies(cartesia): no need to send LLMFullResponseEndFrame Interruptions are already handled by context aggregators. --- src/pipecat/services/cartesia.py | 1 - 1 file changed, 1 deletion(-) diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index c0e4a89a2..21074bbcc 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): From 616aa54f7569148c49f08807fbc8857e1f18533a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 15 Oct 2024 08:57:15 -0700 Subject: [PATCH 4/6] ruff formatting --- src/pipecat/services/openai_realtime_beta/llm_and_context.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) 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): From 3a3bf3fe34e5684b77e4e5eee441aa4294d518d6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 15 Oct 2024 09:02:51 -0700 Subject: [PATCH 5/6] services(cartesia): schedule TTSStoppedFrame after text --- src/pipecat/services/ai_services.py | 3 +++ src/pipecat/services/cartesia.py | 5 +++-- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index f1e01a18e..15b9cb9b7 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -434,6 +434,9 @@ class WordTTSService(TTSService): if word == "LLMFullResponseEndFrame" and timestamp == 0: 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 diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 21074bbcc..f5657126c 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -227,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"])) From 7bbaf4dfe93a12858ab0b1078640f88bf282c675 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 15 Oct 2024 10:24:43 -0700 Subject: [PATCH 6/6] rtvi: merge TTS/TTSText and LLM/LLMText processors --- CHANGELOG.md | 3 ++ src/pipecat/processors/frameworks/rtvi.py | 64 +++-------------------- 2 files changed, 9 insertions(+), 58 deletions(-) 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):