scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-11 22:51:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 124

refactor(realtime-guardrails): reuse pre_call/post_call hooks instead of realtime-specific hooks

Browse source 
Removes realtime_input_transcription and realtime_output_text from
GuardrailEventHooks. Realtime input guardrails now fire on pre_call,
realtime output guardrails now fire on post_call.

pre_call and post_call are cross-modality promises — a guardrail
registered with mode: pre_call runs on regular completions AND realtime
input; post_call runs on completions AND realtime output text.
This commit is contained in:
Ishaan Jaffer 2026-02-25 20:53:29 -08:00
parent b890e3cca5
commit 9578efef61
5 changed files with 17 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
litellm/litellm_core_utils/realtime_streaming.py
View file

@ -214,7 +214,7 @@ class RealTimeStreaming:
executor.submit(self.logging_obj.success_handler(self.messages))
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 pre_call (runs across all modalities)."""
from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.types.guardrails import GuardrailEventHooks
@ -222,7 +222,7 @@ class RealTimeStreaming:
isinstance(cb, CustomGuardrail)
and cb.should_run_guardrail(
data={},
event_type=GuardrailEventHooks.realtime_input_transcription,
event_type=GuardrailEventHooks.pre_call,
)
for cb in litellm.callbacks
)
@ -247,7 +247,7 @@ class RealTimeStreaming:
if (
callback.should_run_guardrail(
data={"transcript": transcript},
event_type=GuardrailEventHooks.realtime_input_transcription,
event_type=GuardrailEventHooks.pre_call,
)
is not True
):
@ -302,7 +302,7 @@ class RealTimeStreaming:
return False
def _has_realtime_output_guardrails(self) -> bool:
"""Return True if any callback is registered for realtime_output_text."""
"""Return True if any callback is registered for post_call (runs across all modalities)."""
from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.types.guardrails import GuardrailEventHooks
@ -310,7 +310,7 @@ class RealTimeStreaming:
isinstance(cb, CustomGuardrail)
and cb.should_run_guardrail(
data={},
event_type=GuardrailEventHooks.realtime_output_text,
event_type=GuardrailEventHooks.post_call,
)
for cb in litellm.callbacks
)
@ -332,7 +332,7 @@ class RealTimeStreaming:
if (
callback.should_run_guardrail(
data={"text": text},
event_type=GuardrailEventHooks.realtime_output_text,
event_type=GuardrailEventHooks.post_call,
)
is not True
):

2
litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py
View file

@ -193,8 +193,6 @@ class ContentFilterGuardrail(CustomGuardrail):
GuardrailEventHooks.pre_call,
GuardrailEventHooks.post_call,
GuardrailEventHooks.during_call,
GuardrailEventHooks.realtime_input_transcription,
GuardrailEventHooks.realtime_output_text,
],
event_hook=event_hook or GuardrailEventHooks.pre_call,
default_on=default_on,

2
litellm/types/guardrails.py
View file

@ -773,8 +773,6 @@ class GuardrailEventHooks(str, Enum):
logging_only = "logging_only"
pre_mcp_call = "pre_mcp_call"
during_mcp_call = "during_mcp_call"
realtime_input_transcription = "realtime_input_transcription"
realtime_output_text = "realtime_output_text"
class DynamicGuardrailParams(TypedDict):

6
tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
View file

@ -383,7 +383,7 @@ async def test_realtime_guardrail_blocks_prompt_injection():
guardrail = PromptInjectionGuardrail(
guardrail_name="test_injection_guard",
event_hook=GuardrailEventHooks.realtime_input_transcription,
event_hook=GuardrailEventHooks.pre_call,
default_on=True,
)
litellm.callbacks = [guardrail]
@ -466,7 +466,7 @@ async def test_realtime_guardrail_allows_clean_transcript():
guardrail = PromptInjectionGuardrail(
guardrail_name="test_injection_guard",
event_hook=GuardrailEventHooks.realtime_input_transcription,
event_hook=GuardrailEventHooks.pre_call,
default_on=True,
)
litellm.callbacks = [guardrail]
@ -530,7 +530,7 @@ async def test_realtime_session_created_injects_create_response_false():
guardrail = DummyGuardrail(
guardrail_name="dummy",
event_hook=GuardrailEventHooks.realtime_input_transcription,
event_hook=GuardrailEventHooks.pre_call,
default_on=True,
)
litellm.callbacks = [guardrail]

16
tests/test_litellm/realtime/test_output_guardrail.py
View file

@ -47,8 +47,8 @@ class _BlockingGuardrail(CustomGuardrail):
def __init__(self):
super().__init__(
guardrail_name="test-content-filter",
supported_event_hooks=[GuardrailEventHooks.realtime_output_text],
event_hook=GuardrailEventHooks.realtime_output_text,
supported_event_hooks=[GuardrailEventHooks.post_call],
event_hook=GuardrailEventHooks.post_call,
default_on=True,
)
@ -71,8 +71,8 @@ class _PassthroughGuardrail(CustomGuardrail):
def __init__(self):
super().__init__(
guardrail_name="test-passthrough",
supported_event_hooks=[GuardrailEventHooks.realtime_output_text],
event_hook=GuardrailEventHooks.realtime_output_text,
supported_event_hooks=[GuardrailEventHooks.post_call],
event_hook=GuardrailEventHooks.post_call,
default_on=True,
)
@ -287,11 +287,11 @@ class TestOutputGuardrailNoGuardrail:
class TestGuardrailEventHook:
"""realtime_output_text is a valid GuardrailEventHooks enum value."""
"""post_call runs across all modalities including realtime output."""
def test_hook_exists(self):
from litellm.types.guardrails import GuardrailEventHooks
assert hasattr(GuardrailEventHooks, "realtime_output_text")
assert GuardrailEventHooks.realtime_output_text == "realtime_output_text"
print(f"\n✅ PASS: GuardrailEventHooks.realtime_output_text = {GuardrailEventHooks.realtime_output_text!r}")
assert hasattr(GuardrailEventHooks, "post_call")
assert GuardrailEventHooks.post_call == "post_call"
print(f"\n✅ PASS: GuardrailEventHooks.post_call = {GuardrailEventHooks.post_call!r}")