Skip to content

fix(realtime): guardrails with pre_call/post_call mode now work on realtime WebSocket #22161

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ishaan-jaff merged 3 commits into from
Feb 26, 2026
Merged

fix(realtime): guardrails with pre_call/post_call mode now work on realtime WebSocket #22161

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f3300e6
fix(realtime): guardrails with pre_call/post_call mode now work on re…
ishaan-jaff Feb 26, 2026
758d548
fix(realtime guardrails): address code review feedback
ishaan-jaff Feb 26, 2026
1d2445d
simplify: remove response.create swallowing after guardrail block
ishaan-jaff Feb 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
144 changes: 87 additions & 57 deletions litellm/litellm_core_utils/realtime_streaming.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ def __init__(
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
Expand All @@ -68,6 +69,7 @@ def __init__(
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 {}

def _should_store_message(
self,
Expand Down Expand Up @@ -231,14 +233,40 @@ async def _send_to_backend(self, message: str) -> None:
await self.backend_ws.send(message)

def _has_realtime_guardrails(self) -> bool:
"""Return True if any callback is registered for realtime_input_transcription."""
"""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.
"""
from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.types.guardrails import GuardrailEventHooks

return any(
isinstance(cb, CustomGuardrail)
and cb.should_run_guardrail(
data={},
data=self.request_data,
event_type=GuardrailEventHooks.realtime_input_transcription,
)
for cb in litellm.callbacks
Expand All @@ -258,17 +286,25 @@ async def run_realtime_guardrails(
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 (
callback.should_run_guardrail(
data={"transcript": transcript},
event_type=GuardrailEventHooks.realtime_input_transcription,
)
is not True
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
Comment on lines +289 to 306

@greptile-apps greptile-apps Bot Feb 26, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

post_call guardrails run on user input, not LLM output

run_realtime_guardrails checks all three event types (pre_call, post_call, realtime_input_transcription) when deciding whether to run a guardrail. This means a guardrail configured with mode: post_call — which is semantically meant to run on the LLM's response — will instead be called on the user's input text during both the conversation.item.create path and the audio transcription path.

A post_call guardrail author may have written detection logic that only makes sense for response content (e.g., checking for hallucinations, formatting violations, or PII in the assistant's reply). Running that same apply_guardrail on user input text could produce false positives or false negatives.

Consider either:

  1. Only matching pre_call and realtime_input_transcription for the input-side check (since those are semantically about guarding input), or
  2. Documenting that post_call guardrails will also gate user input in the realtime context so guardrail authors can account for it.

_already_run.add(id(callback))
try:
await callback.apply_guardrail(
inputs={"texts": [transcript], "images": []},
Expand All @@ -293,20 +329,15 @@ async def run_realtime_guardrails(
safe_msg = str(detail)
else:
safe_msg = str(e) or "I'm sorry, that request was blocked by the content filter."
# Cancel any in-flight response before speaking the warning.
# This handles the race where create_response fired before we could intercept.
await self._send_to_backend(json.dumps({"type": "response.cancel"}))
# Ask the model to speak the warning — TTS audio plays naturally in the client
await self._send_to_backend(
# Return the error directly to the WebSocket consumer.
await self.websocket.send_text(
json.dumps(
{
"type": "response.create",
"response": {
"modalities": ["text", "audio"],
"instructions": (
f"Say exactly and only: \"{safe_msg}\". "
"Do not add anything else."
),
"type": "error",
"error": {
"type": "guardrail_violation",
"message": safe_msg,
"code": "content_policy_violation",
},
}
)
Expand Down Expand Up @@ -348,25 +379,25 @@ async def _handle_provider_config_message(self, raw_response) -> None:
if isinstance(transformed_response, list)
else [transformed_response]
)
for event in events:
## GUARDRAIL: inject create_response=false on session.created
if isinstance(event, dict) and event.get("type") == "session.created":
if self._has_realtime_guardrails():
await self._send_to_backend(
json.dumps(
{
"type": "session.update",
"session": {
"turn_detection": {
"type": "server_vad",
"create_response": False,
}
},
}
)
)
for event in events:

@greptile-apps greptile-apps Bot Feb 26, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Removal of create_response: false injection may cause double responses for audio/VAD path

The previous code injected session.update with create_response: false into the backend session when guardrails were registered. This prevented the LLM from auto-responding to VAD speech completions before the guardrail had a chance to run.

With this code removed, when server VAD detects speech completion, the backend will auto-generate a response (create_response defaults to true). Then, after the transcription arrives and guardrails pass, the proxy also sends a manual response.create (lines 383-384 in the provider_config path, and lines 415-417 in the raw path). This could result in two LLM responses for each clean audio transcription.

For the new text-input (conversation.item.create) path this isn't an issue since the client explicitly controls response.create. But for the existing audio transcription guardrail flow, this removal appears to be a regression. Consider keeping the create_response: false injection for audio-based guardrails, or gating the proxy's manual response.create on whether auto-response is already enabled.

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)
Expand Down Expand Up @@ -397,27 +428,26 @@ async def _handle_raw_backend_message(self, raw_response) -> bool:
try:
event_obj = json.loads(raw_response)

if event_obj.get("type") == "session.created":
# If any realtime guardrails are registered, proactively
# set create_response=false so the LLM never auto-responds
# before our guardrail has a chance to run.
if self._has_realtime_guardrails():
await self._send_to_backend(
json.dumps(
{
"type": "session.update",
"session": {
"turn_detection": {
"type": "server_vad",
"create_response": False,
}
},
}
)
)
verbose_logger.debug(
"[realtime guardrail] injected create_response=false into session"
# 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")
Expand Down
10 changes: 8 additions & 2 deletions litellm/llms/azure/realtime/handler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,13 +6,13 @@

from typing import Any, Optional, cast

from litellm._logging import verbose_proxy_logger
from litellm.constants import REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES

from ....litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
from ....litellm_core_utils.realtime_streaming import RealTimeStreaming
from ....llms.custom_httpx.http_handler import get_shared_realtime_ssl_context
from ..azure import AzureChatCompletion
from litellm._logging import verbose_proxy_logger

# BACKEND_WS_URL = "ws://localhost:8080/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01"

Expand Down Expand Up @@ -77,6 +77,8 @@ async def async_realtime(
client: Optional[Any] = None,
timeout: Optional[float] = None,
realtime_protocol: Optional[str] = None,
user_api_key_dict: Optional[Any] = None,
litellm_metadata: Optional[dict] = None,
):
import websockets
from websockets.asyncio.client import ClientConnection
Expand All @@ -101,7 +103,11 @@ async def async_realtime(
ssl=ssl_context,
) as backend_ws:
realtime_streaming = RealTimeStreaming(
websocket, cast(ClientConnection, backend_ws), logging_obj
websocket,
cast(ClientConnection, backend_ws),
logging_obj,
user_api_key_dict=user_api_key_dict,
request_data={"litellm_metadata": litellm_metadata or {}},
)
await realtime_streaming.bidirectional_forward()

Expand Down
2 changes: 2 additions & 0 deletions litellm/llms/openai/realtime/handler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,7 @@ async def async_realtime(
timeout: Optional[float] = None,
query_params: Optional[RealtimeQueryParams] = None,
user_api_key_dict: Optional[Any] = None,
litellm_metadata: Optional[dict] = None,
**kwargs: Any,
):
import websockets
Expand Down Expand Up @@ -142,6 +143,7 @@ async def async_realtime(
cast(ClientConnection, backend_ws),
logging_obj,
user_api_key_dict=user_api_key_dict,
request_data={"litellm_metadata": litellm_metadata or {}},
)
await realtime_streaming.bidirectional_forward()

Expand Down
14 changes: 14 additions & 0 deletions litellm/realtime_api/main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,15 @@
base_llm_http_handler = BaseLLMHTTPHandler()


def _build_litellm_metadata(kwargs: dict) -> dict:
"""Build the litellm_metadata dict for guardrail checking (internal only, not forwarded to provider)."""
metadata: dict = {**(kwargs.get("litellm_metadata") or {})}
guardrails = (kwargs.get("metadata") or {}).get("guardrails") or kwargs.get("guardrails") or []
if guardrails:
metadata["guardrails"] = guardrails
return metadata


@wrapper_client
async def _arealtime(
model: str,
Expand Down Expand Up @@ -134,6 +143,8 @@ async def _arealtime(
timeout=timeout,
logging_obj=litellm_logging_obj,
realtime_protocol=realtime_protocol,
user_api_key_dict=kwargs.get("user_api_key_dict"),
litellm_metadata=_build_litellm_metadata(kwargs),
)
elif _custom_llm_provider == "openai":
api_base = (
Expand All @@ -160,6 +171,7 @@ async def _arealtime(
timeout=timeout,
query_params=query_params,
user_api_key_dict=kwargs.get("user_api_key_dict"),
litellm_metadata=_build_litellm_metadata(kwargs),
)
elif _custom_llm_provider == "bedrock":
# Extract AWS parameters from kwargs
Expand Down Expand Up @@ -217,6 +229,8 @@ async def _arealtime(
client=None,
timeout=timeout,
query_params=query_params,
user_api_key_dict=kwargs.get("user_api_key_dict"),
litellm_metadata=_build_litellm_metadata(kwargs),
)
elif _custom_llm_provider == "vertex_ai":
vertex_credentials = (
Expand Down
Loading
Loading