From a29be38f48d3c581bd5ae18b03f97ee1af990501 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 2 Apr 2026 12:41:25 -0700 Subject: [PATCH] LLMUserAggregator: remove self-queued frame tracking The _self_queued_frames set and _internal_queue_frame wrapper were used to prevent re-processing SpeechControlParamsFrame that the aggregator queued to itself. Now that the frame is no longer special-cased, this tracking is unnecessary. Also removes unused FrameCallback import. --- .../aggregators/llm_response_universal.py | 21 +++---------------- 1 file changed, 3 insertions(+), 18 deletions(-) diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index fcc4fcf6d..911035fdc 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -73,7 +73,7 @@ from pipecat.processors.aggregators.llm_context_summarizer import ( LLMContextSummarizer, SummaryAppliedEvent, ) -from pipecat.processors.frame_processor import FrameCallback, FrameDirection, FrameProcessor +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.services.settings import LLMSettings from pipecat.turns.user_idle_controller import UserIdleController from pipecat.turns.user_mute import BaseUserMuteStrategy @@ -467,11 +467,6 @@ class LLMUserAggregator(LLMContextAggregator): self._vad_controller.add_event_handler("on_push_frame", self._on_push_frame) self._vad_controller.add_event_handler("on_broadcast_frame", self._on_broadcast_frame) - # NOTE(aleix): Probably just needed temporarily. This was added to - # prevent processing self-queued frames (SpeechControlParamsFrame) - # pushed by strategies. - self._self_queued_frames = set() - async def cleanup(self): """Clean up processor resources.""" await super().cleanup() @@ -654,16 +649,6 @@ class LLMUserAggregator(LLMContextAggregator): ) ) - async def _internal_queue_frame( - self, - frame: Frame, - direction: FrameDirection = FrameDirection.DOWNSTREAM, - callback: Optional[FrameCallback] = None, - ): - """Queues the given frame to ourselves.""" - self._self_queued_frames.add(frame.id) - await self.queue_frame(frame, direction, callback) - async def _queued_broadcast_frame(self, frame_cls: Type[Frame], **kwargs): """Broadcasts a frame upstream and queues it for internal processing. @@ -676,13 +661,13 @@ class LLMUserAggregator(LLMContextAggregator): **kwargs: Keyword arguments to be passed to the frame's constructor. """ - await self._internal_queue_frame(frame_cls(**kwargs)) + await self.queue_frame(frame_cls(**kwargs)) await self.push_frame(frame_cls(**kwargs), FrameDirection.UPSTREAM) async def _on_push_frame( self, controller, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM ): - await self._internal_queue_frame(frame, direction) + await self.queue_frame(frame, direction) async def _on_broadcast_frame(self, controller, frame_cls: Type[Frame], **kwargs): await self._queued_broadcast_frame(frame_cls, **kwargs)