scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-13 08:52:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 174

fix(guardrails): say the decisions warning covers only skipped question instructions

Browse source
This commit is contained in:
mateo-berri 2026-10-28 12:27:16 -07:00
parent 381cba33eb
commit a5e41d91ac
2 changed files with 30 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
litellm/proxy/utils.py
View file

@ -345,6 +345,7 @@ _VIEW_SETUP_GATE_TABLE: Final = '"LiteLLM_SpendLogs"'
_VIEW_SETUP_GATE_PROBE_ROWS: Final = TypeAdapter(tuple[Mapping[str, bool], ...])
_DELEGATED_ATTRIBUTE: Final = TypeAdapter(object)
_REQUESTED_GUARDRAILS: Final = TypeAdapter(tuple[object, ...])
_DECISIONS_CALL_TYPES: Final = frozenset((CallTypes.decisions.value, CallTypes.adecisions.value))
_ViewSetupOutcome: TypeAlias = Literal["ready", "timed_out"]
_ViewSetupAttempt: TypeAlias = Literal["ready", "table_missing"] | Exception
@ -741,12 +742,18 @@ def _warn_on_guardrails_that_cannot_scan(requested: tuple[object, ...], call_typ
)
if unscannable:
verbose_proxy_logger.warning(
"Guardrails %s do not implement apply_guardrail, so they cannot scan the text of %s requests",
"Guardrails %s do not implement apply_guardrail, so they %s",
", ".join(unscannable),
call_type,
_unscanned_request_text(call_type),
)
def _unscanned_request_text(call_type: str) -> str:
if call_type in _DECISIONS_CALL_TYPES:
return f"skip the question instructions of {call_type} requests and see the input only where their own hook reads it"
return f"cannot scan the text of {call_type} requests"
def _merge_pipeline_metadata_bucket(data: dict, bucket_key: str, modified_bucket_value: object) -> None:
if not isinstance(modified_bucket_value, dict):
return

31
tests/unit/proxy/test_proxy_logging_hook_detection.py
View file

@ -1205,9 +1205,24 @@ class _ReadsNoDecisionText(CustomGuardrail):
@pytest.mark.asyncio
@pytest.mark.parametrize(("call_type", "listed"), (("aextraction", False), ("acompletion", True)))
@pytest.mark.parametrize(
("call_type", "listed", "warning"),
(
(
"aextraction",
False,
"Guardrails requested-native, default-native do not implement apply_guardrail, so they cannot scan the text of aextraction requests",
),
(
"adecisions",
False,
"Guardrails requested-native, default-native do not implement apply_guardrail, so they skip the question instructions of adecisions requests and see the input only where their own hook reads it",
),
("acompletion", True, None),
),
)
async def test_requested_guardrail_without_apply_guardrail_is_listed_only_where_it_reads_the_text(
monkeypatch, caplog, call_type, listed
monkeypatch, caplog, call_type, listed, warning
):
requested = _ReadsNoDecisionText("requested-native")
default_on = _ReadsNoDecisionText("default-native", default_on=True)
@ -1228,14 +1243,10 @@ async def test_requested_guardrail_without_apply_guardrail_is_listed_only_where_
assert requested.native_hooks_ran == [call_type]
assert ("requested-native" in _applied_guardrails(result)) is listed
warnings = [record.getMessage() for record in caplog.records if "cannot scan" in record.getMessage()]
assert warnings == (
[]
if listed
else [
"Guardrails requested-native, default-native do not implement apply_guardrail, so they cannot scan the text of aextraction requests"
]
)
warnings = [
record.getMessage() for record in caplog.records if "do not implement apply_guardrail" in record.getMessage()
]
assert warnings == ([] if warning is None else [warning])
class _ReroutesToPrivateModel(CustomGuardrail):