scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-08 03:08:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 115

Merge pull request #34411 from BerriAI/litellm_lit4650_passthrough_guardrail_log

Browse source 
fix(guardrails): stop reporting a no-op guardrail as applied on passthrough
This commit is contained in:
tin-berri 2026-07-24 16:52:46 -07:00 • committed by GitHub
commit d389f837ae
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 277 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
litellm/integrations/custom_guardrail.py
View file

@ -107,6 +107,8 @@ class CustomGuardrail(CustomLogger):
# If True, during_call runs async_moderation_hook instead of the unified apply_guardrail path.
use_native_during_call_hook: ClassVar[bool] = False
records_own_guardrail_information: ClassVar[bool] = False
def __init__(
self,
guardrail_name: Optional[str] = None,
@ -1256,6 +1258,14 @@ def log_guardrail_information(func):
so it stays correct when guardrails run concurrently (asyncio copies the
context into each gathered task): counting shared entries would let one
guardrail's append hide another guardrail's missing record.
A guardrail that only records an entry when it actually runs (e.g.
``HeadroomGuardrail``, which returns the inputs untouched on an endpoint
whose payload it cannot act on) sets ``records_own_guardrail_information =
True`` so the auto-record is skipped even on the return paths where it
recorded nothing; otherwise a no-op early return would be logged as an
"allow"/"success" run even though the guardrail did nothing. The exception
branch below still records so a genuine failure is not lost.
"""
import functools
import inspect
@ -1291,7 +1301,7 @@ def log_guardrail_information(func):
self_recorded_token = _guardrail_self_recorded.set(False)
try:
response = await func(*args, **kwargs)
if _guardrail_self_recorded.get():
if self.records_own_guardrail_information or _guardrail_self_recorded.get():
return response
return self._process_response(
response=response,
@ -1333,7 +1343,7 @@ def log_guardrail_information(func):
self_recorded_token = _guardrail_self_recorded.set(False)
try:
response = func(*args, **kwargs)
if _guardrail_self_recorded.get():
if self.records_own_guardrail_information or _guardrail_self_recorded.get():
return response
return self._process_response(
response=response,

19
litellm/proxy/guardrails/guardrail_hooks/headroom/headroom.py
View file

@ -4,7 +4,7 @@ import json
import re
import time
import uuid
from typing import TYPE_CHECKING, Any, List, Literal, Optional
from typing import TYPE_CHECKING, Any, ClassVar, List, Literal, Optional
import httpx
from fastapi import HTTPException
@ -209,6 +209,8 @@ def _build_responses_followup_items(
class HeadroomGuardrail(CustomGuardrail):
records_own_guardrail_information: ClassVar[bool] = True
@classmethod
def get_supported_event_hooks(cls) -> List[GuardrailEventHooks]:
return [
@ -481,7 +483,21 @@ class HeadroomGuardrail(CustomGuardrail):
)
end_time = time.time()
from litellm.proxy.common_utils.callback_utils import (
add_guardrail_to_applied_guardrails_header,
)
if not compression_succeeded:
self.add_standard_logging_guardrail_information_to_request_data(
guardrail_json_response={"error": "headroom compression unavailable; request forwarded uncompressed"},
request_data=request_data,
guardrail_status="guardrail_failed_to_respond",
guardrail_provider=HEADROOM_GUARDRAIL_PROVIDER,
start_time=start_time,
end_time=end_time,
duration=end_time - start_time,
)
add_guardrail_to_applied_guardrails_header(request_data=request_data, guardrail_name=self.guardrail_name)
return {**inputs, "structured_messages": compressed} # pyright: ignore[reportReturnType]
self.add_standard_logging_guardrail_information_to_request_data(
@ -493,6 +509,7 @@ class HeadroomGuardrail(CustomGuardrail):
end_time=end_time,
duration=end_time - start_time,
)
add_guardrail_to_applied_guardrails_header(request_data=request_data, guardrail_name=self.guardrail_name)
hashes = extract_hashes_from_messages(compressed)
if not hashes:

12
litellm/proxy/guardrails/guardrail_hooks/unified_guardrail/unified_guardrail.py
View file

@ -147,8 +147,10 @@ class UnifiedLLMGuardrails(CustomLogger):
litellm_logging_obj=data.get("litellm_logging_obj"),
)
# Add guardrail to applied guardrails header
add_guardrail_to_applied_guardrails_header(request_data=data, guardrail_name=guardrail_to_apply.guardrail_name)
if not guardrail_to_apply.records_own_guardrail_information:
add_guardrail_to_applied_guardrails_header(
request_data=data, guardrail_name=guardrail_to_apply.guardrail_name
)
return data
async def async_moderation_hook(
@ -274,8 +276,10 @@ class UnifiedLLMGuardrails(CustomLogger):
if e.original_response is None:
e.original_response = response
raise
# Add guardrail to applied guardrails header
add_guardrail_to_applied_guardrails_header(request_data=data, guardrail_name=guardrail_to_apply.guardrail_name)
if not guardrail_to_apply.records_own_guardrail_information:
add_guardrail_to_applied_guardrails_header(
request_data=data, guardrail_name=guardrail_to_apply.guardrail_name
)
return response

59
tests/test_litellm/integrations/test_custom_guardrail.py
View file

@ -3,9 +3,12 @@ from unittest.mock import AsyncMock
import pytest
from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.integrations.custom_guardrail import (
CustomGuardrail,
log_guardrail_information,
)
from litellm.proxy._types import CallTypes, UserAPIKeyAuth
from litellm.types.utils import GuardrailTracingDetail
from litellm.types.utils import GenericGuardrailAPIInputs, GuardrailTracingDetail
class TestCustomGuardrailDeploymentHook:
@ -1947,3 +1950,55 @@ class TestOnlyScanNewMessages:
cache.async_set_cache = AsyncMock(side_effect=RuntimeError("redis down"))
await guardrail.mark_texts_scanned(texts=["a"], request_data={"litellm_session_id": "s1"}, cache=cache)
def _guardrail_entries(request_data: dict) -> list:
container = request_data.get("metadata") or request_data.get("litellm_metadata") or {}
entries = container.get("standard_logging_guardrail_information")
return entries if isinstance(entries, list) else []
class _NoopGuardrail(CustomGuardrail):
"""apply_guardrail that returns the inputs untouched and records nothing."""
@log_guardrail_information
async def apply_guardrail(self, inputs, request_data, input_type, logging_obj=None):
return inputs
class _NoopSelfLoggingGuardrail(_NoopGuardrail):
records_own_guardrail_information = True
class TestRecordsOwnGuardrailInformation:
"""The @log_guardrail_information decorator must not synthesize an "allow"/"success"
entry for a no-op apply_guardrail when the guardrail sets
records_own_guardrail_information (LIT-4650)."""
@pytest.mark.asyncio
async def test_default_noop_apply_guardrail_is_auto_logged(self):
guardrail = _NoopGuardrail(guardrail_name="g1")
request_data: dict = {"model": "gpt-4o"}
await guardrail.apply_guardrail(
inputs=GenericGuardrailAPIInputs(texts=["x"]),
request_data=request_data,
input_type="request",
)
entries = _guardrail_entries(request_data)
assert len(entries) == 1
assert entries[0]["guardrail_status"] == "success"
@pytest.mark.asyncio
async def test_self_logging_noop_apply_guardrail_is_not_logged(self):
guardrail = _NoopSelfLoggingGuardrail(guardrail_name="g2")
request_data: dict = {"model": "gpt-4o"}
await guardrail.apply_guardrail(
inputs=GenericGuardrailAPIInputs(texts=["x"]),
request_data=request_data,
input_type="request",
)
assert _guardrail_entries(request_data) == []

76
tests/test_litellm/proxy/guardrails/guardrail_hooks/test_headroom.py
View file

@ -114,6 +114,24 @@ def guardrail() -> HeadroomGuardrail:
return _make_guardrail()
def _recorded_guardrail_entries(request_data: dict) -> list:
for container_key in ("metadata", "litellm_metadata"):
container = request_data.get(container_key)
if isinstance(container, dict):
entries = container.get("standard_logging_guardrail_information")
if isinstance(entries, list):
return entries
return []
def _applied_guardrails(request_data: dict) -> list:
for container_key in ("metadata", "litellm_metadata"):
container = request_data.get(container_key)
if isinstance(container, dict) and isinstance(container.get("applied_guardrails"), list):
return container["applied_guardrails"]
return []
@pytest.mark.asyncio
async def test_apply_guardrail_compresses_and_returns_structured_messages(
guardrail: HeadroomGuardrail,
@ -123,6 +141,7 @@ async def test_apply_guardrail_compresses_and_returns_structured_messages(
structured_messages=ORIGINAL_MESSAGES,
)
mock_response = _make_compress_response(COMPRESSED_MESSAGES)
request_data = {"model": "gpt-4o"}
with patch.object(
guardrail.async_handler,
@ -132,12 +151,19 @@ async def test_apply_guardrail_compresses_and_returns_structured_messages(
):
result = await guardrail.apply_guardrail(
inputs=inputs,
request_data={"model": "gpt-4o"},
request_data=request_data,
input_type="request",
)
assert result.get("structured_messages") == COMPRESSED_MESSAGES
entries = _recorded_guardrail_entries(request_data)
assert len(entries) == 1
assert entries[0]["guardrail_name"] == "headroom"
assert entries[0]["guardrail_status"] == "success"
assert entries[0]["guardrail_provider"] == "headroom"
assert "headroom" in _applied_guardrails(request_data)
@pytest.mark.asyncio
async def test_apply_guardrail_injects_retrieve_tool_when_hashes_present(
@ -719,6 +745,7 @@ async def test_apply_guardrail_bypass_header_skips_compression(
mock_post.assert_not_called()
assert result.get("structured_messages") == ORIGINAL_MESSAGES
assert _recorded_guardrail_entries(request_data) == []
@pytest.mark.asyncio
@ -729,16 +756,18 @@ async def test_apply_guardrail_response_type_passthrough(
texts=["some response text"],
structured_messages=ORIGINAL_MESSAGES,
)
request_data: dict = {"model": "gpt-4o"}
with patch.object(guardrail.async_handler, "post", new_callable=AsyncMock) as mock_post:
result = await guardrail.apply_guardrail(
inputs=inputs,
request_data={},
request_data=request_data,
input_type="response",
)
mock_post.assert_not_called()
assert result is inputs
assert _recorded_guardrail_entries(request_data) == []
@pytest.mark.asyncio
@ -746,16 +775,48 @@ async def test_apply_guardrail_empty_structured_messages_passthrough(
guardrail: HeadroomGuardrail,
):
inputs = GenericGuardrailAPIInputs(texts=["hello"])
request_data: dict = {"model": "gpt-4o"}
with patch.object(guardrail.async_handler, "post", new_callable=AsyncMock) as mock_post:
result = await guardrail.apply_guardrail(
inputs=inputs,
request_data={},
request_data=request_data,
input_type="request",
)
mock_post.assert_not_called()
assert result is inputs
assert _recorded_guardrail_entries(request_data) == []
assert "headroom" not in _applied_guardrails(request_data)
@pytest.mark.asyncio
async def test_passthrough_handler_does_not_log_headroom_as_run(
guardrail: HeadroomGuardrail,
):
"""Regression for LIT-4650.
A passthrough request drives headroom through PassThroughEndpointHandler, which
only supplies `texts` (no `structured_messages`). Headroom cannot compress that
shape and no-ops, so it must not appear in the spend log's
standard_logging_guardrail_information as a successful run.
"""
from litellm.llms.pass_through.guardrail_translation.handler import (
PassThroughEndpointHandler,
)
data = {"model": "gpt-4o", "messages": [{"role": "user", "content": "hello"}]}
with patch.object(guardrail.async_handler, "post", new_callable=AsyncMock) as mock_post:
await PassThroughEndpointHandler().process_input_messages(
data=data,
guardrail_to_apply=guardrail,
litellm_logging_obj=None,
)
mock_post.assert_not_called()
assert _recorded_guardrail_entries(data) == []
assert "headroom" not in _applied_guardrails(data)
@pytest.mark.asyncio
@ -884,6 +945,7 @@ async def test_apply_guardrail_transport_error_fail_open_forwards_uncompressed()
texts=["hello"],
structured_messages=ORIGINAL_MESSAGES,
)
request_data = {"model": "gpt-4o"}
with patch.object(
guardrail.async_handler,
@ -893,12 +955,18 @@ async def test_apply_guardrail_transport_error_fail_open_forwards_uncompressed()
):
result = await guardrail.apply_guardrail(
inputs=inputs,
request_data={},
request_data=request_data,
input_type="request",
)
assert result["structured_messages"] == ORIGINAL_MESSAGES
entries = _recorded_guardrail_entries(request_data)
assert len(entries) == 1
assert entries[0]["guardrail_name"] == "headroom"
assert entries[0]["guardrail_status"] == "guardrail_failed_to_respond"
assert "headroom" in _applied_guardrails(request_data)
@pytest.mark.asyncio
async def test_apply_guardrail_http_error_fail_open_forwards_uncompressed():

111
tests/test_litellm/proxy/guardrails/guardrail_hooks/unified_guardrails/test_unified_guardrail.py
View file

@ -4,7 +4,10 @@ import pytest
import litellm
from litellm.caching import DualCache
from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.integrations.custom_guardrail import (
CustomGuardrail,
log_guardrail_information,
)
from litellm.llms.base_llm.guardrail_translation.base_translation import BaseTranslation
from litellm.llms.base_llm.guardrail_translation.utils import (
effective_skip_system_message_for_guardrail,
@ -1490,3 +1493,109 @@ class TestStreamingTransform:
# None holdback treated as 0: full text emitted, no crash.
assert "".join(_delta_text(i) for i in out) == "ABCDEF"
def _applied_guardrails(data: dict) -> list:
for key in ("metadata", "litellm_metadata"):
meta = data.get(key)
if isinstance(meta, dict) and isinstance(meta.get("applied_guardrails"), list):
return meta["applied_guardrails"]
return []
class _TextsOnlyTranslation(BaseTranslation):
"""Mimics a passthrough handler: hands the guardrail only `texts`, never
structured_messages, so a structured_messages-based guardrail no-ops."""
async def process_input_messages(self, data, guardrail_to_apply, litellm_logging_obj=None): # type: ignore[override]
await guardrail_to_apply.apply_guardrail(
inputs={"texts": ["payload"]},
request_data=data,
input_type="request",
logging_obj=litellm_logging_obj,
)
return data
async def process_output_response( # type: ignore[override]
self,
response,
guardrail_to_apply,
litellm_logging_obj=None,
user_api_key_dict=None,
request_data=None,
):
return response
class _SelfLoggingGuardrail(CustomGuardrail):
records_own_guardrail_information = True
def __init__(self, *, self_add: bool):
super().__init__(guardrail_name="self-logging")
self._self_add = self_add
def should_run_guardrail(self, data, event_type): # type: ignore[override]
return True
@log_guardrail_information
async def apply_guardrail(self, inputs, request_data, input_type, **kwargs):
if self._self_add:
from litellm.proxy.common_utils.callback_utils import (
add_guardrail_to_applied_guardrails_header,
)
add_guardrail_to_applied_guardrails_header(request_data=request_data, guardrail_name=self.guardrail_name)
return inputs
class _AutoLoggingGuardrail(CustomGuardrail):
def __init__(self):
super().__init__(guardrail_name="auto-logging")
def should_run_guardrail(self, data, event_type): # type: ignore[override]
return True
@log_guardrail_information
async def apply_guardrail(self, inputs, request_data, input_type, **kwargs):
return inputs
class TestAppliedGuardrailsReflectsExecution:
"""The unified hook must not auto-mark a self-logging guardrail
(records_own_guardrail_information) as applied; such a guardrail owns that
decision and marks itself only when it actually ran (LIT-4650). Ordinary
guardrails are still auto-marked by the hook after dispatch."""
@staticmethod
def _data(guardrail):
return {
"guardrail_to_apply": guardrail,
"model": "gpt-4o",
"messages": [{"role": "user", "content": "hello world"}],
}
async def _run(self, guardrail):
unified_module.endpoint_guardrail_translation_mappings = {CallTypes.pass_through: _TextsOnlyTranslation}
data = self._data(guardrail)
await UnifiedLLMGuardrails().async_pre_call_hook(
user_api_key_dict=None,
cache=DualCache(),
data=data,
call_type=CallTypes.pass_through.value,
)
return data
@pytest.mark.asyncio
async def test_self_logging_guardrail_is_not_auto_marked_applied(self):
data = await self._run(_SelfLoggingGuardrail(self_add=False))
assert "self-logging" not in _applied_guardrails(data)
@pytest.mark.asyncio
async def test_self_logging_guardrail_that_self_marks_is_applied(self):
data = await self._run(_SelfLoggingGuardrail(self_add=True))
assert "self-logging" in _applied_guardrails(data)
@pytest.mark.asyncio
async def test_ordinary_guardrail_is_auto_marked_applied(self):
data = await self._run(_AutoLoggingGuardrail())
assert "auto-logging" in _applied_guardrails(data)