FrameProcessor: add broadcast_frame_instance()

src/pipecat/processors/frame_processor.py

@@ -12,7 +12,9 @@ management, and frame flow control mechanisms.
 """
 
 import asyncio
+import dataclasses
 import traceback
+from copy import deepcopy
 from dataclasses import dataclass
 from enum import Enum
 from typing import (
@@ -782,6 +784,38 @@ class FrameProcessor(BaseObject):
         await self.push_frame(frame_cls(**kwargs))
         await self.push_frame(frame_cls(**kwargs), FrameDirection.UPSTREAM)
 
+    async def broadcast_frame_instance(self, frame: Frame):
+        """Broadcasts a frame instance upstream and downstream.
+
+        This method extracts the class and init fields from the given frame
+        instance and creates two new instances to push upstream and downstream.
+
+        Args:
+            frame: The frame instance to broadcast.
+
+        Note:
+            Prefer using `broadcast_frame()` when possible, as it is more
+            efficient. This method should only be used when you are not the
+            creator of the frame and need to broadcast an existing instance.
+        """
+        frame_cls = type(frame)
+        init_fields = {f.name: getattr(frame, f.name) for f in dataclasses.fields(frame) if f.init}
+        extra_fields = {
+            f.name: getattr(frame, f.name)
+            for f in dataclasses.fields(frame)
+            if not f.init and f.name not in ("id", "name")
+        }
+
+        new_frame = frame_cls(**deepcopy(init_fields))
+        for k, v in deepcopy(extra_fields).items():
+            setattr(new_frame, k, v)
+        await self.push_frame(new_frame)
+
+        new_frame = frame_cls(**deepcopy(init_fields))
+        for k, v in deepcopy(extra_fields).items():
+            setattr(new_frame, k, v)
+        await self.push_frame(new_frame, FrameDirection.UPSTREAM)
+
     async def __start(self, frame: StartFrame):
         """Handle the start frame to initialize processor state.
 

src/pipecat/serializers/protobuf.py

@@ -126,7 +126,7 @@ class ProtobufFrameSerializer(FrameSerializer):
         if "pts" in args_dict:
             del args_dict["pts"]
 
-        # Special handling for MessageFrame -> OutputTransportMessageUrgentFrame
+        # Special handling for MessageFrame -> InputTransportMessageFrame
         if class_name == MessageFrame:
             try:
                 msg = json.loads(args_dict["data"])

tests/test_frame_processor.py

@@ -6,7 +6,8 @@
 
 import asyncio
 import unittest
-from dataclasses import dataclass
+from dataclasses import dataclass, field
+from typing import List
 
 from pipecat.frames.frames import (
     DataFrame,
@@ -24,6 +25,15 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.tests.utils import SleepFrame, run_test
 
 
+@dataclass
+class BroadcastTestFrame(DataFrame):
+    """Test frame with init fields for broadcast testing."""
+
+    text: str = ""
+    value: int = 0
+    items: List[str] = field(default_factory=list)
+
+
 class TestFrameProcessor(unittest.IsolatedAsyncioTestCase):
     async def test_before_after_events(self):
         identity = IdentityFilter()
@@ -186,3 +196,157 @@ class TestFrameProcessor(unittest.IsolatedAsyncioTestCase):
             frames_to_send=frames_to_send,
             expected_down_frames=expected_down_frames,
         )
+
+    async def test_broadcast_frame(self):
+        """Test that broadcast_frame creates two separate frames with fresh IDs."""
+        downstream_frames: List[Frame] = []
+        upstream_frames: List[Frame] = []
+
+        class BroadcastTestProcessor(FrameProcessor):
+            async def process_frame(self, frame: Frame, direction: FrameDirection):
+                await super().process_frame(frame, direction)
+                if isinstance(frame, TextFrame):
+                    await self.broadcast_frame(
+                        BroadcastTestFrame, text="hello", value=42, items=["a", "b"]
+                    )
+                else:
+                    await self.push_frame(frame, direction)
+
+        class CaptureProcessor(FrameProcessor):
+            def __init__(self, capture_list: List[Frame], direction: FrameDirection):
+                super().__init__()
+                self._capture_list = capture_list
+                self._capture_direction = direction
+
+            async def process_frame(self, frame: Frame, direction: FrameDirection):
+                await super().process_frame(frame, direction)
+                if direction == self._capture_direction and isinstance(frame, BroadcastTestFrame):
+                    self._capture_list.append(frame)
+                await self.push_frame(frame, direction)
+
+        up_capture = CaptureProcessor(upstream_frames, FrameDirection.UPSTREAM)
+        broadcaster = BroadcastTestProcessor()
+        down_capture = CaptureProcessor(downstream_frames, FrameDirection.DOWNSTREAM)
+
+        pipeline = Pipeline([up_capture, broadcaster, down_capture])
+
+        frames_to_send = [TextFrame(text="trigger")]
+        expected_down_frames = [BroadcastTestFrame]
+        expected_up_frames = [BroadcastTestFrame]
+
+        await run_test(
+            pipeline,
+            frames_to_send=frames_to_send,
+            expected_down_frames=expected_down_frames,
+            expected_up_frames=expected_up_frames,
+        )
+
+        # Verify we got one frame in each direction
+        self.assertEqual(len(downstream_frames), 1)
+        self.assertEqual(len(upstream_frames), 1)
+
+        down_frame = downstream_frames[0]
+        up_frame = upstream_frames[0]
+
+        # Verify the frames have different IDs (they are separate instances)
+        self.assertNotEqual(down_frame.id, up_frame.id)
+
+        # Verify the frames have the correct field values
+        self.assertEqual(down_frame.text, "hello")
+        self.assertEqual(down_frame.value, 42)
+        self.assertEqual(down_frame.items, ["a", "b"])
+        self.assertEqual(up_frame.text, "hello")
+        self.assertEqual(up_frame.value, 42)
+        self.assertEqual(up_frame.items, ["a", "b"])
+
+        # Verify the items lists are separate instances (not shared references)
+        self.assertIsNot(down_frame.items, up_frame.items)
+
+    async def test_broadcast_frame_instance(self):
+        """Test that broadcast_frame_instance copies all fields except id and name."""
+        downstream_frames: List[Frame] = []
+        upstream_frames: List[Frame] = []
+        original_frame: List[Frame] = []
+
+        class BroadcastInstanceTestProcessor(FrameProcessor):
+            async def process_frame(self, frame: Frame, direction: FrameDirection):
+                await super().process_frame(frame, direction)
+                if isinstance(frame, BroadcastTestFrame):
+                    # Set some non-init fields on the frame
+                    frame.pts = 12345
+                    frame.metadata = {"key": "value", "nested": {"a": 1}}
+                    original_frame.append(frame)
+                    await self.broadcast_frame_instance(frame)
+                else:
+                    await self.push_frame(frame, direction)
+
+        class CaptureProcessor(FrameProcessor):
+            def __init__(self, capture_list: List[Frame], direction: FrameDirection):
+                super().__init__()
+                self._capture_list = capture_list
+                self._capture_direction = direction
+
+            async def process_frame(self, frame: Frame, direction: FrameDirection):
+                await super().process_frame(frame, direction)
+                if direction == self._capture_direction and isinstance(frame, BroadcastTestFrame):
+                    self._capture_list.append(frame)
+                await self.push_frame(frame, direction)
+
+        up_capture = CaptureProcessor(upstream_frames, FrameDirection.UPSTREAM)
+        broadcaster = BroadcastInstanceTestProcessor()
+        down_capture = CaptureProcessor(downstream_frames, FrameDirection.DOWNSTREAM)
+
+        pipeline = Pipeline([up_capture, broadcaster, down_capture])
+
+        # Create a frame with mutable fields to test deep copying
+        test_frame = BroadcastTestFrame(text="test", value=99, items=["x", "y", "z"])
+
+        frames_to_send = [test_frame]
+        expected_down_frames = [BroadcastTestFrame]
+        expected_up_frames = [BroadcastTestFrame]
+
+        await run_test(
+            pipeline,
+            frames_to_send=frames_to_send,
+            expected_down_frames=expected_down_frames,
+            expected_up_frames=expected_up_frames,
+        )
+
+        # Verify we got one frame in each direction
+        self.assertEqual(len(downstream_frames), 1)
+        self.assertEqual(len(upstream_frames), 1)
+        self.assertEqual(len(original_frame), 1)
+
+        orig = original_frame[0]
+        down_frame = downstream_frames[0]
+        up_frame = upstream_frames[0]
+
+        # Verify the frames have different IDs and names (fresh values)
+        self.assertNotEqual(down_frame.id, orig.id)
+        self.assertNotEqual(up_frame.id, orig.id)
+        self.assertNotEqual(down_frame.id, up_frame.id)
+        self.assertNotEqual(down_frame.name, orig.name)
+        self.assertNotEqual(up_frame.name, orig.name)
+
+        # Verify init fields are copied correctly
+        self.assertEqual(down_frame.text, "test")
+        self.assertEqual(down_frame.value, 99)
+        self.assertEqual(down_frame.items, ["x", "y", "z"])
+        self.assertEqual(up_frame.text, "test")
+        self.assertEqual(up_frame.value, 99)
+        self.assertEqual(up_frame.items, ["x", "y", "z"])
+
+        # Verify non-init fields (except id/name) are copied
+        self.assertEqual(down_frame.pts, 12345)
+        self.assertEqual(down_frame.metadata, {"key": "value", "nested": {"a": 1}})
+        self.assertEqual(up_frame.pts, 12345)
+        self.assertEqual(up_frame.metadata, {"key": "value", "nested": {"a": 1}})
+
+        # Verify mutable fields are deep copied (not shared references)
+        self.assertIsNot(down_frame.items, orig.items)
+        self.assertIsNot(up_frame.items, orig.items)
+        self.assertIsNot(down_frame.items, up_frame.items)
+        self.assertIsNot(down_frame.metadata, orig.metadata)
+        self.assertIsNot(up_frame.metadata, orig.metadata)
+        self.assertIsNot(down_frame.metadata, up_frame.metadata)
+        self.assertIsNot(down_frame.metadata["nested"], up_frame.metadata["nested"])