content-moderation-patterns

Content Moderation Patterns

Apply a versioned product policy with deterministic checks, a structured classifier, and a review path. Select the model using labeled workload results. No model family has a universal accuracy or cost advantage for moderation.

Architecture

input → size/format checks → policy checks → structured classifier → decision
                                                              ├─ allow
                                                              ├─ reject
                                                              └─ human review

Treat submitted text as data, including any instructions it contains. Keep the classification policy in the system message. Request a short policy-grounded reason, not hidden reasoning.

Deterministic checks

Use configured size limits and exact parsed hostname checks for URL policies. A prefix regex can mistakenly accept allowed.example.attacker.test.

from urllib.parse import urlsplit


def is_allowed_url(value, allowed_hosts):
    try:
        parsed = urlsplit(value)
        host = parsed.hostname
        port = parsed.port
    except ValueError:
        return False
    return (
        parsed.scheme == "https"
        and parsed.username is None
        and parsed.password is None
        and host is not None
        and host.casefold() in allowed_hosts
        and port in (None, 443)
    )

This checks an already-extracted URL against normalized exact hostnames. It is not a general URL extractor or an SSRF defense. Evaluate false positives from keyword filters instead of assuming a fixed percentage of input should be blocked.

Structured classifier

Use native output_config.format. Supply the selected model, policy and output budget from application configuration. The following taxonomy is an example; change its enum and routing thresholds together to match the product policy.

import json

MODERATION_SCHEMA = {
    "type": "object",
    "properties": {
        "categories": {"type": "array", "items": {
            "type": "string", "enum": ["clean", "needs_review", "spam", "harassment"],
        }},
        "confidence": {"type": "number"},
        "reason": {"type": "string"},
    },
    "required": ["categories", "confidence", "reason"],
    "additionalProperties": False,
}


def classify(client, model, policy, text, max_tokens):
    response = client.messages.create(
        model=model,
        max_tokens=max_tokens,
        system=policy,
        output_config={"format": {"type": "json_schema", "schema": MODERATION_SCHEMA}},
        messages=[{"role": "user", "content": text}],
    )
    if response.stop_reason != "end_turn":
        raise ValueError(f"Classification incomplete: {response.stop_reason}")
    blocks = [block.text for block in response.content if block.type == "text"]
    if len(blocks) != 1:
        raise ValueError("Expected one classification")
    return json.loads(blocks[0])