From 979739c1b7983d055f470b3c877dc6a6d372ebeb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 22 May 2024 11:26:08 -0700 Subject: [PATCH] processors: reset aggergator before pushing --- src/pipecat/processors/aggregators/llm_response.py | 9 ++++++--- src/pipecat/processors/aggregators/user_response.py | 9 ++++++++- 2 files changed, 14 insertions(+), 4 deletions(-) diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 9c2c9a214..e01fc9e02 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -117,12 +117,15 @@ class LLMResponseAggregator(FrameProcessor): async def _push_aggregation(self): if len(self._aggregation) > 0: self._messages.append({"role": self._role, "content": self._aggregation}) + + # Reset our accumulator state. Reset it before pushing it down, + # otherwise if the tasks gets cancelled we won't be able to clear + # things up. + self._reset() + frame = LLMMessagesFrame(self._messages) await self.push_frame(frame) - # Reset our accumulator state. - self._reset() - def _reset(self): self._aggregation = "" self._aggregating = False diff --git a/src/pipecat/processors/aggregators/user_response.py b/src/pipecat/processors/aggregators/user_response.py index 172350e26..a30e75b69 100644 --- a/src/pipecat/processors/aggregators/user_response.py +++ b/src/pipecat/processors/aggregators/user_response.py @@ -118,7 +118,14 @@ class ResponseAggregator(FrameProcessor): async def _push_aggregation(self): if len(self._aggregation) > 0: - await self.push_frame(TextFrame(self._aggregation.strip())) + frame = TextFrame(self._aggregation.strip()) + + # Reset our accumulator state. Reset it before pushing it down, + # otherwise if the tasks gets cancelled we won't be able to clear + # things up. + self._reset() + + await self.push_frame(frame) # Reset our accumulator state. self._reset()