diff --git a/litellm/llms/bedrock/realtime/transformation.py b/litellm/llms/bedrock/realtime/transformation.py index 03228fc43ef..8fef84bae29 100644 --- a/litellm/llms/bedrock/realtime/transformation.py +++ b/litellm/llms/bedrock/realtime/transformation.py @@ -118,6 +118,7 @@ class BedrockRealtimeConfig(BaseRealtimeConfig): # so the USER/ASSISTANT split from contentStart is tracked here) self._user_transcript_active = False self._user_transcript_generation_stage: str | None = None + self._assistant_final_text_active = False self._user_item_id: str | None = None self._user_transcript_buffer = "" self._cumulative_usage = BedrockUsageEvent() @@ -748,6 +749,20 @@ class BedrockRealtimeConfig(BaseRealtimeConfig): None, ) + generation_stage: Final = self._parse_generation_stage(content_start.get("additionalModelFields")) + if content_start.get("type") == "TEXT" and generation_stage == "FINAL": + verbose_logger.debug( + "Dropping FINAL ASSISTANT text block: the SPECULATIVE block already carried its sentences" + ) + self._assistant_final_text_active = True + return ( + [], + current_response_id, + current_output_item_id, + current_conversation_id, + None, + ) + verbose_logger.debug("Handling ASSISTANT contentStart") is_new_response: Final = current_response_id is None @@ -943,6 +958,8 @@ class BedrockRealtimeConfig(BaseRealtimeConfig): verbose_logger.debug("Handling textOutput") text_content: Final = event["textOutput"].get("content", "") + if self._assistant_final_text_active: + return [], current_delta_chunks if not current_output_item_id or not current_response_id: return [], current_delta_chunks @@ -1022,6 +1039,9 @@ class BedrockRealtimeConfig(BaseRealtimeConfig): content_end: Final = event["contentEnd"] verbose_logger.debug("Handling contentEnd: %s", content_end) + if self._assistant_final_text_active: + self._assistant_final_text_active = False + return [], current_delta_chunks if not current_output_item_id or not current_response_id: return [], current_delta_chunks diff --git a/tests/e2e/llm_translation/realtime/realtime_client.py b/tests/e2e/llm_translation/realtime/realtime_client.py index 7da7ceac4d3..a280f4bc26b 100644 --- a/tests/e2e/llm_translation/realtime/realtime_client.py +++ b/tests/e2e/llm_translation/realtime/realtime_client.py @@ -11,10 +11,11 @@ models, matching the suite's no-raw-dicts rule. from __future__ import annotations import time -from collections.abc import Generator, Mapping +from collections.abc import Generator, Iterator, Mapping, Sequence from contextlib import contextmanager from dataclasses import dataclass -from typing import TypeVar +from itertools import chain, repeat +from typing import Final, TypeVar from urllib.parse import urlencode from pydantic import BaseModel, ConfigDict @@ -22,6 +23,7 @@ from websockets.sync.client import connect from websockets.sync.connection import Connection from e2e_config import unique_marker, ws_base_url +from e2e_metadata import step from proxy_client import ProxyClient from models import LiteLLMParamsBody @@ -164,6 +166,11 @@ class ResponseCreate(BaseModel): type: str = "response.create" +class InputAudioBufferAppend(BaseModel): + type: str = "input_audio_buffer.append" + audio: str + + def user_message(text: str) -> ConversationItemCreate: return ConversationItemCreate( item=MessageItem(content=[InputTextContent(text=text)]) @@ -214,6 +221,12 @@ class ResponsePayload(BaseModel): output: list[OutputItem] | None = None +class TextDone(BaseModel): + type: str + response_id: str + text: str + + class ResponseDone(BaseModel): type: str response: ResponsePayload @@ -320,6 +333,50 @@ class RealtimeSession: ) + @step("Stream the spoken question as mic chunks, then silence until the turn goes idle") + def stream_and_collect( + self, + chunks: Sequence[InputAudioBufferAppend], + *, + tail: InputAudioBufferAppend, + interval: float, + idle: float, + timeout: float, + ) -> tuple[ReceivedEvent, ...]: + """Send `chunks`, then `tail` every `interval` seconds like a live mic, until + no event arrives for `idle` seconds or `timeout` elapses.""" + return tuple(self._live_mic_events(chunks, tail=tail, interval=interval, idle=idle, timeout=timeout)) + + def _live_mic_events( + self, + chunks: Sequence[InputAudioBufferAppend], + *, + tail: InputAudioBufferAppend, + interval: float, + idle: float, + timeout: float, + ) -> Iterator[ReceivedEvent]: + start: Final = time.monotonic() + last_event = start # rebind-ok: the idle timer restarts at every received event + for frame in chain(chunks, repeat(tail)): + if (now := time.monotonic()) - start >= timeout or now - last_event >= idle: + return + self.send(frame) + for event in self._events_until(now + interval): + last_event = time.monotonic() + yield event + + def _events_until(self, deadline: float) -> Iterator[ReceivedEvent]: + while (remaining := deadline - time.monotonic()) > 0 and (text := self._recv_text(remaining)) is not None: + yield ReceivedEvent(type=ServerEnvelope.model_validate_json(text).type, payload=text) + + def _recv_text(self, timeout: float) -> str | None: + try: + return as_text(self.connection.recv(timeout=timeout)) + except TimeoutError: + return None + + @dataclass(frozen=True, slots=True) class RealtimeClient: proxy: ProxyClient diff --git a/tests/e2e/llm_translation/realtime/test_realtime_bedrock_duplicate_text_e2e.py b/tests/e2e/llm_translation/realtime/test_realtime_bedrock_duplicate_text_e2e.py new file mode 100644 index 00000000000..7a70586b140 --- /dev/null +++ b/tests/e2e/llm_translation/realtime/test_realtime_bedrock_duplicate_text_e2e.py @@ -0,0 +1,134 @@ +"""Live e2e: Bedrock Nova 2 Sonic realtime shows each assistant sentence once. + +User flow: a voice client on /v1/realtime streams mic audio +through input_audio_buffer.append, keeps streaming after the user stops talking, +and renders every response.text.done as a chat bubble. Nova 2 Sonic sends each +assistant text block twice, a SPECULATIVE preview next to the audio and then the +FINAL transcript once the audio turn has ended, tagging each block with +generationStage in its contentStart. The test streams a spoken question the same +way, reads the whole turn including events after the first response.done, and +asserts no assistant sentence reaches the client in more than one +response.text.done. + +NOVA_SONIC pins a vendor-owned model id; see test_realtime_bedrock_e2e.py for +how to re-check it before concluding litellm broke. +""" + +from __future__ import annotations + +import array +import base64 +import wave +from collections import Counter +from pathlib import Path +from typing import Final + +import pytest + +from e2e_config import unique_marker +from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta +from lifecycle import ResourceManager +from models import LiteLLMParamsBody +from realtime_client import ( + InputAudioBufferAppend, + RealtimeClient, + SessionConfig, + SessionUpdate, + TextDone, + events_of_type, +) + +pytestmark = pytest.mark.e2e + +NOVA_SONIC = "bedrock/amazon.nova-2-sonic-v1:0" +QUESTION_WAV = Path(__file__).parent / "fixtures" / "weather_question_24k.wav" +NOVA_INPUT_RATE = 16000 +CHUNK_SECONDS = 0.1 + + +def pcm16_at_nova_input_rate(path: Path) -> bytes: + """Mono PCM16 samples from `path`, linearly resampled to Nova Sonic's 16 kHz input rate.""" + with wave.open(str(path), "rb") as wav: + assert (wav.getnchannels(), wav.getsampwidth()) == (1, 2), "fixture must be mono PCM16" + ratio: Final = wav.getframerate() / NOVA_INPUT_RATE + source: Final = array.array("h", wav.readframes(wav.getnframes())) + positions: Final = (i * ratio for i in range(int(len(source) / ratio) - 1)) + return array.array("h", (_interpolated_sample(source, position) for position in positions)).tobytes() + + +def _interpolated_sample(source: array.array[int], position: float) -> int: + base: Final = int(position) + fraction: Final = position - base + return round(source[base] * (1 - fraction) + source[base + 1] * fraction) + + +def mic_chunks(pcm: bytes) -> list[InputAudioBufferAppend]: + size = int(NOVA_INPUT_RATE * CHUNK_SECONDS) * 2 + return [ + InputAudioBufferAppend(audio=base64.b64encode(pcm[i : i + size]).decode()) for i in range(0, len(pcm), size) + ] + + +class TestNovaSonicAssistantText: + @pytest.mark.covers( + "llm.realtime.bedrock_converse.basic.stream.works", + exercised_on=["realtime"], + ) + @meta( + Subject( + domain=Domain.LLM_TRANSLATION, + route=Route.REALTIME, + providers=(Provider.BEDROCK,), + models=(NOVA_SONIC,), + capabilities=(Capability.AUDIO_INPUT, Capability.AUDIO_OUTPUT), + mode=Mode.WEBSOCKET, + ) + ) + def test_nova_sonic_voice_turn_shows_each_sentence_once( + self, client: RealtimeClient, resources: ResourceManager, scoped_key: str + ) -> None: + model: Final = f"e2e-nova-sonic-dupe-{unique_marker()}" + model_id: Final = client.proxy.create_model( + model, + LiteLLMParamsBody( + model=NOVA_SONIC, + aws_access_key_id="os.environ/AWS_ACCESS_KEY_ID", + aws_secret_access_key="os.environ/AWS_SECRET_ACCESS_KEY", + aws_region_name="os.environ/AWS_REGION", + ), + mode="realtime", + ) + resources.defer(lambda: client.proxy.delete_model(model_id)) + silence: Final = InputAudioBufferAppend( + audio=base64.b64encode(bytes(int(NOVA_INPUT_RATE * CHUNK_SECONDS) * 2)).decode() + ) + + with client.connect(key=scoped_key, model=model) as session: + session.collect_until("session.created", timeout=30) + session.send( + SessionUpdate( + session=SessionConfig( + modalities=["text", "audio"], + instructions="You are a friendly assistant. Answer in two or three short sentences.", + ) + ) + ) + session.collect_until("session.updated", timeout=30) + events: Final = session.stream_and_collect( + mic_chunks(pcm16_at_nova_input_rate(QUESTION_WAV)), + tail=silence, + interval=CHUNK_SECONDS, + idle=15, + timeout=90, + ) + + types: Final = [e.type for e in events] + assert "response.done" in types, f"Nova Sonic never finished a response; types={types}" + spoken: Final = [ + text + for e in events_of_type(events, "response.text.done") + if (text := TextDone.model_validate_json(e.payload).text.strip()) + ] + assert spoken, f"Nova Sonic produced no assistant text; types={types}" + repeated: Final = {t: n for t, n in Counter(spoken).items() if n > 1} + assert not repeated, f"assistant text delivered more than once: {repeated}; all text.done={spoken}" diff --git a/tests/unit/llms/bedrock/realtime/test_bedrock_realtime_transformation.py b/tests/unit/llms/bedrock/realtime/test_bedrock_realtime_transformation.py index a74f03449a1..9d034c9838c 100644 --- a/tests/unit/llms/bedrock/realtime/test_bedrock_realtime_transformation.py +++ b/tests/unit/llms/bedrock/realtime/test_bedrock_realtime_transformation.py @@ -1,4 +1,5 @@ import json +from typing import Final from unittest.mock import MagicMock import pytest @@ -828,8 +829,9 @@ class TestBedrockRealtimeSessionEvents: class TestBedrockRealtimeUserEventsAndUsage: - """Regression tests for #38346: USER ASR transcripts, speech boundary events, - usage propagation, and duplicate response.created""" + """Regression tests for #38346 (USER ASR transcripts, speech boundary events, + usage propagation, duplicate response.created) and for FINAL assistant text + blocks replaying every sentence after the audio turn ended""" @staticmethod def _run(config, messages): @@ -1131,6 +1133,84 @@ class TestBedrockRealtimeUserEventsAndUsage: ) assert [e["type"] for e in events] == ["conversation.item.input_audio_transcription.delta"] + @staticmethod + def _assistant_text_block(stage: str, text: str, stop_reason: str) -> list[dict[str, object]]: + return [ + { + "event": { + "contentStart": { + "role": "ASSISTANT", + "type": "TEXT", + "additionalModelFields": json.dumps({"generationStage": stage}), + } + } + }, + {"event": {"textOutput": {"content": text}}}, + {"event": {"contentEnd": {"stopReason": stop_reason}}}, + ] + + @staticmethod + def _assistant_audio_block(stop_reason: str) -> list[dict[str, object]]: + return [ + {"event": {"contentStart": {"role": "ASSISTANT", "type": "AUDIO"}}}, + {"event": {"audioOutput": {"content": base64.b64encode(b"\x00\x01" * 8).decode()}}}, + {"event": {"contentEnd": {"stopReason": stop_reason}}}, + ] + + def test_final_assistant_text_blocks_do_not_replay_the_spoken_sentences(self): + sentences: Final = ["I can't check live weather.", " Try a weather app.", " It will have the forecast."] + events: Final = self._run( + BedrockRealtimeConfig(), + self._assistant_text_block("SPECULATIVE", sentences[0], "PARTIAL_TURN") + + self._assistant_audio_block("PARTIAL_TURN") + + self._assistant_text_block("SPECULATIVE", sentences[1], "PARTIAL_TURN") + + self._assistant_audio_block("PARTIAL_TURN") + + self._assistant_text_block("SPECULATIVE", sentences[2], "PARTIAL_TURN") + + self._assistant_audio_block("END_TURN") + + self._assistant_text_block("FINAL", sentences[0], "END_TURN") + + self._assistant_text_block("FINAL", sentences[1], "PARTIAL_TURN") + + self._assistant_text_block("FINAL", sentences[2], "PARTIAL_TURN"), + ) + types: Final = [e["type"] for e in events] + assert [e["text"] for e in events if e["type"] == "response.text.done"] == sentences + assert types.count("response.created") == 1 + assert types.count("response.done") == 1 + assert types[-1] == "response.done" + response_ids: Final = {e["response_id"] for e in events if e["type"] == "response.text.delta"} + assert response_ids == {events[0]["response"]["id"]} + + def test_final_assistant_text_after_end_turn_opens_no_response(self): + events: Final = self._run( + BedrockRealtimeConfig(), + self._assistant_text_block("FINAL", "Already spoken.", "END_TURN"), + ) + assert events == [] + + def test_speculative_text_after_a_final_block_is_forwarded(self): + config: Final = BedrockRealtimeConfig() + self._run(config, self._assistant_text_block("FINAL", "Already spoken.", "END_TURN")) + events: Final = self._run(config, self._assistant_text_block("SPECULATIVE", "Next turn.", "PARTIAL_TURN")) + assert [e["type"] for e in events][:1] == ["response.created"] + assert [e["text"] for e in events if e["type"] == "response.text.done"] == ["Next turn."] + + def test_barge_in_marker_in_final_block_is_dropped_and_the_next_answer_still_flows(self): + events: Final = self._run( + BedrockRealtimeConfig(), + self._assistant_text_block("SPECULATIVE", "I can't check live weather.", "PARTIAL_TURN") + + self._assistant_audio_block("INTERRUPTED") + + self._assistant_text_block("FINAL", '{ "interrupted" : true }', "INTERRUPTED") + + self._assistant_text_block("FINAL", "I can't check live weather.", "INTERRUPTED") + + self._assistant_text_block("SPECULATIVE", "Still no live weather.", "PARTIAL_TURN") + + self._assistant_audio_block("END_TURN"), + ) + types: Final = [e["type"] for e in events] + assert [e["text"] for e in events if e["type"] == "response.text.done"] == [ + "I can't check live weather.", + "Still no live weather.", + ] + assert types.count("response.created") == 1 + assert types.count("response.done") == 1 + if __name__ == "__main__": pytest.main([__file__, "-v"])