diff --git a/changelog/3774.added.md b/changelog/3774.added.md new file mode 100644 index 000000000..e72599e60 --- /dev/null +++ b/changelog/3774.added.md @@ -0,0 +1 @@ +- Added `broadcast_sibling_id` field to the base `Frame` class. This field is automatically set by `broadcast_frame()` and `broadcast_frame_instance()` to the ID of the paired frame pushed in the opposite direction, allowing receivers to identify broadcast pairs. diff --git a/changelog/3774.fixed.md b/changelog/3774.fixed.md new file mode 100644 index 000000000..a839f56ed --- /dev/null +++ b/changelog/3774.fixed.md @@ -0,0 +1 @@ +- Fixed `RTVIObserver` not processing upstream-only frames. Previously, all upstream frames were filtered out to avoid duplicate messages from broadcasted frames. Now only upstream copies of broadcasted frames are skipped. diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index c6c4421cd..2d4bbc65e 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -123,6 +123,9 @@ class Frame: id: Unique identifier for the frame instance. name: Human-readable name combining class name and instance count. pts: Presentation timestamp in nanoseconds. + broadcast_sibling_id: ID of the paired frame when this frame was + broadcast in both directions. Set automatically by + ``broadcast_frame()`` and ``broadcast_frame_instance()``. metadata: Dictionary for arbitrary frame metadata. transport_source: Name of the transport source that created this frame. transport_destination: Name of the transport destination for this frame. @@ -131,6 +134,7 @@ class Frame: id: int = field(init=False) name: str = field(init=False) pts: Optional[int] = field(init=False) + broadcast_sibling_id: Optional[int] = field(init=False) metadata: Dict[str, Any] = field(init=False) transport_source: Optional[str] = field(init=False) transport_destination: Optional[str] = field(init=False) @@ -139,6 +143,7 @@ class Frame: self.id: int = obj_id() self.name: str = f"{self.__class__.__name__}#{obj_count(self)}" self.pts: Optional[int] = None + self.broadcast_sibling_id: Optional[int] = None self.metadata: Dict[str, Any] = {} self.transport_source: Optional[str] = None self.transport_destination: Optional[str] = None diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 7423c4845..4cf32c800 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -809,8 +809,12 @@ class FrameProcessor(BaseObject): frame_cls: The class of the frame to be broadcasted. **kwargs: Keyword arguments to be passed to the frame's constructor. """ - await self.push_frame(frame_cls(**kwargs)) - await self.push_frame(frame_cls(**kwargs), FrameDirection.UPSTREAM) + downstream_frame = frame_cls(**kwargs) + upstream_frame = frame_cls(**kwargs) + downstream_frame.broadcast_sibling_id = upstream_frame.id + upstream_frame.broadcast_sibling_id = downstream_frame.id + await self.push_frame(downstream_frame) + await self.push_frame(upstream_frame, FrameDirection.UPSTREAM) async def broadcast_frame_instance(self, frame: Frame): """Broadcasts a frame instance upstream and downstream. @@ -834,15 +838,18 @@ class FrameProcessor(BaseObject): if not f.init and f.name not in ("id", "name") } - new_frame = frame_cls(**init_fields) + downstream_frame = frame_cls(**init_fields) for k, v in extra_fields.items(): - setattr(new_frame, k, v) - await self.push_frame(new_frame) + setattr(downstream_frame, k, v) - new_frame = frame_cls(**init_fields) + upstream_frame = frame_cls(**init_fields) for k, v in extra_fields.items(): - setattr(new_frame, k, v) - await self.push_frame(new_frame, FrameDirection.UPSTREAM) + setattr(upstream_frame, k, v) + + downstream_frame.broadcast_sibling_id = upstream_frame.id + upstream_frame.broadcast_sibling_id = downstream_frame.id + await self.push_frame(downstream_frame) + await self.push_frame(upstream_frame, FrameDirection.UPSTREAM) async def __start(self, frame: StartFrame): """Handle the start frame to initialize processor state. diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index c1497b40b..eb074699a 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -1220,10 +1220,9 @@ class RTVIObserver(BaseObserver): frame = data.frame direction = data.direction - # Only process downstream frames. Some frames are broadcast in both - # directions (e.g. UserStartedSpeakingFrame, FunctionCallResultFrame), - # and we only want to send one RTVI message per event. - if direction != FrameDirection.DOWNSTREAM: + # For broadcast frames (pushed in both directions), only process + # the downstream copy to avoid sending duplicate RTVI messages. + if frame.broadcast_sibling_id is not None and direction != FrameDirection.DOWNSTREAM: return # If we have already seen this frame, let's skip it. diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 6fe31aab5..15eb99f3a 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -625,6 +625,11 @@ class BaseOutputTransport(FrameProcessor): downstream_frame.transport_destination = self._destination upstream_frame = BotStartedSpeakingFrame() upstream_frame.transport_destination = self._destination + + # Setting the siblings id + upstream_frame.broadcast_sibling_id = downstream_frame.id + downstream_frame.broadcast_sibling_id = upstream_frame.id + await self._transport.push_frame(downstream_frame) await self._transport.push_frame(upstream_frame, FrameDirection.UPSTREAM) @@ -647,6 +652,11 @@ class BaseOutputTransport(FrameProcessor): downstream_frame.transport_destination = self._destination upstream_frame = BotStoppedSpeakingFrame() upstream_frame.transport_destination = self._destination + + # Setting the siblings id + upstream_frame.broadcast_sibling_id = downstream_frame.id + downstream_frame.broadcast_sibling_id = upstream_frame.id + await self._transport.push_frame(downstream_frame) await self._transport.push_frame(upstream_frame, FrameDirection.UPSTREAM)