ai-workspace-services/litellm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[Feat] - New guardrail - OpenAI Moderations API (#12519)

Browse Source 
* add OpenAIModerationGuardrailConfigModel

* add OpenAIGuardrailBase

* _extract_response_text

* fixes for openai moderation guard

* docs OpenAI Moderation

* add OpenAIModerationGuardrailConfigModel

* use literal for model

* fix init openai moderation

* test_openai_moderation_guardrail_adds_to_litellm_callbacks

* show default on on UI

* fix code qa
This commit is contained in:
Ishaan Jaff 2025-07-11 11:25:45 -07:00 committed by GitHub
parent 7b44576216
commit cb96f1d740
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 1276 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

312
docs/my-website/docs/proxy/guardrails/openai_moderation.md Normal file
View File

@ -0,0 +1,312 @@
import Image from '@theme/IdealImage';
import Tabs from '@theme/Tabs';
import TabItem from '@theme/TabItem';
# OpenAI Moderation
## Overview
| Property | Details |
|-------|-------|
| Description | Use OpenAI's built-in Moderation API to detect and block harmful content including hate speech, harassment, self-harm, sexual content, and violence. |
| Provider | [OpenAI Moderation API](https://platform.openai.com/docs/guides/moderation) |
| Supported Actions | `BLOCK` (raises HTTP 400 exception when violations detected) |
| Supported Modes | `pre_call`, `during_call`, `post_call` |
| Streaming Support | ✅ Full support for streaming responses |
| API Requirements | OpenAI API key |
## Quick Start
### 1. Define Guardrails on your LiteLLM config.yaml
Define your guardrails under the `guardrails` section:
<Tabs>
<TabItem value="config" label="Config.yaml">
```yaml showLineNumbers title="config.yaml"
model_list:
- model_name: gpt-4
litellm_params:
model: openai/gpt-4
api_key: os.environ/OPENAI_API_KEY
guardrails:
- guardrail_name: "openai-moderation-pre"
litellm_params:
guardrail: openai_moderation
mode: "pre_call"
api_key: os.environ/OPENAI_API_KEY # Optional if already set globally
model: "omni-moderation-latest" # Optional, defaults to omni-moderation-latest
api_base: "https://api.openai.com/v1" # Optional, defaults to OpenAI API
```
#### Supported values for `mode`
- `pre_call` Run **before** LLM call, on **user input**
- `during_call` Run **during** LLM call, on **user input**. Same as `pre_call` but runs in parallel as LLM call. Response not returned until guardrail check completes.
- `post_call` Run **after** LLM call, on **LLM response**
#### Supported OpenAI Moderation Models
- `omni-moderation-latest` (default) - Latest multimodal moderation model
- `text-moderation-latest` - Latest text-only moderation model
</TabItem>
<TabItem value="env" label="Environment Variables">
Set your OpenAI API key:
```bash title="Setup Environment Variables"
export OPENAI_API_KEY="your-openai-api-key"
```
</TabItem>
</Tabs>
### 2. Start LiteLLM Gateway
```shell
litellm --config config.yaml --detailed_debug
```
### 3. Test request
<Tabs>
<TabItem label="Blocked Request" value="blocked">
Expect this to fail since the request contains harmful content:
```shell
curl -i http://0.0.0.0:4000/v1/chat/completions \
-H "Content-Type: application/json" \
-H "Authorization: Bearer sk-1234" \
-d '{
"model": "gpt-4",
"messages": [
{"role": "user", "content": "I hate all people and want to hurt them"}
],
"guardrails": ["openai-moderation-pre"]
}'
```
Expected response on failure:
```json
{
"error": {
"message": {
"error": "Violated OpenAI moderation policy",
"moderation_result": {
"violated_categories": ["hate", "violence"],
"category_scores": {
"hate": 0.95,
"violence": 0.87,
"harassment": 0.12,
"self-harm": 0.01,
"sexual": 0.02
}
}
},
"type": "None",
"param": "None",
"code": "400"
}
}
```
</TabItem>
<TabItem label="Successful Call" value="allowed">
```shell
curl -i http://0.0.0.0:4000/v1/chat/completions \
-H "Content-Type: application/json" \
-H "Authorization: Bearer sk-1234" \
-d '{
"model": "gpt-4",
"messages": [
{"role": "user", "content": "What is the capital of France?"}
],
"guardrails": ["openai-moderation-pre"]
}'
```
Expected response:
```json
{
"id": "chatcmpl-4a1c1a4a-3e1d-4fa4-ae25-7ebe84c9a9a2",
"created": 1741082354,
"model": "gpt-4",
"object": "chat.completion",
"choices": [
{
"finish_reason": "stop",
"index": 0,
"message": {
"content": "The capital of France is Paris.",
"role": "assistant"
}
}
],
"usage": {
"completion_tokens": 8,
"prompt_tokens": 13,
"total_tokens": 21
}
}
```
</TabItem>
</Tabs>
## Advanced Configuration
### Multiple Guardrails for Input and Output
You can configure separate guardrails for user input and LLM responses:
```yaml showLineNumbers title="Multiple Guardrails Config"
guardrails:
- guardrail_name: "openai-moderation-input"
litellm_params:
guardrail: openai_moderation
mode: "pre_call"
api_key: os.environ/OPENAI_API_KEY
- guardrail_name: "openai-moderation-output"
litellm_params:
guardrail: openai_moderation
mode: "post_call"
api_key: os.environ/OPENAI_API_KEY
```
### Custom API Configuration
Configure custom OpenAI API endpoints or different models:
```yaml showLineNumbers title="Custom API Config"
guardrails:
- guardrail_name: "openai-moderation-custom"
litellm_params:
guardrail: openai_moderation
mode: "pre_call"
api_key: os.environ/OPENAI_API_KEY
api_base: "https://your-custom-openai-endpoint.com/v1"
model: "text-moderation-latest"
```
## Streaming Support
The OpenAI Moderation guardrail fully supports streaming responses. When used in `post_call` mode, it will:
1. Collect all streaming chunks
2. Assemble the complete response
3. Apply moderation to the full content
4. Block the entire stream if violations are detected
5. Return the original stream if content is safe
```yaml showLineNumbers title="Streaming Config"
guardrails:
- guardrail_name: "openai-moderation-streaming"
litellm_params:
guardrail: openai_moderation
mode: "post_call" # Works with streaming responses
api_key: os.environ/OPENAI_API_KEY
```
## Content Categories
The OpenAI Moderation API detects the following categories of harmful content:
| Category | Description |
|----------|-------------|
| `hate` | Content that expresses, incites, or promotes hate based on race, gender, ethnicity, religion, nationality, sexual orientation, disability status, or caste |
| `harassment` | Content that harasses, bullies, or intimidates an individual |
| `self-harm` | Content that promotes, encourages, or depicts acts of self-harm |
| `sexual` | Content meant to arouse sexual excitement or promote sexual services |
| `violence` | Content that depicts death, violence, or physical injury |
Each category is evaluated with both a boolean flag and a confidence score (0.0 to 1.0).
## Error Handling
When content violates OpenAI's moderation policy:
- **HTTP Status**: 400 Bad Request
- **Error Type**: `HTTPException`
- **Error Details**: Includes violated categories and confidence scores
- **Behavior**: Request is immediately blocked
## Best Practices
### 1. Use Pre-call for User Input
```yaml
guardrails:
- guardrail_name: "input-moderation"
litellm_params:
guardrail: openai_moderation
mode: "pre_call" # Block harmful user inputs early
```
### 2. Use Post-call for LLM Responses
```yaml
guardrails:
- guardrail_name: "output-moderation"
litellm_params:
guardrail: openai_moderation
mode: "post_call" # Ensure LLM responses are safe
```
### 3. Combine with Other Guardrails
```yaml
guardrails:
- guardrail_name: "openai-moderation"
litellm_params:
guardrail: openai_moderation
mode: "pre_call"
- guardrail_name: "custom-pii-detection"
litellm_params:
guardrail: presidio
mode: "pre_call"
```
## Troubleshooting
### Common Issues
1. **Invalid API Key**: Ensure your OpenAI API key is correctly set
```bash
export OPENAI_API_KEY="sk-your-actual-key"
```
2. **Rate Limiting**: OpenAI Moderation API has rate limits. Monitor usage in high-volume scenarios.
3. **Network Issues**: Verify connectivity to OpenAI's API endpoints.
### Debug Mode
Enable detailed logging to troubleshoot issues:
```shell
litellm --config config.yaml --detailed_debug
```
Look for logs starting with `OpenAI Moderation:` to trace guardrail execution.
## API Costs
The OpenAI Moderation API is **free to use** for content policy compliance. This makes it a cost-effective guardrail option compared to other commercial moderation services.
## Need Help?
For additional support:
- Check the [OpenAI Moderation API documentation](https://platform.openai.com/docs/guides/moderation)
- Review [LiteLLM Guardrails documentation](./quick_start)
- Join our [Discord community](https://discord.gg/wuPM9dRgDw)

1
docs/my-website/sidebars.js
View File

@ -39,6 +39,7 @@ const sidebars = {
"proxy/guardrails/lasso_security",
"proxy/guardrails/guardrails_ai",
"proxy/guardrails/lakera_ai",
"proxy/guardrails/openai_moderation",
"proxy/guardrails/pangea",
"proxy/guardrails/pii_masking_v2",
"proxy/guardrails/panw_prisma_airs",

45
litellm/proxy/guardrails/guardrail_hooks/openai/__init__.py Normal file
View File

@ -0,0 +1,45 @@
from typing import TYPE_CHECKING
import litellm
from litellm.proxy.guardrails.guardrail_hooks.openai.moderations import (
OpenAIModerationGuardrail,
)
from litellm.types.guardrails import SupportedGuardrailIntegrations
if TYPE_CHECKING:
from litellm.types.guardrails import Guardrail, LitellmParams
def initialize_guardrail(litellm_params: "LitellmParams", guardrail: "Guardrail"):
guardrail_name = guardrail.get("guardrail_name")
if not guardrail_name:
raise ValueError("OpenAI Moderation: guardrail_name is required")
openai_moderation_guardrail = OpenAIModerationGuardrail(
guardrail_name=guardrail_name,
**{
**litellm_params.model_dump(exclude_none=True),
"api_key": litellm_params.api_key,
"api_base": litellm_params.api_base,
"default_on": litellm_params.default_on,
"event_hook": litellm_params.mode,
"model": litellm_params.model,
},
)
litellm.logging_callback_manager.add_litellm_callback(
openai_moderation_guardrail
)
return openai_moderation_guardrail
guardrail_initializer_registry = {
SupportedGuardrailIntegrations.OPENAI_MODERATION.value: initialize_guardrail,
}
guardrail_class_registry = {
SupportedGuardrailIntegrations.OPENAI_MODERATION.value: OpenAIModerationGuardrail,
}

52
litellm/proxy/guardrails/guardrail_hooks/openai/base.py Normal file
View File

@ -0,0 +1,52 @@
from typing import TYPE_CHECKING, List, Optional
if TYPE_CHECKING:
from litellm.types.llms.openai import AllMessageValues
class OpenAIGuardrailBase:
"""
Base class for OpenAI guardrails.
"""
def get_user_prompt(self, messages: List["AllMessageValues"]) -> Optional[str]:
"""
Get the last consecutive block of messages from the user.
Example:
messages = [
{"role": "user", "content": "Hello, how are you?"},
{"role": "assistant", "content": "I'm good, thank you!"},
{"role": "user", "content": "What is the weather in Tokyo?"},
]
get_user_prompt(messages) -> "What is the weather in Tokyo?"
"""
from litellm.litellm_core_utils.prompt_templates.common_utils import (
convert_content_list_to_str,
)
if not messages:
return None
# Iterate from the end to find the last consecutive block of user messages
user_messages = []
for message in reversed(messages):
if message.get("role") == "user":
user_messages.append(message)
else:
# Stop when we hit a non-user message
break
if not user_messages:
return None
# Reverse to get the messages in chronological order
user_messages.reverse()
user_prompt = ""
for message in user_messages:
text_content = convert_content_list_to_str(message)
user_prompt += text_content + "\n"
result = user_prompt.strip()
return result if result else None

431
litellm/proxy/guardrails/guardrail_hooks/openai/moderations.py Normal file
View File

@ -0,0 +1,431 @@
#!/usr/bin/env python3
"""
OpenAI Moderation Guardrail Integration for LiteLLM
"""
from typing import (
TYPE_CHECKING,
Any,
AsyncGenerator,
Dict,
List,
Literal,
Optional,
Type,
Union,
)
from fastapi import HTTPException
from litellm._logging import verbose_proxy_logger
from litellm.integrations.custom_guardrail import (
CustomGuardrail,
log_guardrail_information,
)
from litellm.llms.custom_httpx.http_handler import (
get_async_httpx_client,
httpxSpecialProvider,
)
from .base import OpenAIGuardrailBase
if TYPE_CHECKING:
from litellm.proxy._types import UserAPIKeyAuth
from litellm.types.llms.openai import AllMessageValues, OpenAIModerationResponse
from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
from litellm.types.utils import ModelResponse, ModelResponseStream
class OpenAIModerationGuardrail(OpenAIGuardrailBase, CustomGuardrail):
"""
LiteLLM Built-in Guardrail for OpenAI Content Moderation.
This guardrail scans prompts and responses using the OpenAI Moderation API to detect
harmful content, including violence, hate, harassment, self-harm, sexual content, etc.
Configuration:
guardrail_name: Name of the guardrail instance
api_key: OpenAI API key
api_base: OpenAI API endpoint
model: OpenAI moderation model to use
default_on: Whether to enable by default
"""
def __init__(
self,
guardrail_name: str,
api_key: Optional[str] = None,
api_base: Optional[str] = None,
model: Optional[Literal["omni-moderation-latest", "text-moderation-latest"]] = None,
**kwargs,
):
"""Initialize OpenAI Moderation guardrail handler."""
from litellm.types.guardrails import GuardrailEventHooks
# Initialize parent CustomGuardrail
supported_event_hooks = [
GuardrailEventHooks.pre_call,
GuardrailEventHooks.during_call,
GuardrailEventHooks.post_call,
]
super().__init__(
guardrail_name=guardrail_name,
supported_event_hooks=supported_event_hooks,
**kwargs,
)
self.async_handler = get_async_httpx_client(
llm_provider=httpxSpecialProvider.GuardrailCallback
)
# Store configuration
self.api_key = api_key or self._get_api_key()
self.api_base = api_base or "https://api.openai.com/v1"
self.model: Literal["omni-moderation-latest", "text-moderation-latest"] = model or "omni-moderation-latest"
if not self.api_key:
raise ValueError("OpenAI Moderation: api_key is required. Set OPENAI_API_KEY environment variable or pass it in configuration.")
verbose_proxy_logger.info(
f"Initialized OpenAI Moderation Guardrail: {guardrail_name} with model: {self.model}"
)
def _get_api_key(self) -> Optional[str]:
"""Get API key from environment variables or litellm configuration"""
import os
import litellm
from litellm.secret_managers.main import get_secret_str
return (
os.environ.get("OPENAI_API_KEY")
or litellm.api_key
or litellm.openai_key
or get_secret_str("OPENAI_API_KEY")
)
async def async_make_request(
self, input_text: str
) -> "OpenAIModerationResponse":
"""
Make a request to the OpenAI Moderation API.
"""
request_body = {
"model": self.model,
"input": input_text
}
verbose_proxy_logger.debug(
"OpenAI Moderation guard request: %s", request_body
)
response = await self.async_handler.post(
url=f"{self.api_base}/moderations",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
},
json=request_body,
)
verbose_proxy_logger.debug(
"OpenAI Moderation guard response: %s", response.json()
)
if response.status_code != 200:
raise HTTPException(
status_code=response.status_code,
detail={
"error": "OpenAI Moderation API request failed",
"details": response.text,
},
)
from litellm.types.llms.openai import OpenAIModerationResponse
return OpenAIModerationResponse(**response.json())
def _check_moderation_result(self, moderation_response: "OpenAIModerationResponse") -> None:
"""
Check if the moderation response indicates harmful content and raise exception if needed.
"""
if not moderation_response.results:
return
result = moderation_response.results[0]
if result.flagged:
# Build detailed violation information
violated_categories = []
if result.categories:
for category, is_violated in result.categories.items():
if is_violated:
violated_categories.append(category)
violation_details = {
"violated_categories": violated_categories,
"category_scores": result.category_scores or {},
}
verbose_proxy_logger.warning(
"OpenAI Moderation: Content flagged for violations: %s",
violation_details
)
raise HTTPException(
status_code=400,
detail={
"error": "Violated OpenAI moderation policy",
"moderation_result": violation_details,
},
)
@log_guardrail_information
async def async_pre_call_hook(
self,
user_api_key_dict: "UserAPIKeyAuth",
cache: Any,
data: Dict[str, Any],
call_type: Literal[
"completion",
"text_completion",
"embeddings",
"image_generation",
"moderation",
"audio_transcription",
"pass_through_endpoint",
"rerank",
],
) -> Optional[Dict[str, Any]]:
"""
Pre-call hook to scan user prompts before sending to LLM.
Raises HTTPException if content should be blocked.
"""
verbose_proxy_logger.info(
"OpenAI Moderation: Running pre-call prompt scan, on call_type: %s",
call_type,
)
# Skip moderation calls to avoid infinite recursion
if call_type == "moderation":
return data
new_messages: Optional[List["AllMessageValues"]] = data.get("messages")
if new_messages is None:
verbose_proxy_logger.warning(
"OpenAI Moderation: not running guardrail. No messages in data"
)
return data
user_prompt = self.get_user_prompt(new_messages)
if user_prompt:
verbose_proxy_logger.info(
f"OpenAI Moderation: User prompt: {user_prompt[:100]}..." # Log first 100 chars for debugging
)
moderation_response = await self.async_make_request(
input_text=user_prompt,
)
# Check if content is flagged and raise exception if needed
self._check_moderation_result(moderation_response)
else:
verbose_proxy_logger.warning(
"OpenAI Moderation: No user prompt found"
)
return data
@log_guardrail_information
async def async_moderation_hook(
self,
data: Dict[str, Any],
user_api_key_dict: "UserAPIKeyAuth",
call_type: Literal[
"completion",
"embeddings",
"image_generation",
"moderation",
"audio_transcription",
"responses",
],
) -> Optional[Dict[str, Any]]:
"""
Moderation hook to scan user prompts during call processing.
Raises HTTPException if content should be blocked.
"""
verbose_proxy_logger.info(
"OpenAI Moderation: Running moderation hook, on call_type: %s",
call_type,
)
# Skip moderation calls to avoid infinite recursion
if call_type == "moderation":
return data
new_messages: Optional[List["AllMessageValues"]] = data.get("messages")
if new_messages is None:
verbose_proxy_logger.warning(
"OpenAI Moderation: not running guardrail. No messages in data"
)
return data
user_prompt = self.get_user_prompt(new_messages)
if user_prompt:
moderation_response = await self.async_make_request(
input_text=user_prompt,
)
# Check if content is flagged and raise exception if needed
self._check_moderation_result(moderation_response)
return data
@log_guardrail_information
async def async_post_call_hook(
self,
data: Dict[str, Any],
user_api_key_dict: "UserAPIKeyAuth",
response: "ModelResponse",
) -> "ModelResponse":
"""
Post-call hook to scan LLM responses before returning to user.
Raises HTTPException if response should be blocked.
"""
verbose_proxy_logger.info(
"OpenAI Moderation: Running post-call response scan"
)
# Extract response text for moderation
response_text = self._extract_response_text(response)
if response_text:
verbose_proxy_logger.info(
f"OpenAI Moderation: Response text: {response_text[:100]}..." # Log first 100 chars
)
moderation_response = await self.async_make_request(
input_text=response_text,
)
# Check if content is flagged and raise exception if needed
self._check_moderation_result(moderation_response)
return response
@log_guardrail_information
async def async_post_call_streaming_iterator_hook(
self,
user_api_key_dict: "UserAPIKeyAuth",
response: Any,
request_data: Dict[str, Any],
) -> AsyncGenerator["ModelResponseStream", None]:
"""
Process streaming response chunks for OpenAI moderation.
Collects all chunks from the stream, assembles them into a complete response,
and applies moderation check. If content violates moderation policy, raises HTTPException.
"""
# Import here to avoid circular imports
from litellm.llms.base_llm.base_model_iterator import MockResponseIterator
from litellm.main import stream_chunk_builder
from litellm.types.utils import TextCompletionResponse
verbose_proxy_logger.info(
"OpenAI Moderation: Running streaming response scan"
)
# Collect all chunks to process them together
all_chunks: List["ModelResponseStream"] = []
async for chunk in response:
all_chunks.append(chunk)
# Assemble the complete response from chunks
assembled_model_response: Optional[
Union["ModelResponse", TextCompletionResponse]
] = stream_chunk_builder(
chunks=all_chunks,
)
if isinstance(assembled_model_response, (type(None), TextCompletionResponse)):
# If we can't assemble a ModelResponse or it's a text completion,
# just yield the original chunks without moderation
verbose_proxy_logger.warning(
"OpenAI Moderation: Could not assemble ModelResponse from chunks, skipping moderation"
)
for chunk in all_chunks:
yield chunk
return
# Extract response text for moderation
response_text = self._extract_response_text(assembled_model_response)
if response_text:
verbose_proxy_logger.info(
f"OpenAI Moderation: Streaming response text: {response_text[:100]}..." # Log first 100 chars
)
# Make moderation request - this will raise HTTPException if content is flagged
moderation_response = await self.async_make_request(
input_text=response_text,
)
# Check if content is flagged and raise exception if needed
self._check_moderation_result(moderation_response)
# If we reach here, content passed moderation - yield the original chunks
mock_response = MockResponseIterator(
model_response=assembled_model_response
)
# Return the reconstructed stream
async for chunk in mock_response:
yield chunk
def _extract_response_text(self, response: "ModelResponse") -> Optional[str]:
"""
Extract text content from the model response for moderation.
"""
if not hasattr(response, 'choices') or not response.choices:
return None
response_texts = []
for choice in response.choices:
try:
# Try to get content from message (chat completion)
message = getattr(choice, 'message', None)
if message:
content = getattr(message, 'content', None)
if content and isinstance(content, str):
response_texts.append(content)
continue
# Try to get text (text completion)
text = getattr(choice, 'text', None)
if text and isinstance(text, str):
response_texts.append(text)
continue
# Try to get content from delta (streaming)
delta = getattr(choice, 'delta', None)
if delta:
content = getattr(delta, 'content', None)
if content and isinstance(content, str):
response_texts.append(content)
continue
except (AttributeError, TypeError):
# Skip choices that don't have expected attributes
continue
return "\n".join(response_texts) if response_texts else None
@staticmethod
def get_config_model() -> Optional[Type["GuardrailConfigModel"]]:
"""
Get the config model for the OpenAI Moderation guardrail.
"""
from litellm.types.proxy.guardrails.guardrail_hooks.openai.openai_moderation import (
OpenAIModerationGuardrailConfigModel,
)
return OpenAIModerationGuardrailConfigModel

10
litellm/types/guardrails.py
View File

@ -5,6 +5,10 @@ from typing import Any, Dict, List, Literal, Optional, TypedDict, Union
from pydantic import BaseModel, ConfigDict, Field, SecretStr
from typing_extensions import Required, TypedDict
from litellm.types.proxy.guardrails.guardrail_hooks.openai.openai_moderation import (
OpenAIModerationGuardrailConfigModel,
)
"""
Pydantic object defining how to set guardrails on litellm proxy
@ -33,7 +37,7 @@ class SupportedGuardrailIntegrations(Enum):
PANW_PRISMA_AIRS = "panw_prisma_airs"
AZURE_PROMPT_SHIELD = "azure/prompt_shield"
AZURE_TEXT_MODERATIONS = "azure/text_moderations"
OPENAI_MODERATION = "openai_moderation"
class Role(Enum):
SYSTEM = "system"
@ -386,6 +390,10 @@ class BaseLitellmParams(BaseModel): # works for new and patch update guardrails
default=None, description="Recipe for output (LLM response)"
)
model: Optional[str] = Field(
default=None, description="Optional field if guardrail requires a 'model' parameter"
)
model_config = ConfigDict(extra="allow", protected_namespaces=())

32
litellm/types/proxy/guardrails/guardrail_hooks/openai/openai_moderation.py Normal file
View File

@ -0,0 +1,32 @@
from typing import Literal, Optional
from pydantic import BaseModel, Field
from ..base import GuardrailConfigModel
class BaseOpenAIModerationGuardrailConfigModel(GuardrailConfigModel):
"""Base configuration model for the OpenAI Moderation guardrail"""
model: Optional[Literal["omni-moderation-latest", "text-moderation-latest"]] = Field(
default="omni-moderation-latest",
description="The OpenAI moderation model to use. 'omni-moderation-latest' supports more categorization options and multi-modal inputs. Defaults to 'omni-moderation-latest'.",
)
class OpenAIModerationGuardrailConfigModel(BaseOpenAIModerationGuardrailConfigModel):
"""Configuration model for the OpenAI Moderation guardrail"""
api_key: Optional[str] = Field(
default=None,
description="OpenAI API key. Can also be set via OPENAI_API_KEY environment variable.",
)
api_base: Optional[str] = Field(
default="https://api.openai.com/v1",
description="OpenAI API base URL. Defaults to 'https://api.openai.com/v1'.",
)
@staticmethod
def ui_friendly_name() -> str:
return "OpenAI Moderation"

377
tests/test_litellm/proxy/guardrails/guardrail_hooks/openai/test_moderations.py Normal file
View File

@ -0,0 +1,377 @@
#!/usr/bin/env python3
"""
Test OpenAI Moderation Guardrail
"""
import os
import sys
sys.path.insert(0, os.path.abspath("../../../../../.."))
import asyncio
from unittest.mock import MagicMock, patch
import pytest
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.guardrails.guardrail_hooks.openai.moderations import (
OpenAIModerationGuardrail,
)
from litellm.types.llms.openai import OpenAIModerationResponse, OpenAIModerationResult
@pytest.mark.asyncio
async def test_openai_moderation_guardrail_init():
"""Test OpenAI moderation guardrail initialization"""
with patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}):
guardrail = OpenAIModerationGuardrail(
guardrail_name="test-openai-moderation",
)
assert guardrail.guardrail_name == "test-openai-moderation"
assert guardrail.api_key == "test-key"
assert guardrail.model == "omni-moderation-latest"
assert guardrail.api_base == "https://api.openai.com/v1"
@pytest.mark.asyncio
async def test_openai_moderation_guardrail_adds_to_litellm_callbacks():
"""Test that OpenAI moderation guardrail adds itself to litellm callbacks during initialization"""
import litellm
from litellm.proxy.guardrails.guardrail_hooks.openai import (
initialize_guardrail as openai_initialize_guardrail,
)
from litellm.types.guardrails import (
Guardrail,
LitellmParams,
SupportedGuardrailIntegrations,
)
# Clear existing callbacks for clean test
original_callbacks = litellm.callbacks.copy()
litellm.logging_callback_manager._reset_all_callbacks()
try:
with patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}):
guardrail_litellm_params = LitellmParams(
guardrail=SupportedGuardrailIntegrations.OPENAI_MODERATION,
api_key="test-key",
model="omni-moderation-latest",
mode="pre_call"
)
guardrail = openai_initialize_guardrail(
litellm_params=guardrail_litellm_params,
guardrail=Guardrail(
guardrail_name="test-openai-moderation",
litellm_params=guardrail_litellm_params
)
)
# Check that the guardrail was added to litellm callbacks
assert guardrail in litellm.callbacks
assert len(litellm.callbacks) == 1
# Verify it's the correct guardrail
callback = litellm.callbacks[0]
assert isinstance(callback, OpenAIModerationGuardrail)
assert callback.guardrail_name == "test-openai-moderation"
finally:
# Restore original callbacks
litellm.logging_callback_manager._reset_all_callbacks()
for callback in original_callbacks:
litellm.logging_callback_manager.add_litellm_callback(callback)
@pytest.mark.asyncio
async def test_openai_moderation_guardrail_safe_content():
"""Test OpenAI moderation guardrail with safe content"""
with patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}):
guardrail = OpenAIModerationGuardrail(
guardrail_name="test-openai-moderation",
)
# Mock safe moderation response
mock_response = OpenAIModerationResponse(
id="modr-123",
model="omni-moderation-latest",
results=[
OpenAIModerationResult(
flagged=False,
categories={
"sexual": False,
"hate": False,
"harassment": False,
"self-harm": False,
"violence": False,
},
category_scores={
"sexual": 0.001,
"hate": 0.001,
"harassment": 0.001,
"self-harm": 0.001,
"violence": 0.001,
},
category_applied_input_types={
"sexual": [],
"hate": [],
"harassment": [],
"self-harm": [],
"violence": [],
}
)
]
)
with patch.object(guardrail, 'async_make_request', return_value=mock_response):
# Test pre-call hook with safe content
user_api_key_dict = UserAPIKeyAuth(api_key="test")
data = {
"messages": [
{"role": "user", "content": "Hello, how are you today?"}
]
}
result = await guardrail.async_pre_call_hook(
user_api_key_dict=user_api_key_dict,
cache=None,
data=data,
call_type="completion"
)
# Should return the original data unchanged
assert result == data
@pytest.mark.asyncio
async def test_openai_moderation_guardrail_harmful_content():
"""Test OpenAI moderation guardrail with harmful content"""
with patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}):
guardrail = OpenAIModerationGuardrail(
guardrail_name="test-openai-moderation",
)
# Mock harmful moderation response
mock_response = OpenAIModerationResponse(
id="modr-123",
model="omni-moderation-latest",
results=[
OpenAIModerationResult(
flagged=True,
categories={
"sexual": False,
"hate": True,
"harassment": False,
"self-harm": False,
"violence": False,
},
category_scores={
"sexual": 0.001,
"hate": 0.95,
"harassment": 0.001,
"self-harm": 0.001,
"violence": 0.001,
},
category_applied_input_types={
"sexual": [],
"hate": ["text"],
"harassment": [],
"self-harm": [],
"violence": [],
}
)
]
)
with patch.object(guardrail, 'async_make_request', return_value=mock_response):
# Test pre-call hook with harmful content
user_api_key_dict = UserAPIKeyAuth(api_key="test")
data = {
"messages": [
{"role": "user", "content": "This is hateful content"}
]
}
# Should raise HTTPException
from fastapi import HTTPException
with pytest.raises(HTTPException) as exc_info:
await guardrail.async_pre_call_hook(
user_api_key_dict=user_api_key_dict,
cache=None,
data=data,
call_type="completion"
)
assert exc_info.value.status_code == 400
assert "Violated OpenAI moderation policy" in str(exc_info.value.detail)
@pytest.mark.asyncio
async def test_openai_moderation_guardrail_streaming_safe_content():
"""Test OpenAI moderation guardrail with streaming safe content"""
with patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}):
guardrail = OpenAIModerationGuardrail(
guardrail_name="test-openai-moderation",
)
# Mock safe moderation response
mock_response = OpenAIModerationResponse(
id="modr-123",
model="omni-moderation-latest",
results=[
OpenAIModerationResult(
flagged=False,
categories={
"sexual": False,
"hate": False,
"harassment": False,
"self-harm": False,
"violence": False,
},
category_scores={
"sexual": 0.001,
"hate": 0.001,
"harassment": 0.001,
"self-harm": 0.001,
"violence": 0.001,
},
category_applied_input_types={
"sexual": [],
"hate": [],
"harassment": [],
"self-harm": [],
"violence": [],
}
)
]
)
# Mock streaming chunks
async def mock_stream():
# Simulate streaming chunks with safe content
chunks = [
MagicMock(choices=[MagicMock(delta=MagicMock(content="Hello "))]),
MagicMock(choices=[MagicMock(delta=MagicMock(content="world"))]),
MagicMock(choices=[MagicMock(delta=MagicMock(content="!"))])
]
for chunk in chunks:
yield chunk
# Mock the stream_chunk_builder to return a proper ModelResponse
mock_model_response = MagicMock()
mock_model_response.choices = [
MagicMock(message=MagicMock(content="Hello world!"))
]
with patch.object(guardrail, 'async_make_request', return_value=mock_response), \
patch('litellm.main.stream_chunk_builder', return_value=mock_model_response), \
patch('litellm.llms.base_llm.base_model_iterator.MockResponseIterator') as mock_iterator:
# Mock the iterator to yield the original chunks
async def mock_yield_chunks():
chunks = [
MagicMock(choices=[MagicMock(delta=MagicMock(content="Hello "))]),
MagicMock(choices=[MagicMock(delta=MagicMock(content="world"))]),
MagicMock(choices=[MagicMock(delta=MagicMock(content="!"))])
]
for chunk in chunks:
yield chunk
mock_iterator.return_value.__aiter__ = lambda self: mock_yield_chunks()
user_api_key_dict = UserAPIKeyAuth(api_key="test")
request_data = {
"messages": [
{"role": "user", "content": "Hello, how are you today?"}
]
}
# Test streaming hook with safe content
result_chunks = []
async for chunk in guardrail.async_post_call_streaming_iterator_hook(
user_api_key_dict=user_api_key_dict,
response=mock_stream(),
request_data=request_data
):
result_chunks.append(chunk)
# Should return all chunks without blocking
assert len(result_chunks) == 3
@pytest.mark.asyncio
async def test_openai_moderation_guardrail_streaming_harmful_content():
"""Test OpenAI moderation guardrail with streaming harmful content"""
with patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}):
guardrail = OpenAIModerationGuardrail(
guardrail_name="test-openai-moderation",
)
# Mock harmful moderation response
mock_response = OpenAIModerationResponse(
id="modr-123",
model="omni-moderation-latest",
results=[
OpenAIModerationResult(
flagged=True,
categories={
"sexual": False,
"hate": True,
"harassment": False,
"self-harm": False,
"violence": False,
},
category_scores={
"sexual": 0.001,
"hate": 0.95,
"harassment": 0.001,
"self-harm": 0.001,
"violence": 0.001,
},
category_applied_input_types={
"sexual": [],
"hate": ["text"],
"harassment": [],
"self-harm": [],
"violence": [],
}
)
]
)
# Mock streaming chunks with harmful content
async def mock_stream():
chunks = [
MagicMock(choices=[MagicMock(delta=MagicMock(content="This is "))]),
MagicMock(choices=[MagicMock(delta=MagicMock(content="harmful content"))])
]
for chunk in chunks:
yield chunk
# Mock the stream_chunk_builder to return a ModelResponse with harmful content
mock_model_response = MagicMock()
mock_model_response.choices = [
MagicMock(message=MagicMock(content="This is harmful content"))
]
with patch.object(guardrail, 'async_make_request', return_value=mock_response), \
patch('litellm.main.stream_chunk_builder', return_value=mock_model_response):
user_api_key_dict = UserAPIKeyAuth(api_key="test")
request_data = {
"messages": [
{"role": "user", "content": "Generate harmful content"}
]
}
# Should raise HTTPException when processing streaming harmful content
from fastapi import HTTPException
with pytest.raises(HTTPException) as exc_info:
result_chunks = []
async for chunk in guardrail.async_post_call_streaming_iterator_hook(
user_api_key_dict=user_api_key_dict,
response=mock_stream(),
request_data=request_data
):
result_chunks.append(chunk)
assert exc_info.value.status_code == 400
assert "Violated OpenAI moderation policy" in str(exc_info.value.detail)

17
ui/litellm-dashboard/src/components/guardrails/guardrail_table.tsx
View File

@ -16,6 +16,7 @@ import {
ChevronDownIcon,
} from "@heroicons/react/outline";
import { Tooltip } from "antd";
import { Badge } from "@tremor/react";
import {
ColumnDef,
flexRender,
@ -174,6 +175,22 @@ const GuardrailTable: React.FC<GuardrailTableProps> = ({
);
},
},
{
header: "Default On",
accessorKey: "litellm_params.default_on",
cell: ({ row }) => {
const guardrail = row.original;
return (
<Badge
color={guardrail.litellm_params?.default_on ? "green" : "gray"}
className="text-xs font-normal"
size="xs"
>
{guardrail.litellm_params?.default_on ? "Default On" : "Default Off"}
</Badge>
);
},
},
{
header: "Created At",
accessorKey: "created_at",