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

introduce UninterruptibleFrame frames

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-12-04 14:51:53 -08:00
parent fed6a8b669
commit 7ad3969690
4 changed files with 127 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
changelog/3189.added.md Normal file
View File

@@ -0,0 +1,6 @@
- Data and control frames can now be marked as non-interruptible by using the
`UninterruptibleFrame` mixin. Frames marked as `UninterruptibleFrame` will not
be interrupted during processing, and any queued frames of this type will be
retained in the internal queues. This is useful when you need ordered frames
(data or control) that should not be discarded or cancelled due to
interruptions.

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

@@ -186,6 +186,20 @@ class ControlFrame(Frame):
#
@dataclass
class UninterruptibleFrame:
"""A marker for data or control frames that must not be interrupted.
Frames with this mixin are still ordered normally, but unlike other frames,
they are preserved during interruptions: they remain in internal queues and
any task processing them will not be cancelled. This ensures the frame is
always delivered and processed to completion.
"""
pass
@dataclass
class AudioRawFrame:
"""A frame containing a chunk of raw audio.

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

@@ -33,6 +33,7 @@ from pipecat.frames.frames import (
InterruptionTaskFrame,
StartFrame,
SystemFrame,
UninterruptibleFrame,
)
from pipecat.metrics.metrics import LLMTokenUsage, MetricsData
from pipecat.observers.base_observer import BaseObserver, FrameProcessed, FramePushed
@@ -211,6 +212,7 @@ class FrameProcessor(BaseObject):
# The input task that handles all types of frames. It processes system
# frames right away and queues non-system frames for later processing.
self.__should_block_system_frames = False
self.__input_queue = FrameProcessorQueue()
self.__input_event: Optional[asyncio.Event] = None
self.__input_frame_task: Optional[asyncio.Task] = None
@@ -220,8 +222,10 @@ class FrameProcessor(BaseObject):
# called. To resume processing frames we need to call
# `resume_processing_frames()` which will wake up the event.
self.__should_block_frames = False
self.__process_queue = asyncio.Queue()
self.__process_event: Optional[asyncio.Event] = None
self.__process_frame_task: Optional[asyncio.Task] = None
self.__process_current_frame: Optional[Frame] = None
# To interrupt a pipeline, we push an `InterruptionTaskFrame` upstream.
# Then we wait for the corresponding `InterruptionFrame` to travel from
@@ -805,8 +809,12 @@ class FrameProcessor(BaseObject):
# interruption). Instead we just drain the queue because this is
# an interruption.
self.__reset_process_task()
elif isinstance(self.__process_current_frame, UninterruptibleFrame):
# We don't want to cancel UninterruptibleFrame, so we simply
# cleanup the queue.
self.__reset_process_queue()
else:
# Cancel and re-create the process task including the queue.
# Cancel and re-create the process task.
await self.__cancel_process_task()
self.__create_process_task()
except Exception as e:
@@ -872,7 +880,6 @@ class FrameProcessor(BaseObject):
if not self.__input_frame_task:
self.__input_event = asyncio.Event()
self.__input_queue = FrameProcessorQueue()
self.__input_frame_task = self.create_task(self.__input_frame_task_handler())
async def __cancel_input_task(self):
@@ -890,9 +897,7 @@ class FrameProcessor(BaseObject):
return
if not self.__process_frame_task:
self.__should_block_frames = False
self.__process_event = asyncio.Event()
self.__process_queue = asyncio.Queue()
self.__reset_process_task()
self.__process_frame_task = self.create_task(self.__process_frame_task_handler())
def __reset_process_task(self):
@@ -902,10 +907,26 @@ class FrameProcessor(BaseObject):
self.__should_block_frames = False
self.__process_event = asyncio.Event()
self.__reset_process_queue()
def __reset_process_queue(self):
"""Reset non-system frame processing queue."""
# Create a new queue to insert UninterruptibleFrame frames.
new_queue = asyncio.Queue()
# Process current queue and keep UninterruptibleFrame frames.
while not self.__process_queue.empty():
self.__process_queue.get_nowait()
item = self.__process_queue.get_nowait()
if isinstance(item, UninterruptibleFrame):
new_queue.put_nowait(item)
self.__process_queue.task_done()
# Put back UninterruptibleFrame frames into our process queue.
while not new_queue.empty():
item = new_queue.get_nowait()
self.__process_queue.put_nowait(item)
new_queue.task_done()
async def __cancel_process_task(self):
"""Cancel the non-system frame processing task."""
if self.__process_frame_task:
@@ -959,8 +980,12 @@ class FrameProcessor(BaseObject):
async def __process_frame_task_handler(self):
"""Handle non-system frames from the process queue."""
while True:
self.__process_current_frame = None
(frame, direction, callback) = await self.__process_queue.get()
self.__process_current_frame = frame
if self.__should_block_frames and self.__process_event:
logger.trace(f"{self}: frame processing paused")
await self.__process_event.wait()

76
tests/test_frame_processor.py
View File

@@ -6,13 +6,17 @@
import asyncio
import unittest
from dataclasses import dataclass
from pipecat.frames.frames import (
DataFrame,
EndFrame,
Frame,
InterruptionFrame,
OutputTransportMessageUrgentFrame,
SystemFrame,
TextFrame,
UninterruptibleFrame,
)
from pipecat.pipeline.pipeline import Pipeline
from pipecat.processors.filters.identity_filter import IdentityFilter
@@ -110,3 +114,75 @@ class TestFrameProcessor(unittest.IsolatedAsyncioTestCase):
expected_down_frames=expected_down_frames,
send_end_frame=False,
)
async def test_interruptible_frames(self):
@dataclass
class TestInterruptibleFrame(DataFrame):
text: str
class DelayTestFrameProcessor(FrameProcessor):
"""This processor just delays processing frames so we have time to
try to interrupt them.
"""
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
if not isinstance(frame, SystemFrame):
# Sleep more than SleepFrame default.
await asyncio.sleep(0.4)
await self.push_frame(frame, direction)
pipeline = Pipeline([DelayTestFrameProcessor()])
frames_to_send = [
TestInterruptibleFrame(text="Hello from Pipecat!"),
# Make sure we hit the DelayTestFrameProcessor first.
SleepFrame(),
# Just a random interruption. This should cause the interruption of
# TestInterruptibleFrame.
InterruptionFrame(),
]
expected_down_frames = [InterruptionFrame]
await run_test(
pipeline,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
)
async def test_uninterruptible_frames(self):
@dataclass
class TestUninterruptibleFrame(DataFrame, UninterruptibleFrame):
text: str
class DelayTestFrameProcessor(FrameProcessor):
"""This processor just delays processing non-InterruptionFrame so we
have time to try to interrupt them.
"""
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
if not isinstance(frame, SystemFrame):
# Sleep more than SleepFrame default.
await asyncio.sleep(0.4)
await self.push_frame(frame, direction)
pipeline = Pipeline([DelayTestFrameProcessor()])
frames_to_send = [
TestUninterruptibleFrame(text="Hello from Pipecat!"),
# Make sure we hit the DelayTestFrameProcessor first.
SleepFrame(),
# Just a random interruption. This should not cause the interruption
# of TestUninterruptibleFrame.
InterruptionFrame(),
]
expected_down_frames = [
InterruptionFrame,
TestUninterruptibleFrame,
]
await run_test(
pipeline,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
)