From 84d040c6d0aed545091c0884611a2b5e13611e16 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 7 May 2025 16:21:47 -0400 Subject: [PATCH] AWS Nova Sonic service - make interruption handling more reliable, in terms of: - not getting the conversation into a "stuck" state - not losing assistant text that should've made it into the context --- src/pipecat/services/aws_nova_sonic/aws.py | 64 +++++++++++++++------- 1 file changed, 43 insertions(+), 21 deletions(-) diff --git a/src/pipecat/services/aws_nova_sonic/aws.py b/src/pipecat/services/aws_nova_sonic/aws.py index b53578f5a..410481065 100644 --- a/src/pipecat/services/aws_nova_sonic/aws.py +++ b/src/pipecat/services/aws_nova_sonic/aws.py @@ -172,6 +172,7 @@ class AWSNovaSonicLLMService(LLMService): self._content_being_received: Optional[CurrentContent] = None self._assistant_is_responding = False self._ready_to_send_context = False + self._handling_bot_stopped_speaking = False self._triggering_assistant_response = False self._assistant_response_trigger_audio: Optional[bytes] = ( None # Not cleared on _disconnect() @@ -205,7 +206,7 @@ class AWSNovaSonicLLMService(LLMService): async def reset_conversation(self): logger.debug("Resetting conversation") - await self._handle_bot_stopped_speaking() + await self._handle_bot_stopped_speaking(delay_to_catch_trailing_assistant_text=False) # Carry over previous context through disconnect context = self._context @@ -226,7 +227,7 @@ class AWSNovaSonicLLMService(LLMService): elif isinstance(frame, InputAudioRawFrame): await self._handle_input_audio_frame(frame) elif isinstance(frame, BotStoppedSpeakingFrame): - await self._handle_bot_stopped_speaking() + await self._handle_bot_stopped_speaking(delay_to_catch_trailing_assistant_text=True) elif isinstance(frame, AWSNovaSonicFunctionCallResultFrame): await self._handle_function_call_result(frame) @@ -248,25 +249,45 @@ class AWSNovaSonicLLMService(LLMService): await self._send_user_audio_event(frame.audio) - async def _handle_bot_stopped_speaking(self): - if self._assistant_is_responding: - # Consider the assistant finished with their response (after a short delay, to allow for - # any FINAL text block to come in). - # - # TODO: ideally we could base this solely on the LLM output events, but I couldn't - # figure out a reliable way to determine when we've gotten our last FINAL text block - # after the LLM is done talking. - # - # First I looked at stopReason, but it doesn't seem like the last FINAL text block is - # reliably marked END_TURN (sometimes the *first* one is, but not the last...bug?) - # - # Then I considered schemes where we tally or match up SPECULATIVE text blocks with - # FINAL text blocks to know how many or which FINAL blocks to expect, but user - # interruptions throw a wrench in these schemes: depending on the exact timing of the - # interruption, we should or shouldn't expect some FINAL blocks. - await asyncio.sleep(0.25) - self._assistant_is_responding = False - await self._report_assistant_response_ended() + async def _handle_bot_stopped_speaking(self, delay_to_catch_trailing_assistant_text: bool): + # Protect against back-to-back BotStoppedSpeaking calls, which I've observed + if self._handling_bot_stopped_speaking: + return + self._handling_bot_stopped_speaking = True + + async def finalize_assistant_response(): + if self._assistant_is_responding: + # Consider the assistant finished with their response (possibly after a short delay, + # to allow for any trailing FINAL assistant text block to come in that need to make + # it into context). + # + # TODO: ideally we could base this solely on the LLM output events, but I couldn't + # figure out a reliable way to determine when we've gotten our last FINAL text block + # after the LLM is done talking. + # + # First I looked at stopReason, but it doesn't seem like the last FINAL text block + # is reliably marked END_TURN (sometimes the *first* one is, but not the last... + # bug?) + # + # Then I considered schemes where we tally or match up SPECULATIVE text blocks with + # FINAL text blocks to know how many or which FINAL blocks to expect, but user + # interruptions throw a wrench in these schemes: depending on the exact timing of + # the interruption, we should or shouldn't expect some FINAL blocks. + if delay_to_catch_trailing_assistant_text: + # This delay length is a balancing act between "catching" trailing assistant + # text that is quite delayed but not waiting so long that user text comes in + # first and results in a bit of context message order scrambling. + await asyncio.sleep(1.25) + self._assistant_is_responding = False + await self._report_assistant_response_ended() + + self._handling_bot_stopped_speaking = False + + # Finalize the assistant response, either now or after a delay + if delay_to_catch_trailing_assistant_text: + self.create_task(finalize_assistant_response()) + else: + await finalize_assistant_response() async def _handle_function_call_result(self, frame: AWSNovaSonicFunctionCallResultFrame): result = frame.result_frame @@ -391,6 +412,7 @@ class AWSNovaSonicLLMService(LLMService): self._content_being_received = None self._assistant_is_responding = False self._ready_to_send_context = False + self._handling_bot_stopped_speaking = False self._triggering_assistant_response = False self._disconnecting = False self._connected_time = None