diff --git a/litellm/llms/bedrock/realtime/__init__.py b/litellm/llms/bedrock/realtime/__init__.py new file mode 100644 index 00000000000..74bfe707589 --- /dev/null +++ b/litellm/llms/bedrock/realtime/__init__.py @@ -0,0 +1 @@ +# Bedrock Realtime API implementation diff --git a/litellm/llms/bedrock/realtime/handler.py b/litellm/llms/bedrock/realtime/handler.py new file mode 100644 index 00000000000..4b3c4bf0a19 --- /dev/null +++ b/litellm/llms/bedrock/realtime/handler.py @@ -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 diff --git a/litellm/llms/bedrock/realtime/transformation.py b/litellm/llms/bedrock/realtime/transformation.py new file mode 100644 index 00000000000..7d3023b7d0e --- /dev/null +++ b/litellm/llms/bedrock/realtime/transformation.py @@ -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) diff --git a/litellm/realtime_api/main.py b/litellm/realtime_api/main.py index 0a78fb7b72a..b5063051635 100644 --- a/litellm/realtime_api/main.py +++ b/litellm/realtime_api/main.py @@ -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, diff --git a/litellm/utils.py b/litellm/utils.py index ad9e36795af..8cffd2f6825 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -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,