Content moderation at scale

Lesson updated November 3, 2026 based on OpenAI documentation: temperature and top_p are only accepted with a reasoning effort of none, whereas GPT-5.6 models reason by default (medium). Calls that set these parameters must therefore specify reasoning={"effort": "none"} (Responses API) or reasoning_effort="none" (Chat Completions). Sources: Using GPT-6 · GPT-5.6 Terra.

Protecting your users in production

As soon as your application goes live, you must handle inappropriate content—both on input (malicious prompts) and output (problematic responses). This two-way requirement often catches teams off guard. Most people think first about filtering what users send, but reputational risk concentrates on what your application publishes under your name. A well-aligned model rarely generates shocking content on its own, but it can do so when pushed—and your brand appears at the top of the screenshot.

We’ll move from a single API call to a high-throughput asynchronous pipeline, scaling up step by step.

The first layer: the Moderations API

The Moderations API classifies text across several risk categories and returns both a binary verdict and detailed scores. It’s fast and free, making it the natural first filter for any public application. The function below only surfaces categories with scores above 0.3—a deliberately low threshold. At this stage, you want to see what’s approaching the line, not just what crosses it.

import openai

client = openai.OpenAI()

def moderate_content(text: str) -> dict:
    """Check whether text contains inappropriate content."""
    response = client.moderations.create(
        model="omni-moderation-latest",
        input=text,
    )

    result = response.results[0]
    return {
        "flagged": result.flagged,
        "categories": {
            cat: score
            for cat, score in result.category_scores.__dict__.items()
            if score > 0.3
        },
    }

# Example
result = moderate_content("How do I bake a chocolate cake?")
print(f"Flagged: {result['flagged']}")

In production, a binary verdict isn’t enough. Depending on the content and context, you may want to allow it, rewrite it, block it, or escalate it to a human—four distinct decisions you should name explicitly in code rather than improvising with booleans. The pipeline below formalizes these actions, applies per-category thresholds, and adds a second layer for prompt-injection detection.

from enum import Enum
from dataclasses import dataclass

class ModerationAction(Enum):
    ALLOW = "allow"
    MODIFY = "modify"
    BLOCK = "block"
    ESCALATE = "escalate"

@dataclass
class ModerationResult:
    action: ModerationAction
    reason: str | None = None
    modified_content: str | None = None

class ModerationPipeline:
    """Multi-layer moderation pipeline."""

    def __init__(self, thresholds: dict | None = None):
        self.thresholds = thresholds or {
            "harassment": 0.7,
            "hate": 0.7,
            "self-harm": 0.5,
            "sexual": 0.7,
            "violence": 0.7,
        }

    def moderate_input(self, text: str) -> ModerationResult:
        """Moderate input content (user prompt)."""
        moderation = client.moderations.create(
            model="omni-moderation-latest",
            input=text,
        )

        result = moderation.results[0]
        if result.flagged:
            problematic_categories = [
                cat for cat, flagged in result.categories.__dict__.items()
                if flagged
            ]
            return ModerationResult(
                action=ModerationAction.BLOCK,
                reason=f"Inappropriate content: {', '.join(problematic_categories)}",
            )

        # Prompt injection detection
        if self._detect_injection(text):
            return ModerationResult(
                action=ModerationAction.BLOCK,
                reason="Prompt injection attempt detected",
            )

        return ModerationResult(action=ModerationAction.ALLOW)

    def moderate_output(self, response: str) -> ModerationResult:
        """Moderate output content (LLM response)."""
        moderation = client.moderations.create(
            model="omni-moderation-latest",
            input=response,
        )

        if moderation.results[0].flagged:
            return ModerationResult(
                action=ModerationAction.BLOCK,
                reason="Generated response contains inappropriate content",
            )

        return ModerationResult(action=ModerationAction.ALLOW)

    def _detect_injection(self, text: str) -> bool:
        """Detect prompt injection attempts."""
        markers = [
            "ignore previous instructions",
            "ignore your instructions",
            "forget your guidelines",
            "you are now",
            "new role",
            "system prompt",
            "jailbreak",
        ]
        text_lower = text.lower()
        return any(m in text_lower for m in markers)

The 0.5 threshold for self-harm—stricter than the 0.7 used for other categories—illustrates the reasoning you should apply to each threshold. Here, the cost of a false positive is nowhere near the cost of a false negative, so you lower the bar. Make this trade-off category by category, based on your audience, rather than applying a uniform value.

Marker-based injection detection comes with a caveat. It catches naive attempts, which represent the bulk of the volume, and nothing more: a slightly different phrasing or a translated version slips through. Treat it as a noise filter, not a defense. Real protection against injection is architectural—never give the model capabilities that a hijacked prompt could exploit.

Handling load

On a high-traffic application, synchronous moderation becomes a bottleneck: every message waits for its verdict before processing. An asynchronous moderator decouples these steps by queuing content for consumption by multiple workers, while also tracking useful statistics along the way.

import asyncio

class AsyncModerator:
    """Asynchronously moderate content for high throughput."""

    def __init__(self, num_workers: int = 10):
        self.pipeline = ModerationPipeline()
        self.queue: asyncio.Queue = asyncio.Queue()
        self.num_workers = num_workers
        self.stats = {"allowed": 0, "blocked": 0, "escalated": 0}

    async def worker(self):
        while True:
            task = await self.queue.get()
            try:
                result = self.pipeline.moderate_input(task["text"])
                task["callback"](result)

                match result.action:
                    case ModerationAction.ALLOW:
                        self.stats["allowed"] += 1
                    case ModerationAction.BLOCK:
                        self.stats["blocked"] += 1
                    case ModerationAction.ESCALATE:
                        self.stats["escalated"] += 1
            finally:
                self.queue.task_done()

    async def start(self):
        for _ in range(self.num_workers):
            asyncio.create_task(self.worker())

    def report(self) -> str:
        total = sum(self.stats.values())
        if total == 0:
            return "No content moderated."
        return (
            f"Total: {total} | "
            f"Allowed: {self.stats['allowed']} ({self.stats['allowed']/total:.1%}) | "
            f"Blocked: {self.stats['blocked']} ({self.stats['blocked']/total:.1%})"
        )

The block rate produced by report is your primary calibration metric. If it suddenly jumps to 15% after running at 0.5%, only two explanations exist: you’re under a coordinated attack, or you just broke your thresholds. In either case, you want to know within the hour, not at the next monthly review.

One final limitation deserves mention: the Moderations API knows nothing about your business. A message that’s perfectly acceptable in general may violate your own rules—requesting medical advice from a banking assistant, trying to obtain an off-grid rate, or bypassing a procedure. Contextual moderation adds this layer by submitting the message to a model with your explicit rules, but only after the standard filter has allowed it through.

def contextual_moderation(
    text: str,
    app_context: str,
    business_rules: list[str],
) -> ModerationResult:
    """Context-aware moderation tailored to the application."""
    pipeline = ModerationPipeline()
    standard_result = pipeline.moderate_input(text)

    if standard_result.action == ModerationAction.BLOCK:
        return standard_result

    verification_prompt = (
        f"Application context: {app_context}\n\n"
        f"Business rules to verify:\n"
        + "\n".join(f"- {r}" for r in business_rules)
        + f"\n\nUser message: {text}\n\n"
        "Does the message comply with all rules? "
        "Answer yes or no with a brief justification."
    )

    response = client.responses.create(
        model="gpt-5.6-terra",
        reasoning={"effort": "none"},  # temperature requires effort none
        input=verification_prompt,
        max_output_tokens=100,
        temperature=0.0,
    )

    if "no" in response.output_text.lower()[:10]:
        return ModerationResult(
            action=ModerationAction.BLOCK,
            reason=response.output_text,
        )

    return ModerationResult(action=ModerationAction.ALLOW)

The order of these two checks matters. The free, fast filter runs first, and the paid model call happens only for messages that pass it. On real traffic, this covers nearly all messages. Doing the reverse—paying for a model call on content the Moderations API would have rejected in milliseconds—would be wasteful.

Keeping a defensible record

A moderation decision is a decision you’ll be held accountable for, one way or another. You need a log—but one that doesn’t become a problem itself. The comment in the code highlights the most important line in the file: record the action, the reason, and the length, never the raw text. Storing user messages in a log file exposes you to data-protection obligations far heavier than the service provided.

import json
import time
from pathlib import Path

class ModerationLog:
    """Log all moderation decisions for audit purposes."""

    def __init__(self, log_path: str = "logs/moderation.jsonl"):
        self.path = Path(log_path)
        self.path.parent.mkdir(parents=True, exist_ok=True)

    def record(
        self,
        text: str,
        result: ModerationResult,
        source: str = "input",
    ):
        entry = {
            "timestamp": time.time(),
            "source": source,
            "action": result.action.value,
            "reason": result.reason,
            "text_length": len(text),
            # DO NOT log raw text for privacy
        }

        with open(self.path, "a") as f:
            f.write(json.dumps(entry) + "\n")

If you must still be able to retrieve a specific message during a dispute, log a hash of the text rather than the text itself. This lets you confirm that a given piece of content matches a recorded decision without retaining the content on your side.

Key takeaways

• Moderate both inputs (prompts) and outputs (responses)
• Use the Moderations API as your first, fast, free layer
• Adjust thresholds category by category based on the real cost of a false negative
• Add injection detection knowing it only catches naive attempts
• Log all decisions for audit without storing raw content