Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/guardrails/guardrail_hooks/openai/moderations.py: 22%
104 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1#!/usr/bin/env python3
2"""
3OpenAI Moderation Guardrail Integration for LiteLLM
4"""
6from typing import TYPE_CHECKING, Final, Literal, Optional
8from fastapi import HTTPException
10from litellm._logging import verbose_proxy_logger
11from litellm.integrations.custom_guardrail import (
12 CustomGuardrail,
13 log_guardrail_information,
14)
15from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
16from litellm.llms.custom_httpx.http_handler import (
17 get_async_httpx_client,
18 httpxSpecialProvider,
19)
20from litellm.proxy.common_utils.callback_utils import add_guardrail_scan_id
21from litellm.types.guardrails import GuardrailEventHooks, SupportedGuardrailIntegrations
22from litellm.types.utils import (
23 GenericGuardrailAPIInputs,
24 GuardrailStatus,
25 GuardrailTracingDetail,
26)
28from .base import OpenAIGuardrailBase
30if TYPE_CHECKING: 30 ↛ 31line 30 didn't jump to line 31 because the condition on line 30 was never true
31 from litellm.types.llms.openai import OpenAIModerationResponse
32 from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
35class OpenAIModerationGuardrail(OpenAIGuardrailBase, CustomGuardrail):
36 """
37 LiteLLM Built-in Guardrail for OpenAI Content Moderation.
39 This guardrail scans prompts and responses using the OpenAI Moderation API to detect
40 harmful content, including violence, hate, harassment, self-harm, sexual content, etc.
42 Configuration:
43 guardrail_name: Name of the guardrail instance
44 api_key: OpenAI API key
45 api_base: OpenAI API endpoint
46 model: OpenAI moderation model to use
47 default_on: Whether to enable by default
48 """
50 def __init__(
51 self,
52 guardrail_name: str,
53 api_key: str | None = None,
54 api_base: str | None = None,
55 model: Literal["omni-moderation-latest", "text-moderation-latest"] | None = None,
56 streaming_end_of_stream_only: bool | None = None,
57 streaming_sampling_rate: int | None = None,
58 **kwargs,
59 ):
60 """Initialize OpenAI Moderation guardrail handler."""
61 super().__init__(
62 guardrail_name=guardrail_name,
63 supported_event_hooks=list(self.get_supported_event_hooks()),
64 **kwargs,
65 )
67 self.async_handler = get_async_httpx_client(llm_provider=httpxSpecialProvider.GuardrailCallback)
69 # Store configuration
70 self.api_key = api_key or self._get_api_key()
71 self.api_base = api_base or "https://api.openai.com/v1"
72 self.model: Literal["omni-moderation-latest", "text-moderation-latest"] = model or "omni-moderation-latest"
74 # Read by UnifiedLLMGuardrails.async_post_call_streaming_iterator_hook
75 # via getattr(guardrail_to_apply, "streaming_*", default).
76 self.streaming_end_of_stream_only: bool = (
77 False if streaming_end_of_stream_only is None else streaming_end_of_stream_only
78 )
79 self.streaming_sampling_rate: int = 5 if streaming_sampling_rate is None else streaming_sampling_rate
81 if not self.api_key:
82 raise ValueError(
83 "OpenAI Moderation: api_key is required. Set OPENAI_API_KEY environment variable or pass it in configuration."
84 )
86 verbose_proxy_logger.debug(
87 "Initialized OpenAI Moderation Guardrail: %s with model: %s", guardrail_name, self.model
88 )
90 def _get_api_key(self) -> str | None:
91 """Get API key from environment variables or litellm configuration"""
92 import os
94 import litellm
95 from litellm.secret_managers.main import get_secret_str
97 return (
98 os.environ.get("OPENAI_API_KEY")
99 or litellm.api_key
100 or litellm.openai_key
101 or get_secret_str("OPENAI_API_KEY")
102 )
104 async def async_make_request(self, input_text: str) -> "OpenAIModerationResponse":
105 """
106 Make a request to the OpenAI Moderation API.
107 """
108 request_body: Final = {"model": self.model, "input": input_text}
110 verbose_proxy_logger.debug("OpenAI Moderation guard request: %s", request_body)
112 response: Final = await self.async_handler.post(
113 url=f"{self.api_base}/moderations",
114 headers={
115 "Authorization": f"Bearer {self.api_key}",
116 "Content-Type": "application/json",
117 },
118 json=request_body,
119 )
121 verbose_proxy_logger.debug("OpenAI Moderation guard response: %s", response.json())
123 if response.status_code != 200:
124 raise HTTPException(
125 status_code=response.status_code,
126 detail={
127 "error": "OpenAI Moderation API request failed",
128 "details": response.text,
129 },
130 )
132 from litellm.types.llms.openai import OpenAIModerationResponse
134 return OpenAIModerationResponse(**response.json())
136 def _check_moderation_result(self, moderation_response: "OpenAIModerationResponse") -> None:
137 """
138 Check if the moderation response indicates harmful content and raise exception if needed.
139 """
140 if not moderation_response.results:
141 return
143 result: Final = moderation_response.results[0]
144 if result.flagged:
145 # Build detailed violation information
146 violated_categories: Final = []
147 if result.categories:
148 for category, is_violated in result.categories.items():
149 if is_violated:
150 violated_categories.append(category)
152 violation_details: Final = {
153 "violated_categories": violated_categories,
154 "category_scores": result.category_scores or {},
155 }
157 verbose_proxy_logger.warning(
158 "OpenAI Moderation: Content flagged for violations: %s",
159 violation_details,
160 )
162 raise HTTPException(
163 status_code=400,
164 detail={
165 "error": "Violated OpenAI moderation policy",
166 "moderation_result": violation_details,
167 },
168 )
170 @log_guardrail_information
171 async def apply_guardrail(
172 self,
173 inputs: GenericGuardrailAPIInputs,
174 request_data: dict,
175 input_type: Literal["request", "response"],
176 logging_obj: Optional["LiteLLMLoggingObj"] = None,
177 ) -> GenericGuardrailAPIInputs:
178 """
179 Apply OpenAI moderation guardrail using the unified guardrail interface.
181 This method is called by the UnifiedLLMGuardrails system for all endpoint types
182 (chat completions, embeddings, responses API, etc.).
184 Args:
185 inputs: GenericGuardrailAPIInputs containing texts and/or structured_messages
186 request_data: The original request data
187 input_type: Whether this is a "request" (pre-call) or "response" (post-call)
188 logging_obj: Optional logging object
190 Returns:
191 The inputs unchanged (moderation doesn't modify content, only blocks)
193 Raises:
194 HTTPException: If content violates moderation policy
195 """
196 # Extract text to moderate from inputs
197 text_to_moderate: str | None = None
199 # Prefer structured_messages if available (has role context)
200 if structured_messages := inputs.get("structured_messages"):
201 text_to_moderate = self.get_user_prompt(structured_messages)
203 # Fall back to texts
204 if not text_to_moderate:
205 if texts := inputs.get("texts"):
206 # Join all texts for moderation
207 text_to_moderate = "\n".join(texts)
209 if not text_to_moderate:
210 verbose_proxy_logger.debug("OpenAI Moderation: No text content to moderate in inputs")
211 return inputs
213 # Make moderation request
214 moderation_response: Final = await self.async_make_request(input_text=text_to_moderate)
216 # Stash full moderation response in request_data for logging
217 # (Model Armor pattern — per-request dict avoids race conditions)
218 if isinstance(request_data, dict):
219 metadata: Final = request_data.get("metadata") or {}
220 request_data["metadata"] = metadata
221 metadata["_openai_moderation_response"] = moderation_response.model_dump()
222 add_guardrail_scan_id(
223 request_data=request_data,
224 scan_id=moderation_response.id,
225 guardrail_name=self.guardrail_name,
226 provider=SupportedGuardrailIntegrations.OPENAI_MODERATION.value,
227 stage=GuardrailEventHooks.post_call if input_type == "response" else GuardrailEventHooks.pre_call,
228 )
230 # Check if content is flagged and raise exception if needed
231 self._check_moderation_result(moderation_response)
233 # Moderation doesn't modify content, just blocks - return inputs unchanged
234 return inputs
236 def _process_response(
237 self,
238 response: dict | None,
239 request_data: dict,
240 start_time: float | None = None,
241 end_time: float | None = None,
242 duration: float | None = None,
243 event_type: GuardrailEventHooks | None = None,
244 original_inputs: dict | None = None,
245 ):
246 """
247 Override to log the full OpenAI Moderation API response instead of
248 the decorator's simplified "allow"/"mask" string.
250 Follows the Model Armor pattern (model_armor.py:325-360).
251 """
252 if isinstance(request_data, dict):
253 metadata = request_data.get("metadata") or {}
254 request_data["metadata"] = metadata # anchor so pop() mutates the real dict
255 else:
256 metadata = {}
258 # .pop() cleans up the internal key so it doesn't leak to downstream
259 # loggers. Falls back to "allow" when no moderation call was made
260 # (e.g. no text to moderate — early return in apply_guardrail).
261 guardrail_response: Final = metadata.pop("_openai_moderation_response", "allow")
263 self.add_standard_logging_guardrail_information_to_request_data(
264 guardrail_json_response=guardrail_response,
265 request_data=request_data,
266 guardrail_status="success",
267 duration=duration,
268 start_time=start_time,
269 end_time=end_time,
270 event_type=event_type,
271 tracing_detail=self._build_tracing_detail(guardrail_response),
272 )
273 return response
275 def _process_error(
276 self,
277 e: Exception,
278 request_data: dict,
279 start_time: float | None = None,
280 end_time: float | None = None,
281 duration: float | None = None,
282 event_type: GuardrailEventHooks | None = None,
283 ):
284 """
285 Override to log the full OpenAI Moderation API response on error
286 instead of the stringified exception.
287 """
288 guardrail_status: Final[GuardrailStatus] = (
289 "guardrail_intervened" if self._is_guardrail_intervention(e) else "guardrail_failed_to_respond"
290 )
292 if isinstance(request_data, dict):
293 metadata = request_data.get("metadata") or {}
294 request_data["metadata"] = metadata # anchor so pop() mutates the real dict
295 else:
296 metadata = {}
298 # Use the stashed moderation response if available, fall back to exception
299 guardrail_response: Final[dict | Exception | str] = metadata.pop("_openai_moderation_response", e)
301 self.add_standard_logging_guardrail_information_to_request_data(
302 guardrail_json_response=guardrail_response,
303 request_data=request_data,
304 guardrail_status=guardrail_status,
305 duration=duration,
306 start_time=start_time,
307 end_time=end_time,
308 event_type=event_type,
309 tracing_detail=self._build_tracing_detail(guardrail_response),
310 )
311 raise e
313 @staticmethod
314 def _build_tracing_detail(
315 guardrail_response: dict | str | Exception,
316 ) -> GuardrailTracingDetail | None:
317 """
318 Pull the flagged category names out of the moderation response so trace
319 backends can index a short, queryable ``guardrail_violation_categories``
320 attribute instead of the full ``guardrail_response`` blob, whose
321 ``category_scores`` map (one float per category) blows past indexed-field
322 length limits on backends like ELK (1024 chars).
323 """
324 if not isinstance(guardrail_response, dict):
325 return None
327 results: Final = guardrail_response.get("results") or []
328 violation_categories: Final = [
329 category
330 for result in results
331 if isinstance(result, dict)
332 for category, is_flagged in (result.get("categories") or {}).items()
333 if is_flagged
334 ]
335 if not violation_categories:
336 return None
337 return GuardrailTracingDetail(violation_categories=violation_categories)
339 @staticmethod
340 def get_config_model() -> type["GuardrailConfigModel"] | None:
341 """
342 Get the config model for the OpenAI Moderation guardrail.
343 """
344 from litellm.types.proxy.guardrails.guardrail_hooks.openai.openai_moderation import (
345 OpenAIModerationGuardrailConfigModel,
346 )
348 return OpenAIModerationGuardrailConfigModel
350 @classmethod
351 def get_supported_event_hooks(cls) -> list[GuardrailEventHooks]:
352 return [
353 GuardrailEventHooks.pre_call,
354 GuardrailEventHooks.during_call,
355 GuardrailEventHooks.post_call,
356 ]