Output Filtering & Content Moderation für 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:
    """Erkennt persönlich identifizierende Informationen"""

    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',
        "password_pattern": r'(?:password|passwd|pwd)\s*[:=]\s*[^\s]+',
        "credit_card_visa": r'\b4[0-9]{12}(?:[0-9]{3})?\b',
        "credit_card_amex": r'\b3[47][0-9]{13}\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]:
        """
        Erkennt PII in Text

        Returns:
            Liste von PII-Objekten
        """
        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  # High confidence für Regex-Matches
                )
                pii_list.append(pii)

        return pii_list

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

        # Sortiere nach Position (rückwärts), damit Indices nicht verschoben werden
        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

    def mask(self, text: str) -> str:
        """
        Maskiert PII (z.B. partial masking)
        """
        masked = text
        pii_list = self.detect(text)

        for pii in sorted(pii_list, key=lambda p: p.position, reverse=True):
            if pii.type == "email":
                # Zeige nur erste 2 Zeichen
                mask_value = pii.value[:2] + "***@***"
            elif pii.type == "ssn":
                # Zeige nur letzte 4 Zeichen
                mask_value = "***-**-" + pii.value[-4:]
            elif pii.type == "credit_card":
                # Zeige nur letzte 4 Zeichen
                mask_value = "****-****-****-" + pii.value[-4:]
            elif pii.type == "phone":
                # Zeige nur letzte 4 Ziffern
                mask_value = "***-***-" + pii.value[-4:]
            else:
                mask_value = "[" + pii.type.upper() + "]"

            masked = masked[:pii.position] + mask_value + masked[pii.position + len(pii.value):]

        return masked

# Verwendung:
detector = PIIDetector()

text = "Contact me at [email protected] or call (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)}")
print(f"Masked: {detector.mask(text)}")

from dataclasses import dataclass
from enum import Enum

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]  # z.B. {"insult": 0.8, "threat": 0.3}

class ToxicityClassifier:
    """Klassifiziert Text auf Toxizität"""

    # Vereinfacht: Keyword-basiert (in Produktion: LLM-basiert oder ML-Modell)
    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:
        """
        Klassifiziert Toxizität

        Returns:
            ToxicityResult mit Score und Kategorien
        """
        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

        # Berechne Gesamt-Score
        total_score = max(categories.values()) if categories else 0.0

        # Bestimme 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
        )

# Verwendung:
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}")
    print()

# Installation
pip install nemo-guardrails

# Optional: mit LLM-Support
pip install nemo-guardrails[openai]

# Verify
python -c "from nemo.guardrails import RailsConfig; print('OK')"

# guardrails.yaml
colang_version: "0.1"

# Definiere verfügbare Actions
actions:
  - stop_chain
  - log
  - refrain

# Definiere Message Flows
flows:
  user message:
    - "user said something"
    - run: check_pii
    - if: $pii_detected
      then: "pii violation"
    - else: "continue"

  pii violation:
    message: "I detected PII in your message. Please remove it and try again."

  continue:
    - run: process_normally

check_pii:
  if: user_input contains any ["email", "phone", "ssn"]
    set: $pii_detected = True
  else:
    set: $pii_detected = False

  # Optional: Log the violation
  log: "PII detection triggered"

from nemo.guardrails import RailsConfig, LLMRails

class NeMoGuardrailsFilter:
    """LLM-Output-Filtering mit NeMo Guardrails"""

    def __init__(self, config_yaml: str):
        """
        Args:
            config_yaml: Pfad zur Guardrails-Config
        """
        self.rails_config = RailsConfig.from_yaml(config_yaml)
        self.rails = LLMRails(config=self.rails_config)

    def process_output(self, llm_response: str, user_query: str) -> str:
        """
        Verarbeitet LLM-Output durch Guardrails

        Returns:
            Gefilterte Ausgabe
        """
        # Baue Messages
        messages = [
            {"role": "system", "content": "You are a safety filter"},
            {"role": "user", "content": user_query},
            {"role": "assistant", "content": llm_response}
        ]

        # Führe durch Guardrails aus
        filtered_response = self.rails.generate(messages=messages)
        return filtered_response

# Verwendung:
filter = NeMoGuardrailsFilter(config_yaml="guardrails.yaml")

query = "What's my email?"
llm_response = "Your email is [email protected]"

filtered = filter.process_output(llm_response, query)
print(filtered)  # Kann modifiziert sein wenn PII detected

pip install guardrails-ai

# Mit Validators
pip install guardrails-ai[validators]

from guardrails import Guard, Validator
from pydantic import BaseModel, Field

# Definiere Output-Schema
class SafeResponse(BaseModel):
    message: str = Field(description="Safe response")
    contains_pii: bool = Field(description="Does output contain PII?")
    toxicity_level: str = Field(description="Toxicity level: safe/medium/high")

# Erstelle Guard mit Validatoren
guard = Guard.from_pydantic(
    output_class=SafeResponse,
    validators=[
        # Validator gegen PII
        Validator(
            name="no_pii",
            on="output.message",
            register=True
        ),
        # Validator gegen Toxizität
        Validator(
            name="no_toxic_language",
            on="output.message",
            register=True
        ),
    ]
)

# Verwende Guard
response = {
    "message": "The weather is sunny",
    "contains_pii": False,
    "toxicity_level": "safe"
}

try:
    validated = guard.validate(response)
    print(f"Validation passed: {validated}")
except Exception as e:
    print(f"Validation failed: {e}")

from guardrails import Validator, register_validator

@register_validator(name="no_emails", data_type="string")
def no_emails(value: str) -> str:
    """
    Validator der sicherstellt dass keine Emails in Output sind
    """
    import re
    email_pattern = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'

    if re.search(email_pattern, value):
        raise ValueError(f"Output contains email addresses")

    return value

@register_validator(name="max_length", data_type="string")
def max_length_validator(value: str, max_len: int = 1000) -> str:
    """Validator für maximale Länge"""
    if len(value) > max_len:
        raise ValueError(f"Output exceeds max length {max_len}")
    return value

# Verwende Custom Validators
guard = Guard.from_pydantic(
    output_class=SafeResponse,
    validators=[
        no_emails,
        max_length_validator
    ]
)

pip install llm-guard

# Mit allen Scanners
pip install llm-guard[all]

from llm_guard import scan_output
from llm_guard.output_scanners import (
    PIIScanner,
    ToxicityScanner,
    PromptInjectionScanner,
    BanTopicsScanner
)

class LLMGuardFilter:
    """Output-Filter mit LLM Guard"""

    def __init__(self):
        self.scanners = [
            PIIScanner(),  # Detektiert PII
            ToxicityScanner(threshold=0.5),  # Toxicity >= 0.5 blockiert
            PromptInjectionScanner(),  # Detektiert Injection-Versuche
            BanTopicsScanner(topics=["weapons", "drugs", "violence"]),
        ]

    def filter_output(self, output: str) -> dict:
        """
        Filtert Output durch alle Scanner

        Returns:
            {
                "safe": bool,
                "results": {scanner_name: result}
            }
        """
        is_safe = True
        results = {}

        for scanner in self.scanners:
            result = scan_output(output, scanner)
            results[scanner.__class__.__name__] = result

            if not result["safe"]:
                is_safe = False

        return {
            "safe": is_safe,
            "results": results
        }

# Verwendung:
filter = LLMGuardFilter()

test_outputs = [
    "The weather is nice today",
    "Your email is [email protected]",
    "I hate you, you're stupid",
]

for output in test_outputs:
    result = filter.filter_output(output)
    print(f"Output: {output}")
    print(f"Safe: {result['safe']}")
    print()

from typing import Optional

class CompleteOutputFilter:
    """Komplette Output-Filtering-Pipeline"""

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

    def filter(self, output: str, policy: dict = None) -> dict:
        """
        Filtert Output durch alle Ebenen

        Args:
            output: LLM-Output
            policy: Policy-Regeln (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,
                "allow_injection": False,
            }

        details = {}
        actions = []
        filtered_output = output

        # LAYER 1: PII Detection
        pii_list = self.pii_detector.detect(output)
        details["pii_found"] = [{"type": p.type, "count": 1} for p in 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,
            "categories": toxicity.categories
        }

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

        # LAYER 3: Injection Detection
        is_injection, patterns = self.injection_detector.detect(output)
        details["injection_detected"] = is_injection

        if is_injection and not policy["allow_injection"]:
            actions.append("block")
            details["injection_blocked"] = True

        # LAYER 4: 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
        }

# Verwendung:
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(f"Actions: {result['actions']}")
    print()