AI Content Moderation System: Python + LLM + OpenCV for Enterprise Content Safety

Why Content Moderation is AI's Killer App

ByteDance processes billions of pieces of content daily — every piece must be reviewed in milliseconds. This is AI's most hardcore production scenario — not "chatting," but "judging."

Content moderation isn't nice-to-have — it's a compliance red line. One missed review can lead to app removal.

---

Three-Layer Moderation Architecture

┌──────────────────────────────────────────────────┐
│              Content Moderation Architecture       │
├──────────────────────────────────────────────────┤
│  Layer 1: Text Moderation                         │
│  ├── Sensitive word detection (AC automaton)      │
│  ├── Semantic understanding (LLM classification)  │
│  └── Sentiment analysis                           │
├──────────────────────────────────────────────────┤
│  Layer 2: Image Moderation                        │
│  ├── OCR text extraction → Text moderation        │
│  ├── Object detection (violence/porn/politics)    │
│  └── Face detection (public figures/minors)       │
├──────────────────────────────────────────────────┤
│  Layer 3: Video Moderation                        │
│  ├── Keyframe extraction → Image moderation       │
│  ├── Audio to text → Text moderation              │
│  └── Behavior recognition                         │
└──────────────────────────────────────────────────┘

---

Text Moderation

Sensitive Word Detection (AC Automaton)

from pyahocorasick import Automaton

class SensitiveWordDetector:
    def __init__(self):
        self.automaton = Automaton()
        self._load_words()

    def _load_words(self):
        with open("sensitive_words.txt", "r") as f:
            for idx, word in enumerate(f):
                self.automaton.add_word(word.strip(), (idx, word.strip()))
        self.automaton.make_automaton()

    def detect(self, text: str) -> list:
        results = []
        for end_idx, (word_idx, word) in self.automaton.iter(text):
            start_idx = end_idx - len(word) + 1
            results.append({"word": word, "start": start_idx, "end": end_idx + 1})
        return results

LLM Semantic Moderation

from openai import OpenAI

class SemanticModerator:
    def __init__(self):
        self.client = OpenAI()
        self.system_prompt = """You are a content moderation expert. Determine if the content violates rules.
Output JSON: {"is_violation": true/false, "category": "...", "confidence": 0.0-1.0, "reason": "..."}"""

    def moderate(self, text: str) -> dict:
        response = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": text}
            ],
            response_format={"type": "json_object"},
            temperature=0
        )
        return json.loads(response.choices[0].message.content)

---

Image Moderation

import easyocr

class ImageModerator:
    def __init__(self):
        self.ocr_reader = easyocr.Reader(['ch_sim', 'en'])
        self.text_moderator = SemanticModerator()

    def moderate(self, image_path: str) -> dict:
        ocr_results = self.ocr_reader.readtext(image_path)
        text = " ".join([result[1] for result in ocr_results])

        if text.strip():
            text_result = self.text_moderator.moderate(text)
        else:
            text_result = {"is_violation": False}

        image_result = self._detect_objects(image_path)

        return {
            "ocr_text": text,
            "text_moderation": text_result,
            "image_moderation": image_result,
            "is_violation": text_result["is_violation"] or image_result["is_violation"]
        }

---

Multi-Level Moderation Pipeline

class ModerationPipeline:
    def __init__(self):
        self.sensitive_detector = SensitiveWordDetector()
        self.semantic_moderator = SemanticModerator()
        self.image_moderator = ImageModerator()

    def moderate(self, content: dict) -> dict:
        # Level 1: Fast rule filtering (<10ms)
        sensitive_hits = self.sensitive_detector.detect(content.get("text", ""))
        if sensitive_hits:
            return {"is_violation": True, "level": "high", "details": sensitive_hits}

        # Level 2: AI semantic moderation (<500ms)
        results = {}
        if content["type"] == "text":
            results["semantic"] = self.semantic_moderator.moderate(content["text"])
        elif content["type"] == "image":
            results["image"] = self.image_moderator.moderate(content["url"])

        # Level 3: Human review (low confidence triggers)
        for key, result in results.items():
            if result.get("confidence", 1.0) < 0.85:
                results[key]["need_human_review"] = True

        is_violation = any(r.get("is_violation", False) for r in results.values())
        return {"is_violation": is_violation, "results": results}

---

Summary

Enterprise content moderation system core design:

1. Multi-level filtering: Rules (fast) → AI (accurate) → Human review (safety net)
2. Multi-modal coverage: Text + Image + Video + Audio
3. Python ecosystem: OpenCV, easyocr, whisper out of the box
4. LLM enhancement: From "keyword matching" to "semantic understanding"

Content moderation is AI's most hardcore production scenario — not "can chat," but "can judge."