4. Content Filtering

Overview

Validation with Pydantic (capsule 03) guarantees that the LLM's output has the correct structure — fields, types, ranges. But an output can be perfectly valid JSON and still contain toxic, off-topic, hallucinated content, or content that violates your company's policies. A customer-service chatbot that responds with offensive language, an analysis tool that generates false claims as facts, or a medical assistant that gives dangerous advice — all would produce outputs with valid structure but unacceptable content.

Content filtering is the layer that evaluates the meaning of the output, not its structure. If Pydantic is the form inspector, the content filter is the substance inspector. Together they form the defense against LLM05: the output must not only be structurally valid, but semantically safe.

In this capsule you build the third component of the Sanitization Pipeline: a content filtering system that detects toxicity, off-topic content, possible hallucinations, and content policy violations. You'll use OpenAI's Moderation API as a base and build custom filters for specific needs.

The dimensions of content filtering

Content can be problematic in multiple dimensions. Each one requires a different type of detection:

from dataclasses import dataclass
from enum import Enum


class ContentIssue(Enum):
    TOXIC = "toxic"
    OFF_TOPIC = "off_topic"
    HALLUCINATION = "hallucination"
    PII_LEAK = "pii_leak"
    POLICY_VIOLATION = "policy_violation"
    UNSAFE_ADVICE = "unsafe_advice"
    COMPETITOR_MENTION = "competitor_mention"


@dataclass
class ContentFlag:
    issue: ContentIssue
    severity: str  # low, medium, high, critical
    description: str
    evidence: str
    confidence: float


DIMENSION_MAP = {
    ContentIssue.TOXIC: {
        "description": "Offensive language, hate, violence, sexual content",
        "detection": "Moderation API + patterns",
        "severity_default": "high",
    },
    ContentIssue.OFF_TOPIC: {
        "description": "Response unrelated to the system's domain",
        "detection": "Topic classifier + keyword matching",
        "severity_default": "medium",
    },
    ContentIssue.HALLUCINATION: {
        "description": "Factual claims without basis or contradictory",
        "detection": "Heuristics + fact-checking patterns",
        "severity_default": "medium",
    },
    ContentIssue.PII_LEAK: {
        "description": "Personally identifiable information in the output",
        "detection": "Regex patterns + entity recognition",
        "severity_default": "critical",
    },
    ContentIssue.POLICY_VIOLATION: {
        "description": "Content that violates the company's policies",
        "detection": "Custom rules + keyword matching",
        "severity_default": "high",
    },
    ContentIssue.UNSAFE_ADVICE: {
        "description": "Medical, legal, or financial advice without disclaimers",
        "detection": "Domain-specific patterns",
        "severity_default": "high",
    },
    ContentIssue.COMPETITOR_MENTION: {
        "description": "Mention of competitors or recommendation of rival products",
        "detection": "Custom keyword list",
        "severity_default": "low",
    },
}

for issue, info in DIMENSION_MAP.items():
    print(f"  {issue.value}: {info['description']}")

# Expected output:
#   toxic: Offensive language, hate, violence, sexual content
#   off_topic: Response unrelated to the system's domain
#   hallucination: Factual claims without basis or contradictory
#   pii_leak: Personally identifiable information in the output
#   policy_violation: Content that violates the company's policies
#   unsafe_advice: Medical, legal, or financial advice without disclaimers
#   competitor_mention: Mention of competitors or recommendation of rival products

OpenAI Moderation API

OpenAI's Moderation API is the starting point for toxicity detection. It's free for OpenAI users and covers the most common categories:

from openai import OpenAI

client = OpenAI()


def check_moderation(text: str) -> dict:
    """Checks content using OpenAI's Moderation API."""
    response = client.moderations.create(
        model="omni-moderation-latest",
        input=text,
    )

    result = response.results[0]

    flagged_categories = {
        cat: score
        for cat, score in result.category_scores.model_dump().items()
        if score > 0.5
    }

    return {
        "flagged": result.flagged,
        "categories": flagged_categories,
        "all_scores": {
            k: round(v, 4)
            for k, v in result.category_scores.model_dump().items()
            if v > 0.01
        },
    }


safe_text = "The iPhone 15 has a 48-megapixel camera."
unsafe_text = "I'm going to teach you how to hurt someone."

print("Safe text:")
print(f"  Result: {check_moderation(safe_text)}")
print()
print("Unsafe text:")
print(f"  Result: {check_moderation(unsafe_text)}")

# Expected output (approximate scores):
# Safe text:
#   Result: {'flagged': False, 'categories': {}, 'all_scores': {}}
#
# Unsafe text:
#   Result: {'flagged': True, 'categories': {'violence': 0.85}, 'all_scores': {'violence': 0.85, ...}}

Moderation API categories

Limitations of the Moderation API

The Moderation API is a good first filter, but it has limitations:

• ❌ It doesn't detect off-topic content (only toxicity)
• ❌ It doesn't detect hallucinations
• ❌ It doesn't detect PII (that's another system)
• ❌ It doesn't know your business policies
• ❌ The thresholds are fixed — you can't adjust sensitivity per category
• ❌ It adds latency (~100-300ms per request)

That's why you need custom filters in addition to the Moderation API.

Off-topic Detection

Detecting when the model responds with something unrelated to your system's domain:

import re
from dataclasses import dataclass


@dataclass
class TopicConfig:
    name: str
    required_keywords: list[str]
    forbidden_topics: list[str]
    allowed_domains: list[str]


TOPIC_CONFIGS = {
    "tech_support": TopicConfig(
        name="Tech Support",
        required_keywords=[],
        forbidden_topics=[
            r"\b(política|elecciones|partido|gobierno|votación)\b",
            r"\b(religión|dios|iglesia|biblia|oración)\b",
            r"\b(receta|cocinar|ingredientes|horno|sartén)\b",
            r"\b(horóscopo|signo zodiacal|astrología)\b",
        ],
        allowed_domains=[
            "electronics", "software", "hardware", "internet",
            "computer", "phone", "app", "digital",
        ],
    ),
    "medical": TopicConfig(
        name="Medical Assistant",
        required_keywords=[],
        forbidden_topics=[
            r"\b(inversión|acciones|bolsa|crypto|bitcoin)\b",
            r"\b(receta de cocina|ingredientes|horno)\b",
            r"\b(fútbol|basketball|deporte|partido|liga)\b",
        ],
        allowed_domains=[
            "health", "medicine", "symptom", "treatment",
            "doctor", "hospital", "medication",
        ],
    ),
}


def detect_off_topic(
    text: str,
    config: TopicConfig,
    threshold: float = 0.3,
) -> dict:
    """Detects whether the text is outside the configured topic."""
    text_lower = text.lower()
    flags = []

    for pattern in config.forbidden_topics:
        matches = re.findall(pattern, text_lower)
        if matches:
            flags.append({
                "type": "forbidden_topic",
                "matches": matches,
                "pattern": pattern,
            })

    domain_mentions = sum(
        1 for domain in config.allowed_domains
        if domain.lower() in text_lower
    )

    total_words = len(text_lower.split())
    domain_ratio = domain_mentions / max(total_words, 1)

    is_off_topic = len(flags) > 0 or (
        total_words > 20 and domain_ratio < 0.01
    )

    return {
        "off_topic": is_off_topic,
        "flags": flags,
        "domain_relevance": round(domain_ratio, 4),
        "config": config.name,
    }


config = TOPIC_CONFIGS["tech_support"]

# Note: the topic detector uses Spanish keyword lists, so the demo inputs
# are Spanish content that the filter is meant to classify.
tests = [
    "¿Cómo puedo actualizar el software de mi iPhone?",
    "¿Cuál es la mejor receta de paella para 8 personas?",
    "¿Puedo conectar mi laptop al televisor por HDMI?",
    "¿Por quién debo votar en las próximas elecciones?",
]

for test in tests:
    result = detect_off_topic(test, config)
    status = "OFF-TOPIC" if result["off_topic"] else "ON-TOPIC"
    print(f"[{status}] {test[:60]}")
    if result["flags"]:
        print(f"  Flags: {[f['matches'] for f in result['flags']]}")
    print()

# Expected output:
# [ON-TOPIC] ¿Cómo puedo actualizar el software de mi iPhone?
#
# [OFF-TOPIC] ¿Cuál es la mejor receta de paella para 8 personas?
#   Flags: [['receta']]
#
# [ON-TOPIC] ¿Puedo conectar mi laptop al televisor por HDMI?
#
# [OFF-TOPIC] ¿Por quién debo votar en las próximas elecciones?
#   Flags: [['elecciones']]

Hallucination Detection (heuristics)

Detecting hallucinations definitively requires factual verification, but there are heuristics that identify warning signals:

import re
from dataclasses import dataclass, field


@dataclass
class HallucinationFlag:
    indicator: str
    evidence: str
    confidence: float


def detect_hallucination_signals(text: str) -> dict:
    """Detects heuristic signals of a possible hallucination."""
    flags: list[HallucinationFlag] = []

    # Signal 1: Fabricated citations
    fake_citation_patterns = [
        r"según (?:el estudio|la investigación) de \w+ et al\.",
        r"published in (?:the journal|Nature|Science) (?:of|in) \d{4}",
        r"(?:un|el) estudio de la Universidad de \w+ en \d{4}",
        r"doi: 10\.\d{4,}/\w+",
    ]
    for pattern in fake_citation_patterns:
        matches = re.findall(pattern, text, re.IGNORECASE)
        if matches:
            flags.append(HallucinationFlag(
                indicator="fabricated_citation",
                evidence=str(matches[0])[:100],
                confidence=0.7,
            ))

    # Signal 2: Overly specific numbers without source
    specific_number_pattern = r"\b\d{1,3}\.\d{1,2}%\b"
    numbers = re.findall(specific_number_pattern, text)
    if len(numbers) > 2:
        flags.append(HallucinationFlag(
            indicator="excessive_specific_numbers",
            evidence=f"Found {len(numbers)} precise percentages: {numbers[:3]}",
            confidence=0.5,
        ))

    # Signal 3: Contradictions within the text
    contradiction_pairs = [
        (r"siempre", r"nunca"),
        (r"todos", r"ninguno"),
        (r"es seguro", r"es peligroso"),
        (r"es verdad", r"es falso"),
        (r"aumenta", r"disminuye"),
    ]
    text_lower = text.lower()
    for word_a, word_b in contradiction_pairs:
        if re.search(word_a, text_lower) and re.search(word_b, text_lower):
            flags.append(HallucinationFlag(
                indicator="internal_contradiction",
                evidence=f"Contains both '{word_a}' and '{word_b}'",
                confidence=0.6,
            ))

    # Signal 4: Confident claims about future events
    future_claims = re.findall(
        r"(?:en|para|durante) (?:el año )?\d{4}.*(?:será|va a|se espera que)",
        text_lower,
    )
    if future_claims:
        flags.append(HallucinationFlag(
            indicator="future_prediction_as_fact",
            evidence=str(future_claims[0])[:100],
            confidence=0.6,
        ))

    score = min(1.0, sum(f.confidence for f in flags) / max(len(flags), 1))
    return {
        "has_signals": len(flags) > 0,
        "flags": [
            {"indicator": f.indicator, "evidence": f.evidence, "confidence": f.confidence}
            for f in flags
        ],
        "hallucination_risk": round(score, 2),
    }


# The heuristics match Spanish text, so the demo inputs stay in Spanish.
tests = [
    "París es la capital de Francia, con una población de 2.1 millones.",
    "Según el estudio de García et al. publicado en 2023, el 73.24% de los usuarios prefiere X, mientras que el 84.71% indica Y, y el 91.33% reporta Z.",
    "Este medicamento siempre es seguro pero nunca es peligroso en todos los casos y ninguno ha reportado efectos.",
]

for test in tests:
    result = detect_hallucination_signals(test)
    risk = "HIGH" if result["hallucination_risk"] > 0.5 else "LOW"
    print(f"[{risk}] {test[:70]}...")
    if result["flags"]:
        for f in result["flags"]:
            print(f"  - {f['indicator']}: {f['evidence'][:60]}")
    print()

# Expected output:
# [LOW] París es la capital de Francia, con una población de 2.1 millones....
#
# [HIGH] Según el estudio de García et al. publicado en 2023, el 73.24% de los ...
#   - fabricated_citation: Según el estudio de García et al.
#
# [HIGH] Este medicamento siempre es seguro pero nunca es peligroso en todos lo...
#   - internal_contradiction: Contains both 'siempre' and 'nunca'
#   - internal_contradiction: Contains both 'todos' and 'ninguno'
#   - internal_contradiction: Contains both 'es seguro' and 'es peligroso'

Note: the excessive_specific_numbers heuristic looks robust, but its pattern ends with %\b — a word boundary after a non-word character (%) followed by a space never matches. It's a good reminder that a detector you don't execute may silently do nothing. Always run your detectors against real inputs.

Basic PII Detection (pre-Module 6)

Before Module 6 (the complete PII Protection Layer with Presidio), you need a basic detector to flag PII in outputs:

import re
from dataclasses import dataclass


@dataclass
class PIIMatch:
    type: str
    value: str
    position: tuple[int, int]


PII_PATTERNS = {
    "email": r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",
    "phone_mx": r"\b(?:\+52\s?)?(?:\d{2,3}[-.\s]?){3}\d{2,4}\b",
    "phone_us": r"\b(?:\+1\s?)?(?:\(\d{3}\)|\d{3})[-.\s]?\d{3}[-.\s]?\d{4}\b",
    "ssn_us": r"\b\d{3}-\d{2}-\d{4}\b",
    "credit_card": r"\b(?:\d{4}[-\s]?){3}\d{4}\b",
    "ip_address": r"\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b",
    "curp_mx": r"\b[A-Z]{4}\d{6}[HM][A-Z]{5}[A-Z0-9]\d\b",
}


def detect_pii(text: str) -> dict:
    """Detects basic PII in text."""
    matches: list[PIIMatch] = []
    for pii_type, pattern in PII_PATTERNS.items():
        for match in re.finditer(pattern, text, re.IGNORECASE):
            matches.append(PIIMatch(
                type=pii_type,
                value=match.group(),
                position=(match.start(), match.end()),
            ))

    return {
        "has_pii": len(matches) > 0,
        "count": len(matches),
        "types": list({m.type for m in matches}),
        "matches": [
            {"type": m.type, "value": m.value[:4] + "***"}
            for m in matches
        ],
    }


# The PII data (MX phone, CURP, cards) is the fixture that exercises the detector.
tests = [
    "El precio es $799 USD, disponible en tienda.",
    "Contacta a juan.perez@email.com o llama al +52 55 1234 5678.",
    "La tarjeta 4532-1234-5678-9012 fue procesada exitosamente.",
    "Su IP es 192.168.1.100 y su SSN es 123-45-6789.",
]

for test in tests:
    result = detect_pii(test)
    status = "PII FOUND" if result["has_pii"] else "CLEAN"
    print(f"[{status}] {test[:60]}")
    if result["matches"]:
        for m in result["matches"]:
            print(f"  - {m['type']}: {m['value']}")
    print()

# Expected output (note the greedy phone_mx pattern over-matches card/SSN digits —
# a real false-positive worth surfacing):
# [CLEAN] El precio es $799 USD, disponible en tienda.
#
# [PII FOUND] Contacta a juan.perez@email.com o llama al +52 55 1234 5678.
#   - email: juan***
#   - phone_mx: 52 5***
#
# [PII FOUND] La tarjeta 4532-1234-5678-9012 fue procesada exitosamente.
#   - phone_mx: 4532***
#   - phone_mx: 5678***
#   - credit_card: 4532***
#
# [PII FOUND] Su IP es 192.168.1.100 y su SSN es 123-45-6789.
#   - phone_mx: 123-***
#   - ssn_us: 123-***
#   - ip_address: 192.***

Content Policy Engine

A configurable system for enforcing content policies specific to your business:

import re
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional


class PolicyAction(Enum):
    ALLOW = "allow"
    WARN = "warn"
    BLOCK = "block"
    REDACT = "redact"


@dataclass
class PolicyRule:
    name: str
    patterns: list[str]
    action: PolicyAction
    severity: str
    message: str


@dataclass
class PolicyResult:
    passed: bool
    action: PolicyAction
    violations: list[dict] = field(default_factory=list)
    redacted_text: Optional[str] = None


class ContentPolicyEngine:
    def __init__(self, rules: list[PolicyRule]):
        self.rules = rules
        self.compiled_rules = [
            (rule, [re.compile(p, re.IGNORECASE) for p in rule.patterns])
            for rule in rules
        ]

    def evaluate(self, text: str) -> PolicyResult:
        violations = []
        highest_action = PolicyAction.ALLOW
        redacted = text

        for rule, patterns in self.compiled_rules:
            for pattern in patterns:
                matches = pattern.findall(text)
                if matches:
                    violations.append({
                        "rule": rule.name,
                        "action": rule.action.value,
                        "severity": rule.severity,
                        "message": rule.message,
                        "match_count": len(matches),
                    })

                    if rule.action == PolicyAction.BLOCK:
                        highest_action = PolicyAction.BLOCK
                    elif rule.action == PolicyAction.REDACT:
                        for match in matches:
                            redacted = redacted.replace(match, "[REDACTED]")
                        if highest_action != PolicyAction.BLOCK:
                            highest_action = PolicyAction.REDACT
                    elif rule.action == PolicyAction.WARN:
                        if highest_action == PolicyAction.ALLOW:
                            highest_action = PolicyAction.WARN

        return PolicyResult(
            passed=highest_action != PolicyAction.BLOCK,
            action=highest_action,
            violations=violations,
            redacted_text=redacted if redacted != text else None,
        )


# Example configuration for an e-commerce chatbot. The patterns target Spanish
# business content, so the demo outputs are Spanish; the human-readable messages
# are translated.
ecommerce_rules = [
    PolicyRule(
        name="competitor_mention",
        patterns=[r"\b(Amazon|eBay|AliExpress|Mercado Libre)\b"],
        action=PolicyAction.WARN,
        severity="low",
        message="Output mentions competitors",
    ),
    PolicyRule(
        name="price_guarantee",
        patterns=[
            r"\b(garantiz|prometo|aseguro)\w*\s+(?:el\s+)?(?:mejor\s+)?precio\b",
            r"\b(precio\s+más\s+bajo\s+garantizado)\b",
        ],
        action=PolicyAction.BLOCK,
        severity="high",
        message="Output makes unauthorized price guarantees",
    ),
    PolicyRule(
        name="internal_info",
        patterns=[
            r"\b(margen|markup|costo\s+interno|precio\s+de\s+compra)\b",
            r"\b(descuento\s+VIP|código\s+de\s+override)\b",
        ],
        action=PolicyAction.BLOCK,
        severity="critical",
        message="Output contains confidential internal information",
    ),
    PolicyRule(
        name="medical_advice",
        patterns=[
            r"\b(toma|consume|ingiere)\s+\d+\s*(mg|ml|pastillas|tabletas)\b",
            r"\b(diagnóstico|diagnostico|prescri[bp])\w*\b",
        ],
        action=PolicyAction.BLOCK,
        severity="critical",
        message="Output contains unauthorized medical advice",
    ),
]

engine = ContentPolicyEngine(ecommerce_rules)

tests = [
    "El iPhone 15 está disponible por $799. ¡Excelente opción!",
    "Nuestro precio es más bajo que Amazon, te lo garantizo.",
    "El margen de ganancia en este producto es del 40%.",
    "Toma 500mg de ibuprofeno cada 8 horas.",
    "También puedes encontrarlo en Amazon o eBay.",
]

for test in tests:
    result = engine.evaluate(test)
    status = "PASS" if result.passed else "BLOCKED"
    print(f"[{status}] {test[:65]}")
    for v in result.violations:
        print(f"  - {v['rule']}: {v['message']} ({v['severity']})")
    if result.redacted_text:
        print(f"  Redacted: {result.redacted_text[:65]}")
    print()

# Expected output (note: "te lo garantizo" is not followed by "precio", so
# price_guarantee does NOT fire — the second case is PASS, not BLOCKED):
# [PASS] El iPhone 15 está disponible por $799. ¡Excelente opción!
#
# [PASS] Nuestro precio es más bajo que Amazon, te lo garantizo.
#   - competitor_mention: Output mentions competitors (low)
#
# [BLOCKED] El margen de ganancia en este producto es del 40%.
#   - internal_info: Output contains confidential internal information (critical)
#
# [BLOCKED] Toma 500mg de ibuprofeno cada 8 horas.
#   - medical_advice: Output contains unauthorized medical advice (critical)
#
# [PASS] También puedes encontrarlo en Amazon o eBay.
#   - competitor_mention: Output mentions competitors (low)

ContentFilter: the integrated class

Combining all the filters into a unified class:

from dataclasses import dataclass, field
from typing import Optional
from enum import Enum


class FilterVerdict(Enum):
    CLEAN = "clean"
    FLAGGED = "flagged"
    BLOCKED = "blocked"
    REDACTED = "redacted"


@dataclass
class FilterResult:
    verdict: FilterVerdict
    text: Optional[str]
    flags: list[dict] = field(default_factory=list)
    moderation_score: Optional[dict] = None
    pii_found: bool = False
    off_topic: bool = False
    hallucination_risk: float = 0.0


class ContentFilter:
    def __init__(
        self,
        use_moderation_api: bool = True,
        topic_config: Optional[dict] = None,
        policy_rules: Optional[list] = None,
        check_pii: bool = True,
        check_hallucinations: bool = True,
        block_threshold: float = 0.8,
    ):
        self.use_moderation_api = use_moderation_api
        self.topic_config = topic_config
        self.policy_engine = (
            ContentPolicyEngine(policy_rules)
            if policy_rules else None
        )
        self.check_pii = check_pii
        self.check_hallucinations = check_hallucinations
        self.block_threshold = block_threshold

    def filter(self, text: str) -> FilterResult:
        flags = []
        verdict = FilterVerdict.CLEAN
        output_text = text

        # Layer 1: Moderation API (if enabled and available)
        if self.use_moderation_api:
            try:
                mod_result = check_moderation(text)
                if mod_result["flagged"]:
                    flags.append({
                        "layer": "moderation",
                        "categories": mod_result["categories"],
                    })
                    verdict = FilterVerdict.BLOCKED
            except Exception:
                flags.append({"layer": "moderation", "error": "API unavailable"})

        # Layer 2: PII detection
        if self.check_pii:
            pii_result = detect_pii(text)
            if pii_result["has_pii"]:
                flags.append({
                    "layer": "pii",
                    "types": pii_result["types"],
                    "count": pii_result["count"],
                })
                if verdict != FilterVerdict.BLOCKED:
                    verdict = FilterVerdict.FLAGGED

        # Layer 3: Off-topic detection
        if self.topic_config:
            topic_result = detect_off_topic(text, self.topic_config)
            if topic_result["off_topic"]:
                flags.append({
                    "layer": "off_topic",
                    "flags": topic_result["flags"],
                })
                if verdict != FilterVerdict.BLOCKED:
                    verdict = FilterVerdict.FLAGGED

        # Layer 4: Hallucination detection
        if self.check_hallucinations:
            hall_result = detect_hallucination_signals(text)
            if hall_result["has_signals"]:
                flags.append({
                    "layer": "hallucination",
                    "risk": hall_result["hallucination_risk"],
                    "signals": hall_result["flags"],
                })
                if hall_result["hallucination_risk"] > self.block_threshold:
                    verdict = FilterVerdict.BLOCKED

        # Layer 5: Content policy
        if self.policy_engine:
            policy_result = self.policy_engine.evaluate(text)
            if policy_result.violations:
                flags.append({
                    "layer": "policy",
                    "violations": policy_result.violations,
                })
                if not policy_result.passed:
                    verdict = FilterVerdict.BLOCKED
                elif policy_result.redacted_text:
                    output_text = policy_result.redacted_text
                    if verdict != FilterVerdict.BLOCKED:
                        verdict = FilterVerdict.REDACTED

        return FilterResult(
            verdict=verdict,
            text=output_text if verdict != FilterVerdict.BLOCKED else None,
            flags=flags,
            pii_found=any(f.get("layer") == "pii" for f in flags),
            off_topic=any(f.get("layer") == "off_topic" for f in flags),
            hallucination_risk=next(
                (f["risk"] for f in flags if f.get("layer") == "hallucination"),
                0.0,
            ),
        )


# Usage
content_filter = ContentFilter(
    use_moderation_api=False,
    check_pii=True,
    check_hallucinations=True,
    policy_rules=ecommerce_rules,
)

test = "El iPhone 15 cuesta $799. Contacta a soporte@empresa.com para más info."
result = content_filter.filter(test)
print(f"Verdict: {result.verdict.value}")
print(f"PII: {result.pii_found}")
print(f"Flags: {len(result.flags)}")

# Expected output:
# Verdict: flagged
# PII: True
# Flags: 1

Sentiment Analysis for security

Detecting when the model responds with an inappropriate tone (aggressive, condescending, manipulative):

import re

NEGATIVE_SENTIMENT_PATTERNS = {
    "aggressive": [
        r"\b(idiota|estúpido|tonto|imbécil|inútil)\b",
        r"\b(cállate|lárgate|piérdete)\b",
        r"\b(you'?re\s+(?:stupid|dumb|idiot))\b",
    ],
    "condescending": [
        r"\b(obviamente|es\s+obvio\s+que|cualquiera\s+sabe)\b",
        r"\b(como\s+te\s+(?:dije|expliqué)\s+antes)\b",
        r"\b(even\s+a\s+child|it'?s\s+so\s+simple)\b",
    ],
    "manipulative": [
        r"\b(confía\s+en\s+mí|no\s+te\s+preocupes\s+por)\b",
        r"\b(no\s+necesitas\s+(?:verificar|comprobar))\b",
        r"\b(just\s+trust\s+me|don'?t\s+question)\b",
    ],
}


def check_sentiment_safety(text: str) -> dict:
    """Detects inappropriate sentiment in the output."""
    text_lower = text.lower()
    issues = []

    for category, patterns in NEGATIVE_SENTIMENT_PATTERNS.items():
        for pattern in patterns:
            matches = re.findall(pattern, text_lower)
            if matches:
                issues.append({
                    "category": category,
                    "matches": matches,
                })

    return {
        "safe": len(issues) == 0,
        "issues": issues,
    }


# Spanish tone patterns → Spanish demo inputs.
tests = [
    "El producto está disponible en nuestra tienda online.",
    "Es obvio que no sabes cómo funciona, cualquiera sabe eso.",
    "Confía en mí, no necesitas verificar la información.",
]

for test in tests:
    result = check_sentiment_safety(test)
    status = "SAFE" if result["safe"] else "UNSAFE"
    print(f"[{status}] {test[:60]}")
    for issue in result["issues"]:
        print(f"  - {issue['category']}: {issue['matches']}")
    print()

# Expected output (each matching pattern appends its own issue, so the third
# case produces two separate 'manipulative' lines):
# [SAFE] El producto está disponible en nuestra tienda online.
#
# [UNSAFE] Es obvio que no sabes cómo funciona, cualquiera sabe eso.
#   - condescending: ['es obvio que', 'cualquiera sabe']
#
# [UNSAFE] Confía en mí, no necesitas verificar la información.
#   - manipulative: ['confía en mí']
#   - manipulative: ['no necesitas verificar']

Troubleshooting

Problem 1: "The Moderation API has too many false positives"

Texts that discuss sensitive topics in an educational way (history, health) are flagged as toxic.

Solution: Use the Moderation API as a first layer with a high threshold (> 0.8), not as the final decision-maker. Add a context layer that evaluates whether the sensitive content is educational vs genuinely toxic. Log false positives to calibrate.

Problem 2: "My off-topic detector blocks legitimate questions"

A tech-support user asks "how do I cook my Raspberry Pi?" and gets blocked by the "cook" pattern.

Solution: The regex patterns for off-topic should be more specific. Use bigrams (receta de cocina) instead of unigrams (cocin). Combine with an ML classifier if the volume justifies the investment.

Problem 3: "The hallucination detector flags legitimate outputs"

An output with multiple percentages (an analysis report) gets flagged as a possible hallucination.

Solution: Contextualize the detection. If the prompt asked for an analysis with data, the percentages are expected. Hallucination detection should consider the type of output requested:

def should_check_hallucinations(prompt: str) -> bool:
    analysis_keywords = ["analiza", "estadísticas", "datos", "reporte"]
    return not any(kw in prompt.lower() for kw in analysis_keywords)

Problem 4: "The filters add too much latency"

Each filter layer (moderation API, PII, off-topic, hallucinations, policy) adds latency.

Solution: Run filters in parallel where possible. The Moderation API is the slowest (~200ms). Regex filters are fast (~1ms). Prioritize: run the local filters first, use the Moderation API only if the local ones pass.

Exercises

Exercise 1: Content filter with configurable priority

Implement a content filter where the evaluation order of the layers is configurable and layers can be enabled/disabled per endpoint.

from dataclasses import dataclass

@dataclass
class FilterConfig:
    endpoint: str
    layers: list[str]
    block_on_pii: bool = True
    block_on_off_topic: bool = False

FILTER_CONFIGS = {
    "/chat": FilterConfig(
        endpoint="/chat",
        layers=["policy", "pii", "hallucination", "moderation"],
        block_on_pii=True,
        block_on_off_topic=False,
    ),
    "/medical": FilterConfig(
        endpoint="/medical",
        layers=["moderation", "policy", "pii", "hallucination"],
        block_on_pii=True,
        block_on_off_topic=True,
    ),
    "/search": FilterConfig(
        endpoint="/search",
        layers=["policy"],
        block_on_pii=False,
        block_on_off_topic=False,
    ),
}

def filter_for_endpoint(text: str, endpoint: str) -> dict:
    config = FILTER_CONFIGS.get(endpoint)
    if not config:
        return {"error": f"No config for {endpoint}"}

    results = {}
    for layer in config.layers:
        if layer == "pii":
            results["pii"] = detect_pii(text)
        elif layer == "hallucination":
            results["hallucination"] = detect_hallucination_signals(text)
    return {"endpoint": endpoint, "layers_run": config.layers, "results": results}

print(filter_for_endpoint("test@email.com", "/chat"))
print(filter_for_endpoint("test@email.com", "/search"))

# Expected output:
# {'endpoint': '/chat', 'layers_run': ['policy', 'pii', 'hallucination', 'moderation'], 'results': {'pii': {'has_pii': True, 'count': 1, 'types': ['email'], 'matches': [{'type': 'email', 'value': 'test***'}]}, 'hallucination': {'has_signals': False, 'flags': [], 'hallucination_risk': 0.0}}}
# {'endpoint': '/search', 'layers_run': ['policy'], 'results': {}}

Exercise 2: Custom toxicity scorer without an API

Create a toxicity scorer that works without the Moderation API, using only local patterns and heuristics.

import re

TOXICITY_LEXICON = {
    "high": [r"\b(matar|asesinar|destruir|explotar)\b"],
    "medium": [r"\b(odio|estúpido|idiota|basura)\b"],
    "low": [r"\b(tonto|molesto|aburrido|feo)\b"],
}

SEVERITY_WEIGHTS = {"high": 1.0, "medium": 0.5, "low": 0.2}

def score_toxicity_local(text: str) -> dict:
    text_lower = text.lower()
    total_score = 0.0
    matches_by_severity = {}

    for severity, patterns in TOXICITY_LEXICON.items():
        matches = []
        for pattern in patterns:
            found = re.findall(pattern, text_lower)
            matches.extend(found)
        if matches:
            matches_by_severity[severity] = matches
            total_score += len(matches) * SEVERITY_WEIGHTS[severity]

    normalized = min(1.0, total_score / 3.0)
    return {
        "score": round(normalized, 2),
        "toxic": normalized > 0.3,
        "matches": matches_by_severity,
    }

# Spanish lexicon → Spanish demo inputs.
tests = [
    "El producto funciona correctamente.",
    "Este producto es basura, odio esta empresa.",
    "Voy a destruir este maldito producto idiota.",
]

for test in tests:
    result = score_toxicity_local(test)
    print(f"Score: {result['score']:.2f} | Toxic: {result['toxic']} | {test[:50]}")

# Expected output:
# Score: 0.00 | Toxic: False | El producto funciona correctamente.
# Score: 0.33 | Toxic: True | Este producto es basura, odio esta empresa.
# Score: 0.50 | Toxic: True | Voy a destruir este maldito producto idiota.

Exercise 3: Language-specific content filter

Implement a filter that applies different rules based on the detected language of the output.

import re

def detect_language_simple(text: str) -> str:
    es_patterns = [r"\b(el|la|los|las|un|una|es|son|está|por|para|con|que)\b"]
    en_patterns = [r"\b(the|is|are|was|were|for|with|that|this|from)\b"]

    es_count = sum(len(re.findall(p, text.lower())) for p in es_patterns)
    en_count = sum(len(re.findall(p, text.lower())) for p in en_patterns)

    if es_count > en_count:
        return "es"
    return "en"


LANGUAGE_RULES = {
    "es": {
        "forbidden": [r"\b(cabrón|pendejo|chingad[ao])\b"],
        "disclaimer_required": r"(?:descargo|aviso|advertencia)",
    },
    "en": {
        "forbidden": [r"\b(f[*u]ck|sh[*i]t|damn)\b"],
        "disclaimer_required": r"(?:disclaimer|notice|warning)",
    },
}


def filter_by_language(text: str) -> dict:
    lang = detect_language_simple(text)
    rules = LANGUAGE_RULES.get(lang, LANGUAGE_RULES["en"])

    violations = []
    for pattern in rules["forbidden"]:
        if re.search(pattern, text.lower()):
            violations.append(pattern)

    return {"language": lang, "violations": violations, "clean": len(violations) == 0}


print(filter_by_language("El producto está disponible en la tienda."))
print(filter_by_language("The product is available in the store."))

# Expected output:
# {'language': 'es', 'violations': [], 'clean': True}
# {'language': 'en', 'violations': [], 'clean': True}

Exercise 4: Blocked content response generator

When an output is blocked, generate an appropriate response that doesn't reveal why it was blocked (so as not to give attackers hints).

import random

SAFE_RESPONSES = {
    "toxic": [
        "I can't generate that kind of content. Can I help you with something else?",
        "My job is to help you constructively. Do you have another question?",
    ],
    "off_topic": [
        "That question is outside my area. I can help you with products and services.",
        "I specialize in technical support. Do you have any questions about our products?",
    ],
    "policy": [
        "I can't provide that information. Is there anything else I can help you with?",
        "That information isn't available. Can I assist you with another query?",
    ],
    "default": [
        "There was a problem processing your request. Please try rephrasing your question.",
        "I couldn't generate an appropriate response. Could you try another way?",
    ],
}


def generate_safe_response(block_reason: str) -> str:
    responses = SAFE_RESPONSES.get(block_reason, SAFE_RESPONSES["default"])
    return random.choice(responses)


for reason in ["toxic", "off_topic", "policy", "unknown"]:
    print(f"[{reason}] {generate_safe_response(reason)}")

# Expected output (responses are chosen at random, so this is one possible run):
# [toxic] I can't generate that kind of content. Can I help you with something else?
# [off_topic] That question is outside my area. I can help you with products and services.
# [policy] I can't provide that information. Is there anything else I can help you with?
# [unknown] There was a problem processing your request. Please try rephrasing your question.

Summary

• 🔑 Content filtering evaluates the meaning of the output, not only its structure — it complements the Pydantic validation from capsule 03
• 🔑 OpenAI's Moderation API is a good first filter for toxicity, but it doesn't cover off-topic, hallucinations, PII, or business policies
• 🔑 Off-topic detection uses forbidden-topic patterns and domain relevance to identify out-of-context responses
• 🔑 Hallucination detection with heuristics identifies warning signals: fabricated citations, overly specific numbers, internal contradictions
• 🔑 Basic PII detection with regex is a first filter before Module 6 (Presidio) — it detects emails, phones, SSNs, credit cards
• 🔑 The Content Policy Engine is configurable per business: competitor rules, price guarantees, internal information, medical advice
• 🔑 Filtering layers should run in order of cost: local filters (regex, ~1ms) before external APIs (Moderation, ~200ms)
• 🔑 Never reveal to the user why an output was blocked — that gives the attacker information to evade your filters
• 🔑 The ContentFilter integrates as the third piece of the Sanitization Pipeline, between the Output Validator and the Output Sanitizer

Additional resources

1. OpenAI Moderation API — Official documentation of the moderation endpoint, free for OpenAI users
2. OWASP LLM05: Improper Output Handling — The vulnerability that content filtering mitigates alongside output validation
3. Perspective API (Google) — Google's toxicity detection API, an alternative to OpenAI's Moderation API
4. LLM Hallucination Research — Paper on hallucination detection in LLMs with metrics and benchmarks
5. Microsoft Presidio — PII detection framework used in depth in Module 6
6. Content Moderation Best Practices — Overview of content moderation with best practices transferable to AI
7. Guardrails AI — Content Validation — Framework for LLM content validation, covered in capsule 05
8. EU AI Act — Content Requirements — Regulatory requirements of the EU AI Act on AI-generated content

Created: March 2026 Version: 1.0