From b9d996ff41826d9cbed91bcf0e90c80212d10fa9 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 18 Mar 2026 16:04:12 -0400 Subject: [PATCH] Improvements for Nova Sonic LLM and TTS output frames (#4042) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Fix empty user transcription causing spurious interruption in Nova Sonic Skip _report_user_transcription_ended() when _user_text_buffer is empty, which happens when the initial prompt is text-only. Previously, an empty TranscriptionFrame was pushed upstream, triggering a chain reaction: on_user_turn_stopped → UserStartedSpeakingFrame → interruption → premature BotStoppedSpeaking → multiple response start/stop cycles. * Improve TextFrame and assistant end of turn logic Now, SPECULATIVE text results are used to push the LLMTextFrame, AggregatedTextFrame, and TTSTextFrame. Additionally, the TTSTextFrames are push at the end of the corresponding audio segment. * Remove BotStoppedSpeakingFrame fallback from Nova Sonic Now that assistant response end is detected directly from Nova Sonic contentEnd events (END_TURN and INTERRUPTED), the BotStoppedSpeakingFrame handler is no longer needed. Inline the cleanup logic in reset_conversation. --- changelog/4042.changed.md | 1 + changelog/4042.fixed.md | 1 + src/pipecat/services/aws/nova_sonic/llm.py | 185 +++++++-------------- 3 files changed, 66 insertions(+), 121 deletions(-) create mode 100644 changelog/4042.changed.md create mode 100644 changelog/4042.fixed.md diff --git a/changelog/4042.changed.md b/changelog/4042.changed.md new file mode 100644 index 000000000..f83c32f59 --- /dev/null +++ b/changelog/4042.changed.md @@ -0,0 +1 @@ +- Nova Sonic assistant text transcripts are now delivered in real-time using speculative text events instead of delayed final text events. Previously, assistant text only arrived after all audio had finished playing, causing laggy transcripts in client UIs. Speculative text arrives before each audio chunk, providing text synchronized with what the bot is saying. This also simplifies the internal text handling by removing the interruption re-push hack and assistant text buffer. diff --git a/changelog/4042.fixed.md b/changelog/4042.fixed.md new file mode 100644 index 000000000..a48996208 --- /dev/null +++ b/changelog/4042.fixed.md @@ -0,0 +1 @@ +- Fixed empty user transcriptions in Nova Sonic causing spurious interruptions. Previously, an empty transcription could trigger an interruption of the assistant's response even though the user hadn't actually spoken. diff --git a/src/pipecat/services/aws/nova_sonic/llm.py b/src/pipecat/services/aws/nova_sonic/llm.py index 0947ba1d7..ffcbb5e5d 100644 --- a/src/pipecat/services/aws/nova_sonic/llm.py +++ b/src/pipecat/services/aws/nova_sonic/llm.py @@ -27,8 +27,8 @@ from pydantic import BaseModel, Field from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.adapters.services.aws_nova_sonic_adapter import AWSNovaSonicLLMAdapter, Role from pipecat.frames.frames import ( + AggregatedTextFrame, AggregationType, - BotStoppedSpeakingFrame, CancelFrame, EndFrame, Frame, @@ -424,18 +424,16 @@ class AWSNovaSonicLLMService(LLMService): self._input_audio_content_name: Optional[str] = None self._content_being_received: Optional[CurrentContent] = None self._assistant_is_responding = False - self._may_need_repush_assistant_text = False self._ready_to_send_context = False - self._handling_bot_stopped_speaking = False self._triggering_assistant_response = False self._waiting_for_trigger_transcription = False self._disconnecting = False self._connected_time: Optional[float] = None self._wants_connection = False self._user_text_buffer = "" - self._assistant_text_buffer = "" self._completed_tool_calls = set() self._audio_input_started = False + self._pending_speculative_text: Optional[str] = None file_path = files("pipecat.services.aws.nova_sonic").joinpath("ready.wav") with wave.open(file_path.open("rb"), "rb") as wav_file: @@ -505,11 +503,13 @@ class AWSNovaSonicLLMService(LLMService): async def reset_conversation(self): """Reset the conversation state while preserving context. - Handles bot stopped speaking event, disconnects from the service, - and reconnects with the preserved context. + Cleans up any in-progress assistant response, disconnects from the + service, and reconnects with the preserved context. """ logger.debug("Resetting conversation") - await self._handle_bot_stopped_speaking(delay_to_catch_trailing_assistant_text=False) + if self._assistant_is_responding: + self._assistant_is_responding = False + await self._report_assistant_response_ended() # Grab context to carry through disconnect/reconnect context = self._context @@ -540,8 +540,6 @@ class AWSNovaSonicLLMService(LLMService): await self._handle_context(context) elif isinstance(frame, InputAudioRawFrame): await self._handle_input_audio_frame(frame) - elif isinstance(frame, BotStoppedSpeakingFrame): - await self._handle_bot_stopped_speaking(delay_to_catch_trailing_assistant_text=True) elif isinstance(frame, InterruptionFrame): await self._handle_interruption_frame() @@ -569,49 +567,8 @@ class AWSNovaSonicLLMService(LLMService): await self._send_user_audio_event(frame.audio) - 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_interruption_frame(self): - if self._assistant_is_responding: - self._may_need_repush_assistant_text = True + pass # # LLM communication: lifecycle @@ -771,17 +728,15 @@ class AWSNovaSonicLLMService(LLMService): self._input_audio_content_name = None self._content_being_received = None self._assistant_is_responding = False - self._may_need_repush_assistant_text = False self._ready_to_send_context = False - self._handling_bot_stopped_speaking = False self._triggering_assistant_response = False self._waiting_for_trigger_transcription = False self._disconnecting = False self._connected_time = None self._user_text_buffer = "" - self._assistant_text_buffer = "" self._completed_tool_calls = set() self._audio_input_started = False + self._pending_speculative_text = None logger.info("Finished disconnecting") except Exception as e: @@ -1153,10 +1108,11 @@ class AWSNovaSonicLLMService(LLMService): self._content_being_received = content if content.role == Role.ASSISTANT: - if content.type == ContentType.AUDIO: - # Note that an assistant response can comprise of multiple audio blocks - if not self._assistant_is_responding: - # The assistant has started responding. + if content.type == ContentType.TEXT: + if ( + content.text_stage == TextStage.SPECULATIVE + and not self._assistant_is_responding + ): self._assistant_is_responding = True await self._report_user_transcription_ended() # Consider user turn over await self._report_assistant_response_started() @@ -1232,18 +1188,30 @@ class AWSNovaSonicLLMService(LLMService): if content.role == Role.ASSISTANT: if content.type == ContentType.TEXT: - # Ignore non-final text, and the "interrupted" message (which isn't meaningful text) - if content.text_stage == TextStage.FINAL and stop_reason != "INTERRUPTED": - if self._assistant_is_responding: - # Text added to the ongoing assistant response - await self._report_assistant_response_text_added(content.text_content) + if stop_reason != "INTERRUPTED": + if content.text_stage == TextStage.SPECULATIVE: + await self._report_llm_text(content.text_content) + elif self._assistant_is_responding: + # TEXT INTERRUPTED with no audio means the user interrupted + # before audio started. End the response here since no AUDIO + # contentEnd will arrive. + self._assistant_is_responding = False + await self._report_assistant_response_ended() + elif content.type == ContentType.AUDIO: + # Emit deferred TTSTextFrame after all audio chunks have been sent + await self._report_tts_text() + if stop_reason in ("END_TURN", "INTERRUPTED"): + # END_TURN: normal completion. INTERRUPTED: user interrupted + # mid-audio. Both mean no more audio for this turn. + self._assistant_is_responding = False + await self._report_assistant_response_ended() elif content.role == Role.USER: if content.type == ContentType.TEXT: if content.text_stage == TextStage.FINAL: # User transcription text added await self._report_user_transcription_text_added(content.text_content) - async def _handle_completion_end_event(self, event_json): + async def _handle_completion_end_event(self, _): pass # @@ -1256,29 +1224,40 @@ class AWSNovaSonicLLMService(LLMService): async def _report_assistant_response_started(self): logger.debug("Assistant response started") - - # Report the start of the assistant response. await self.push_frame(LLMFullResponseStartFrame()) # Report that equivalent of TTS (this is a speech-to-speech model) started await self.push_frame(TTSStartedFrame()) - async def _report_assistant_response_text_added(self, text): - if not self._context: # should never happen - return + async def _report_llm_text(self, text): + """Push speculative assistant text and defer TTSTextFrame. - logger.debug(f"Assistant response text added: {text}") + Speculative text arrives before each audio chunk, providing real-time + text that is synchronized with what the bot is saying. LLMTextFrame and + AggregatedTextFrame are pushed immediately for real-time text display. + TTSTextFrame emission is deferred to audio contentEnd so it aligns with + audio playout timing. + """ + logger.debug(f"Assistant speculative text: {text}") - # Report the text of the assistant response. - await self._push_assistant_response_text_frames(text) + llm_text_frame = LLMTextFrame(text) + llm_text_frame.append_to_context = False + await self.push_frame(llm_text_frame) - # HACK: here we're also buffering the assistant text ourselves as a - # backup rather than relying solely on the assistant context aggregator - # to do it, because the text arrives from Nova Sonic only after all the - # assistant audio frames have been pushed, meaning that if an - # interruption frame were to arrive we would lose all of it (the text - # frames sitting in the queue would be wiped). - self._assistant_text_buffer += text + aggregated_text_frame = AggregatedTextFrame(text, aggregated_by=AggregationType.SENTENCE) + aggregated_text_frame.append_to_context = False + await self.push_frame(aggregated_text_frame) + + self._pending_speculative_text = text + + async def _report_tts_text(self): + if self._pending_speculative_text: + tts_text_frame = TTSTextFrame( + self._pending_speculative_text, aggregated_by=AggregationType.SENTENCE + ) + tts_text_frame.includes_inter_frame_spaces = True + await self.push_frame(tts_text_frame) + self._pending_speculative_text = None async def _report_assistant_response_ended(self): if not self._context: # should never happen @@ -1286,54 +1265,12 @@ class AWSNovaSonicLLMService(LLMService): logger.debug("Assistant response ended") - # If an interruption frame arrived while the assistant was responding - # we may have lost all of the assistant text (see HACK, above), so - # re-push it downstream to the aggregator now. - if self._may_need_repush_assistant_text: - # Just in case, check that assistant text hasn't already made it - # into the context (sometimes it does, despite the interruption). - messages = self._context.get_messages() - last_message = messages[-1] if messages else None - if ( - not last_message - or last_message.get("role") != "assistant" - or last_message.get("content") != self._assistant_text_buffer - ): - # We also need to re-push the LLMFullResponseStartFrame since the - # TTSTextFrame would be ignored otherwise (the interruption frame - # would have cleared the assistant aggregator state). - await self.push_frame(LLMFullResponseStartFrame()) - await self._push_assistant_response_text_frames(self._assistant_text_buffer) - self._may_need_repush_assistant_text = False - # Report the end of the assistant response. await self.push_frame(LLMFullResponseEndFrame()) # Report that equivalent of TTS (this is a speech-to-speech model) stopped. await self.push_frame(TTSStoppedFrame()) - # Clear out the buffered assistant text - self._assistant_text_buffer = "" - - async def _push_assistant_response_text_frames(self, text: str): - # In a typical "cascade" LLM + TTS setup, LLMTextFrames would not - # proceed beyond the TTS service. Therefore, since a speech-to-speech - # service like Nova Sonic combines both LLM and TTS functionality, you - # would think we wouldn't need to push LLMTextFrames at all. However, - # RTVI relies on LLMTextFrames being pushed to trigger its - # "bot-llm-text" event. So here we push an LLMTextFrame, too, but avoid - # appending it to context to avoid context message duplication. - - # Push LLMTextFrame - llm_text_frame = LLMTextFrame(text) - llm_text_frame.append_to_context = False - await self.push_frame(llm_text_frame) - - # Push TTSTextFrame - tts_text_frame = TTSTextFrame(text, aggregated_by=AggregationType.SENTENCE) - tts_text_frame.includes_inter_frame_spaces = True - await self.push_frame(tts_text_frame) - # # user transcription reporting # @@ -1363,6 +1300,12 @@ class AWSNovaSonicLLMService(LLMService): if not self._context: # should never happen return + # Nothing to report if no user speech was transcribed (e.g. the prompt + # was text-only, which is the case on the first user turn when the bot + # starts the conversation). + if not self._user_text_buffer: + return + logger.debug(f"User transcription ended") # Report to the upstream user context aggregator that some new user