Handling the case where user stopped speaking but no new aggregation received.

src/pipecat/processors/aggregators/llm_response.py

@@ -266,6 +266,7 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
 
         self._user_speaking = False
         self._bot_speaking = False
+        self._was_bot_speaking = False
         self._emulating_vad = False
         self._seen_interim_results = False
         self._waiting_for_aggregation = False
@@ -275,6 +276,7 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
 
     async def reset(self):
         await super().reset()
+        self._was_bot_speaking = False
         self._seen_interim_results = False
         self._waiting_for_aggregation = False
         [await s.reset() for s in self._interruption_strategies]
@@ -355,6 +357,20 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
             else:
                 # No interruption config - normal behavior (always push aggregation)
                 await self._process_aggregation()
+        # Handles the case where both the user and the bot are not speaking,
+        # and the bot was previously speaking before the user interruption.
+        # Normally, when the user stops speaking, new text is expected,
+        # which triggers the bot to respond. However, if no new text
+        # is received, this safeguard ensures
+        # the bot doesn't hang indefinitely while waiting to speak again.
+        elif not self._seen_interim_results and self._was_bot_speaking and not self._bot_speaking:
+            logger.warning(
+                "User stopped speaking but no new aggregation received. Forcing aggregation processing to resume bot response."
+            )
+            # Resetting it so we don't trigger this twice
+            self._was_bot_speaking = False
+            # We are just pushing the same previous context to be processed again in this case
+            await self.push_frame(OpenAILLMContextFrame(self._context))
 
     async def _should_interrupt_based_on_strategies(self) -> bool:
         """Check if interruption should occur based on configured strategies."""
@@ -381,6 +397,7 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
     async def _handle_user_started_speaking(self, frame: UserStartedSpeakingFrame):
         self._user_speaking = True
         self._waiting_for_aggregation = True
+        self._was_bot_speaking = self._bot_speaking
 
         # 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
@@ -393,7 +410,7 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
         # We just stopped speaking. Let's see if there's some aggregation to
         # push. If the last thing we saw is an interim transcription, let's wait
         # pushing the aggregation as we will probably get a final transcription.
-        if not self._seen_interim_results:
+        if len(self._aggregation) > 0 and not self._seen_interim_results:
             await self.push_aggregation()
 
     async def _handle_bot_started_speaking(self, _: BotStartedSpeakingFrame):