The mental model for content moderation is a typed pipeline, not a wall. Think of it as a series of transforms on a message object: the message enters, each stage annotates it with scores and metadata, and the final stage dispatches to an action. That action might be 'allow', 'block with user-visible reason', 'silently rewrite', 'add a warning banner', or 'queue for human review'. The key insight is that the classifier produces a score, not a decision. You write the policy that converts scores into decisions, and that policy lives in your code, not in the model.
Under the hood, dedicated toxicity classifiers like Google's Perspective API, OpenAI's Moderation endpoint, or open-source models like unitary/toxic-bert and cardiffnlp/twitter-roberta-base-offensive are fine-tuned text classifiers. They output per-category probability scores: harassment, hate speech, sexual content, violence, self-harm, and so on. The OpenAI Moderation API returns a results array with categories (boolean) and category_scores (float). You should use the float scores, not the booleans, because the boolean thresholds are calibrated for OpenAI's own use case, not yours. If you let the provider decide the threshold, you lose control of your false-positive rate. A score of 0.78 on harassment means something different on a teen gaming platform than on a professional enterprise tool.
A real-world scenario: you're building a customer support chat product where an LLM answers questions on behalf of a brand. You need two moderation passes. The first is pre-LLM: check the incoming user message for prompt injection attempts, attempts to extract confidential data, and genuinely harmful content. The second is post-LLM: check the model's response before it reaches the user to catch cases where the model was manipulated into producing something harmful or off-brand. These two passes serve different purposes. Pre-LLM moderation protects your system. Post-LLM moderation protects your users from your system. A senior engineer also adds a third, async pass: log flagged items and surface them to a human moderator queue so your policy can evolve based on real patterns. Tools like Langfuse or a simple Postgres queue with an internal review UI work well here.
The tradeoffs between approaches are real. Using a dedicated classifier (Perspective API, OpenAI Moderation) is fast (often sub-100ms), inexpensive, and purpose-built. The downside is that these models have their own biases: they frequently over-flag African American Vernacular English, LGBTQ+ content, and discussion of violence in historical or fictional contexts. An LLM-as-judge approach (sending content to GPT-4o-mini with a structured moderation prompt) is more contextually aware and handles nuance better, but adds 300-800ms of latency and costs per-call. Hybrid architectures use the fast classifier as a first pass; only when the score falls in a middle band (say, 0.4-0.75) do you escalate to the more expensive LLM judgment call. Below 0.4 you pass, above 0.75 you block, and in the middle you pay for the smarter call. This keeps median-case latency low while handling edge cases well.
What changes at scale: at 10 users, you can afford a synchronous classifier call on every message with no optimization. At 10,000 users you start noticing that moderation adds 80-150ms to p99 latency, and you consider running the classifier asynchronously while optimistically allowing the message, then retroactively suppressing it if it scores badly. At 10 million users you need distributed queuing, per-tenant policy configuration stored in a fast cache like Redis, dedicated classifier infrastructure separate from your LLM inference cluster, category-specific thresholds stored in a database tunable by your trust-and-safety team without code deploys, and feedback loops that retrain or recalibrate models on your actual traffic distribution. You also need legal clarity on data retention of flagged content across jurisdictions, which ties directly to the compliance topics covered in the regulations subtopic.
Key Takeaways
- Treat moderation as a pipeline stage, not a single boolean check, with distinct pre- and post-LLM hooks.
- Tune classification thresholds per context, not globally — a gaming chat and a medical app need different calibrations.
- Always log every moderation decision with scores so you can audit, retrain, and catch drift over time.
- Combine a fast, cheap dedicated classifier with an LLM fallback for ambiguous or high-stakes cases.
Pro tips
- Never use the provider's boolean
flaggedoutput directly. Pull the floatcategory_scoresand apply your own thresholds stored in config, not in code. This lets your trust-and-safety team adjust sensitivity without a deploy. - Run your moderation classifier against a labeled holdout set from your own traffic at least monthly. Classifier drift is real: slang evolves, your user base shifts, and a model that was well-calibrated six months ago may now have a 15% false-positive rate on your core use case.
- Fail open, not closed, when the moderation API is unavailable, but flag every failure for async review. Silently blocking all messages during an outage is a worse user experience than occasionally missing a moderation call that gets caught in review.
- The 'ambiguous middle band' pattern (use cheap classifier first, escalate only scores between 0.4-0.75 to an LLM judge) cuts your LLM moderation costs by roughly 80-90% on typical traffic distributions while preserving nuanced judgment exactly where it's needed.
Common pitfalls
- Mistake: Applying one global threshold across all categories and contexts. Fix: Store per-category, per-context thresholds in a config database so different product surfaces can be tuned independently without code changes.
- Mistake: Only moderating user input, not LLM output. Fix: Add a post-generation moderation pass before the response is returned; jailbreaks can cause models to produce harmful content that your input filter never saw.
- Mistake: Logging only final block/allow decisions without storing the raw scores. Fix: Always persist full category_scores to a queryable store so you can audit decisions, tune thresholds retroactively, and build training data for future classifiers.
- Mistake: Treating moderation failures (API timeout, 500 error) as blocks. Fix: Implement graceful degradation with fail-open behavior plus an async review queue so legitimate users are not silently denied service during outages.
When to use dedicated classifiers vs LLM-as-judge vs hybrid
| Option | Use when | Avoid when |
|---|---|---|
| Dedicated classifier (OpenAI Moderation, Perspective API, toxic-bert) | High-throughput, latency-sensitive paths where you need sub-100ms screening and costs must stay near zero per call. | Your content involves heavy domain-specific language, satire, fiction, or community-specific slang the classifier was not trained on. |
| LLM-as-judge (GPT-4o-mini with structured moderation prompt) | You need nuanced contextual judgment, handling ambiguous cases, multi-turn context awareness, or explanation generation for moderation decisions. | Every message must be checked synchronously at scale; the added latency and per-call cost become prohibitive above moderate traffic volumes. |
| Hybrid (classifier first, LLM escalation for middle band) | You need both low median latency and high accuracy on edge cases; most production consumer products at scale land here. | Your traffic is very low volume, where the added architectural complexity outweighs the cost savings of skipping LLM calls. |
| Self-hosted open-source classifier (unitary/toxic-bert, Detoxify) | Data privacy requirements prevent sending user content to third-party APIs, or you need to fine-tune on proprietary labeled data. | You lack ML infrastructure for model serving and monitoring; running your own inference adds significant operational burden. |
Code Example
# openai>=1.0.0
from openai import OpenAI
client = OpenAI() # reads OPENAI_API_KEY from env
def check_moderation(text: str) -> dict:
"""Returns category scores dict. Use scores, not boolean flags."""
response = client.moderations.create(input=text)
result = response.results[0]
return {
"flagged": result.flagged,
"scores": result.category_scores.model_dump(),
}
# Example usage
user_message = "I want to hurt someone"
mod_result = check_moderation(user_message)
HARASSMENT_THRESHOLD = 0.6
if mod_result["scores"]["harassment"] > HARASSMENT_THRESHOLD:
print("BLOCKED: harassment score too high")
else:
print("ALLOWED:", mod_result["scores"])How this code works
This code demonstrates how to implement a basic content moderation system using OpenAI's Moderation API to identify potentially harmful user input. Its primary job is to analyze a given text, categorize its potential toxicity, and then decide whether to allow or block it based on specific risk thresholds.
The process begins by initializing an OpenAI client, which securely reads the necessary API key from an environment variable. The core logic resides in the check_moderation function. This function takes a text string, sends it to the client.moderations.create endpoint, and receives a detailed response. From this response, it extracts a flagged status and, crucially, a scores dictionary containing individual category scores like "harassment" and "violence". The category_scores.model_dump() method is vital here, converting the API's internal object into a standard Python dictionary for easy access. Finally, the example usage sets a HARASSMENT_THRESHOLD. If the harassment score from mod_result["scores"] surpasses this threshold, the content is "BLOCKED"; otherwise, it's "ALLOWED". The lesson emphasizes using these granular scores for decision-making, rather than just the general flagged boolean, to provide more nuanced control over moderation outcomes.
Production-grade example
Adds retries, per-category thresholds, structured logging, timeouts, and graceful degradation on API failure.
# openai>=1.0.0, tenacity>=8.0, structlog>=23.0
import os
import time
import structlog
from openai import OpenAI, APIStatusError, APITimeoutError
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
log = structlog.get_logger()
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"], timeout=5.0)
# Per-category thresholds tuned for your product context
THRESHOLDS = {
"harassment": 0.65,
"hate": 0.70,
"self-harm": 0.50,
"sexual": 0.80,
"violence": 0.75,
}
@retry(
retry=retry_if_exception_type((APIStatusError, APITimeoutError)),
wait=wait_exponential(multiplier=0.5, min=0.5, max=8),
stop=stop_after_attempt(3),
)
def _call_moderation_api(text: str) -> dict:
return client.moderations.create(input=text).results[0]
def moderate(text: str, request_id: str) -> dict:
"""Returns {allowed: bool, action: str, scores: dict, latency_ms: float}."""
start = time.perf_counter()
try:
result = _call_moderation_api(text)
scores = result.category_scores.model_dump()
latency_ms = (time.perf_counter() - start) * 1000
triggered = [
cat for cat, threshold in THRESHOLDS.items()
if scores.get(cat, 0.0) > threshold
]
allowed = len(triggered) == 0
action = "allow" if allowed else "block"
log.info(
"moderation_decision",
request_id=request_id,
allowed=allowed,
action=action,
triggered_categories=triggered,
scores={k: round(v, 4) for k, v in scores.items()},
latency_ms=round(latency_ms, 2),
)
return {"allowed": allowed, "action": action, "scores": scores, "latency_ms": latency_ms}
except (APIStatusError, APITimeoutError) as exc:
latency_ms = (time.perf_counter() - start) * 1000
log.error("moderation_api_failure", request_id=request_id,
error=str(exc), latency_ms=round(latency_ms, 2))
# Graceful degradation: fail open with a flag for async review
return {"allowed": True, "action": "allow_on_failure", "scores": {}, "latency_ms": latency_ms}How this code works
This code provides a robust content moderation service using the OpenAI API. Its job is to evaluate text for harmful content, like harassment or violence, and decide whether to allow or block it based on toxicity levels. It starts by importing necessary tools: OpenAI for interacting with the moderation API, structlog for structured logging, and tenacity for making API calls more resilient. The client is set up with an API key, and THRESHOLDS defines custom acceptable toxicity scores for various categories. This allows tailoring the moderation strictness for different types of harmful content relevant to a product's context.
The core _call_moderation_api function sends text to OpenAI. It's wrapped with the @retry decorator, which is crucial for handling flaky network or API issues by automatically retrying failed calls (specifically APIStatusError or APITimeoutError) a few times before giving up. The main moderate function measures latency_ms, calls the API, and then compares the returned category_scores against the predefined THRESHOLDS. If any score exceeds its threshold, the content is triggered and the action becomes "block." All decisions and relevant scores are recorded using structlog.info. A subtle but important detail is the try...except block: if the moderation API itself fails, the code doesn't crash or indiscriminately block. Instead, it logs the moderation_api_failure and gracefully "fails open" by returning allowed: True with action: "allow_on_failure", ensuring the application continues to function while signaling that human review is needed.
Practice & master
Try the exercise, check your understanding, then mark this lesson mastered to track your path to pro.
Exercise
Build a two-pass moderation function that checks a user message before sending it to an LLM, then checks the LLM's response before returning it. Use the OpenAI Moderation API for the pre-pass and a GPT-4o-mini prompt for the post-pass. Log both decisions with scores. Return the final response or a safe error message if either pass blocks.
# openai>=1.0.0
import os
from openai import OpenAI
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
PRE_THRESHOLDS = {"harassment": 0.6, "hate": 0.65, "violence": 0.7}
def pre_moderate(user_message: str) -> bool:
"""Return True if the message is safe to send to the LLM."""
# TODO: call client.moderations.create and check category_scores
# against PRE_THRESHOLDS. Return False if any threshold exceeded.
pass
def post_moderate(llm_response: str) -> bool:
"""Return True if the LLM response is safe to show the user."""
# TODO: send llm_response to GPT-4o-mini with a system prompt asking
# it to return JSON: {"safe": true/false, "reason": "..."}.
# Parse the JSON and return the safe field.
pass
def safe_chat(user_message: str) -> str:
if not pre_moderate(user_message):
return "I can't respond to that."
# TODO: call GPT-4o-mini to get a response
llm_response = ""
if not post_moderate(llm_response):
return "I ran into an issue generating a safe response."
return llm_response
if __name__ == "__main__":
print(safe_chat("Tell me about the history of chemical weapons."))
print(safe_chat("What is the capital of France?"))Quick check
You're getting too many false positives on legitimate user messages. What is the most targeted fix?
The moderation API goes down during peak traffic. What should your system do?
Why should you store the raw float category_scores rather than just the final allow/block decision?