processors: fix user response processors

CHANGELOG.md

@@ -19,6 +19,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed an issue in `LLMUserResponseAggregator` and `UserResponseAggregator`
+  that would cause frames after a brief pause to not be pushed to the LLM.
+
 - Clear the audio output buffer if we are interrupted.
 
 - Re-add exponential smoothing after volume calculation. This makes sure the

src/pipecat/processors/aggregators/llm_response.py

@@ -78,10 +78,14 @@ class LLMResponseAggregator(FrameProcessor):
         send_aggregation = False
 
         if isinstance(frame, self._start_frame):
-            self._seen_start_frame = True
+            self._aggregation = ""
             self._aggregating = True
+            self._seen_start_frame = True
+            self._seen_end_frame = False
+            self._seen_interim_results = False
         elif isinstance(frame, self._end_frame):
             self._seen_end_frame = True
+            self._seen_start_frame = False
 
             # We might have received the end frame but we might still be
             # aggregating (i.e. we have seen interim results but not the final
@@ -118,10 +122,9 @@ class LLMResponseAggregator(FrameProcessor):
         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()
+            # Reset the aggregation. Reset it before pushing it down, otherwise
+            # if the tasks gets cancelled we won't be able to clear things up.
+            self._aggregation = ""
 
             frame = LLMMessagesFrame(self._messages)
             await self.push_frame(frame)

src/pipecat/processors/aggregators/user_response.py

@@ -85,10 +85,13 @@ class ResponseAggregator(FrameProcessor):
         send_aggregation = False
 
         if isinstance(frame, self._start_frame):
-            self._seen_start_frame = True
             self._aggregating = True
+            self._seen_start_frame = True
+            self._seen_end_frame = False
+            self._seen_interim_results = False
         elif isinstance(frame, self._end_frame):
             self._seen_end_frame = True
+            self._seen_start_frame = False
 
             # We might have received the end frame but we might still be
             # aggregating (i.e. we have seen interim results but not the final
@@ -120,10 +123,9 @@ class ResponseAggregator(FrameProcessor):
         if len(self._aggregation) > 0:
             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()
+            # Reset the aggregation. Reset it before pushing it down, otherwise
+            # if the tasks gets cancelled we won't be able to clear things up.
+            self._aggregation = ""
 
             await self.push_frame(frame)