wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix RTVIObserver missing upstream-only frames by adding broadcasted flag

Browse Source 
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.
This commit is contained in:
Mark Backman
2026-02-19 07:43:20 -07:00
parent 43869a499d
commit 63df4642b5
3 changed files with 13 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/pipecat/frames/frames.py
View File

@@ -131,6 +131,7 @@ class Frame:
id: int = field(init=False) id: int = field(init=False)
name: str = field(init=False) name: str = field(init=False)
pts: Optional[int] = field(init=False) pts: Optional[int] = field(init=False)
broadcasted: bool = field(init=False)
metadata: Dict[str, Any] = field(init=False) metadata: Dict[str, Any] = field(init=False)
transport_source: Optional[str] = field(init=False) transport_source: Optional[str] = field(init=False)
transport_destination: Optional[str] = field(init=False) transport_destination: Optional[str] = field(init=False)
@@ -139,6 +140,7 @@ class Frame:
self.id: int = obj_id() self.id: int = obj_id()
self.name: str = f"{self.__class__.__name__}#{obj_count(self)}" self.name: str = f"{self.__class__.__name__}#{obj_count(self)}"
self.pts: Optional[int] = None self.pts: Optional[int] = None
self.broadcasted: bool = False
self.metadata: Dict[str, Any] = {} self.metadata: Dict[str, Any] = {}
self.transport_source: Optional[str] = None self.transport_source: Optional[str] = None
self.transport_destination: Optional[str] = None self.transport_destination: Optional[str] = None

10
src/pipecat/processors/frame_processor.py
View File

@@ -787,8 +787,12 @@ class FrameProcessor(BaseObject):
frame_cls: The class of the frame to be broadcasted. frame_cls: The class of the frame to be broadcasted.
**kwargs: Keyword arguments to be passed to the frame's constructor. **kwargs: Keyword arguments to be passed to the frame's constructor.
""" """
await self.push_frame(frame_cls(**kwargs)) downstream_frame = frame_cls(**kwargs)
await self.push_frame(frame_cls(**kwargs), FrameDirection.UPSTREAM) 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): async def broadcast_frame_instance(self, frame: Frame):
"""Broadcasts a frame instance upstream and downstream. """Broadcasts a frame instance upstream and downstream.
@@ -815,11 +819,13 @@ class FrameProcessor(BaseObject):
new_frame = frame_cls(**init_fields) new_frame = frame_cls(**init_fields)
for k, v in extra_fields.items(): for k, v in extra_fields.items():
setattr(new_frame, k, v) setattr(new_frame, k, v)
new_frame.broadcasted = True
await self.push_frame(new_frame) await self.push_frame(new_frame)
new_frame = frame_cls(**init_fields) new_frame = frame_cls(**init_fields)
for k, v in extra_fields.items(): for k, v in extra_fields.items():
setattr(new_frame, k, v) setattr(new_frame, k, v)
new_frame.broadcasted = True
await self.push_frame(new_frame, FrameDirection.UPSTREAM) await self.push_frame(new_frame, FrameDirection.UPSTREAM)
async def __start(self, frame: StartFrame): async def __start(self, frame: StartFrame):

7
src/pipecat/processors/frameworks/rtvi.py
View File

@@ -1220,10 +1220,9 @@ class RTVIObserver(BaseObserver):
frame = data.frame frame = data.frame
direction = data.direction direction = data.direction
# Only process downstream frames. Some frames are broadcast in both # For broadcasted frames (pushed in both directions), only process
# directions (e.g. UserStartedSpeakingFrame, FunctionCallResultFrame), # the downstream copy to avoid sending duplicate RTVI messages.
# and we only want to send one RTVI message per event. if frame.broadcasted and direction != FrameDirection.DOWNSTREAM:
if direction != FrameDirection.DOWNSTREAM:
return return
# If we have already seen this frame, let's skip it. # If we have already seen this frame, let's skip it.