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

LLMUserAggregator: remove self-queued frame tracking

Browse Source 
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.
This commit is contained in:
Aleix Conchillo Flaqué
2026-04-02 12:41:25 -07:00
parent 976c644f90
commit a29be38f48
1 changed files with 3 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
src/pipecat/processors/aggregators/llm_response_universal.py
View File

@@ -73,7 +73,7 @@ from pipecat.processors.aggregators.llm_context_summarizer import (
LLMContextSummarizer, LLMContextSummarizer,
SummaryAppliedEvent, 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.services.settings import LLMSettings
from pipecat.turns.user_idle_controller import UserIdleController from pipecat.turns.user_idle_controller import UserIdleController
from pipecat.turns.user_mute import BaseUserMuteStrategy 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_push_frame", self._on_push_frame)
self._vad_controller.add_event_handler("on_broadcast_frame", self._on_broadcast_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): async def cleanup(self):
"""Clean up processor resources.""" """Clean up processor resources."""
await super().cleanup() 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): async def _queued_broadcast_frame(self, frame_cls: Type[Frame], **kwargs):
"""Broadcasts a frame upstream and queues it for internal processing. """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. **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) await self.push_frame(frame_cls(**kwargs), FrameDirection.UPSTREAM)
async def _on_push_frame( async def _on_push_frame(
self, controller, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM 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): async def _on_broadcast_frame(self, controller, frame_cls: Type[Frame], **kwargs):
await self._queued_broadcast_frame(frame_cls, **kwargs) await self._queued_broadcast_frame(frame_cls, **kwargs)