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

LLMAssistantContextAggregator: stop emulations if the user starts speaking

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-03-26 14:39:12 -07:00
parent 750bb88586
commit 055a3f1c53
1 changed files with 11 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
src/pipecat/processors/aggregators/llm_response.py
View File

@@ -249,7 +249,7 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
self._waiting_for_aggregation = False self._waiting_for_aggregation = False
async def handle_aggregation(self, aggregation: str): async def handle_aggregation(self, aggregation: str):
self._context.add_message({"role": self.role, "content": self._aggregation}) self._context.add_message({"role": self.role, "content": aggregation})
async def process_frame(self, frame: Frame, direction: FrameDirection): async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction) await super().process_frame(frame, direction)
@@ -290,12 +290,14 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
async def push_aggregation(self): async def push_aggregation(self):
if len(self._aggregation) > 0: if len(self._aggregation) > 0:
await self.handle_aggregation(self._aggregation) aggregation = self._aggregation
# Reset the aggregation. Reset it before pushing it down, otherwise # Reset the aggregation. Reset it before pushing it down, otherwise
# if the tasks gets cancelled we won't be able to clear things up. # if the tasks gets cancelled we won't be able to clear things up.
self.reset() self.reset()
await self.handle_aggregation(aggregation)
frame = OpenAILLMContextFrame(self._context) frame = OpenAILLMContextFrame(self._context)
await self.push_frame(frame) await self.push_frame(frame)
@@ -308,10 +310,16 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
async def _cancel(self, frame: CancelFrame): async def _cancel(self, frame: CancelFrame):
await self._cancel_aggregation_task() await self._cancel_aggregation_task()
async def _handle_user_started_speaking(self, _: UserStartedSpeakingFrame): async def _handle_user_started_speaking(self, frame: UserStartedSpeakingFrame):
self._user_speaking = True self._user_speaking = True
self._waiting_for_aggregation = True self._waiting_for_aggregation = True
# If we get a non-emulated UserStartedSpeakingFrame but we are in the
# middle of emulating VAD, let's stop emulating VAD (i.e. don't send the
# EmulateUserStoppedSpeakingFrame).
if not frame.emulated and self._emulating_vad:
self._emulating_vad = False
async def _handle_user_stopped_speaking(self, _: UserStoppedSpeakingFrame): async def _handle_user_stopped_speaking(self, _: UserStoppedSpeakingFrame):
self._user_speaking = False self._user_speaking = False
# We just stopped speaking. Let's see if there's some aggregation to # We just stopped speaking. Let's see if there's some aggregation to