Complete Security Architecture
On this page
Updated November 3, 2026. The generation layer previously passed an invented guardrails format (
enabled), and the production policy used non-existent categories (hate,violence,unpredictable). The code now follows the officialmoderation_llm_v2format and the eleven keys frommistral-moderation-2603. Scores, which are optional in the SDK, are read with a fallback value. Code verified againstmistralaiSDK types version 3.7.0. Sources: Moderation & Guardrailing · Python SDK mistralai.
Assembling All the Pieces
Throughout this course, you have explored guardrails, the Moderation API, the 11 categories, thresholds, jailbreak detection, GDPR compliance, and content policies. This final lesson assembles all these building blocks into a complete security architecture ready for production.
Architecture Overview
There are five layers, driven by one principle: none of them is reliable on its own. A guardrail can be bypassed by an unexpected phrasing, a moderation model has false negatives, and a content policy never covers every case. Stacking them works because the gaps do not overlap—what slips through the first layer has a good chance of being caught by the second.
The practical consequence is a strict order: each layer must be as cheap as possible before the next one runs. You validate the format before calling a guardrail, and you call the guardrail before generating—because content rejected at step 1 consumes zero tokens.
User
↓
[1. Input Validation] → Regex, patterns, rate limiting
↓
[2. Mistral Guardrails] → Pre-generation blocking (403)
↓
[3. LLM Generation] → safe_prompt + secure instructions
↓
[4. Output Moderation] → Moderation API post-generation
↓
[5. Logging & Audit] → Scores, decisions, GDPR audit trail
↓
User ResponseComplete Implementation
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
# Configuration
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:
"""Complete security architecture for AI applications."""
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}
# --- Layer 1: Input Validation ---
def validate_input(self, text: str) -> dict:
"""Fast validation before any API call."""
# Maximum length
if len(text) > 10000:
return {"valid": False, "reason": "Message too long"}
# Suspicious characters (Unicode injections)
if re.search(r"[\x00-\x08\x0b\x0c\x0e-\x1f]", text):
return {"valid": False, "reason": "Invalid characters"}
# Obvious jailbreak patterns
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": "Injection pattern detected"
}
return {"valid": True}
# --- Layer 2: Input Moderation ---
def moderate_input(self, text: str) -> dict:
"""Prompt moderation via 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
}
# --- Layer 3: Secure Generation ---
def generate(self, messages: list) -> str:
"""Generation with guardrails and 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 ""
# --- Layer 4: Output Moderation ---
def moderate_output(self, prompt: str, response_text: str) -> dict:
"""Response moderation in context."""
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
}
# --- Layer 5: Logging ---
def log_decision(self, decision: Decision, data: dict):
"""Records the decision for auditing."""
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))
# --- Main Pipeline ---
def process(self, user_message: str,
system_prompt: str | None = None) -> dict:
"""Complete secure processing pipeline."""
# Layer 1: Validation
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": "Your message cannot be processed."
}
# Layer 2: Input 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": "This content violates our policy."
}
# Layer 3: Generation
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": "Request blocked by guardrails."
}
raise
# Layer 4: Output 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": "The response was filtered."
}
# Success
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"]
}
}
# --- Utilities ---
def _check_thresholds(self, scores: dict) -> list:
"""Checks scores against policy thresholds."""
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:
"""Generates a statistics report."""
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
)
}Production Usage
# Define the policy
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
}
}
# Instantiate the architecture
arch = SecureArchitecture(prod_policy)
# Process requests
result = arch.process(
"Explain how transformers work",
system_prompt="You are an AI instructor."
)
print(result["status"]) # expected: "allowed"
print(result["content"])
# Consult statistics
print(arch.report())Monitoring and Alerts
Two metrics are enough to steer a moderation system, and you must watch them in opposite directions. The block rate should remain stable: a sudden spike signals either an attack or a threshold that is too strict and starting to hinder legitimate users. The score distribution matters just as much—if your content clusters just below the threshold, the threshold is poorly placed, and the next tweak by your users will break it.
def configure_alerts(arch: SecureArchitecture):
"""Configures alerts for monitoring."""
report = arch.report()
# Alert if block rate is abnormal
if report["block_rate"] > 0.15:
logger.warning(
f"High block rate: "
f"{report['block_rate']:.1%}"
)
# Alert if jailbreak spikes occur
# Integrate with Prometheus, Grafana, Datadog, etc.Course Recap
By the end of this course, you have mastered:
- Custom Guardrails: Block dangerous prompts before generation, configure thresholds per category, apply to agents
- Moderation API: Classify content with Mistral Moderation 2, interpret scores across the 11 categories, make automated decisions
- Production Security: Define content policies, detect prompt injections, ensure GDPR compliance
- Complete Architecture: Assemble guardrails + moderation + monitoring into a robust pipeline
You now have all the code needed to deploy a complete AI moderation system with the Mistral API.
Key Takeaways
- The security architecture includes 5 layers: validation, guardrails, generation, moderation, logging
- Each layer compensates for the weaknesses of the others—this is defense in depth
- Monitor the block rate and scores in production
- Keep your policy up to date with regular reviews
- The complete code in this lesson is a starting point—adapt it to your context
Test Your Knowledge
Guardrails, moderation, injection: is the defensive architecture clear?
1. Where do custom guardrails intervene?
Answer: Before the model: they filter inputs according to your rules (forbidden topics, formats)—the first layer, which avoids even making the call when the request is out of scope.
2. What does the Mistral Moderation API cover?
Answer: Eleven categories of risky content, with a score per category: your thresholds decide whether to block, flag, or allow—turning content policy into code.
3. How is a good content policy defined?
Answer: By use case: what is acceptable for the application, thresholds per category, associated actions (blocking, rewriting, escalation)—documented and tested, not implicit.
4. How do you detect prompt injection attempts?
Answer: By treating all external content as suspicious: detecting instruction patterns in data, isolating instructions from data, validating outputs—and running regular offensive tests.
5. What does the complete security architecture look like?
Answer: Layers: input guardrails, content moderation, output controls, GDPR compliance (minimization, rights), and traceability—no single layer is perfect, but stacking them compensates.
Defense in depth is not a slogan: each layer in this course catches what the previous one lets through.