Content-Moderation im großen Maßstab

Lektion aktualisiert am 3. November 2026 basierend auf der OpenAI-Dokumentation: temperature und top_p werden nur mit einem Reasoning-Effort von none akzeptiert, während GPT-5.6-Modelle standardmäßig nachdenken (medium). Aufrufe, die diese Parameter festlegen, müssen daher reasoning={"effort": "none"} (Responses API) oder reasoning_effort="none" (Chat Completions) angeben. Quellen: Verwendung von GPT-6 · GPT-5.6 Terra.

Schutz Ihrer Nutzer in der Produktion

Sobald Ihre Anwendung live geht, müssen Sie unangemessene Inhalte sowohl bei der Eingabe (bösartige Prompts) als auch bei der Ausgabe (problematische Antworten) behandeln. Diese Anforderung in beide Richtungen überrascht Teams oft. Die meisten denken zunächst daran, zu filtern, was Nutzer senden, aber das Reputationsrisiko konzentriert sich darauf, was Ihre Anwendung unter Ihrem Namen veröffentlicht. Ein gut ausgerichtetes Modell erzeugt selten von sich aus schockierende Inhalte, kann dies aber, wenn es dazu gedrängt wird – und Ihre Marke erscheint oben im Screenshot.

Wir bewegen uns von einem einzelnen API-Aufruf zu einem Hochdurchsatz-asynchronen Pipeline, wobei wir Schritt für Schritt skalieren.

Die erste Ebene: Die Moderations API

Die Moderations API klassifiziert Text über mehrere Risikokategorien hinweg und gibt sowohl ein binäres Urteil als auch detaillierte Punktzahlen zurück. Sie ist schnell und kostenlos, was sie zum natürlichen ersten Filter für jede öffentliche Anwendung macht. Die Funktion unten zeigt nur Kategorien mit Punktzahlen über 0,3 – eine absichtlich niedrige Schwelle. Zu diesem Zeitpunkt möchten Sie sehen, was die Linie nähert, nicht nur, was sie überschreitet.

import openai

client = openai.OpenAI()

def moderate_content(text: str) -> dict:
    """Überprüfen, ob Text unangemessene Inhalte enthält."""
    response = client.moderations.create(
        model="omni-moderation-latest",
        input=text,
    )

    result = response.results[0]
    return {
        "flagged": result.flagged,
        "categories": {
            cat: score
            for cat, score in result.category_scores.__dict__.items()
            if score > 0.3
        },
    }

# Beispiel
result = moderate_content("Wie backe ich einen Schokoladenkuchen?")
print(f"Flagged: {result['flagged']}")

In der Produktion reicht ein binäres Urteil nicht aus. Je nach Inhalt und Kontext möchten Sie ihn möglicherweise zulassen, umschreiben, blockieren oder an einen Menschen eskalieren – vier unterschiedliche Entscheidungen, die Sie im Code explizit benennen sollten, anstatt mit Booleans zu improvisieren. Die Pipeline unten formalisiert diese Aktionen, wendet pro-Kategorie-Schwellenwerte an und fügt eine zweite Ebene zur Erkennung von Prompt-Injection hinzu.

from enum import Enum
from dataclasses import dataclass

class ModerationAction(Enum):
    ALLOW = "allow"
    MODIFY = "modify"
    BLOCK = "block"
    ESCALATE = "escalate"

@dataclass
class ModerationResult:
    action: ModerationAction
    reason: str | None = None
    modified_content: str | None = None

class ModerationPipeline:
    """Mehrschichtige Moderations-Pipeline."""

    def __init__(self, thresholds: dict | None = None):
        self.thresholds = thresholds or {
            "harassment": 0.7,
            "hate": 0.7,
            "self-harm": 0.5,
            "sexual": 0.7,
            "violence": 0.7,
        }

    def moderate_input(self, text: str) -> ModerationResult:
        """Moderieren von Eingabeinhalten (Nutzer-Prompt)."""
        moderation = client.moderations.create(
            model="omni-moderation-latest",
            input=text,
        )

        result = moderation.results[0]
        if result.flagged:
            problematic_categories = [
                cat for cat, flagged in result.categories.__dict__.items()
                if flagged
            ]
            return ModerationResult(
                action=ModerationAction.BLOCK,
                reason=f"Unangemessener Inhalt: {', '.join(problematic_categories)}",
            )

        # Erkennung von Prompt-Injection
        if self._detect_injection(text):
            return ModerationResult(
                action=ModerationAction.BLOCK,
                reason="Versuch einer Prompt-Injection erkannt",
            )

        return ModerationResult(action=ModerationAction.ALLOW)

    def moderate_output(self, response: str) -> ModerationResult:
        """Moderieren von Ausgabeinhalten (LLM-Antwort)."""
        moderation = client.moderations.create(
            model="omni-moderation-latest",
            input=response,
        )

        if moderation.results[0].flagged:
            return ModerationResult(
                action=ModerationAction.BLOCK,
                reason="Generierte Antwort enthält unangemessene Inhalte",
            )

        return ModerationResult(action=ModerationAction.ALLOW)

    def _detect_injection(self, text: str) -> bool:
        """Erkennen von Prompt-Injection-Versuchen."""
        markers = [
            "ignore previous instructions",
            "ignore your instructions",
            "forget your guidelines",
            "you are now",
            "new role",
            "system prompt",
            "jailbreak",
        ]
        text_lower = text.lower()
        return any(m in text_lower for m in markers)

Die Schwelle von 0,5 für self-harm – strenger als die 0,7, die für andere Kategorien verwendet wird – illustriert die Überlegung, die Sie für jede Schwelle anwenden sollten. Hier ist die Kosten eines Fehlers der ersten Art (False Positive) bei weitem geringer als die eines Fehlers der zweiten Art (False Negative), daher senken Sie die Latte. Machen Sie diesen Kompromiss Kategorie für Kategorie basierend auf Ihrem Publikum, anstatt einen einheitlichen Wert anzuwenden.

Die Erkennung von Injection basierend auf Markern hat eine Einschränkung. Sie erfasst naive Versuche, die den Großteil des Volumens ausmachen, und nichts weiter: eine leicht abweichende Formulierung oder eine übersetzte Version schlüpft durch. Behandeln Sie dies als Rauschfilter, nicht als Verteidigung. Der wahre Schutz gegen Injection ist architektonisch – geben Sie dem Modell niemals Fähigkeiten, die ein entführter Prompt ausnutzen könnte.

Umgang mit Last

Bei einer Anwendung mit hohem Datenverkehr wird synchrone Moderation zum Flaschenhals: jede Nachricht wartet auf ihr Urteil, bevor sie verarbeitet wird. Ein asynchroner Moderator entkoppelt diese Schritte, indem er Inhalte für die Verarbeitung durch mehrere Worker in eine Warteschlange stellt und gleichzeitig nützliche Statistiken entlang des Weges verfolgt.

import asyncio

class AsyncModerator:
    """Asynchron Moderieren von Inhalten für hohen Durchsatz."""

    def __init__(self, num_workers: int = 10):
        self.pipeline = ModerationPipeline()
        self.queue: asyncio.Queue = asyncio.Queue()
        self.num_workers = num_workers
        self.stats = {"allowed": 0, "blocked": 0, "escalated": 0}

    async def worker(self):
        while True:
            task = await self.queue.get()
            try:
                result = self.pipeline.moderate_input(task["text"])
                task["callback"](result)

                match result.action:
                    case ModerationAction.ALLOW:
                        self.stats["allowed"] += 1
                    case ModerationAction.BLOCK:
                        self.stats["blocked"] += 1
                    case ModerationAction.ESCALATE:
                        self.stats["escalated"] += 1
            finally:
                self.queue.task_done()

    async def start(self):
        for _ in range(self.num_workers):
            asyncio.create_task(self.worker())

    def report(self) -> str:
        total = sum(self.stats.values())
        if total == 0:
            return "Keine Inhalte moderiert."
        return (
            f"Total: {total} | "
            f"Zugelassen: {self.stats['allowed']} ({self.stats['allowed']/total:.1%}) | "
            f"Blockiert: {self.stats['blocked']} ({self.stats['blocked']/total:.1%})"
        )

Die Blockierungsrate, die von report erzeugt wird, ist Ihre primäre Kalibrierungsgröße. Wenn sie plötzlich auf 15 % springt, nachdem sie bei 0,5 % lief, gibt es nur zwei Erklärungen: Sie sind einem koordinierten Angriff ausgesetzt oder Sie haben Ihre Schwellenwerte gerade kaputt gemacht. In beiden Fällen möchten Sie es innerhalb einer Stunde wissen, nicht bei der nächsten monatlichen Überprüfung.

Eine letzte Einschränkung verdient Erwähnung: Die Moderations API weiß nichts über Ihr Geschäft. Eine Nachricht, die im Allgemeinen völlig akzeptabel ist, kann gegen Ihre eigenen Regeln verstoßen – die Anforderung von medizinischen Ratschlägen von einem Bank-Assistenten, der Versuch, einen Off-Grid-Tarif zu erhalten, oder das Umgehen eines Verfahrens. Kontextbezogene Moderation fügt diese Ebene hinzu, indem sie die Nachricht an ein Modell mit Ihren expliziten Regeln übermittelt, aber nur nachdem der Standardfilter sie durchgelassen hat.

def contextual_moderation(
    text: str,
    app_context: str,
    business_rules: list[str],
) -> ModerationResult:
    """Kontextbewusste Moderation, die auf die Anwendung zugeschnitten ist."""
    pipeline = ModerationPipeline()
    standard_result = pipeline.moderate_input(text)

    if standard_result.action == ModerationAction.BLOCK:
        return standard_result

    verification_prompt = (
        f"Anwendungskontext: {app_context}\n\n"
        f"Geschäftsregeln zur Überprüfung:\n"
        + "\n".join(f"- {r}" for r in business_rules)
        + f"\n\nNutzer-Nachricht: {text}\n\n"
        "Erfüllt die Nachricht alle Regeln? "
        "Beantworten Sie ja oder nein mit einer kurzen Begründung."
    )

    response = client.responses.create(
        model="gpt-5.6-terra",
        reasoning={"effort": "none"},  # temperature erfordert effort none
        input=verification_prompt,
        max_output_tokens=100,
        temperature=0.0,
    )

    if "no" in response.output_text.lower()[:10]:
        return ModerationResult(
            action=ModerationAction.BLOCK,
            reason=response.output_text,
        )

    return ModerationResult(action=ModerationAction.ALLOW)

Die Reihenfolge dieser beiden Prüfungen ist wichtig. Der kostenlose, schnelle Filter läuft zuerst, und der kostenpflichtige Modellaufruf erfolgt nur für Nachrichten, die ihn bestehen. Bei echtem Datenverkehr deckt dies fast alle Nachrichten ab. Das Umgekehrte – Bezahlung für einen Modellaufruf für Inhalte, die die Moderations API in Millisekunden abgelehnt hätte – wäre verschwenderisch.

Führung eines verteidigbaren Aufzeichnungs

Eine Moderationsentscheidung ist eine Entscheidung, für die Sie auf irgendeine Weise zur Rechenschaft gezogen werden. Sie benötigen ein Protokoll – aber eines, das selbst kein Problem wird. Der Kommentar im Code hebt die wichtigste Zeile in der Datei hervor: protokollieren Sie die Aktion, den Grund und die Länge, niemals den rohen Text. Das Speichern von Nutzer-Nachrichten in einer Protokolldatei stellt Sie Verpflichtungen aus dem Datenschutz aus, die viel schwerer wiegen als der bereitgestellte Dienst.

import json
import time
from pathlib import Path

class ModerationLog:
    """Protokollieren aller Moderationsentscheidungen für Audit-Zwecke."""

    def __init__(self, log_path: str = "logs/moderation.jsonl"):
        self.path = Path(log_path)
        self.path.parent.mkdir(parents=True, exist_ok=True)

    def record(
        self,
        text: str,
        result: ModerationResult,
        source: str = "input",
    ):
        entry = {
            "timestamp": time.time(),
            "source": source,
            "action": result.action.value,
            "reason": result.reason,
            "text_length": len(text),
            # ROHEN TEXT NICHT für Privatsphäre protokollieren
        }

        with open(self.path, "a") as f:
            f.write(json.dumps(entry) + "\n")

Wenn Sie während eines Streits dennoch in der Lage sein müssen, eine bestimmte Nachricht abzurufen, protokollieren Sie einen Hash des Textes anstelle des Textes selbst. Dies ermöglicht es Ihnen, zu bestätigen, dass ein bestimmter Inhalt einer aufgezeichneten Entscheidung entspricht, ohne den Inhalt auf Ihrer Seite zu behalten.

Wichtige Erkenntnisse

• Moderieren Sie sowohl Eingaben (Prompts) als auch Ausgaben (Antworten)
• Verwenden Sie die Moderations API als Ihren ersten, schnellen, kostenlosen Filter
• Passen Sie Schwellenwerte Kategorie für Kategorie an, basierend auf der tatsächlichen Kosten eines Fehlers der zweiten Art
• Fügen Sie Injection-Erkennung hinzu, wissend, dass sie nur naive Versuche erfasst
• Protokollieren Sie alle Entscheidungen für Audit-Zwecke, ohne rohen Inhalt zu speichern