How to Build an LLM Content Moderation Service

The architecture

The service is a thin wrapper: FastAPI receives POST /api/moderate, calls the LLM with a classifying system prompt, parses the JSON it returns, and sends back the label plus a reasoning string. A second endpoint, POST /api/moderate/batch, runs the same logic over a list of items.

The model does the judgment; your job is a clean request/response contract and error handling at every boundary.

The system prompt

The prompt is the product. It must tell the model exactly what labels to use and what format to return - otherwise you get prose instead of JSON:

SYSTEM_PROMPT = (
    "You are a content moderator. Classify the user's content as one of: "
    "safe, warning, or blocked. Respond with ONLY a JSON object of the form "
    '{"classification": "safe|warning|blocked", "reasoning": "one short sentence"}. '
    "No markdown, no prose, JSON only."
)

Calling the LLM and parsing the response

Use httpx to POST to the OpenAI-compatible endpoint, pull the model's reply, and extract the JSON. Always strip markdown fences - models ignore "JSON only" sometimes:

import re, json, httpx

def classify(content: str) -> dict:
    try:
        with httpx.Client(timeout=60.0) as client:
            r = client.post(
                f"{LLM_API_URL}/v1/chat/completions",
                json={
                    "model": "mock-llm",
                    "messages": [
                        {"role": "system", "content": SYSTEM_PROMPT},
                        {"role": "user",   "content": content},
                    ],
                    "temperature": 0,
                    "max_tokens": 100,
                },
            )
            r.raise_for_status()
            raw = r.json()["choices"][0]["message"]["content"]
    except Exception as e:
        return {"content": content, "classification": "safe",
                "reasoning": f"moderator unavailable: {e}"}

    # Strip fences, extract the first {...} block
    m = re.search(r"\{.*\}", raw, re.DOTALL)
    parsed = json.loads(m.group(0)) if m else {}
    label = parsed.get("classification", "safe").lower()
    if label not in {"safe", "warning", "blocked"}:
        label = "safe"
    return {"content": content, "classification": label,
            "reasoning": parsed.get("reasoning", raw.strip())}

The endpoints

Wire both endpoints to the same classify helper:

from fastapi import FastAPI
from pydantic import BaseModel
from typing import List

app = FastAPI()

class ModerationRequest(BaseModel):
    content: str

class BatchModerationRequest(BaseModel):
    items: List[str]

@app.post("/api/moderate")
def moderate_content(req: ModerationRequest):
    return classify(req.content)

@app.post("/api/moderate/batch")
def moderate_batch(req: BatchModerationRequest):
    return {"results": [classify(item) for item in req.items]}

Hardening the output

LLMs are inconsistent. After parsing, always validate the label against your allowed set and fall back to a default rather than passing unknown strings downstream. For production, add a cheap regex or keyword pre-filter before the LLM call - it handles obvious cases instantly and keeps inference costs down.