fix(bedrock): surface Nova Sonic user transcripts, speech events, and usage in realtime API

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Devin AI 2026-08-28 00:07:40 +00:00
parent 49affa7c01
commit b7a7754b05
5 changed files with 383 additions and 18 deletions

View file

@ -7,7 +7,7 @@ Transforms between OpenAI Realtime API format and Bedrock Nova Sonic format.
import base64
import json
import uuid as uuid_lib
from typing import Any, Final
from typing import Any, Final, cast
from pydantic import BaseModel
@ -20,16 +20,21 @@ from litellm.types.llms.openai import (
OpenAIRealtimeContentPartDone,
OpenAIRealtimeDoneEvent,
OpenAIRealtimeEvents,
OpenAIRealtimeInputAudioBufferSpeechEvent,
OpenAIRealtimeInputAudioTranscriptionCompleted,
OpenAIRealtimeInputAudioTranscriptionDelta,
OpenAIRealtimeOutputItemDone,
OpenAIRealtimeResponseAudioDone,
OpenAIRealtimeResponseContentPartAdded,
OpenAIRealtimeResponseDelta,
OpenAIRealtimeResponseDoneObject,
OpenAIRealtimeResponseTextDone,
OpenAIRealtimeResponseUsage,
OpenAIRealtimeStreamResponseBaseObject,
OpenAIRealtimeStreamResponseOutputItemAdded,
OpenAIRealtimeStreamSession,
OpenAIRealtimeStreamSessionEvents,
OpenAIRealtimeUsageTokenDetails,
)
from litellm.types.realtime import (
ALL_DELTA_TYPES,
@ -43,6 +48,27 @@ class BedrockContentEnd(BaseModel):
stopReason: str | None = None
class BedrockUsageTokenDetails(BaseModel):
speechTokens: int = 0
textTokens: int = 0
class BedrockUsageDetailsTotal(BaseModel):
input: BedrockUsageTokenDetails = BedrockUsageTokenDetails()
output: BedrockUsageTokenDetails = BedrockUsageTokenDetails()
class BedrockUsageDetails(BaseModel):
total: BedrockUsageDetailsTotal = BedrockUsageDetailsTotal()
class BedrockUsageEvent(BaseModel):
totalInputTokens: int = 0
totalOutputTokens: int = 0
totalTokens: int = 0
details: BedrockUsageDetails = BedrockUsageDetails()
TRIGGER_AUDIO_SAMPLE_RATE_HERTZ: Final = 16000
TRIGGER_AUDIO_BYTES_PER_SECOND: Final = TRIGGER_AUDIO_SAMPLE_RATE_HERTZ * 2
TRIGGER_LEADING_SILENCE: Final = bytes(TRIGGER_AUDIO_BYTES_PER_SECOND // 2)
@ -87,6 +113,12 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
# Text configuration
self.text_media_type = "text/plain"
# Response-stream state (Bedrock events carry no role on textOutput,
# so the USER/ASSISTANT split from contentStart is tracked here)
self._user_transcript_active = False
self._user_transcript_generation_stage: str | None = None
self._latest_usage: OpenAIRealtimeResponseUsage | None = None
def validate_environment(self, headers: dict, model: str, api_key: str | None = None) -> dict:
"""Validate environment - no special validation needed for Bedrock."""
return headers
@ -691,6 +723,11 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
role: Final = content_start.get("role")
if role != "ASSISTANT":
if role == "USER" and content_start.get("type") == "TEXT":
self._user_transcript_active = True
self._user_transcript_generation_stage = self._parse_generation_stage(
content_start.get("additionalModelFields")
)
return (
[],
current_response_id,
@ -700,6 +737,7 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
)
verbose_logger.debug("Handling ASSISTANT contentStart")
is_new_response: Final = current_response_id is None
# Initialize IDs if needed
if not current_response_id:
@ -715,7 +753,8 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
returned_messages: Final[list[OpenAIRealtimeEvents]] = []
# Send response.created
# Send response.created only once per response (a response can contain
# multiple content blocks, e.g. TEXT then AUDIO)
response_created: Final = OpenAIRealtimeStreamResponseBaseObject(
type="response.created",
event_id=f"event_{uuid.uuid4()}",
@ -727,7 +766,8 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
"conversation_id": current_conversation_id,
},
)
returned_messages.append(response_created)
if is_new_response:
returned_messages.append(response_created)
# Send response.output_item.added
output_item_added: Final = OpenAIRealtimeStreamResponseOutputItemAdded(
@ -767,6 +807,70 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
current_delta_type,
)
@staticmethod
def _parse_generation_stage(additional_model_fields: object) -> str | None:
if not isinstance(additional_model_fields, str):
return None
try:
parsed: Final = json.loads(additional_model_fields)
except json.JSONDecodeError:
return None
stage: Final = parsed.get("generationStage") if isinstance(parsed, dict) else None
return stage if isinstance(stage, str) else None
def transform_user_speech_event(self, is_speech_start: bool) -> tuple[OpenAIRealtimeEvents, ...]:
"""Transform Bedrock userSpeechStart/userSpeechEnd to OpenAI speech boundary events."""
verbose_logger.debug("Handling userSpeech%s", "Start" if is_speech_start else "End")
speech_event: Final[OpenAIRealtimeInputAudioBufferSpeechEvent] = {
"type": "input_audio_buffer.speech_started" if is_speech_start else "input_audio_buffer.speech_stopped",
"event_id": f"event_{uuid.uuid4()}",
"item_id": f"item_{uuid.uuid4()}",
}
return (speech_event,)
def transform_usage_event(self, usage_event: BedrockUsageEvent) -> None:
"""Record Bedrock usageEvent token totals for the next response.done."""
verbose_logger.debug("Handling usageEvent")
input_details: Final[OpenAIRealtimeUsageTokenDetails] = {
"audio_tokens": usage_event.details.total.input.speechTokens,
"text_tokens": usage_event.details.total.input.textTokens,
"cached_tokens": 0,
}
output_details: Final[OpenAIRealtimeUsageTokenDetails] = {
"audio_tokens": usage_event.details.total.output.speechTokens,
"text_tokens": usage_event.details.total.output.textTokens,
}
latest_usage: Final[OpenAIRealtimeResponseUsage] = {
"input_tokens": usage_event.totalInputTokens,
"output_tokens": usage_event.totalOutputTokens,
"total_tokens": usage_event.totalTokens,
"input_token_details": input_details,
"output_token_details": output_details,
}
self._latest_usage = latest_usage
def transform_user_transcript_event(self, transcript: str) -> tuple[OpenAIRealtimeEvents, ...]:
"""Transform a USER-role Bedrock textOutput (ASR transcript) to OpenAI transcription events."""
verbose_logger.debug("Handling USER textOutput (ASR transcript)")
item_id: Final = f"item_{uuid.uuid4()}"
delta_event: Final[OpenAIRealtimeInputAudioTranscriptionDelta] = {
"type": "conversation.item.input_audio_transcription.delta",
"event_id": f"event_{uuid.uuid4()}",
"item_id": item_id,
"content_index": 0,
"delta": transcript,
}
if self._user_transcript_generation_stage == "SPECULATIVE":
return (delta_event,)
completed_event: Final[OpenAIRealtimeInputAudioTranscriptionCompleted] = {
"type": "conversation.item.input_audio_transcription.completed",
"event_id": f"event_{uuid.uuid4()}",
"item_id": item_id,
"content_index": 0,
"transcript": transcript,
}
return (delta_event, completed_event)
def transform_text_output_event(
self,
event: dict,
@ -985,7 +1089,14 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
if not current_response_id or not current_conversation_id:
return [], None, None, None
usage_obj: Final = get_empty_usage()
empty_usage: Final = get_empty_usage()
zero_usage: Final[OpenAIRealtimeResponseUsage] = {
"input_tokens": empty_usage.prompt_tokens,
"output_tokens": empty_usage.completion_tokens,
"total_tokens": empty_usage.total_tokens,
}
usage: Final = self._latest_usage or zero_usage
self._latest_usage = None
response_done: Final = OpenAIRealtimeDoneEvent(
type="response.done",
event_id=f"event_{uuid.uuid4()}",
@ -995,11 +1106,7 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
status="completed",
output=[],
conversation_id=current_conversation_id,
usage={
"prompt_tokens": usage_obj.prompt_tokens,
"completion_tokens": usage_obj.completion_tokens,
"total_tokens": usage_obj.total_tokens,
},
usage=dict(usage),
),
)
@ -1042,8 +1149,6 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
# Create a function call arguments done event
# This is a custom event format that matches what clients expect
from typing import cast
function_call_event: Final[dict[str, Any]] = {
"type": "response.function_call_arguments.done",
"event_id": f"event_{uuid.uuid4()}",
@ -1194,18 +1299,25 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
returned_messages.extend(events)
elif "textOutput" in event:
events, current_delta_chunks = self.transform_text_output_event(
event,
current_output_item_id,
current_response_id,
current_delta_chunks,
)
returned_messages.extend(events)
if self._user_transcript_active:
returned_messages.extend(self.transform_user_transcript_event(event["textOutput"].get("content", "")))
else:
events, current_delta_chunks = self.transform_text_output_event(
event,
current_output_item_id,
current_response_id,
current_delta_chunks,
)
returned_messages.extend(events)
elif "audioOutput" in event:
events = self.transform_audio_output_event(event, current_output_item_id, current_response_id)
returned_messages.extend(events)
elif "contentEnd" in event and self._user_transcript_active:
self._user_transcript_active = False
self._user_transcript_generation_stage = None
elif "contentEnd" in event:
events, current_delta_chunks = self.transform_content_end_event(
event,
@ -1224,6 +1336,12 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
) = self._response_done_events(current_response_id, current_conversation_id)
returned_messages.extend(done_events)
elif "userSpeechStart" in event or "userSpeechEnd" in event:
returned_messages.extend(self.transform_user_speech_event("userSpeechStart" in event))
elif "usageEvent" in event:
self.transform_usage_event(BedrockUsageEvent.model_validate(event["usageEvent"]))
elif "toolUse" in event:
events, tool_call_id, tool_name = self.transform_tool_use_event(
event, current_output_item_id, current_response_id

View file

@ -553,6 +553,26 @@
"supports_response_schema": true,
"supports_vision": true
},
"amazon.nova-sonic-v1:0": {
"input_cost_per_audio_token": 3.4e-06,
"input_cost_per_token": 6e-08,
"litellm_provider": "bedrock",
"mode": "realtime",
"output_cost_per_audio_token": 1.36e-05,
"output_cost_per_token": 2.4e-07,
"supports_audio_input": true,
"supports_audio_output": true
},
"amazon.nova-2-sonic-v1:0": {
"input_cost_per_audio_token": 3e-06,
"input_cost_per_token": 3.3e-07,
"litellm_provider": "bedrock",
"mode": "realtime",
"output_cost_per_audio_token": 1.2e-05,
"output_cost_per_token": 2.75e-06,
"supports_audio_input": true,
"supports_audio_output": true
},
"amazon.rerank-v1:0": {
"input_cost_per_query": 0.001,
"input_cost_per_token": 0.0,

View file

@ -2162,6 +2162,42 @@ class OpenAIRealtimeDoneEvent(TypedDict):
type: Literal["response.done"]
class OpenAIRealtimeInputAudioBufferSpeechEvent(TypedDict):
type: ReadOnly[Literal["input_audio_buffer.speech_started", "input_audio_buffer.speech_stopped"]]
event_id: ReadOnly[str]
item_id: ReadOnly[str]
class OpenAIRealtimeInputAudioTranscriptionDelta(TypedDict):
type: ReadOnly[Literal["conversation.item.input_audio_transcription.delta"]]
event_id: ReadOnly[str]
item_id: ReadOnly[str]
content_index: ReadOnly[int]
delta: ReadOnly[str]
class OpenAIRealtimeInputAudioTranscriptionCompleted(TypedDict):
type: ReadOnly[Literal["conversation.item.input_audio_transcription.completed"]]
event_id: ReadOnly[str]
item_id: ReadOnly[str]
content_index: ReadOnly[int]
transcript: ReadOnly[str]
class OpenAIRealtimeUsageTokenDetails(TypedDict):
audio_tokens: ReadOnly[int]
text_tokens: ReadOnly[int]
cached_tokens: NotRequired[ReadOnly[int]]
class OpenAIRealtimeResponseUsage(TypedDict):
input_tokens: ReadOnly[int]
output_tokens: ReadOnly[int]
total_tokens: ReadOnly[int]
input_token_details: NotRequired[ReadOnly[OpenAIRealtimeUsageTokenDetails]]
output_token_details: NotRequired[ReadOnly[OpenAIRealtimeUsageTokenDetails]]
class OpenAIRealtimeEventTypes(Enum):
SESSION_CREATED = "session.created"
# Beta delta event names
@ -2199,6 +2235,9 @@ OpenAIRealtimeEvents = (
| OpenAIRealtimeOutputItemDone
| OpenAIRealtimeFunctionCallArgumentsDone
| OpenAIRealtimeDoneEvent
| OpenAIRealtimeInputAudioBufferSpeechEvent
| OpenAIRealtimeInputAudioTranscriptionDelta
| OpenAIRealtimeInputAudioTranscriptionCompleted
)
OpenAIRealtimeStreamList = list[OpenAIRealtimeEvents]

View file

@ -553,6 +553,26 @@
"supports_response_schema": true,
"supports_vision": true
},
"amazon.nova-sonic-v1:0": {
"input_cost_per_audio_token": 3.4e-06,
"input_cost_per_token": 6e-08,
"litellm_provider": "bedrock",
"mode": "realtime",
"output_cost_per_audio_token": 1.36e-05,
"output_cost_per_token": 2.4e-07,
"supports_audio_input": true,
"supports_audio_output": true
},
"amazon.nova-2-sonic-v1:0": {
"input_cost_per_audio_token": 3e-06,
"input_cost_per_token": 3.3e-07,
"litellm_provider": "bedrock",
"mode": "realtime",
"output_cost_per_audio_token": 1.2e-05,
"output_cost_per_token": 2.75e-06,
"supports_audio_input": true,
"supports_audio_output": true
},
"amazon.rerank-v1:0": {
"input_cost_per_query": 0.001,
"input_cost_per_token": 0.0,

View file

@ -827,5 +827,173 @@ class TestBedrockRealtimeSessionEvents:
assert event["session"]["modalities"] == ["text", "audio"]
class TestBedrockRealtimeUserEventsAndUsage:
"""Regression tests for #38346: USER ASR transcripts, speech boundary events,
usage propagation, and duplicate response.created"""
@staticmethod
def _run(config, messages):
logging_obj = MagicMock()
logging_obj.litellm_trace_id = "trace_123"
state = {
"session_configuration_request": json.dumps({"configured": True}),
"current_output_item_id": None,
"current_response_id": None,
"current_conversation_id": None,
"current_delta_chunks": [],
"current_item_chunks": [],
"current_delta_type": None,
}
all_events = []
for msg in messages:
result = config.transform_realtime_response(
json.dumps(msg),
"amazon.nova-2-sonic-v1:0",
logging_obj,
realtime_response_transform_input=dict(state),
)
all_events.extend(result["response"])
state.update(
{
"current_output_item_id": result["current_output_item_id"],
"current_response_id": result["current_response_id"],
"current_conversation_id": result["current_conversation_id"],
"current_delta_chunks": result["current_delta_chunks"],
"current_delta_type": result["current_delta_type"],
}
)
return all_events
def test_user_speech_start_and_stop_events(self):
events = self._run(
BedrockRealtimeConfig(),
[{"event": {"userSpeechStart": {}}}, {"event": {"userSpeechEnd": {}}}],
)
assert [e["type"] for e in events] == [
"input_audio_buffer.speech_started",
"input_audio_buffer.speech_stopped",
]
assert all(e["event_id"] and e["item_id"] for e in events)
def test_user_transcript_emits_input_audio_transcription_events(self):
events = self._run(
BedrockRealtimeConfig(),
[
{
"event": {
"contentStart": {
"role": "USER",
"type": "TEXT",
"additionalModelFields": json.dumps({"generationStage": "FINAL"}),
}
}
},
{"event": {"textOutput": {"content": "ready"}}},
{"event": {"contentEnd": {"stopReason": "PARTIAL_TURN"}}},
],
)
deltas = [e for e in events if e["type"] == "conversation.item.input_audio_transcription.delta"]
completed = [e for e in events if e["type"] == "conversation.item.input_audio_transcription.completed"]
assert len(deltas) == 1 and deltas[0]["delta"] == "ready"
assert len(completed) == 1 and completed[0]["transcript"] == "ready"
assert deltas[0]["item_id"] == completed[0]["item_id"]
assert not any(e["type"] == "response.text.delta" for e in events)
def test_speculative_user_transcript_emits_delta_only(self):
events = self._run(
BedrockRealtimeConfig(),
[
{
"event": {
"contentStart": {
"role": "USER",
"type": "TEXT",
"additionalModelFields": json.dumps({"generationStage": "SPECULATIVE"}),
}
}
},
{"event": {"textOutput": {"content": "rea"}}},
],
)
assert [e["type"] for e in events] == ["conversation.item.input_audio_transcription.delta"]
def test_user_transcript_state_resets_on_content_end(self):
events = self._run(
BedrockRealtimeConfig(),
[
{"event": {"contentStart": {"role": "USER", "type": "TEXT"}}},
{"event": {"contentEnd": {"stopReason": "PARTIAL_TURN"}}},
{"event": {"contentStart": {"role": "ASSISTANT", "type": "TEXT"}}},
{"event": {"textOutput": {"content": "Hi there"}}},
],
)
text_deltas = [e for e in events if e["type"] == "response.text.delta"]
assert len(text_deltas) == 1 and text_deltas[0]["delta"] == "Hi there"
assert not any(e["type"].startswith("conversation.item.input_audio_transcription") for e in events)
def test_response_created_emitted_once_per_response(self):
events = self._run(
BedrockRealtimeConfig(),
[
{"event": {"contentStart": {"role": "ASSISTANT", "type": "TEXT"}}},
{"event": {"textOutput": {"content": "Hi"}}},
{"event": {"contentEnd": {"stopReason": "PARTIAL_TURN"}}},
{"event": {"contentStart": {"role": "ASSISTANT", "type": "AUDIO"}}},
],
)
assert sum(1 for e in events if e["type"] == "response.created") == 1
def test_usage_event_propagates_to_response_done(self):
events = self._run(
BedrockRealtimeConfig(),
[
{
"event": {
"usageEvent": {
"totalInputTokens": 25,
"totalOutputTokens": 40,
"totalTokens": 65,
"details": {
"total": {
"input": {"speechTokens": 20, "textTokens": 5},
"output": {"speechTokens": 30, "textTokens": 10},
}
},
}
}
},
{"event": {"contentStart": {"role": "ASSISTANT", "type": "TEXT"}}},
{"event": {"textOutput": {"content": "Hi"}}},
{"event": {"contentEnd": {"stopReason": "END_TURN"}}},
],
)
done_events = [e for e in events if e["type"] == "response.done"]
assert len(done_events) == 1
usage = done_events[0]["response"]["usage"]
assert usage["input_tokens"] == 25
assert usage["output_tokens"] == 40
assert usage["total_tokens"] == 65
assert usage["input_token_details"]["audio_tokens"] == 20
assert usage["input_token_details"]["text_tokens"] == 5
assert usage["output_token_details"]["audio_tokens"] == 30
assert usage["output_token_details"]["text_tokens"] == 10
def test_response_done_without_usage_event_reports_zero_usage(self):
events = self._run(
BedrockRealtimeConfig(),
[
{"event": {"contentStart": {"role": "ASSISTANT", "type": "TEXT"}}},
{"event": {"textOutput": {"content": "Hi"}}},
{"event": {"contentEnd": {"stopReason": "END_TURN"}}},
],
)
done_events = [e for e in events if e["type"] == "response.done"]
assert len(done_events) == 1
usage = done_events[0]["response"]["usage"]
assert usage["input_tokens"] == 0
assert usage["output_tokens"] == 0
assert usage["total_tokens"] == 0
if __name__ == "__main__":
pytest.main([__file__, "-v"])