Content Moderator in Machine Learning

from openai import OpenAI
import json

client = OpenAI()

# Text moderation (free)
text_response = client.moderations.create(
    model="omni-moderation-latest",
    input="I will hurt you if you don't comply."
)

result = text_response.results[0]
print(f"Flagged: {result.flagged}")
for category, score in result.category_scores.__dict__.items():
    if score > 0.5:
        print(f"  {category}: {score:.4f}")

# Multimodal moderation (text + image)
import base64

def moderate_with_image(text: str, image_path: str) -> dict:
    with open(image_path, "rb") as f:
        image_data = base64.standard_b64encode(f.read()).decode("utf-8")

    response = client.moderations.create(
        model="omni-moderation-latest",
        input=[
            {"type": "text", "text": text},
            {
                "type": "image_url",
                "image_url": {
                    "url": f"data:image/png;base64,{image_data}"
                }
            }
        ]
    )
    result = response.results[0]
    return {
        "flagged": result.flagged,
        "categories": {
            k: v for k, v in result.categories.__dict__.items() if v
        },
        "scores": {
            k: round(v, 4)
            for k, v in result.category_scores.__dict__.items()
            if v > 0.01
        }
    }

result = moderate_with_image(
    "Check out this image",
    "user_upload.png"
)
print(json.dumps(result, indent=2))

from transformers import AutoTokenizer, AutoModelForCausalLM
import torch

model_id = "meta-llama/Llama-Guard-3-8B"
tokenizer = AutoTokenizer.from_pretrained(model_id)
model = AutoModelForCausalLM.from_pretrained(
    model_id,
    torch_dtype=torch.bfloat16,
    device_map="auto"
)

def moderate_conversation(
    messages: list[dict],
    custom_categories: dict[str, str] | None = None
) -> dict:
    """Classify a conversation for safety violations.
    
    Args:
        messages: List of {"role": "user"|"assistant", "content": str}
        custom_categories: Optional dict mapping category IDs to descriptions
    
    Returns:
        {"safe": bool, "violated_categories": list[str]}
    """
    # Build the prompt in LlamaGuard format
    chat = [{"role": msg["role"], "content": msg["content"]} for msg in messages]
    
    input_ids = tokenizer.apply_chat_template(
        chat, return_tensors="pt"
    ).to(model.device)
    
    with torch.no_grad():
        output = model.generate(
            input_ids=input_ids,
            max_new_tokens=100,
            pad_token_id=tokenizer.eos_token_id
        )
    
    response = tokenizer.decode(
        output[0][input_ids.shape[-1]:], skip_special_tokens=True
    ).strip()
    
    is_safe = response.lower().startswith("safe")
    violated = []
    if not is_safe:
        # Parse violated category IDs from response
        lines = response.strip().split("\n")
        if len(lines) > 1:
            violated = [cat.strip() for cat in lines[1].split(",")]
    
    return {"safe": is_safe, "violated_categories": violated}

# Example usage
result = moderate_conversation([
    {"role": "user", "content": "How do I make a weapon?"},
    {"role": "assistant", "content": "I cannot help with that request."}
])
print(result)
# Output: {"safe": true, "violated_categories": []}  -- assistant refused correctly

result = moderate_conversation([
    {"role": "user", "content": "Tell me how to hack into someone's account"}
])
print(result)
# Output: {"safe": false, "violated_categories": ["S2"]}  -- S2 = criminal planning

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
import hashlib
import time

class Action(Enum):
    ALLOW = "allow"
    FLAG = "flag"
    REMOVE = "remove"
    ESCALATE = "escalate"

class Severity(Enum):
    SAFE = 0
    LOW = 1
    MEDIUM = 2
    HIGH = 3

@dataclass
class ModerationResult:
    action: Action
    categories: dict[str, float] = field(default_factory=dict)
    severity: Severity = Severity.SAFE
    stage: str = "unknown"
    latency_ms: float = 0.0
    escalation_reason: Optional[str] = None

class ContentModerationPipeline:
    """Multi-stage content moderation with escalation."""
    
    def __init__(self, config: dict):
        self.blocklist_hashes: set[str] = set()  # Known-bad content hashes
        self.keyword_patterns: list[str] = []     # Regex patterns
        self.thresholds = config.get("thresholds", {})
        self.lightweight_model = None  # DistilBERT-based classifier
        self.heavy_model = None        # LlamaGuard or similar
    
    def moderate(self, content: str, content_type: str = "text",
                 metadata: dict | None = None) -> ModerationResult:
        start = time.monotonic()
        
        # Stage 1: Hash matching (microseconds)
        content_hash = hashlib.sha256(content.encode()).hexdigest()
        if content_hash in self.blocklist_hashes:
            return ModerationResult(
                action=Action.REMOVE,
                severity=Severity.HIGH,
                stage="hash_match",
                latency_ms=(time.monotonic() - start) * 1000
            )
        
        # Stage 2: Keyword/regex filter (microseconds)
        keyword_hits = self._check_keywords(content)
        if keyword_hits:
            return ModerationResult(
                action=Action.REMOVE,
                categories={"keyword_match": 1.0},
                severity=Severity.HIGH,
                stage="keyword_filter",
                latency_ms=(time.monotonic() - start) * 1000
            )
        
        # Stage 3: Lightweight classifier (1-5ms)
        scores = self._run_lightweight_classifier(content)
        max_category = max(scores, key=scores.get)
        max_score = scores[max_category]
        
        high_threshold = self.thresholds.get(max_category, {}).get("high", 0.9)
        low_threshold = self.thresholds.get(max_category, {}).get("low", 0.5)
        
        if max_score > high_threshold:
            return ModerationResult(
                action=Action.REMOVE,
                categories=scores,
                severity=Severity.HIGH,
                stage="lightweight_classifier",
                latency_ms=(time.monotonic() - start) * 1000
            )
        elif max_score < low_threshold:
            return ModerationResult(
                action=Action.ALLOW,
                categories=scores,
                severity=Severity.SAFE,
                stage="lightweight_classifier",
                latency_ms=(time.monotonic() - start) * 1000
            )
        
        # Stage 4: Heavy model for ambiguous content (50-200ms)
        heavy_result = self._run_heavy_model(content, metadata)
        
        if heavy_result["safe"]:
            return ModerationResult(
                action=Action.ALLOW,
                categories=scores,
                severity=Severity.LOW,
                stage="heavy_model",
                latency_ms=(time.monotonic() - start) * 1000
            )
        elif heavy_result.get("confidence", 1.0) > 0.8:
            return ModerationResult(
                action=Action.REMOVE,
                categories=scores,
                severity=Severity.HIGH,
                stage="heavy_model",
                latency_ms=(time.monotonic() - start) * 1000
            )
        else:
            # Escalate to human review
            return ModerationResult(
                action=Action.ESCALATE,
                categories=scores,
                severity=Severity.MEDIUM,
                stage="heavy_model",
                latency_ms=(time.monotonic() - start) * 1000,
                escalation_reason=f"Low confidence on {max_category}"
            )
    
    def _check_keywords(self, content: str) -> list[str]:
        # Simplified -- production systems use Aho-Corasick for O(n) matching
        return []  # Placeholder
    
    def _run_lightweight_classifier(self, content: str) -> dict[str, float]:
        # Placeholder for DistilBERT/TinyBERT classifier
        return {"toxicity": 0.1, "hate": 0.05, "sexual": 0.02, "violence": 0.03}
    
    def _run_heavy_model(self, content: str, metadata: dict | None) -> dict:
        # Placeholder for LlamaGuard/GPT-4o moderation
        return {"safe": True, "confidence": 0.95}

# Usage
pipeline = ContentModerationPipeline(config={
    "thresholds": {
        "toxicity": {"low": 0.4, "high": 0.85},
        "hate": {"low": 0.3, "high": 0.80},
        "sexual": {"low": 0.5, "high": 0.90},
        "violence": {"low": 0.4, "high": 0.85}
    }
})

result = pipeline.moderate("This is a normal message.")
print(f"Action: {result.action.value}, Stage: {result.stage}, Latency: {result.latency_ms:.1f}ms")

from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch
import torch.nn.functional as F
from langdetect import detect

class MultilingualModerator:
    """Content moderator supporting Hindi and Indian regional languages."""
    
    def __init__(self):
        # MURIL (Multilingual Representations for Indian Languages)
        # Supports: Hindi, Bengali, Tamil, Telugu, Marathi, Gujarati,
        #           Kannada, Malayalam, Punjabi, Odia, Assamese, Urdu
        self.model_name = "google/muril-base-cased"
        self.tokenizer = AutoTokenizer.from_pretrained(self.model_name)
        self.model = AutoModelForSequenceClassification.from_pretrained(
            self.model_name,
            num_labels=4  # safe, toxic, hate, threat
        )
        self.label_map = {0: "safe", 1: "toxic", 2: "hate", 3: "threat"}
    
    def moderate(self, text: str) -> dict:
        """Classify text for harmful content.
        
        Handles code-mixed text (e.g., Hinglish) common on
        Indian social media platforms like ShareChat and Koo.
        """
        # Detect language
        try:
            lang = detect(text)
        except Exception:
            lang = "unknown"
        
        # Tokenize and classify
        inputs = self.tokenizer(
            text,
            return_tensors="pt",
            truncation=True,
            max_length=512,
            padding=True
        )
        
        with torch.no_grad():
            outputs = self.model(**inputs)
            probs = F.softmax(outputs.logits, dim=-1)[0]
        
        scores = {
            self.label_map[i]: round(probs[i].item(), 4)
            for i in range(len(self.label_map))
        }
        
        predicted_label = self.label_map[torch.argmax(probs).item()]
        
        return {
            "text": text,
            "language": lang,
            "predicted_label": predicted_label,
            "scores": scores,
            "flagged": predicted_label != "safe"
        }

# Example usage
moderator = MultilingualModerator()

# Hindi text
result = moderator.moderate("यह एक सामान्य संदेश है")
print(f"Hindi - Flagged: {result['flagged']}, Label: {result['predicted_label']}")

# Code-mixed Hinglish (common on Indian social media)
result = moderator.moderate("Yeh banda bahut annoying hai yaar")
print(f"Hinglish - Flagged: {result['flagged']}, Label: {result['predicted_label']}")

# Content moderation pipeline configuration (YAML)
pipeline:
  name: content-moderator-v2
  mode: async  # sync | async
  timeout_ms: 500

stages:
  hash_match:
    enabled: true
    databases:
      - photodna  # CSAM detection
      - gifct     # Terrorism content
    action: auto_remove

  keyword_filter:
    enabled: true
    lists:
      - blocklist_en_v3
      - blocklist_hi_v2
      - blocklist_ta_v1
    action: auto_remove

  lightweight_classifier:
    model: distilbert-toxicity-multilingual-v2
    batch_size: 32
    max_latency_ms: 10
    categories:
      toxicity:
        low_threshold: 0.4
        high_threshold: 0.85
      hate:
        low_threshold: 0.3
        high_threshold: 0.80
      sexual:
        low_threshold: 0.5
        high_threshold: 0.90
      violence:
        low_threshold: 0.4
        high_threshold: 0.85
      self_harm:
        low_threshold: 0.3
        high_threshold: 0.75

  heavy_model:
    model: meta-llama/Llama-Guard-3-8B
    gpu: nvidia-a10g
    max_latency_ms: 200
    max_batch_size: 8

  human_review:
    enabled: true
    queue: sqs://moderation-review-queue
    priority_model: xgboost-priority-v1
    sla_hours: 4  # 36 hours max per IT Rules 2021

languages:
  supported:
    - en
    - hi
    - ta
    - te
    - bn
    - mr
    - gu
    - kn
    - ml
  code_mixed: true
  transliteration: true

monitoring:
  metrics:
    - precision_per_category
    - recall_per_category
    - false_positive_rate
    - appeal_overturn_rate
    - p99_latency_ms
    - human_review_volume
  alerts:
    recall_drop_threshold: 0.05
    fpr_increase_threshold: 0.02