mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-05 08:07:05 +00:00
Merge pull request #38597 from BerriAI/litellm_fix_nova_sonic_realtime_user_asr_usage
fix(bedrock): surface Nova Sonic user transcripts, speech events, and usage in realtime API
This commit is contained in:
Mateo Wang
GitHub
commit
336269cfec
11 changed files with 772 additions and 45 deletions
|
|
@ -7,6 +7,9 @@ warnings.filterwarnings("ignore", message=".*conflict with protected namespace.*
|
|||
# Suppress Pydantic 2.11+ deprecation warning about accessing model_fields on instances
|
||||
# This warning can accumulate during streaming and cause memory leaks
|
||||
warnings.filterwarnings("ignore", message=".*Accessing the.*attribute on the instance is deprecated.*")
|
||||
# ReadOnly on TypedDict fields is repo-wide static discipline (LIT012); pydantic warns it
|
||||
# cannot enforce it at runtime, which floods proxy boot once such a type is schema-walked
|
||||
warnings.filterwarnings("ignore", message=".*`ReadOnly` qualifier.*")
|
||||
### INIT VARIABLES #########################
|
||||
import threading
|
||||
import os
|
||||
|
|
|
|||
|
|
@ -7,13 +7,18 @@ This uses aws_sdk_bedrock_runtime for bidirectional streaming with Nova Sonic.
|
|||
import asyncio
|
||||
import contextlib
|
||||
import json
|
||||
from collections.abc import AsyncIterator, Mapping
|
||||
from typing import Final, Protocol
|
||||
|
||||
from pydantic import JsonValue, TypeAdapter
|
||||
|
||||
import litellm
|
||||
from litellm._logging import _redact_string, verbose_proxy_logger
|
||||
from litellm.litellm_core_utils.aws_partition import get_aws_dns_suffix
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
|
||||
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
|
||||
from litellm.litellm_core_utils.realtime_streaming import DefaultLoggedRealTimeEventTypes
|
||||
from litellm.types.llms.openai import OpenAIRealtimeEvents
|
||||
from litellm.types.realtime import RealtimeResponseTransformInput
|
||||
|
||||
from ..base_aws_llm import BaseAWSLLM
|
||||
|
|
@ -32,6 +37,17 @@ def _json_str(value: JsonValue) -> str | None:
|
|||
return value if isinstance(value, str) else None
|
||||
|
||||
|
||||
def _should_log_event(openai_message: Mapping[str, object]) -> bool:
|
||||
logged_types: Final = (
|
||||
litellm.logged_real_time_event_types
|
||||
if litellm.logged_real_time_event_types is not None
|
||||
else DefaultLoggedRealTimeEventTypes
|
||||
)
|
||||
if logged_types == "*":
|
||||
return True
|
||||
return openai_message.get("type") in logged_types
|
||||
|
||||
|
||||
class RealtimeClientWebSocket(Protocol):
|
||||
"""The client-facing websocket surface the realtime bridge talks to."""
|
||||
|
||||
|
|
@ -205,16 +221,22 @@ class BedrockRealtime(BaseAWSLLM):
|
|||
)
|
||||
)
|
||||
|
||||
bedrock_to_client_task: Final = asyncio.create_task(
|
||||
self._forward_bedrock_to_client(
|
||||
bedrock_stream,
|
||||
websocket,
|
||||
transformation_config,
|
||||
model,
|
||||
logging_obj,
|
||||
session_state,
|
||||
async def forward_bedrock_and_collect_logged_events() -> tuple[OpenAIRealtimeEvents, ...]:
|
||||
return tuple(
|
||||
[
|
||||
event
|
||||
async for event in self._forward_bedrock_to_client(
|
||||
bedrock_stream,
|
||||
websocket,
|
||||
transformation_config,
|
||||
model,
|
||||
logging_obj,
|
||||
session_state,
|
||||
)
|
||||
]
|
||||
)
|
||||
)
|
||||
|
||||
bedrock_to_client_task: Final = asyncio.create_task(forward_bedrock_and_collect_logged_events())
|
||||
|
||||
# Wait for both tasks to complete
|
||||
await asyncio.gather(
|
||||
|
|
@ -223,6 +245,27 @@ class BedrockRealtime(BaseAWSLLM):
|
|||
return_exceptions=True,
|
||||
)
|
||||
|
||||
forwarded_logged_events: Final = (
|
||||
bedrock_to_client_task.result()
|
||||
if not bedrock_to_client_task.cancelled() and bedrock_to_client_task.exception() is None
|
||||
else ()
|
||||
)
|
||||
logged_events: Final = (
|
||||
*forwarded_logged_events,
|
||||
*(
|
||||
leftover_event
|
||||
for leftover_event in transformation_config.leftover_usage_done_events()
|
||||
if _should_log_event(leftover_event)
|
||||
),
|
||||
)
|
||||
if logged_events:
|
||||
GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue(
|
||||
logging_obj.dispatch_success_handlers(
|
||||
list(logged_events), # mutable-ok: realtime spend logging requires a list result
|
||||
prefer_async_handlers=True,
|
||||
)
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
verbose_proxy_logger.exception("Error in BedrockRealtime.async_realtime: %s", e)
|
||||
try:
|
||||
|
|
@ -304,8 +347,8 @@ class BedrockRealtime(BaseAWSLLM):
|
|||
model: str,
|
||||
logging_obj: LiteLLMLogging,
|
||||
session_state: RealtimeResponseTransformInput,
|
||||
):
|
||||
"""Forward messages from Bedrock stream to client WebSocket."""
|
||||
) -> AsyncIterator[OpenAIRealtimeEvents]:
|
||||
"""Forward messages from Bedrock to the client, yielding the ones to record for spend logging."""
|
||||
try:
|
||||
while True:
|
||||
# Receive from Bedrock
|
||||
|
|
@ -353,11 +396,14 @@ class BedrockRealtime(BaseAWSLLM):
|
|||
)
|
||||
|
||||
# Send transformed messages to client
|
||||
openai_messages = transformed_response.get("response", [])
|
||||
response_value = transformed_response["response"]
|
||||
openai_messages = response_value if isinstance(response_value, list) else (response_value,)
|
||||
for openai_message in openai_messages:
|
||||
message_json = json.dumps(openai_message)
|
||||
await client_ws.send_text(message_json)
|
||||
verbose_proxy_logger.debug("Bedrock Realtime: Sent to client: %s", message_json[:200])
|
||||
if _should_log_event(openai_message):
|
||||
yield openai_message
|
||||
|
||||
except Exception as e:
|
||||
verbose_proxy_logger.debug("Bedrock to client forwarding ended: %s", e, exc_info=True)
|
||||
|
|
|
|||
|
|
@ -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,29 +20,54 @@ 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,
|
||||
RealtimeResponseTransformInput,
|
||||
RealtimeResponseTypedDict,
|
||||
)
|
||||
from litellm.utils import get_empty_usage
|
||||
|
||||
|
||||
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 +112,15 @@ 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._user_item_id: str | None = None
|
||||
self._user_transcript_buffer = ""
|
||||
self._cumulative_usage = BedrockUsageEvent()
|
||||
self._reported_usage = BedrockUsageEvent()
|
||||
|
||||
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 +725,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 +739,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 +755,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 +768,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 +809,108 @@ 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 _current_user_item_id(self, new_utterance: bool = False) -> str:
|
||||
"""Item id shared by all events of one user utterance (speech boundaries and transcript)."""
|
||||
if new_utterance or self._user_item_id is None:
|
||||
self._user_item_id = f"item_{uuid.uuid4()}"
|
||||
return self._user_item_id
|
||||
|
||||
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": self._current_user_item_id(new_utterance=is_speech_start),
|
||||
}
|
||||
return (speech_event,)
|
||||
|
||||
def transform_usage_event(self, usage_event: BedrockUsageEvent) -> None:
|
||||
"""Record Bedrock's session-cumulative usage totals for the next response.done."""
|
||||
verbose_logger.debug("Handling usageEvent")
|
||||
self._cumulative_usage = usage_event
|
||||
|
||||
def _take_usage_delta(self) -> OpenAIRealtimeResponseUsage:
|
||||
"""Usage for the response now completing: cumulative totals minus what prior response.done events reported."""
|
||||
prior: Final = self._reported_usage
|
||||
latest: Final = self._cumulative_usage
|
||||
self._reported_usage = latest
|
||||
input_details: Final[OpenAIRealtimeUsageTokenDetails] = {
|
||||
"audio_tokens": latest.details.total.input.speechTokens - prior.details.total.input.speechTokens,
|
||||
"text_tokens": latest.details.total.input.textTokens - prior.details.total.input.textTokens,
|
||||
"cached_tokens": 0,
|
||||
}
|
||||
output_details: Final[OpenAIRealtimeUsageTokenDetails] = {
|
||||
"audio_tokens": latest.details.total.output.speechTokens - prior.details.total.output.speechTokens,
|
||||
"text_tokens": latest.details.total.output.textTokens - prior.details.total.output.textTokens,
|
||||
}
|
||||
usage_delta: Final[OpenAIRealtimeResponseUsage] = {
|
||||
"input_tokens": latest.totalInputTokens - prior.totalInputTokens,
|
||||
"output_tokens": latest.totalOutputTokens - prior.totalOutputTokens,
|
||||
"total_tokens": latest.totalTokens - prior.totalTokens,
|
||||
"input_token_details": input_details,
|
||||
"output_token_details": output_details,
|
||||
}
|
||||
return usage_delta
|
||||
|
||||
def leftover_usage_done_events(self) -> tuple[OpenAIRealtimeEvents, ...]:
|
||||
"""Logged-only response.done for usage Bedrock reports after the final turn's contentEnd."""
|
||||
if self._cumulative_usage == self._reported_usage:
|
||||
return ()
|
||||
usage: Final = self._take_usage_delta()
|
||||
leftover_done: Final = OpenAIRealtimeDoneEvent(
|
||||
type="response.done",
|
||||
event_id=f"event_{uuid.uuid4()}",
|
||||
response=OpenAIRealtimeResponseDoneObject(
|
||||
object="realtime.response",
|
||||
id=f"resp_{uuid.uuid4()}",
|
||||
status="completed",
|
||||
conversation_id=f"conv_{uuid.uuid4()}",
|
||||
usage=dict(usage), # mutable-ok: OpenAIRealtimeResponseDoneObject types usage as plain dict
|
||||
),
|
||||
)
|
||||
return (leftover_done,)
|
||||
|
||||
def transform_user_transcript_event(self, transcript: str) -> tuple[OpenAIRealtimeEvents, ...]:
|
||||
"""Transform a USER-role Bedrock textOutput (ASR transcript) to an OpenAI transcription delta."""
|
||||
verbose_logger.debug("Handling USER textOutput (ASR transcript)")
|
||||
delta_event: Final[OpenAIRealtimeInputAudioTranscriptionDelta] = {
|
||||
"type": "conversation.item.input_audio_transcription.delta",
|
||||
"event_id": f"event_{uuid.uuid4()}",
|
||||
"item_id": self._current_user_item_id(),
|
||||
"content_index": 0,
|
||||
"delta": transcript,
|
||||
}
|
||||
if self._user_transcript_generation_stage != "SPECULATIVE":
|
||||
self._user_transcript_buffer += transcript
|
||||
return (delta_event,)
|
||||
|
||||
def user_transcript_completed_events(self) -> tuple[OpenAIRealtimeEvents, ...]:
|
||||
"""One completed event with the full transcript once the FINAL user content block ends."""
|
||||
transcript: Final = self._user_transcript_buffer
|
||||
if not transcript:
|
||||
return ()
|
||||
self._user_transcript_buffer = ""
|
||||
completed_event: Final[OpenAIRealtimeInputAudioTranscriptionCompleted] = {
|
||||
"type": "conversation.item.input_audio_transcription.completed",
|
||||
"event_id": f"event_{uuid.uuid4()}",
|
||||
"item_id": self._current_user_item_id(),
|
||||
"content_index": 0,
|
||||
"transcript": transcript,
|
||||
}
|
||||
return (completed_event,)
|
||||
|
||||
def transform_text_output_event(
|
||||
self,
|
||||
event: dict,
|
||||
|
|
@ -985,7 +1129,7 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
|
|||
if not current_response_id or not current_conversation_id:
|
||||
return [], None, None, None
|
||||
|
||||
usage_obj: Final = get_empty_usage()
|
||||
usage: Final = self._take_usage_delta()
|
||||
response_done: Final = OpenAIRealtimeDoneEvent(
|
||||
type="response.done",
|
||||
event_id=f"event_{uuid.uuid4()}",
|
||||
|
|
@ -995,11 +1139,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 +1182,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 +1332,26 @@ 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
|
||||
returned_messages.extend(self.user_transcript_completed_events())
|
||||
|
||||
elif "contentEnd" in event:
|
||||
events, current_delta_chunks = self.transform_content_end_event(
|
||||
event,
|
||||
|
|
@ -1224,6 +1370,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
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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]
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -2,3 +2,8 @@
|
|||
id = "GHSA-w8v5-vhqr-4h9v"
|
||||
ignoreUntil = 2026-09-09
|
||||
reason = "diskcache has no fixed release published; remove this entry once one exists"
|
||||
|
||||
[[IgnoredVulns]]
|
||||
id = "GHSA-h7x2-h6g9-p789"
|
||||
ignoreUntil = 2026-09-14
|
||||
reason = "mlflow has no fixed release published; remove this entry once one exists"
|
||||
|
|
|
|||
|
|
@ -11,7 +11,7 @@
|
|||
"user": "",
|
||||
"team_id": "",
|
||||
"organization_id": "",
|
||||
"metadata": "{\"applied_guardrails\": [], \"attempted_fallbacks\": null, \"original_model_group\": null, \"batch_models\": null, \"batch_successful_requests\": null, \"batch_failed_requests\": null, \"mcp_tool_call_metadata\": null, \"vector_store_request_metadata\": null, \"routing_decision\": null, \"internal_call_origin\": null, \"guardrail_information\": null, \"compression_savings\": null, \"litellm_gateway_injected_cache\": null, \"usage_object\": {\"completion_tokens\": 20, \"prompt_tokens\": 10, \"total_tokens\": 30, \"completion_tokens_details\": null, \"prompt_tokens_details\": null}, \"model_map_information\": {\"model_map_key\": \"gpt-4o\", \"model_map_value\": {\"key\": \"gpt-4o\", \"max_tokens\": 16384, \"max_input_tokens\": 128000, \"max_output_tokens\": 16384, \"input_cost_per_token\": 2.5e-06, \"cache_creation_input_token_cost\": null, \"cache_read_input_token_cost\": 1.25e-06, \"input_cost_per_character\": null, \"input_cost_per_token_above_128k_tokens\": null, \"input_cost_per_token_above_200k_tokens\": null, \"input_cost_per_query\": null, \"input_cost_per_second\": null, \"input_cost_per_audio_token\": null, \"input_cost_per_token_batches\": 1.25e-06, \"output_cost_per_token_batches\": 5e-06, \"output_cost_per_token\": 1e-05, \"output_cost_per_audio_token\": null, \"output_cost_per_character\": null, \"output_cost_per_token_above_128k_tokens\": null, \"output_cost_per_character_above_128k_tokens\": null, \"output_cost_per_token_above_200k_tokens\": null, \"output_cost_per_second\": null, \"output_cost_per_image\": null, \"output_vector_size\": null, \"litellm_provider\": \"openai\", \"mode\": \"chat\", \"supports_system_messages\": true, \"supports_response_schema\": true, \"supports_vision\": true, \"supports_function_calling\": true, \"supports_tool_choice\": true, \"supports_assistant_prefill\": false, \"supports_prompt_caching\": true, \"supports_audio_input\": false, \"supports_audio_output\": false, \"supports_pdf_input\": false, \"supports_embedding_image_input\": false, \"supports_native_streaming\": null, \"supports_web_search\": true, \"supports_reasoning\": false, \"search_context_cost_per_query\": {\"search_context_size_low\": 0.03, \"search_context_size_medium\": 0.035, \"search_context_size_high\": 0.05}, \"tpm\": null, \"rpm\": null, \"supported_openai_params\": [\"frequency_penalty\", \"logit_bias\", \"logprobs\", \"top_logprobs\", \"max_tokens\", \"max_completion_tokens\", \"modalities\", \"prediction\", \"n\", \"presence_penalty\", \"seed\", \"stop\", \"stream\", \"stream_options\", \"temperature\", \"top_p\", \"tools\", \"tool_choice\", \"function_call\", \"functions\", \"max_retries\", \"extra_headers\", \"parallel_tool_calls\", \"audio\", \"response_format\", \"user\"]}}, \"additional_usage_values\": {\"completion_tokens_details\": null, \"prompt_tokens_details\": null}, \"user_api_key\": null, \"user_api_key_alias\": null, \"user_api_key_team_id\": null, \"user_api_key_project_id\": null, \"user_api_key_project_alias\": null, \"user_api_key_org_id\": null, \"user_api_key_user_id\": null, \"user_api_key_team_alias\": null, \"spend_logs_metadata\": null, \"requester_ip_address\": null, \"status\": null, \"proxy_server_request\": null, \"error_information\": null, \"attempted_retries\": null, \"max_retries\": null}",
|
||||
"metadata": "{\"applied_guardrails\": [], \"attempted_fallbacks\": null, \"original_model_group\": null, \"batch_models\": null, \"batch_successful_requests\": null, \"batch_failed_requests\": null, \"mcp_tool_call_metadata\": null, \"vector_store_request_metadata\": null, \"routing_decision\": null, \"internal_call_origin\": null, \"router_metadata\": null, \"guardrail_information\": null, \"compression_savings\": null, \"litellm_gateway_injected_cache\": null, \"usage_object\": {\"completion_tokens\": 20, \"prompt_tokens\": 10, \"total_tokens\": 30, \"completion_tokens_details\": null, \"prompt_tokens_details\": null}, \"model_map_information\": {\"model_map_key\": \"gpt-4o\", \"model_map_value\": {\"key\": \"gpt-4o\", \"max_tokens\": 16384, \"max_input_tokens\": 128000, \"max_output_tokens\": 16384, \"input_cost_per_token\": 2.5e-06, \"cache_creation_input_token_cost\": null, \"cache_read_input_token_cost\": 1.25e-06, \"input_cost_per_character\": null, \"input_cost_per_token_above_128k_tokens\": null, \"input_cost_per_token_above_200k_tokens\": null, \"input_cost_per_query\": null, \"input_cost_per_second\": null, \"input_cost_per_audio_token\": null, \"input_cost_per_token_batches\": 1.25e-06, \"output_cost_per_token_batches\": 5e-06, \"output_cost_per_token\": 1e-05, \"output_cost_per_audio_token\": null, \"output_cost_per_character\": null, \"output_cost_per_token_above_128k_tokens\": null, \"output_cost_per_character_above_128k_tokens\": null, \"output_cost_per_token_above_200k_tokens\": null, \"output_cost_per_second\": null, \"output_cost_per_image\": null, \"output_vector_size\": null, \"litellm_provider\": \"openai\", \"mode\": \"chat\", \"supports_system_messages\": true, \"supports_response_schema\": true, \"supports_vision\": true, \"supports_function_calling\": true, \"supports_tool_choice\": true, \"supports_assistant_prefill\": false, \"supports_prompt_caching\": true, \"supports_audio_input\": false, \"supports_audio_output\": false, \"supports_pdf_input\": false, \"supports_embedding_image_input\": false, \"supports_native_streaming\": null, \"supports_web_search\": true, \"supports_reasoning\": false, \"search_context_cost_per_query\": {\"search_context_size_low\": 0.03, \"search_context_size_medium\": 0.035, \"search_context_size_high\": 0.05}, \"tpm\": null, \"rpm\": null, \"supported_openai_params\": [\"frequency_penalty\", \"logit_bias\", \"logprobs\", \"top_logprobs\", \"max_tokens\", \"max_completion_tokens\", \"modalities\", \"prediction\", \"n\", \"presence_penalty\", \"seed\", \"stop\", \"stream\", \"stream_options\", \"temperature\", \"top_p\", \"tools\", \"tool_choice\", \"function_call\", \"functions\", \"max_retries\", \"extra_headers\", \"parallel_tool_calls\", \"audio\", \"response_format\", \"user\"]}}, \"additional_usage_values\": {\"completion_tokens_details\": null, \"prompt_tokens_details\": null}, \"user_api_key\": null, \"user_api_key_alias\": null, \"user_api_key_team_id\": null, \"user_api_key_project_id\": null, \"user_api_key_project_alias\": null, \"user_api_key_org_id\": null, \"user_api_key_user_id\": null, \"user_api_key_team_alias\": null, \"spend_logs_metadata\": null, \"requester_ip_address\": null, \"status\": null, \"proxy_server_request\": null, \"error_information\": null, \"attempted_retries\": null, \"max_retries\": null}",
|
||||
"cache_key": "Cache OFF",
|
||||
"spend": 0.00022500000000000002,
|
||||
"total_tokens": 30,
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ from unittest.mock import MagicMock
|
|||
|
||||
import pytest
|
||||
|
||||
|
||||
import litellm
|
||||
from litellm.llms.bedrock.common_utils import BedrockError
|
||||
from litellm.llms.bedrock.realtime.handler import BedrockRealtime
|
||||
from litellm.llms.bedrock.realtime.transformation import BedrockRealtimeConfig
|
||||
|
|
@ -104,12 +104,24 @@ class RealtimeClientWS:
|
|||
self.closed = True
|
||||
|
||||
|
||||
class ImmediatelyEndingBedrockStream:
|
||||
def __init__(self):
|
||||
class ScriptedBedrockReceiver:
|
||||
def __init__(self, payloads):
|
||||
self._payloads = list(payloads)
|
||||
|
||||
async def receive(self):
|
||||
if not self._payloads:
|
||||
return None
|
||||
payload = self._payloads.pop(0)
|
||||
return SimpleNamespace(value=SimpleNamespace(bytes_=payload.encode("utf-8")))
|
||||
|
||||
|
||||
class ScriptedBedrockStream:
|
||||
def __init__(self, payloads):
|
||||
self.input_stream = FakeInputStream()
|
||||
self._receiver = ScriptedBedrockReceiver(payloads)
|
||||
|
||||
async def await_output(self):
|
||||
return (None, EndedBedrockReceiver())
|
||||
return (None, self._receiver)
|
||||
|
||||
|
||||
class FakeStaticCredentialsResolver:
|
||||
|
|
@ -151,7 +163,7 @@ def stub_aws_sdk_client(monkeypatch):
|
|||
|
||||
async def invoke_model_with_bidirectional_stream(self, operation_input):
|
||||
captured["operation_input"] = operation_input
|
||||
return ImmediatelyEndingBedrockStream()
|
||||
return ScriptedBedrockStream(captured.get("scripted_payloads", []))
|
||||
|
||||
package = types.ModuleType("aws_sdk_bedrock_runtime")
|
||||
client_module = types.ModuleType("aws_sdk_bedrock_runtime.client")
|
||||
|
|
@ -271,19 +283,132 @@ class TestBedrockRealtimeHandler:
|
|||
assert "sessionEnd" in event_names
|
||||
assert stream.input_stream.closed
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_forwarded_events_are_filtered_to_logged_types_for_spend_logging(self):
|
||||
handler = BedrockRealtime()
|
||||
stream = ScriptedBedrockStream(
|
||||
[
|
||||
json.dumps({"event": {"userSpeechStart": {}}}),
|
||||
json.dumps({"event": {"contentStart": {"role": "ASSISTANT", "type": "TEXT"}}}),
|
||||
json.dumps({"event": {"textOutput": {"content": "Hi"}}}),
|
||||
json.dumps({"event": {"contentEnd": {"stopReason": "END_TURN"}}}),
|
||||
]
|
||||
)
|
||||
client_ws = RealtimeClientWS()
|
||||
|
||||
logged_events = [
|
||||
event
|
||||
async for event in handler._forward_bedrock_to_client(
|
||||
stream,
|
||||
client_ws,
|
||||
BedrockRealtimeConfig(),
|
||||
"amazon.nova-sonic-v1:0",
|
||||
FakeLogging(),
|
||||
{},
|
||||
)
|
||||
]
|
||||
|
||||
assert [event["type"] for event in logged_events] == ["response.done"]
|
||||
sent_types = [json.loads(message)["type"] for message in client_ws.sent_to_client]
|
||||
assert "input_audio_buffer.speech_started" in sent_types
|
||||
assert "response.text.delta" in sent_types
|
||||
assert "response.done" in sent_types
|
||||
assert client_ws.closed
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_logged_event_types_star_collects_every_forwarded_event(self, monkeypatch):
|
||||
monkeypatch.setattr(litellm, "logged_real_time_event_types", "*")
|
||||
handler = BedrockRealtime()
|
||||
stream = ScriptedBedrockStream(
|
||||
[
|
||||
json.dumps({"event": {"userSpeechStart": {}}}),
|
||||
json.dumps({"event": {"userSpeechEnd": {}}}),
|
||||
]
|
||||
)
|
||||
client_ws = RealtimeClientWS()
|
||||
|
||||
logged_events = [
|
||||
event
|
||||
async for event in handler._forward_bedrock_to_client(
|
||||
stream,
|
||||
client_ws,
|
||||
BedrockRealtimeConfig(),
|
||||
"amazon.nova-sonic-v1:0",
|
||||
FakeLogging(),
|
||||
{},
|
||||
)
|
||||
]
|
||||
|
||||
assert [event["type"] for event in logged_events] == [
|
||||
"input_audio_buffer.speech_started",
|
||||
"input_audio_buffer.speech_stopped",
|
||||
]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_trailing_usage_after_last_done_is_dispatched_for_spend(self, stub_aws_sdk_client, monkeypatch):
|
||||
import litellm.llms.bedrock.realtime.handler as handler_module
|
||||
|
||||
dispatched = {}
|
||||
|
||||
class RecordingLogging(FakeLogging):
|
||||
async def dispatch_success_handlers(self, result=None, prefer_async_handlers=False, **kwargs):
|
||||
dispatched["events"] = result
|
||||
|
||||
class RecordingLoggingWorker:
|
||||
def ensure_initialized_and_enqueue(self, coro):
|
||||
dispatched["coro"] = coro
|
||||
|
||||
monkeypatch.setattr(handler_module, "GLOBAL_LOGGING_WORKER", RecordingLoggingWorker())
|
||||
stub_aws_sdk_client["scripted_payloads"] = [
|
||||
json.dumps(
|
||||
{
|
||||
"event": {
|
||||
"usageEvent": {
|
||||
"totalInputTokens": 3,
|
||||
"totalOutputTokens": 6,
|
||||
"totalTokens": 9,
|
||||
"details": {
|
||||
"total": {
|
||||
"input": {"speechTokens": 3, "textTokens": 0},
|
||||
"output": {"speechTokens": 0, "textTokens": 6},
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
)
|
||||
]
|
||||
|
||||
await BedrockRealtime().async_realtime(
|
||||
model="amazon.nova-sonic-v1:0",
|
||||
websocket=RealtimeClientWS(),
|
||||
logging_obj=RecordingLogging(),
|
||||
aws_region_name="us-east-1",
|
||||
aws_access_key_id="k",
|
||||
aws_secret_access_key="s",
|
||||
)
|
||||
await dispatched["coro"]
|
||||
|
||||
assert [event["type"] for event in dispatched["events"]] == ["response.done"]
|
||||
usage = dispatched["events"][0]["response"]["usage"]
|
||||
assert (usage["input_tokens"], usage["output_tokens"], usage["total_tokens"]) == (3, 6, 9)
|
||||
assert usage["input_token_details"] == {"audio_tokens": 3, "text_tokens": 0, "cached_tokens": 0}
|
||||
assert usage["output_token_details"] == {"audio_tokens": 0, "text_tokens": 6}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bedrock_stream_end_closes_client_websocket(self):
|
||||
handler = BedrockRealtime()
|
||||
client_ws = ClosableClientWS()
|
||||
|
||||
await handler._forward_bedrock_to_client(
|
||||
async for _ in handler._forward_bedrock_to_client(
|
||||
EndedBedrockStream(),
|
||||
client_ws,
|
||||
BedrockRealtimeConfig(),
|
||||
"amazon.nova-sonic-v1:0",
|
||||
MagicMock(),
|
||||
{},
|
||||
)
|
||||
):
|
||||
pass
|
||||
|
||||
assert client_ws.closed
|
||||
|
||||
|
|
@ -320,9 +445,7 @@ class TestBedrockRealtimeSessionLifecycle:
|
|||
[json.dumps({"type": "session.update", "session": {"instructions": "hi", "modalities": ["text"]}})]
|
||||
)
|
||||
|
||||
await handler._forward_client_to_bedrock(
|
||||
client_ws, stream, config, "amazon.nova-sonic-v1:0", {}, FakeLogging()
|
||||
)
|
||||
await handler._forward_client_to_bedrock(client_ws, stream, config, "amazon.nova-sonic-v1:0", {}, FakeLogging())
|
||||
|
||||
acked = [json.loads(message) for message in client_ws.sent_to_client]
|
||||
updated = [event for event in acked if event["type"] == "session.updated"]
|
||||
|
|
@ -334,9 +457,7 @@ class TestBedrockRealtimeSessionLifecycle:
|
|||
handler = BedrockRealtime()
|
||||
config = BedrockRealtimeConfig()
|
||||
stream = FakeBedrockStream()
|
||||
client_ws = DisconnectingClientWS(
|
||||
[json.dumps({"type": "session.update", "session": {"instructions": "hi"}})]
|
||||
)
|
||||
client_ws = DisconnectingClientWS([json.dumps({"type": "session.update", "session": {"instructions": "hi"}})])
|
||||
|
||||
await handler._forward_client_to_bedrock(client_ws, stream, config, "amazon.nova-sonic-v1:0", {})
|
||||
|
||||
|
|
|
|||
|
|
@ -827,5 +827,310 @@ 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)
|
||||
assert events[0]["item_id"] == events[1]["item_id"]
|
||||
|
||||
def test_utterance_lifecycle_shares_one_item_id(self):
|
||||
events = self._run(
|
||||
BedrockRealtimeConfig(),
|
||||
[
|
||||
{"event": {"userSpeechStart": {}}},
|
||||
{"event": {"userSpeechEnd": {}}},
|
||||
{
|
||||
"event": {
|
||||
"contentStart": {
|
||||
"role": "USER",
|
||||
"type": "TEXT",
|
||||
"additionalModelFields": json.dumps({"generationStage": "FINAL"}),
|
||||
}
|
||||
}
|
||||
},
|
||||
{"event": {"textOutput": {"content": "ready"}}},
|
||||
{"event": {"contentEnd": {"stopReason": "PARTIAL_TURN"}}},
|
||||
],
|
||||
)
|
||||
item_ids = {e["item_id"] for e in events if "item_id" in e}
|
||||
assert len(item_ids) == 1
|
||||
|
||||
def test_new_utterance_gets_new_item_id(self):
|
||||
config = BedrockRealtimeConfig()
|
||||
first = self._run(config, [{"event": {"userSpeechStart": {}}}, {"event": {"userSpeechEnd": {}}}])
|
||||
second = self._run(config, [{"event": {"userSpeechStart": {}}}, {"event": {"userSpeechEnd": {}}}])
|
||||
assert first[0]["item_id"] == first[1]["item_id"]
|
||||
assert second[0]["item_id"] == second[1]["item_id"]
|
||||
assert first[0]["item_id"] != second[0]["item_id"]
|
||||
|
||||
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
|
||||
|
||||
@staticmethod
|
||||
def _usage_event(total_input, total_output, in_speech, in_text, out_speech, out_text):
|
||||
return {
|
||||
"event": {
|
||||
"usageEvent": {
|
||||
"totalInputTokens": total_input,
|
||||
"totalOutputTokens": total_output,
|
||||
"totalTokens": total_input + total_output,
|
||||
"details": {
|
||||
"total": {
|
||||
"input": {"speechTokens": in_speech, "textTokens": in_text},
|
||||
"output": {"speechTokens": out_speech, "textTokens": out_text},
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
_ASSISTANT_TURN = (
|
||||
{"event": {"contentStart": {"role": "ASSISTANT", "type": "TEXT"}}},
|
||||
{"event": {"textOutput": {"content": "Hi"}}},
|
||||
{"event": {"contentEnd": {"stopReason": "END_TURN"}}},
|
||||
)
|
||||
|
||||
def test_multi_turn_usage_reports_per_response_deltas_not_cumulative_totals(self):
|
||||
events = self._run(
|
||||
BedrockRealtimeConfig(),
|
||||
[
|
||||
self._usage_event(25, 40, in_speech=20, in_text=5, out_speech=30, out_text=10),
|
||||
*self._ASSISTANT_TURN,
|
||||
self._usage_event(40, 100, in_speech=30, in_text=10, out_speech=75, out_text=25),
|
||||
*self._ASSISTANT_TURN,
|
||||
],
|
||||
)
|
||||
usages = [e["response"]["usage"] for e in events if e["type"] == "response.done"]
|
||||
assert len(usages) == 2
|
||||
assert (usages[0]["input_tokens"], usages[0]["output_tokens"], usages[0]["total_tokens"]) == (25, 40, 65)
|
||||
assert (usages[1]["input_tokens"], usages[1]["output_tokens"], usages[1]["total_tokens"]) == (15, 60, 75)
|
||||
assert usages[1]["input_token_details"] == {"audio_tokens": 10, "text_tokens": 5, "cached_tokens": 0}
|
||||
assert usages[1]["output_token_details"] == {"audio_tokens": 45, "text_tokens": 15}
|
||||
assert sum(u["total_tokens"] for u in usages) == 140
|
||||
|
||||
def test_usage_reported_after_last_response_done_flushes_as_logged_only_done(self):
|
||||
config = BedrockRealtimeConfig()
|
||||
self._run(
|
||||
config,
|
||||
[
|
||||
self._usage_event(25, 40, in_speech=20, in_text=5, out_speech=30, out_text=10),
|
||||
*self._ASSISTANT_TURN,
|
||||
],
|
||||
)
|
||||
assert config.leftover_usage_done_events() == ()
|
||||
|
||||
self._run(config, [self._usage_event(25, 46, in_speech=20, in_text=5, out_speech=30, out_text=16)])
|
||||
leftover = config.leftover_usage_done_events()
|
||||
assert len(leftover) == 1
|
||||
assert leftover[0]["type"] == "response.done"
|
||||
usage = leftover[0]["response"]["usage"]
|
||||
assert (usage["input_tokens"], usage["output_tokens"], usage["total_tokens"]) == (0, 6, 6)
|
||||
assert usage["output_token_details"] == {"audio_tokens": 0, "text_tokens": 6}
|
||||
assert config.leftover_usage_done_events() == ()
|
||||
|
||||
def test_final_transcript_fragments_emit_one_completed_with_full_transcript(self):
|
||||
events = self._run(
|
||||
BedrockRealtimeConfig(),
|
||||
[
|
||||
{
|
||||
"event": {
|
||||
"contentStart": {
|
||||
"role": "USER",
|
||||
"type": "TEXT",
|
||||
"additionalModelFields": json.dumps({"generationStage": "FINAL"}),
|
||||
}
|
||||
}
|
||||
},
|
||||
{"event": {"textOutput": {"content": "What is the "}}},
|
||||
{"event": {"textOutput": {"content": "capital of France?"}}},
|
||||
{"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 [d["delta"] for d in deltas] == ["What is the ", "capital of France?"]
|
||||
assert len(completed) == 1
|
||||
assert completed[0]["transcript"] == "What is the capital of France?"
|
||||
assert {e["item_id"] for e in deltas + completed} == {completed[0]["item_id"]}
|
||||
|
||||
def test_speculative_transcript_block_end_emits_no_completed(self):
|
||||
events = self._run(
|
||||
BedrockRealtimeConfig(),
|
||||
[
|
||||
{
|
||||
"event": {
|
||||
"contentStart": {
|
||||
"role": "USER",
|
||||
"type": "TEXT",
|
||||
"additionalModelFields": json.dumps({"generationStage": "SPECULATIVE"}),
|
||||
}
|
||||
}
|
||||
},
|
||||
{"event": {"textOutput": {"content": "rea"}}},
|
||||
{"event": {"contentEnd": {"stopReason": "PARTIAL_TURN"}}},
|
||||
],
|
||||
)
|
||||
assert [e["type"] for e in events] == ["conversation.item.input_audio_transcription.delta"]
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
|
|
|
|||
|
|
@ -35,6 +35,22 @@ def mock_mcp_client_ip():
|
|||
yield
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def isolate_global_mcp_registry():
|
||||
"""Restore the module-global MCP server registry after each test.
|
||||
|
||||
Tests here register servers on ``global_mcp_server_manager`` directly; without a
|
||||
restore, entries leak into other test modules sharing the same worker and break
|
||||
assertions over the full registry contents.
|
||||
"""
|
||||
from litellm.proxy._experimental.mcp_server.mcp_server_manager import global_mcp_server_manager
|
||||
|
||||
snapshot = dict(global_mcp_server_manager.registry)
|
||||
yield
|
||||
global_mcp_server_manager.registry.clear()
|
||||
global_mcp_server_manager.registry.update(snapshot)
|
||||
|
||||
|
||||
def _mock_callback_request(base_url: str = "http://localhost:3000/"):
|
||||
"""Return a MagicMock Request for callback/authorize same-origin tests.
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue