Whether you run a community forum, a social platform, or an e-commerce marketplace with user-generated content, you need moderation. Manual review doesn't scale, and rule-based filters miss the nuance of human language. AI-powered content moderation is the solution — and with modern LLM APIs, it's easier to build than ever.
This tutorial walks through building a production-ready content moderation system using AI APIs, complete with code examples in Python and JavaScript.
Why AI Content Moderation?
Traditional content moderation approaches have severe limitations:
- Keyword blocklists — Easily bypassed with misspellings, slang, or encoded language. They also generate false positives on legitimate content.
- Manual review — Expensive, slow, and psychologically harmful to human moderators who must review toxic content.
- Rule-based systems — Brittle and require constant updating as new forms of abuse emerge.
AI-powered moderation, by contrast, understands context, nuance, and intent. It can detect harassment even when no profanity is used, identify coordinated manipulation campaigns, and classify content across multiple dimensions simultaneously — all in milliseconds.
Architecture Overview
Here's the system we'll build:
User Content → Pre-filter (keywords) → AI Moderation API → Decision Engine → Action
↓
Audit Log + AnalyticsThe pre-filter catches obvious violations cheaply. The AI model handles the nuanced cases. The decision engine applies your policies and logs everything for compliance.
Step 1: Set Up Your API Client
We'll use the Qubax AI API for this tutorial because it provides access to multiple models (GPT-5.6, Claude, Gemini) through a single endpoint with unified pricing. But the same approach works with any OpenAI-compatible API.
Python Setup
import httpx
import json
import asyncio
from dataclasses import dataclass
from enum import Enum
class Severity(Enum):
SAFE = "safe"
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
@dataclass
class ModerationResult:
is_safe: bool
severity: Severity
categories: list[str]
explanation: str
confidence: float
class AIModerator:
def __init__(self, api_key: str, base_url: str = "https://api.qubax.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.client = httpx.AsyncClient(
timeout=30.0,
headers={"Authorization": f"Bearer {api_key}"}
)
async def moderate(self, content: str) -> ModerationResult:
response = await self.client.post(
f"{self.base_url}/chat/completions",
json={
"model": "gpt-5.6-mini", # Fast and cost-effective
"messages": [
{
"role": "system",
"content": self._get_system_prompt()
},
{
"role": "user",
"content": f"Moderate this content:\n\n{content}"
}
],
"temperature": 0.1, # Low temperature for consistency
"max_tokens": 200,
"response_format": {"type": "json_object"}
}
)
result = response.json()
analysis = json.loads(result["choices"][0]["message"]["content"])
return ModerationResult(
is_safe=analysis["is_safe"],
severity=Severity(analysis["severity"]),
categories=analysis.get("categories", []),
explanation=analysis.get("explanation", ""),
confidence=analysis.get("confidence", 0.0)
)
def _get_system_prompt(self) -> str:
return """You are a content moderation AI. Analyze the content and return a JSON object with:
- is_safe (boolean): whether the content is appropriate
- severity (string): "safe", "low", "medium", "high", "critical"
- categories (array): any violated categories (e.g., "harassment", "spam", "violence", "hate_speech", "sexual_content", "personal_info")
- explanation (string): brief reason for the decision
- confidence (float): 0.0 to 1.0
Be conservative with user safety. When in doubt, flag for review."""
async def close(self):
await self.client.aclose()JavaScript/Node.js Setup
class AIModerator {
constructor(apiKey, baseUrl = 'https://api.qubax.ai/v1') {
this.apiKey = apiKey;
this.baseUrl = baseUrl;
}
async moderate(content) {
const response = await fetch(`${this.baseUrl}/chat/completions`, {
method: 'POST',
headers: {
'Authorization': `Bearer ${this.apiKey}`,
'Content-Type': 'application/json'
},
body: JSON.stringify({
model: 'gpt-5.6-mini',
messages: [
{ role: 'system', content: this.getSystemPrompt() },
{ role: 'user', content: `Moderate this content:\n\n${content}` }
],
temperature: 0.1,
max_tokens: 200,
response_format: { type: 'json_object' }
})
});
const result = await response.json();
return JSON.parse(result.choices[0].message.content);
}
getSystemPrompt() {
return `You are a content moderation AI. Analyze the content and return a JSON object with:
- is_safe (boolean)
- severity (string): "safe", "low", "medium", "high", "critical"
- categories (array): violated categories
- explanation (string): brief reason
- confidence (float): 0.0 to 1.0`;
}
}
module.exports = { AIModerator };Step 2: Add a Pre-Filter Layer
Before sending content to the AI model, run a fast keyword check. This saves API costs on obvious violations.
import re
class PreFilter:
BANNED_PATTERNS = [
r'(?i)(bitcoin\\s+investment|crypto\\s+giveaway)',
r'(?i)(free\\s+(iphone|money|gift))',
r'https?://(?:spam|phishing|scam)\\.',
]
def __init__(self):
self.patterns = [re.compile(p) for p in self.BANNED_PATTERNS]
def check(self, content: str) -> bool:
"""Returns True if content should be blocked immediately."""
return any(pattern.search(content) for pattern in self.patterns)This catches obvious spam and scams without spending a single API call.
Step 3: Build the Decision Engine
The decision engine takes the AI's analysis and applies your specific business rules.
from typing import Optional
class DecisionEngine:
def __init__(self, auto_block_severity: str = "high"):
self.auto_block_threshold = {
"safe": 0, "low": 1, "medium": 2,
"high": 3, "critical": 4
}[auto_block_severity]
def decide(self, result: ModerationResult) -> dict:
action = "allow"
reason = "Content approved"
severity_level = self.auto_block_threshold
if self._severity_value(result.severity) >= severity_level:
action = "block"
reason = f"Blocked: {', '.join(result.categories)}"
elif result.severity in [Severity.MEDIUM, Severity.LOW]:
if result.confidence < 0.8:
action = "flag_for_review"
reason = "Low confidence - sent for human review"
else:
action = "allow_with_warning"
reason = "Content allowed with warning"
return {
"action": action,
"reason": reason,
"result": result
}
def _severity_value(self, severity: Severity) -> int:
values = {
Severity.SAFE: 0, Severity.LOW: 1,
Severity.MEDIUM: 2, Severity.HIGH: 3,
Severity.CRITICAL: 4
}
return values[severity]Step 4: Add Caching for Cost Optimization
Many pieces of content are duplicates or near-duplicates. Cache moderation results to avoid paying for the same analysis twice.
import hashlib
class ModerationCache:
def __init__(self, redis_client):
self.redis = redis_client
self.ttl = 86400 # 24 hours
def _key(self, content: str) -> str:
return f"mod:{hashlib.sha256(content.encode()).hexdigest()}"
async def get(self, content: str):
result = await self.redis.get(self._key(content))
return json.loads(result) if result else None
async def set(self, content: str, result: dict):
await self.redis.setex(
self._key(content),
self.ttl,
json.dumps(result)
)Step 5: Put It All Together
class ContentModerationPipeline:
def __init__(self, api_key: str, redis_client=None):
self.pre_filter = PreFilter()
self.moderator = AIModerator(api_key)
self.decision_engine = DecisionEngine(auto_block_severity="high")
self.cache = ModerationCache(redis_client) if redis_client else None
async def process(self, content: str) -> dict:
# Step 1: Pre-filter
if self.pre_filter.check(content):
return {
"action": "block",
"reason": "Caught by pre-filter",
"source": "pre_filter"
}
# Step 2: Check cache
if self.cache:
cached = await self.cache.get(content)
if cached:
return {**cached, "source": "cache"}
# Step 3: AI moderation
try:
result = await self.moderator.moderate(content)
except Exception as e:
# Fail open or closed depending on your risk tolerance
return {
"action": "flag_for_review",
"reason": f"AI moderation failed: {e}",
"source": "error_fallback"
}
# Step 4: Decision
decision = self.decision_engine.decide(result)
# Step 5: Cache and return
if self.cache:
await self.cache.set(content, decision)
return {**decision, "source": "ai"}
# Usage
pipeline = ContentModerationPipeline(api_key="your-qubax-api-key")
# Process content
result = await pipeline.process("This is a great product, highly recommend!")
print(result)
# Output: {"action": "allow", "reason": "Content approved", ...}Step 6: Batch Processing for Scale
If you're moderating large volumes of content (e.g., thousands of comments per minute), process them in batches:
async def moderate_batch(moderator: AIModerator, contents: list[str]) -> list[dict]:
tasks = [moderator.moderate(content) for content in contents]
results = await asyncio.gather(*tasks, return_exceptions=True)
return [
{"content": contents[i], "result": r if not isinstance(r, Exception) else None}
for i, r in enumerate(results)
]Cost Optimization Tips
- Use smaller models first — GPT-5.6-mini or Gemini 3.5 Flash-Lite are 10x cheaper than flagship models. Only escalate to a larger model for borderline cases.
- Cache aggressively — Deduplicate content before sending to the API.
- Batch requests — Process multiple pieces of content in parallel.
- Set token limits — Always cap max_tokens to prevent runaway costs.
- Monitor usage — Use Qubax AI's dashboard to track spending and set budget alerts.
Model Selection Guide
| Use Case | Recommended Model | Approx. Cost per 1K calls |
|---|---|---|
| Basic spam detection | GPT-5.6-mini | ~$0.15 |
| Harassment detection | Claude Sonnet | ~$1.20 |
| Complex policy enforcement | GPT-5.6 | ~$2.50 |
| Multilingual moderation | Gemini 3.6 Flash | ~$0.30 |
FAQ
What AI model is best for content moderation?
For most use cases, a fast and affordable model like GPT-5.6-mini or Gemini 3.5 Flash-Lite is sufficient. For nuanced cases like harassment or hate speech, upgrade to Claude Sonnet or GPT-5.6. Explore all options at Qubax AI models.
How much does AI content moderation cost?
With caching and pre-filtering, you can moderate content for as little as $0.01-$0.05 per 1,000 pieces of content. Without optimization, expect $0.15-$2.50 per 1,000 calls depending on the model.
Should I fail open or closed if the AI API is unavailable?
This depends on your risk tolerance. For platforms with vulnerable users (children, healthcare), fail closed (block content). For general platforms, fail open (allow content and flag for later review).
Can AI detect coordinated manipulation campaigns?
Yes, but it requires analyzing patterns across multiple users and posts rather than individual pieces of content. This typically involves graph analysis combined with AI classification.
How do I handle false positives?
Implement a human review queue for borderline cases, collect feedback from users, and periodically fine-tune your moderation prompts. Low confidence scores should always trigger human review.
