scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-22 00:31:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 51

Feat: add nova sonic sts model

Browse source
This commit is contained in:
Sameer Kankute 2026-01-23 13:03:43 +05:30
parent ebf0beda97
commit 650426b5ac
5 changed files with 1231 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
litellm/llms/bedrock/realtime/__init__.py Normal file
View file

@ -0,0 +1 @@
# Bedrock Realtime API implementation

463
litellm/llms/bedrock/realtime/handler.py Normal file
View file

@ -0,0 +1,463 @@
"""
Bedrock Nova Sonic realtime handler using native HTTP/2 bidirectional streaming.
This bridges the user's WebSocket connection to LiteLLM with Bedrock's HTTP/2
bidirectional streaming protocol using AWS SigV4 signing.
"""
import asyncio
import json
from typing import Any, Optional
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
from ..base_aws_llm import BaseAWSLLM
from .transformation import BedrockRealtimeConfig
class BedrockRealtime(BaseAWSLLM):
"""Handler for Bedrock Nova Sonic realtime using native HTTP/2 bidirectional streaming."""
def __init__(self):
super().__init__()
self.config = BedrockRealtimeConfig()
async def async_realtime(
self,
model: str,
websocket: Any,
logging_obj: LiteLLMLogging,
api_base: Optional[str] = None,
api_key: Optional[str] = None,
client: Optional[Any] = None,
timeout: Optional[float] = None,
aws_region_name: Optional[str] = None,
aws_access_key_id: Optional[str] = None,
aws_secret_access_key: Optional[str] = None,
aws_session_token: Optional[str] = None,
):
"""
Bridge user's WebSocket to Bedrock HTTP/2 bidirectional stream.
Args:
model: Model name (e.g., "bedrock/amazon.nova-sonic-v1:0")
websocket: FastAPI WebSocket connection from user
logging_obj: LiteLLM logging object
api_base: Optional API base URL
api_key: AWS access key ID (or None to use environment)
aws_region_name: AWS region
aws_access_key_id: AWS access key ID
aws_secret_access_key: AWS secret access key
aws_session_token: AWS session token
"""
try:
# Import required libraries for HTTP/2 and AWS signing
import httpx
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
except ImportError as e:
error_msg = f"Required libraries not available: {str(e)}. Install with: pip install httpx botocore"
verbose_logger.error(error_msg)
await websocket.close(code=1011, reason=error_msg)
return
# Determine AWS region
if aws_region_name is None:
# Try to get region from model ARN
region = self._get_aws_region_from_model_arn(model)
if region is None:
# Default to us-west-2 if no region specified
region = "us-west-2"
else:
region = aws_region_name
# Get AWS credentials
credentials = self.get_credentials(
aws_access_key_id=aws_access_key_id or api_key,
aws_secret_access_key=aws_secret_access_key,
aws_session_token=aws_session_token,
aws_region_name=region,
)
model_id = model.replace("bedrock/", "")
endpoint = api_base or f"https://bedrock-runtime.{region}.amazonaws.com"
verbose_logger.debug(f"Connecting to Bedrock Nova Sonic: model={model_id}, region={region}")
try:
# Log the request
logging_obj.pre_call(
input=None,
api_key=credentials.access_key,
additional_args={
"api_base": endpoint,
"model": model_id,
"region": region,
},
)
# Create HTTP/2 bidirectional stream to Bedrock
await self._bridge_streams_http2(
websocket=websocket,
endpoint=endpoint,
model_id=model_id,
credentials=credentials,
region=region,
logging_obj=logging_obj,
model=model,
)
except Exception as e:
verbose_logger.exception(f"Error in Bedrock realtime: {e}")
try:
await websocket.close(
code=1011, reason=f"Internal server error: {str(e)}"
)
except RuntimeError as close_error:
if "already completed" in str(close_error) or "websocket.close" in str(
close_error
):
pass
else:
raise
async def _bridge_streams_http2(
self,
websocket: Any,
endpoint: str,
model_id: str,
credentials: Any,
region: str,
logging_obj: LiteLLMLogging,
model: str,
):
"""
Bridge messages between user's WebSocket and Bedrock's HTTP/2 bidirectional stream.
Args:
websocket: User's WebSocket connection
endpoint: Bedrock endpoint URL
model_id: Model ID
credentials: AWS credentials
region: AWS region
logging_obj: Logging object
model: Model name
"""
import httpx
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
# Construct the URL for bidirectional streaming
url = f"{endpoint}/model/{model_id}/invoke-with-bidirectional-stream"
# Create a signed request
headers = {
"Content-Type": "application/vnd.amazon.eventstream",
"Accept": "application/vnd.amazon.eventstream",
}
# Sign the request using AWS SigV4
request = AWSRequest(method="POST", url=url, headers=headers, data=b"")
SigV4Auth(credentials, "bedrock", region).add_auth(request)
# Extract signed headers
signed_headers = dict(request.headers)
verbose_logger.debug(f"Connecting to Bedrock HTTP/2 stream: {url}")
# Create a queue for request data
import asyncio
request_queue = asyncio.Queue()
# Create HTTP/2 client
async with httpx.AsyncClient(http2=True, timeout=None) as http_client:
# Create bidirectional stream using HTTP/2
async with http_client.stream(
"POST",
url,
headers=signed_headers,
content=self._generate_request_stream_from_queue(request_queue),
) as response:
verbose_logger.debug(f"Bedrock stream established: status={response.status_code}")
if response.status_code != 200:
error_msg = f"Failed to establish stream: {response.status_code}"
try:
error_body = await response.aread()
verbose_logger.error(f"Error response: {error_body}")
except:
pass
verbose_logger.error(error_msg)
await websocket.close(code=1011, reason=error_msg[:100]) # Limit reason length
return
# Run request handling and response handling concurrently
await asyncio.gather(
self._handle_websocket_to_bedrock(websocket, model, request_queue),
self._forward_bedrock_to_user(
response=response,
websocket=websocket,
logging_obj=logging_obj,
model=model,
),
return_exceptions=True,
)
async def _generate_request_stream(self, websocket: Any, model: str):
"""
Generate request stream from user WebSocket messages.
Args:
websocket: User's WebSocket connection
model: Model name
Yields:
Encoded event stream data
"""
try:
while True:
# Receive message from user's WebSocket
message = await websocket.receive_text()
verbose_logger.debug(f"Received from user: {message[:200]}...")
# Transform OpenAI format to Bedrock format
bedrock_messages = self.config.transform_realtime_request(
message=message,
model=model,
session_configuration_request=None,
)
# Yield each transformed message as event stream data
for bedrock_msg in bedrock_messages:
# Encode as event stream format
event_data = self._encode_event_stream_message(bedrock_msg)
yield event_data
verbose_logger.debug(f"Sent to Bedrock: {bedrock_msg[:200]}...")
except Exception as e:
verbose_logger.debug(f"Request stream ended: {e}")
async def _generate_request_stream_from_queue(self, request_queue):
"""
Generate request stream from a queue.
Args:
request_queue: asyncio.Queue containing encoded event stream data
Yields:
Encoded event stream data
"""
try:
while True:
# Get data from queue
data = await request_queue.get()
if data is None: # Sentinel value to end stream
break
yield data
except Exception as e:
verbose_logger.debug(f"Request stream from queue ended: {e}")
async def _handle_websocket_to_bedrock(self, websocket: Any, model: str, request_queue):
"""
Handle messages from WebSocket and put them in the request queue.
Args:
websocket: User's WebSocket connection
model: Model name
request_queue: asyncio.Queue to put encoded messages
"""
try:
while True:
# Receive message from user's WebSocket
message = await websocket.receive_text()
verbose_logger.debug(f"Received from user: {message[:200]}...")
# Transform OpenAI format to Bedrock format
bedrock_messages = self.config.transform_realtime_request(
message=message,
model=model,
session_configuration_request=None,
)
# Put each transformed message in the queue
for bedrock_msg in bedrock_messages:
# Encode as event stream format
event_data = self._encode_event_stream_message(bedrock_msg)
await request_queue.put(event_data)
verbose_logger.debug(f"Sent to Bedrock: {bedrock_msg[:200]}...")
except Exception as e:
verbose_logger.debug(f"WebSocket to Bedrock handler ended: {e}")
finally:
# Signal end of stream
await request_queue.put(None)
def _encode_event_stream_message(self, message: str) -> bytes:
"""
Encode a message in AWS event stream format.
Args:
message: JSON message string
Returns:
Encoded event stream bytes
"""
import struct
import binascii
# Convert message to bytes
payload = message.encode('utf-8')
# AWS event stream format:
# Prelude (12 bytes):
# - Total byte length (4 bytes, big-endian uint32)
# - Headers byte length (4 bytes, big-endian uint32)
# - Prelude CRC (4 bytes, big-endian uint32)
# Headers (variable, can be 0 bytes)
# Payload (variable)
# Message CRC (4 bytes, big-endian uint32)
headers_bytes = b"" # No headers for now
headers_length = len(headers_bytes)
# Calculate total length (prelude + headers + payload + message CRC)
total_length = 12 + headers_length + len(payload) + 4
# Build prelude (without CRC yet)
prelude_without_crc = struct.pack('>II', total_length, headers_length)
# Calculate prelude CRC
prelude_crc = binascii.crc32(prelude_without_crc) & 0xFFFFFFFF
prelude = prelude_without_crc + struct.pack('>I', prelude_crc)
# Build message (without final CRC)
message_without_crc = prelude + headers_bytes + payload
# Calculate message CRC
message_crc = binascii.crc32(message_without_crc) & 0xFFFFFFFF
# Build complete message
complete_message = message_without_crc + struct.pack('>I', message_crc)
return complete_message
async def _forward_bedrock_to_user(
self,
response: Any,
websocket: Any,
logging_obj: LiteLLMLogging,
model: str,
):
"""
Forward responses from Bedrock to user's WebSocket.
Args:
response: HTTP/2 streaming response
websocket: User's WebSocket connection
logging_obj: Logging object
model: Model name
"""
from botocore.eventstream import EventStreamBuffer
try:
current_state = {
"current_output_item_id": None,
"current_response_id": None,
"current_conversation_id": None,
"current_delta_chunks": None,
"current_item_chunks": None,
"current_delta_type": None,
"session_configuration_request": None,
}
# Create event stream buffer for decoding
event_buffer = EventStreamBuffer()
# Read event stream from response
async for chunk in response.aiter_bytes():
if not chunk:
continue
verbose_logger.debug(f"Received chunk from Bedrock: {len(chunk)} bytes")
# Add chunk to event stream buffer
event_buffer.add_data(chunk)
# Decode messages from buffer
messages = self._decode_event_stream_buffer(event_buffer)
for raw_response in messages:
verbose_logger.debug(f"Received from Bedrock: {raw_response[:200]}...")
# Transform Bedrock format to OpenAI format
result = self.config.transform_realtime_response(
message=raw_response,
model=model,
logging_obj=logging_obj,
realtime_response_transform_input=current_state,
)
# Update state
current_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_item_chunks": result["current_item_chunks"],
"current_delta_type": result["current_delta_type"],
"session_configuration_request": result["session_configuration_request"],
})
# Send transformed events to user
response_data = result["response"]
if isinstance(response_data, list):
for event_obj in response_data:
event_str = json.dumps(event_obj)
await websocket.send_text(event_str)
verbose_logger.debug(f"Sent to user: {event_str[:200]}...")
else:
event_str = json.dumps(response_data)
await websocket.send_text(event_str)
verbose_logger.debug(f"Sent to user: {event_str[:200]}...")
except Exception as e:
verbose_logger.exception(f"Forward to user ended: {e}")
def _decode_event_stream_buffer(self, buffer: Any) -> list[str]:
"""
Decode AWS event stream using EventStreamBuffer.
Args:
buffer: EventStreamBuffer instance
Returns:
List of decoded JSON message strings
"""
messages = []
for event in buffer:
try:
# Get the payload from the event
if hasattr(event, 'payload'):
payload = event.payload
elif hasattr(event, 'to_response_dict'):
response_dict = event.to_response_dict()
if 'body' in response_dict:
payload = response_dict['body']
else:
continue
else:
continue
# Decode payload
if isinstance(payload, bytes):
message = payload.decode('utf-8')
messages.append(message)
elif isinstance(payload, str):
messages.append(payload)
except Exception as e:
verbose_logger.warning(f"Failed to decode event: {e}")
return messages

717
litellm/llms/bedrock/realtime/transformation.py Normal file
View file

@ -0,0 +1,717 @@
"""
This file contains the transformation logic for the Bedrock Nova Sonic realtime API.
Bedrock Nova Sonic uses bidirectional streaming with the InvokeModelWithBidirectionalStream API.
"""
import json
import uuid as uuid_module
from typing import Any, Dict, List, Optional, Union, cast
from litellm import verbose_logger
from litellm._uuid import uuid
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig
from litellm.types.llms.openai import (
OpenAIRealtimeContentPartDone,
OpenAIRealtimeConversationItemCreated,
OpenAIRealtimeDoneEvent,
OpenAIRealtimeEvents,
OpenAIRealtimeEventTypes,
OpenAIRealtimeOutputItemDone,
OpenAIRealtimeResponseAudioDone,
OpenAIRealtimeResponseContentPartAdded,
OpenAIRealtimeResponseDelta,
OpenAIRealtimeResponseDoneObject,
OpenAIRealtimeResponseTextDone,
OpenAIRealtimeStreamResponseBaseObject,
OpenAIRealtimeStreamResponseOutputItemAdded,
OpenAIRealtimeStreamSession,
OpenAIRealtimeStreamSessionEvents,
OpenAIRealtimeTurnDetection,
)
from litellm.types.realtime import (
ALL_DELTA_TYPES,
RealtimeModalityResponseTransformOutput,
RealtimeResponseTransformInput,
RealtimeResponseTypedDict,
)
from litellm.utils import get_empty_usage
from ..base_aws_llm import BaseAWSLLM
# Map OpenAI voice names to Bedrock Nova Sonic voice IDs
OPENAI_TO_BEDROCK_VOICE_MAP = {
"alloy": "matthew",
"echo": "matthew",
"fable": "ruth",
"onyx": "matthew",
"nova": "ruth",
"shimmer": "ruth",
}
class BedrockRealtimeConfig(BaseRealtimeConfig, BaseAWSLLM):
"""
Configuration for Bedrock Nova Sonic realtime API.
Transforms between OpenAI realtime API format and Bedrock's bidirectional streaming format.
"""
def __init__(self):
super().__init__()
self.prompt_name = str(uuid_module.uuid4())
self.content_name = str(uuid_module.uuid4())
def validate_environment(
self, headers: dict, model: str, api_key: Optional[str] = None
) -> dict:
"""Validate AWS credentials are available."""
return headers
def get_complete_url(
self, api_base: Optional[str], model: str, api_key: Optional[str] = None
) -> str:
"""
Get the API base URL for Bedrock (not WebSocket - AWS SDK handles the connection).
Example output:
"https://bedrock-runtime.us-west-1.amazonaws.com"
"""
# If api_base is not provided, get region from credentials
if api_base is None:
from botocore.credentials import Credentials
credentials = self.get_credentials(
aws_access_key_id=api_key,
aws_secret_access_key=None,
aws_session_token=None,
aws_region_name=None,
)
# credentials can be either Boto3CredentialsInfo or Credentials (from cache)
if isinstance(credentials, Credentials):
# If cached, it's just a Credentials object - use default region
region = "us-west-1"
else:
# It's a Boto3CredentialsInfo object
region = credentials.aws_region_name
api_base = f"https://bedrock-runtime.{region}.amazonaws.com"
return api_base
def map_openai_voice_to_bedrock(self, voice: Optional[str]) -> str:
"""Map OpenAI voice names to Bedrock voice IDs."""
if voice is None:
return "matthew" # Default voice
return OPENAI_TO_BEDROCK_VOICE_MAP.get(voice.lower(), "matthew")
def transform_realtime_request(
self,
message: str,
model: str,
session_configuration_request: Optional[str] = None,
) -> List[str]:
"""
Transform OpenAI realtime request format to Bedrock format.
Bedrock expects events in this sequence:
1. sessionStart
2. promptStart
3. contentStart
4. textInput (or audioInput)
5. contentEnd
6. promptEnd
7. sessionEnd (when done)
"""
try:
json_message = json.loads(message)
except json.JSONDecodeError:
if isinstance(message, bytes):
message_str = message.decode("utf-8", errors="replace")
else:
message_str = str(message)
raise ValueError(f"Invalid JSON message: {message_str}")
messages: List[str] = []
# Handle session.update - this sets up the session configuration
if "type" in json_message and json_message["type"] == "session.update":
session = json_message.get("session", {})
# Extract configuration
temperature = session.get("temperature", 0.7)
max_tokens = session.get("max_response_output_tokens", 1024)
voice = session.get("voice", "alloy")
bedrock_voice = self.map_openai_voice_to_bedrock(voice)
# Store configuration for later use
self.session_config = {
"temperature": temperature,
"max_tokens": max_tokens,
"voice": bedrock_voice,
}
# Don't send anything yet - wait for actual content
return messages
# Handle input_audio_buffer.append - audio input
elif "type" in json_message and json_message["type"] == "input_audio_buffer.append":
audio_data = json_message.get("audio", "")
# Create full event sequence for audio input
events = self._create_audio_input_events(audio_data)
messages.append(json.dumps({"events": events}))
# Handle response.create - text input
elif "type" in json_message and json_message["type"] == "response.create":
# Extract text from the conversation
response_data = json_message.get("response", {})
# For now, we'll handle this as a trigger to start processing
# The actual text should come from previous messages
pass
# Handle conversation.item.create - text message
elif "type" in json_message and json_message["type"] == "conversation.item.create":
item = json_message.get("item", {})
content = item.get("content", [])
# Extract text from content
text_content = ""
for part in content:
if part.get("type") == "input_text":
text_content = part.get("text", "")
elif part.get("type") == "text":
text_content = part.get("text", "")
if text_content:
events = self._create_text_input_events(text_content)
messages.append(json.dumps({"events": events}))
return messages
def _create_text_input_events(self, text: str) -> List[Dict[str, Any]]:
"""Create the event sequence for text input."""
config = getattr(self, "session_config", {})
temperature = config.get("temperature", 0.7)
max_tokens = config.get("max_tokens", 1024)
voice = config.get("voice", "matthew")
return [
{
"event": {
"sessionStart": {
"inferenceConfiguration": {
"maxTokens": max_tokens,
"topP": 0.9,
"temperature": temperature,
}
}
}
},
{
"event": {
"promptStart": {
"promptName": self.prompt_name,
"textOutputConfiguration": {"mediaType": "text/plain"},
"audioOutputConfiguration": {
"mediaType": "audio/lpcm",
"sampleRateHertz": 24000,
"sampleSizeBits": 16,
"channelCount": 1,
"voiceId": voice,
"encoding": "base64",
"audioType": "SPEECH",
},
}
}
},
{
"event": {
"contentStart": {
"promptName": self.prompt_name,
"contentName": self.content_name,
"type": "TEXT",
"interactive": False,
"role": "USER",
"textInputConfiguration": {"mediaType": "text/plain"},
}
}
},
{
"event": {
"textInput": {
"promptName": self.prompt_name,
"contentName": self.content_name,
"content": text,
}
}
},
{
"event": {
"contentEnd": {
"promptName": self.prompt_name,
"contentName": self.content_name,
}
}
},
{
"event": {
"promptEnd": {
"promptName": self.prompt_name,
}
}
},
{"event": {"sessionEnd": {}}},
]
def _create_audio_input_events(self, audio_data: str) -> List[Dict[str, Any]]:
"""Create the event sequence for audio input."""
config = getattr(self, "session_config", {})
temperature = config.get("temperature", 0.7)
max_tokens = config.get("max_tokens", 1024)
voice = config.get("voice", "matthew")
return [
{
"event": {
"sessionStart": {
"inferenceConfiguration": {
"maxTokens": max_tokens,
"topP": 0.9,
"temperature": temperature,
}
}
}
},
{
"event": {
"promptStart": {
"promptName": self.prompt_name,
"audioInputConfiguration": {
"mediaType": "audio/lpcm",
"sampleRateHertz": 16000,
"sampleSizeBits": 16,
"channelCount": 1,
"encoding": "base64",
},
"audioOutputConfiguration": {
"mediaType": "audio/lpcm",
"sampleRateHertz": 24000,
"sampleSizeBits": 16,
"channelCount": 1,
"voiceId": voice,
"encoding": "base64",
"audioType": "SPEECH",
},
}
}
},
{
"event": {
"contentStart": {
"promptName": self.prompt_name,
"contentName": self.content_name,
"type": "AUDIO",
"interactive": False,
"role": "USER",
"audioInputConfiguration": {
"mediaType": "audio/lpcm",
"sampleRateHertz": 16000,
"sampleSizeBits": 16,
"channelCount": 1,
"encoding": "base64",
},
}
}
},
{
"event": {
"audioInput": {
"promptName": self.prompt_name,
"contentName": self.content_name,
"content": audio_data,
}
}
},
{
"event": {
"contentEnd": {
"promptName": self.prompt_name,
"contentName": self.content_name,
}
}
},
{
"event": {
"promptEnd": {
"promptName": self.prompt_name,
}
}
},
{"event": {"sessionEnd": {}}},
]
def transform_realtime_response(
self,
message: Union[str, bytes],
model: str,
logging_obj: LiteLLMLoggingObj,
realtime_response_transform_input: RealtimeResponseTransformInput,
) -> RealtimeResponseTypedDict:
"""
Transform Bedrock realtime response to OpenAI format.
Bedrock sends events like:
- sessionStarted
- promptStarted
- contentStarted
- textOutput / audioOutput (streaming chunks)
- contentEnded
- promptEnded
- sessionEnded
"""
try:
json_message = json.loads(message)
except json.JSONDecodeError:
if isinstance(message, bytes):
message_str = message.decode("utf-8", errors="replace")
else:
message_str = str(message)
raise ValueError(f"Invalid JSON message: {message_str}")
logging_session_id = logging_obj.litellm_trace_id
current_output_item_id = realtime_response_transform_input["current_output_item_id"]
current_response_id = realtime_response_transform_input["current_response_id"]
current_conversation_id = realtime_response_transform_input["current_conversation_id"]
current_delta_chunks = realtime_response_transform_input["current_delta_chunks"]
session_configuration_request = realtime_response_transform_input["session_configuration_request"]
current_item_chunks = realtime_response_transform_input["current_item_chunks"]
current_delta_type = realtime_response_transform_input["current_delta_type"]
returned_message: List[OpenAIRealtimeEvents] = []
# Handle different Bedrock event types
if "sessionStarted" in json_message:
# Create session.created event
session_event = self._create_session_event(model, logging_session_id)
returned_message.append(session_event)
session_configuration_request = json.dumps(session_event)
elif "textOutput" in json_message:
# Handle text output chunks
text_chunk = json_message["textOutput"].get("content", "")
if not current_response_id:
current_response_id = f"resp_{uuid.uuid4()}"
if not current_output_item_id:
current_output_item_id = f"item_{uuid.uuid4()}"
if not current_conversation_id:
current_conversation_id = f"conv_{uuid.uuid4()}"
# Create initial events if this is the first chunk
if current_delta_chunks is None:
current_delta_chunks = []
current_delta_type = "text"
initial_events = self._create_initial_response_events(
current_response_id,
current_output_item_id,
current_conversation_id,
"text"
)
returned_message.extend(initial_events)
# Create text delta event
delta_event = OpenAIRealtimeResponseDelta(
type="response.text.delta",
content_index=0,
event_id=f"event_{uuid.uuid4()}",
item_id=current_output_item_id,
output_index=0,
response_id=current_response_id,
delta=text_chunk,
)
returned_message.append(delta_event)
current_delta_chunks.append(delta_event)
elif "audioOutput" in json_message:
# Handle audio output chunks
audio_chunk = json_message["audioOutput"].get("content", "")
if not current_response_id:
current_response_id = f"resp_{uuid.uuid4()}"
if not current_output_item_id:
current_output_item_id = f"item_{uuid.uuid4()}"
if not current_conversation_id:
current_conversation_id = f"conv_{uuid.uuid4()}"
# Create initial events if this is the first chunk
if current_delta_chunks is None:
current_delta_chunks = []
current_delta_type = "audio"
initial_events = self._create_initial_response_events(
current_response_id,
current_output_item_id,
current_conversation_id,
"audio"
)
returned_message.extend(initial_events)
# Create audio delta event
delta_event = OpenAIRealtimeResponseDelta(
type="response.audio.delta",
content_index=0,
event_id=f"event_{uuid.uuid4()}",
item_id=current_output_item_id,
output_index=0,
response_id=current_response_id,
delta=audio_chunk,
)
returned_message.append(delta_event)
# Don't accumulate audio chunks to avoid memory issues
elif "contentEnded" in json_message or "promptEnded" in json_message or "sessionEnded" in json_message:
# Create done events
if current_delta_type and current_output_item_id and current_response_id:
done_events = self._create_done_events(
current_output_item_id,
current_response_id,
current_conversation_id,
current_delta_chunks,
current_delta_type,
)
returned_message.extend(done_events)
# Reset state
current_delta_chunks = None
current_output_item_id = None
current_response_id = None
current_delta_type = None
return {
"response": returned_message,
"current_output_item_id": current_output_item_id,
"current_response_id": current_response_id,
"current_delta_chunks": current_delta_chunks,
"current_conversation_id": current_conversation_id,
"current_item_chunks": current_item_chunks,
"current_delta_type": current_delta_type,
"session_configuration_request": session_configuration_request,
}
def _create_session_event(
self, model: str, session_id: str
) -> OpenAIRealtimeStreamSessionEvents:
"""Create a session.created event."""
return OpenAIRealtimeStreamSessionEvents(
type="session.created",
session=OpenAIRealtimeStreamSession(
id=session_id,
model=model.replace("bedrock/", ""),
modalities=["text", "audio"],
),
event_id=f"event_{uuid.uuid4()}",
)
def _create_initial_response_events(
self,
response_id: str,
output_item_id: str,
conversation_id: str,
delta_type: ALL_DELTA_TYPES,
) -> List[OpenAIRealtimeEvents]:
"""Create initial events when starting a new response."""
events: List[OpenAIRealtimeEvents] = []
# response.created
events.append(
OpenAIRealtimeStreamResponseBaseObject(
type="response.created",
event_id=f"event_{uuid.uuid4()}",
response={
"object": "realtime.response",
"id": response_id,
"status": "in_progress",
"output": [],
"conversation_id": conversation_id,
"modalities": [delta_type],
},
)
)
# response.output_item.added
events.append(
OpenAIRealtimeStreamResponseOutputItemAdded(
type="response.output_item.added",
response_id=response_id,
output_index=0,
item={
"id": output_item_id,
"object": "realtime.item",
"type": "message",
"status": "in_progress",
"role": "assistant",
"content": [],
},
)
)
# conversation.item.created
events.append(
OpenAIRealtimeConversationItemCreated(
type="conversation.item.created",
event_id=f"event_{uuid.uuid4()}",
item={
"id": output_item_id,
"object": "realtime.item",
"type": "message",
"status": "in_progress",
"role": "assistant",
"content": [],
},
)
)
# response.content_part.added
events.append(
OpenAIRealtimeResponseContentPartAdded(
type="response.content_part.added",
content_index=0,
output_index=0,
event_id=f"event_{uuid.uuid4()}",
item_id=output_item_id,
part={
"type": delta_type,
"text": "" if delta_type == "text" else None,
"transcript": "" if delta_type == "audio" else None,
},
response_id=response_id,
)
)
return events
def _create_done_events(
self,
output_item_id: str,
response_id: str,
conversation_id: Optional[str],
delta_chunks: Optional[List[OpenAIRealtimeResponseDelta]],
delta_type: ALL_DELTA_TYPES,
) -> List[OpenAIRealtimeEvents]:
"""Create done events when response is complete."""
events: List[OpenAIRealtimeEvents] = []
# Accumulate text if available
text_content = ""
if delta_chunks and delta_type == "text":
text_content = "".join([chunk["delta"] for chunk in delta_chunks])
# response.text.done or response.audio.done
if delta_type == "text":
events.append(
OpenAIRealtimeResponseTextDone(
type="response.text.done",
content_index=0,
event_id=f"event_{uuid.uuid4()}",
item_id=output_item_id,
output_index=0,
response_id=response_id,
text=text_content,
)
)
else:
events.append(
OpenAIRealtimeResponseAudioDone(
type="response.audio.done",
content_index=0,
event_id=f"event_{uuid.uuid4()}",
item_id=output_item_id,
output_index=0,
response_id=response_id,
)
)
# response.content_part.done
events.append(
OpenAIRealtimeContentPartDone(
type="response.content_part.done",
content_index=0,
event_id=f"event_{uuid.uuid4()}",
item_id=output_item_id,
output_index=0,
part={
"type": delta_type,
"text": text_content if delta_type == "text" else None,
"transcript": "" if delta_type == "audio" else None,
},
response_id=response_id,
)
)
# response.output_item.done
events.append(
OpenAIRealtimeOutputItemDone(
type="response.output_item.done",
event_id=f"event_{uuid.uuid4()}",
output_index=0,
response_id=response_id,
item={
"id": output_item_id,
"object": "realtime.item",
"type": "message",
"status": "completed",
"role": "assistant",
"content": [
{
"type": delta_type,
"text": text_content if delta_type == "text" else None,
"transcript": "" if delta_type == "audio" else None,
}
],
},
)
)
# response.done
events.append(
OpenAIRealtimeDoneEvent(
type="response.done",
event_id=f"event_{uuid.uuid4()}",
response=OpenAIRealtimeResponseDoneObject(
object="realtime.response",
id=response_id,
status="completed",
output=[
{
"id": output_item_id,
"object": "realtime.item",
"type": "message",
"status": "completed",
"role": "assistant",
"content": [
{
"type": delta_type,
"text": text_content if delta_type == "text" else None,
"transcript": "" if delta_type == "audio" else None,
}
],
}
],
conversation_id=conversation_id or f"conv_{uuid.uuid4()}",
modalities=[delta_type],
usage=get_empty_usage().model_dump(),
),
)
)
return events
def requires_session_configuration(self) -> bool:
"""Bedrock requires session configuration."""
return True
def session_configuration_request(self, model: str) -> Optional[str]:
"""Return default session configuration."""
session_event = self._create_session_event(model, str(uuid.uuid4()))
return json.dumps(session_event)

40
litellm/realtime_api/main.py
View file

@ -16,11 +16,13 @@ from litellm.utils import ProviderConfigManager
from ..litellm_core_utils.get_litellm_params import get_litellm_params
from ..litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
from ..llms.azure.realtime.handler import AzureOpenAIRealtime
from ..llms.bedrock.realtime.handler import BedrockRealtime
from ..llms.openai.realtime.handler import OpenAIRealtime
from ..utils import client as wrapper_client
from ..llms.custom_httpx.http_handler import get_shared_realtime_ssl_context
azure_realtime = AzureOpenAIRealtime()
bedrock_realtime = BedrockRealtime()
openai_realtime = OpenAIRealtime()
base_llm_http_handler = BaseLLMHTTPHandler()
@ -79,7 +81,43 @@ async def _arealtime(
model=model,
provider=LlmProviders(_custom_llm_provider),
)
if provider_config is not None:
# Special handling for Bedrock - uses AWS SDK, not WebSocket
if _custom_llm_provider == "bedrock":
# Get AWS credentials
aws_access_key_id = (
dynamic_api_key
or kwargs.get("aws_access_key_id")
or get_secret_str("AWS_ACCESS_KEY_ID")
)
aws_secret_access_key = (
kwargs.get("aws_secret_access_key")
or get_secret_str("AWS_SECRET_ACCESS_KEY")
)
aws_session_token = (
kwargs.get("aws_session_token")
or get_secret_str("AWS_SESSION_TOKEN")
)
aws_region_name = (
kwargs.get("aws_region_name")
or get_secret_str("AWS_REGION_NAME")
or "us-west-1"
)
await bedrock_realtime.async_realtime(
model=model,
websocket=websocket,
logging_obj=litellm_logging_obj,
api_base=api_base,
api_key=aws_access_key_id,
aws_access_key_id=aws_access_key_id,
aws_secret_access_key=aws_secret_access_key,
aws_session_token=aws_session_token,
aws_region_name=aws_region_name,
client=None,
timeout=timeout,
)
elif provider_config is not None:
await base_llm_http_handler.async_realtime(
model=model,
websocket=websocket,

11
litellm/utils.py
View file

@ -8592,6 +8592,10 @@ class ProviderConfigManager:
from litellm.llms.gemini.realtime.transformation import GeminiRealtimeConfig
return GeminiRealtimeConfig()
elif LlmProviders.BEDROCK == provider:
from litellm.llms.bedrock.realtime.transformation import BedrockRealtimeConfig
return BedrockRealtimeConfig()
return None
@staticmethod
@ -8737,6 +8741,13 @@ class ProviderConfigManager:
)
return AzureAVATextToSpeechConfig()
elif litellm.LlmProviders.BEDROCK == provider:
# Bedrock Nova Sonic speech-to-speech models
if "sonic" in model.lower():
from litellm.llms.bedrock.text_to_speech.transformation import (
BedrockNovaSonicConfig,
)
return BedrockNovaSonicConfig()
elif litellm.LlmProviders.ELEVENLABS == provider:
from litellm.llms.elevenlabs.text_to_speech.transformation import (
ElevenLabsTextToSpeechConfig,