← Articles

Building a content moderation pipeline with LLMs

By Seun Adetoye · 16 July 20260 views
#llm#content-moderation#pipeline#backend#safety

Content moderation at scale is one of the most operationally demanding problems a platform can face. Rule-based keyword filters catch obvious violations but miss context. Manual review queues are expensive and slow. LLMs occupy the middle ground: they understand nuance, follow complex policies, and can be updated by changing a prompt rather than retraining a model. This article walks through building a complete moderation pipeline in production, from first classification call to audit logging.

Framing the problem correctly

Before writing any code, you need to decide what the pipeline is actually responsible for. Content moderation is not a binary spam/not-spam classifier. A mature system typically handles several distinct categories simultaneously:

  • Toxicity and hate speech — slurs, dehumanisation, targeted harassment
  • Sexual content — explicit material, content involving minors
  • Violence and threats — credible threats, graphic descriptions, glorification
  • Spam and manipulation — coordinated inauthentic behaviour, scam links, fake reviews
  • Misinformation — health disinformation, election interference claims
  • Self-harm — content that may encourage or instruct harmful behaviour

Each category may have different thresholds (a news platform tolerates graphic violence in journalism; a children's app does not) and different escalation paths. Map these before designing your schema.

The pipeline architecture has three layers:

  1. Pre-filter — cheap heuristics that catch obvious violations without an LLM call (regex, blocklists, image hashes)
  2. LLM classifier — structured classification against your policy
  3. Human review queue — uncertain cases forwarded to moderators

Getting the routing logic right is more important than any single model choice.

Structured classification with tool use

The most reliable way to get consistent output from an LLM classifier is to define an explicit tool call schema. This avoids free-text parsing and gives you typed confidence scores.

import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic();

const moderationTool: Anthropic.Tool = {
  name: "classify_content",
  description: "Classify user-generated content against the platform policy",
  input_schema: {
    type: "object" as const,
    properties: {
      categories: {
        type: "object",
        properties: {
          toxicity: { type: "number", minimum: 0, maximum: 1 },
          sexual_content: { type: "number", minimum: 0, maximum: 1 },
          violence: { type: "number", minimum: 0, maximum: 1 },
          spam: { type: "number", minimum: 0, maximum: 1 },
          self_harm: { type: "number", minimum: 0, maximum: 1 },
        },
        required: ["toxicity", "sexual_content", "violence", "spam", "self_harm"],
      },
      overall_verdict: {
        type: "string",
        enum: ["allow", "flag_for_review", "reject"],
      },
      policy_violations: {
        type: "array",
        items: { type: "string" },
        description: "Specific policy sections violated, empty if none",
      },
      reasoning: {
        type: "string",
        description: "One-sentence explanation of the verdict for the audit log",
      },
    },
    required: ["categories", "overall_verdict", "policy_violations", "reasoning"],
  },
};

interface ModerationResult {
  categories: Record<string, number>;
  overall_verdict: "allow" | "flag_for_review" | "reject";
  policy_violations: string[];
  reasoning: string;
}

async function classifyContent(
  content: string,
  systemPrompt: string
): Promise<ModerationResult> {
  const response = await client.messages.create({
    model: "claude-sonnet-4-5",
    max_tokens: 512,
    system: systemPrompt,
    tools: [moderationTool],
    tool_choice: { type: "any" },
    messages: [
      {
        role: "user",
        content: `Classify the following user-generated content:\n\n<content>${content}</content>`,
      },
    ],
  });

  const toolUse = response.content.find((block) => block.type === "tool_use");
  if (!toolUse || toolUse.type !== "tool_use") {
    throw new Error("Model did not return a tool call");
  }

  return toolUse.input as ModerationResult;
}

The system prompt is where your actual policy lives. Keep it versioned in source control alongside the code:

const MODERATION_SYSTEM_PROMPT = `
You are a content moderation classifier for Acme Platform.

## Platform context
Acme is a professional networking site for adult users. Content must be appropriate
for a business environment.

## Policy summary
- REJECT: hate speech targeting protected characteristics, explicit sexual content,
  credible threats of violence, spam/phishing links, content encouraging self-harm
- FLAG FOR REVIEW: borderline cases, satire involving sensitive topics, news content
  referencing violence, content with ambiguous intent
- ALLOW: professional discussion, criticism of products/services, strong but non-hateful
  language, clearly humorous content

## Confidence scores
Assign a float from 0.0 to 1.0 for each category representing how strongly the
content exhibits that characteristic. Use 0.0 for clear absence, 1.0 for unambiguous
presence.

## Important
- Evaluate intent and context, not surface-level keywords
- A sentence containing the word "kill" in "I could kill for a coffee right now" is not violence
- Apply the policy consistently regardless of the target group
`;

Building the routing layer

Classification is one step. Routing is where business logic lives. You need to combine category scores, apply platform-specific thresholds, and decide what happens next.

interface RoutingConfig {
  autoRejectThresholds: Record<string, number>;
  autoAllowThresholds: Record<string, number>;
  humanReviewQueue: string;
}

const DEFAULT_CONFIG: RoutingConfig = {
  autoRejectThresholds: {
    toxicity: 0.85,
    sexual_content: 0.7,
    violence: 0.8,
    spam: 0.9,
    self_harm: 0.75,
  },
  autoAllowThresholds: {
    toxicity: 0.15,
    sexual_content: 0.05,
    violence: 0.1,
    spam: 0.1,
    self_harm: 0.05,
  },
  humanReviewQueue: "moderation-review",
};

type RouteDecision = {
  action: "allow" | "reject" | "human_review";
  reason: string;
  triggered_by?: string;
};

function routeDecision(
  result: ModerationResult,
  config: RoutingConfig = DEFAULT_CONFIG
): RouteDecision {
  // Hard reject: any category exceeds auto-reject threshold
  for (const [category, score] of Object.entries(result.categories)) {
    if (score >= (config.autoRejectThresholds[category] ?? 1)) {
      return {
        action: "reject",
        reason: `Auto-rejected: ${category} score ${score.toFixed(2)} exceeds threshold`,
        triggered_by: category,
      };
    }
  }

  // Model explicitly flagged for review
  if (result.overall_verdict === "flag_for_review") {
    return {
      action: "human_review",
      reason: `Model flagged for review: ${result.reasoning}`,
    };
  }

  // All categories below auto-allow threshold
  const allBelowAllowThreshold = Object.entries(result.categories).every(
    ([category, score]) => score <= (config.autoAllowThresholds[category] ?? 0)
  );

  if (allBelowAllowThreshold && result.overall_verdict === "allow") {
    return { action: "allow", reason: "All scores below allow threshold" };
  }

  // Uncertain: send to humans
  return {
    action: "human_review",
    reason: `Uncertain: scores outside threshold bands, verdict was ${result.overall_verdict}`,
  };
}

Pre-filtering and cost control

LLM calls are expensive. A realistic platform with millions of daily posts cannot run every piece of content through a frontier model. A layered pre-filter slashes costs without reducing safety:

interface PreFilterResult {
  skip_llm: boolean;
  auto_action?: "allow" | "reject";
  reason: string;
}

// Extremely fast pre-filter — runs before any LLM call
function preFilter(content: string): PreFilterResult {
  // Empty or trivially short content
  if (!content || content.trim().length < 3) {
    return { skip_llm: true, auto_action: "allow", reason: "Too short to classify" };
  }

  // Hard blocklist (maintained separately, loaded at startup)
  const blocklist = getBlocklist(); // your implementation
  const lowerContent = content.toLowerCase();
  for (const term of blocklist) {
    if (lowerContent.includes(term)) {
      return { skip_llm: true, auto_action: "reject", reason: `Blocklist match: ${term}` };
    }
  }

  // Known-safe patterns (verified user posts a simple link, etc.)
  const urlOnlyPattern = /^https?:\/\/[^\s]+$/;
  if (urlOnlyPattern.test(content.trim())) {
    // URL-only: run through URL reputation checker, not LLM
    return { skip_llm: true, reason: "URL-only, route to URL checker" };
  }

  return { skip_llm: false, reason: "Requires LLM classification" };
}

In practice, a well-tuned pre-filter catches 60–80% of clearly safe content and 15–25% of clearly violating content, leaving a fraction of borderline posts for the LLM.

Async pipeline with queues

Synchronous moderation blocks content before display. Asynchronous moderation allows immediate posting with retroactive removal. The right choice depends on your product:

  • Synchronous — higher latency, no violations visible even briefly, appropriate for payment messages, healthcare content, content involving minors
  • Asynchronous — low latency, violations briefly visible, appropriate for social feeds, comments, reviews

For asynchronous processing, use a queue:

import { Queue, Worker } from "bullmq";
import Redis from "ioredis";

const connection = new Redis(process.env.REDIS_URL!);

interface ModerationJob {
  contentId: string;
  content: string;
  authorId: string;
  contentType: "post" | "comment" | "review" | "profile_bio";
  submittedAt: string;
}

// Producer — called when user submits content
const moderationQueue = new Queue<ModerationJob>("content-moderation", {
  connection,
  defaultJobOptions: {
    attempts: 3,
    backoff: { type: "exponential", delay: 2000 },
    removeOnComplete: { count: 1000 },
    removeOnFail: { count: 5000 },
  },
});

export async function enqueueForModeration(job: ModerationJob) {
  await moderationQueue.add("classify", job, {
    priority: job.contentType === "profile_bio" ? 10 : 1, // bios get lower priority
  });
}

// Consumer — runs on separate worker processes
const worker = new Worker<ModerationJob>(
  "content-moderation",
  async (job) => {
    const preFilterResult = preFilter(job.data.content);
    if (preFilterResult.skip_llm && preFilterResult.auto_action) {
      await applyModerationDecision(job.data.contentId, {
        action: preFilterResult.auto_action,
        reason: preFilterResult.reason,
        source: "pre_filter",
      });
      return;
    }

    const classification = await classifyContent(
      job.data.content,
      MODERATION_SYSTEM_PROMPT
    );
    const decision = routeDecision(classification);

    await applyModerationDecision(job.data.contentId, {
      action: decision.action,
      reason: decision.reason,
      source: "llm",
      llmResult: classification,
    });

    if (decision.action === "human_review") {
      await enqueueHumanReview(job.data, classification, decision);
    }
  },
  { connection, concurrency: 20 }
);

Audit logging and explainability

Every moderation decision must be logged with enough context to reconstruct why it was made. This matters for appeals, regulatory compliance, and debugging systematic errors.

import { Firestore, FieldValue } from "firebase-admin/firestore";

const db = new Firestore();

interface ModerationAuditEntry {
  contentId: string;
  authorId: string;
  content: string; // store only if legally required; hash otherwise
  contentHash: string;
  action: "allow" | "reject" | "human_review";
  source: "pre_filter" | "llm" | "human";
  reason: string;
  llmResult?: ModerationResult;
  policyVersion: string;
  modelVersion: string;
  decidedAt: FirebaseFirestore.FieldValue;
  reviewerId?: string; // set when human overrides
  appealStatus?: "pending" | "upheld" | "overturned";
}

async function logModerationDecision(entry: Omit<ModerationAuditEntry, "decidedAt">) {
  const docRef = db.collection("moderation_log").doc();
  await docRef.set({
    ...entry,
    decidedAt: FieldValue.serverTimestamp(),
  });

  // Separate collection for fast appeals lookup
  if (entry.action === "reject") {
    await db.collection("moderation_rejections").doc(entry.contentId).set({
      logRef: docRef.id,
      authorId: entry.authorId,
      reason: entry.reason,
      appealStatus: "none",
      decidedAt: FieldValue.serverTimestamp(),
    });
  }
}

Track policy version and model version separately. When you update either, you can compare decision distributions before and after to detect regressions.

Handling human review queues

The human review queue is not an afterthought — it is where your policy is actually refined. Build it properly from day one:

interface ReviewQueueItem {
  contentId: string;
  content: string;
  llmClassification: ModerationResult;
  llmDecision: RouteDecision;
  priority: number; // higher = review sooner
  createdAt: string;
  assignedTo?: string;
  reviewedAt?: string;
  reviewDecision?: "allow" | "reject";
  reviewNotes?: string;
}

// Priority scoring: higher scores get reviewed first
function calculateReviewPriority(
  classification: ModerationResult,
  contentType: string
): number {
  let priority = 0;

  // High self-harm score → urgent
  if (classification.categories.self_harm > 0.5) priority += 100;

  // Violence with credible threat language
  if (classification.categories.violence > 0.6) priority += 50;

  // Verified users get faster review (reputational risk)
  if (contentType === "verified_user_post") priority += 30;

  // Recency — older items lose urgency
  return priority;
}

// When a moderator submits their decision, feed it back as training signal
async function submitHumanReview(
  itemId: string,
  reviewerId: string,
  decision: "allow" | "reject",
  notes: string
) {
  const item = await getReviewQueueItem(itemId);
  const wasLlmCorrect = item.llmDecision.action === decision ||
    (item.llmDecision.action === "human_review"); // escalation = correct

  await db.collection("moderation_reviews").doc(itemId).update({
    assignedTo: reviewerId,
    reviewedAt: FieldValue.serverTimestamp(),
    reviewDecision: decision,
    reviewNotes: notes,
    llmAgreement: wasLlmCorrect,
  });

  // Track disagreement rate by category for prompt tuning
  if (!wasLlmCorrect) {
    await logLlmDisagreement(item, decision, reviewerId);
  }
}

Review queue disagreement rate is your primary signal for prompt quality. If moderators are overturning LLM decisions at more than 15%, your prompt is miscalibrated — either too aggressive or too permissive in specific categories.

Measuring pipeline health

A moderation pipeline without metrics is flying blind. Track these at minimum:

interface PipelineMetrics {
  // Volume
  contentSubmittedPerMinute: number;
  preFilteredRate: number; // % caught by pre-filter
  llmClassifiedRate: number;
  humanReviewRate: number;

  // Decisions
  autoAllowRate: number;
  autoRejectRate: number;
  escalationRate: number; // % sent to human review

  // Quality (requires human labels)
  llmPrecision: number; // of LLM rejects, % that humans agree are violations
  llmRecall: number; // of true violations, % that LLM caught
  humanOverrideRate: number; // % of LLM decisions humans change

  // Latency
  p50LatencyMs: number;
  p99LatencyMs: number;

  // Cost
  llmTokensPerContent: number;
  costPerThousandItems: number;
}

Set up alerts on escalation rate (sudden spike = prompt broke), human override rate (steady increase = policy drift), and p99 latency (indicates queue backup).

Prompt versioning and regression testing

Every prompt change is a code change. Treat it the same way:

// moderation-prompt-v2.ts — check into source control
export const PROMPT_VERSION = "2.1.0";
export const MODERATION_SYSTEM_PROMPT = `...`;

// test/moderation.test.ts
import { describe, it, expect } from "vitest";
import { classifyContent, routeDecision } from "../src/moderation";
import { MODERATION_SYSTEM_PROMPT } from "../src/moderation-prompt-v2";

const testCases = [
  {
    name: "obvious spam",
    content: "CLICK HERE FOR FREE IPHONE https://scam.example.com",
    expectedAction: "reject",
    expectedCategory: "spam",
  },
  {
    name: "idiomatic violence",
    content: "I could absolutely kill for a coffee right now",
    expectedAction: "allow",
  },
  {
    name: "borderline political satire",
    content: "That politician is a clown who should be thrown out",
    expectedAction: "allow",
  },
  {
    name: "targeted harassment",
    content: "Everyone should report @user and get them fired",
    expectedAction: "flag_for_review",
  },
];

describe("content moderation classifier", () => {
  for (const tc of testCases) {
    it(tc.name, async () => {
      const result = await classifyContent(tc.content, MODERATION_SYSTEM_PROMPT);
      const decision = routeDecision(result);
      expect(decision.action).toBe(tc.expectedAction);
    });
  }
});

Run this test suite before deploying any prompt change. Add every moderator-corrected case to it over time. The suite becomes a regression oracle for your policy.

Scaling considerations

At 10,000 requests per minute, a few architectural decisions become critical:

Model selection by risk tier. Not every piece of content needs your best model. Route clearly structured posts (job listings, product reviews with structured fields) to a smaller, faster model. Reserve the frontier model for unstructured text and appeal re-reviews.

Batch API for non-urgent content. Most platforms can tolerate 30-minute moderation latency for historical content re-review, user-flagged content, and profile audit passes. Use the Anthropic Batch API for these workloads at 50% cost reduction.

Cache identical content. Profile bios, standard templates, and repeated spam campaigns often produce identical content. A content hash → decision cache with a short TTL (30 minutes) eliminates redundant LLM calls. Invalidate on policy version change.

Idempotent workers. Your queue worker will process the same job multiple times during failures. Make applyModerationDecision idempotent by checking existing decision state before writing.

A well-built content moderation pipeline reduces moderator workload by 70–90%, catches violations in seconds rather than hours, and produces an audit trail that satisfies regulatory requirements. The investment in proper schema, routing logic, and metrics pays off quickly as the platform scales.

Comments

No comments yet. Be the first!

Sign in to leave a comment.

Building a content moderation pipeline with LLMs — ANN Tech