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

Merge pull request #3592 from pipecat-ai/aleix/broadcast-frame-no-deepcopy

Browse Source 
don't deep copy fields when broadcasting frames
This commit is contained in:
Aleix Conchillo Flaqué
2026-01-29 11:50:20 -08:00
committed by GitHub
parent f3b72e9263 0423acd8a0
commit ca613bcb79
3 changed files with 21 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -14,7 +14,6 @@ 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 (
@@ -775,20 +774,21 @@ class FrameProcessor(BaseObject):
"""Broadcasts a frame of the specified class upstream and downstream.
This method creates two instances of the given frame class using the
provided keyword arguments and pushes them upstream and downstream.
provided keyword arguments (without deep-copying them) and pushes them
upstream and downstream.
Args:
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(**deepcopy(kwargs)))
await self.push_frame(frame_cls(**deepcopy(kwargs)), FrameDirection.UPSTREAM)
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 creates two new frame instances copying all fields from the
original frame except `id` and `name`, which get fresh values.
This method creates two new frame instances shallow-copying all fields
from the original frame except `id` and `name`, which get fresh values.
Args:
frame: The frame instance to broadcast.
@@ -806,13 +806,13 @@ class FrameProcessor(BaseObject):
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():
new_frame = frame_cls(**init_fields)
for k, v in 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():
new_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)

4
src/pipecat/services/stt_service.py
View File

@@ -497,11 +497,11 @@ class SegmentedSTTService(STTService):
wav.close()
content.seek(0)
await self.process_generator(self.run_stt(content.read()))
# Start clean.
self._audio_buffer.clear()
await self.process_generator(self.run_stt(content.read()))
async def process_audio_frame(self, frame: AudioRawFrame, direction: FrameDirection):
"""Process audio frames by buffering them for segmented transcription.

21
tests/test_frame_processor.py
View File

@@ -259,11 +259,11 @@ class TestFrameProcessor(unittest.IsolatedAsyncioTestCase):
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)
# Verify the items lists are shared references (no deep copy)
self.assertIs(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."""
"""Test that broadcast_frame_instance shallow-copies all fields except id and name."""
downstream_frames: List[Frame] = []
upstream_frames: List[Frame] = []
original_frame: List[Frame] = []
@@ -298,7 +298,7 @@ class TestFrameProcessor(unittest.IsolatedAsyncioTestCase):
pipeline = Pipeline([up_capture, broadcaster, down_capture])
# Create a frame with mutable fields to test deep copying
# Create a frame with mutable fields to test shallow copying
test_frame = BroadcastTestFrame(text="test", value=99, items=["x", "y", "z"])
frames_to_send = [test_frame]
@@ -342,11 +342,8 @@ class TestFrameProcessor(unittest.IsolatedAsyncioTestCase):
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"])
# Verify mutable fields are shallow-copied (shared references)
self.assertIs(down_frame.items, orig.items)
self.assertIs(up_frame.items, orig.items)
self.assertIs(down_frame.metadata, orig.metadata)
self.assertIs(up_frame.metadata, orig.metadata)