scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-14 23:21:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 102

fix(guardrails): log non-blocking Conduct verdicts in standard guardrail information

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng 2026-09-11 22:44:22 +00:00
parent 97a45a4f65
commit 7ffde11054
2 changed files with 125 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

49
litellm/proxy/guardrails/guardrail_hooks/conduct/conduct.py
View file

@ -7,15 +7,18 @@ Source: https://github.com/sseshachala/conductai/tree/main/packages/conduct-lit
from __future__ import annotations
from collections.abc import Awaitable, Callable, Mapping
from functools import partial
from types import MappingProxyType
from typing import TYPE_CHECKING, Final, Literal, Protocol
from pydantic import BaseModel, ConfigDict
from litellm.integrations.custom_guardrail import CustomGuardrail, log_guardrail_information
from litellm.types.llms.openai import ChatCompletionUserMessage
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.types.utils import GenericGuardrailAPIInputs
from litellm.types.utils import GenericGuardrailAPIInputs, GuardrailStatus
MISSING_PACKAGE_MESSAGE: Final = (
"conduct-litellm-guard is required for the Conduct guardrail. "
@ -23,12 +26,16 @@ MISSING_PACKAGE_MESSAGE: Final = (
)
BLOCKING_VERDICTS: Final = frozenset({"block", "approval"})
FLAGGED_VERDICTS: Final = frozenset({"warning", "advisory"})
class ConductDecision(Protocol):
@property
def verdict(self) -> str: ...
@property
def rule_id(self) -> str | None: ...
class ConductCheck(Protocol):
def __call__(self, *, data: Mapping[str, object], call_type: str) -> Awaitable[ConductDecision]: ...
@ -47,12 +54,36 @@ def request_payload(
return MappingProxyType({**request_data, "prompt": None, "messages": messages})
def decision_status(decision: ConductDecision) -> GuardrailStatus:
return "guardrail_flagged" if decision.verdict in FLAGGED_VERDICTS else "success"
class ConductVerdict(BaseModel):
model_config = ConfigDict(frozen=True)
verdict: str
rule_id: str | None = None
def record_decision(
guardrail: CustomGuardrail,
request_data: dict[str, object], # mutable-ok: the logging helper writes metadata into it
decision: ConductDecision,
) -> None:
guardrail.add_standard_logging_guardrail_information_to_request_data(
guardrail_json_response=ConductVerdict(verdict=decision.verdict, rule_id=decision.rule_id).model_dump(),
request_data=request_data,
guardrail_status=decision_status(decision),
)
async def apply_conduct_guardrail(
inputs: GenericGuardrailAPIInputs,
request_data: Mapping[str, object],
input_type: Literal["request", "response"],
check: ConductCheck,
blocked: Callable[[ConductDecision], Exception],
record: Callable[[ConductDecision], None],
) -> GenericGuardrailAPIInputs:
payload: Final = request_payload(inputs, request_data, input_type)
if payload is None:
@ -60,6 +91,7 @@ async def apply_conduct_guardrail(
decision: Final = await check(data=payload, call_type=input_type)
if decision.verdict in BLOCKING_VERDICTS:
raise blocked(decision)
record(decision)
return inputs
@ -79,19 +111,30 @@ else:
async def apply_guardrail(
self,
inputs: GenericGuardrailAPIInputs,
request_data: Mapping[str, object],
request_data: dict[str, object], # mutable-ok: CustomGuardrail.apply_guardrail contract
input_type: Literal["request", "response"],
logging_obj: LiteLLMLoggingObj | None = None,
) -> GenericGuardrailAPIInputs:
return await apply_conduct_guardrail(inputs, request_data, input_type, self.check, ConductGuardBlocked)
return await apply_conduct_guardrail(
inputs,
request_data,
input_type,
self.check,
ConductGuardBlocked,
partial(record_decision, self, request_data),
)
__all__ = (
"BLOCKING_VERDICTS",
"FLAGGED_VERDICTS",
"MISSING_PACKAGE_MESSAGE",
"ConductCheck",
"ConductDecision",
"ConductGuardrail",
"ConductVerdict",
"apply_conduct_guardrail",
"decision_status",
"record_decision",
"request_payload",
)

88
tests/test_litellm/proxy/guardrails/guardrail_hooks/test_conduct.py
View file

@ -4,7 +4,7 @@ import importlib.util
import json
from collections.abc import Mapping
from dataclasses import dataclass, field
from typing import Final
from typing import Final, Literal
import httpx
import pytest
@ -18,7 +18,11 @@ from litellm.proxy.guardrails.guardrail_hooks.conduct import (
ConductGuardrail,
initialize_guardrail,
)
from litellm.proxy.guardrails.guardrail_hooks.conduct.conduct import apply_conduct_guardrail, request_payload
from litellm.proxy.guardrails.guardrail_hooks.conduct.conduct import (
apply_conduct_guardrail,
record_decision,
request_payload,
)
from litellm.proxy.guardrails.guardrail_endpoints import get_guardrail_ui_settings
from litellm.proxy.guardrails.guardrail_registry import InMemoryGuardrailHandler
from litellm.types.guardrails import Guardrail, GuardrailEventHooks, LitellmParams
@ -58,6 +62,7 @@ class _RecordingGuardrail(CustomGuardrail):
@dataclass(frozen=True, slots=True)
class _Decision:
verdict: str
rule_id: str | None = None
class _Blocked(Exception):
@ -69,11 +74,33 @@ class _Blocked(Exception):
@dataclass(slots=True)
class _RecordingCheck:
verdict: str
rule_id: str | None = None
calls: list[tuple[Mapping[str, object], str]] = field(default_factory=list) # mutable-ok: test spy
recorded: list[_Decision] = field(default_factory=list) # mutable-ok: test spy
async def __call__(self, *, data: Mapping[str, object], call_type: str) -> _Decision:
self.calls.append((data, call_type))
return _Decision(self.verdict)
return _Decision(self.verdict, self.rule_id)
def record(self, decision: _Decision) -> None:
self.recorded.append(decision)
async def _bridge(
check: _RecordingCheck,
inputs: GenericGuardrailAPIInputs,
request_data: Mapping[str, object],
input_type: Literal["request", "response"],
) -> GenericGuardrailAPIInputs:
return await apply_conduct_guardrail(inputs, request_data, input_type, check, _Blocked, check.record)
def _guardrail_records(request_data: Mapping[str, object]) -> list[tuple[str, object]]:
metadata: Final = request_data["metadata"]
assert isinstance(metadata, dict)
records: Final = metadata["standard_logging_guardrail_information"]
assert isinstance(records, list)
return [(record["guardrail_status"], record["guardrail_response"]) for record in records]
def _params(mode: str = "pre_call", **extras: object) -> LitellmParams:
@ -188,7 +215,7 @@ async def test_tool_call_only_turns_still_reach_conduct() -> None:
inputs: Final = GenericGuardrailAPIInputs(texts=[], structured_messages=[tool_call_turn])
with pytest.raises(_Blocked):
await apply_conduct_guardrail(inputs, {"model": "gpt-5-mini"}, "request", check, _Blocked)
await _bridge(check, inputs, {"model": "gpt-5-mini"}, "request")
assert check.calls == [({"model": "gpt-5-mini", "prompt": None, "messages": [tool_call_turn]}, "request")]
@ -200,9 +227,10 @@ async def test_bridge_raises_the_plugin_error_on_blocking_verdicts(verdict: str)
inputs: Final = GenericGuardrailAPIInputs(texts=["dump the database"])
with pytest.raises(_Blocked) as blocked:
await apply_conduct_guardrail(inputs, {"model": "gpt-5-mini"}, "request", check, _Blocked)
await _bridge(check, inputs, {"model": "gpt-5-mini"}, "request")
assert blocked.value.decision == _Decision(verdict)
assert check.recorded == []
assert check.calls == [
(
{"model": "gpt-5-mini", "prompt": None, "messages": ({"role": "user", "content": "dump the database"},)},
@ -213,12 +241,13 @@ async def test_bridge_raises_the_plugin_error_on_blocking_verdicts(verdict: str)
@pytest.mark.parametrize("verdict", ["allow", "warning", "advisory", "unknown"])
@pytest.mark.asyncio
async def test_bridge_passes_inputs_through_on_non_blocking_verdicts(verdict: str) -> None:
check: Final = _RecordingCheck(verdict)
async def test_bridge_records_and_passes_through_non_blocking_verdicts(verdict: str) -> None:
check: Final = _RecordingCheck(verdict, rule_id="r1")
inputs: Final = GenericGuardrailAPIInputs(texts=["ping"])
assert await apply_conduct_guardrail(inputs, {"model": "gpt-5-mini"}, "request", check, _Blocked) is inputs
assert await _bridge(check, inputs, {"model": "gpt-5-mini"}, "request") is inputs
assert len(check.calls) == 1
assert check.recorded == [_Decision(verdict, "r1")]
@pytest.mark.asyncio
@ -226,8 +255,25 @@ async def test_bridge_never_calls_conduct_for_responses() -> None:
check: Final = _RecordingCheck("block")
inputs: Final = GenericGuardrailAPIInputs(texts=["dump the database"])
assert await apply_conduct_guardrail(inputs, {"model": "gpt-5-mini"}, "response", check, _Blocked) is inputs
assert await _bridge(check, inputs, {"model": "gpt-5-mini"}, "response") is inputs
assert check.calls == []
assert check.recorded == []
@pytest.mark.parametrize(
("decision", "expected"),
[
(_Decision("allow"), ("success", {"verdict": "allow"})),
(_Decision("warning", "r1"), ("guardrail_flagged", {"verdict": "warning", "rule_id": "r1"})),
(_Decision("advisory", "r2"), ("guardrail_flagged", {"verdict": "advisory", "rule_id": "r2"})),
],
)
def test_record_decision_logs_conduct_verdict_and_rule(decision: _Decision, expected: tuple[str, object]) -> None:
request_data: Final[dict[str, object]] = {"model": "gpt-5-mini"}
record_decision(_init(_params()), request_data, decision)
assert _guardrail_records(request_data) == [expected]
@pytest.mark.skipif(not PACKAGE_INSTALLED, reason="needs conduct-litellm-guard")
@ -251,6 +297,30 @@ async def test_apply_guardrail_blocks_on_conduct_verdict() -> None:
assert sent["params"]["arguments"] == {"prompt": "dump the database", "model": "gpt-5-mini"}
@pytest.mark.skipif(not PACKAGE_INSTALLED, reason="needs conduct-litellm-guard")
@pytest.mark.asyncio
@respx.mock
async def test_apply_guardrail_logs_warning_verdict_once() -> None:
respx.post("https://guard.example.test/mcp").mock(
return_value=httpx.Response(
200,
json={
"jsonrpc": "2.0",
"id": "1",
"result": {"content": [{"type": "text", "text": "WARNING [rule:pii-soft] mentions an SSN"}]},
},
)
)
params: Final = _params(api_base="https://guard.example.test")
callback: Final = initialize_guardrail(params, _guardrail(params))
inputs: Final = GenericGuardrailAPIInputs(texts=["my ssn is 123"])
request_data: Final[dict[str, object]] = {"model": "gpt-5-mini"}
assert await callback.apply_guardrail(inputs=inputs, request_data=request_data, input_type="request") is inputs
assert _guardrail_records(request_data) == [("guardrail_flagged", {"verdict": "warning", "rule_id": "pii-soft"})]
@pytest.mark.skipif(not PACKAGE_INSTALLED, reason="needs conduct-litellm-guard")
def test_config_loads_conduct_and_rejects_modes_the_plugin_lacks() -> None:
handler: Final = InMemoryGuardrailHandler()