Merge pull request #552 from pipecat-ai/aleix/rtvi-user-llm-text

rtvi: add RTVIUserLLMTextProcessor

CHANGELOG.md

@@ -5,10 +5,19 @@ All notable changes to **Pipecat** will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [0.0.43] - 2024-10-03
+## [Unreleased]
+
+### Added
+
+- Added new `RTVIUserLLMTextProcessor`. This processor will send an RTVI
+  `user-llm-text` message with the user content's that was sent to the LLM.
 
 ### Changed
 
+- `TransportMessageFrame` doesn't have an `urgent` field anymore, instead
+  there's now a `TransportMessageUrgentFrame` which is a `SystemFrame` and
+  therefore skip all internal queuing.
+
 - For TTS services, convert inputted languages to match each service's language
   format
 

src/pipecat/frames/frames.py

@@ -269,7 +269,6 @@ class TTSSpeakFrame(DataFrame):
 @dataclass
 class TransportMessageFrame(DataFrame):
     message: Any
-    urgent: bool = False
 
     def __str__(self):
         return f"{self.name}(message: {self.message})"
@@ -405,6 +404,14 @@ class BotInterruptionFrame(SystemFrame):
     pass
 
 
+@dataclass
+class TransportMessageUrgentFrame(SystemFrame):
+    message: Any
+
+    def __str__(self):
+        return f"{self.name}(message: {self.message})"
+
+
 @dataclass
 class MetricsFrame(SystemFrame):
     """Emitted by processor that can compute metrics like latencies."""

src/pipecat/processors/frameworks/rtvi.py

@@ -6,10 +6,11 @@
 
 import asyncio
 import base64
-
-from typing import Any, Awaitable, Callable, Dict, List, Literal, Optional, Union
-from pydantic import BaseModel, Field, PrivateAttr, ValidationError
 from dataclasses import dataclass
+from typing import Any, Awaitable, Callable, Dict, List, Literal, Optional, Union
+
+from loguru import logger
+from pydantic import BaseModel, Field, PrivateAttr, ValidationError
 
 from pipecat.frames.frames import (
     BotInterruptionFrame,
@@ -20,27 +21,28 @@ from pipecat.frames.frames import (
     EndFrame,
     ErrorFrame,
     Frame,
+    FunctionCallResultFrame,
     InterimTranscriptionFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     OutputAudioRawFrame,
     StartFrame,
     SystemFrame,
-    TTSStartedFrame,
-    TTSStoppedFrame,
     TextFrame,
     TranscriptionFrame,
     TransportMessageFrame,
+    TransportMessageUrgentFrame,
+    TTSStartedFrame,
+    TTSStoppedFrame,
     UserStartedSpeakingFrame,
-    FunctionCallResultFrame,
     UserStoppedSpeakingFrame,
 )
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.openai_llm_context import (
+    OpenAILLMContext,
+    OpenAILLMContextFrame,
+)
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 
-from loguru import logger
-
-
 RTVI_PROTOCOL_VERSION = "0.2"
 
 ActionResult = Union[bool, int, float, str, list, dict]
@@ -291,22 +293,12 @@ class RTVIAudioMessageData(BaseModel):
     num_channels: int
 
 
-class RTVIBotAudioMessage(BaseModel):
+class RTVIBotTTSAudioMessage(BaseModel):
     label: Literal["rtvi-ai"] = "rtvi-ai"
-    type: Literal["bot-audio"] = "bot-audio"
+    type: Literal["bot-tts-audio"] = "bot-tts-audio"
     data: RTVIAudioMessageData
 
 
-class RTVIBotTranscriptionMessageData(BaseModel):
-    text: str
-
-
-class RTVIBotTranscriptionMessage(BaseModel):
-    label: Literal["rtvi-ai"] = "rtvi-ai"
-    type: Literal["bot-transcription"] = "bot-transcription"
-    data: RTVIBotTranscriptionMessageData
-
-
 class RTVIUserTranscriptionMessageData(BaseModel):
     text: str
     user_id: str
@@ -320,6 +312,12 @@ class RTVIUserTranscriptionMessage(BaseModel):
     data: RTVIUserTranscriptionMessageData
 
 
+class RTVIUserLLMTextMessage(BaseModel):
+    label: Literal["rtvi-ai"] = "rtvi-ai"
+    type: Literal["user-llm-text"] = "user-llm-text"
+    data: RTVITextMessageData
+
+
 class RTVIUserStartedSpeakingMessage(BaseModel):
     label: Literal["rtvi-ai"] = "rtvi-ai"
     type: Literal["user-started-speaking"] = "user-started-speaking"
@@ -350,9 +348,11 @@ class RTVIFrameProcessor(FrameProcessor):
         self._direction = direction
 
     async def _push_transport_message(self, model: BaseModel, exclude_none: bool = True):
-        frame = TransportMessageFrame(
+        frame = TransportMessageFrame(message=model.model_dump(exclude_none=exclude_none))
-            message=model.model_dump(exclude_none=exclude_none), urgent=True
+        await self.push_frame(frame, self._direction)
-        )
+
+    async def _push_transport_message_urgent(self, model: BaseModel, exclude_none: bool = True):
+        frame = TransportMessageUrgentFrame(message=model.model_dump(exclude_none=exclude_none))
         await self.push_frame(frame, self._direction)
 
 
@@ -378,7 +378,7 @@ class RTVISpeakingProcessor(RTVIFrameProcessor):
             message = RTVIUserStoppedSpeakingMessage()
 
         if message:
-            await self._push_transport_message(message)
+            await self._push_transport_message_urgent(message)
 
     async def _handle_bot_speaking(self, frame: Frame):
         message = None
@@ -388,7 +388,7 @@ class RTVISpeakingProcessor(RTVIFrameProcessor):
             message = RTVIBotStoppedSpeakingMessage()
 
         if message:
-            await self._push_transport_message(message)
+            await self._push_transport_message_urgent(message)
 
 
 class RTVIUserTranscriptionProcessor(RTVIFrameProcessor):
@@ -419,7 +419,36 @@ class RTVIUserTranscriptionProcessor(RTVIFrameProcessor):
             )
 
         if message:
-            await self._push_transport_message(message)
+            await self._push_transport_message_urgent(message)
+
+
+class RTVIUserLLMTextProcessor(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, OpenAILLMContextFrame):
+            await self._handle_context(frame)
+
+    async def _handle_context(self, frame: OpenAILLMContextFrame):
+        messages = frame.context.messages
+        if len(messages) > 0:
+            message = messages[-1]
+            if message["role"] == "user":
+                content = message["content"]
+                if isinstance(content, list):
+                    print("LIST")
+                    text = " ".join(item["text"] for item in content if "text" in item)
+                else:
+                    print("STRING")
+                    text = content
+
+                rtvi_message = RTVIUserLLMTextMessage(data=RTVITextMessageData(text=text))
+                await self._push_transport_message_urgent(rtvi_message)
 
 
 class RTVIBotLLMProcessor(RTVIFrameProcessor):
@@ -432,9 +461,9 @@ class RTVIBotLLMProcessor(RTVIFrameProcessor):
         await self.push_frame(frame, direction)
 
         if isinstance(frame, LLMFullResponseStartFrame):
-            await self._push_transport_message(RTVIBotLLMStartedMessage())
+            await self._push_transport_message_urgent(RTVIBotLLMStartedMessage())
         elif isinstance(frame, LLMFullResponseEndFrame):
-            await self._push_transport_message(RTVIBotLLMStoppedMessage())
+            await self._push_transport_message_urgent(RTVIBotLLMStoppedMessage())
 
 
 class RTVIBotTTSProcessor(RTVIFrameProcessor):
@@ -447,9 +476,9 @@ class RTVIBotTTSProcessor(RTVIFrameProcessor):
         await self.push_frame(frame, direction)
 
         if isinstance(frame, TTSStartedFrame):
-            await self._push_transport_message(RTVIBotTTSStartedMessage())
+            await self._push_transport_message_urgent(RTVIBotTTSStartedMessage())
         elif isinstance(frame, TTSStoppedFrame):
-            await self._push_transport_message(RTVIBotTTSStoppedMessage())
+            await self._push_transport_message_urgent(RTVIBotTTSStoppedMessage())
 
 
 class RTVIBotLLMTextProcessor(RTVIFrameProcessor):
@@ -466,7 +495,7 @@ class RTVIBotLLMTextProcessor(RTVIFrameProcessor):
 
     async def _handle_text(self, frame: TextFrame):
         message = RTVIBotLLMTextMessage(data=RTVITextMessageData(text=frame.text))
-        await self._push_transport_message(message)
+        await self._push_transport_message_urgent(message)
 
 
 class RTVIBotTTSTextProcessor(RTVIFrameProcessor):
@@ -486,7 +515,7 @@ class RTVIBotTTSTextProcessor(RTVIFrameProcessor):
         await self._push_transport_message(message)
 
 
-class RTVIBotAudioProcessor(RTVIFrameProcessor):
+class RTVIBotTTSAudioProcessor(RTVIFrameProcessor):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
 
@@ -500,7 +529,7 @@ class RTVIBotAudioProcessor(RTVIFrameProcessor):
 
     async def _handle_audio(self, frame: OutputAudioRawFrame):
         encoded = base64.b64encode(frame.audio).decode("utf-8")
-        message = RTVIBotAudioMessage(
+        message = RTVIBotTTSAudioMessage(
             data=RTVIAudioMessageData(
                 audio=encoded, sample_rate=frame.sample_rate, num_channels=frame.num_channels
             )
@@ -647,9 +676,7 @@ class RTVIProcessor(FrameProcessor):
             self._message_task = None
 
     async def _push_transport_message(self, model: BaseModel, exclude_none: bool = True):
-        frame = TransportMessageFrame(
+        frame = TransportMessageUrgentFrame(message=model.model_dump(exclude_none=exclude_none))
-            message=model.model_dump(exclude_none=exclude_none), urgent=True
-        )
         await self.push_frame(frame)
 
     async def _action_task_handler(self):

src/pipecat/transports/base_output.py

@@ -33,6 +33,7 @@ from pipecat.frames.frames import (
     TTSStoppedFrame,
     TextFrame,
     TransportMessageFrame,
+    TransportMessageUrgentFrame,
 )
 from pipecat.transports.base_transport import TransportParams
 
@@ -148,7 +149,7 @@ class BaseOutputTransport(FrameProcessor):
             await self._audio_out_task
             self._audio_out_task = None
 
-    async def send_message(self, frame: TransportMessageFrame):
+    async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame):
         pass
 
     async def send_metrics(self, frame: MetricsFrame):
@@ -186,6 +187,8 @@ class BaseOutputTransport(FrameProcessor):
         elif isinstance(frame, MetricsFrame):
             await self.push_frame(frame, direction)
             await self.send_metrics(frame)
+        elif isinstance(frame, TransportMessageUrgentFrame):
+            await self.send_message(frame)
         elif isinstance(frame, SystemFrame):
             await self.push_frame(frame, direction)
         # Control frames.
@@ -198,8 +201,6 @@ class BaseOutputTransport(FrameProcessor):
             await self._handle_audio(frame)
         elif isinstance(frame, (OutputImageRawFrame, SpriteFrame)):
             await self._handle_image(frame)
-        elif isinstance(frame, TransportMessageFrame) and frame.urgent:
-            await self.send_message(frame)
         # TODO(aleix): Images and audio should support presentation timestamps.
         elif frame.pts:
             await self._sink_clock_queue.put((frame.pts, frame.id, frame))

src/pipecat/transports/services/daily.py

@@ -35,6 +35,7 @@ from pipecat.frames.frames import (
     StartFrame,
     TranscriptionFrame,
     TransportMessageFrame,
+    TransportMessageUrgentFrame,
     UserImageRawFrame,
     UserImageRequestFrame,
 )
@@ -70,6 +71,11 @@ class DailyTransportMessageFrame(TransportMessageFrame):
     participant_id: str | None = None
 
 
+@dataclass
+class DailyTransportMessageUrgentFrame(TransportMessageUrgentFrame):
+    participant_id: str | None = None
+
+
 class WebRTCVADAnalyzer(VADAnalyzer):
     def __init__(self, *, sample_rate=16000, num_channels=1, params: VADParams = VADParams()):
         super().__init__(sample_rate=sample_rate, num_channels=num_channels, params=params)
@@ -234,12 +240,12 @@ class DailyTransportClient(EventHandler):
     def set_callbacks(self, callbacks: DailyCallbacks):
         self._callbacks = callbacks
 
-    async def send_message(self, frame: TransportMessageFrame):
+    async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame):
         if not self._client:
             return
 
         participant_id = None
-        if isinstance(frame, DailyTransportMessageFrame):
+        if isinstance(frame, (DailyTransportMessageFrame, DailyTransportMessageUrgentFrame)):
             participant_id = frame.participant_id
 
         future = self._loop.create_future()
@@ -736,7 +742,7 @@ class DailyOutputTransport(BaseOutputTransport):
         await super().cleanup()
         await self._client.cleanup()
 
-    async def send_message(self, frame: TransportMessageFrame):
+    async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame):
         await self._client.send_message(frame)
 
     async def send_metrics(self, frame: MetricsFrame):

src/pipecat/transports/services/livekit.py

@@ -22,6 +22,7 @@ from pipecat.frames.frames import (
     MetricsFrame,
     StartFrame,
     TransportMessageFrame,
+    TransportMessageUrgentFrame,
 )
 from pipecat.metrics.metrics import (
     LLMUsageMetricsData,
@@ -51,6 +52,11 @@ class LiveKitTransportMessageFrame(TransportMessageFrame):
     participant_id: str | None = None
 
 
+@dataclass
+class LiveKitTransportMessageUrgentFrame(TransportMessageUrgentFrame):
+    participant_id: str | None = None
+
+
 class LiveKitParams(TransportParams):
     audio_out_sample_rate: int = 48000
     audio_out_channels: int = 1
@@ -420,8 +426,8 @@ class LiveKitOutputTransport(BaseOutputTransport):
         await super().cancel(frame)
         await self._client.disconnect()
 
-    async def send_message(self, frame: TransportMessageFrame):
+    async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame):
-        if isinstance(frame, LiveKitTransportMessageFrame):
+        if isinstance(frame, (LiveKitTransportMessageFrame, LiveKitTransportMessageUrgentFrame)):
             await self._client.send_data(frame.message.encode(), frame.participant_id)
         else:
             await self._client.send_data(frame.message.encode())
@@ -596,6 +602,13 @@ class LiveKitTransport(BaseTransport):
             frame = LiveKitTransportMessageFrame(message=message, participant_id=participant_id)
             await self._output.send_message(frame)
 
+    async def send_message_urgent(self, message: str, participant_id: str | None = None):
+        if self._output:
+            frame = LiveKitTransportMessageUrgentFrame(
+                message=message, participant_id=participant_id
+            )
+            await self._output.send_message(frame)
+
     async def cleanup(self):
         if self._input:
             await self._input.cleanup()