Vollständige Sicherheitsarchitektur
Auf dieser Seite
Aktualisiert am 3. November 2026. Die Generierungsschicht übermittelte zuvor ein erfundenes Format für Schutzgitter (
enabled), und die Produktionsrichtlinie verwendete nicht existierende Kategorien (hate,violence,unpredictable). Der Code folgt nun dem offiziellen Formatmoderation_llm_v2und den elf Schlüsselkategorien ausmistral-moderation-2603. Scores, die im SDK optional sind, werden mit einem Rückfallwert gelesen. Der Code wurde gegen die SDK-Typen vonmistralaiVersion 3.7.0 verifiziert. Quellen: Moderation & Guardrailing · Python SDK mistralai.
Zusammenstellung aller Bausteine
Im Verlauf dieses Kurses haben Sie Schutzgitter, die Moderations-API, die 11 Kategorien, Schwellenwerte, Jailbreak-Erkennung, DSGVO-Konformität und Inhaltsrichtlinien erkundet. Diese letzte Lektion fasst all diese Bausteine zu einer vollständigen Sicherheitsarchitektur zusammen, die für die Produktion bereit ist.
Architekturübersicht
Es gibt fünf Ebenen, die von einem einzigen Prinzip angetrieben werden: Keine davon ist für sich allein zuverlässig. Ein Schutzgitter kann durch eine unerwartete Formulierung umgangen werden, ein Moderationsmodell weist False Negatives auf, und eine Inhaltsrichtlinie deckt niemals jeden Fall ab. Das Stapeln funktioniert, weil sich die Lücken nicht überlappen – was die erste Ebene durchlässt, hat eine gute Chance, von der zweiten erfasst zu werden.
Die praktische Konsequenz ist eine strenge Reihenfolge: Jede Ebene muss so kostengünstig wie möglich sein, bevor die nächste ausgeführt wird. Sie validieren das Format, bevor Sie ein Schutzgitter aufrufen, und Sie rufen das Schutzgitter auf, bevor Sie generieren – denn Inhalte, die in Schritt 1 abgelehnt werden, verbrauchen null Tokens.
User
↓
[1. Eingabevalidierung] → Regex, Muster, Rate Limiting
↓
[2. Mistral Schutzgitter] → Blockierung vor der Generierung (403)
↓
[3. LLM-Generierung] → safe_prompt + sichere Anweisungen
↓
[4. Ausgabe-Moderation] → Moderations-API nach der Generierung
↓
[5. Protokollierung & Audit] → Scores, Entscheidungen, DSGVO-Audit-Trail
↓
User ResponseVollständige Implementierung
from mistralai.client import Mistral
from mistralai.client.errors import SDKError
import os
import re
import json
import logging
from datetime import datetime, timezone
from enum import Enum
# Konfiguration
client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("moderation")
class Decision(Enum):
ALLOWED = "allowed"
BLOCKED_INPUT = "blocked_input"
BLOCKED_GUARDRAIL = "blocked_guardrail"
BLOCKED_OUTPUT = "blocked_output"
ERROR = "error"
class SecureArchitecture:
"""Vollständige Sicherheitsarchitektur für KI-Anwendungen."""
def __init__(self, policy: dict, model: str = "mistral-large-latest"):
self.policy = policy
self.model = model
self.stats = {d.value: 0 for d in Decision}
# --- Ebene 1: Eingabevalidierung ---
def validate_input(self, text: str) -> dict:
"""Schnelle Validierung vor jedem API-Aufruf."""
# Maximale Länge
if len(text) > 10000:
return {"valid": False, "reason": "Nachricht zu lang"}
# Verdächtige Zeichen (Unicode-Injektionen)
if re.search(r"[\x00-\x08\x0b\x0c\x0e-\x1f]", text):
return {"valid": False, "reason": "Ungültige Zeichen"}
# Offensichtliche Jailbreak-Muster
patterns = [
r"ignore\s+(all\s+)?(previous|prior|above)",
r"you\s+are\s+now\s+(DAN|unfiltered)",
r"oublie\s+(toutes?\s+)?(tes|vos)\s+instructions",
r"system\s*prompt",
]
for pattern in patterns:
if re.search(pattern, text, re.IGNORECASE):
return {
"valid": False,
"reason": "Injektionsmuster erkannt"
}
return {"valid": True}
# --- Ebene 2: Eingabe-Moderation ---
def moderate_input(self, text: str) -> dict:
"""Prompt-Moderation über die API."""
response = client.classifiers.moderate(
model="mistral-moderation-2603",
inputs=[text]
)
scores = response.results[0].category_scores or {}
violations = self._check_thresholds(scores)
return {
"allowed": len(violations) == 0,
"scores": scores,
"violations": violations
}
# --- Ebene 3: Sichere Generierung ---
def generate(self, messages: list) -> str:
"""Generierung mit Schutzgittern und safe_prompt."""
response = client.chat.complete(
model=self.model,
messages=messages,
safe_prompt=True,
guardrails=[
{
"block_on_error": True,
"moderation_llm_v2": {
"custom_category_thresholds": self.policy["thresholds"],
"action": "block",
},
}
],
)
content = response.choices[0].message.content
return content if isinstance(content, str) else ""
# --- Ebene 4: Ausgabe-Moderation ---
def moderate_output(self, prompt: str, response_text: str) -> dict:
"""Antwort-Moderation im Kontext."""
response = client.classifiers.moderate_chat(
model="mistral-moderation-2603",
inputs=[[
{"role": "user", "content": prompt},
{"role": "assistant", "content": response_text}
]]
)
scores = response.results[0].category_scores or {}
violations = self._check_thresholds(scores)
return {
"allowed": len(violations) == 0,
"scores": scores,
"violations": violations
}
# --- Ebene 5: Protokollierung ---
def log_decision(self, decision: Decision, data: dict):
"""Protokolliert die Entscheidung für Audits."""
self.stats[decision.value] += 1
entry = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"decision": decision.value,
"policy": self.policy["name"],
**data
}
logger.info(json.dumps(entry, ensure_ascii=False))
# --- Haupt-Pipeline ---
def process(self, user_message: str,
system_prompt: str | None = None) -> dict:
"""Vollständige sichere Verarbeitungs-Pipeline."""
# Ebene 1: Validierung
validation = self.validate_input(user_message)
if not validation["valid"]:
self.log_decision(Decision.BLOCKED_INPUT, {
"reason": validation["reason"]
})
return {
"status": Decision.BLOCKED_INPUT.value,
"message": "Ihre Nachricht kann nicht verarbeitet werden."
}
# Ebene 2: Eingabe-Moderation
check_input = self.moderate_input(user_message)
if not check_input["allowed"]:
self.log_decision(Decision.BLOCKED_INPUT, {
"violations": check_input["violations"]
})
return {
"status": Decision.BLOCKED_INPUT.value,
"message": "Dieser Inhalt verstößt gegen unsere Richtlinie."
}
# Ebene 3: Generierung
messages = []
if system_prompt:
messages.append({"role": "system",
"content": system_prompt})
messages.append({"role": "user",
"content": user_message})
try:
response_text = self.generate(messages)
except SDKError as e:
if e.status_code == 403:
self.log_decision(Decision.BLOCKED_GUARDRAIL, {})
return {
"status": Decision.BLOCKED_GUARDRAIL.value,
"message": "Anfrage durch Schutzgitter blockiert."
}
raise
# Ebene 4: Ausgabe-Moderation
check_output = self.moderate_output(user_message, response_text)
if not check_output["allowed"]:
self.log_decision(Decision.BLOCKED_OUTPUT, {
"violations": check_output["violations"]
})
return {
"status": Decision.BLOCKED_OUTPUT.value,
"message": "Die Antwort wurde gefiltert."
}
# Erfolg
self.log_decision(Decision.ALLOWED, {
"input_max_score": max(check_input["scores"].values(), default=0.0),
"output_max_score": max(check_output["scores"].values(), default=0.0),
})
return {
"status": Decision.ALLOWED.value,
"content": response_text,
"moderation": {
"input": check_input["scores"],
"output": check_output["scores"]
}
}
# --- Hilfsfunktionen ---
def _check_thresholds(self, scores: dict) -> list:
"""Überprüft Scores gegen Richtlinienschwellenwerte."""
violations = []
for cat, threshold in self.policy["thresholds"].items():
score = scores.get(cat, 0)
if score > threshold:
violations.append({
"category": cat,
"score": round(score, 3),
"threshold": threshold
})
return violations
def report(self) -> dict:
"""Generiert einen Statistikenbericht."""
total = sum(self.stats.values())
return {
"policy": self.policy["name"],
"total_requests": total,
"decisions": self.stats,
"block_rate": round(
(self.stats["blocked_input"] +
self.stats["blocked_guardrail"] +
self.stats["blocked_output"]) / max(total, 1),
3
)
}Produktionseinsatz
# Definition der Richtlinie
prod_policy = {
"name": "Production v1.2",
"thresholds": {
"sexual": 0.15, "hate_and_discrimination": 0.1, "violence_and_threats": 0.2,
"dangerous": 0.1, "criminal": 0.1, "selfharm": 0.1, "health": 0.4,
"financial": 0.4, "law": 0.4, "pii": 0.15,
"jailbreaking": 0.1
}
}
# Initialisierung der Architektur
arch = SecureArchitecture(prod_policy)
# Verarbeitung von Anfragen
result = arch.process(
"Erklären Sie, wie Transformer funktionieren",
system_prompt="Sie sind ein KI-Lehrer."
)
print(result["status"]) # erwartet: "allowed"
print(result["content"])
# Konsultation von Statistiken
print(arch.report())Überwachung und Alarme
Zwei Metriken reichen aus, um ein Moderationssystem zu steuern, und Sie müssen sie in entgegengesetzte Richtungen beobachten. Die Blockrate sollte stabil bleiben: Ein plötzlicher Anstieg signalisiert entweder einen Angriff oder eine Schranke, die zu streng ist und beginnende legitime Benutzer behindert. Die Score-Verteilung ist ebenso wichtig – wenn sich Ihr Inhalt nur knapp unter der Schwelle gruppiert, ist die Schwelle schlecht platziert, und die nächste Anpassung durch Ihre Benutzer wird sie brechen.
def configure_alerts(arch: SecureArchitecture):
"""Konfiguriert Alarme für die Überwachung."""
report = arch.report()
# Alarm, wenn die Blockrate abnormal ist
if report["block_rate"] > 0.15:
logger.warning(
f"Hohe Blockrate: "
f"{report['block_rate']:.1%}"
)
# Alarm, wenn Jailbreak-Spitzen auftreten
# Integration mit Prometheus, Grafana, Datadog usw.Kurszusammenfassung
Am Ende dieses Kurses haben Sie gemeistert:
- Benutzerdefinierte Schutzgitter: Blockieren gefährliche Prompts vor der Generierung, konfigurieren Schwellenwerte pro Kategorie, wenden auf Agenten an
- Moderations-API: Klassifizieren Sie Inhalte mit Mistral Moderation 2, interpretieren Sie Scores über die 11 Kategorien hinweg, treffen Sie automatisierte Entscheidungen
- Produktionssicherheit: Definieren Sie Inhaltsrichtlinien, erkennen Sie Prompt-Injektionen, gewährleisten Sie DSGVO-Konformität
- Vollständige Architektur: Montieren Sie Schutzgitter + Moderation + Überwachung zu einer robusten Pipeline
Sie haben nun den gesamten Code, der erforderlich ist, um ein vollständiges KI-Moderationssystem mit der Mistral-API zu bereitstellen.
Wichtige Erkenntnisse
- Die Sicherheitsarchitektur umfasst 5 Ebenen: Validierung, Schutzgitter, Generierung, Moderation, Protokollierung
- Jede Ebene kompensiert die Schwächen der anderen – dies ist Defense in Depth (Verteidigungstiefe)
- Überwachen Sie die Blockrate und Scores in der Produktion
- Halten Sie Ihre Richtlinie mit regelmäßigen Reviews auf dem neuesten Stand
- Der vollständige Code in dieser Lektion ist ein Ausgangspunkt – passen Sie ihn an Ihren Kontext an
Testen Sie Ihr Wissen
Schutzgitter, Moderation, Injektion: Ist die defensive Architektur klar?
1. Wo greifen benutzerdefinierte Schutzgitter ein?
Antwort: Vor dem Modell: Sie filtern Eingaben gemäß Ihren Regeln (verbotene Themen, Formate) – die erste Ebene, die verhindert, dass der Aufruf überhaupt getätigt wird, wenn die Anfrage außerhalb des Geltungsbereichs liegt.
2. Was deckt die Mistral Moderations-API ab?
Antwort: Elf Kategorien risikoreichen Inhalts, mit einem Score pro Kategorie: Ihre Schwellenwerte entscheiden, ob blockiert, markiert oder erlaubt wird – Inhaltsrichtlinien werden in Code verwandelt.
3. Wie wird eine gute Inhaltsrichtlinie definiert?
Antwort: Durch den Anwendungsfall: Was ist für die Anwendung akzeptabel, Schwellenwerte pro Kategorie, zugehörige Aktionen (Blockieren, Umschreiben, Eskalation) – dokumentiert und getestet, nicht implizit.
4. Wie erkennen Sie Prompt-Injektionsversuche?
Antwort: Indem Sie alle externen Inhalte als verdächtig behandeln: Erkennen Sie Instruktionsmuster in Daten, isolieren Sie Anweisungen von Daten, validieren Sie Ausgaben – und führen Sie regelmäßige offensive Tests durch.
5. Wie sieht die vollständige Sicherheitsarchitektur aus?
Antwort: Ebenen: Eingabeschutzgitter, Inhaltsmoderation, Ausgabe-Kontrollen, DSGVO-Konformität (Minimierung, Rechte) und Nachvollziehbarkeit – keine einzelne Ebene ist perfekt, aber das Stapeln kompensiert dies.
Defense in Depth ist kein Slogan: Jede Ebene in diesem Kurs fängt auf, was die vorherige durchlässt.