import asyncio import concurrent.futures import json from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union, cast import litellm from litellm._logging import verbose_logger from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig from litellm.types.llms.openai import ( OpenAIRealtimeEvents, OpenAIRealtimeOutputItemDone, OpenAIRealtimeResponseDelta, OpenAIRealtimeStreamResponseBaseObject, OpenAIRealtimeStreamSessionEvents, ) from litellm.types.realtime import ALL_DELTA_TYPES from .litellm_logging import Logging as LiteLLMLogging if TYPE_CHECKING: from websockets.asyncio.client import ClientConnection CLIENT_CONNECTION_CLASS = ClientConnection else: CLIENT_CONNECTION_CLASS = Any # Create a thread pool with a maximum of 10 threads executor = concurrent.futures.ThreadPoolExecutor(max_workers=10) DefaultLoggedRealTimeEventTypes = [ "session.created", "response.create", "response.done", ] class RealTimeStreaming: def __init__( self, websocket: Any, backend_ws: CLIENT_CONNECTION_CLASS, logging_obj: LiteLLMLogging, provider_config: Optional[BaseRealtimeConfig] = None, model: str = "", user_api_key_dict: Optional[Any] = None, request_data: Optional[Dict] = None, ): self.websocket = websocket self.backend_ws = backend_ws self.logging_obj = logging_obj self.messages: List[OpenAIRealtimeEvents] = [] self.input_message: Dict = {} self.input_messages: List[Dict[str, str]] = [] self.session_tools: List[Dict] = [] self.tool_calls: List[Dict] = [] _logged_real_time_event_types = litellm.logged_real_time_event_types if _logged_real_time_event_types is None: _logged_real_time_event_types = DefaultLoggedRealTimeEventTypes self.logged_real_time_event_types = _logged_real_time_event_types self.provider_config = provider_config self.model = model self.current_delta_chunks: Optional[List[OpenAIRealtimeResponseDelta]] = None self.current_output_item_id: Optional[str] = None self.current_response_id: Optional[str] = None self.current_conversation_id: Optional[str] = None self.current_item_chunks: Optional[List[OpenAIRealtimeOutputItemDone]] = None self.current_delta_type: Optional[ALL_DELTA_TYPES] = None self.session_configuration_request: Optional[str] = None self.user_api_key_dict = user_api_key_dict self.request_data: Dict = request_data or {} # Violation counter for end_session_after_n_fails support self._violation_count: int = 0 # When a text message is blocked, hold the guardrail reason so the next # response.create can be rewritten to include the failure context. self._pending_guardrail_message: Optional[str] = None def _should_store_message( self, message_obj: Union[dict, OpenAIRealtimeEvents], ) -> bool: _msg_type = message_obj["type"] if "type" in message_obj else None if self.logged_real_time_event_types == "*": return True if _msg_type and _msg_type in self.logged_real_time_event_types: return True return False def store_message(self, message: Union[str, bytes, OpenAIRealtimeEvents]): """Store message in list""" if isinstance(message, bytes): message = message.decode("utf-8") if isinstance(message, dict): message_obj = message else: message_obj = json.loads(message) self._collect_tool_calls_from_response_done(cast(dict, message_obj)) try: if ( not isinstance(message, dict) or message_obj.get("type") == "session.created" or message_obj.get("type") == "session.updated" ): message_obj = OpenAIRealtimeStreamSessionEvents(**message_obj) # type: ignore elif not isinstance(message, dict): message_obj = OpenAIRealtimeStreamResponseBaseObject(**message_obj) # type: ignore except Exception as e: verbose_logger.debug(f"Error parsing message for logging: {e}") raise e if self._should_store_message(message_obj): self.messages.append(message_obj) def _collect_user_input_from_client_event( self, message: Union[str, dict] ) -> None: """Extract user text content from client WebSocket events for spend logging.""" try: if isinstance(message, str): msg_obj = json.loads(message) elif isinstance(message, dict): msg_obj = message else: return msg_type = msg_obj.get("type", "") if msg_type == "conversation.item.create": item = msg_obj.get("item", {}) if item.get("role") == "user": content_list = item.get("content", []) for content in content_list: if ( isinstance(content, dict) and content.get("type") == "input_text" ): text = content.get("text", "") if text: self.input_messages.append( {"role": "user", "content": text} ) elif msg_type == "session.update": session = msg_obj.get("session", {}) instructions = session.get("instructions", "") if instructions: self.input_messages.append( {"role": "system", "content": instructions} ) tools = session.get("tools") if tools and isinstance(tools, list): self.session_tools = tools except (json.JSONDecodeError, AttributeError, TypeError): pass def _collect_user_input_from_backend_event( self, event_obj: Union[dict, OpenAIRealtimeEvents] ) -> None: """Extract user voice transcription from backend events for spend logging.""" try: event_type = event_obj.get("type", "") if ( event_type == "conversation.item.input_audio_transcription.completed" ): transcript = cast(str, event_obj.get("transcript", "")) if transcript: self.input_messages.append( {"role": "user", "content": transcript} ) except (AttributeError, TypeError): pass def _collect_tool_calls_from_response_done( self, event_obj: Union[dict, OpenAIRealtimeEvents] ) -> None: """Extract function_call items from response.done events for spend logging.""" try: if event_obj.get("type") != "response.done": return response = cast(Dict[str, Any], event_obj.get("response", {})) for item in response.get("output", []): if item.get("type") == "function_call": self.tool_calls.append( { "id": item.get("call_id", ""), "type": "function", "function": { "name": item.get("name", ""), "arguments": item.get("arguments", "{}"), }, } ) except (AttributeError, TypeError): pass def store_input(self, message: Union[str, dict]): """Store input message""" self.input_message = message if isinstance(message, dict) else {} self._collect_user_input_from_client_event(message) if self.logging_obj: self.logging_obj.pre_call(input=message, api_key="") async def log_messages(self): """Log messages in list""" if self.logging_obj: if self.input_messages: self.logging_obj.model_call_details["messages"] = ( self.input_messages ) if self.session_tools or self.tool_calls: self.logging_obj.model_call_details[ "realtime_tools" ] = self.session_tools self.logging_obj.model_call_details[ "realtime_tool_calls" ] = self.tool_calls ## ASYNC LOGGING # Create an event loop for the new thread asyncio.create_task(self.logging_obj.async_success_handler(self.messages)) ## SYNC LOGGING executor.submit(self.logging_obj.success_handler(self.messages)) async def _send_to_backend(self, message: str) -> None: """Send a message to the backend WebSocket. If a provider_config is set the message is first passed through transform_realtime_request so that provider-specific translation (e.g. dropping session.update for Vertex AI) is applied even for guardrail-injected messages. """ if self.provider_config: transformed = self.provider_config.transform_realtime_request( message, self.model, self.session_configuration_request ) for msg in transformed: await self.backend_ws.send(msg) # type: ignore[union-attr, attr-defined] else: await self.backend_ws.send(message) # type: ignore[union-attr, attr-defined] def _has_realtime_guardrails(self) -> bool: """Return True if any callback is registered for realtime guardrail event types.""" from litellm.integrations.custom_guardrail import CustomGuardrail from litellm.types.guardrails import GuardrailEventHooks _realtime_event_types = [ GuardrailEventHooks.realtime_input_transcription, GuardrailEventHooks.pre_call, GuardrailEventHooks.post_call, ] return any( isinstance(cb, CustomGuardrail) and any( cb.should_run_guardrail( data=self.request_data, event_type=et, ) for et in _realtime_event_types ) for cb in litellm.callbacks ) def _has_audio_transcription_guardrails(self) -> bool: """Return True if any callback needs to run on audio transcriptions (VAD path). When this returns True, we inject a session.update to disable the LLM's auto-response so the guardrail can gate it first. Must match the same hook criteria as run_realtime_guardrails() so that any guardrail that would actually check the transcript also disables auto-response before the transcript arrives. """ return self._has_realtime_guardrails() async def run_realtime_guardrails( self, transcript: str, item_id: Optional[str] = None, ) -> bool: """ Run registered guardrails on a completed speech transcription. Returns True if blocked (synthetic warning already sent to client). Returns False if clean (caller should send response.create to the backend). """ from litellm.integrations.custom_guardrail import CustomGuardrail from litellm.types.guardrails import GuardrailEventHooks _realtime_event_types = [ GuardrailEventHooks.realtime_input_transcription, GuardrailEventHooks.pre_call, GuardrailEventHooks.post_call, ] _check_data = {**self.request_data, "transcript": transcript} _already_run: set = set() for callback in litellm.callbacks: if not isinstance(callback, CustomGuardrail): continue if id(callback) in _already_run: continue if not any( callback.should_run_guardrail(data=_check_data, event_type=et) for et in _realtime_event_types ): continue _already_run.add(id(callback)) try: await callback.apply_guardrail( inputs={"texts": [transcript], "images": []}, request_data={"user_api_key_dict": self.user_api_key_dict}, input_type="request", ) except Exception as e: # Re-raise unexpected errors (no status_code/detail = programming bug, not a block). # HTTPException and guardrail-raised exceptions have a status_code or detail attr. is_guardrail_block = hasattr(e, "status_code") or isinstance(e, ValueError) if not is_guardrail_block: verbose_logger.exception( "[realtime guardrail] unexpected error in apply_guardrail: %s", e ) raise # Extract the human-readable error from the detail dict (HTTPException) # or fall back to str(e) for plain ValueError. detail = getattr(e, "detail", None) if isinstance(detail, dict): safe_msg = detail.get("error") or str(e) elif detail is not None: safe_msg = str(detail) else: safe_msg = str(e) or "I'm sorry, that request was blocked by the content filter." # Use realtime_violation_message if configured; fall back to guardrail error text. error_msg = getattr(callback, "realtime_violation_message", None) or safe_msg # Cancel any in-progress LLM response (e.g. VAD auto-response). await self._send_to_backend(json.dumps({"type": "response.cancel"})) # Send the policy violation hint (shows as small gray status text in UI). await self.websocket.send_text( json.dumps({ "type": "error", "error": { "type": "guardrail_violation", "message": error_msg, "code": "content_policy_violation", }, }) ) # Ask the LLM to voice the exact guardrail message so the # user hears it as audio in voice sessions (not just text). guardrail_prompt = ( f"Say exactly the following message to the user, word for word, " f"do not add anything else: {error_msg}" ) await self._send_to_backend(json.dumps({ "type": "conversation.item.create", "item": { "type": "message", "role": "user", "content": [{"type": "input_text", "text": guardrail_prompt}], }, })) await self._send_to_backend( json.dumps({"type": "response.create"}) ) self._violation_count += 1 end_session_after: Optional[int] = getattr( callback, "end_session_after_n_fails", None ) should_end = getattr(callback, "on_violation", None) == "end_session" or ( end_session_after is not None and self._violation_count >= end_session_after ) if should_end: verbose_logger.warning( "[realtime guardrail] ending session after violation %d", self._violation_count, ) await self.backend_ws.close() # type: ignore[union-attr, attr-defined] verbose_logger.warning( "[realtime guardrail] BLOCKED transcript (violation %d): %r", self._violation_count, transcript[:80], ) return True return False async def _handle_provider_config_message(self, raw_response) -> None: """Process a backend message when a provider_config is set (transformed path).""" returned_object = self.provider_config.transform_realtime_response( # type: ignore[union-attr] raw_response, self.model, self.logging_obj, realtime_response_transform_input={ "session_configuration_request": self.session_configuration_request, "current_output_item_id": self.current_output_item_id, "current_response_id": self.current_response_id, "current_delta_chunks": self.current_delta_chunks, "current_conversation_id": self.current_conversation_id, "current_item_chunks": self.current_item_chunks, "current_delta_type": self.current_delta_type, }, ) transformed_response = returned_object["response"] self.current_output_item_id = returned_object["current_output_item_id"] self.current_response_id = returned_object["current_response_id"] self.current_delta_chunks = returned_object["current_delta_chunks"] self.current_conversation_id = returned_object["current_conversation_id"] self.current_item_chunks = returned_object["current_item_chunks"] self.current_delta_type = returned_object["current_delta_type"] self.session_configuration_request = returned_object["session_configuration_request"] events = ( transformed_response if isinstance(transformed_response, list) else [transformed_response] ) for event in events: event_str = json.dumps(event) ## For audio/VAD guardrail path: forward session.created first, then inject. if ( isinstance(event, dict) and event.get("type") == "session.created" and self._has_audio_transcription_guardrails() ): self.store_message(event_str) await self.websocket.send_text(event_str) await self._send_to_backend( json.dumps( { "type": "session.update", "session": {"turn_detection": {"create_response": False}}, } ) ) continue ## GUARDRAIL: run on transcription events in provider_config path too if ( isinstance(event, dict) and event.get("type") == "conversation.item.input_audio_transcription.completed" ): transcript = event.get("transcript", "") self._collect_user_input_from_backend_event(cast(dict, event)) self.store_message(event_str) await self.websocket.send_text(event_str) blocked = await self.run_realtime_guardrails( cast(str, transcript), item_id=cast(Optional[str], event.get("item_id")) ) if not blocked: await self._send_to_backend( json.dumps({"type": "response.create"}) ) continue ## LOGGING self.store_message(event_str) await self.websocket.send_text(event_str) async def _handle_raw_backend_message(self, raw_response) -> bool: """Process a backend message without provider_config (raw path). Returns True if the caller should skip the default store+forward (i.e. continue the loop). """ try: event_obj = json.loads(raw_response) # For audio/VAD guardrail path: once the session is ready, tell the backend # not to auto-respond after VAD detects end-of-speech. We send the # session.created to the client FIRST so the client is always in sync, then # inject the session.update so a potential error from the backend doesn't # arrive before the client sees session.created. if ( event_obj.get("type") == "session.created" and self._has_audio_transcription_guardrails() ): self.store_message(raw_response) await self.websocket.send_text(raw_response) await self._send_to_backend( json.dumps( { "type": "session.update", "session": {"turn_detection": {"create_response": False}}, } ) ) return True if ( event_obj.get("type") == "conversation.item.input_audio_transcription.completed" ): transcript = event_obj.get("transcript", "") self._collect_user_input_from_backend_event(event_obj) ## LOGGING — must happen before continue below self.store_message(raw_response) # Forward transcript to client so user sees what they said await self.websocket.send_text(raw_response) blocked = await self.run_realtime_guardrails( transcript, item_id=event_obj.get("item_id"), ) if not blocked: # Clean — trigger LLM response await self._send_to_backend( json.dumps({"type": "response.create"}) ) return True except (json.JSONDecodeError, AttributeError): pass return False async def backend_to_client_send_messages(self): import websockets try: while True: try: raw_response = await self.backend_ws.recv( # type: ignore[union-attr] decode=False ) # improves performance except TypeError: raw_response = await self.backend_ws.recv() # type: ignore[union-attr, assignment] if self.provider_config: try: await self._handle_provider_config_message(raw_response) except Exception as e: verbose_logger.exception( f"Error processing backend message, skipping: {e}" ) continue else: handled = await self._handle_raw_backend_message(raw_response) if handled: continue ## LOGGING self.store_message(raw_response) await self.websocket.send_text(raw_response) except websockets.exceptions.ConnectionClosed as e: # type: ignore verbose_logger.exception( f"Connection closed in backend to client send messages - {e}" ) except Exception as e: verbose_logger.exception(f"Error in backend to client send messages: {e}") finally: await self.log_messages() async def client_ack_messages(self): try: while True: message = await self.websocket.receive_text() ## GUARDRAIL: intercept conversation.item.create for text-based injection. try: msg_obj = json.loads(message) msg_type = msg_obj.get("type") if msg_type == "conversation.item.create": # Check user text messages for prompt injection item = msg_obj.get("item", {}) if item.get("role") == "user": content_list = item.get("content", []) texts = [ c.get("text", "") for c in content_list if isinstance(c, dict) and c.get("type") == "input_text" ] combined_text = " ".join(texts) if combined_text: blocked = await self.run_realtime_guardrails( combined_text ) if blocked: # Store the guardrail reason so the next response.create # (sent automatically by the client) is rewritten to # include it as response instructions. self._pending_guardrail_message = combined_text continue # don't forward the original blocked message if msg_type == "response.create" and self._pending_guardrail_message: # The guardrail already sent the synthetic AI bubble — drop this # response.create so OpenAI doesn't generate an additional response. self._pending_guardrail_message = None continue except (json.JSONDecodeError, AttributeError): pass ## LOGGING self.store_input(message=message) ## FORWARD TO BACKEND if self.provider_config: message = self.provider_config.transform_realtime_request( message, self.model ) for msg in message: await self.backend_ws.send(msg) # type: ignore[union-attr] else: await self.backend_ws.send(message) # type: ignore[union-attr] except Exception as e: verbose_logger.debug(f"Error in client ack messages: {e}") async def bidirectional_forward(self): forward_task = asyncio.create_task(self.backend_to_client_send_messages()) try: await self.client_ack_messages() except self.websocket.exceptions.ConnectionClosed: # type: ignore verbose_logger.debug("Connection closed") forward_task.cancel() finally: if not forward_task.done(): forward_task.cancel() try: await forward_task except asyncio.CancelledError: pass