Output Filtering & Content Moderation for LLMs

┌────────────────────────────────────────────┐
│ LLM-Output (Raw)                           │
│ "The password for admin is [LEAKED_DATA]"  │
└─────────────────────┬──────────────────────┘
                      ↓
┌────────────────────────────────────────────┐
│ LAYER 1: Content Classification            │
│ - Toxicity Score                           │
│ - Harm Category Detection                  │
│ - PII Detection                            │
└─────────────────────┬──────────────────────┘
                      ↓
┌────────────────────────────────────────────┐
│ LAYER 2: Policy Check                      │
│ - Does output violate policies?            │
│ - Severity assessment                      │
│ - Confidence scoring                       │
└─────────────────────┬──────────────────────┘
                      ↓
┌────────────────────────────────────────────┐
│ LAYER 3: Action                            │
│ - SAFE: Return to user                     │
│ - CAUTION: Log + Allow                     │
│ - DANGEROUS: Block or Redact               │
│ - UNKNOWN: Human review                    │
└────────────────────────────────────────────┘

import re
from typing import List, Dict
from dataclasses import dataclass

@dataclass
class PII:
    type: str  # "email", "ssn", "phone", "credit_card", etc.
    value: str
    position: int
    confidence: float

class PIIDetector:
    """Detects personally identifiable information"""

    PATTERNS = {
        "email": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
        "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(?:\+1[-.\s]?)?\(?([0-9]{3})\)?[-.\s]?([0-9]{3})[-.\s]?([0-9]{4})\b',
        "ip_address": r'\b(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\b',
        "api_key": r'\b(?:sk|pk)-[A-Za-z0-9]{20,}\b',
    }

    def __init__(self):
        self.compiled_patterns = {
            key: re.compile(pattern, re.IGNORECASE)
            for key, pattern in self.PATTERNS.items()
        }

    def detect(self, text: str) -> List[PII]:
        """
        Detects PII in text

        Returns:
            List of PII objects
        """
        pii_list = []

        for pii_type, pattern in self.compiled_patterns.items():
            for match in pattern.finditer(text):
                pii = PII(
                    type=pii_type,
                    value=match.group(),
                    position=match.start(),
                    confidence=0.95
                )
                pii_list.append(pii)

        return pii_list

    def redact(self, text: str, replacement: str = "[REDACTED]") -> str:
        """Redacts PII in text"""
        redacted = text
        pii_list = self.detect(text)

        # Sort by position (reverse) so indices don't shift
        for pii in sorted(pii_list, key=lambda p: p.position, reverse=True):
            redacted = redacted[:pii.position] + replacement + redacted[pii.position + len(pii.value):]

        return redacted

# Usage:
detector = PIIDetector()

text = "Contact me at [email protected] or (555) 123-4567. SSN: 123-45-6789"
pii_found = detector.detect(text)

print(f"Found {len(pii_found)} PII:")
for p in pii_found:
    print(f"  {p.type}: {p.value}")

print(f"\nRedacted: {detector.redact(text)}")

from dataclasses import dataclass
from enum import Enum
from typing import Dict, List

class ToxicityLevel(Enum):
    SAFE = "safe"
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

@dataclass
class ToxicityResult:
    level: ToxicityLevel
    score: float  # 0.0 - 1.0
    toxic_spans: List[str]
    categories: Dict[str, float]

class ToxicityClassifier:
    """Classifies text for toxicity"""

    TOXIC_KEYWORDS = {
        "insult": ["stupid", "idiot", "dumb", "moron"],
        "threat": ["kill", "die", "destroy", "harm"],
        "hate": ["hate", "despise", "loathe"],
        "abuse": ["abuse", "attack", "assault"],
    }

    SEVERITY_WEIGHTS = {
        "insult": 0.3,
        "threat": 0.9,
        "hate": 0.8,
        "abuse": 0.9,
    }

    def classify(self, text: str) -> ToxicityResult:
        """Classifies toxicity level"""
        lower_text = text.lower()
        categories = {}
        toxic_spans = []

        for category, keywords in self.TOXIC_KEYWORDS.items():
            found = []
            for keyword in keywords:
                if keyword in lower_text:
                    found.append(keyword)
                    toxic_spans.append(keyword)

            if found:
                categories[category] = len(found) * self.SEVERITY_WEIGHTS[category]
            else:
                categories[category] = 0.0

        # Calculate total score
        total_score = max(categories.values()) if categories else 0.0

        # Determine level
        if total_score == 0.0:
            level = ToxicityLevel.SAFE
        elif total_score < 0.3:
            level = ToxicityLevel.LOW
        elif total_score < 0.6:
            level = ToxicityLevel.MEDIUM
        elif total_score < 0.85:
            level = ToxicityLevel.HIGH
        else:
            level = ToxicityLevel.CRITICAL

        return ToxicityResult(
            level=level,
            score=total_score,
            toxic_spans=list(set(toxic_spans)),
            categories=categories
        )

# Usage:
classifier = ToxicityClassifier()

test_texts = [
    "Hello, how are you?",
    "You are stupid and dumb",
    "I will kill you",
]

for text in test_texts:
    result = classifier.classify(text)
    print(f"Text: {text}")
    print(f"  Level: {result.level.value}")
    print(f"  Score: {result.score:.2f}\n")

from typing import Optional

class CompleteOutputFilter:
    """Complete output filtering pipeline"""

    def __init__(self):
        self.pii_detector = PIIDetector()
        self.toxicity_classifier = ToxicityClassifier()

    def filter(self, output: str, policy: dict = None) -> dict:
        """
        Filters output through all layers

        Args:
            output: LLM output
            policy: Policy rules (optional)

        Returns:
            {
                "safe": bool,
                "severity": "safe" | "caution" | "dangerous",
                "actions": ["redact", "block", "log"],
                "filtered_output": str,
                "details": {...}
            }
        """

        if policy is None:
            policy = {
                "allow_pii": False,
                "max_toxicity": 0.5,
            }

        details = {}
        actions = []
        filtered_output = output

        # LAYER 1: PII Detection
        pii_list = self.pii_detector.detect(output)
        details["pii_found"] = len(pii_list)

        if pii_list and not policy["allow_pii"]:
            actions.append("redact")
            filtered_output = self.pii_detector.redact(filtered_output)
            details["pii_redacted"] = True

        # LAYER 2: Toxicity Check
        toxicity = self.toxicity_classifier.classify(output)
        details["toxicity"] = {
            "level": toxicity.level.value,
            "score": toxicity.score,
        }

        if toxicity.score > policy["max_toxicity"]:
            actions.append("block")
            details["toxicity_blocked"] = True

        # LAYER 3: Determine Severity
        if "block" in actions:
            severity = "dangerous"
            filtered_output = "[Output blocked due to policy violation]"
        elif "redact" in actions:
            severity = "caution"
        else:
            severity = "safe"

        is_safe = severity == "safe"

        return {
            "safe": is_safe,
            "severity": severity,
            "actions": actions,
            "filtered_output": filtered_output,
            "original_output": output,
            "details": details
        }

# Usage:
filter = CompleteOutputFilter()

outputs = [
    "The capital of France is Paris",
    "Contact [email protected] or call (555) 123-4567",
    "I hate you, you're stupid and should die",
]

for output in outputs:
    result = filter.filter(output)
    print(f"Original: {output}")
    print(f"Safe: {result['safe']}")
    print(f"Severity: {result['severity']}")
    print(f"Filtered: {result['filtered_output']}")
    print()