From d459465eb652a923a8df731c1eb8898fbf832d71 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 30 Dec 2025 14:27:03 -0800 Subject: [PATCH] FrameProcessor: add queue_task_frame() and queue_task_frames() --- changelog/3326.added.md | 1 + src/pipecat/processors/frame_processor.py | 17 +++++++++++++++++ 2 files changed, 18 insertions(+) create mode 100644 changelog/3326.added.md diff --git a/changelog/3326.added.md b/changelog/3326.added.md new file mode 100644 index 000000000..d362fa09d --- /dev/null +++ b/changelog/3326.added.md @@ -0,0 +1 @@ +- Frame processors can now push frames from the top of the pipeline using new methods `queue_task_frame()` and `queue_task_frames()`. diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 09ec36b9a..cc1f89c6b 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -41,6 +41,7 @@ from pipecat.frames.frames import ( FrameProcessorResumeUrgentFrame, InterruptionFrame, InterruptionTaskFrame, + QueueTaskFrame, StartFrame, SystemFrame, UninterruptibleFrame, @@ -612,6 +613,22 @@ class FrameProcessor(BaseObject): else: await self.__input_queue.put((frame, direction, callback)) + async def queue_task_frame(self, frame: Frame): + """Queue a single frame to be pushed from the pipeline task down to the pipeline. + + Args: + frame: A single frame to push downstream from the top of the pipeline. + """ + await self.queue_task_frames([frame]) + + async def queue_task_frames(self, frames: Sequence[Frame]): + """Queue multiple frames to be pushed from the pipeline task down to the pipeline. + + Args: + frames: A sequence of frames to push downstream from the top of the pipeline. + """ + await self.push_frame(QueueTaskFrame(frames=frames), FrameDirection.UPSTREAM) + async def pause_processing_frames(self): """Pause processing of queued frames.""" logger.trace(f"{self}: pausing frame processing")