From 26c68ccd7c3ddcf9ffcea2b576c1a4f273c03426 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 26 Feb 2025 18:59:06 -0500 Subject: [PATCH 1/2] Add a new generic server to client message and frame type --- CHANGELOG.md | 10 ++++++++++ src/pipecat/frames/frames.py | 16 ++++++++++++++-- src/pipecat/processors/frameworks/rtvi.py | 18 ++++++++++++++++++ 3 files changed, 42 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ff7b62e9b..9b6de3430 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,16 @@ 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). +## [Unreleased] + +### Added + +- Added a new frame, `ServerMessageFrame`, and RTVI message `RTVIServerMessage` + which provides a generic mechanism for sending custom messages from server to + client. The `ServerMessageFrame` can be processed by either a `RTVIProcessor` + or `RTVIObserver` and will be delivered to the client's `onServerMessage` + callback. + ## [0.0.58] - 2025-02-26 ### Added diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index b43aaaeae..b771b765f 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -568,7 +568,8 @@ class UserStoppedSpeakingFrame(SystemFrame): @dataclass class EmulateUserStartedSpeakingFrame(SystemFrame): """Emitted by internal processors upstream to emulate VAD behavior when a - user starts speaking.""" + user starts speaking. + """ pass @@ -576,7 +577,8 @@ class EmulateUserStartedSpeakingFrame(SystemFrame): @dataclass class EmulateUserStoppedSpeakingFrame(SystemFrame): """Emitted by internal processors upstream to emulate VAD behavior when a - user stops speaking.""" + user stops speaking. + """ pass @@ -704,6 +706,16 @@ class VisionImageRawFrame(InputImageRawFrame): return f"{self.name}(pts: {pts}, text: [{self.text}], size: {self.size}, format: {self.format})" +@dataclass +class ServerMessageFrame(SystemFrame): + """A frame for sending server messages to the client.""" + + data: Any + + def __str__(self): + return f"{self.name}(data: {self.data})" + + # # Control frames # diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index af55fbd42..4edef4203 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -38,6 +38,7 @@ from pipecat.frames.frames import ( LLMFullResponseStartFrame, LLMTextFrame, MetricsFrame, + ServerMessageFrame, StartFrame, SystemFrame, TranscriptionFrame, @@ -375,6 +376,12 @@ class RTVIMetricsMessage(BaseModel): data: Mapping[str, Any] +class RTVIServerMessage(BaseModel): + label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL + type: Literal["server-message"] = "server-message" + data: Any + + class RTVIFrameProcessor(FrameProcessor): def __init__(self, direction: FrameDirection = FrameDirection.DOWNSTREAM, **kwargs): super().__init__(**kwargs) @@ -710,6 +717,9 @@ class RTVIObserver(BaseObserver): mark_as_seen = False elif isinstance(frame, MetricsFrame): await self._handle_metrics(frame) + elif isinstance(frame, ServerMessageFrame): + message = RTVIServerMessage(data=frame.data) + await self.push_transport_message_urgent(message) if mark_as_seen: self._frames_seen.add(frame.id) @@ -895,6 +905,11 @@ class RTVIProcessor(FrameProcessor): async def handle_message(self, message: RTVIMessage): await self._message_queue.put(message) + async def _handle_server_message(self, frame: ServerMessageFrame): + """Handle server message frame by converting it to a transport message.""" + message = RTVIServerMessage(data=frame.data) + await self._push_transport_message(message) + async def handle_function_call( self, function_name: str, @@ -934,6 +949,9 @@ class RTVIProcessor(FrameProcessor): await self.push_frame(frame, direction) elif isinstance(frame, TransportMessageUrgentFrame): await self._handle_transport_message(frame) + elif isinstance(frame, ServerMessageFrame): + await self._handle_server_message(frame) + await self.push_frame(frame, direction) # All other system frames elif isinstance(frame, SystemFrame): await self.push_frame(frame, direction) From 53d403af4bce22a4e8e77c40a7eac5dfd6319c3e Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 27 Feb 2025 12:50:43 -0500 Subject: [PATCH 2/2] Remove the RTVIServerMessage logic from the RTVIProcessor --- CHANGELOG.md | 6 +++--- src/pipecat/processors/frameworks/rtvi.py | 8 -------- 2 files changed, 3 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9b6de3430..cdf100d41 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,9 +11,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Added a new frame, `ServerMessageFrame`, and RTVI message `RTVIServerMessage` which provides a generic mechanism for sending custom messages from server to - client. The `ServerMessageFrame` can be processed by either a `RTVIProcessor` - or `RTVIObserver` and will be delivered to the client's `onServerMessage` - callback. + client. The `ServerMessageFrame` is processed by the `RTVIObserver` and will + be delivered to the client's `onServerMessage` callback or `ServerMessage` + event. ## [0.0.58] - 2025-02-26 diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 4edef4203..9835a9648 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -905,11 +905,6 @@ class RTVIProcessor(FrameProcessor): async def handle_message(self, message: RTVIMessage): await self._message_queue.put(message) - async def _handle_server_message(self, frame: ServerMessageFrame): - """Handle server message frame by converting it to a transport message.""" - message = RTVIServerMessage(data=frame.data) - await self._push_transport_message(message) - async def handle_function_call( self, function_name: str, @@ -949,9 +944,6 @@ class RTVIProcessor(FrameProcessor): await self.push_frame(frame, direction) elif isinstance(frame, TransportMessageUrgentFrame): await self._handle_transport_message(frame) - elif isinstance(frame, ServerMessageFrame): - await self._handle_server_message(frame) - await self.push_frame(frame, direction) # All other system frames elif isinstance(frame, SystemFrame): await self.push_frame(frame, direction)