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

Revert "fix interruption task frame context ordering"

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-11-05 12:14:03 -08:00
committed by GitHub
parent 11b101e8a6
commit d844829538
5 changed files with 127 additions and 346 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -11,7 +11,6 @@ including data frames, system frames, and control frames for audio, video, text,
and LLM processing.
"""
import asyncio
from dataclasses import dataclass, field
from typing import (
TYPE_CHECKING,
@@ -860,13 +859,9 @@ class InterruptionFrame(SystemFrame):
speaking (i.e. is interrupting). This is similar to
UserStartedSpeakingFrame except that it should be pushed concurrently
with other frames (so the order is not guaranteed).
Parameters:
finished_event: If not None, the event will be set when the frame
reaches the end of the pipeline.
"""
finished_event: Optional[asyncio.Event] = None
pass
@dataclass
@@ -1427,13 +1422,9 @@ class InterruptionTaskFrame(TaskFrame):
same actions as if the user interrupted except that the
UserStartedSpeakingFrame and UserStoppedSpeakingFrame won't be generated.
This frame should be pushed upstream.
Parameters:
finished_event: If not None, the event will be set when the generated
InterruptionFrame reaches the end of the pipeline.
"""
finished_event: Optional[asyncio.Event] = None
pass
@dataclass

6
src/pipecat/pipeline/task.py
View File

@@ -745,7 +745,7 @@ class PipelineTask(BasePipelineTask):
# pipeline. This is in case the push task is blocked waiting for a
# pipeline-ending frame to finish traversing the pipeline.
logger.debug(f"{self}: received interruption task frame {frame}")
await self._pipeline.queue_frame(InterruptionFrame(finished_event=frame.finished_event))
await self._pipeline.queue_frame(InterruptionFrame())
elif isinstance(frame, ErrorFrame):
await self._call_event_handler("on_pipeline_error", frame)
if frame.fatal:
@@ -786,10 +786,6 @@ class PipelineTask(BasePipelineTask):
self._pipeline_end_event.set()
elif isinstance(frame, HeartbeatFrame):
await self._heartbeat_queue.put(frame)
elif isinstance(frame, InterruptionFrame) and frame.finished_event:
# This should unblock any code waiting for the interruption to
# complete.
frame.finished_event.set()
async def _heartbeat_push_handler(self):
"""Push heartbeat frames at regular intervals."""

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

@@ -93,7 +93,7 @@ class FrameProcessorQueue(asyncio.PriorityQueue):
self.__high_counter = 0
self.__low_counter = 0
async def put(self, item: Tuple[Frame, FrameDirection, Optional[FrameCallback]]):
async def put(self, item: Tuple[Frame, FrameDirection, FrameCallback]):
"""Put an item into the priority queue.
System frames (`SystemFrame`) have higher priority than any other
@@ -228,9 +228,11 @@ class FrameProcessor(BaseObject):
# To interrupt a pipeline, we push an `InterruptionTaskFrame` upstream.
# Then we wait for the corresponding `InterruptionFrame` to travel from
# start to end of the pipeline. When it reaches the end we will be
# notified through the assigned event.
# the start of the pipeline back to the processor that sent the
# `InterruptionTaskFrame`. This wait is handled using the following
# event.
self._wait_for_interruption = False
self._wait_interruption_event = asyncio.Event()
# Frame processor events.
self._register_event_handler("on_before_process_frame", sync=True)
@@ -565,6 +567,10 @@ class FrameProcessor(BaseObject):
if self._cancelling:
return
# If we are waiting for an interruption we will bypass all queued system
# frames and we will process the frame right away. This is because a
# previous system frame might be waiting for the interruption frame and
# it's blocking the input task.
if self._wait_for_interruption and isinstance(frame, InterruptionFrame):
await self.__process_frame(frame, direction, callback)
return
@@ -655,27 +661,31 @@ class FrameProcessor(BaseObject):
await self._call_event_handler("on_after_push_frame", frame)
# If we are waiting for an interruption and we get an interruption, then
# we can unblock `push_interruption_task_frame_and_wait()`.
if self._wait_for_interruption and isinstance(frame, InterruptionFrame):
self._wait_interruption_event.set()
async def push_interruption_task_frame_and_wait(self):
"""Interrupt the pipeline and wait for the interruption to complete.
This function sends an `InterruptionTaskFrame` upstream with an
associated asyncio event. It then waits for the generated
`InterruptionFrame` to reach the end of the pipeline where the event
will be set.
"""Push an interruption task frame upstream and wait for the interruption.
This function sends an `InterruptionTaskFrame` upstream to the pipeline
task and waits to receive the corresponding `InterruptionFrame`. When
the function finishes it is guaranteed that the `InterruptionFrame` has
been pushed downstream.
"""
self._wait_for_interruption = True
finished_event = asyncio.Event()
await self.push_frame(InterruptionTaskFrame(), FrameDirection.UPSTREAM)
await self.push_frame(
InterruptionTaskFrame(finished_event=finished_event), FrameDirection.UPSTREAM
)
# Wait for an `InterruptionFrame` to come to this processor and be
# pushed. Take a look at `push_frame()` to see how we first push the
# `InterruptionFrame` and then we set the event in order to maintain
# frame ordering.
await self._wait_interruption_event.wait()
# Wait for the event to be set. This event is set when the
# `InterruptionFrame` pushed by the pipeline task reaches the end of the
# pipeline.
await finished_event.wait()
# Clean the event.
self._wait_interruption_event.clear()
self._wait_for_interruption = False