Ai Moderating Content

Auto-Moderate What Users Post

Guide the user through building AI content moderation — classify user-generated content, score severity, and route decisions (auto-approve, human-review, auto-reject). The pattern: classify, score, route.

When NOT to use AI moderation

• Low-volume content — if a human can review everything in under an hour per day, skip AI. The complexity of maintaining a moderation pipeline is not worth it.
• Exact-match violations only — if your policy is just a blocklist of words or regex patterns (SSNs, emails, phone numbers), use pattern matching directly. No LM needed.
• Legal-grade decisions — AI moderation is a first pass, not a legal ruling. If a wrong moderation decision has legal consequences (DMCA takedowns, defamation claims), always route to human review.

Consider /ai-sorting instead if you just need classification without severity scoring or routing logic.

Step 1: Define your moderation policy

Ask the user: 1. What content do you need to catch? (hate speech, spam, NSFW, harassment, self-harm, illegal activity, PII) 2. What are the severity levels? (warning, remove, ban) 3. What is the tolerance for false positives? (over-moderating frustrates users) 4. Is human review in the loop? (auto-only vs. auto + human escalation)

Step 2: Choose your approach

Step 3: Build the moderator

Classification + severity scoring + routing decision:

import dspy
from typing import Literal

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

VIOLATIONS = Literal[
    "safe", "spam", "hate_speech", "harassment",
    "violence", "nsfw", "self_harm", "illegal",
]

class ModerateContent(dspy.Signature):
    """Assess user-generated content against platform policies."""
    content: str = dspy.InputField(desc="user-generated content to moderate")
    platform_context: str = dspy.InputField(desc="where this content appears, e.g. 'product review'")
    violation_type: VIOLATIONS = dspy.OutputField()
    severity: Literal["none", "low", "medium", "high"] = dspy.OutputField()
    explanation: str = dspy.OutputField(desc="brief reason for the decision")

class ContentModerator(dspy.Module):
    def __init__(self):
        self.assess = dspy.ChainOfThought(ModerateContent)

    def forward(self, content, platform_context="social media post"):
        result = self.assess(content=content, platform_context=platform_context)

        # Route based on severity
        if result.severity == "high":
            decision = "remove"
        elif result.severity == "medium":
            decision = "human_review"
        elif result.severity == "low":
            decision = "warn"
        else:
            decision = "approve"

        return dspy.Prediction(
            violation_type=result.violation_type,
            severity=result.severity,
            decision=decision,
            explanation=result.explanation,
        )

# Usage
moderator = ContentModerator()
result = moderator(content="Great product, works exactly as described!")
print(result.decision)  # "approve"

result = moderator(content="This seller is a scammer, I'll find where they live")
print(result.decision)  # "remove"
print(result.violation_type)  # "harassment"

Step 4: Multi-label moderation

Content can violate multiple policies at once (e.g., spam and contains PII):

VIOLATION_TYPES = ["safe", "spam", "hate_speech", "harassment", "violence", "nsfw", "self_harm", "illegal"]

class MultiLabelModerate(dspy.Signature):
    """Flag all policy violations in user content. Content may have multiple violations."""
    content: str = dspy.InputField()
    platform_context: str = dspy.InputField()
    violations: list[str] = dspy.OutputField(desc=f"all that apply from: {VIOLATION_TYPES}")
    severity: Literal["none", "low", "medium", "high"] = dspy.OutputField(
        desc="overall severity based on the worst violation"
    )
    explanation: str = dspy.OutputField()

class MultiLabelModerator(dspy.Module):
    def __init__(self):
        self.assess = dspy.ChainOfThought(MultiLabelModerate)

    def forward(self, content, platform_context=""):
        return self.assess(content=content, platform_context=platform_context)

def multi_label_reward(args, pred):
    # Validate that returned violations are from the allowed set
    if all(v in VIOLATION_TYPES for v in pred.violations):
        return 1.0
    return 0.0

validated_moderator = dspy.Refine(
    module=MultiLabelModerator(),
    N=3,
    reward_fn=multi_label_reward,
    threshold=1.0,
)

Step 5: Hard blocks with pattern matching

For zero-tolerance patterns, block instantly with pattern matching — no LM needed:

import re

class StrictModerator(dspy.Module):
    def __init__(self):
        self.assess = dspy.ChainOfThought(ModerateContent)

    def forward(self, content, platform_context=""):
        # Pattern-based hard blocks (instant, no LM needed)
        if re.search(r"\b\d{3}-\d{2}-\d{4}\b", content):
            return dspy.Prediction(
                violation_type="illegal",
                severity="high",
                decision="remove",
                explanation="Content contains SSN pattern — auto-reject",
            )
        if re.search(
            r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",
            content,
        ):
            return dspy.Prediction(
                violation_type="illegal",
                severity="high",
                decision="remove",
                explanation="Content contains email addresses — redact before posting",
            )
        if re.search(r"\b\d{16}\b", content):
            return dspy.Prediction(
                violation_type="illegal",
                severity="high",
                decision="remove",
                explanation="Content contains potential credit card number — auto-reject",
            )

        # LM-based assessment for everything else
        return self.assess(content=content, platform_context=platform_context)

Pattern-based blocks are faster, cheaper, and more reliable than LM-based detection for well-defined patterns (SSNs, credit cards, emails). Use regex for structure, LMs for semantics.

Step 6: Confidence-based routing

Route uncertain decisions to human reviewers instead of making bad calls:

class ConfidentModerate(dspy.Signature):
    """Moderate content and rate your confidence in the assessment."""
    content: str = dspy.InputField()
    platform_context: str = dspy.InputField()
    violation_type: VIOLATIONS = dspy.OutputField()
    severity: Literal["none", "low", "medium", "high"] = dspy.OutputField()
    confidence: float = dspy.OutputField(desc="0.0 to 1.0 — how sure are you about this assessment?")
    explanation: str = dspy.OutputField()

class ConfidentModerator(dspy.Module):
    def __init__(self, confidence_threshold=0.7):
        self.assess = dspy.ChainOfThought(ConfidentModerate)
        self.confidence_threshold = confidence_threshold

    def forward(self, content, platform_context=""):
        result = self.assess(content=content, platform_context=platform_context)

        # Clamp confidence to valid range
        confidence = max(0.0, min(1.0, result.confidence))

        # Route based on confidence + severity
        if confidence < self.confidence_threshold:
            decision = "human_review"  # uncertain — always escalate
        elif result.severity == "high":
            decision = "remove"
        elif result.severity == "medium":
            decision = "human_review"
        elif result.severity == "low":
            decision = "warn"
        else:
            decision = "approve"

        return dspy.Prediction(
            violation_type=result.violation_type,
            severity=result.severity,
            confidence=confidence,
            decision=decision,
            explanation=result.explanation,
        )

Step 7: Metrics and optimization

Define moderation metrics

def moderation_metric(example, prediction, trace=None):
    """Weighted score: type matters more than severity."""
    type_correct = float(prediction.violation_type == example.violation_type)
    severity_correct = float(prediction.severity == example.severity)
    return 0.7 * type_correct + 0.3 * severity_correct

Per-category metrics (more useful than overall accuracy)

def make_category_metric(category):
    """Create a precision metric for a specific violation category."""
    def metric(example, prediction, trace=None):
        if example.violation_type == category:
            return float(prediction.violation_type == category)  # recall
        else:
            return float(prediction.violation_type != category)  # precision
    return metric

# Track each category separately
hate_speech_metric = make_category_metric("hate_speech")
spam_metric = make_category_metric("spam")

Optimize the moderator

trainset = [
    dspy.Example(
        content="Buy cheap watches at spam-site.com!!!",
        platform_context="product review",
        violation_type="spam",
        severity="medium",
    ).with_inputs("content", "platform_context"),
    dspy.Example(
        content="This product changed my life, highly recommend!",
        platform_context="product review",
        violation_type="safe",
        severity="none",
    ).with_inputs("content", "platform_context"),
    # 50-200 labeled examples for good optimization
]

optimizer = dspy.MIPROv2(metric=moderation_metric, auto="medium")
optimized = optimizer.compile(moderator, trainset=trainset)

Step 8: Handle tricky cases

• Sarcasm and satire — "Oh sure, what a great product" is not hate speech. Context matters. The platform_context field helps here.
• Quoting to criticize — "The seller said 'you are an idiot'" is reporting harassment, not committing it. Include instructions in your signature to distinguish.
• Code snippets — Variable names or test strings might contain offensive words. If your platform has code, add a code-detection step before moderation.
• Non-English content — LMs handle major languages well but may miss nuance in less-common languages. Consider language-specific test sets.
• Adversarial evasion — Users will try to bypass moderation (leetspeak, Unicode tricks, word splitting). Test your moderator with /ai-testing-safety.

Gotchas

• Claude adds a `reasoning` field to signatures used with ChainOfThought. Do not add your own reasoning output field — DSPy injects one automatically. Adding a second causes duplicate or conflicting reasoning outputs.
• Use programmatic checks (not `dspy.Refine`) for hard PII blocks. For zero-tolerance patterns like SSNs or credit card numbers, check with regex before calling the LM and return a structured rejection immediately. dspy.Refine is for output quality constraints that benefit from retrying the LM, not for instant pattern-based blocks.
• Claude uses `Literal[list]` instead of `Literal[tuple(list)]` for dynamic categories. If violation types come from a database or config, you must use Literal[tuple(categories)] — Literal[list] silently fails type validation.
• LM confidence scores are not calibrated probabilities. When Claude builds a confidence-based router, it treats the 0.0-1.0 confidence output as if 0.7 means 70% accurate. LM self-reported confidence is directionally useful but not calibrated — tune the threshold empirically on your dev set, not based on the number itself.
• Over-moderating borderline content is worse than under-moderating. Claude defaults to being cautious and tends to classify borderline content as violations. For moderation, false positives (removing safe content) hurt user engagement more than false negatives. Bias your metric toward precision over recall for low-severity categories.

Cross-references

Install any skill: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill <name>

• Classification patterns for general sorting and categorization -- see /ai-sorting
• Output guardrails for moderating your own AI responses -- see /ai-checking-outputs
• Adversarial testing to stress-test your moderator -- see /ai-testing-safety
• Production monitoring to track moderation quality over time -- see /ai-monitoring
• Signatures for defining input/output contracts -- see /dspy-signatures
• ChainOfThought for the reasoning module used in moderation -- see /dspy-chain-of-thought
• Install `/ai-do` if you do not have it — it routes any AI problem to the right skill and is the fastest way to work: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill ai-do

Additional resources

• For complete worked examples, see examples.md

Content Moderation Examples

Example 1: Community forum moderation

A community forum needs to auto-moderate user comments. Categories: safe, spam, toxic, off-topic. Human reviewers handle uncertain cases.

Set up the moderator

import dspy
from typing import Literal

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

class ModerateComment(dspy.Signature):
    """Classify a forum comment for moderation."""
    comment: str = dspy.InputField(desc="user comment to moderate")
    thread_topic: str = dspy.InputField(desc="what the thread is about")
    category: Literal["safe", "spam", "toxic", "off_topic"] = dspy.OutputField()
    severity: Literal["none", "low", "medium", "high"] = dspy.OutputField()
    confidence: float = dspy.OutputField(desc="0.0 to 1.0")
    explanation: str = dspy.OutputField(desc="brief reason")

class ForumModerator(dspy.Module):
    def __init__(self):
        self.assess = dspy.ChainOfThought(ModerateComment)

    def forward(self, comment, thread_topic="general discussion"):
        result = self.assess(comment=comment, thread_topic=thread_topic)

        # Confidence-based routing
        if result.confidence < 0.7:
            decision = "human_review"
        elif result.severity == "high":
            decision = "remove"
        elif result.severity == "medium":
            decision = "human_review"
        elif result.severity == "low":
            decision = "warn"
        else:
            decision = "approve"

        return dspy.Prediction(
            category=result.category,
            severity=result.severity,
            confidence=result.confidence,
            decision=decision,
            explanation=result.explanation,
        )

Prepare labeled training data

trainset = [
    dspy.Example(
        comment="Has anyone tried the new update? It fixed the crash I was having.",
        thread_topic="software updates",
        category="safe",
        severity="none",
    ).with_inputs("comment", "thread_topic"),
    dspy.Example(
        comment="BUY CHEAP WATCHES AT www.spam-site.com BEST PRICES!!!",
        thread_topic="software updates",
        category="spam",
        severity="medium",
    ).with_inputs("comment", "thread_topic"),
    dspy.Example(
        comment="Anyone who uses this software is a complete moron",
        thread_topic="software updates",
        category="toxic",
        severity="high",
    ).with_inputs("comment", "thread_topic"),
    dspy.Example(
        comment="Hey does anyone know a good recipe for banana bread?",
        thread_topic="software updates",
        category="off_topic",
        severity="low",
    ).with_inputs("comment", "thread_topic"),
    dspy.Example(
        comment="I disagree with the previous poster. The old version was more stable.",
        thread_topic="software updates",
        category="safe",
        severity="none",
    ).with_inputs("comment", "thread_topic"),
    dspy.Example(
        comment="Check out my profile for amazing deals on electronics!",
        thread_topic="software updates",
        category="spam",
        severity="low",
    ).with_inputs("comment", "thread_topic"),
    dspy.Example(
        comment="You're all idiots if you think this feature is good",
        thread_topic="software updates",
        category="toxic",
        severity="medium",
    ).with_inputs("comment", "thread_topic"),
    dspy.Example(
        comment="This is somewhat related but has anyone compared it to CompetitorApp?",
        thread_topic="software updates",
        category="safe",
        severity="none",
    ).with_inputs("comment", "thread_topic"),
    # ... 200 labeled examples total for production quality
]

# Split into train/dev
split = int(len(trainset) * 0.8)
train, dev = trainset[:split], trainset[split:]

Evaluate baseline and optimize

from dspy.evaluate import Evaluate

def moderation_metric(example, prediction, trace=None):
    category_correct = float(prediction.category == example.category)
    severity_correct = float(prediction.severity == example.severity)
    return 0.7 * category_correct + 0.3 * severity_correct

evaluator = Evaluate(devset=dev, metric=moderation_metric, num_threads=4, display_table=5)

moderator = ForumModerator()
baseline = evaluator(moderator)
print(f"Baseline: {baseline:.1f}%")
# Output: Baseline: 72.0%

# Optimize
optimizer = dspy.MIPROv2(metric=moderation_metric, auto="medium")
optimized = optimizer.compile(moderator, trainset=train)

optimized_score = evaluator(optimized)
print(f"Optimized: {optimized_score:.1f}%")
# Output: Optimized: 89.0%

Check per-category performance

for category in ["safe", "spam", "toxic", "off_topic"]:
    cat_examples = [e for e in dev if e.category == category]
    if cat_examples:
        cat_evaluator = Evaluate(devset=cat_examples, metric=moderation_metric, num_threads=4)
        score = cat_evaluator(optimized)
        print(f"  {category}: {score:.1f}%")

# Output:
#   safe: 95.0%
#   spam: 88.0%
#   toxic: 85.0%
#   off_topic: 78.0%

Deploy with confidence routing

# In production, track routing distribution
stats = {"approve": 0, "warn": 0, "human_review": 0, "remove": 0}

for comment_text in incoming_comments:
    result = optimized(comment=comment_text, thread_topic=thread_topic)
    stats[result.decision] += 1

    if result.decision == "remove":
        hide_comment(comment_text)
    elif result.decision == "human_review":
        queue_for_review(comment_text, result.explanation)
    elif result.decision == "warn":
        add_warning_label(comment_text)

print(f"Routing: {stats}")
# Typical: approve 75%, warn 10%, human_review 10%, remove 5%

---

Example 2: Marketplace listing moderation

An online marketplace needs to moderate product listings for prohibited items, misleading claims, and PII in descriptions. Listings can violate multiple policies at once.

Set up multi-label moderation with hard blocks

import dspy
import re
from typing import Literal

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

LISTING_VIOLATIONS = [
    "clean", "prohibited_item", "misleading_claims",
    "counterfeit", "pii_exposed", "inappropriate_images_described",
]

class ModerateListing(dspy.Signature):
    """Moderate a marketplace product listing. Flag all policy violations."""
    title: str = dspy.InputField(desc="product listing title")
    description: str = dspy.InputField(desc="product listing description")
    price: str = dspy.InputField(desc="listed price")
    violations: list[str] = dspy.OutputField(desc=f"all that apply from: {LISTING_VIOLATIONS}")
    severity: Literal["none", "low", "medium", "high"] = dspy.OutputField()
    explanation: str = dspy.OutputField()

class ListingModerator(dspy.Module):
    def __init__(self):
        self.assess = dspy.ChainOfThought(ModerateListing)

    def forward(self, title, description, price):
        full_text = f"{title} {description}"

        # Hard blocks: PII patterns (instant, no LM needed) — return rejection immediately
        if re.search(r"\b\d{3}-\d{2}-\d{4}\b", full_text):
            return dspy.Prediction(
                violations=["pii_exposed"], severity="high",
                decision="reject", explanation="Listing contains SSN — auto-reject and notify seller",
            )
        if re.search(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b", full_text):
            return dspy.Prediction(
                violations=["pii_exposed"], severity="high",
                decision="reject", explanation="Listing contains email — ask seller to remove before publishing",
            )
        if re.search(r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b", full_text):
            return dspy.Prediction(
                violations=["pii_exposed"], severity="high",
                decision="reject", explanation="Listing contains phone number — ask seller to remove",
            )

        # LM-based assessment
        result = self.assess(
            title=title,
            description=description,
            price=price,
        )

        # Route
        if result.severity == "high" or "prohibited_item" in result.violations:
            decision = "reject"
        elif result.severity == "medium" or len(result.violations) > 1:
            decision = "human_review"
        elif result.severity == "low":
            decision = "request_edit"
        else:
            decision = "approve"

        return dspy.Prediction(
            violations=result.violations,
            severity=result.severity,
            decision=decision,
            explanation=result.explanation,
        )


def listing_violations_reward(args, pred):
    """Reward function: penalize if violations contain values outside the allowed set."""
    score = 1.0
    if not all(v in LISTING_VIOLATIONS for v in pred.violations):
        score -= 0.5  # soft penalty — LM should stay within known categories
    return score

validated_moderator = dspy.Refine(
    module=ListingModerator(), N=3, reward_fn=listing_violations_reward, threshold=0.8
)

Training data with multi-label examples

trainset = [
    dspy.Example(
        title="Vintage Leather Jacket - Size M",
        description="Genuine leather, barely worn. Great condition. Smoke-free home.",
        price="$85",
        violations=["clean"],
        severity="none",
    ).with_inputs("title", "description", "price"),
    dspy.Example(
        title="GUARANTEED Weight Loss Pills - Lose 30lbs in 1 Week!",
        description="Doctor-recommended miracle supplement. 100% guaranteed results or your money back. FDA approved.",
        price="$29.99",
        violations=["misleading_claims"],
        severity="high",
    ).with_inputs("title", "description", "price"),
    dspy.Example(
        title="Designer Handbag - Looks Just Like Gucci",
        description="High quality replica. Indistinguishable from the real thing. Same materials and craftsmanship.",
        price="$45",
        violations=["counterfeit"],
        severity="high",
    ).with_inputs("title", "description", "price"),
    dspy.Example(
        title="Used Textbook - Calculus 101",
        description="Some highlighting. Contact me at seller@email.com for bundle deals. Call 555-123-4567.",
        price="$30",
        violations=["pii_exposed"],
        severity="medium",
    ).with_inputs("title", "description", "price"),
    dspy.Example(
        title="AMAZING Deal Electronics - Best Price EVER!!!",
        description="Buy now before they're gone! Limited stock! We beat ANY price! Not sold in stores!",
        price="$9.99",
        violations=["misleading_claims"],
        severity="low",
    ).with_inputs("title", "description", "price"),
    dspy.Example(
        title="Replica Rolex + Weight Loss Combo",
        description="Get a luxury watch AND lose weight! Both guaranteed authentic and effective.",
        price="$99",
        violations=["counterfeit", "misleading_claims"],
        severity="high",
    ).with_inputs("title", "description", "price"),
    # ... 100+ examples for production
]

Evaluate and optimize

from dspy.evaluate import Evaluate

def listing_metric(example, prediction, trace=None):
    """Multi-label metric: check violation overlap and severity."""
    expected = set(example.violations)
    predicted = set(prediction.violations)

    if not expected and not predicted:
        violation_score = 1.0
    elif not expected or not predicted:
        violation_score = 0.0
    else:
        intersection = expected & predicted
        union = expected | predicted
        violation_score = len(intersection) / len(union)  # Jaccard similarity

    severity_score = float(prediction.severity == example.severity)
    return 0.6 * violation_score + 0.4 * severity_score

split = int(len(trainset) * 0.8)
train, dev = trainset[:split], trainset[split:]

evaluator = Evaluate(devset=dev, metric=listing_metric, num_threads=4, display_table=5)

moderator = ListingModerator()
baseline = evaluator(moderator)
print(f"Baseline: {baseline:.1f}%")

optimizer = dspy.MIPROv2(metric=listing_metric, auto="medium")
optimized = optimizer.compile(moderator, trainset=train)

optimized_score = evaluator(optimized)
print(f"Optimized: {optimized_score:.1f}%")

# Save for production
optimized.save("listing_moderator.json")

Production integration

# Process incoming listings
moderator = ListingModerator()
moderator.load("listing_moderator.json")

def moderate_new_listing(listing):
    result = moderator(
        title=listing["title"],
        description=listing["description"],
        price=listing["price"],
    )
    return {
        "decision": result.decision,
        "violations": result.violations,
        "severity": result.severity,
        "explanation": result.explanation,
    }

Result: The multi-label moderator catches listings that violate multiple policies simultaneously (e.g., counterfeit + misleading claims), while pattern-based hard blocks instantly catch PII before the LM even runs. dspy.Refine retries the LM-based assessment up to 3 times when violation categories look off, ensuring outputs stay within the known category set. Confidence-based routing sends ~12% of listings to human reviewers — the ones where the moderator is least certain.