diff --git a/CHANGELOG.md b/CHANGELOG.md index bae0c5361..8c2cf3aee 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index e01fc9e02..8739c8ab3 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/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) diff --git a/src/pipecat/processors/aggregators/user_response.py b/src/pipecat/processors/aggregators/user_response.py index a30e75b69..5ce62b859 100644 --- a/src/pipecat/processors/aggregators/user_response.py +++ b/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)