Merge pull request #162 from pipecat-ai/reset-before-pushing

processors: reset aggergator before pushing

src/pipecat/processors/aggregators/llm_response.py

@@ -117,12 +117,15 @@ async def process_frame(self, frame: Frame, direction: FrameDirection):
     async def _push_aggregation(self):
         if len(self._aggregation) > 0:
             self._messages.append({"role": self._role, "content": self._aggregation})
-            frame = LLMMessagesFrame(self._messages)
-            await self.push_frame(frame)
 
-            # Reset our accumulator state.
+            # 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)
+
     def _reset(self):
         self._aggregation = ""
         self._aggregating = False

src/pipecat/processors/aggregators/user_response.py

@@ -118,7 +118,14 @@ async def process_frame(self, frame: Frame, direction: FrameDirection):
 
     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()