add TransportMessageSystemFrame

CHANGELOG.md

@@ -14,6 +14,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### 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

@@ -31,6 +31,7 @@ from pipecat.frames.frames import (
     TextFrame,
     TranscriptionFrame,
     TransportMessageFrame,
+    TransportMessageUrgentFrame,
     UserStartedSpeakingFrame,
     FunctionCallResultFrame,
     UserStoppedSpeakingFrame,
@@ -349,9 +350,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)
 
 
@@ -377,7 +380,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
@@ -387,7 +390,7 @@ class RTVISpeakingProcessor(RTVIFrameProcessor):
             message = RTVIBotStoppedSpeakingMessage()
 
         if message:
-            await self._push_transport_message(message)
+            await self._push_transport_message_urgent(message)
 
 
 class RTVIUserTranscriptionProcessor(RTVIFrameProcessor):
@@ -418,7 +421,7 @@ class RTVIUserTranscriptionProcessor(RTVIFrameProcessor):
             )
 
         if message:
-            await self._push_transport_message(message)
+            await self._push_transport_message_urgent(message)
 
 
 class RTVIUserLLMTextProcessor(RTVIFrameProcessor):
@@ -439,7 +442,7 @@ class RTVIUserLLMTextProcessor(RTVIFrameProcessor):
             message = messages[-1]
             if message["role"] == "user":
                 message = RTVIUserLLMTextMessage(data=RTVITextMessageData(text=message["content"]))
-                await self._push_transport_message(message)
+                await self._push_transport_message_urgent(message)
 
 
 class RTVIBotLLMProcessor(RTVIFrameProcessor):
@@ -452,9 +455,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):
@@ -467,9 +470,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):
@@ -486,7 +489,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):

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()