From 63df4642b5fdcf01df84185795986f2b2bbeafa6 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 19 Feb 2026 07:43:20 -0700 Subject: [PATCH 1/6] Fix RTVIObserver missing upstream-only frames by adding broadcasted flag RTVIObserver previously filtered out all upstream frames to avoid duplicate messages from broadcasted frames. This caused upstream-only frames to be silently ignored. Instead, add a `broadcasted` field to the Frame base class that is set by broadcast_frame() and broadcast_frame_instance(), and only skip upstream copies of broadcasted frames. --- src/pipecat/frames/frames.py | 2 ++ src/pipecat/processors/frame_processor.py | 10 ++++++++-- src/pipecat/processors/frameworks/rtvi.py | 7 +++---- 3 files changed, 13 insertions(+), 6 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 8d237defc..c3f6226c6 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -131,6 +131,7 @@ class Frame: id: int = field(init=False) name: str = field(init=False) pts: Optional[int] = field(init=False) + broadcasted: bool = 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 +140,7 @@ class Frame: self.id: int = obj_id() self.name: str = f"{self.__class__.__name__}#{obj_count(self)}" self.pts: Optional[int] = None + self.broadcasted: bool = False 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 f0c9e7183..48c8c718c 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -787,8 +787,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) + downstream_frame.broadcasted = True + await self.push_frame(downstream_frame) + upstream_frame = frame_cls(**kwargs) + upstream_frame.broadcasted = True + await self.push_frame(upstream_frame, FrameDirection.UPSTREAM) async def broadcast_frame_instance(self, frame: Frame): """Broadcasts a frame instance upstream and downstream. @@ -815,11 +819,13 @@ class FrameProcessor(BaseObject): new_frame = frame_cls(**init_fields) for k, v in extra_fields.items(): setattr(new_frame, k, v) + new_frame.broadcasted = True await self.push_frame(new_frame) new_frame = frame_cls(**init_fields) for k, v in extra_fields.items(): setattr(new_frame, k, v) + new_frame.broadcasted = True await self.push_frame(new_frame, FrameDirection.UPSTREAM) async def __start(self, frame: StartFrame): diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index c1497b40b..8c7bb15ef 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 broadcasted frames (pushed in both directions), only process + # the downstream copy to avoid sending duplicate RTVI messages. + if frame.broadcasted and direction != FrameDirection.DOWNSTREAM: return # If we have already seen this frame, let's skip it. From 50ef4909e3a97020b103d27716d3a7c9c9e28a10 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 19 Feb 2026 07:44:52 -0700 Subject: [PATCH 2/6] Add changelog entries for PR #3774 --- changelog/3774.added.md | 1 + changelog/3774.fixed.md | 1 + 2 files changed, 2 insertions(+) create mode 100644 changelog/3774.added.md create mode 100644 changelog/3774.fixed.md diff --git a/changelog/3774.added.md b/changelog/3774.added.md new file mode 100644 index 000000000..032d8238f --- /dev/null +++ b/changelog/3774.added.md @@ -0,0 +1 @@ +- Added `broadcasted` field to the base `Frame` class. This field is automatically set to `True` by `broadcast_frame()` and `broadcast_frame_instance()` to distinguish broadcasted frames from single-direction frames. 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. From b1cee140b9cabc6fb7bc78727a85e47a2251d134 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Thu, 19 Feb 2026 15:06:50 -0300 Subject: [PATCH 3/6] Refactoring to use broadcasted_sibling_id instead of broadcasted field. --- changelog/3774.added.md | 2 +- src/pipecat/frames/frames.py | 7 +++++-- src/pipecat/processors/frame_processor.py | 23 ++++++++++++----------- src/pipecat/processors/frameworks/rtvi.py | 2 +- 4 files changed, 19 insertions(+), 15 deletions(-) diff --git a/changelog/3774.added.md b/changelog/3774.added.md index 032d8238f..8e442999c 100644 --- a/changelog/3774.added.md +++ b/changelog/3774.added.md @@ -1 +1 @@ -- Added `broadcasted` field to the base `Frame` class. This field is automatically set to `True` by `broadcast_frame()` and `broadcast_frame_instance()` to distinguish broadcasted frames from single-direction frames. +- Added `broadcasted_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/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index c3f6226c6..e6bc008ab 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. + broadcasted_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,7 +134,7 @@ class Frame: id: int = field(init=False) name: str = field(init=False) pts: Optional[int] = field(init=False) - broadcasted: bool = field(init=False) + broadcasted_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) @@ -140,7 +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.broadcasted: bool = False + self.broadcasted_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 48c8c718c..6bdd9ae1d 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -788,10 +788,10 @@ class FrameProcessor(BaseObject): **kwargs: Keyword arguments to be passed to the frame's constructor. """ downstream_frame = frame_cls(**kwargs) - downstream_frame.broadcasted = True - await self.push_frame(downstream_frame) upstream_frame = frame_cls(**kwargs) - upstream_frame.broadcasted = True + downstream_frame.broadcasted_sibling_id = upstream_frame.id + upstream_frame.broadcasted_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): @@ -816,17 +816,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) - new_frame.broadcasted = True - 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) - new_frame.broadcasted = True - await self.push_frame(new_frame, FrameDirection.UPSTREAM) + setattr(upstream_frame, k, v) + + downstream_frame.broadcasted_sibling_id = upstream_frame.id + upstream_frame.broadcasted_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 8c7bb15ef..55be46452 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -1222,7 +1222,7 @@ class RTVIObserver(BaseObserver): # For broadcasted frames (pushed in both directions), only process # the downstream copy to avoid sending duplicate RTVI messages. - if frame.broadcasted and direction != FrameDirection.DOWNSTREAM: + if frame.broadcasted_sibling_id is not None and direction != FrameDirection.DOWNSTREAM: return # If we have already seen this frame, let's skip it. From d608c400f93eba752aee43d534a4d5f089bf6555 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Thu, 19 Feb 2026 15:49:22 -0300 Subject: [PATCH 4/6] Preventing the duplicated BotStartedSpeakingFrame and BotStoppedSpeakingFrame. --- src/pipecat/transports/base_output.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index c4f59a61d..4f072c0df 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -613,6 +613,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.broadcasted_sibling_id = downstream_frame.id + downstream_frame.broadcasted_sibling_id = upstream_frame.id + await self._transport.push_frame(downstream_frame) await self._transport.push_frame(upstream_frame, FrameDirection.UPSTREAM) @@ -635,6 +640,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.broadcasted_sibling_id = downstream_frame.id + downstream_frame.broadcasted_sibling_id = upstream_frame.id + await self._transport.push_frame(downstream_frame) await self._transport.push_frame(upstream_frame, FrameDirection.UPSTREAM) From 5b2fa69bdc5f6a888508a0c52189d2add6899b32 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Thu, 19 Feb 2026 16:24:07 -0300 Subject: [PATCH 5/6] Renaming from broadcasted_sibling_id to broadcast_sibling_id --- changelog/3774.added.md | 2 +- src/pipecat/frames/frames.py | 6 +++--- src/pipecat/processors/frame_processor.py | 8 ++++---- src/pipecat/processors/frameworks/rtvi.py | 2 +- src/pipecat/transports/base_output.py | 8 ++++---- 5 files changed, 13 insertions(+), 13 deletions(-) diff --git a/changelog/3774.added.md b/changelog/3774.added.md index 8e442999c..e72599e60 100644 --- a/changelog/3774.added.md +++ b/changelog/3774.added.md @@ -1 +1 @@ -- Added `broadcasted_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. +- 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/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index e6bc008ab..62c0d1352 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -123,7 +123,7 @@ class Frame: id: Unique identifier for the frame instance. name: Human-readable name combining class name and instance count. pts: Presentation timestamp in nanoseconds. - broadcasted_sibling_id: ID of the paired frame when this frame was + 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. @@ -134,7 +134,7 @@ class Frame: id: int = field(init=False) name: str = field(init=False) pts: Optional[int] = field(init=False) - broadcasted_sibling_id: 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) @@ -143,7 +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.broadcasted_sibling_id: 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 6bdd9ae1d..0d20214b2 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -789,8 +789,8 @@ class FrameProcessor(BaseObject): """ downstream_frame = frame_cls(**kwargs) upstream_frame = frame_cls(**kwargs) - downstream_frame.broadcasted_sibling_id = upstream_frame.id - upstream_frame.broadcasted_sibling_id = downstream_frame.id + 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) @@ -824,8 +824,8 @@ class FrameProcessor(BaseObject): for k, v in extra_fields.items(): setattr(upstream_frame, k, v) - downstream_frame.broadcasted_sibling_id = upstream_frame.id - upstream_frame.broadcasted_sibling_id = downstream_frame.id + 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) diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 55be46452..a8b66ccdd 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -1222,7 +1222,7 @@ class RTVIObserver(BaseObserver): # For broadcasted frames (pushed in both directions), only process # the downstream copy to avoid sending duplicate RTVI messages. - if frame.broadcasted_sibling_id is not None and direction != FrameDirection.DOWNSTREAM: + 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 4f072c0df..1a5083128 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -615,8 +615,8 @@ class BaseOutputTransport(FrameProcessor): upstream_frame.transport_destination = self._destination # Setting the siblings id - upstream_frame.broadcasted_sibling_id = downstream_frame.id - downstream_frame.broadcasted_sibling_id = upstream_frame.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) @@ -642,8 +642,8 @@ class BaseOutputTransport(FrameProcessor): upstream_frame.transport_destination = self._destination # Setting the siblings id - upstream_frame.broadcasted_sibling_id = downstream_frame.id - downstream_frame.broadcasted_sibling_id = upstream_frame.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) From 63caa403cb51f092329d6ef554e4361a0c75ad6b Mon Sep 17 00:00:00 2001 From: filipi87 Date: Thu, 19 Feb 2026 17:31:25 -0300 Subject: [PATCH 6/6] Improving RTVI doc description. --- src/pipecat/processors/frameworks/rtvi.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index a8b66ccdd..eb074699a 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -1220,7 +1220,7 @@ class RTVIObserver(BaseObserver): frame = data.frame direction = data.direction - # For broadcasted frames (pushed in both directions), only process + # 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