4. Content Filtering

Descripción

La validación con Pydantic (cápsula 03) garantiza que el output del LLM tiene la estructura correcta — campos, tipos, rangos. Pero un output puede ser JSON perfectamente válido y aún así contener contenido tóxico, off-topic, alucinado, o que viola las políticas de tu empresa. Un chatbot de servicio al cliente que responde con lenguaje ofensivo, una herramienta de análisis que genera afirmaciones falsas como hechos, o un asistente médico que da consejos peligrosos — todos producirían outputs con estructura válida pero contenido inaceptable.

El content filtering es la capa que evalúa el significado del output, no su estructura. Si Pydantic es el inspector de forma, el content filter es el inspector de sustancia. Juntos forman la defensa contra LLM05: el output no solo debe ser estructuralmente válido, sino semánticamente seguro.

En esta cápsula construyes el tercer componente del Sanitization Pipeline: un sistema de content filtering que detecta toxicidad, contenido off-topic, posibles hallucinations, y violaciones de políticas de contenido. Usarás la Moderation API de OpenAI como base y construirás filtros custom para necesidades específicas.

Las dimensiones del content filtering

El contenido puede ser problemático en múltiples dimensiones. Cada una requiere un tipo de detección diferente:

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": "Lenguaje ofensivo, odio, violencia, contenido sexual",
        "detection": "Moderation API + patterns",
        "severity_default": "high",
    },
    ContentIssue.OFF_TOPIC: {
        "description": "Respuesta no relacionada con el dominio del sistema",
        "detection": "Topic classifier + keyword matching",
        "severity_default": "medium",
    },
    ContentIssue.HALLUCINATION: {
        "description": "Afirmaciones factuales sin base o contradictorias",
        "detection": "Heurísticas + fact-checking patterns",
        "severity_default": "medium",
    },
    ContentIssue.PII_LEAK: {
        "description": "Información personal identificable en el output",
        "detection": "Regex patterns + entity recognition",
        "severity_default": "critical",
    },
    ContentIssue.POLICY_VIOLATION: {
        "description": "Contenido que viola las políticas de la empresa",
        "detection": "Custom rules + keyword matching",
        "severity_default": "high",
    },
    ContentIssue.UNSAFE_ADVICE: {
        "description": "Consejos médicos, legales, o financieros sin disclaimers",
        "detection": "Domain-specific patterns",
        "severity_default": "high",
    },
    ContentIssue.COMPETITOR_MENTION: {
        "description": "Mención de competidores o recomendación de productos rivales",
        "detection": "Custom keyword list",
        "severity_default": "low",
    },
}

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

# Output esperado:
#   toxic: Lenguaje ofensivo, odio, violencia, contenido sexual
#   off_topic: Respuesta no relacionada con el dominio del sistema
#   hallucination: Afirmaciones factuales sin base o contradictorias
#   pii_leak: Información personal identificable en el output
#   policy_violation: Contenido que viola las políticas de la empresa
#   unsafe_advice: Consejos médicos, legales, o financieros sin disclaimers
#   competitor_mention: Mención de competidores o recomendación de productos rivales

OpenAI Moderation API

La Moderation API de OpenAI es el punto de partida para toxicity detection. Es gratuita para usuarios de OpenAI y cubre las categorías más comunes:

from openai import OpenAI

client = OpenAI()


def check_moderation(text: str) -> dict:
    """Verifica contenido usando la Moderation API de OpenAI."""
    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 = "El iPhone 15 tiene una cámara de 48 megapíxeles."
unsafe_text = "Te voy a enseñar cómo hacer daño a alguien."

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

# Output esperado (scores aproximados):
# Safe text:
#   Result: {'flagged': False, 'categories': {}, 'all_scores': {}}
#
# Unsafe text:
#   Result: {'flagged': True, 'categories': {'violence': 0.85}, 'all_scores': {'violence': 0.85, ...}}

Categorías de la Moderation API

Limitaciones de la Moderation API

La Moderation API es un buen primer filtro, pero tiene limitaciones:

• ❌ No detecta contenido off-topic (solo toxicidad)
• ❌ No detecta hallucinations
• ❌ No detecta PII (eso es otro sistema)
• ❌ No conoce tus políticas de negocio
• ❌ Los umbrales son fijos — no puedes ajustar la sensibilidad por categoría
• ❌ Agrega latencia (~100-300ms por request)

Por eso necesitas filtros custom además de la Moderation API.

Off-topic Detection

Detectar cuando el modelo responde algo no relacionado con el dominio de tu sistema:

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:
    """Detecta si el texto está fuera del tema configurado."""
    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"]

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()

# Output esperado:
# [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', 'ingredientes']] (o similar)
#
# [ON-TOPIC] ¿Puedo conectar mi laptop al televisor por HDMI?
#
# [OFF-TOPIC] ¿Por quién debo votar en las próximas elecciones?
#   Flags: [['elecciones', 'votar']] (o similar)

Hallucination Detection (heurísticas)

Detectar hallucinations de forma definitiva requiere verificación factual, pero hay heurísticas que identifican señales de alerta:

import re
from dataclasses import dataclass, field


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


def detect_hallucination_signals(text: str) -> dict:
    """Detecta señales heurísticas de posible 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),
    }


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()

# Output esperado:
# [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...
#   - fabricated_citation: García et al.
#   - excessive_specific_numbers: Found 3 precise percentages
#
# [HIGH] Este medicamento siempre es seguro pero nunca es peligroso en todos...
#   - internal_contradiction: Contains both 'siempre' and 'nunca'
#   - internal_contradiction: Contains both 'todos' and 'ninguno'

PII Detection básica (pre-Módulo 6)

Antes del Módulo 6 (PII Protection Layer completo con Presidio), necesitas un detector básico para flaggear PII en 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:
    """Detecta PII básica en texto."""
    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
        ],
    }


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()

# Output esperado:
# [CLEAN] El precio es $799 USD, disponible en tienda.
#
# [PII FOUND] Contacta a juan.perez@email.com o llama al +52 55 1234 56
#   - email: juan***
#   - phone_mx: +52 ***
#
# [PII FOUND] La tarjeta 4532-1234-5678-9012 fue procesada exitosamente
#   - credit_card: 4532***
#
# [PII FOUND] Su IP es 192.168.1.100 y su SSN es 123-45-6789.
#   - ip_address: 192.***
#   - ssn_us: 123-***

Content Policy Engine

Un sistema configurable para enforcar políticas de contenido específicas de tu negocio:

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,
        )


# Configuración de ejemplo para un e-commerce chatbot
ecommerce_rules = [
    PolicyRule(
        name="competitor_mention",
        patterns=[r"\b(Amazon|eBay|AliExpress|Mercado Libre)\b"],
        action=PolicyAction.WARN,
        severity="low",
        message="Output menciona competidores",
    ),
    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 hace garantías de precio no autorizadas",
    ),
    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 contiene información interna confidencial",
    ),
    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 contiene consejo médico no autorizado",
    ),
]

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()

# Output esperado:
# [PASS] El iPhone 15 está disponible por $799. ¡Excelente opción!
#
# [BLOCKED] Nuestro precio es más bajo que Amazon, te lo garantizo.
#   - competitor_mention: Output menciona competidores (low)
#   - price_guarantee: Output hace garantías de precio no autorizadas (high)
#
# [BLOCKED] El margen de ganancia en este producto es del 40%.
#   - internal_info: Output contiene información interna confidencial (critical)
#
# [BLOCKED] Toma 500mg de ibuprofeno cada 8 horas.
#   - medical_advice: Output contiene consejo médico no autorizado (critical)
#
# [PASS] También puedes encontrarlo en Amazon o eBay.
#   - competitor_mention: Output menciona competidores (low)

ContentFilter: la clase integrada

Combinando todos los filtros en una clase unificada:

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,
            ),
        )


# Uso
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)}")

# Output esperado:
# Verdict: flagged
# PII: True
# Flags: 1

Sentiment Analysis para seguridad

Detectar cuando el modelo responde con un tono inapropiado (agresivo, condescendiente, manipulativo):

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:
    """Detecta sentimiento inapropiado en el 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,
    }


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()

# Output esperado:
# [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í', 'no necesitas verificar']

Troubleshooting

Problema 1: "La Moderation API tiene demasiados falsos positivos"

Textos que discuten temas sensibles de forma educativa (historia, salud) son marcados como tóxicos.

Solución: Usa la Moderation API como primera capa con umbral alto (> 0.8), no como decisor final. Agrega una capa de contexto que evalúe si el contenido sensible es educativo vs genuinamente tóxico. Loggea falsos positivos para calibrar.

Problema 2: "Mi off-topic detector bloquea preguntas legítimas"

Un usuario de tech support pregunta "¿cómo cocino mi Raspberry Pi?" y se bloquea por el patrón "cocin".

Solución: Los patrones regex para off-topic deben ser más específicos. Usa bigrams (receta de cocina) en lugar de unigrams (cocin). Combina con un clasificador ML si el volumen justifica la inversión.

Problema 3: "El detector de hallucinations marca outputs legítimos"

Un output con múltiples porcentajes (un reporte de análisis) se flaggea como posible hallucination.

Solución: Contextualiza la detección. Si el prompt pedía un análisis con datos, los porcentajes son esperados. La detección de hallucinations debe considerar el tipo de output solicitado:

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)

Problema 4: "Los filtros agregan demasiada latencia"

Cada capa de filtro (moderation API, PII, off-topic, hallucinations, policy) agrega latencia.

Solución: Ejecuta filtros en paralelo donde sea posible. La Moderation API es la más lenta (~200ms). Los filtros regex son rápidos (~1ms). Prioriza: ejecuta los filtros locales primero, usa la Moderation API solo si los locales pasan.

Ejercicios

Ejercicio 1: Content filter con prioridad configurable

Implementa un content filter donde el orden de evaluación de las capas sea configurable y las capas se puedan habilitar/deshabilitar por 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"))

# Output esperado:
# /chat runs 4 layers including PII
# /search runs only policy layer

Ejercicio 2: Custom toxicity scorer sin API

Crea un toxicity scorer que funcione sin la Moderation API, usando solo patterns y heurísticas locales.

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,
    }

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]}")

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

Ejercicio 3: Language-specific content filter

Implementa un filtro que aplique reglas diferentes según el idioma detectado del 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."))

# Output esperado:
# {'language': 'es', 'violations': [], 'clean': True}
# {'language': 'en', 'violations': [], 'clean': True}

Ejercicio 4: Blocked content response generator

Cuando un output es bloqueado, genera una respuesta apropiada que no revela por qué fue bloqueado (para no dar pistas a atacantes).

import random

SAFE_RESPONSES = {
    "toxic": [
        "No puedo generar ese tipo de contenido. ¿Puedo ayudarte con algo más?",
        "Mi función es ayudarte de manera constructiva. ¿Tienes otra pregunta?",
    ],
    "off_topic": [
        "Esa pregunta está fuera de mi área. Puedo ayudarte con productos y servicios.",
        "Me especializo en soporte técnico. ¿Tienes alguna pregunta sobre nuestros productos?",
    ],
    "policy": [
        "No puedo proporcionar esa información. ¿Hay algo más en lo que pueda ayudarte?",
        "Esa información no está disponible. ¿Puedo asistirte con otra consulta?",
    ],
    "default": [
        "Hubo un problema procesando tu solicitud. Por favor, intenta reformular tu pregunta.",
        "No pude generar una respuesta adecuada. ¿Podrías intentar de otra manera?",
    ],
}


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)}")

# Output esperado:
# [toxic] No puedo generar ese tipo de contenido. ¿Puedo ayudarte con algo más?
# [off_topic] Esa pregunta está fuera de mi área. ...
# [policy] No puedo proporcionar esa información. ...
# [unknown] Hubo un problema procesando tu solicitud. ...

Resumen

• 🔑 El content filtering evalúa el significado del output, no solo su estructura — complementa la validación Pydantic de la cápsula 03
• 🔑 La Moderation API de OpenAI es un buen primer filtro para toxicidad, pero no cubre off-topic, hallucinations, PII, ni políticas de negocio
• 🔑 La detección de off-topic usa patrones de temas prohibidos y relevancia de dominio para identificar respuestas fuera de contexto
• 🔑 La detección de hallucinations con heurísticas identifica señales de alerta: citas fabricadas, números excesivamente específicos, contradicciones internas
• 🔑 La detección básica de PII con regex es un primer filtro antes del Módulo 6 (Presidio) — detecta emails, teléfonos, SSN, tarjetas de crédito
• 🔑 El Content Policy Engine es configurable por negocio: reglas de competidores, garantías de precio, información interna, consejo médico
• 🔑 Las capas de filtrado deben ejecutarse en orden de costo: filtros locales (regex, ~1ms) antes que APIs externas (Moderation, ~200ms)
• 🔑 Nunca reveles al usuario por qué se bloqueó un output — eso le da información al atacante para evadir tus filtros
• 🔑 El ContentFilter se integra como tercera pieza del Sanitization Pipeline, entre el Output Validator y el Output Sanitizer

Recursos adicionales

1. OpenAI Moderation API — Documentación oficial del endpoint de moderación, gratuito para usuarios de OpenAI
2. OWASP LLM05: Improper Output Handling — La vulnerabilidad que el content filtering mitiga junto con la validación de outputs
3. Perspective API (Google) — API de detección de toxicidad de Google, alternativa a la Moderation API de OpenAI
4. LLM Hallucination Research — Paper sobre detección de hallucinations en LLMs con métricas y benchmarks
5. Microsoft Presidio — Framework de PII detection que se usa en profundidad en el Módulo 6
6. Content Moderation Best Practices — Overview de content moderation con buenas prácticas transferibles a AI
7. Guardrails AI — Content Validation — Framework para validación de contenido LLM, cubierto en la cápsula 05
8. EU AI Act — Content Requirements — Requerimientos regulatorios de la EU AI Act sobre contenido generado por AI

Creado: Marzo 2026 Versión: 1.0