diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py index 449a4892621..294f9c485c1 100644 --- a/litellm/litellm_core_utils/realtime_streaming.py +++ b/litellm/litellm_core_utils/realtime_streaming.py @@ -72,6 +72,9 @@ class RealTimeStreaming: 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, @@ -261,18 +264,12 @@ class RealTimeStreaming: When this returns True, we inject a session.update to disable the LLM's auto-response so the guardrail can gate it first. - """ - from litellm.integrations.custom_guardrail import CustomGuardrail - from litellm.types.guardrails import GuardrailEventHooks - return any( - isinstance(cb, CustomGuardrail) - and cb.should_run_guardrail( - data=self.request_data, - event_type=GuardrailEventHooks.realtime_input_transcription, - ) - for cb in litellm.callbacks - ) + 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, @@ -335,18 +332,35 @@ class RealTimeStreaming: # Use realtime_violation_message if configured; fall back to guardrail error text. error_msg = getattr(callback, "realtime_violation_message", None) or safe_msg - # Return the error directly to the WebSocket consumer. + # 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", - }, - } - ) + 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 @@ -559,7 +573,17 @@ class RealTimeStreaming: combined_text ) if blocked: - continue # don't forward to backend + # 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 diff --git a/litellm/llms/custom_httpx/llm_http_handler.py b/litellm/llms/custom_httpx/llm_http_handler.py index b09a36be60f..d6fdc58099f 100644 --- a/litellm/llms/custom_httpx/llm_http_handler.py +++ b/litellm/llms/custom_httpx/llm_http_handler.py @@ -1,4 +1,5 @@ import json +import ssl from typing import ( TYPE_CHECKING, Any, @@ -4659,6 +4660,8 @@ class BaseLLMHTTPHandler: api_key: Optional[str] = None, client: Optional[Any] = None, timeout: Optional[float] = None, + user_api_key_dict: Optional[Any] = None, + litellm_metadata: Optional[Dict[str, Any]] = None, ): import websockets from websockets.asyncio.client import ClientConnection @@ -4672,6 +4675,11 @@ class BaseLLMHTTPHandler: try: ssl_context = get_shared_realtime_ssl_context() + if url.startswith("wss://") and ssl_context is False: + # Keep TLS for wss:// while honoring SSL_VERIFY=False semantics. + ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + ssl_context.check_hostname = False + ssl_context.verify_mode = ssl.CERT_NONE async with websockets.connect( # type: ignore url, additional_headers=headers, @@ -4686,12 +4694,17 @@ class BaseLLMHTTPHandler: if _session_config: await backend_ws.send(_session_config) + _request_data: Dict[str, Any] = {} + if litellm_metadata: + _request_data["litellm_metadata"] = litellm_metadata realtime_streaming = RealTimeStreaming( websocket, cast(ClientConnection, backend_ws), logging_obj, provider_config, model, + user_api_key_dict=user_api_key_dict, + request_data=_request_data, ) if _session_config: realtime_streaming.session_configuration_request = _session_config diff --git a/litellm/llms/gemini/realtime/transformation.py b/litellm/llms/gemini/realtime/transformation.py index 2e0e678e69f..d9465c95e3b 100644 --- a/litellm/llms/gemini/realtime/transformation.py +++ b/litellm/llms/gemini/realtime/transformation.py @@ -867,6 +867,52 @@ class GeminiRealtimeConfig(BaseRealtimeConfig): ) returned_message: List[OpenAIRealtimeEvents] = [] + # Handle transcription events that arrive independently from model + # content. Gemini sends inputTranscription / outputTranscription + # inside serverContent, separately from modelTurn / turnComplete. + server_content = json_message.get("serverContent") + if isinstance(server_content, dict): + input_tx = server_content.get("inputTranscription") + if isinstance(input_tx, dict) and input_tx.get("text"): + returned_message.append( + cast(OpenAIRealtimeEvents, { + "type": "conversation.item.input_audio_transcription.completed", + "event_id": "event_{}".format(uuid.uuid4()), + "transcript": input_tx["text"], + "item_id": "item_{}".format(uuid.uuid4()), + "content_index": 0, + }) + ) + + output_tx = server_content.get("outputTranscription") + if isinstance(output_tx, dict) and output_tx.get("text"): + returned_message.append( + cast(OpenAIRealtimeEvents, { + "type": "response.audio_transcript.delta", + "event_id": "event_{}".format(uuid.uuid4()), + "delta": output_tx["text"], + "item_id": current_output_item_id or "item_{}".format(uuid.uuid4()), + "response_id": current_response_id or "resp_{}".format(uuid.uuid4()), + "output_index": 0, + "content_index": 0, + }) + ) + + # If serverContent only contained transcription(s) and no model + # content, return early — the main loop would fail on unknown keys. + _model_content_keys = {"modelTurn", "turnComplete", "interrupted", "generationComplete"} + if not any(k in server_content for k in _model_content_keys): + 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, + } + for key, value in json_message.items(): # Check if this key or any nested key matches our mapping openai_event = self.map_openai_event( @@ -974,6 +1020,8 @@ class GeminiRealtimeConfig(BaseRealtimeConfig): setup_config: BidiGenerateContentSetup = { "model": f"models/{model}", "generationConfig": {"responseModalities": response_modalities}, + # Return input transcript so guardrails can inspect user speech. + "inputAudioTranscription": {}, } if output_audio_transcription: setup_config["outputAudioTranscription"] = {} diff --git a/litellm/llms/vertex_ai/realtime/transformation.py b/litellm/llms/vertex_ai/realtime/transformation.py index eaa9844f108..5eae143175b 100644 --- a/litellm/llms/vertex_ai/realtime/transformation.py +++ b/litellm/llms/vertex_ai/realtime/transformation.py @@ -124,6 +124,8 @@ class VertexAIRealtimeConfig(GeminiRealtimeConfig): "silenceDurationMs": 800, } }, + # Return input transcript so guardrails can inspect user speech. + "inputAudioTranscription": {}, # Return output transcript so clients can read what the model said. "outputAudioTranscription": {}, } diff --git a/litellm/realtime_api/main.py b/litellm/realtime_api/main.py index ac597fc623d..3e64f61abdb 100644 --- a/litellm/realtime_api/main.py +++ b/litellm/realtime_api/main.py @@ -106,6 +106,8 @@ async def _arealtime( # noqa: PLR0915 client=client, timeout=timeout, headers=headers, + user_api_key_dict=kwargs.get("user_api_key_dict"), + litellm_metadata=_build_litellm_metadata(kwargs), ) elif _custom_llm_provider == "azure": api_base = ( @@ -277,6 +279,8 @@ async def _arealtime( # noqa: PLR0915 client=client, timeout=timeout, headers=headers, + user_api_key_dict=kwargs.get("user_api_key_dict"), + litellm_metadata=_build_litellm_metadata(kwargs), ) else: raise ValueError(f"Unsupported model: {model}") diff --git a/tests/llm_translation/realtime/test_realtime_guardrails_openai.py b/tests/llm_translation/realtime/test_realtime_guardrails_openai.py new file mode 100644 index 00000000000..a7913e6d761 --- /dev/null +++ b/tests/llm_translation/realtime/test_realtime_guardrails_openai.py @@ -0,0 +1,355 @@ +""" +Integration tests for RealTimeStreaming guardrails against a live OpenAI backend. + +These tests require OPENAI_API_KEY and are skipped if not set. + +They verify end-to-end that: + 1. A text message blocked by a guardrail -> error event sent to client, NO AI response. + 2. A voice transcript blocked by a guardrail -> error event sent, response.create NOT sent. + 3. A clean text message passes through and triggers a real OpenAI response. + +Run with: + poetry run pytest tests/llm_translation/realtime/test_realtime_guardrails_openai.py -v -s +""" + +import asyncio +import json +import os +from typing import List +from unittest.mock import AsyncMock, MagicMock + +import pytest + +import litellm +from litellm.integrations.custom_guardrail import CustomGuardrail +from litellm.litellm_core_utils.realtime_streaming import RealTimeStreaming +from litellm.types.guardrails import GuardrailEventHooks + +OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY") +OPENAI_REALTIME_URL = ( + "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-12-17" +) + +pytestmark = pytest.mark.skipif( + not OPENAI_API_KEY, + reason="OPENAI_API_KEY not set - skipping OpenAI realtime integration tests", +) + +# A unique phrase guaranteed NOT to appear in normal assistant output. +BLOCKED_PHRASE = "XSECRETBLOCKTESTPHRASEX" + + +class PhraseBlockingGuardrail(CustomGuardrail): + """Blocks any message containing BLOCKED_PHRASE.""" + + async def apply_guardrail( + self, inputs, request_data, input_type, logging_obj=None + ): + for text in inputs.get("texts", []): + if BLOCKED_PHRASE in text: + raise ValueError( + "Content blocked: contains forbidden test phrase." + ) + return inputs + + +def _make_guardrail(event_hook=GuardrailEventHooks.pre_call): + return PhraseBlockingGuardrail( + guardrail_name="integration-test-guard", + event_hook=event_hook, + default_on=True, + ) + + +async def _wait_for_event( + client_events: List[dict], event_type: str, timeout: float = 15.0 +) -> dict: + """Poll client_events list until an event with matching type appears.""" + deadline = asyncio.get_event_loop().time() + timeout + while asyncio.get_event_loop().time() < deadline: + matching = [e for e in client_events if e.get("type") == event_type] + if matching: + return matching[0] + await asyncio.sleep(0.05) + raise TimeoutError( + f"Timed out waiting for '{event_type}'. Got so far: {[e.get('type') for e in client_events]}" + ) + + +async def _build_streaming(client_events: List[dict], backend_ws, request_data=None): + """Create a RealTimeStreaming with a mock client WebSocket that captures events.""" + client_ws = MagicMock() + input_queue: asyncio.Queue = asyncio.Queue() + + async def send_text(data: str): + client_events.append(json.loads(data)) + + client_ws.send_text = send_text + client_ws.receive_text = input_queue.get + + logging_obj = MagicMock() + logging_obj.pre_call = MagicMock() + logging_obj.async_success_handler = AsyncMock() + logging_obj.success_handler = MagicMock() + logging_obj.model_call_details = {} + + streaming = RealTimeStreaming( + websocket=client_ws, + backend_ws=backend_ws, + logging_obj=logging_obj, + request_data=request_data or {"guardrails": ["integration-test-guard"]}, + ) + return streaming, input_queue + + +@pytest.mark.asyncio +async def test_text_message_blocked_by_guardrail_no_ai_response(): + """ + Send a text message containing the blocked phrase. + Guardrail must: + - Send error event (guardrail_violation) to client. + - Send response.audio_transcript.delta with the block message to client. + - NOT forward response.create to OpenAI (no AI response). + """ + import websockets + + guardrail = _make_guardrail(GuardrailEventHooks.pre_call) + litellm.callbacks = [guardrail] + + client_events: List[dict] = [] + + try: + async with websockets.connect( + OPENAI_REALTIME_URL, + additional_headers={ + "Authorization": f"Bearer {OPENAI_API_KEY}", + "OpenAI-Beta": "realtime=v1", + }, + ) as backend_ws: + streaming, input_queue = await _build_streaming(client_events, backend_ws) + + # Start backend -> client forwarding + backend_task = asyncio.create_task( + streaming.backend_to_client_send_messages() + ) + # Start client -> backend forwarding (reads from input_queue) + client_task = asyncio.create_task(streaming.client_ack_messages()) + + try: + # Wait until session is ready + await _wait_for_event(client_events, "session.created", timeout=15) + + # Send the blocked message + response.create + blocked_item = json.dumps( + { + "type": "conversation.item.create", + "item": { + "role": "user", + "content": [ + { + "type": "input_text", + "text": f"Hello {BLOCKED_PHRASE}", + } + ], + }, + } + ) + await input_queue.put(blocked_item) + # Give guardrail time to process before the follow-up response.create + await asyncio.sleep(0.3) + await input_queue.put(json.dumps({"type": "response.create"})) + + # Allow time for guardrail round-trip + await asyncio.sleep(3.0) + + finally: + backend_task.cancel() + client_task.cancel() + await asyncio.gather(backend_task, client_task, return_exceptions=True) + + # --- Assertions --- + event_types = [e.get("type") for e in client_events] + + # 1. Must have received guardrail error + error_events = [e for e in client_events if e.get("type") == "error"] + assert len(error_events) >= 1, ( + f"Expected at least one error event but got: {event_types}" + ) + assert error_events[0]["error"]["type"] == "guardrail_violation", ( + f"Wrong error type: {error_events[0]}" + ) + + # 2. Must have the guardrail message surfaced as an AI transcript delta + transcript_deltas = [ + e + for e in client_events + if e.get("type") == "response.audio_transcript.delta" + ] + assert len(transcript_deltas) >= 1, ( + f"Expected guardrail message in transcript delta, got: {event_types}" + ) + + # 3. No real AI response should have been generated - response.done would only + # appear if we sent a response.create and OpenAI replied. We allow it in the + # synthetic form (empty output=[]) but NOT with actual AI content. + done_events = [e for e in client_events if e.get("type") == "response.done"] + for done in done_events: + output = done.get("response", {}).get("output", []) + ai_texts = [ + c.get("text", "") or c.get("transcript", "") + for item in output + for c in item.get("content", []) + ] + real_ai_text = " ".join(ai_texts).strip() + assert real_ai_text == "", ( + f"AI responded with real content even though message was blocked: {real_ai_text!r}" + ) + + finally: + litellm.callbacks = [] + + +@pytest.mark.asyncio +async def test_voice_transcript_blocked_by_guardrail(): + """ + Simulate a backend-side voice transcription event containing the blocked phrase. + Guardrail must block it - no response.create sent to OpenAI. + """ + from websockets.exceptions import ConnectionClosed + + guardrail = _make_guardrail(GuardrailEventHooks.realtime_input_transcription) + litellm.callbacks = [guardrail] + + client_events: List[dict] = [] + + # Build the transcript event that would come from the OpenAI backend + transcript_event = json.dumps( + { + "type": "conversation.item.input_audio_transcription.completed", + "transcript": f"This is {BLOCKED_PHRASE} in my voice message", + "item_id": "item_integ_test", + } + ).encode() + + # Mock backend that delivers the transcript then closes + backend_ws = MagicMock() + backend_ws.recv = AsyncMock( + side_effect=[ + transcript_event, + ConnectionClosed(None, None), + ] + ) + backend_ws.send = AsyncMock() + + try: + streaming, _ = await _build_streaming(client_events, backend_ws) + await streaming.backend_to_client_send_messages() + + event_types = [e.get("type") for e in client_events] + + # 1. Error event must be sent to client + error_events = [e for e in client_events if e.get("type") == "error"] + assert len(error_events) >= 1, ( + f"Expected guardrail error event, got: {event_types}" + ) + assert error_events[0]["error"]["type"] == "guardrail_violation" + + # 2. response.create must NOT have been sent to backend + sent_to_backend = [ + json.loads(c.args[0]) + for c in backend_ws.send.call_args_list + if c.args and isinstance(c.args[0], str) + ] + response_creates = [ + e for e in sent_to_backend if e.get("type") == "response.create" + ] + assert len(response_creates) == 0, ( + f"Guardrail should have stopped response.create, got: {sent_to_backend}" + ) + + # 3. Guardrail message surfaced as AI transcript delta + transcript_deltas = [ + e + for e in client_events + if e.get("type") == "response.audio_transcript.delta" + ] + assert len(transcript_deltas) >= 1, ( + f"Expected guardrail message in transcript delta, got: {event_types}" + ) + + finally: + litellm.callbacks = [] + + +@pytest.mark.asyncio +async def test_clean_text_message_passes_through_to_openai(): + """ + A clean message (no blocked phrase) must pass the guardrail and result in a real + AI response from OpenAI (response.done with non-empty output). + """ + import websockets + + guardrail = _make_guardrail(GuardrailEventHooks.pre_call) + litellm.callbacks = [guardrail] + + client_events: List[dict] = [] + + try: + async with websockets.connect( + OPENAI_REALTIME_URL, + additional_headers={ + "Authorization": f"Bearer {OPENAI_API_KEY}", + "OpenAI-Beta": "realtime=v1", + }, + ) as backend_ws: + streaming, input_queue = await _build_streaming(client_events, backend_ws) + + backend_task = asyncio.create_task( + streaming.backend_to_client_send_messages() + ) + client_task = asyncio.create_task(streaming.client_ack_messages()) + + try: + await _wait_for_event(client_events, "session.created", timeout=15) + + # Send a clean message + clean_item = json.dumps( + { + "type": "conversation.item.create", + "item": { + "role": "user", + "content": [ + {"type": "input_text", "text": "Reply with just: OK"} + ], + }, + } + ) + await input_queue.put(clean_item) + await asyncio.sleep(0.1) + await input_queue.put(json.dumps({"type": "response.create"})) + + # Wait for OpenAI to respond + await _wait_for_event(client_events, "response.done", timeout=30) + + finally: + backend_task.cancel() + client_task.cancel() + await asyncio.gather(backend_task, client_task, return_exceptions=True) + + # No guardrail error should have been sent + error_events = [e for e in client_events if e.get("type") == "error"] + guardrail_errors = [ + e for e in error_events if e.get("error", {}).get("type") == "guardrail_violation" + ] + assert len(guardrail_errors) == 0, ( + f"Clean message should not trigger guardrail, got: {guardrail_errors}" + ) + + # AI response must be present + done_events = [e for e in client_events if e.get("type") == "response.done"] + assert len(done_events) >= 1, ( + f"Expected response.done from OpenAI, got: {[e.get('type') for e in client_events]}" + ) + + finally: + litellm.callbacks = [] diff --git a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py index bcda3c7bfac..11d6bb028d8 100644 --- a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py +++ b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py @@ -637,9 +637,10 @@ async def test_realtime_text_input_guardrail_uses_pre_call_mode(): assert streaming._has_realtime_guardrails() is True, ( "pre_call guardrail should be recognized as a realtime guardrail" ) - # pre_call guardrail should NOT trigger the audio/VAD session.update injection - assert streaming._has_audio_transcription_guardrails() is False, ( - "pre_call guardrail should not trigger audio transcription guardrail path" + # pre_call guardrail SHOULD trigger the audio/VAD session.update injection so + # that the LLM does not auto-respond before the guardrail can check the transcript. + assert streaming._has_audio_transcription_guardrails() is True, ( + "pre_call guardrail should trigger audio transcription guardrail path" ) litellm.callbacks = [] # cleanup @@ -711,10 +712,11 @@ async def test_realtime_session_created_injects_session_update_for_audio_guardra @pytest.mark.asyncio -async def test_realtime_session_created_no_injection_for_pre_call_only(): +async def test_realtime_session_created_injects_session_update_for_pre_call_guardrail(): """ - Test that when only a pre_call guardrail is configured (no audio transcription), - session.created does NOT trigger the session.update injection. + Test that when a pre_call guardrail is configured, session.created triggers the + session.update injection (create_response: false) so the LLM does not auto-respond + before the guardrail can check the voice transcript. """ import litellm from litellm.integrations.custom_guardrail import CustomGuardrail @@ -751,14 +753,15 @@ async def test_realtime_session_created_no_injection_for_pre_call_only(): streaming = RealTimeStreaming(client_ws, backend_ws, logging_obj) await streaming.backend_to_client_send_messages() - # No session.update should be injected + # session.update SHOULD be injected so the LLM waits for guardrail approval sent_to_backend = [ json.loads(c.args[0]) for c in backend_ws.send.call_args_list if c.args ] session_updates = [e for e in sent_to_backend if e.get("type") == "session.update"] - assert len(session_updates) == 0, ( - f"pre_call guardrail should NOT inject session.update, got: {sent_to_backend}" + assert len(session_updates) == 1, ( + f"pre_call guardrail should inject session.update to gate audio responses, got: {sent_to_backend}" ) + assert session_updates[0]["session"]["turn_detection"]["create_response"] is False litellm.callbacks = [] # cleanup