Módulo 6: Data Privacy & PII Protection

4. PII Redaction: Antes y Después del LLM

Siguiente →

Descripción

En la cápsula anterior construiste un PIIScanner que detecta más de 30 tipos de PII con Presidio, spaCy y reconocedores custom. Detectar PII es el primer paso — ahora necesitas decidir qué hacer con los datos detectados. ¿Los eliminas? ¿Los reemplazas con un placeholder? ¿Los hasheas para poder auditar? ¿Mantienes una versión reversible para reconstruir la respuesta después?

La redacción de PII no es una operación simple de "buscar y reemplazar". Cada estrategia tiene trade-offs de utilidad, privacidad y compliance. Y el momento en que aplicas la redacción — antes del LLM (pre-LLM) o después del LLM (post-LLM) — cambia completamente el diseño y las implicaciones.

En esta cápsula construyes dos componentes del PII Protection Layer: el Pre-LLM Redactor que sanitiza datos antes de enviarlos al modelo, y el Post-LLM Redactor que filtra PII de los outputs del modelo. Ambos se reutilizan directamente en el proyecto de la cápsula 08.

Estrategias de redacción

Hay cinco estrategias principales para manejar PII detectado. Cada una tiene un perfil diferente de privacidad vs utilidad:

from dataclasses import dataclass
from enum import Enum


class RedactionStrategy(Enum):
    MASK = "mask"
    REPLACE = "replace"
    HASH = "hash"
    GENERALIZE = "generalize"
    SYNTHETIC = "synthetic"


@dataclass
class StrategyProfile:
    strategy: RedactionStrategy
    description: str
    example_input: str
    example_output: str
    reversible: bool
    privacy_level: str
    utility_preserved: str
    use_case: str


strategies = [
    StrategyProfile(
        strategy=RedactionStrategy.MASK,
        description="Reemplaza con placeholder genérico del tipo",
        example_input="Contact john@acme.com for info",
        example_output="Contact <EMAIL_ADDRESS> for info",
        reversible=False,
        privacy_level="Alta",
        utility_preserved="Baja — se pierde el dato",
        use_case="Logs, auditoría, datos que no necesitan reconstrucción",
    ),
    StrategyProfile(
        strategy=RedactionStrategy.REPLACE,
        description="Reemplaza con un valor ficticio pero del mismo tipo",
        example_input="Call María García at 555-1234",
        example_output="Call Jane Doe at 000-0000",
        reversible=False,
        privacy_level="Alta",
        utility_preserved="Media — mantiene estructura",
        use_case="Testing, demos, datasets de entrenamiento",
    ),
    StrategyProfile(
        strategy=RedactionStrategy.HASH,
        description="Reemplaza con hash del valor original",
        example_input="Email: maria@test.com",
        example_output="Email: <EMAIL_a1b2c3d4>",
        reversible=True,
        privacy_level="Media — el hash es determinístico",
        utility_preserved="Baja — pero permite correlación",
        use_case="Auditoría donde necesitas rastrear sin exponer",
    ),
    StrategyProfile(
        strategy=RedactionStrategy.GENERALIZE,
        description="Reduce la especificidad del dato",
        example_input="Born on 03/15/1990 in New York, NY 10001",
        example_output="Born in 1990s in New York area",
        reversible=False,
        privacy_level="Media — se pierde precisión",
        utility_preserved="Alta — mantiene contexto general",
        use_case="Análisis donde necesitas patrones sin identificar individuos",
    ),
    StrategyProfile(
        strategy=RedactionStrategy.SYNTHETIC,
        description="Reemplaza con datos sintéticos realistas",
        example_input="John Smith, john@real.com, 555-123-4567",
        example_output="Alex Johnson, alex@example.com, 555-000-0000",
        reversible=True,
        privacy_level="Alta — los datos son ficticios",
        utility_preserved="Alta — el LLM recibe datos 'realistas'",
        use_case="Pre-LLM redaction donde el modelo necesita contexto",
    ),
]

print("Estrategias de redacción de PII:\n")
for s in strategies:
    print(f"  {s.strategy.value.upper()}")
    print(f"    {s.description}")
    print(f"    Input:  \"{s.example_input}\"")
    print(f"    Output: \"{s.example_output}\"")
    print(f"    Reversible: {s.reversible} | Privacidad: {s.privacy_level}")
    print(f"    Caso de uso: {s.use_case}")
    print()

Presidio Anonymizer: redacción con el engine oficial

Presidio Anonymizer trabaja con los resultados del Analyzer para aplicar redacción automática.

Masking básico

from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()

text = (
    "My name is John Smith. My email is john@acme.com "
    "and my SSN is 078-05-1120."
)

analyzer_results = analyzer.analyze(text=text, language="en")

anonymized = anonymizer.anonymize(
    text=text,
    analyzer_results=analyzer_results,
)

print(f"Original:   {text}")
print(f"Anonymized: {anonymized.text}")
print(f"\nItems anonymized:")
for item in anonymized.items:
    print(f"  {item.entity_type}: \"{item.text}\" → \"{item.operator}\"")

# Output esperado:
# Original:   My name is John Smith. My email is john@acme.com and my SSN is 078-05-1120.
# Anonymized: My name is <PERSON>. My email is <EMAIL_ADDRESS> and my SSN is <US_SSN>.

Operadores de anonimización

from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig

anonymizer = AnonymizerEngine()

text = (
    "Contact María García at maria@test.com or 555-123-4567. "
    "SSN: 123-45-6789."
)

analyzer = AnalyzerEngine()
results = analyzer.analyze(text=text, language="en")

operators = {
    "PERSON": OperatorConfig("replace", {"new_value": "[NOMBRE_REDACTADO]"}),
    "EMAIL_ADDRESS": OperatorConfig("replace", {"new_value": "redacted@example.com"}),
    "PHONE_NUMBER": OperatorConfig("mask", {
        "type": "mask",
        "masking_char": "*",
        "chars_to_mask": 8,
        "from_end": False,
    }),
    "US_SSN": OperatorConfig("replace", {"new_value": "***-**-****"}),
    "DEFAULT": OperatorConfig("replace", {"new_value": "[REDACTED]"}),
}

anonymized = anonymizer.anonymize(
    text=text,
    analyzer_results=results,
    operators=operators,
)

print(f"Original:   {text}")
print(f"Anonymized: {anonymized.text}")

# Output esperado:
# Original:   Contact María García at maria@test.com or 555-123-4567. SSN: 123-45-6789.
# Anonymized: Contact [NOMBRE_REDACTADO] at redacted@example.com or ********4567. SSN: ***-**-****.

Hash operator

import hashlib
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig

anonymizer = AnonymizerEngine()

text = "Email: maria@test.com, Phone: 555-123-4567"

analyzer = AnalyzerEngine()
results = analyzer.analyze(text=text, language="en")

operators = {
    "DEFAULT": OperatorConfig("hash", {"hash_type": "sha256"}),
}

anonymized = anonymizer.anonymize(
    text=text,
    analyzer_results=results,
    operators=operators,
)

print(f"Original:   {text}")
print(f"Hashed:     {anonymized.text}")

# Output esperado:
# Original:   Email: maria@test.com, Phone: 555-123-4567
# Hashed:     Email: <hash_sha256>, Phone: <hash_sha256>

Pre-LLM Redaction: sanitizar antes de enviar al modelo

La redacción pre-LLM es la defensa más efectiva contra LLM02. Si el dato nunca llega al modelo, el modelo no puede filtrarlo.

from dataclasses import dataclass, field
from typing import Optional
import hashlib
import json

from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig


@dataclass
class RedactionMapping:
    """Mapeo para reconstruir datos después del LLM."""
    placeholder: str
    original: str
    entity_type: str


@dataclass
class PreLLMRedactionResult:
    original_text: str
    redacted_text: str
    redaction_count: int
    mappings: list[RedactionMapping] = field(default_factory=list)
    reversible: bool = False


class PreLLMRedactor:
    """Redacta PII antes de enviar al LLM."""

    def __init__(
        self,
        strategy: str = "mask",
        reversible: bool = False,
        entities_to_redact: Optional[list[str]] = None,
        score_threshold: float = 0.5,
    ):
        self.strategy = strategy
        self.reversible = reversible
        self.entities_to_redact = entities_to_redact
        self.score_threshold = score_threshold
        self.analyzer = AnalyzerEngine()
        self.anonymizer = AnonymizerEngine()

    def redact(self, text: str, language: str = "en") -> PreLLMRedactionResult:
        """Redacta PII del texto."""
        results = self.analyzer.analyze(
            text=text,
            language=language,
            entities=self.entities_to_redact,
            score_threshold=self.score_threshold,
        )

        if not results:
            return PreLLMRedactionResult(
                original_text=text,
                redacted_text=text,
                redaction_count=0,
            )

        mappings = []

        if self.reversible:
            operators = {}
            counter = {}
            for r in sorted(results, key=lambda x: x.start):
                entity_text = text[r.start:r.end]
                count = counter.get(r.entity_type, 0) + 1
                counter[r.entity_type] = count
                placeholder = f"<{r.entity_type}_{count}>"
                mappings.append(RedactionMapping(
                    placeholder=placeholder,
                    original=entity_text,
                    entity_type=r.entity_type,
                ))
        
            operators = {
                "DEFAULT": OperatorConfig("replace", {"new_value": "<REDACTED>"}),
            }
            anonymized = self.anonymizer.anonymize(
                text=text,
                analyzer_results=results,
                operators=operators,
            )
            
            redacted = anonymized.text
            for mapping in mappings:
                redacted = redacted.replace("<REDACTED>", mapping.placeholder, 1)
        else:
            operators = {
                "DEFAULT": OperatorConfig("replace", {"new_value": "<REDACTED>"}),
            }
            
            if self.strategy == "mask":
                operators = {}
            
            anonymized = self.anonymizer.anonymize(
                text=text,
                analyzer_results=results,
                operators=operators,
            )
            redacted = anonymized.text

        return PreLLMRedactionResult(
            original_text=text,
            redacted_text=redacted,
            redaction_count=len(results),
            mappings=mappings,
            reversible=self.reversible,
        )


# --- Demostración ---

redactor = PreLLMRedactor(strategy="mask", reversible=False)

user_input = (
    "Mi nombre es María García, mi email es maria@empresa.com "
    "y mi SSN es 123-45-6789. ¿Cuál es el estado de mi pedido?"
)

result = redactor.redact(user_input)

print(f"Original:  {result.original_text}")
print(f"Redacted:  {result.redacted_text}")
print(f"Count:     {result.redaction_count}")
print(f"\nEl LLM recibe el texto redactado y nunca ve el PII real.")

# Output esperado:
# Original:  Mi nombre es María García, mi email es maria@empresa.com y mi SSN es 123-45-6789. ¿Cuál es el estado de mi pedido?
# Redacted:  Mi nombre es <PERSON>, mi email es <EMAIL_ADDRESS> y mi SSN es <US_SSN>. ¿Cuál es el estado de mi pedido?
# Count:     3

Redacción reversible

redactor_reversible = PreLLMRedactor(
    strategy="mask",
    reversible=True,
)

text = "Contact John Smith at john@test.com about order #12345"
result = redactor_reversible.redact(text)

print(f"Redacted: {result.redacted_text}")
print(f"Mappings:")
for m in result.mappings:
    print(f"  {m.placeholder} → \"{m.original}\" ({m.entity_type})")


def reconstruct(redacted_text: str, mappings: list[RedactionMapping]) -> str:
    """Reconstruye el texto original usando los mappings."""
    text = redacted_text
    for mapping in mappings:
        text = text.replace(mapping.placeholder, mapping.original)
    return text


reconstructed = reconstruct(result.redacted_text, result.mappings)
print(f"\nReconstructed: {reconstructed}")
print(f"Matches original: {reconstructed == text}")

# Output esperado:
# Redacted: Contact <PERSON_1> at <EMAIL_ADDRESS_1> about order #12345
# Mappings:
#   <PERSON_1> → "John Smith" (PERSON)
#   <EMAIL_ADDRESS_1> → "john@test.com" (EMAIL_ADDRESS)
#
# Reconstructed: Contact John Smith at john@test.com about order #12345
# Matches original: True

Post-LLM Redaction: filtrar outputs del modelo

Aunque redactes PII antes del LLM, el modelo puede generar PII en su output — datos memorizados del training, datos del contexto RAG, o simplemente datos fabricados que parecen reales.

@dataclass
class PostLLMRedactionResult:
    original_output: str
    redacted_output: str
    pii_found: int
    entities_redacted: list[dict] = field(default_factory=list)
    action: str = "pass"


class PostLLMRedactor:
    """Redacta PII de outputs del LLM antes de enviar al usuario."""

    def __init__(
        self,
        score_threshold: float = 0.5,
        block_on_critical: bool = True,
        redact_strategy: str = "mask",
    ):
        self.score_threshold = score_threshold
        self.block_on_critical = block_on_critical
        self.redact_strategy = redact_strategy
        self.analyzer = AnalyzerEngine()
        self.anonymizer = AnonymizerEngine()
        
        self.critical_entities = {"US_SSN", "CREDIT_CARD", "IBAN_CODE"}

    def redact(
        self,
        llm_output: str,
        language: str = "en",
    ) -> PostLLMRedactionResult:
        """Escanea y redacta PII del output del LLM."""
        results = self.analyzer.analyze(
            text=llm_output,
            language=language,
            score_threshold=self.score_threshold,
        )

        if not results:
            return PostLLMRedactionResult(
                original_output=llm_output,
                redacted_output=llm_output,
                pii_found=0,
                action="pass",
            )

        has_critical = any(
            r.entity_type in self.critical_entities for r in results
        )

        if has_critical and self.block_on_critical:
            return PostLLMRedactionResult(
                original_output=llm_output,
                redacted_output=(
                    "Lo siento, no puedo proporcionar esa información "
                    "por razones de privacidad."
                ),
                pii_found=len(results),
                entities_redacted=[
                    {
                        "type": r.entity_type,
                        "text": llm_output[r.start:r.end],
                        "score": r.score,
                    }
                    for r in results
                ],
                action="blocked",
            )

        anonymized = self.anonymizer.anonymize(
            text=llm_output,
            analyzer_results=results,
        )

        return PostLLMRedactionResult(
            original_output=llm_output,
            redacted_output=anonymized.text,
            pii_found=len(results),
            entities_redacted=[
                {
                    "type": r.entity_type,
                    "text": llm_output[r.start:r.end],
                    "score": r.score,
                }
                for r in results
            ],
            action="redacted",
        )


# --- Demostración ---

post_redactor = PostLLMRedactor(block_on_critical=True)

outputs = [
    "El producto cuesta $49.99 y está disponible en tienda.",
    "Claro, el email de soporte es support@acme.com. Contacta a Juan.",
    "Tu SSN 123-45-6789 está asociado a la cuenta 4111-1111-1111-1111.",
]

for output in outputs:
    result = post_redactor.redact(output)
    print(f"Output LLM: \"{output[:60]}...\"")
    print(f"  Action: {result.action}")
    print(f"  PII found: {result.pii_found}")
    if result.action != "pass":
        print(f"  Redacted:  \"{result.redacted_output[:60]}...\"")
    print()

# Output esperado:
# Output LLM: "El producto cuesta $49.99 y está disponible en tienda...."
#   Action: pass
#   PII found: 0
#
# Output LLM: "Claro, el email de soporte es support@acme.com. Contacta..."
#   Action: redacted
#   PII found: 2
#   Redacted:  "Claro, el email de soporte es <EMAIL_ADDRESS>. Contacta a..."
#
# Output LLM: "Tu SSN 123-45-6789 está asociado a la cuenta 4111-1111-11..."
#   Action: blocked
#   PII found: 2
#   Redacted:  "Lo siento, no puedo proporcionar esa información por razo..."

Redacción en pipelines RAG

En un pipeline RAG, hay tres puntos donde puedes redactar PII:

class RAGPIIProtector:
    """Protege PII en cada etapa del pipeline RAG."""

    def __init__(self):
        self.analyzer = AnalyzerEngine()
        self.anonymizer = AnonymizerEngine()

    def redact_at_index_time(self, document: str) -> dict:
        """Redacta PII cuando indexas documentos en el vector store."""
        results = self.analyzer.analyze(text=document, language="en")
        if not results:
            return {"document": document, "redacted": False, "pii_count": 0}

        anonymized = self.anonymizer.anonymize(
            text=document, analyzer_results=results,
        )
        return {
            "document": anonymized.text,
            "redacted": True,
            "pii_count": len(results),
        }

    def redact_post_retrieval(self, chunks: list[str]) -> list[dict]:
        """Redacta PII de chunks recuperados antes de inyectar en prompt."""
        redacted_chunks = []
        for chunk in chunks:
            results = self.analyzer.analyze(text=chunk, language="en")
            if results:
                anonymized = self.anonymizer.anonymize(
                    text=chunk, analyzer_results=results,
                )
                redacted_chunks.append({
                    "chunk": anonymized.text,
                    "pii_found": len(results),
                })
            else:
                redacted_chunks.append({"chunk": chunk, "pii_found": 0})
        return redacted_chunks

    def redact_in_query(self, query: str) -> dict:
        """Redacta PII del query del usuario antes de buscar."""
        results = self.analyzer.analyze(text=query, language="en")
        if not results:
            return {"query": query, "redacted": False}

        anonymized = self.anonymizer.anonymize(
            text=query, analyzer_results=results,
        )
        return {"query": anonymized.text, "redacted": True}


# --- Demostración ---

protector = RAGPIIProtector()

document = (
    "Customer John Smith (john@acme.com) reported issue with "
    "order #12345. Phone: 555-123-4567."
)

indexed = protector.redact_at_index_time(document)
print(f"Index-time redaction:")
print(f"  Original: {document[:60]}...")
print(f"  Indexed:  {indexed['document'][:60]}...")
print(f"  PII removed: {indexed['pii_count']}")

retrieved = protector.redact_post_retrieval([
    "Call María at 555-0100 for help",
    "Product ships in 2-3 days",
])
print(f"\nPost-retrieval redaction:")
for r in retrieved:
    print(f"  Chunk: \"{r['chunk'][:40]}...\" (PII: {r['pii_found']})")

query_result = protector.redact_in_query(
    "My name is Carlos and my email is carlos@test.com, what's my order status?"
)
print(f"\nQuery redaction:")
print(f"  Redacted query: \"{query_result['query'][:60]}...\"")

# Output esperado:
# Index-time redaction:
#   Original: Customer John Smith (john@acme.com) reported issue with...
#   Indexed:  Customer <PERSON> (<EMAIL_ADDRESS>) reported issue with...
#   PII removed: 3
#
# Post-retrieval redaction:
#   Chunk: "Call <PERSON> at <PHONE_NUMBER> for h..." (PII: 2)
#   Chunk: "Product ships in 2-3 days..." (PII: 0)
#
# Query redaction:
#   Redacted query: "My name is <PERSON> and my email is <EMAIL_ADDRESS>,..."

Trade-offs de cada punto de redacción

rag_redaction_tradeoffs = {
    "index_time": {
        "pros": [
            "PII nunca se almacena en el vector store",
            "Una sola redacción para todas las queries",
            "Reduce riesgo de exposición en caso de breach del vector store",
        ],
        "cons": [
            "Pérdida permanente del dato original (si no guardas backup)",
            "La búsqueda semántica puede verse afectada por placeholders",
            "No puedes re-procesar con mejores detectores sin re-indexar",
        ],
        "recommendation": "Ideal para datos sensibles permanentes (SSN, tarjetas)",
    },
    "post_retrieval": {
        "pros": [
            "Documentos originales intactos para re-procesamiento",
            "Puedes actualizar la lógica de redacción sin re-indexar",
            "Solo redactas lo que se recupera, no todo el corpus",
        ],
        "cons": [
            "PII almacenado en el vector store (riesgo en breach)",
            "Latencia adicional en cada query (scan + redacción)",
            "Necesitas garantizar que la redacción siempre se ejecute",
        ],
        "recommendation": "Ideal cuando necesitas los documentos originales para otros usos",
    },
    "query_time": {
        "pros": [
            "Protege al usuario de enviar su propio PII al sistema",
            "Complementa las otras estrategias",
        ],
        "cons": [
            "Puede afectar la calidad del retrieval",
            "No protege contra PII en los documentos",
        ],
        "recommendation": "Siempre recomendado como capa adicional",
    },
}

print("Trade-offs de redacción en RAG:\n")
for point, info in rag_redaction_tradeoffs.items():
    print(f"  {point.upper().replace('_', ' ')}:")
    print(f"    Recomendación: {info['recommendation']}")
    print(f"    Pros: {', '.join(info['pros'][:2])}")
    print(f"    Cons: {', '.join(info['cons'][:2])}")
    print()

Manteniendo contexto mientras proteges PII

El desafío más grande de la redacción pre-LLM es mantener suficiente contexto para que el modelo genere respuestas útiles.

def redact_with_context_preservation(
    text: str,
    preserve_entity_types: bool = True,
) -> dict:
    """
    Redacta PII preservando el tipo de entidad
    para que el LLM entienda el contexto.
    """
    analyzer = AnalyzerEngine()
    anonymizer = AnonymizerEngine()
    
    results = analyzer.analyze(text=text, language="en")
    
    if not results:
        return {"text": text, "redacted": False}
    
    if preserve_entity_types:
        counter = {}
        operators = {}
        for r in sorted(results, key=lambda x: x.start):
            count = counter.get(r.entity_type, 0) + 1
            counter[r.entity_type] = count
            
        operators = {}
        
        anonymized = anonymizer.anonymize(
            text=text,
            analyzer_results=results,
        )
        return {
            "text": anonymized.text,
            "redacted": True,
            "strategy": "type_preserving",
        }
    else:
        operators = {
            "DEFAULT": OperatorConfig(
                "replace", {"new_value": "[REDACTED]"}
            ),
        }
        anonymized = anonymizer.anonymize(
            text=text,
            analyzer_results=results,
            operators=operators,
        )
        return {
            "text": anonymized.text,
            "redacted": True,
            "strategy": "generic",
        }


text = (
    "The customer John Smith from New York called about "
    "his order. His email is john@acme.com."
)

result_typed = redact_with_context_preservation(text, preserve_entity_types=True)
result_generic = redact_with_context_preservation(text, preserve_entity_types=False)

print(f"Original:       {text}")
print(f"Type-preserving: {result_typed['text']}")
print(f"Generic:         {result_generic['text']}")
print()
print("El LLM entiende mejor '<PERSON> from <LOCATION>'")
print("que '[REDACTED] from [REDACTED]'")

# Output esperado:
# Original:       The customer John Smith from New York called about his order. His email is john@acme.com.
# Type-preserving: The customer <PERSON> from <LOCATION> called about his order. His email is <EMAIL_ADDRESS>.
# Generic:         The customer [REDACTED] from [REDACTED] called about his order. His email is [REDACTED].

Reversible vs irreversible redaction

class ReversibleRedactor:
    """Redactor que permite reconstruir el texto original."""

    def __init__(self):
        self.analyzer = AnalyzerEngine()
        self._mapping_store: dict[str, dict[str, str]] = {}

    def redact(self, text: str, request_id: str) -> str:
        """Redacta y guarda mappings para reversión."""
        results = self.analyzer.analyze(text=text, language="en")
        if not results:
            return text

        mappings = {}
        redacted = text

        for r in sorted(results, key=lambda x: x.end, reverse=True):
            original_value = text[r.start:r.end]
            placeholder = f"<{r.entity_type}_{hash(original_value) % 10000:04d}>"
            mappings[placeholder] = original_value
            redacted = redacted[:r.start] + placeholder + redacted[r.end:]

        self._mapping_store[request_id] = mappings
        return redacted

    def reverse(self, redacted_text: str, request_id: str) -> str:
        """Reconstruye el texto original."""
        mappings = self._mapping_store.get(request_id, {})
        result = redacted_text
        for placeholder, original in mappings.items():
            result = result.replace(placeholder, original)
        return result

    def clear_mapping(self, request_id: str):
        """Elimina los mappings de un request (para retention)."""
        self._mapping_store.pop(request_id, None)


# --- Demostración ---

redactor = ReversibleRedactor()

text = "Contact María García at maria@test.com about order"
request_id = "req-001"

redacted = redactor.redact(text, request_id)
print(f"Original: {text}")
print(f"Redacted: {redacted}")

reversed_text = redactor.reverse(redacted, request_id)
print(f"Reversed: {reversed_text}")
print(f"Match:    {reversed_text == text}")

redactor.clear_mapping(request_id)
print(f"Mapping cleared for retention compliance")

# Output esperado:
# Original: Contact María García at maria@test.com about order
# Redacted: Contact <PERSON_XXXX> at <EMAIL_ADDRESS_XXXX> about order
# Reversed: Contact María García at maria@test.com about order
# Match:    True
# Mapping cleared for retention compliance

Conexión con el proyecto

Los componentes de esta cápsula se integran en el PII Protection Layer:

PII Protection Layer
├── PIIScanner (Cápsula 03)
├── PreLLMRedactor (ESTA CÁPSULA)     ← Redacción antes del LLM
├── PostLLMRedactor (ESTA CÁPSULA)    ← Redacción después del LLM
├── Data Minimizer (Cápsula 05)
├── Retention Scheduler (Cápsula 06)
└── Audit Logger

El flujo del pipeline con redacción:

User Input
  │
  ▼
PIIScanner → detecta PII
  │
  ▼
PreLLMRedactor → redacta PII del input
  │
  ▼
Data Minimizer → envía solo lo necesario
  │
  ▼
LLM Processing
  │
  ▼
PostLLMRedactor → redacta PII del output
  │
  ▼
[Opcional] Reconstrucción reversible
  │
  ▼
Response al usuario

Troubleshooting

Problema 1: "La redacción rompe la estructura del JSON"

Si el texto contiene JSON y los valores se redactan, el JSON puede quedar malformado.

Solución: Parsea el JSON primero, redacta los valores individuales, y reconstruye el JSON:

import json

def redact_json_values(data: dict, redactor) -> dict:
    result = {}
    for key, value in data.items():
        if isinstance(value, str):
            redacted = redactor.redact(value)
            result[key] = redacted.redacted_text
        elif isinstance(value, dict):
            result[key] = redact_json_values(value, redactor)
        else:
            result[key] = value
    return result

Problema 2: "El LLM genera respuestas peores con textos redactados"

Los placeholders como <PERSON> confunden al modelo y producen respuestas de menor calidad.

Solución: Usa la estrategia SYNTHETIC para pre-LLM redaction. El modelo recibe datos "realistas" ficticios en lugar de placeholders, y la respuesta se reconstruye con los datos reales usando redacción reversible.

Problema 3: "La redacción reversible es un riesgo de seguridad"

Los mappings de la redacción reversible contienen los datos originales. Si se filtran, la redacción fue inútil.

Solución: Guarda los mappings en memoria (no en logs), cifra los mappings at rest, y elimínalos inmediatamente después de la reconstrucción. Usa la función clear_mapping() como parte del flujo.

Problema 4: "Presidio no detecta ciertos tipos de PII en outputs del LLM"

El modelo puede generar PII en formatos que Presidio no reconoce, como "su número de teléfono es cinco-cinco-cinco, ciento veintitrés, cuarenta y cinco, sesenta y siete".

Solución: Agrega reconocedores custom para formatos verbalizados, o usa una capa de normalización que convierta texto verbalizado a formato estándar antes de pasar por Presidio.

Ejercicios

Ejercicio 1: Redactor con estrategia configurable por tipo de entidad

Crea un redactor que aplique diferentes estrategias según el tipo de PII.

Ver solución 
from presidio_anonymizer.entities import OperatorConfig

class SmartRedactor:
    STRATEGY_MAP = {
        "PERSON": OperatorConfig("replace", {"new_value": "[PERSONA]"}),
        "EMAIL_ADDRESS": OperatorConfig("replace", {"new_value": "redacted@example.com"}),
        "PHONE_NUMBER": OperatorConfig("mask", {
            "type": "mask", "masking_char": "*",
            "chars_to_mask": 7, "from_end": False,
        }),
        "US_SSN": OperatorConfig("replace", {"new_value": "***-**-****"}),
        "CREDIT_CARD": OperatorConfig("replace", {"new_value": "****-****-****-****"}),
    }

    def __init__(self):
        self.analyzer = AnalyzerEngine()
        self.anonymizer = AnonymizerEngine()

    def redact(self, text: str) -> str:
        results = self.analyzer.analyze(text=text, language="en")
        if not results:
            return text

        operators = {
            entity_type: config
            for entity_type, config in self.STRATEGY_MAP.items()
        }
        operators["DEFAULT"] = OperatorConfig("replace", {"new_value": "[REDACTED]"})

        anonymized = self.anonymizer.anonymize(
            text=text, analyzer_results=results, operators=operators,
        )
        return anonymized.text


smart = SmartRedactor()
text = "John Smith, john@test.com, 555-123-4567, SSN: 123-45-6789"
print(f"Original: {text}")
print(f"Redacted: {smart.redact(text)}")

# Output esperado:
# Original: John Smith, john@test.com, 555-123-4567, SSN: 123-45-6789
# Redacted: [PERSONA], redacted@example.com, *******4567, SSN: ***-**-****

Ejercicio 2: Comparador pre-LLM vs post-LLM

Crea una función que compare qué PII se detecta en el input vs el output del LLM.

Ver solución 
def compare_pii_flow(user_input: str, llm_output: str) -> dict:
    analyzer = AnalyzerEngine()
    
    input_results = analyzer.analyze(text=user_input, language="en")
    output_results = analyzer.analyze(text=llm_output, language="en")
    
    input_types = {r.entity_type for r in input_results}
    output_types = {r.entity_type for r in output_results}
    
    new_in_output = output_types - input_types
    
    return {
        "input_pii_count": len(input_results),
        "output_pii_count": len(output_results),
        "input_pii_types": list(input_types),
        "output_pii_types": list(output_types),
        "new_pii_in_output": list(new_in_output),
        "risk": "HIGH" if new_in_output else "LOW",
        "explanation": (
            f"El LLM introdujo nuevos tipos de PII: {new_in_output}"
            if new_in_output
            else "El LLM no introdujo PII nuevo"
        ),
    }


result = compare_pii_flow(
    user_input="What is my order status?",
    llm_output="Your order is ready, John. Contact support@acme.com for pickup.",
)

print(f"Input PII: {result['input_pii_count']}")
print(f"Output PII: {result['output_pii_count']}")
print(f"New PII in output: {result['new_pii_in_output']}")
print(f"Risk: {result['risk']}")
print(f"Explanation: {result['explanation']}")

Ejercicio 3: Redactor con audit trail

Agrega un audit trail al redactor que registre cada redacción con timestamp y metadata.

Ver solución 
from datetime import datetime, timezone


class AuditedRedactor:
    def __init__(self):
        self.analyzer = AnalyzerEngine()
        self.anonymizer = AnonymizerEngine()
        self.audit_log: list[dict] = []

    def redact(self, text: str, request_id: str = "unknown") -> str:
        results = self.analyzer.analyze(text=text, language="en")
        
        entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "request_id": request_id,
            "text_length": len(text),
            "pii_count": len(results),
            "entity_types": [r.entity_type for r in results],
            "action": "redacted" if results else "clean",
        }
        self.audit_log.append(entry)
        
        if not results:
            return text
        
        anonymized = self.anonymizer.anonymize(
            text=text, analyzer_results=results,
        )
        return anonymized.text

    def get_audit_summary(self) -> dict:
        if not self.audit_log:
            return {"total": 0}
        return {
            "total_requests": len(self.audit_log),
            "redacted": sum(1 for e in self.audit_log if e["action"] == "redacted"),
            "clean": sum(1 for e in self.audit_log if e["action"] == "clean"),
            "total_pii_found": sum(e["pii_count"] for e in self.audit_log),
        }


audited = AuditedRedactor()
audited.redact("Call John at 555-1234", "req-1")
audited.redact("The sky is blue", "req-2")
audited.redact("Email: test@example.com", "req-3")

import json
print(json.dumps(audited.get_audit_summary(), indent=2))

Ejercicio 4: Pipeline RAG con redacción en tres puntos

Implementa un mini-pipeline RAG que redacte PII en indexación, retrieval y query.

Ver solución 
class SecureRAGPipeline:
    def __init__(self):
        self.protector = RAGPIIProtector()
        self.indexed_docs: list[str] = []

    def index_document(self, doc: str) -> dict:
        result = self.protector.redact_at_index_time(doc)
        self.indexed_docs.append(result["document"])
        return result

    def search_and_protect(self, query: str) -> dict:
        query_result = self.protector.redact_in_query(query)
        safe_query = query_result["query"]
        
        retrieved = [
            doc for doc in self.indexed_docs
            if any(word in doc.lower() for word in safe_query.lower().split()[:3])
        ]
        
        protected_chunks = self.protector.redact_post_retrieval(retrieved)
        
        return {
            "original_query": query,
            "safe_query": safe_query,
            "chunks_found": len(retrieved),
            "protected_chunks": protected_chunks,
        }


pipeline = SecureRAGPipeline()
pipeline.index_document("John Smith (john@acme.com) ordered product A")
pipeline.index_document("Product A costs $49.99 and ships in 2 days")

result = pipeline.search_and_protect("What did John order?")
print(f"Query: {result['original_query']}")
print(f"Safe query: {result['safe_query']}")
print(f"Chunks: {result['chunks_found']}")
for chunk in result['protected_chunks']:
    print(f"  {chunk['chunk'][:50]}... (PII: {chunk['pii_found']})")

Resumen

  • 🔑 Hay 5 estrategias de redacción: mask (placeholder), replace (valor ficticio), hash (determinístico), generalize (menos específico), synthetic (datos ficticios realistas) — cada una con trade-offs diferentes
  • 🔑 Presidio Anonymizer implementa las estrategias de redacción y trabaja con los resultados del Analyzer — soporta operators configurables por tipo de entidad
  • 🔑 Pre-LLM redaction es la defensa más efectiva: si el dato nunca llega al modelo, no puede filtrarse — pero reduce el contexto disponible
  • 🔑 Post-LLM redaction es la última línea de defensa: filtra PII que el modelo genera por memorización, context leakage, o fabricación
  • 🔑 Redacción reversible permite reconstruir el texto original usando mappings — útil para personalizar respuestas después de procesar con datos redactados
  • 🔑 En RAG pipelines, puedes redactar en tres puntos: indexación (permanente), post-retrieval (por query), y query (protege al usuario)
  • 🔑 Preservar el tipo de entidad (<PERSON> vs [REDACTED]) mejora la calidad de las respuestas del LLM porque el modelo entiende el contexto
  • 🔑 Los mappings de redacción reversible son datos sensibles en sí mismos — deben cifrarse, guardarse en memoria, y eliminarse después de uso

Recursos adicionales

  1. Presidio Anonymizer Documentation — Documentación oficial del Anonymizer
  2. Presidio Operators — Guía de operators disponibles y cómo crear custom operators
  3. Data Anonymization Techniques (ENISA) — Técnicas de anonimización y pseudonimización de ENISA
  4. k-Anonymity, l-Diversity, t-Closeness — Modelos formales de anonimización para conjuntos de datos
  5. NIST De-identification Guidelines — Guía del NIST para de-identificación de datos
  6. Presidio Tutorial: Anonymize and Deanonymize — Tutorial de anonimización reversible con Presidio
  7. GDPR Art. 4(5) — Pseudonymisation — Definición legal de pseudonimización bajo GDPR
  8. ARX Data Anonymization Tool — Herramienta open-source de anonimización con modelos formales de privacidad

Creado: Marzo 2026 Versión: 1.0

← Anterior3. PII Detection: Patrones, Regex y PresidioSiguiente →5. Data Minimization