Building a Content Moderation Pipeline for AI Applications

Every AI application that accepts user input or generates text needs content moderation. It's not optional. Whether you're building a chatbot, a document summarizer, or an AI-powered search engine, unmoderated content will eventually cause real harm: users exposed to toxic outputs, PII leaking into responses, or your product being weaponized through prompt injection. The OWASP Top 10 for LLM Applications lists insecure output handling (LLM02) and sensitive information disclosure (LLM06) as top risks, both of which a proper moderation pipeline directly addresses.

The question isn't whether to moderate. It's how to build a moderation pipeline that's fast enough for production, accurate enough to avoid blocking legitimate users, and flexible enough to evolve alongside the threat landscape.

In this guide, we'll walk through how to design and implement a content moderation pipeline for AI applications from scratch. We'll cover architecture decisions, pre-processing and post-processing strategies, code examples you can adapt, and techniques for scaling to high throughput.

Why AI Applications Need a Different Approach

Traditional content moderation focused on user-generated text: forum posts, comments, chat messages. The rules were relatively static, the content was human-authored, and the moderation system only needed to check inputs.

AI applications change the equation in three fundamental ways.

You need to moderate both inputs and outputs. Users can craft prompts designed to make your model produce harmful content, leak sensitive data, or bypass safety guidelines. Even if the input looks benign, the output might not be. A request to "summarize this document" is harmless on its own, but if the document contains hidden instructions, the output could include anything from hate speech to stolen credentials.

Latency budgets are tighter. Users expect AI responses in under a second. Your moderation pipeline needs to complete its work within that window, or you need an async architecture that doesn't block the response.

The threat surface is dynamic. New jailbreak techniques, prompt injection strategies, and adversarial patterns emerge weekly. MITRE ATLAS tracks over 100 documented adversarial ML techniques targeting AI systems, and the catalog grows with each quarter. A static blocklist won't cut it.

Pipeline Architecture Overview

A well-designed moderation pipeline has three stages: pre-processing (before the LLM sees the input), inference-time guardrails, and post-processing (before the user sees the output). Each stage serves a different purpose and catches a different class of problems.

Here's the high-level flow:

User Input
    |
    v
[Pre-Processing Layer]
    ├── Input validation (length, encoding, format)
    ├── Rule-based pattern matching
    └── ML-based threat classification
    |
    v
[LLM Inference]
    |
    v
[Post-Processing Layer]
    ├── Output threat classification
    ├── PII detection and redaction
    └── Content policy enforcement
    |
    v
Safe Response → User

Each layer runs independently, and you can tune the strictness of each one based on your application's risk tolerance. A children's educational app will have much tighter post-processing filters than an internal developer tool.

Pre-Processing: Catching Problems Early

Pre-processing is your first line of defense. Everything that gets caught here saves you an LLM inference call, which means lower cost and lower latency. The goal is to reject clearly malicious or invalid inputs before they reach your model.

Input Validation

Start with the basics. These checks are fast, deterministic, and catch a surprising number of issues:

interface ValidationResult {
  valid: boolean;
  reason?: string;
}
 
function validateInput(text: string): ValidationResult {
  // Reject empty or whitespace-only inputs
  if (!text.trim()) {
    return { valid: false, reason: "empty_input" };
  }
 
  // Enforce length limits (prevents context window abuse)
  if (text.length > 10_000) {
    return { valid: false, reason: "input_too_long" };
  }
 
  // Normalize Unicode to prevent homoglyph attacks
  const normalized = text.normalize("NFKC");
 
  // Check for excessive special characters (common in encoded payloads)
  const specialCharRatio =
    (normalized.replace(/[a-zA-Z0-9\s]/g, "").length) / normalized.length;
  if (specialCharRatio > 0.5) {
    return { valid: false, reason: "suspicious_encoding" };
  }
 
  return { valid: true };
}

Input validation is essentially free from a performance standpoint. There's no reason to skip it.

Threat Classification

After validation, run the input through a threat classifier. This is where you catch prompt injection attempts, jailbreak patterns, and other adversarial inputs before they reach your LLM.

You have two options here: rule-based pattern matching and ML-based classification. In practice, you should use both.

Rule-based patterns are fast and predictable. They catch the most common attack patterns, the ones that account for roughly 70% of real-world attacks. ML-based classification handles the sophisticated cases, like novel phrasings, encoded payloads, and multi-language attacks that regex will never catch.

Here's how to combine both approaches using Wardstone:

import { Wardstone } from "wardstone";
 
const wardstone = new Wardstone({
  apiKey: process.env.WARDSTONE_API_KEY,
});
 
async function preProcess(text: string) {
  // Step 1: Validate input format
  const validation = validateInput(text);
  if (!validation.valid) {
    return {
      allowed: false,
      stage: "validation",
      reason: validation.reason,
    };
  }
 
  // Step 2: ML-based threat classification
  const result = await wardstone.detect(text);
 
  if (result.flagged) {
    return {
      allowed: false,
      stage: "threat_detection",
      category: result.primary_category,
      risk_bands: result.risk_bands,
    };
  }
 
  return { allowed: true, text };
}

The Wardstone detect API classifies inputs across multiple threat categories simultaneously: prompt attacks, content violations, data leakage, and unknown links. A single API call covers all of them, which keeps your pre-processing pipeline simple.

Deciding What to Block vs. Flag

Not every detection needs to result in a hard block. Consider a tiered response strategy:

This approach reduces false positive impact. A user asking about "how SQL injection works" in a cybersecurity context shouldn't be blocked, but it should be logged. Context matters, and your pipeline should reflect that.

Post-Processing: The Last Line of Defense

Post-processing inspects the LLM's output before it reaches the user. This layer is critical because some attacks are designed specifically to bypass input filters.

Indirect prompt injection is the clearest example. As demonstrated by Greshake et al. in their foundational research on indirect prompt injection (arXiv:2302.12173), an attacker embeds instructions in a document, email, or web page that your LLM processes. The user's input is perfectly benign ("summarize this article"), but the LLM's output contains the attacker's payload. Without post-processing, this goes straight to the user.

Output Scanning

Run the same classification pipeline on outputs that you run on inputs:

from wardstone import Wardstone
 
client = Wardstone(api_key="YOUR_API_KEY")
 
def post_process(ai_response: str) -> dict:
    """Scan LLM output before delivering to user."""
    result = client.detect(ai_response)
 
    if result.flagged:
        # Check specific categories
        if result.risk_bands.data_leakage.level != "Low Risk":
            return {
                "safe": False,
                "action": "block",
                "reason": "potential_data_leak",
                "details": result.risk_bands.data_leakage,
            }
 
        if result.risk_bands.content_violation.level != "Low Risk":
            return {
                "safe": False,
                "action": "block",
                "reason": "content_violation",
                "details": result.risk_bands.content_violation,
            }
 
    return {"safe": True, "response": ai_response}

PII Detection and Redaction

Data leakage is one of the most common post-processing concerns. Research from the NIST AI Risk Management Framework (specifically NIST AI 600-1, the Generative AI Profile) identifies information disclosure as one of 12 key risks unique to generative AI systems. LLMs can inadvertently include phone numbers, email addresses, social security numbers, or other PII in their outputs, especially when they've been fed documents containing personal data.

Wardstone's data_leakage category catches structured PII patterns. For additional coverage, you can layer on regex-based detection for domain-specific identifiers:

import re
 
PII_PATTERNS = {
    "ssn": r"\b\d{3}-\d{2}-\d{4}\b",
    "credit_card": r"\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b",
    "phone": r"\b\(\d{3}\)\s?\d{3}-\d{4}\b",
    "email": r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",
}
 
def redact_pii(text: str) -> tuple[str, list[str]]:
    """Redact detected PII and return list of found types."""
    found_types = []
    redacted = text
 
    for pii_type, pattern in PII_PATTERNS.items():
        if re.search(pattern, redacted):
            found_types.append(pii_type)
            redacted = re.sub(pattern, f"[REDACTED_{pii_type.upper()}]", redacted)
 
    return redacted, found_types

Combine ML-based detection with pattern matching for the best coverage. ML catches context-dependent leakage (like someone's full name paired with their address), while regex catches structured formats that are easy to miss in natural language.

The Full Pipeline

Here's a complete implementation that ties pre-processing, LLM inference, and post-processing together:

import { Wardstone } from "wardstone";
 
const wardstone = new Wardstone({
  apiKey: process.env.WARDSTONE_API_KEY,
});
 
interface ModerationResult {
  allowed: boolean;
  response?: string;
  blocked_stage?: string;
  blocked_reason?: string;
  metadata?: Record<string, unknown>;
}
 
async function moderatedInference(
  userInput: string,
  callLLM: (input: string) => Promise<string>
): Promise<ModerationResult> {
  // === PRE-PROCESSING ===
 
  // 1. Input validation
  const validation = validateInput(userInput);
  if (!validation.valid) {
    return {
      allowed: false,
      blocked_stage: "pre_validation",
      blocked_reason: validation.reason,
    };
  }
 
  // 2. Input threat detection
  const inputScan = await wardstone.detect(userInput);
 
  if (inputScan.flagged) {
    logModeration("pre_processing", userInput, inputScan);
    return {
      allowed: false,
      blocked_stage: "pre_threat_detection",
      blocked_reason: `Detected: ${inputScan.primary_category}`,
      metadata: { risk_bands: inputScan.risk_bands },
    };
  }
 
  // === LLM INFERENCE ===
  const aiResponse = await callLLM(userInput);
 
  // === POST-PROCESSING ===
 
  // 3. Output threat detection
  const outputScan = await wardstone.detect(aiResponse);
 
  if (outputScan.flagged) {
    logModeration("post_processing", aiResponse, outputScan);
    return {
      allowed: false,
      blocked_stage: "post_threat_detection",
      blocked_reason: "Response filtered for safety",
      metadata: { risk_bands: outputScan.risk_bands },
    };
  }
 
  return { allowed: true, response: aiResponse };
}

This pipeline adds roughly 50-60ms of total overhead: two Wardstone API calls at ~25-30ms each. For most applications, that's well within acceptable latency budgets. If you need lower latency, read on for scaling strategies.

Handling Edge Cases

Production moderation pipelines encounter situations that clean architecture diagrams don't prepare you for. Here are the most common edge cases and how to handle them.

Context-Dependent Content

The word "kill" appears in violent threats, murder mystery discussions, and software process management. The phrase "how to make a bomb" could be a chemistry student's homework or an actual threat. Context determines whether content is harmful.

This is where ML-based classification outperforms keyword matching. Models trained on diverse, labeled datasets learn to distinguish between these contexts. Wardstone's classifier is trained on over 900,000 labeled examples from 30+ sources, which gives it strong contextual understanding across domains.

For applications where context is especially important (medical, legal, educational), consider adding domain-specific rules that adjust your moderation thresholds:

type AppContext = "medical" | "legal" | "education" | "general";
 
function getThresholds(context: AppContext) {
  const thresholds: Record<AppContext, Record<string, string>> = {
    medical: {
      // Allow clinical discussions of self-harm, drugs, etc.
      content_violation: "Critical Risk",
      data_leakage: "Medium Risk",
    },
    education: {
      // Allow discussion of weapons in historical context
      content_violation: "High Risk",
      data_leakage: "Medium Risk",
    },
    general: {
      content_violation: "Medium Risk",
      data_leakage: "Low Risk",
    },
    legal: {
      content_violation: "High Risk",
      data_leakage: "Low Risk",
    },
  };
  return thresholds[context];
}

Multilingual Content

If your application serves users in multiple languages, your moderation pipeline must handle all of them. This is harder than it sounds. Slurs, coded language, and cultural context vary enormously across languages. A phrase that's innocuous in one language might be deeply offensive in another.

Three practical approaches:

1. Use multilingual models. Transformer-based classifiers trained on multilingual data handle cross-language detection well. Wardstone's model supports content in multiple languages out of the box.

2. Normalize before scanning. Transliteration, Unicode normalization, and script detection help catch evasion techniques like mixing scripts (Cyrillic "a" looks identical to Latin "a" but has a different codepoint).

3. Accept higher uncertainty. For languages with less training data, your model's confidence will be lower. Adjust your thresholds accordingly, and route low-confidence decisions to human review if your application supports it.

Multi-Turn Conversations

A single message might be harmless, but a sequence of messages might constitute an attack. Multi-turn manipulation involves gradually steering a conversation toward restricted territory, with each individual message appearing benign.

To catch this, scan conversation windows rather than individual messages:

def scan_conversation_window(
    messages: list[dict],
    window_size: int = 5
) -> dict:
    """Scan recent conversation history for multi-turn attacks."""
    recent_user_msgs = [
        msg["content"]
        for msg in messages[-window_size:]
        if msg["role"] == "user"
    ]
 
    # Concatenate and scan as a single block
    combined = " ".join(recent_user_msgs)
    result = client.detect(combined)
 
    return {
        "flagged": result.flagged,
        "category": result.primary_category if result.flagged else None,
        "window_size": len(recent_user_msgs),
    }

This adds minimal overhead because you're making the same number of API calls. You're just passing more text per call.

Scaling Your Pipeline

A moderation pipeline that works at 10 requests per second might fall over at 10,000. Here's how to scale without compromising safety.

Async Processing for Non-Blocking Moderation

For applications where blocking on moderation isn't acceptable (streaming responses, real-time chat), use an async architecture. Process the response optimistically and scan it in the background:

import { Wardstone } from "wardstone";
 
const wardstone = new Wardstone({
  apiKey: process.env.WARDSTONE_API_KEY,
});
 
async function streamWithAsyncModeration(
  userInput: string,
  onChunk: (chunk: string) => void,
  onViolation: (result: unknown) => void
) {
  // Pre-processing is still synchronous (fast, blocks bad input)
  const inputScan = await wardstone.detect(userInput);
  if (inputScan.flagged) {
    onViolation(inputScan);
    return;
  }
 
  // Collect chunks for post-processing
  const chunks: string[] = [];
 
  // Stream response to user
  await streamLLMResponse(userInput, (chunk) => {
    chunks.push(chunk);
    onChunk(chunk);
  });
 
  // Post-process asynchronously (don't block the response)
  const fullResponse = chunks.join("");
  wardstone.detect(fullResponse).then((outputScan) => {
    if (outputScan.flagged) {
      // Log violation, alert team, take corrective action
      onViolation(outputScan);
    }
  });
}

The tradeoff is clear: the user might briefly see flagged content before it's caught. For many applications, this is acceptable because you log the violation and can take action (redact the message, warn the user, flag the conversation). For high-risk applications, stick with synchronous post-processing.

Caching Detection Results

If your application sees repeated or similar inputs (common in search, FAQ-style chatbots, or template-driven applications), cache moderation results to avoid redundant API calls:

const moderationCache = new Map<string, {
  result: unknown;
  timestamp: number;
}>();
 
const CACHE_TTL = 5 * 60 * 1000; // 5 minutes
 
async function cachedDetect(text: string) {
  const hash = createHash("sha256").update(text).digest("hex");
  const cached = moderationCache.get(hash);
 
  if (cached && Date.now() - cached.timestamp < CACHE_TTL) {
    return cached.result;
  }
 
  const result = await wardstone.detect(text);
  moderationCache.set(hash, { result, timestamp: Date.now() });
  return result;
}

Keep TTLs short (5-10 minutes). Moderation policies can change, and you don't want stale cache entries allowing content that should now be blocked.

Queue-Based Architecture for High Throughput

At very high volumes, consider a queue-based architecture where moderation workers process requests from a message queue. This decouples your application's throughput from your moderation pipeline's capacity:

User Request → App Server → Message Queue → Moderation Workers
                                  ↓
                          Results Cache/DB

This pattern is especially useful for batch processing scenarios: moderating uploaded documents, scanning conversation histories, or processing backfill jobs. Route time-sensitive requests through a priority queue with stricter latency SLAs, and batch everything else.

Monitoring and Observability

A moderation pipeline without monitoring is a liability. You need to know what's being blocked, why, and whether your pipeline is performing correctly.

Track these key metrics:

Log every moderation decision with enough context for forensic analysis: the input text (or a hash of it), the detection result, the action taken, and a request ID for correlation. This data is invaluable for tuning thresholds, investigating incidents, and demonstrating compliance.

Getting Started

Building a content moderation pipeline doesn't need to be a month-long project. Here's a practical roadmap:

Day 1: Basic input validation and threat detection. Add input length checks and integrate the Wardstone API for threat classification. This alone blocks the majority of attacks and harmful content.

Week 1: Add post-processing. Scan LLM outputs for content violations and data leakage before they reach users. Add structured logging for all moderation events.

Week 2: Handle edge cases. Implement context-aware thresholds, conversation window scanning for multi-turn attacks, and PII redaction for outputs.

Month 1: Scale and monitor. Add caching, set up dashboards for key metrics, and build alerting for anomalies. Run the pipeline against test datasets to measure accuracy.

You can try the detection capabilities right now in the Wardstone Playground to see how different inputs and outputs are classified. When you're ready to integrate, our OpenAI integration guide shows how to add moderation to your existing LLM calls in under 10 minutes. Check out our pricing page for details on API limits and plans.

Content moderation for AI isn't a solved problem. According to the Stanford AI Index Report, AI-related security incidents increased 56.4% year over year from 2023 to 2024, reaching 233 documented cases. The threat landscape evolves constantly, models change, and user expectations shift. But a well-designed pipeline gives you the foundation to adapt. Start simple, measure everything, and iterate.