Merge pull request #583 from pipecat-ai/aleix/add-pts-to-llm-full-response-end-frame

add pts to llm full response end frame

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.
 

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,64 +500,11 @@ class RTVIBotTTSProcessor(RTVIFrameProcessor):
             await self._push_transport_message_urgent(RTVIBotTTSStartedMessage())
         elif isinstance(frame, TTSStoppedFrame):
             await self._push_transport_message_urgent(RTVIBotTTSStoppedMessage())
-
+        elif type(frame) is TextFrame:
-
-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)
-
-
 class RTVIProcessor(FrameProcessor):
     def __init__(
         self,

src/pipecat/services/ai_services.py

@@ -427,14 +427,20 @@ 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
+                last_pts = frame.pts
                 await self.push_frame(frame)
                 self._words_queue.task_done()
             except asyncio.CancelledError:

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"]))

src/pipecat/services/openai_realtime_beta/llm_and_context.py

@@ -364,7 +364,6 @@ class OpenAILLMServiceRealtimeBeta(LLMService):
                 )
             )
 
-
     #
     # frame processing
     #

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,9 +308,7 @@ class BaseOutputTransport(FrameProcessor):
                 # time to process it.
                 if running:
                     current_time = self.get_clock().get_time()
-                    if timestamp <= current_time:
+                    if timestamp > current_time:
-                        await self._sink_clock_frame_handler(frame)
-                    else:
                         wait_time = nanoseconds_to_seconds(timestamp - current_time)
                         await asyncio.sleep(wait_time)
                     await self._sink_frame_handler(frame)