Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/guardrails/guardrail_hooks/qualifire/qualifire.py: 12%
186 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1# +-------------------------------------------------------------+
2#
3# Use Qualifire for your LLM calls
4#
5# +-------------------------------------------------------------+
6# Qualifire - Evaluate LLM outputs for quality, safety, and reliability
8import json
9import os
10from typing import Any, Final, Literal
12from fastapi import HTTPException
14from litellm._logging import verbose_proxy_logger
15from litellm.integrations.custom_guardrail import (
16 CustomGuardrail,
17 log_guardrail_information,
18)
19from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
20from litellm.llms.custom_httpx.http_handler import (
21 get_async_httpx_client,
22 httpxSpecialProvider,
23)
24from litellm.secret_managers.main import get_secret_str
25from litellm.types.guardrails import GuardrailEventHooks, LitellmParams
26from litellm.types.llms.openai import AllMessageValues
27from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
28from litellm.types.utils import GenericGuardrailAPIInputs
30GUARDRAIL_NAME: Final = "qualifire"
31DEFAULT_QUALIFIRE_API_BASE: Final = "https://proxy.qualifire.ai"
34class QualifireGuardrail(CustomGuardrail):
35 @classmethod
36 def get_supported_event_hooks(cls) -> list[GuardrailEventHooks]:
37 return [
38 GuardrailEventHooks.pre_call,
39 GuardrailEventHooks.during_call,
40 GuardrailEventHooks.post_call,
41 ]
43 def __init__(
44 self,
45 api_key: str | None = None,
46 api_base: str | None = None,
47 evaluation_id: str | None = None,
48 prompt_injections: bool | None = None,
49 hallucinations_check: bool | None = None,
50 grounding_check: bool | None = None,
51 pii_check: bool | None = None,
52 content_moderation_check: bool | None = None,
53 tool_selection_quality_check: bool | None = None,
54 assertions: list[str] | None = None,
55 on_flagged: str | None = "block",
56 **kwargs,
57 ):
58 """
59 Initialize the QualifireGuardrail class.
61 Args:
62 api_key: API key for Qualifire (or use QUALIFIRE_API_KEY env var)
63 api_base: Optional custom API base URL (defaults to https://api.qualifire.ai)
64 evaluation_id: Pre-configured evaluation ID from Qualifire dashboard
65 prompt_injections: Enable prompt injection detection (default if no other checks)
66 hallucinations_check: Enable hallucination detection
67 grounding_check: Enable grounding verification
68 pii_check: Enable PII detection
69 content_moderation_check: Enable content moderation
70 tool_selection_quality_check: Enable tool selection quality check
71 assertions: Custom assertions to validate against the output
72 on_flagged: Action when content is flagged: "block" or "monitor"
73 """
74 self.qualifire_api_key = api_key or get_secret_str("QUALIFIRE_API_KEY") or os.environ.get("QUALIFIRE_API_KEY")
75 self.qualifire_api_base = (
76 api_base
77 or get_secret_str("QUALIFIRE_BASE_URL")
78 or os.environ.get("QUALIFIRE_BASE_URL")
79 or DEFAULT_QUALIFIRE_API_BASE
80 )
81 self.evaluation_id = evaluation_id
82 self.prompt_injections = prompt_injections
83 self.hallucinations_check = hallucinations_check
84 self.grounding_check = grounding_check
85 self.pii_check = pii_check
86 self.content_moderation_check = content_moderation_check
87 self.tool_selection_quality_check = tool_selection_quality_check
88 self.assertions = assertions
89 self.on_flagged = on_flagged or "block"
90 self._validate_on_flagged(self.on_flagged)
92 # If no checks are specified and no evaluation_id, default to prompt_injections
93 if not self._has_any_check_enabled() and not self.evaluation_id:
94 self.prompt_injections = True
96 # Initialize async HTTP client for direct API calls
97 self.async_handler = get_async_httpx_client(llm_provider=httpxSpecialProvider.GuardrailCallback)
99 kwargs.setdefault("supported_event_hooks", list(self.get_supported_event_hooks()))
100 super().__init__(**kwargs)
102 def _validate_on_flagged(self, on_flagged: str) -> None:
103 if on_flagged not in ("block", "monitor"):
104 # on_flagged is defined on LakeraV2GuardrailConfigModel but LitellmParams
105 # flattens every guardrail config mixin together, so a value Lakera
106 # supports (e.g. "inject_system_message") type-checks for any guardrail,
107 # including this one, which never implements it. Reject it explicitly
108 # instead of silently falling through to a block-on-anything-else branch.
109 raise ValueError(
110 f"Qualifire guardrail does not support on_flagged={on_flagged!r}; "
111 "only 'block' and 'monitor' are supported."
112 )
114 def update_in_memory_litellm_params(self, litellm_params: LitellmParams) -> None:
115 """
116 The base implementation blindly ``setattr``s every field on ``litellm_params``
117 (including ``on_flagged``) onto this live instance with no revalidation, so an
118 in-place config update (via the DB/UI, without a restart) could otherwise
119 reintroduce the exact invalid on_flagged value __init__ rejects. Validate the
120 prospective post-update value *before* mutating, so a rejected update leaves
121 the live instance untouched instead of raising after it's already been
122 corrupted. Mirrors LakeraAIGuardrail's own override of this same method.
123 """
124 prospective_on_flagged: Final = litellm_params.on_flagged or self.on_flagged
125 self._validate_on_flagged(prospective_on_flagged)
126 super().update_in_memory_litellm_params(litellm_params=litellm_params)
128 def _has_any_check_enabled(self) -> bool:
129 """Check if any evaluation check is explicitly enabled."""
130 return any(
131 [
132 self.prompt_injections,
133 self.hallucinations_check,
134 self.grounding_check,
135 self.pii_check,
136 self.content_moderation_check,
137 self.tool_selection_quality_check,
138 self.assertions,
139 ]
140 )
142 def _convert_messages_to_api_format(self, messages: list[AllMessageValues]) -> list[dict[str, object]]:
143 """
144 Convert LiteLLM messages to Qualifire API format.
145 Supports tool calls for tool_selection_quality_check.
147 Returns a list of dicts matching the API's ModelInvocationCanonicalMessage schema:
148 {
149 "role": "user" | "assistant" | "system" | "tool",
150 "content": "...",
151 "tool_call_id": "...", # optional
152 "tool_calls": [{"id": "...", "name": "...", "arguments": {...}}] # optional
153 }
154 """
155 api_messages: Final = []
156 for msg in messages:
157 role = msg.get("role", "user")
158 content = msg.get("content", "")
160 # Handle content that might be a list (multimodal)
161 if isinstance(content, list):
162 text_parts = []
163 for part in content:
164 if isinstance(part, dict) and part.get("type") == "text":
165 text_parts.append(part.get("text", ""))
166 elif isinstance(part, str):
167 text_parts.append(part)
168 content = "\n".join(text_parts)
170 api_message: dict[str, object] = {
171 "role": role,
172 "content": content if isinstance(content, str) else str(content),
173 }
175 # Handle tool_call_id for tool response messages
176 tool_call_id = msg.get("tool_call_id")
177 if tool_call_id:
178 api_message["tool_call_id"] = tool_call_id
180 # Handle tool calls if present
181 tool_calls = msg.get("tool_calls")
182 if tool_calls and isinstance(tool_calls, list):
183 api_tool_calls = []
184 for tc in tool_calls:
185 if isinstance(tc, dict):
186 function_info = tc.get("function", {})
187 # Arguments can be a string (JSON) or dict
188 args = function_info.get("arguments", {})
189 if isinstance(args, str):
190 try:
191 args = json.loads(args)
192 except json.JSONDecodeError:
193 args = {}
194 api_tool_calls.append(
195 {
196 "id": tc.get("id") or "",
197 "name": function_info.get("name") or "",
198 "arguments": args if isinstance(args, dict) else {},
199 }
200 )
201 if api_tool_calls:
202 api_message["tool_calls"] = api_tool_calls
204 api_messages.append(api_message)
206 return api_messages
208 def _convert_tools_to_api_format(self, tools: list[object] | None) -> list[dict[str, object]] | None:
209 """
210 Convert OpenAI-format tools to Qualifire API format.
212 Returns a list of dicts matching the API's ModelInvocationToolDefinition schema:
213 {
214 "name": "...",
215 "description": "...",
216 "parameters": {...}
217 }
218 """
219 if not tools:
220 return None
222 api_tools: Final = []
223 for tool in tools:
224 if isinstance(tool, dict):
225 # Handle OpenAI function tool format
226 if tool.get("type") == "function":
227 function_def = tool.get("function", {})
228 api_tools.append(
229 {
230 "name": function_def.get("name", ""),
231 "description": function_def.get("description", ""),
232 "parameters": function_def.get("parameters", {}),
233 }
234 )
235 # Handle direct tool format
236 elif "name" in tool:
237 api_tools.append(
238 {
239 "name": tool.get("name", ""),
240 "description": tool.get("description", ""),
241 "parameters": tool.get("parameters", {}),
242 }
243 )
245 return api_tools if api_tools else None
247 def _check_if_flagged(self, result: dict[str, Any]) -> bool:
248 """
249 Check if the Qualifire evaluation result indicates flagged content.
251 Returns True only if there are explicitly flagged items in the evaluation results.
252 A high score (close to 100) indicates GOOD content, low score indicates problems.
253 """
254 # Check evaluation results for any flagged items
255 evaluation_results: Final = result.get("evaluationResults", []) or []
257 for eval_result in evaluation_results:
258 results = eval_result.get("results", []) or []
259 for r in results:
260 if r.get("flagged"):
261 return True
263 return False
265 def _build_evaluate_payload(
266 self,
267 api_messages: list[dict[str, object]],
268 output: str | None,
269 assertions: list[str] | None,
270 available_tools: list[dict[str, object]] | None,
271 ) -> dict[str, object]:
272 """Build payload dictionary for the /api/evaluation/evaluate endpoint."""
273 payload: Final[dict[str, object]] = {"messages": api_messages}
275 if output is not None:
276 payload["output"] = output
278 # Add enabled checks
279 if self.prompt_injections:
280 payload["prompt_injections"] = True
281 if self.hallucinations_check:
282 payload["hallucinations_check"] = True
283 if self.grounding_check:
284 payload["grounding_check"] = True
285 if self.pii_check:
286 payload["pii_check"] = True
287 if self.content_moderation_check:
288 payload["content_moderation_check"] = True
289 if self.tool_selection_quality_check:
290 # Only enable tool_selection_quality_check if available_tools is provided
291 if available_tools:
292 payload["tool_selection_quality_check"] = True
293 payload["available_tools"] = available_tools
294 else:
295 verbose_proxy_logger.debug(
296 "Qualifire Guardrail: tool_selection_quality_check enabled but no available_tools provided, skipping this check"
297 )
298 if assertions:
299 payload["assertions"] = assertions
301 return payload
303 async def _run_qualifire_check(
304 self,
305 messages: list[AllMessageValues],
306 output: str | None,
307 dynamic_params: dict[str, Any],
308 available_tools: list[object] | None = None,
309 ) -> None:
310 """
311 Core Qualifire check logic - shared between hooks.
313 Args:
314 messages: The conversation messages
315 output: The LLM output text (for post_call)
316 dynamic_params: Dynamic parameters from request body
317 available_tools: Available tools from the request (for tool_selection_quality_check)
319 Raises:
320 HTTPException: If content is blocked
321 """
322 # Apply dynamic param overrides
323 evaluation_id: Final = dynamic_params.get("evaluation_id") or self.evaluation_id
324 assertions: Final = dynamic_params.get("assertions") or self.assertions
325 on_flagged: Final = dynamic_params.get("on_flagged") or self.on_flagged
327 # Prepare headers
328 headers: Final = {
329 "X-Qualifire-API-Key": self.qualifire_api_key or "",
330 "Content-Type": "application/json",
331 }
333 try:
334 # Convert messages to API format
335 api_messages: Final = self._convert_messages_to_api_format(messages)
337 # Use invoke endpoint if evaluation_id is provided
338 if evaluation_id:
339 # For invoke_evaluation, we need to extract input/output
340 input_text = ""
342 # Get the last user message as input
343 for msg in reversed(messages):
344 if msg.get("role") == "user":
345 content = msg.get("content", "")
346 if isinstance(content, str):
347 input_text = content
348 break
350 payload = {
351 "evaluation_id": evaluation_id,
352 "input": input_text,
353 "output": output or "",
354 "messages": api_messages,
355 }
357 # Convert tools if provided
358 api_tools = self._convert_tools_to_api_format(available_tools)
359 if api_tools:
360 payload["available_tools"] = api_tools
362 url = f"{self.qualifire_api_base}/api/evaluation/invoke"
363 else:
364 # Use evaluate endpoint with individual checks
365 api_tools = self._convert_tools_to_api_format(available_tools)
366 payload = self._build_evaluate_payload(
367 api_messages=api_messages,
368 output=output,
369 assertions=assertions,
370 available_tools=api_tools,
371 )
372 url = f"{self.qualifire_api_base}/api/evaluation/evaluate"
374 verbose_proxy_logger.debug("Qualifire Guardrail: Making request to %s", url)
376 # Make the API request
377 response: Final = await self.async_handler.post(
378 url=url,
379 headers=headers,
380 json=payload,
381 )
382 response.raise_for_status()
383 result: Final = response.json()
385 # Extract response info for logging
386 qualifire_response: Final[dict[str, object]] = {
387 "score": result.get("score"),
388 "status": result.get("status"),
389 }
391 verbose_proxy_logger.debug(
392 "Qualifire Guardrail: Got result from API, score=%s, status=%s",
393 qualifire_response["score"],
394 qualifire_response["status"],
395 )
397 # Check if any evaluation flagged the content
398 is_flagged: Final = self._check_if_flagged(result)
400 if is_flagged:
401 if on_flagged == "monitor":
402 verbose_proxy_logger.warning(
403 "Qualifire Guardrail: Monitoring mode - violation detected but allowing request. Response: %s",
404 qualifire_response,
405 )
406 else:
407 # Block the request
408 raise HTTPException(
409 status_code=400,
410 detail={
411 "error": "Violated guardrail policy",
412 "qualifire_response": qualifire_response,
413 },
414 )
416 except HTTPException:
417 raise
418 except Exception as e:
419 verbose_proxy_logger.exception("Qualifire Guardrail error: %s", e)
420 raise
422 @log_guardrail_information
423 async def apply_guardrail(
424 self,
425 inputs: GenericGuardrailAPIInputs,
426 request_data: dict,
427 input_type: Literal["request", "response"],
428 logging_obj: LiteLLMLoggingObj | None = None,
429 ) -> GenericGuardrailAPIInputs:
430 """
431 Apply Qualifire guardrail to the given inputs.
433 This method is called by the unified guardrail system for both
434 input (request) and output (response) validation.
436 Args:
437 inputs: Dictionary containing:
438 - texts: List of texts to check
439 - structured_messages: Structured messages from the request (pre-call only)
440 - tool_calls: Tool calls if present
441 request_data: The original request data
442 input_type: "request" for pre-call, "response" for post-call
443 logging_obj: Optional logging object
445 Returns:
446 GenericGuardrailAPIInputs - unchanged if allowed through
448 Raises:
449 HTTPException: If content is blocked
450 """
451 # Get dynamic params from request body (allows runtime overrides)
452 dynamic_params: Final = self.get_guardrail_dynamic_request_body_params(request_data=request_data)
454 # Extract messages from structured_messages or request_data
455 messages: list[AllMessageValues] | None = inputs.get("structured_messages")
456 if not messages:
457 messages = request_data.get("messages")
459 # For response (post_call), messages may not be available in the inputs
460 # We need to work with texts instead and construct messages if needed
461 output: str | None = None
462 texts: Final = inputs.get("texts", [])
464 if input_type == "response":
465 # For post_call, extract output from texts
466 if texts:
467 output = texts[-1] if isinstance(texts, list) else str(texts)
469 # If no structured messages available, construct from texts
470 if not messages and texts:
471 # Create a simple message structure for the output
472 messages = [{"role": "assistant", "content": output or ""}]
474 if not messages:
475 # For pre_call with no messages, try to construct from texts
476 if texts:
477 messages = [{"role": "user", "content": texts[-1] if texts else ""}]
478 else:
479 verbose_proxy_logger.debug("Qualifire Guardrail: No messages or texts found, skipping")
480 return inputs
482 # Get available tools from request_data for tool_selection_quality_check
483 available_tools: Final = request_data.get("tools")
485 await self._run_qualifire_check(
486 messages=messages,
487 output=output,
488 dynamic_params=dynamic_params,
489 available_tools=available_tools,
490 )
492 return inputs
494 @staticmethod
495 def get_config_model() -> type["GuardrailConfigModel"] | None:
496 from litellm.types.proxy.guardrails.guardrail_hooks.qualifire import (
497 QualifireGuardrailConfigModel,
498 )
500 return QualifireGuardrailConfigModel