Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/guardrails/guardrail_hooks/llm_as_a_judge/__init__.py: 32%
185 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""LLM-as-a-Judge guardrail: uses an LLM to score requests or responses against weighted criteria."""
3from collections.abc import Callable, Mapping, Sequence
4from datetime import datetime
5from types import MappingProxyType
6from typing import TYPE_CHECKING, Final, Generic, Literal, Optional, TypeVar
8from fastapi import HTTPException
9from pydantic import BaseModel, ConfigDict, ValidationError
10from typing_extensions import NotRequired, ReadOnly, TypedDict, Unpack
12import litellm
13from litellm._logging import verbose_logger
14from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY
15from litellm.integrations.custom_guardrail import CustomGuardrail
16from litellm.litellm_core_utils.llm_judge import (
17 default_router_provider,
18 extract_text_from_content,
19 judge_acompletion,
20 parse_json_verdict,
21)
22from litellm.litellm_core_utils.prompt_templates.common_utils import get_last_user_message
23from litellm.types.guardrails import GuardrailEventHooks, Mode, SupportedGuardrailIntegrations
24from litellm.types.utils import LLM_AS_A_JUDGE_GUARDRAIL_CALL_ORIGIN, GenericGuardrailAPIInputs, GuardrailStatus
26if TYPE_CHECKING: 26 ↛ 27line 26 didn't jump to line 27 because the condition on line 26 was never true
27 from litellm import Router
28 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
29 from litellm.types.guardrails import Guardrail, LitellmParams
30 from litellm.types.llms.openai import AllMessageValues
31 from litellm.types.utils import StandardLoggingEvalInformation
33JudgeInputType = Literal["request", "response"]
34JudgeEventHook = GuardrailEventHooks | list[GuardrailEventHooks] | Mode
35JudgeModeParam = str | list[str] | Mode | GuardrailEventHooks | list[GuardrailEventHooks] | None
37_JUDGE_SYSTEM_PROMPT_TEMPLATE: Final = """You are a quality judge. Evaluate the {subject} against the criteria provided.
38{focus}For each criterion, assign a score from 0 to 100 and provide concise reasoning.
39Return ONLY valid JSON in this exact format:
40{{
41 "verdicts": [
42 {{"criterion_name": "<name>", "score": <0-100>, "reasoning": "<one sentence>", "passed": <true|false>, "weight": <weight>}}
43 ],
44 "overall_score": <weighted average 0-100>
45}}"""
47JUDGE_SYSTEM_PROMPTS: Final[MappingProxyType[JudgeInputType, str]] = MappingProxyType(
48 {
49 "request": _JUDGE_SYSTEM_PROMPT_TEMPLATE.format(
50 subject="request",
51 focus="Judge the most recent user turn; treat earlier turns in the conversation only as context.\n",
52 ),
53 "response": _JUDGE_SYSTEM_PROMPT_TEMPLATE.format(subject="assistant's response", focus=""),
54 }
55)
57_JUDGE_SUBJECT_LABELS: Final[MappingProxyType[JudgeInputType, str]] = MappingProxyType(
58 {"request": "Latest request turn to evaluate", "response": "Assistant response to evaluate"}
59)
61_LIFECYCLE_HOOKS: Final[MappingProxyType[JudgeInputType, tuple[GuardrailEventHooks, ...]]] = MappingProxyType(
62 {
63 "request": (GuardrailEventHooks.pre_call, GuardrailEventHooks.during_call, GuardrailEventHooks.logging_only),
64 "response": (GuardrailEventHooks.post_call, GuardrailEventHooks.logging_only),
65 }
66)
68_VALID_ON_FAILURE: Final = frozenset({"block", "log"})
70_JUDGE_CALL_METADATA: Final = MappingProxyType(
71 {INTERNAL_CALL_ORIGIN_METADATA_KEY: LLM_AS_A_JUDGE_GUARDRAIL_CALL_ORIGIN}
72)
75class _LoggedCallParams(BaseModel):
76 model_config = ConfigDict(frozen=True)
78 metadata: Mapping[str, object] | None = None
81def _is_logged_judge_call(data: Mapping[str, object], event_type: GuardrailEventHooks) -> bool:
82 """logging_only is the only event whose ``data`` is the SDK's model_call_details rather than the client body."""
83 if event_type is not GuardrailEventHooks.logging_only:
84 return False
85 try:
86 params: Final = _LoggedCallParams.model_validate(data.get("litellm_params") or {})
87 except ValidationError:
88 return False
89 return (params.metadata or {}).get(INTERNAL_CALL_ORIGIN_METADATA_KEY) == LLM_AS_A_JUDGE_GUARDRAIL_CALL_ORIGIN
92_default_router_provider: Final = default_router_provider
93_parse_judge_verdict: Final = parse_json_verdict
94_extract_text_from_content: Final = extract_text_from_content
96_ParamT = TypeVar("_ParamT")
99class _LitellmParamView(TypedDict, Generic[_ParamT]):
100 """Typed read of a single entry in an untyped ``litellm_params`` mapping."""
102 value: ReadOnly[_ParamT]
105class JudgeCriterion(TypedDict):
106 """A single weighted criterion the judge scores the response against."""
108 name: ReadOnly[NotRequired[str]]
109 description: ReadOnly[NotRequired[str]]
110 weight: ReadOnly[NotRequired[float]]
113class JudgeMessage(TypedDict):
114 """The parts of a conversation message the judge prompt renders."""
116 role: ReadOnly[NotRequired[str]]
117 content: ReadOnly[NotRequired[object]]
120def _get_litellm_param(
121 litellm_params: "LitellmParams",
122 guardrail: "Guardrail",
123 key: str,
124 default: _ParamT,
125) -> _ParamT:
126 val: Final[_ParamT | None] = getattr(litellm_params, key, None)
127 if val is not None:
128 return val
129 raw: Final = guardrail.get("litellm_params")
130 if isinstance(raw, dict) and key in raw:
131 entry: Final[_LitellmParamView[_ParamT]] = {"value": raw[key]}
132 return entry["value"]
133 if raw is not None and not isinstance(raw, dict):
134 attr: Final[_ParamT | None] = getattr(raw, key, None)
135 if attr is not None:
136 return attr
137 return default
140def _coerce_event_hook(mode: JudgeModeParam) -> JudgeEventHook:
141 if mode is None:
142 return GuardrailEventHooks.post_call
143 if isinstance(mode, Mode):
144 return mode
145 if isinstance(mode, list):
146 return [GuardrailEventHooks(hook) for hook in mode]
147 return GuardrailEventHooks(mode)
150def _text_under_review(inputs: GenericGuardrailAPIInputs, input_type: JudgeInputType) -> str:
151 all_text: Final = "\n".join(inputs.get("texts") or [])
152 if input_type == "response":
153 return all_text
154 latest_user_turn: Final = get_last_user_message(inputs.get("structured_messages") or [])
155 return latest_user_turn if latest_user_turn is not None else all_text
158def _build_judge_prompt(
159 criteria: Sequence[JudgeCriterion],
160 messages: Sequence[JudgeMessage],
161 text_under_review: str,
162 input_type: JudgeInputType = "response",
163) -> str:
164 criteria_block: Final = "\n".join(
165 f"- {c.get('name', '')} (weight {c.get('weight', 0)}%): {c.get('description', '')}" for c in criteria
166 )
167 conversation: Final = "\n".join(
168 f"{m.get('role', 'user').upper()}: {_extract_text_from_content(m.get('content', ''))}"
169 for m in messages
170 if m.get("content") is not None
171 )
172 conversation_block: Final = f"Conversation:\n{conversation}\n\n" if conversation or input_type == "response" else ""
173 return (
174 f"Criteria to evaluate:\n{criteria_block}\n\n"
175 f"{conversation_block}"
176 f"{_JUDGE_SUBJECT_LABELS[input_type]}:\n{text_under_review}"
177 )
180class _CustomGuardrailOptions(TypedDict, total=False):
181 """The ``CustomGuardrail`` options this guardrail accepts and forwards untouched."""
183 mask_request_content: ReadOnly[bool]
184 mask_response_content: ReadOnly[bool]
185 violation_message_template: ReadOnly[str | None]
186 end_session_after_n_fails: ReadOnly[int | None]
187 on_violation: ReadOnly[str | None]
188 realtime_violation_message: ReadOnly[str | None]
189 on_sensitive_data: ReadOnly[str | None]
190 sensitive_data_route_to_model: ReadOnly[str | None]
191 sticky_session_routing: ReadOnly[bool]
192 run_in_parallel: ReadOnly[bool]
193 scan_raw_request: ReadOnly[bool]
194 only_scan_new_messages: ReadOnly[bool]
197class LLMAsAJudgeGuardrail(CustomGuardrail):
198 """Guardrail that judges request (pre_call/during_call) or response (post_call) quality via an LLM."""
200 def __init__(
201 self,
202 guardrail_name: str,
203 judge_model: str,
204 criteria: Sequence[JudgeCriterion],
205 overall_threshold: float = 80.0,
206 on_failure: Literal["block", "log"] = "block",
207 event_hook: JudgeModeParam = None,
208 default_on: bool = False,
209 router_provider: "Callable[[], Router | None] | None" = None,
210 **kwargs: Unpack[_CustomGuardrailOptions],
211 ) -> None:
212 super().__init__(
213 guardrail_name=guardrail_name,
214 supported_event_hooks=list(self.get_supported_event_hooks()),
215 event_hook=_coerce_event_hook(event_hook),
216 default_on=default_on,
217 **kwargs,
218 )
219 self.judge_model = judge_model
220 self.criteria = criteria
221 self.overall_threshold = overall_threshold
222 self.on_failure = on_failure
223 self._router_provider = router_provider or _default_router_provider
225 @classmethod
226 def get_supported_event_hooks(cls) -> list[GuardrailEventHooks]:
227 return [GuardrailEventHooks.pre_call, GuardrailEventHooks.during_call, GuardrailEventHooks.post_call]
229 def should_run_guardrail(self, data: Mapping[str, object], event_type: GuardrailEventHooks) -> bool:
230 if _is_logged_judge_call(data, event_type):
231 return False
232 return super().should_run_guardrail(data, event_type)
234 async def _run_judge(
235 self,
236 messages: Sequence[JudgeMessage],
237 text_under_review: str,
238 input_type: JudgeInputType = "response",
239 ) -> dict[str, object]:
240 judge_messages: Final[list[AllMessageValues]] = [
241 {"role": "system", "content": JUDGE_SYSTEM_PROMPTS[input_type]},
242 {
243 "role": "user",
244 "content": _build_judge_prompt(self.criteria, messages, text_under_review, input_type),
245 },
246 ]
247 response: Final = await judge_acompletion(
248 self._router_provider(),
249 self.judge_model,
250 judge_messages,
251 response_format={"type": "json_object"},
252 temperature=0,
253 metadata=dict(_JUDGE_CALL_METADATA),
254 )
255 raw: Final = response.choices[0].message.content or "{}"
256 return _parse_judge_verdict(raw)
258 async def apply_guardrail(
259 self,
260 inputs: GenericGuardrailAPIInputs,
261 request_data: dict,
262 input_type: Literal["request", "response"],
263 logging_obj: Optional["LiteLLMLoggingObj"] = None,
264 ) -> GenericGuardrailAPIInputs:
265 text_under_review: Final = _text_under_review(inputs, input_type)
266 if not text_under_review:
267 return inputs
269 start_time: Final = datetime.now()
270 status: GuardrailStatus = "success"
271 judge_result: dict[str, object] = {}
273 try:
274 messages: Final[Sequence[JudgeMessage]] = (
275 inputs.get("structured_messages") or request_data.get("messages") or []
276 )
278 try:
279 judge_result = await self._run_judge(messages, text_under_review, input_type)
280 except Exception as judge_err:
281 verbose_logger.warning(
282 "llm_as_a_judge guardrail: judge call failed, failing open. Error: %s", judge_err
283 )
284 status = "guardrail_failed_to_respond"
285 return inputs
287 try:
288 overall_score: Final = max(0.0, min(100.0, float(judge_result.get("overall_score", 100))))
289 except (TypeError, ValueError):
290 verbose_logger.warning("llm_as_a_judge: invalid overall_score from judge, failing open")
291 return inputs
293 passed: Final = overall_score >= self.overall_threshold
295 eval_info: Final[StandardLoggingEvalInformation] = {
296 "eval_name": self.guardrail_name or "",
297 "overall_score": overall_score,
298 "passed": passed,
299 "judge_model": self.judge_model,
300 "threshold": self.overall_threshold,
301 "verdicts": judge_result.get("verdicts", []),
302 }
303 _metadata: Final = request_data.setdefault("metadata", {})
304 existing: Final = _metadata.get("eval_information")
305 if isinstance(existing, list):
306 existing.append(eval_info)
307 elif existing is not None:
308 _metadata["eval_information"] = [existing, eval_info]
309 else:
310 _metadata["eval_information"] = eval_info
312 if not passed:
313 status = "guardrail_intervened"
314 if self.on_failure == "block":
315 raise HTTPException(
316 status_code=422,
317 detail={
318 "error": f"LLM judge rejected {input_type}: score below threshold",
319 "overall_score": overall_score,
320 "threshold": self.overall_threshold,
321 "verdicts": judge_result.get("verdicts", []),
322 },
323 )
325 return inputs
327 except HTTPException:
328 raise
329 except Exception as e:
330 verbose_logger.warning("llm_as_a_judge guardrail unexpected error: %s", e)
331 return inputs
332 finally:
333 self.add_standard_logging_guardrail_information_to_request_data(
334 guardrail_provider="llm_as_a_judge",
335 guardrail_json_response=judge_result,
336 request_data=request_data,
337 guardrail_status=status,
338 start_time=start_time.timestamp(),
339 end_time=datetime.now().timestamp(),
340 event_type=self._event_type_for(input_type),
341 )
343 def _event_type_for(self, input_type: JudgeInputType) -> GuardrailEventHooks | None:
344 configured: Final = tuple(hook for hook in _LIFECYCLE_HOOKS[input_type] if self._event_hook_is_event_type(hook))
345 return configured[0] if len(configured) == 1 else None
348def initialize_guardrail(
349 litellm_params: "LitellmParams",
350 guardrail: "Guardrail",
351) -> LLMAsAJudgeGuardrail:
352 guardrail_name: Final = guardrail.get("guardrail_name")
353 if not guardrail_name:
354 raise ValueError("llm_as_a_judge guardrail requires a guardrail_name")
356 judge_model: Final[str] = _get_litellm_param(litellm_params, guardrail, "judge_model", "")
357 if not judge_model:
358 raise ValueError("llm_as_a_judge guardrail requires judge_model in litellm_params")
360 criteria: Final[Sequence[JudgeCriterion]] = _get_litellm_param(litellm_params, guardrail, "criteria", ()) or ()
361 if not criteria:
362 raise ValueError("llm_as_a_judge guardrail requires at least one criterion")
364 weight_total: Final = sum(float(c.get("weight", 0)) for c in criteria)
365 if abs(weight_total - 100) > 0.5:
366 raise ValueError(f"llm_as_a_judge criterion weights must sum to 100 (got {weight_total})")
368 on_failure: Final[Literal["block", "log"]] = _get_litellm_param(litellm_params, guardrail, "on_failure", "block")
369 if on_failure not in _VALID_ON_FAILURE:
370 raise ValueError(f"llm_as_a_judge on_failure must be 'block' or 'log', got '{on_failure}'")
372 overall_threshold: Final = float(_get_litellm_param(litellm_params, guardrail, "overall_threshold", 80.0))
374 mode: Final[JudgeModeParam] = _get_litellm_param(litellm_params, guardrail, "mode", None)
376 instance: Final = LLMAsAJudgeGuardrail(
377 guardrail_name=guardrail_name,
378 judge_model=judge_model,
379 criteria=criteria,
380 overall_threshold=overall_threshold,
381 on_failure=on_failure,
382 event_hook=mode,
383 default_on=bool(_get_litellm_param(litellm_params, guardrail, "default_on", False)),
384 )
385 litellm.logging_callback_manager.add_litellm_callback(instance)
386 return instance
389guardrail_class_registry: Final = {
390 SupportedGuardrailIntegrations.LLM_AS_A_JUDGE.value: LLMAsAJudgeGuardrail,
391}
394__all__ = [
395 "LLMAsAJudgeGuardrail",
396 "guardrail_class_registry",
397 "initialize_guardrail",
398]