mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-05 08:07:05 +00:00
fix(realtime): fix guardrails not firing for Gemini/Vertex AI and provider_config realtime WebSocket sessions (#22168)
* fix(gemini): enable inputAudioTranscription and handle transcription events for realtime guardrails Gemini sends inputTranscription/outputTranscription inside serverContent separately from modelTurn/turnComplete. This adds handling to convert them into OpenAI-compatible events so the guardrail pipeline can inspect voice input, and enables inputAudioTranscription in the session setup config. Made-with: Cursor * fix(vertex_ai): enable inputAudioTranscription in realtime session config Add inputAudioTranscription to the Vertex AI realtime setup so the backend returns transcripts of user speech, allowing guardrails to inspect voice input. Made-with: Cursor * fix(realtime): pass user_api_key_dict and guardrail metadata through async_realtime handler The base LLM HTTP handler's async_realtime method was not accepting or forwarding user_api_key_dict and litellm_metadata to RealTimeStreaming. This meant guardrails configured with default_on=false were silently skipped for all provider_config-based realtime connections (Gemini, Vertex AI, etc). Also fixes wss:// connections when SSL_VERIFY=False by overriding ssl=False for secure WebSocket URLs. Made-with: Cursor * fix(realtime): forward guardrail metadata for generic provider_config and vertex_ai paths The _arealtime function was not passing user_api_key_dict or litellm_metadata to base_llm_http_handler.async_realtime() for the generic provider_config path and the vertex_ai-specific path. This broke guardrail resolution since RealTimeStreaming.request_data was empty, causing should_run_guardrail to return False. Made-with: Cursor * fix(realtime): voice guardrail responses and block duplicate response.create on text input When a guardrail blocks voice input, send a conversation.item.create + response.create to the backend so the LLM voices the guardrail message as audio instead of only returning text. Also adds pending_guardrail_message tracking to suppress the automatic response.create the client sends after a blocked text message, and broadens _has_audio_transcription_guardrails to match pre_call/post_call modes. Made-with: Cursor * test(realtime): update guardrail tests for broadened audio transcription check and add integration tests Update existing tests to reflect that pre_call guardrails now correctly trigger the audio/VAD session.update injection. Add integration test file for live OpenAI realtime guardrail testing. Made-with: Cursor * fix(realtime): instruct LLM to say exact guardrail message verbatim The previous prompt gave the LLM creative freedom to paraphrase the guardrail violation message. Now it instructs the LLM to repeat the exact configured message word for word. Made-with: Cursor * fix(realtime): preserve wss ssl semantics and move live guardrail test Keep TLS enabled for wss realtime sessions while honoring SSL_VERIFY=False via a no-verify SSLContext, move the OpenAI live guardrail test into llm_translation, and dedupe duplicated guardrail-detection helpers to prevent drift. Made-with: Cursor
This commit is contained in:
Ishaan Jaff
Sameer Kankute
parent
cbeaebf826
commit
e40a2a2bd6
7 changed files with 481 additions and 32 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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"] = {}
|
||||
|
|
|
|||
|
|
@ -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": {},
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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}")
|
||||
|
|
|
|||
|
|
@ -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 = []
|
||||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue