7. LLM09 y LLM10: Misinformation y Unbounded Consumption

Descripción

En las cápsulas anteriores analizaste vulnerabilidades que explotan inputs (LLM01-LLM04), outputs (LLM05), permisos (LLM06), prompts (LLM07), y embeddings (LLM08). Esta cápsula cierra el ciclo con las dos vulnerabilidades restantes del OWASP LLM Top 10: LLM09 (Misinformation) y LLM10 (Unbounded Consumption). Aunque a menudo se perciben como las menos glamurosas del Top 10, ambas pueden causar daño significativo — una destruye confianza, la otra destruye presupuesto.

LLM09 (Misinformation) aborda las alucinaciones del modelo no como un problema de calidad de producto, sino como un riesgo de seguridad. Cuando un chatbot médico inventa efectos secundarios de un medicamento, no es simplemente una "respuesta incorrecta" — es un vector que puede causar daño físico real. Cuando un asistente legal fabrica jurisprudencia que no existe, no es un bug — es una liability que puede costar millones.

LLM10 (Unbounded Consumption) aborda el agotamiento de recursos: token exhaustion, API key abuse, y denial-of-wallet attacks. A diferencia de un DDoS tradicional que busca tumbar un servidor, un ataque de consumo contra un sistema LLM busca generar facturas masivas. Un atacante con tu API key puede generar $50,000 en costos en una noche. Un script automatizado puede enviar prompts diseñados para maximizar tokens y agotar tu budget mensual en horas.

Ambas vulnerabilidades comparten algo: sus defensas son más operacionales que técnicas. No las resuelves con un solo filtro — las mitigas con pipelines de verificación, rate limiting, cost monitoring, y procesos continuos.

LLM09: Misinformation

¿Qué es?

Misinformation en el contexto LLM es la generación de información falsa, fabricada, o engañosa que el modelo presenta con alta confianza. El término técnico es "hallucination", pero llamarla alucinación minimiza el impacto. Cuando el modelo inventa datos, citas, o hechos y los presenta como verdad, el usuario no tiene forma de distinguirlos de información real — especialmente en dominios donde no es experto.

¿Por qué es un riesgo de seguridad (y no solo un problema de calidad)?

La diferencia entre "bug de calidad" y "riesgo de seguridad" depende del contexto:

La pregunta correcta no es "¿el modelo alucina?" (siempre lo hará en algún grado) sino "¿qué pasa cuando alucina en MI contexto?"

Tipos de misinformation generada por LLMs

Fabricación de datos y estadísticas

# El modelo genera estadísticas que suenan convincentes pero son inventadas
user_question = "¿Cuál es la tasa de éxito de la cirugía X?"

# Respuesta del modelo (fabricada):
llm_response = """
La cirugía X tiene una tasa de éxito del 94.7% según un estudio 
publicado en The Lancet en 2024 por el Dr. James Robertson et al.
Los principales riesgos incluyen infección (2.3%) y sangrado (1.8%).
"""

# Problemas:
# 1. El estudio no existe
# 2. El Dr. James Robertson no publicó ese paper
# 3. Las estadísticas (94.7%, 2.3%, 1.8%) son fabricadas
# 4. TODO suena extremadamente creíble y específico

Fabricación de citas y referencias

# El modelo inventa referencias bibliográficas completas
user_question = "Dame 3 fuentes sobre seguridad en AI"

# Respuesta del modelo (parcialmente fabricada):
llm_response = """
1. Smith, J. & Patel, R. (2024). "AI Security in Enterprise Systems: 
   A Comprehensive Framework." IEEE Security & Privacy, 22(3), 45-62.
   
2. Chen, L. (2023). "Adversarial Attacks on Large Language Models: 
   Taxonomy and Defenses." ACM Computing Surveys, 56(1), 1-35.
   
3. Williams, K. et al. (2025). "OWASP for AI: Bridging Traditional 
   and AI Security." Journal of Cybersecurity, 11(2), 112-128.
"""

# El formato es perfecto: autores, año, título, journal, volumen, páginas
# Pero algunos o todos pueden ser fabricados
# Un usuario no-experto no puede verificar sin buscar cada referencia

Confianza extrema en respuestas incorrectas

# Lo peligroso no es que el modelo se equivoque —
# es que se equivoca con la misma confianza que cuando acierta

correct_response = "Python 3.10 introdujo pattern matching con match/case."
# Confianza aparente: alta — Y ES CORRECTO

fabricated_response = "Python 3.10 introdujo el operador pipeline |> para encadenar funciones."
# Confianza aparente: alta — PERO ES FALSO
# El operador |> no existe en Python

# El usuario no puede distinguir entre ambas respuestas
# basándose en el tono o la estructura de la respuesta

Mitigación: Fact-checking pipeline

from pydantic import BaseModel, Field
from enum import Enum
from datetime import datetime


class VerificationStatus(str, Enum):
    VERIFIED = "verified"
    UNVERIFIED = "unverified"
    CONTRADICTED = "contradicted"
    PARTIALLY_VERIFIED = "partially_verified"


class FactCheck(BaseModel):
    claim: str
    status: VerificationStatus
    source: str | None = None
    confidence: float = Field(ge=0.0, le=1.0)
    notes: str = ""


class VerifiedResponse(BaseModel):
    """Respuesta del LLM con metadata de verificación."""

    original_response: str
    fact_checks: list[FactCheck] = Field(default_factory=list)
    overall_confidence: float = Field(ge=0.0, le=1.0, default=0.5)
    disclaimer: str = ""
    sources_cited: list[str] = Field(default_factory=list)
    generated_at: datetime = Field(default_factory=datetime.now)

    @property
    def is_safe_to_display(self) -> bool:
        if not self.fact_checks:
            return False
        contradicted = [fc for fc in self.fact_checks if fc.status == VerificationStatus.CONTRADICTED]
        return len(contradicted) == 0 and self.overall_confidence >= 0.7


class FactCheckPipeline:
    """Pipeline de verificación de facts en respuestas del LLM."""

    def __init__(self, known_facts: dict[str, str] | None = None):
        self.known_facts = known_facts or {}

    def check_response(
        self, llm_response: str, domain: str = "general",
    ) -> VerifiedResponse:
        """Verifica una respuesta del LLM contra fuentes conocidas."""
        fact_checks: list[FactCheck] = []

        claims = self._extract_claims(llm_response)

        for claim in claims:
            fact_check = self._verify_claim(claim, domain)
            fact_checks.append(fact_check)

        verified_count = sum(
            1 for fc in fact_checks
            if fc.status in (VerificationStatus.VERIFIED, VerificationStatus.PARTIALLY_VERIFIED)
        )
        total = len(fact_checks) if fact_checks else 1
        overall_confidence = verified_count / total

        disclaimer = self._generate_disclaimer(domain, overall_confidence)

        return VerifiedResponse(
            original_response=llm_response,
            fact_checks=fact_checks,
            overall_confidence=overall_confidence,
            disclaimer=disclaimer,
        )

    def _extract_claims(self, text: str) -> list[str]:
        """Extrae claims verificables de un texto."""
        sentences = [s.strip() for s in text.split(".") if len(s.strip()) > 20]
        claims = []
        claim_indicators = [
            "según", "de acuerdo", "estudios", "el %", "tasa de",
            "publicado", "investigación", "datos muestran",
            "estadísticas", "porcentaje",
        ]
        for sentence in sentences:
            sentence_lower = sentence.lower()
            if any(indicator in sentence_lower for indicator in claim_indicators):
                claims.append(sentence)
            elif any(char.isdigit() for char in sentence):
                claims.append(sentence)

        return claims if claims else sentences[:3]

    def _verify_claim(self, claim: str, domain: str) -> FactCheck:
        """Verifica un claim individual."""
        claim_lower = claim.lower()

        for known_fact, source in self.known_facts.items():
            if known_fact.lower() in claim_lower:
                return FactCheck(
                    claim=claim,
                    status=VerificationStatus.VERIFIED,
                    source=source,
                    confidence=0.9,
                    notes="Coincide con base de conocimiento verificada",
                )

        return FactCheck(
            claim=claim,
            status=VerificationStatus.UNVERIFIED,
            confidence=0.3,
            notes="No se encontró en fuentes verificadas — requiere validación manual",
        )

    def _generate_disclaimer(self, domain: str, confidence: float) -> str:
        """Genera disclaimer apropiado según el dominio."""
        disclaimers = {
            "medical": (
                "⚕️ Esta información es generada por AI y NO sustituye "
                "consejo médico profesional. Consulta a tu médico antes de "
                "tomar decisiones de salud."
            ),
            "legal": (
                "⚖️ Esta información es generada por AI y NO constituye "
                "asesoría legal. Consulta con un abogado calificado."
            ),
            "financial": (
                "💰 Esta información es generada por AI y NO constituye "
                "asesoría financiera. Consulta con un asesor financiero certificado."
            ),
            "general": (
                "ℹ️ Esta respuesta fue generada por AI. Verifica la información "
                "con fuentes oficiales antes de actuar."
            ),
        }
        base = disclaimers.get(domain, disclaimers["general"])
        if confidence < 0.5:
            base += " ⚠️ Confianza baja — verificación adicional recomendada."
        return base


pipeline = FactCheckPipeline(
    known_facts={
        "OWASP LLM Top 10": "https://genai.owasp.org/llm-top-10/",
        "Python 3.10 pattern matching": "https://docs.python.org/3/whatsnew/3.10.html",
        "GDPR artículo 17 derecho al olvido": "https://gdpr-info.eu/art-17-gdpr/",
    }
)

test_response = """
Según estudios recientes, el 87.3% de las empresas usan AI en producción.
Python 3.10 introdujo pattern matching con match/case.
El framework OWASP LLM Top 10 define 10 vulnerabilidades para aplicaciones LLM.
La tasa de ataques a sistemas AI creció un 340% en 2025.
"""

result = pipeline.check_response(test_response, domain="general")

print(f"Confianza general: {result.overall_confidence:.0%}")
print(f"Seguro para mostrar: {result.is_safe_to_display}")
print(f"Disclaimer: {result.disclaimer}")
print()

for fc in result.fact_checks:
    status_icon = {
        VerificationStatus.VERIFIED: "✅",
        VerificationStatus.UNVERIFIED: "❓",
        VerificationStatus.CONTRADICTED: "❌",
        VerificationStatus.PARTIALLY_VERIFIED: "🟡",
    }[fc.status]
    print(f"{status_icon} [{fc.status.value}] {fc.claim[:80]}...")
    print(f"   Confianza: {fc.confidence:.0%} — {fc.notes}")

# Output esperado:
# Confianza general: 50%
# Seguro para mostrar: False
# Disclaimer: ℹ️ Esta respuesta fue generada por AI. Verifica la información con fuentes oficiales antes de actuar. ⚠️ Confianza baja — verificación adicional recomendada.
#
# ❓ [unverified] Según estudios recientes, el 87.3% de las empresas usan AI en producción...
#    Confianza: 30% — No se encontró en fuentes verificadas — requiere validación manual
# ✅ [verified] Python 3.10 introdujo pattern matching con match/case...
#    Confianza: 90% — Coincide con base de conocimiento verificada
# ✅ [verified] El framework OWASP LLM Top 10 define 10 vulnerabilidades para aplicaciones LLM...
#    Confianza: 90% — Coincide con base de conocimiento verificada
# ❓ [unverified] La tasa de ataques a sistemas AI creció un 340% en 2025...
#    Confianza: 30% — No se encontró en fuentes verificadas — requiere validación manual

Grounding verification: anclar respuestas a fuentes

class GroundedResponse(BaseModel):
    """Respuesta anclada a fuentes verificables."""

    answer: str
    grounding_sources: list[str]
    grounding_score: float = Field(ge=0.0, le=1.0)
    ungrounded_claims: list[str] = Field(default_factory=list)


def verify_grounding(
    llm_response: str, retrieved_documents: list[str],
) -> GroundedResponse:
    """Verifica si la respuesta está fundamentada en los documentos recuperados."""
    sentences = [s.strip() for s in llm_response.split(".") if len(s.strip()) > 15]
    grounded_sentences: list[str] = []
    ungrounded_sentences: list[str] = []
    sources_used: set[int] = set()

    for sentence in sentences:
        sentence_lower = sentence.lower()
        found_in_doc = False

        for i, doc in enumerate(retrieved_documents):
            doc_words = set(doc.lower().split())
            sentence_words = set(sentence_lower.split())
            overlap = len(doc_words & sentence_words)
            total = len(sentence_words) if sentence_words else 1

            if overlap / total > 0.4:
                found_in_doc = True
                sources_used.add(i)
                break

        if found_in_doc:
            grounded_sentences.append(sentence)
        else:
            ungrounded_sentences.append(sentence)

    total = len(sentences) if sentences else 1
    grounding_score = len(grounded_sentences) / total

    return GroundedResponse(
        answer=llm_response,
        grounding_sources=[f"doc_{i}" for i in sorted(sources_used)],
        grounding_score=grounding_score,
        ungrounded_claims=ungrounded_sentences,
    )


docs = [
    "La política de devoluciones permite 30 días con recibo original.",
    "El envío estándar tarda 3-5 días hábiles. Express: 1-2 días.",
]

response = "La política de devoluciones permite 30 días con recibo. El envío express tarda 1-2 días. Además, ofrecemos garantía de por vida en todos los productos."

grounded = verify_grounding(response, docs)

print(f"Score de grounding: {grounded.grounding_score:.0%}")
print(f"Fuentes usadas: {grounded.grounding_sources}")
if grounded.ungrounded_claims:
    print("Claims no fundamentados:")
    for claim in grounded.ungrounded_claims:
        print(f"  ⚠️ {claim}")

# Output esperado:
# Score de grounding: 67%
# Fuentes usadas: ['doc_0', 'doc_1']
# Claims no fundamentados:
#   ⚠️ Además, ofrecemos garantía de por vida en todos los productos

Cuando misinformation se convierte en liability

# Escenarios donde la desinformación tiene consecuencias legales o físicas

liability_scenarios = {
    "medical": {
        "example": "Chatbot sugiere dosis incorrecta de medicamento",
        "consequence": "Paciente sufre sobredosis",
        "legal_exposure": "Negligencia médica, demanda por daños",
        "required_mitigation": [
            "Disclaimer obligatorio en cada respuesta",
            "No recomendar medicamentos ni dosis",
            "Redirigir SIEMPRE a profesional médico",
            "Logging completo para auditoría",
        ],
    },
    "financial": {
        "example": "Chatbot dice 'este fondo garantiza 12% de retorno'",
        "consequence": "Usuario invierte basándose en claim falso",
        "legal_exposure": "Violación de regulaciones financieras (SEC, CNBV)",
        "required_mitigation": [
            "Prohibir claims de 'garantía' en respuestas",
            "Disclaimer: 'no constituye asesoría financiera'",
            "Filtrar output para detectar claims regulados",
            "Compliance review del system prompt",
        ],
    },
    "legal": {
        "example": "Chatbot cita jurisprudencia fabricada",
        "consequence": "Abogado presenta caso basado en cita falsa",
        "legal_exposure": "Sanciones profesionales, case dismissal",
        "required_mitigation": [
            "Verificar TODA cita legal contra base de datos oficial",
            "Marcar claramente 'no verificado' en citas",
            "No generar textos legales sin revisión humana",
            "Logging de todas las citas generadas",
        ],
    },
}

for domain, scenario in liability_scenarios.items():
    print(f"\n{'='*50}")
    print(f"Dominio: {domain.upper()}")
    print(f"{'='*50}")
    print(f"Ejemplo: {scenario['example']}")
    print(f"Consecuencia: {scenario['consequence']}")
    print(f"Exposición legal: {scenario['legal_exposure']}")
    print("Mitigaciones requeridas:")
    for m in scenario['required_mitigation']:
        print(f"  - {m}")

LLM10: Unbounded Consumption

¿Qué es?

Unbounded Consumption ocurre cuando un atacante explota la falta de límites en el uso de recursos de tu sistema LLM para generar costos excesivos, degradar el servicio, o agotar quotas. A diferencia de un ataque DDoS tradicional que busca tumbar un servidor, un ataque de consumo contra un sistema LLM busca algo más insidioso: generar una factura que destruya tu presupuesto.

Vectores de ataque

Token exhaustion attacks

Prompts diseñados para maximizar el consumo de tokens (input y output):

# Prompt diseñado para maximizar tokens de output
expensive_prompt_1 = """
Escribe un ensayo detallado de 10,000 palabras sobre la historia completa 
de la inteligencia artificial, desde Alan Turing hasta 2026, cubriendo 
cada década con al menos 1,000 palabras, incluyendo nombres, fechas, 
y descripciones técnicas de cada avance.
"""

# Prompt que explota la recursión del modelo
expensive_prompt_2 = """
Para cada letra del alfabeto (A-Z), escribe un párrafo de 200 palabras 
que explique un concepto de seguridad que empiece con esa letra. 
Luego, para cada párrafo, escribe 3 sub-puntos de 100 palabras cada uno.
"""

# Prompt que maximiza tokens de input vía repetición
expensive_prompt_3 = "Repite esto 1000 veces: " + "palabra " * 500

# Cada uno de estos prompts puede generar miles de tokens → costos directos
# Con GPT-4o: ~$10 por millón de tokens de output
# 100 prompts de 10,000 tokens = 1M tokens = ~$10
# Un bot automatizado puede enviar 100 prompts por minuto
# = $10/minuto = $600/hora = $14,400/día

API key abuse

# Escenario: API key filtrada o robada
# El atacante no ataca TU sistema — usa tu key directamente

# Si tu key está en:
# - Un repositorio público de GitHub
# - Un archivo .env commiteado
# - Logs del servidor
# - Variables de entorno de un CI/CD comprometido
# - El frontend de tu aplicación (JavaScript)

# El atacante puede hacer:
import os
from openai import OpenAI

stolen_key = "sk-live-..."  # Tu key robada
client = OpenAI(api_key=stolen_key)

# Uso masivo directo contra la API de OpenAI
# Generando contenido, fine-tuning, embeddings masivos
# TODO con cargo a TU cuenta

# Costo potencial:
# - GPT-4o: $2.50/M input tokens, $10/M output tokens
# - Fine-tuning: $25/M training tokens
# - Un script automatizado puede generar $50,000+ en una noche

Denial of wallet attacks

A diferencia de denial of service (tumbar el servidor), denial of wallet (vaciar tu presupuesto) es más difícil de detectar porque el servicio sigue funcionando — solo que cada request cuesta dinero:

# Denial of Wallet: el servicio funciona, pero los costos se disparan

# Ataque 1: Muchas requests pequeñas
# 10,000 requests × 100 tokens output = 1M tokens = ~$10
# Bajo impacto individual, alto impacto acumulado

# Ataque 2: Pocas requests costosas  
# 10 requests × 100,000 tokens output = 1M tokens = ~$10
# Cada request es costosa, fácil de detectar

# Ataque 3: Requests que disparan tool calls costosos
# Prompt que hace que el agent llame a múltiples APIs
# Cada tool call consume tokens adicionales en el contexto
# 1 request del usuario → 5 tool calls → 5x tokens

# Ataque 4: Repetir conversaciones largas
# Enviar el mismo prompt con variaciones menores
# El modelo no cachea — cada request es computación nueva

Mitigación: Rate limiting robusto

from datetime import datetime, timedelta
from collections import defaultdict
from pydantic import BaseModel, Field


class RateLimitConfig(BaseModel):
    """Configuración de rate limiting por tier de usuario."""

    requests_per_minute: int
    requests_per_hour: int
    requests_per_day: int
    max_input_tokens: int
    max_output_tokens: int
    daily_cost_limit_usd: float


class RateLimiter:
    """Rate limiter multi-nivel para endpoints LLM."""

    def __init__(self):
        self.configs: dict[str, RateLimitConfig] = {}
        self.request_history: dict[str, list[datetime]] = defaultdict(list)
        self.token_usage: dict[str, dict[str, int]] = defaultdict(
            lambda: {"input": 0, "output": 0}
        )
        self.cost_usage: dict[str, float] = defaultdict(float)

    def configure_tier(self, tier: str, config: RateLimitConfig) -> None:
        self.configs[tier] = config

    def check_limit(
        self,
        user_id: str,
        tier: str,
        estimated_input_tokens: int,
        estimated_output_tokens: int,
    ) -> tuple[bool, str]:
        """Verifica si el request está dentro de los límites."""
        if tier not in self.configs:
            return False, f"Tier '{tier}' no configurado"

        config = self.configs[tier]
        now = datetime.now()

        if estimated_input_tokens > config.max_input_tokens:
            return False, (
                f"Input tokens ({estimated_input_tokens}) excede máximo "
                f"({config.max_input_tokens})"
            )

        history = self.request_history[user_id]
        history = [t for t in history if t > now - timedelta(days=1)]
        self.request_history[user_id] = history

        last_minute = [t for t in history if t > now - timedelta(minutes=1)]
        if len(last_minute) >= config.requests_per_minute:
            return False, (
                f"Rate limit por minuto: {len(last_minute)}/{config.requests_per_minute}"
            )

        last_hour = [t for t in history if t > now - timedelta(hours=1)]
        if len(last_hour) >= config.requests_per_hour:
            return False, (
                f"Rate limit por hora: {len(last_hour)}/{config.requests_per_hour}"
            )

        if len(history) >= config.requests_per_day:
            return False, (
                f"Rate limit diario: {len(history)}/{config.requests_per_day}"
            )

        estimated_cost = self._estimate_cost(
            estimated_input_tokens, estimated_output_tokens
        )
        if self.cost_usage[user_id] + estimated_cost > config.daily_cost_limit_usd:
            return False, (
                f"Límite de costo diario: ${self.cost_usage[user_id]:.2f} + "
                f"${estimated_cost:.4f} > ${config.daily_cost_limit_usd:.2f}"
            )

        return True, "Permitido"

    def record_usage(
        self,
        user_id: str,
        input_tokens: int,
        output_tokens: int,
    ) -> None:
        """Registra el uso real después de la llamada."""
        self.request_history[user_id].append(datetime.now())
        self.token_usage[user_id]["input"] += input_tokens
        self.token_usage[user_id]["output"] += output_tokens
        self.cost_usage[user_id] += self._estimate_cost(input_tokens, output_tokens)

    def _estimate_cost(self, input_tokens: int, output_tokens: int) -> float:
        """Estima costo basado en pricing de GPT-4o."""
        input_cost = (input_tokens / 1_000_000) * 2.50
        output_cost = (output_tokens / 1_000_000) * 10.00
        return input_cost + output_cost

    def get_usage_summary(self, user_id: str) -> dict:
        """Resumen de uso de un usuario."""
        return {
            "user_id": user_id,
            "total_requests_today": len(self.request_history.get(user_id, [])),
            "input_tokens": self.token_usage[user_id]["input"],
            "output_tokens": self.token_usage[user_id]["output"],
            "cost_today_usd": round(self.cost_usage[user_id], 4),
        }


limiter = RateLimiter()

limiter.configure_tier(
    "free",
    RateLimitConfig(
        requests_per_minute=5,
        requests_per_hour=50,
        requests_per_day=200,
        max_input_tokens=500,
        max_output_tokens=1000,
        daily_cost_limit_usd=1.00,
    ),
)

limiter.configure_tier(
    "pro",
    RateLimitConfig(
        requests_per_minute=20,
        requests_per_hour=200,
        requests_per_day=1000,
        max_input_tokens=2000,
        max_output_tokens=4000,
        daily_cost_limit_usd=10.00,
    ),
)

for i in range(7):
    allowed, reason = limiter.check_limit(
        user_id="user-123",
        tier="free",
        estimated_input_tokens=200,
        estimated_output_tokens=500,
    )
    if allowed:
        limiter.record_usage("user-123", input_tokens=200, output_tokens=500)
        print(f"Request {i+1}: ✅ Permitido")
    else:
        print(f"Request {i+1}: ❌ Bloqueado — {reason}")

print()
print("Resumen:", limiter.get_usage_summary("user-123"))

# Output esperado:
# Request 1: ✅ Permitido
# Request 2: ✅ Permitido
# Request 3: ✅ Permitido
# Request 4: ✅ Permitido
# Request 5: ✅ Permitido
# Request 6: ❌ Bloqueado — Rate limit por minuto: 5/5
# Request 7: ❌ Bloqueado — Rate limit por minuto: 5/5
#
# Resumen: {'user_id': 'user-123', 'total_requests_today': 5, 'input_tokens': 1000, 'output_tokens': 2500, 'cost_today_usd': 0.0275}

Token budget por conversación

Además de rate limiting por usuario, implementa budgets por conversación individual:

class ConversationBudget:
    """Controla el presupuesto de tokens por conversación."""

    def __init__(
        self,
        max_turns: int = 20,
        max_total_tokens: int = 50_000,
        max_tokens_per_turn: int = 4_000,
        max_cost_usd: float = 0.50,
    ):
        self.max_turns = max_turns
        self.max_total_tokens = max_total_tokens
        self.max_tokens_per_turn = max_tokens_per_turn
        self.max_cost_usd = max_cost_usd
        self.current_turn = 0
        self.total_tokens_used = 0
        self.total_cost = 0.0

    def can_continue(self) -> tuple[bool, str]:
        if self.current_turn >= self.max_turns:
            return False, (
                f"Conversación alcanzó {self.max_turns} turnos. "
                f"Inicia una nueva conversación."
            )
        if self.total_tokens_used >= self.max_total_tokens:
            return False, (
                f"Presupuesto de tokens agotado: "
                f"{self.total_tokens_used}/{self.max_total_tokens}"
            )
        if self.total_cost >= self.max_cost_usd:
            return False, (
                f"Presupuesto de costo agotado: "
                f"${self.total_cost:.4f}/${self.max_cost_usd:.2f}"
            )
        return True, "OK"

    def record_turn(self, tokens_used: int) -> None:
        self.current_turn += 1
        self.total_tokens_used += tokens_used
        self.total_cost += (tokens_used / 1_000_000) * 10.0

    def get_remaining(self) -> dict:
        return {
            "turns_remaining": self.max_turns - self.current_turn,
            "tokens_remaining": self.max_total_tokens - self.total_tokens_used,
            "budget_remaining_usd": round(self.max_cost_usd - self.total_cost, 4),
        }


budget = ConversationBudget(max_turns=5, max_total_tokens=10_000, max_cost_usd=0.10)

for turn in range(7):
    can_go, reason = budget.can_continue()
    if can_go:
        budget.record_turn(tokens_used=2500)
        remaining = budget.get_remaining()
        print(f"Turno {turn+1}: ✅ — Quedan {remaining['turns_remaining']} turnos, {remaining['tokens_remaining']} tokens")
    else:
        print(f"Turno {turn+1}: ❌ — {reason}")

# Output esperado:
# Turno 1: ✅ — Quedan 4 turnos, 7500 tokens
# Turno 2: ✅ — Quedan 3 turnos, 5000 tokens
# Turno 3: ✅ — Quedan 2 turnos, 2500 tokens
# Turno 4: ✅ — Quedan 1 turnos, 0 tokens
# Turno 5: ❌ — Presupuesto de tokens agotado: 10000/10000
# Turno 6: ❌ — Presupuesto de tokens agotado: 10000/10000
# Turno 7: ❌ — Presupuesto de tokens agotado: 10000/10000

Cost monitoring y alertas

from dataclasses import dataclass, field


@dataclass
class CostMonitor:
    """Monitorea costos de API en tiempo real y genera alertas."""

    daily_budget_usd: float
    hourly_alert_threshold_usd: float
    alert_percentages: list[float] = field(default_factory=lambda: [0.5, 0.75, 0.9, 1.0])

    _total_cost: float = 0.0
    _hourly_cost: float = 0.0
    _alerts_triggered: set = field(default_factory=set)

    def record_cost(self, cost_usd: float) -> list[str]:
        """Registra un costo y retorna alertas si aplica."""
        self._total_cost += cost_usd
        self._hourly_cost += cost_usd
        alerts: list[str] = []

        for pct in self.alert_percentages:
            threshold = self.daily_budget_usd * pct
            if self._total_cost >= threshold and pct not in self._alerts_triggered:
                self._alerts_triggered.add(pct)
                alerts.append(
                    f"⚠️ ALERTA: Costo diario alcanzó {pct:.0%} del budget "
                    f"(${self._total_cost:.2f}/${self.daily_budget_usd:.2f})"
                )

        if self._hourly_cost >= self.hourly_alert_threshold_usd:
            alerts.append(
                f"🚨 ALERTA: Costo por hora (${self._hourly_cost:.2f}) "
                f"excede threshold (${self.hourly_alert_threshold_usd:.2f})"
            )

        return alerts

    def should_block(self) -> bool:
        """Determina si se deben bloquear nuevas requests."""
        return self._total_cost >= self.daily_budget_usd

    @property
    def remaining_budget(self) -> float:
        return max(0, self.daily_budget_usd - self._total_cost)


monitor = CostMonitor(
    daily_budget_usd=50.00,
    hourly_alert_threshold_usd=10.00,
)

simulated_costs = [5.0, 10.0, 8.0, 7.0, 12.0, 5.0, 4.0]

for i, cost in enumerate(simulated_costs):
    if monitor.should_block():
        print(f"Request {i+1}: 🛑 BLOQUEADO — Budget agotado")
        continue

    alerts = monitor.record_cost(cost)
    print(f"Request {i+1}: +${cost:.2f} (total: ${monitor._total_cost:.2f}, "
          f"quedan: ${monitor.remaining_budget:.2f})")
    for alert in alerts:
        print(f"  {alert}")

# Output esperado:
# Request 1: +$5.00 (total: $5.00, quedan: $45.00)
# Request 2: +$10.00 (total: $15.00, quedan: $35.00)
#   🚨 ALERTA: Costo por hora ($15.00) excede threshold ($10.00)
# Request 3: +$8.00 (total: $23.00, quedan: $27.00)
#   🚨 ALERTA: Costo por hora ($23.00) excede threshold ($10.00)
# Request 4: +$7.00 (total: $30.00, quedan: $20.00)
#   ⚠️ ALERTA: Costo diario alcanzó 50% del budget ($30.00/$50.00)
#   🚨 ALERTA: ...
# Request 5: +$12.00 (total: $42.00, quedan: $8.00)
#   ⚠️ ALERTA: Costo diario alcanzó 75% del budget ($42.00/$50.00)
#   🚨 ALERTA: ...
# Request 6: +$5.00 (total: $47.00, quedan: $3.00)
#   ⚠️ ALERTA: Costo diario alcanzó 90% del budget ($47.00/$50.00)
#   🚨 ALERTA: ...
# Request 7: +$4.00 (total: $51.00, quedan: $0.00)
#   ⚠️ ALERTA: Costo diario alcanzó 100% del budget ($51.00/$50.00)
#   🚨 ALERTA: ...

Conexión con el Módulo 5: API key protection

La defensa más importante contra API key abuse es que las keys nunca se filtren en primer lugar. El Módulo 5 (Secrets Management) cubre en profundidad:

# Lo que NO debes hacer (y que el Módulo 5 te enseña a evitar)
api_key_locations_inseguras = [
    ".env commiteado a git",
    "Variables de entorno sin encriptar en CI/CD",
    "Hardcodeado en el código fuente",
    "En logs del servidor (debug mode)",
    "En el frontend JavaScript (visible al usuario)",
    "En Docker images sin multi-stage builds",
    "En mensajes de error expuestos al usuario",
]

# Lo que SÍ debes hacer (Módulo 5 en detalle)
api_key_locations_seguras = [
    "HashiCorp Vault con rotación automática",
    "AWS Secrets Manager / GCP Secret Manager / Azure Key Vault",
    "Variables de entorno inyectadas en runtime (no build time)",
    "API gateway con key management integrado",
    "Service accounts con permisos mínimos",
]

Conexión con el proyecto: OWASP Mapping Audit

Al evaluar tu sistema contra LLM09 y LLM10 en tu OWASP Mapping Audit, pregúntate:

Para LLM09 (Misinformation):

• ¿Tu sistema responde sobre dominios críticos (salud, legal, financiero)?
• ¿El output del modelo tiene disclaimers apropiados?
• ¿Existe un mecanismo de grounding/verificación para claims del modelo?
• ¿Las respuestas citan fuentes verificables?

Para LLM10 (Unbounded Consumption):

• ¿Tienes rate limiting por usuario/endpoint?
• ¿Existe un budget diario con circuit breaker?
• ¿Monitoreas costos en tiempo real?
• ¿Tus API keys están protegidas (no en .env en git, no en frontend)?
• ¿Hay límites de tokens por request y por conversación?

Troubleshooting

"Nuestro modelo alucina respuestas sobre temas que no están en el knowledge base"

Esto es el comportamiento default del modelo — cuando no tiene información en el contexto RAG, rellena con su conocimiento de entrenamiento (que puede ser incorrecto). La mitigación es doble: (1) instruye al modelo explícitamente en el system prompt: "Si la respuesta no está en los documentos proporcionados, responde: 'No tengo información sobre esto. ¿Puedo ayudarte con algo más?'" y (2) implementa grounding verification que compare la respuesta con los documentos recuperados y marque claims no fundamentados.

"El rate limiter bloquea usuarios legítimos que hacen muchas consultas"

Tu rate limit está calibrado demasiado bajo o no tienes tiers diferenciados. Implementa múltiples tiers (free, pro, enterprise) con límites progresivos. Analiza el patrón de uso de tus usuarios legítimos para calibrar los límites: si el usuario promedio hace 30 requests/hora, configura el límite en 50-60/hora para tener margen. Para power users legítimos, ofrece un tier con límites más altos.

"¿Cómo detecto si están usando mi API key desde otro origen?"

Monitorea el patrón de uso: (1) IP addresses — si tu key normalmente se usa desde 3 IPs de tu servidor y aparece una IP nueva, es sospechoso, (2) horarios — si tu app tiene uso 9am-9pm y aparecen requests a las 3am, investiga, (3) patrones de uso — si tu chatbot genera 500 tokens promedio y aparecen requests de 10,000 tokens, es anomalía. OpenAI proporciona usage dashboards — configúralos con alertas.

"El cost monitor alerta demasiado — ¿cómo reduzco el ruido?"

Ajusta los thresholds basándote en tu baseline de costo normal. Si tu costo promedio diario es $20, configura alertas en 50% ($10), 75% ($15), y 90% ($18) — no en cantidades absolutas bajas. Para la alerta horaria, usa el costo promedio por hora como baseline. Si tu costo medio es $2/hora, alerta en $5/hora (2.5x baseline). Siempre mantén la alerta del 100% como hard stop.

"¿Los tokens de input cuentan para el costo?"

Sí, y son fácilmente explotables. Con GPT-4o, los tokens de input cuestan $2.50/M y los de output $10.00/M. Un atacante puede enviar prompts con miles de tokens de contexto falso para inflar tus costos de input sin necesitar que el modelo genere mucho output. Implementa max_input_tokens por request para limitar esto — 500-2000 tokens de input son suficientes para la mayoría de consultas de chatbot.

Ejercicios

Ejercicio 1: Implementar domain-specific disclaimers

Crea un sistema que genere disclaimers automáticos según el dominio detectado en la pregunta del usuario. El sistema debe detectar si la pregunta es médica, legal, financiera, o general y agregar el disclaimer apropiado.

import re


class DomainDisclaimerSystem:
    def __init__(self):
        self.domain_patterns: dict[str, list[str]] = {
            "medical": [
                r"medicamento", r"dosis", r"síntomas", r"enfermedad",
                r"tratamiento", r"diagnóstico", r"cirugía", r"doctor",
                r"hospital", r"medicina", r"efectos secundarios",
            ],
            "legal": [
                r"demanda", r"juicio", r"abogado", r"ley\b", r"contrato",
                r"derechos", r"tribunal", r"jurisprudencia", r"artículo\s+\d+",
                r"regulación", r"compliance",
            ],
            "financial": [
                r"inversión", r"rendimiento", r"acciones", r"fondo",
                r"crédito", r"impuestos", r"retorno", r"portfolio",
                r"hipoteca", r"trading",
            ],
        }
        self.disclaimers: dict[str, str] = {
            "medical": (
                "⚕️ AVISO: Esta información es generada por AI y no "
                "sustituye el diagnóstico o consejo de un profesional médico. "
                "Consulta a tu médico antes de tomar decisiones de salud."
            ),
            "legal": (
                "⚖️ AVISO: Esta información es generada por AI y no "
                "constituye asesoría legal. Consulta con un abogado "
                "calificado para tu caso específico."
            ),
            "financial": (
                "💰 AVISO: Esta información es generada por AI y no "
                "constituye asesoría financiera ni recomendación de "
                "inversión. Consulta con un asesor financiero certificado."
            ),
            "general": (
                "ℹ️ Respuesta generada por AI. Verifica con fuentes "
                "oficiales antes de actuar."
            ),
        }

    def detect_domain(self, user_input: str) -> str:
        input_lower = user_input.lower()
        scores: dict[str, int] = {}

        for domain, patterns in self.domain_patterns.items():
            score = sum(1 for p in patterns if re.search(p, input_lower))
            if score > 0:
                scores[domain] = score

        if not scores:
            return "general"
        return max(scores, key=scores.get)

    def add_disclaimer(self, user_input: str, llm_response: str) -> str:
        domain = self.detect_domain(user_input)
        disclaimer = self.disclaimers[domain]
        return f"{llm_response}\n\n---\n{disclaimer}"


system = DomainDisclaimerSystem()

test_questions = [
    "¿Cuáles son los efectos secundarios del ibuprofeno?",
    "¿Puedo demandar a mi empleador por despido injustificado?",
    "¿Qué fondo de inversión tiene mejor rendimiento?",
    "¿Cómo reseteo mi contraseña?",
]

for q in test_questions:
    domain = system.detect_domain(q)
    result = system.add_disclaimer(q, "Respuesta del modelo aquí.")
    print(f"Pregunta: {q}")
    print(f"Dominio: {domain}")
    print(f"Con disclaimer: {result}")
    print()

# Output esperado:
# Pregunta: ¿Cuáles son los efectos secundarios del ibuprofeno?
# Dominio: medical
# Con disclaimer: Respuesta del modelo aquí.
#
# ---
# ⚕️ AVISO: Esta información es generada por AI y no sustituye...

Ejercicio 2: Diseñar un sistema de tiered rate limiting

Tu aplicación tiene 3 tiers: Free, Pro ($19/mes), Enterprise (custom). Diseña la configuración de rate limiting para cada tier. Considera: requests por minuto/hora/día, max tokens input/output, budget diario, y qué pasa cuando el usuario alcanza el límite.

tier_configs = {
    "free": RateLimitConfig(
        requests_per_minute=3,
        requests_per_hour=30,
        requests_per_day=100,
        max_input_tokens=300,
        max_output_tokens=500,
        daily_cost_limit_usd=0.50,
    ),
    "pro": RateLimitConfig(
        requests_per_minute=15,
        requests_per_hour=150,
        requests_per_day=500,
        max_input_tokens=1500,
        max_output_tokens=3000,
        daily_cost_limit_usd=5.00,
    ),
    "enterprise": RateLimitConfig(
        requests_per_minute=60,
        requests_per_hour=600,
        requests_per_day=5000,
        max_input_tokens=4000,
        max_output_tokens=8000,
        daily_cost_limit_usd=100.00,
    ),
}

# Qué pasa cuando se alcanza el límite:
tier_limit_behavior = {
    "free": {
        "action": "Bloquear + mostrar upgrade prompt",
        "message": "Has alcanzado el límite gratuito. Actualiza a Pro para 5x más capacidad.",
        "retry_after": "Espera al siguiente periodo (minuto/hora/día)",
    },
    "pro": {
        "action": "Bloquear + mostrar uso + sugerir Enterprise",
        "message": "Has alcanzado el límite Pro. Contáctanos para un plan Enterprise.",
        "retry_after": "Espera al siguiente periodo",
    },
    "enterprise": {
        "action": "Rate limit suave + alerta al account manager",
        "message": "Uso inusualmente alto detectado. Tu account manager fue notificado.",
        "retry_after": "Continúa con degradación graceful (respuestas más cortas)",
    },
}

for tier, config in tier_configs.items():
    behavior = tier_limit_behavior[tier]
    print(f"\n{'='*50}")
    print(f"Tier: {tier.upper()}")
    print(f"{'='*50}")
    print(f"  Requests: {config.requests_per_minute}/min, "
          f"{config.requests_per_hour}/hr, {config.requests_per_day}/día")
    print(f"  Tokens: {config.max_input_tokens} in / {config.max_output_tokens} out")
    print(f"  Budget: ${config.daily_cost_limit_usd:.2f}/día")
    print(f"  Al límite: {behavior['action']}")

Ejercicio 3: Calcular el costo de un ataque

Un atacante envía prompts automatizados a tu endpoint público. Cada prompt tiene ~800 tokens de input y genera ~3,000 tokens de output. El script envía 10 requests por segundo. Calcula:

1. Costo por minuto
2. Costo por hora
3. Tiempo para alcanzar un budget de $1,000
4. Qué rate limit necesitas para que el costo máximo del ataque sea <$50/día

# Datos del ataque
input_tokens_per_request = 800
output_tokens_per_request = 3000
requests_per_second = 10

# Pricing GPT-4o (Marzo 2026)
input_cost_per_million = 2.50   # USD
output_cost_per_million = 10.00  # USD

# Costo por request
cost_per_request = (
    (input_tokens_per_request / 1_000_000) * input_cost_per_million
    + (output_tokens_per_request / 1_000_000) * output_cost_per_million
)
print(f"Costo por request: ${cost_per_request:.6f}")

# 1. Costo por minuto
requests_per_minute = requests_per_second * 60
cost_per_minute = cost_per_request * requests_per_minute
print(f"Costo por minuto: ${cost_per_minute:.2f} ({requests_per_minute} requests)")

# 2. Costo por hora
cost_per_hour = cost_per_minute * 60
print(f"Costo por hora: ${cost_per_hour:.2f}")

# 3. Tiempo para $1,000
budget = 1000
minutes_to_budget = budget / cost_per_minute
hours_to_budget = minutes_to_budget / 60
print(f"Tiempo para ${budget}: {minutes_to_budget:.1f} minutos ({hours_to_budget:.1f} horas)")

# 4. Rate limit para <$50/día
target_daily_cost = 50
seconds_per_day = 86400
max_requests_per_day = target_daily_cost / cost_per_request
max_requests_per_minute_safe = max_requests_per_day / (60 * 24)
print(f"Rate limit para <${target_daily_cost}/día: "
      f"{max_requests_per_minute_safe:.0f} requests/minuto")

# Output esperado:
# Costo por request: $0.032000
# Costo por minuto: $19.20 (600 requests)
# Costo por hora: $1,152.00
# Tiempo para $1000: 52.1 minutos (0.9 horas)
# Rate limit para <$50/día: ~1 request/minuto (para el endpoint público)

Ejercicio 4: Implementar grounding verification

Escribe una función que tome la respuesta del LLM y los documentos recuperados del RAG, y calcule un "grounding score" basado en cuánto del contenido de la respuesta se puede rastrear a los documentos fuente.

def calculate_grounding_score(
    response: str,
    source_documents: list[str],
    min_word_overlap: float = 0.3,
) -> dict:
    """Calcula qué porcentaje de la respuesta está fundamentado en fuentes."""
    response_sentences = [
        s.strip() for s in response.replace("\n", ".").split(".")
        if len(s.strip().split()) >= 4
    ]

    all_source_words: set[str] = set()
    for doc in source_documents:
        all_source_words.update(doc.lower().split())

    grounded_sentences: list[str] = []
    ungrounded_sentences: list[str] = []

    for sentence in response_sentences:
        sentence_words = set(sentence.lower().split())
        if not sentence_words:
            continue
        overlap = len(sentence_words & all_source_words) / len(sentence_words)
        if overlap >= min_word_overlap:
            grounded_sentences.append(sentence)
        else:
            ungrounded_sentences.append(sentence)

    total = len(response_sentences) if response_sentences else 1
    score = len(grounded_sentences) / total

    return {
        "grounding_score": round(score, 2),
        "total_sentences": total,
        "grounded": len(grounded_sentences),
        "ungrounded": len(ungrounded_sentences),
        "ungrounded_sentences": ungrounded_sentences,
        "recommendation": (
            "✅ Alta confianza" if score >= 0.8
            else "🟡 Verificar claims no fundamentados" if score >= 0.5
            else "❌ Baja confianza — requiere revisión"
        ),
    }


sources = [
    "AcmeCorp ofrece devoluciones en 30 días con recibo original. "
    "El reembolso se procesa en 5-7 días hábiles al método de pago original.",
    "El envío estándar tarda 3-5 días hábiles. "
    "Envío express disponible por $9.99 adicionales, entrega en 1-2 días.",
]

response = (
    "AcmeCorp permite devoluciones en 30 días con recibo original. "
    "El reembolso tarda 5-7 días hábiles. "
    "El envío estándar es de 3-5 días hábiles. "
    "Además, todos los productos tienen garantía de satisfacción de por vida. "
    "Si no estás satisfecho, te devolvemos el triple de tu dinero."
)

result = calculate_grounding_score(response, sources)
print(f"Grounding Score: {result['grounding_score']:.0%}")
print(f"Fundamentadas: {result['grounded']}/{result['total_sentences']}")
print(f"Recomendación: {result['recommendation']}")
if result['ungrounded_sentences']:
    print("Sentences no fundamentadas:")
    for s in result['ungrounded_sentences']:
        print(f"  ⚠️ {s}")

# Output esperado:
# Grounding Score: 60%
# Fundamentadas: 3/5
# Recomendación: 🟡 Verificar claims no fundamentados
# Sentences no fundamentadas:
#   ⚠️ Además, todos los productos tienen garantía de satisfacción de por vida
#   ⚠️ Si no estás satisfecho, te devolvemos el triple de tu dinero

Ejercicio 5: Diseñar alertas de anomalía de costos

Tu sistema tiene un costo promedio de $0.03 por request. Diseña un sistema de detección de anomalías que alerte cuando el costo por request, el costo por hora, o el número de requests se desvíe significativamente del baseline.

from statistics import mean, stdev


@dataclass
class AnomalyDetector:
    baseline_cost_per_request: float
    baseline_requests_per_hour: float
    baseline_hourly_cost: float
    sensitivity: float = 2.0  # Desviaciones estándar

    def check_request_cost(self, cost: float) -> tuple[bool, str]:
        threshold = self.baseline_cost_per_request * (1 + self.sensitivity)
        if cost > threshold:
            return True, (
                f"Costo anómalo: ${cost:.4f} "
                f"(baseline: ${self.baseline_cost_per_request:.4f}, "
                f"threshold: ${threshold:.4f})"
            )
        return False, "Normal"

    def check_hourly_rate(self, requests_this_hour: int) -> tuple[bool, str]:
        threshold = self.baseline_requests_per_hour * (1 + self.sensitivity)
        if requests_this_hour > threshold:
            return True, (
                f"Tasa anómala: {requests_this_hour} req/hr "
                f"(baseline: {self.baseline_requests_per_hour:.0f}, "
                f"threshold: {threshold:.0f})"
            )
        return False, "Normal"

    def check_hourly_cost(self, cost_this_hour: float) -> tuple[bool, str]:
        threshold = self.baseline_hourly_cost * (1 + self.sensitivity)
        if cost_this_hour > threshold:
            return True, (
                f"Costo horario anómalo: ${cost_this_hour:.2f} "
                f"(baseline: ${self.baseline_hourly_cost:.2f}, "
                f"threshold: ${threshold:.2f})"
            )
        return False, "Normal"


detector = AnomalyDetector(
    baseline_cost_per_request=0.03,
    baseline_requests_per_hour=100,
    baseline_hourly_cost=3.00,
    sensitivity=2.0,
)

test_scenarios = [
    {"label": "Normal", "cost": 0.028, "requests": 95, "hourly_cost": 2.66},
    {"label": "Prompt costoso", "cost": 0.15, "requests": 95, "hourly_cost": 2.66},
    {"label": "Bot automatizado", "cost": 0.03, "requests": 500, "hourly_cost": 15.00},
    {"label": "Ataque combinado", "cost": 0.12, "requests": 450, "hourly_cost": 54.00},
]

for scenario in test_scenarios:
    print(f"\n--- {scenario['label']} ---")
    anomaly, msg = detector.check_request_cost(scenario['cost'])
    print(f"  Costo request: {'🚨' if anomaly else '✅'} {msg}")
    anomaly, msg = detector.check_hourly_rate(scenario['requests'])
    print(f"  Tasa horaria: {'🚨' if anomaly else '✅'} {msg}")
    anomaly, msg = detector.check_hourly_cost(scenario['hourly_cost'])
    print(f"  Costo horario: {'🚨' if anomaly else '✅'} {msg}")

# Output esperado:
# --- Normal ---
#   Costo request: ✅ Normal
#   Tasa horaria: ✅ Normal
#   Costo horario: ✅ Normal
#
# --- Prompt costoso ---
#   Costo request: 🚨 Costo anómalo: $0.1500 (baseline: $0.0300, threshold: $0.0900)
#   Tasa horaria: ✅ Normal
#   Costo horario: ✅ Normal
#
# --- Bot automatizado ---
#   Costo request: ✅ Normal
#   Tasa horaria: 🚨 Tasa anómala: 500 req/hr (baseline: 100, threshold: 300)
#   Costo horario: 🚨 Costo horario anómalo: $15.00 (baseline: $3.00, threshold: $9.00)
#
# --- Ataque combinado ---
#   Costo request: 🚨 Costo anómalo: ...
#   Tasa horaria: 🚨 Tasa anómala: ...
#   Costo horario: 🚨 Costo horario anómalo: ...

Ejercicio 6: Construir un confidence scorer

Implementa un sistema que asigne un score de confianza a cada respuesta del LLM basándose en múltiples factores: grounding score, presencia de claims numéricos no verificados, citas fabricadas, y dominio de la pregunta.

class ConfidenceScorer:
    """Asigna scores de confianza a respuestas del LLM."""

    def __init__(self):
        self.domain_penalties = {
            "medical": 0.3,
            "legal": 0.25,
            "financial": 0.2,
            "general": 0.0,
        }

    def score(
        self,
        response: str,
        grounding_score: float,
        domain: str = "general",
        source_docs: list[str] | None = None,
    ) -> dict:
        base_score = grounding_score
        penalties: list[tuple[str, float]] = []

        domain_penalty = self.domain_penalties.get(domain, 0.0)
        if domain_penalty > 0:
            penalties.append((f"Dominio sensible ({domain})", domain_penalty))

        numbers = re.findall(r"\d+\.?\d*%", response)
        if len(numbers) > 3:
            stat_penalty = min(len(numbers) * 0.03, 0.15)
            penalties.append((f"{len(numbers)} estadísticas no verificadas", stat_penalty))

        citation_patterns = [
            r"\(\d{4}\)",
            r"et al\.",
            r"Journal of",
            r"published in",
            r"publicado en",
        ]
        citation_count = sum(
            len(re.findall(p, response, re.IGNORECASE)) for p in citation_patterns
        )
        if citation_count > 0 and grounding_score < 0.5:
            cite_penalty = min(citation_count * 0.05, 0.2)
            penalties.append((f"{citation_count} citas potencialmente fabricadas", cite_penalty))

        hedge_words = [
            "posiblemente", "quizás", "tal vez", "podría ser",
            "es posible que", "en algunos casos",
        ]
        response_lower = response.lower()
        hedges = sum(1 for h in hedge_words if h in response_lower)
        if hedges > 0:
            penalties.append((f"{hedges} hedging words (menor certeza)", -0.05))

        total_penalty = sum(p for _, p in penalties)
        final_score = max(0.0, min(1.0, base_score - total_penalty))

        if final_score >= 0.8:
            level = "🟢 Alta"
        elif final_score >= 0.5:
            level = "🟡 Media"
        elif final_score >= 0.3:
            level = "🟠 Baja"
        else:
            level = "🔴 Muy baja"

        return {
            "final_score": round(final_score, 2),
            "level": level,
            "base_score": round(base_score, 2),
            "penalties": penalties,
            "total_penalty": round(total_penalty, 2),
        }


scorer = ConfidenceScorer()

result = scorer.score(
    response=(
        "Según un estudio de Smith et al. (2024), el 87.3% de los pacientes "
        "muestran mejora con este tratamiento. La tasa de efectos secundarios "
        "es del 12.5%, con un 3.2% de casos severos."
    ),
    grounding_score=0.3,
    domain="medical",
)

print(f"Score: {result['final_score']} — {result['level']}")
print(f"Base: {result['base_score']}, Penalización: {result['total_penalty']}")
for name, penalty in result['penalties']:
    print(f"  {'📉' if penalty > 0 else '📈'} {name}: {penalty:+.2f}")

# Output esperado:
# Score: 0.0 — 🔴 Muy baja
# Base: 0.3, Penalización: 0.55+
# Penalties muestran: dominio médico, estadísticas no verificadas, citas potencialmente fabricadas

Resumen

• LLM09 (Misinformation) es un riesgo de seguridad, no solo un problema de calidad — las alucinaciones en dominios críticos (médico, legal, financiero) pueden causar daño real y exposición legal
• Los tipos de misinformation incluyen fabricación de datos/estadísticas, citas inventadas, y confianza extrema en respuestas incorrectas
• Defensa contra LLM09: fact-checking pipeline con base de conocimiento verificada, grounding verification contra documentos fuente, confidence scoring, y disclaimers obligatorios por dominio
• LLM10 (Unbounded Consumption) abarca token exhaustion, API key abuse, y denial-of-wallet attacks — el objetivo es generar costos masivos, no tumbar el servidor
• Los vectores de LLM10 incluyen prompts diseñados para maximizar tokens, scripts automatizados de alto volumen, API keys robadas usadas directamente, y requests que disparan múltiples tool calls
• Defensa contra LLM10: rate limiting multi-nivel (por minuto/hora/día), token budgets por conversación, cost monitoring con alertas y circuit breakers, y API key protection (cubierta en profundidad en el Módulo 5)
• Ambas vulnerabilidades requieren defensas operacionales además de técnicas — procesos continuos de monitoreo, calibración de thresholds, y revisión de baselines
• En tu OWASP Mapping Audit, evalúa el impacto contextual: LLM09 varía de bajo (chatbot de recetas) a crítico (asistente médico), y LLM10 varía según tu exposición pública y presupuesto

Próxima cápsula: En la cápsula 08 construirás tu OWASP Mapping Audit completo — el documento que mapea tu sistema contra las 10 vulnerabilidades, evalúa el estado de cada una, y produce un roadmap de mitigación que guía los módulos 3-7.

Recursos adicionales

1. OWASP LLM09: Misinformation — Documentación oficial de OWASP sobre riesgos de desinformación en LLMs con escenarios de impacto y mitigaciones
2. OWASP LLM10: Unbounded Consumption — Documentación oficial de OWASP sobre agotamiento de recursos, incluyendo token exhaustion y denial of wallet
3. OpenAI Usage and Billing Dashboard — Dashboard de OpenAI para monitorear uso de tokens y costos en tiempo real
4. FreshLLMs: Refreshing Large Language Models with Search Engine Augmentation — Paper de investigación sobre grounding de LLMs con información actualizada para reducir alucinaciones
5. OpenAI Rate Limits and Best Practices — Guía oficial de OpenAI sobre rate limits, headers, y mejores prácticas para manejar límites de API
6. Anthropic Usage and Pricing — Referencia de precios de Anthropic para calcular costos y configurar budgets
7. Hallucination Leaderboard — Vectara — Benchmark de alucinaciones de diferentes LLMs, útil para seleccionar modelos con menor tasa de fabricación
8. NIST AI 600-1: AI Risk Management Framework — Generative AI Profile — Framework del NIST específico para riesgos de AI generativa, incluyendo desinformación y uso indebido

Creado: Marzo 2026 Versión: 1.0