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

LLMUserAggregator: queue frames internally so strategies and controllers can process them

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2026-01-28 16:17:47 -08:00
parent a69a037ffa
commit 183c0aa4ef
1 changed files with 21 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -64,7 +64,7 @@ from pipecat.processors.aggregators.llm_context import (
LLMSpecificMessage,
NotGiven,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.processors.frame_processor import FrameCallback, FrameDirection, FrameProcessor
from pipecat.turns.user_idle_controller import UserIdleController
from pipecat.turns.user_mute import BaseUserMuteStrategy
from pipecat.turns.user_start import BaseUserTurnStartStrategy, UserTurnStartedParams
@@ -452,7 +452,7 @@ class LLMUserAggregator(LLMContextAggregator):
await self.push_frame(frame, direction)
elif isinstance(frame, LLMSetToolChoiceFrame):
self.set_tool_choice(frame.tool_choice)
elif isinstance(frame, SpeechControlParamsFrame):
elif isinstance(frame, SpeechControlParamsFrame) and not self._vad_controller:
await self._handle_speech_control_params(frame)
else:
await self.push_frame(frame, direction)
@@ -587,25 +587,36 @@ class LLMUserAggregator(LLMContextAggregator):
)
)
async def _queued_broadcast_frame(self, frame_cls: Type[Frame], **kwargs):
"""Broadcasts a frame upstream and queues it downstream for internal processing.
Queues the frame downstream so it flows through ``process_frame`` and
is handled internally (e.g. by the ``UserTurnController``). The
upstream frame is pushed directly.
Args:
frame_cls: The class of the frame to be broadcasted.
**kwargs: Keyword arguments to be passed to the frame's constructor.
"""
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.push_frame(frame, direction)
await self.queue_frame(frame, direction)
async def _on_broadcast_frame(self, controller, frame_cls: Type[Frame], **kwargs):
await self.broadcast_frame(frame_cls, **kwargs)
await self._queued_broadcast_frame(frame_cls, **kwargs)
async def _on_vad_speech_started(self, controller):
await self.queue_frame(VADUserStartedSpeakingFrame())
await self.broadcast_frame(VADUserStartedSpeakingFrame)
await self._queued_broadcast_frame(VADUserStartedSpeakingFrame)
async def _on_vad_speech_stopped(self, controller):
await self.queue_frame(VADUserStoppedSpeakingFrame())
await self.broadcast_frame(VADUserStoppedSpeakingFrame)
await self._queued_broadcast_frame(VADUserStoppedSpeakingFrame)
async def _on_vad_speech_activity(self, controller):
await self.queue_frame(UserSpeakingFrame())
await self.broadcast_frame(UserSpeakingFrame)
await self._queued_broadcast_frame(UserSpeakingFrame)
async def _on_user_turn_started(
self,