wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improvements for Nova Sonic LLM and TTS output frames (#4042)

Browse Source 
* 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.
This commit is contained in:
Mark Backman
2026-03-18 16:04:12 -04:00
parent 5de4256ab1
commit b9d996ff41
3 changed files with 66 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog/4042.changed.md Normal file
View File

@@ -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.

1
changelog/4042.fixed.md Normal file
View File

@@ -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.

185
src/pipecat/services/aws/nova_sonic/llm.py
View File

@@ -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