Módulo 6: Data Privacy & PII Protection

5. Data Minimization

Descripción

En las cápsulas 03 y 04 aprendiste a detectar y redactar PII en los datos que pasan por tu sistema AI. Pero hay una pregunta más fundamental que deberías hacerte antes de redactar: ¿necesitas enviar esos datos al LLM en primer lugar?

Data minimization es el principio de enviar solo la cantidad mínima de datos necesaria para que el LLM cumpla su tarea. Si un usuario pregunta "¿cuál es el estado de mi pedido?", el LLM no necesita su SSN, su historial médico, ni su dirección completa — necesita el número de pedido y su estado. Cada dato adicional que envías al modelo es un dato que puede filtrarse.

Este principio viene directamente de GDPR (Art. 5(1)(c): "los datos serán adecuados, pertinentes y limitados a lo necesario") y es una de las defensas más efectivas contra LLM02 porque reduce la superficie de ataque: si el dato nunca llega al modelo, el modelo no puede revelarlo.

En esta cápsula construyes un Data Minimizer que filtra, clasifica y reduce los datos antes de enviarlos al LLM. Este componente se integra entre el PII Scanner/Redactor y la llamada al modelo en el PII Protection Layer.


El principio de necesidad de saber

La data minimization aplicada a AI es equivalente al principio de "need-to-know" en seguridad de la información: cada componente del sistema solo accede a los datos que necesita para cumplir su función.

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


class AccessLevel(Enum):
    FULL = "full"
    PARTIAL = "partial"
    MINIMAL = "minimal"
    NONE = "none"


@dataclass
class DataAccessPolicy:
    """Define qué datos puede ver cada componente."""
    component: str
    access_level: AccessLevel
    allowed_fields: list[str]
    denied_fields: list[str]
    rationale: str


access_policies = [
    DataAccessPolicy(
        component="LLM (general query)",
        access_level=AccessLevel.MINIMAL,
        allowed_fields=["query", "product_info", "order_status"],
        denied_fields=["ssn", "credit_card", "full_address", "medical_history"],
        rationale="El LLM solo necesita el contexto de la pregunta y datos del producto",
    ),
    DataAccessPolicy(
        component="LLM (personalized response)",
        access_level=AccessLevel.PARTIAL,
        allowed_fields=["first_name", "query", "order_history", "preferences"],
        denied_fields=["ssn", "credit_card", "full_address", "email", "phone"],
        rationale="Personalización requiere nombre y preferencias, no datos financieros",
    ),
    DataAccessPolicy(
        component="RAG retrieval",
        access_level=AccessLevel.PARTIAL,
        allowed_fields=["query_embedding", "document_metadata"],
        denied_fields=["raw_documents_with_pii"],
        rationale="El retriever busca por similitud semántica, no necesita PII",
    ),
    DataAccessPolicy(
        component="Audit log",
        access_level=AccessLevel.MINIMAL,
        allowed_fields=["request_id", "timestamp", "action", "redaction_count"],
        denied_fields=["original_text", "pii_values", "user_data"],
        rationale="Los logs registran eventos, no datos personales",
    ),
]

print("Políticas de acceso a datos:\n")
for policy in access_policies:
    print(f"  {policy.component} [{policy.access_level.value}]")
    print(f"    Permitido: {', '.join(policy.allowed_fields[:3])}")
    print(f"    Denegado:  {', '.join(policy.denied_fields[:3])}")
    print(f"    Razón: {policy.rationale[:60]}...")
    print()

Clasificación de datos para acceso LLM

Antes de minimizar, necesitas clasificar cada campo de datos por su necesidad para la tarea del LLM.

from enum import Enum
from typing import Optional


class DataNecessity(Enum):
    REQUIRED = "required"
    USEFUL = "useful"
    UNNECESSARY = "unnecessary"
    FORBIDDEN = "forbidden"


@dataclass
class FieldClassification:
    field_name: str
    necessity: DataNecessity
    sensitivity: str
    minimization_action: str


def classify_fields_for_task(
    user_data: dict,
    task_type: str,
) -> list[FieldClassification]:
    """Clasifica cada campo según la necesidad para la tarea."""
    
    task_requirements = {
        "order_status": {
            "required": ["order_id", "query"],
            "useful": ["first_name", "order_date"],
            "unnecessary": ["email", "phone", "address"],
            "forbidden": ["ssn", "credit_card", "password"],
        },
        "product_recommendation": {
            "required": ["query", "preferences"],
            "useful": ["purchase_history", "first_name"],
            "unnecessary": ["full_name", "address", "phone"],
            "forbidden": ["ssn", "credit_card", "date_of_birth"],
        },
        "support_ticket": {
            "required": ["query", "ticket_id", "product_name"],
            "useful": ["first_name", "purchase_date"],
            "unnecessary": ["address", "payment_method"],
            "forbidden": ["ssn", "credit_card", "medical_info"],
        },
    }
    
    requirements = task_requirements.get(task_type, {})
    classifications = []
    
    sensitivity_map = {
        "ssn": "critical",
        "credit_card": "critical",
        "password": "critical",
        "email": "high",
        "phone": "high",
        "full_name": "medium",
        "first_name": "low",
        "address": "high",
        "date_of_birth": "medium",
        "order_id": "low",
        "query": "low",
        "preferences": "low",
    }
    
    action_map = {
        DataNecessity.REQUIRED: "include_as_is",
        DataNecessity.USEFUL: "include_redacted",
        DataNecessity.UNNECESSARY: "exclude",
        DataNecessity.FORBIDDEN: "block",
    }
    
    for field_name in user_data.keys():
        if field_name in requirements.get("required", []):
            necessity = DataNecessity.REQUIRED
        elif field_name in requirements.get("useful", []):
            necessity = DataNecessity.USEFUL
        elif field_name in requirements.get("forbidden", []):
            necessity = DataNecessity.FORBIDDEN
        else:
            necessity = DataNecessity.UNNECESSARY
        
        classifications.append(FieldClassification(
            field_name=field_name,
            necessity=necessity,
            sensitivity=sensitivity_map.get(field_name, "unknown"),
            minimization_action=action_map[necessity],
        ))
    
    return classifications


# --- Demostración ---

user_data = {
    "query": "¿Cuál es el estado de mi pedido?",
    "order_id": "ORD-12345",
    "first_name": "María",
    "full_name": "María García López",
    "email": "maria@empresa.com",
    "phone": "555-123-4567",
    "ssn": "123-45-6789",
    "credit_card": "4111-1111-1111-1111",
    "address": "Calle Reforma 123, CDMX",
}

classifications = classify_fields_for_task(user_data, "order_status")

print("Clasificación de campos para 'order_status':\n")
for c in classifications:
    icon = {
        DataNecessity.REQUIRED: "✅",
        DataNecessity.USEFUL: "🔶",
        DataNecessity.UNNECESSARY: "❌",
        DataNecessity.FORBIDDEN: "🚫",
    }[c.necessity]
    print(f"  {icon} {c.field_name}: {c.necessity.value}{c.minimization_action}")

# Output esperado:
#   ✅ query: required → include_as_is
#   ✅ order_id: required → include_as_is
#   🔶 first_name: useful → include_redacted
#   ❌ full_name: unnecessary → exclude
#   ❌ email: unnecessary → exclude
#   ❌ phone: unnecessary → exclude
#   🚫 ssn: forbidden → block
#   🚫 credit_card: forbidden → block
#   ❌ address: unnecessary → exclude

Data Minimizer: la clase completa

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


class MinimizationAction(Enum):
    INCLUDED = "included"
    REDACTED = "redacted"
    EXCLUDED = "excluded"
    BLOCKED = "blocked"
    GENERALIZED = "generalized"


@dataclass
class MinimizedField:
    field_name: str
    action: MinimizationAction
    original_value: Optional[str] = None
    minimized_value: Optional[str] = None


@dataclass
class MinimizationResult:
    original_field_count: int
    minimized_data: dict
    actions: list[MinimizedField] = field(default_factory=list)
    excluded_count: int = 0
    blocked_count: int = 0
    data_reduction_percent: float = 0.0


class DataMinimizer:
    """Minimiza los datos enviados al LLM según la necesidad de la tarea."""

    FORBIDDEN_FIELDS = {
        "ssn", "social_security", "credit_card", "card_number",
        "password", "secret", "api_key", "token",
        "medical_record", "health_data", "biometric",
    }

    SENSITIVE_FIELDS = {
        "email", "phone", "telephone", "address", "full_address",
        "date_of_birth", "dob", "full_name", "last_name",
        "passport", "license_number", "account_number",
    }

    def __init__(
        self,
        task_allowed_fields: Optional[list[str]] = None,
        include_sensitive: bool = False,
        generalize_dates: bool = True,
        generalize_locations: bool = True,
    ):
        self.task_allowed_fields = task_allowed_fields
        self.include_sensitive = include_sensitive
        self.generalize_dates = generalize_dates
        self.generalize_locations = generalize_locations

    def minimize(
        self,
        data: dict,
        task_type: Optional[str] = None,
    ) -> MinimizationResult:
        """Minimiza un dict de datos según las políticas configuradas."""
        minimized = {}
        actions = []
        excluded = 0
        blocked = 0

        original_size = sum(
            len(str(v)) for v in data.values() if v is not None
        )

        for key, value in data.items():
            normalized_key = key.lower().replace("-", "_").replace(" ", "_")

            if normalized_key in self.FORBIDDEN_FIELDS:
                actions.append(MinimizedField(
                    field_name=key,
                    action=MinimizationAction.BLOCKED,
                ))
                blocked += 1
                continue

            if self.task_allowed_fields and key not in self.task_allowed_fields:
                actions.append(MinimizedField(
                    field_name=key,
                    action=MinimizationAction.EXCLUDED,
                ))
                excluded += 1
                continue

            if normalized_key in self.SENSITIVE_FIELDS:
                if not self.include_sensitive:
                    actions.append(MinimizedField(
                        field_name=key,
                        action=MinimizationAction.EXCLUDED,
                    ))
                    excluded += 1
                    continue

                generalized = self._generalize(key, value)
                if generalized != value:
                    minimized[key] = generalized
                    actions.append(MinimizedField(
                        field_name=key,
                        action=MinimizationAction.GENERALIZED,
                        minimized_value=str(generalized),
                    ))
                    continue

            minimized[key] = value
            actions.append(MinimizedField(
                field_name=key,
                action=MinimizationAction.INCLUDED,
            ))

        minimized_size = sum(
            len(str(v)) for v in minimized.values() if v is not None
        )
        reduction = (
            (1 - minimized_size / original_size) * 100
            if original_size > 0 else 0
        )

        return MinimizationResult(
            original_field_count=len(data),
            minimized_data=minimized,
            actions=actions,
            excluded_count=excluded,
            blocked_count=blocked,
            data_reduction_percent=round(reduction, 1),
        )

    def _generalize(self, key: str, value: Any) -> Any:
        """Generaliza un valor para reducir su especificidad."""
        if value is None:
            return None

        str_value = str(value)
        normalized_key = key.lower()

        if normalized_key in ("email",) and "@" in str_value:
            domain = str_value.split("@")[1]
            return f"***@{domain}"

        if normalized_key in ("phone", "telephone"):
            if len(str_value) >= 4:
                return "***" + str_value[-4:]
            return "***"

        if normalized_key in ("address", "full_address"):
            parts = str_value.split(",")
            if len(parts) >= 2:
                return parts[-1].strip()
            return "[Location]"

        if normalized_key in ("date_of_birth", "dob"):
            import re
            year_match = re.search(r"(19|20)\d{2}", str_value)
            if year_match:
                year = int(year_match.group())
                decade = (year // 10) * 10
                return f"{decade}s"
            return "[Date]"

        if normalized_key in ("full_name",):
            parts = str_value.split()
            if parts:
                return parts[0]
            return "[Name]"

        return value


# --- Demostración ---

user_data = {
    "query": "¿Cuál es el estado de mi pedido?",
    "order_id": "ORD-12345",
    "first_name": "María",
    "full_name": "María García López",
    "email": "maria@empresa.com",
    "phone": "555-123-4567",
    "ssn": "123-45-6789",
    "credit_card": "4111-1111-1111-1111",
    "address": "Calle Reforma 123, CDMX, México",
    "date_of_birth": "03/15/1990",
    "order_status": "shipped",
}

minimizer = DataMinimizer(
    task_allowed_fields=[
        "query", "order_id", "first_name", "order_status",
    ],
)

result = minimizer.minimize(user_data)

print("Data Minimization Results:\n")
print(f"  Original fields: {result.original_field_count}")
print(f"  Minimized fields: {len(result.minimized_data)}")
print(f"  Excluded: {result.excluded_count}")
print(f"  Blocked: {result.blocked_count}")
print(f"  Data reduction: {result.data_reduction_percent}%")
print(f"\n  Minimized data:")
for key, value in result.minimized_data.items():
    print(f"    {key}: {value}")
print(f"\n  Actions:")
for action in result.actions:
    icon = {
        MinimizationAction.INCLUDED: "✅",
        MinimizationAction.REDACTED: "🔶",
        MinimizationAction.EXCLUDED: "❌",
        MinimizationAction.BLOCKED: "🚫",
        MinimizationAction.GENERALIZED: "📐",
    }[action.action]
    print(f"    {icon} {action.field_name}: {action.action.value}")

# Output esperado:
#   Original fields: 11
#   Minimized fields: 4
#   Excluded: 5
#   Blocked: 2
#   Data reduction: ~70%
#
#   Minimized data:
#     query: ¿Cuál es el estado de mi pedido?
#     order_id: ORD-12345
#     first_name: María
#     order_status: shipped

Truncación inteligente de contexto

Cuando el contexto para el LLM es un texto largo (documento, historial de chat, resultados de RAG), necesitas truncar inteligentemente para maximizar la relevancia dentro del token budget.

from dataclasses import dataclass


@dataclass
class TruncationResult:
    original_length: int
    truncated_length: int
    estimated_tokens: int
    strategy_used: str
    text: str


class ContextTruncator:
    """Trunca contexto de forma inteligente para el LLM."""

    def __init__(
        self,
        max_tokens: int = 2000,
        chars_per_token: int = 4,
    ):
        self.max_tokens = max_tokens
        self.max_chars = max_tokens * chars_per_token
        self.chars_per_token = chars_per_token

    def truncate(
        self,
        text: str,
        strategy: str = "smart",
    ) -> TruncationResult:
        """Trunca texto según la estrategia."""
        if len(text) <= self.max_chars:
            return TruncationResult(
                original_length=len(text),
                truncated_length=len(text),
                estimated_tokens=len(text) // self.chars_per_token,
                strategy_used="none",
                text=text,
            )

        if strategy == "start":
            truncated = text[:self.max_chars]
        elif strategy == "end":
            truncated = text[-self.max_chars:]
        elif strategy == "smart":
            truncated = self._smart_truncate(text)
        elif strategy == "middle_out":
            truncated = self._middle_out(text)
        else:
            truncated = text[:self.max_chars]

        return TruncationResult(
            original_length=len(text),
            truncated_length=len(truncated),
            estimated_tokens=len(truncated) // self.chars_per_token,
            strategy_used=strategy,
            text=truncated,
        )

    def _smart_truncate(self, text: str) -> str:
        """Mantiene inicio y final, trunca el medio."""
        keep_start = self.max_chars // 3
        keep_end = self.max_chars // 3
        start = text[:keep_start]
        end = text[-keep_end:]
        return f"{start}\n\n[... {len(text) - keep_start - keep_end} chars omitted ...]\n\n{end}"

    def _middle_out(self, text: str) -> str:
        """Prioriza el contenido del medio (útil para documentos con headers/footers)."""
        total_to_remove = len(text) - self.max_chars
        remove_start = total_to_remove // 2
        remove_end = total_to_remove - remove_start
        return text[remove_start:len(text) - remove_end]


# --- Demostración ---

truncator = ContextTruncator(max_tokens=100)

long_text = "Important intro. " + "Middle content. " * 200 + "Critical conclusion."

for strategy in ["start", "end", "smart", "middle_out"]:
    result = truncator.truncate(long_text, strategy=strategy)
    print(f"Strategy: {strategy}")
    print(f"  Original: {result.original_length} chars")
    print(f"  Truncated: {result.truncated_length} chars")
    print(f"  Tokens: ~{result.estimated_tokens}")
    print(f"  Preview: \"{result.text[:50]}...\"")
    print()

Prompt engineering para privacidad

Puedes diseñar tus prompts para que el LLM genere respuestas útiles sin necesitar datos sensibles.

privacy_prompt_patterns = {
    "reference_by_id": {
        "bad": (
            "El usuario María García (maria@test.com, SSN: 123-45-6789) "
            "pregunta sobre su pedido #12345"
        ),
        "good": (
            "El usuario [ID: USR-789] pregunta sobre el pedido #12345. "
            "Estado del pedido: enviado, entrega estimada: mañana."
        ),
        "principle": "Referencia por ID, no por datos personales",
    },
    "provide_answer_not_data": {
        "bad": (
            "Datos del cliente:\n"
            "Nombre: Juan López\nEmail: juan@test.com\n"
            "Historial: 15 compras, total $5,432\n"
            "Pregunta: ¿Soy elegible para descuento VIP?"
        ),
        "good": (
            "Un cliente con 15 compras y total histórico >$5,000 "
            "pregunta si es elegible para descuento VIP. "
            "Política: VIP requiere 10+ compras y >$3,000."
        ),
        "principle": "Proporciona la respuesta, no los datos para calcularla",
    },
    "aggregate_not_individual": {
        "bad": (
            "Empleados del departamento:\n"
            "- Ana Ruiz, $85,000\n- Pedro Soto, $92,000\n"
            "- Carmen Vega, $78,000\n"
            "Pregunta: ¿Cuál es el promedio salarial?"
        ),
        "good": (
            "Departamento con 3 empleados. "
            "Rango salarial: $78,000-$92,000. Promedio: $85,000. "
            "Pregunta: resume la información salarial."
        ),
        "principle": "Envía agregados, no datos individuales",
    },
    "separate_context_from_query": {
        "bad": (
            "Mi nombre es Carlos Méndez, tengo 42 años, vivo en "
            "Guadalajara, mi email es carlos@test.com. "
            "¿Qué restaurantes me recomiendas?"
        ),
        "good": (
            "Un usuario en Guadalajara busca recomendaciones de "
            "restaurantes. Preferencias: cocina mexicana, rango "
            "de precio medio."
        ),
        "principle": "Extrae solo lo relevante de la query del usuario",
    },
}

print("Patrones de prompt engineering para privacidad:\n")
for pattern, info in privacy_prompt_patterns.items():
    print(f"  {pattern.upper().replace('_', ' ')}")
    print(f"    Principio: {info['principle']}")
    print(f"    ❌ Bad: \"{info['bad'][:60]}...\"")
    print(f"    ✅ Good: \"{info['good'][:60]}...\"")
    print()

Selective field inclusion por endpoint

from typing import Optional


class EndpointMinimizer:
    """Minimiza datos según el endpoint de la API."""

    ENDPOINT_FIELDS = {
        "/chat": {
            "include": ["query", "session_id", "language"],
            "context_fields": ["first_name"],
            "max_context_chars": 500,
        },
        "/search": {
            "include": ["query"],
            "context_fields": [],
            "max_context_chars": 200,
        },
        "/support": {
            "include": ["query", "ticket_id", "product_name", "error_code"],
            "context_fields": ["first_name", "purchase_date"],
            "max_context_chars": 1000,
        },
        "/recommendation": {
            "include": ["query", "preferences", "category"],
            "context_fields": ["purchase_history_summary"],
            "max_context_chars": 800,
        },
    }

    def minimize_for_endpoint(
        self,
        data: dict,
        endpoint: str,
    ) -> dict:
        """Filtra datos según la configuración del endpoint."""
        config = self.ENDPOINT_FIELDS.get(endpoint)
        if not config:
            return {"query": data.get("query", "")}

        result = {}
        for field in config["include"]:
            if field in data:
                result[field] = data[field]

        for field in config["context_fields"]:
            if field in data:
                value = str(data[field])
                max_chars = config["max_context_chars"]
                result[field] = value[:max_chars]

        return result


# --- Demostración ---

endpoint_minimizer = EndpointMinimizer()

full_data = {
    "query": "¿Tienen laptops con descuento?",
    "session_id": "sess-123",
    "first_name": "María",
    "full_name": "María García López",
    "email": "maria@test.com",
    "phone": "555-1234",
    "preferences": "electronics, budget",
    "purchase_history_summary": "5 compras en últimos 6 meses",
    "language": "es",
}

for endpoint in ["/chat", "/search", "/recommendation"]:
    minimized = endpoint_minimizer.minimize_for_endpoint(full_data, endpoint)
    print(f"  {endpoint}: {list(minimized.keys())}")

# Output esperado:
#   /chat: ['query', 'session_id', 'language', 'first_name']
#   /search: ['query']
#   /recommendation: ['query', 'preferences', 'purchase_history_summary']

Minimización de contexto RAG

@dataclass
class RAGContextMinimizer:
    """Minimiza el contexto RAG antes de enviarlo al LLM."""
    max_chunks: int = 3
    max_chars_per_chunk: int = 500
    max_total_chars: int = 2000
    remove_metadata: bool = True

    def minimize_context(
        self,
        chunks: list[dict],
        query: str,
    ) -> dict:
        """Reduce el contexto RAG al mínimo necesario."""
        selected = chunks[:self.max_chunks]

        minimized_chunks = []
        total_chars = 0

        for chunk in selected:
            text = chunk.get("content", chunk.get("text", ""))
            
            if len(text) > self.max_chars_per_chunk:
                text = text[:self.max_chars_per_chunk] + "..."

            if total_chars + len(text) > self.max_total_chars:
                remaining = self.max_total_chars - total_chars
                if remaining > 100:
                    text = text[:remaining] + "..."
                else:
                    break

            minimized_chunk = {"content": text}
            if not self.remove_metadata:
                minimized_chunk["source"] = chunk.get("source", "unknown")

            minimized_chunks.append(minimized_chunk)
            total_chars += len(text)

        return {
            "chunks": minimized_chunks,
            "original_count": len(chunks),
            "selected_count": len(minimized_chunks),
            "total_chars": total_chars,
            "estimated_tokens": total_chars // 4,
        }


# --- Demostración ---

chunks = [
    {"content": "Product A is a laptop with 16GB RAM. " * 20, "source": "catalog.pdf", "score": 0.95},
    {"content": "Product B is a tablet with 8GB RAM. " * 15, "source": "catalog.pdf", "score": 0.87},
    {"content": "Warranty covers 2 years. " * 10, "source": "warranty.pdf", "score": 0.82},
    {"content": "Return policy is 30 days. " * 10, "source": "returns.pdf", "score": 0.75},
    {"content": "Shipping takes 3-5 days. " * 10, "source": "shipping.pdf", "score": 0.60},
]

minimizer = RAGContextMinimizer(max_chunks=3, max_total_chars=1000)
result = minimizer.minimize_context(chunks, "laptop specs")

print(f"RAG Context Minimization:")
print(f"  Original chunks: {result['original_count']}")
print(f"  Selected chunks: {result['selected_count']}")
print(f"  Total chars: {result['total_chars']}")
print(f"  Estimated tokens: {result['estimated_tokens']}")

Conexión con el proyecto

El Data Minimizer se integra en el PII Protection Layer entre el scanner/redactor y la llamada al LLM:

User Input + Context
  │
  ▼
PIIScanner → detecta PII
  │
  ▼
PreLLMRedactor → redacta PII
  │
  ▼
DataMinimizer → reduce al mínimo        ← ESTE COMPONENTE
  │
  ▼
LLM Processing (con datos mínimos)
  │
  ▼
PostLLMRedactor → filtra PII del output

Troubleshooting

Problema 1: "La minimización reduce tanto el contexto que el LLM da respuestas genéricas"

Si el LLM no tiene suficiente contexto, genera respuestas vagas como "No tengo información suficiente."

Solución: Calibra los task_allowed_fields por tipo de tarea. Empieza con más campos y reduce gradualmente hasta encontrar el mínimo que mantiene la calidad. Mide la calidad de las respuestas con y sin minimización.

Problema 2: "No sé qué campos son necesarios para cada tarea"

Solución: Empieza con una política permisiva (incluir todo excepto FORBIDDEN), monitorea qué campos el LLM realmente usa en sus respuestas, y gradualmente excluye los que nunca aparecen en el output.

Problema 3: "La generalización de fechas pierde demasiada información"

Convertir "03/15/1990" a "1990s" puede ser demasiado agresivo para algunas tareas.

Solución: Ajusta la granularidad de generalización por tarea. Para verificación de edad, "adulto" puede ser suficiente. Para análisis de cohortes, el año exacto puede ser necesario.


Ejercicios

Ejercicio 1: Minimizer con métricas de reducción

Extiende el DataMinimizer para que calcule métricas detalladas de reducción de datos.

Ver solución
def calculate_minimization_metrics(
    original: dict, minimized: dict,
) -> dict:
    original_chars = sum(len(str(v)) for v in original.values())
    minimized_chars = sum(len(str(v)) for v in minimized.values())
    
    return {
        "original_fields": len(original),
        "minimized_fields": len(minimized),
        "field_reduction": f"{(1 - len(minimized)/len(original))*100:.0f}%",
        "original_chars": original_chars,
        "minimized_chars": minimized_chars,
        "char_reduction": f"{(1 - minimized_chars/original_chars)*100:.0f}%",
        "original_tokens_est": original_chars // 4,
        "minimized_tokens_est": minimized_chars // 4,
        "token_savings": (original_chars - minimized_chars) // 4,
    }


data = {
    "query": "order status", "order_id": "ORD-123",
    "name": "John Smith", "email": "john@test.com",
    "ssn": "123-45-6789", "address": "123 Main St, NYC",
}
minimized = {"query": "order status", "order_id": "ORD-123"}

metrics = calculate_minimization_metrics(data, minimized)
for k, v in metrics.items():
    print(f"  {k}: {v}")

Ejercicio 2: Policy engine configurable

Crea un engine de políticas que lea las reglas de minimización desde un diccionario de configuración.

Ver solución
MINIMIZATION_POLICIES = {
    "strict": {
        "forbidden": ["ssn", "credit_card", "password", "medical"],
        "exclude": ["email", "phone", "address", "full_name", "dob"],
        "include_all_else": False,
        "allowed": ["query", "session_id"],
    },
    "moderate": {
        "forbidden": ["ssn", "credit_card", "password"],
        "exclude": ["address", "medical"],
        "include_all_else": True,
        "allowed": [],
    },
}


def apply_policy(data: dict, policy_name: str) -> dict:
    policy = MINIMIZATION_POLICIES.get(policy_name)
    if not policy:
        raise ValueError(f"Unknown policy: {policy_name}")
    
    result = {}
    for key, value in data.items():
        key_lower = key.lower()
        if key_lower in policy["forbidden"]:
            continue
        if key_lower in policy["exclude"]:
            continue
        if policy["include_all_else"] or key_lower in policy["allowed"]:
            result[key] = value
    return result


data = {
    "query": "help", "email": "test@test.com",
    "ssn": "123-45-6789", "name": "John",
}
print("Strict:", apply_policy(data, "strict"))
print("Moderate:", apply_policy(data, "moderate"))

# Output:
# Strict: {'query': 'help'}
# Moderate: {'query': 'help', 'email': 'test@test.com', 'name': 'John'}

Ejercicio 3: Context builder que arma prompts minimizados

Crea una función que construya un prompt para el LLM usando solo los datos minimizados.

Ver solución
def build_minimized_prompt(
    minimized_data: dict,
    system_template: str = "Eres un asistente. Responde basándote solo en el contexto proporcionado.",
) -> list[dict]:
    context_parts = []
    query = minimized_data.pop("query", "")
    
    for key, value in minimized_data.items():
        context_parts.append(f"{key}: {value}")
    
    context_str = "\n".join(context_parts) if context_parts else "Sin contexto adicional."
    
    return [
        {"role": "system", "content": system_template},
        {"role": "user", "content": f"Contexto:\n{context_str}\n\nPregunta: {query}"},
    ]


minimized = {
    "query": "¿Cuál es el estado de mi pedido?",
    "order_id": "ORD-12345",
    "order_status": "shipped",
    "first_name": "María",
}

messages = build_minimized_prompt(minimized)
for msg in messages:
    print(f"  [{msg['role']}]: {msg['content'][:80]}...")

Ejercicio 4: Detector de datos innecesarios en prompts existentes

Crea una función que analice un prompt existente e identifique datos que podrían eliminarse.

Ver solución
from presidio_analyzer import AnalyzerEngine


def audit_prompt_data(prompt: str) -> dict:
    """Identifica datos potencialmente innecesarios en un prompt."""
    analyzer = AnalyzerEngine()
    results = analyzer.analyze(text=prompt, language="en", score_threshold=0.4)
    
    unnecessary = []
    for r in results:
        entity_text = prompt[r.start:r.end]
        unnecessary.append({
            "type": r.entity_type,
            "text": entity_text,
            "score": r.score,
            "recommendation": (
                "REMOVE" if r.entity_type in ("US_SSN", "CREDIT_CARD")
                else "CONSIDER_REMOVING" if r.entity_type in ("EMAIL_ADDRESS", "PHONE_NUMBER")
                else "REVIEW"
            ),
        })
    
    return {
        "prompt_length": len(prompt),
        "pii_found": len(unnecessary),
        "items": unnecessary,
        "estimated_reduction": f"{sum(len(i['text']) for i in unnecessary)} chars removable",
    }


prompt = (
    "User John Smith (john@test.com, SSN: 123-45-6789) "
    "asks about product pricing. His account is ACC-123."
)
audit = audit_prompt_data(prompt)
print(f"PII in prompt: {audit['pii_found']}")
for item in audit['items']:
    print(f"  [{item['recommendation']}] {item['type']}: \"{item['text']}\"")

Resumen

  • 🔑 Data minimization es enviar solo los datos estrictamente necesarios al LLM — si el dato no llega al modelo, no puede filtrarse
  • 🔑 El principio viene de GDPR Art. 5(1)(c) y es una de las defensas más efectivas contra LLM02 porque reduce la superficie de ataque
  • 🔑 Clasificar campos por necesidad (REQUIRED, USEFUL, UNNECESSARY, FORBIDDEN) permite automatizar qué datos se envían al modelo por tarea
  • 🔑 El DataMinimizer filtra, excluye y generaliza datos automáticamente según políticas configurables por endpoint y tarea
  • 🔑 La truncación inteligente de contexto maximiza la relevancia dentro del token budget — la estrategia "smart" preserva inicio y final
  • 🔑 El prompt engineering para privacidad permite obtener respuestas útiles sin enviar datos personales: referencia por ID, agregados en lugar de individuales
  • 🔑 La minimización por endpoint permite políticas diferentes: /search solo necesita la query, /support necesita ticket y producto
  • 🔑 El trade-off es privacidad vs calidad: menos datos = más seguro pero respuestas potencialmente menos personalizadas

Recursos adicionales

  1. GDPR Art. 5 — Data Minimisation — Texto oficial del principio de minimización de datos
  2. NIST Privacy Framework — Data Minimization — Framework del NIST que incluye prácticas de minimización
  3. Data Minimization in Machine Learning (ACM) — Paper sobre minimización de datos en ML
  4. OpenAI Token Counter (tiktoken) — Librería oficial de OpenAI para contar tokens
  5. OWASP Data Minimization — Principio de minimización en el contexto de seguridad web
  6. Microsoft Privacy Principles — Principios de privacidad incluyendo minimización
  7. ICO Guide to Data Minimisation — Guía del regulador UK sobre minimización

Creado: Marzo 2026 Versión: 1.0