Module 6: Data Privacy & PII Protection
4. PII Redaction: Before and After the LLM
Overview
In the previous capsule you built a PIIScanner that detects more than 30 types of PII with Presidio, spaCy, and custom recognizers. Detecting PII is the first step — now you need to decide what to do with the detected data. Do you remove it? Replace it with a placeholder? Hash it so you can audit? Keep a reversible version to reconstruct the response later?
PII redaction isn't a simple "find and replace" operation. Each strategy has utility, privacy, and compliance trade-offs. And the moment when you apply the redaction — before the LLM (pre-LLM) or after the LLM (post-LLM) — completely changes the design and the implications.
In this capsule you build two components of the PII Protection Layer: the Pre-LLM Redactor that sanitizes data before sending it to the model, and the Post-LLM Redactor that filters PII from the model's outputs. Both are reused directly in the capsule 08 project.
Redaction strategies
There are five main strategies for handling detected PII. Each has a different privacy vs utility profile:
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="Replaces with a generic placeholder of the type",
example_input="Contact john@acme.com for info",
example_output="Contact <EMAIL_ADDRESS> for info",
reversible=False,
privacy_level="High",
utility_preserved="Low — the data is lost",
use_case="Logs, auditing, data that doesn't need reconstruction",
),
StrategyProfile(
strategy=RedactionStrategy.REPLACE,
description="Replaces with a fictional value of the same type",
example_input="Call María García at 555-1234",
example_output="Call Jane Doe at 000-0000",
reversible=False,
privacy_level="High",
utility_preserved="Medium — keeps the structure",
use_case="Testing, demos, training datasets",
),
StrategyProfile(
strategy=RedactionStrategy.HASH,
description="Replaces with a hash of the original value",
example_input="Email: maria@test.com",
example_output="Email: <EMAIL_a1b2c3d4>",
reversible=True,
privacy_level="Medium — the hash is deterministic",
utility_preserved="Low — but allows correlation",
use_case="Auditing where you need to trace without exposing",
),
StrategyProfile(
strategy=RedactionStrategy.GENERALIZE,
description="Reduces the specificity of the data",
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="Medium — precision is lost",
utility_preserved="High — keeps general context",
use_case="Analysis where you need patterns without identifying individuals",
),
StrategyProfile(
strategy=RedactionStrategy.SYNTHETIC,
description="Replaces with realistic synthetic data",
example_input="John Smith, john@real.com, 555-123-4567",
example_output="Alex Johnson, alex@example.com, 555-000-0000",
reversible=True,
privacy_level="High — the data is fictional",
utility_preserved="High — the LLM receives 'realistic' data",
use_case="Pre-LLM redaction where the model needs context",
),
]
print("PII redaction strategies:\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} | Privacy: {s.privacy_level}")
print(f" Use case: {s.use_case}")
print()Presidio Anonymizer: redaction with the official engine
Presidio Anonymizer works with the Analyzer's results to apply automatic redaction.
Basic masking
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}\"")
# Expected output:
# 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>.Anonymization operators
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": "[NAME_REDACTED]"}),
"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}")
# Expected output:
# Original: Contact María García at maria@test.com or 555-123-4567. SSN: 123-45-6789.
# Anonymized: Contact [NAME_REDACTED] 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}")
# Expected output:
# Original: Email: maria@test.com, Phone: 555-123-4567
# Hashed: Email: <hash_sha256>, Phone: <hash_sha256>Pre-LLM Redaction: sanitize before sending to the model
Pre-LLM redaction is the most effective defense against LLM02. If the data never reaches the model, the model can't leak it.
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:
"""Mapping to reconstruct data after the 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:
"""Redacts PII before sending to the 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:
"""Redacts PII from the text."""
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,
)
# --- Demo ---
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"\nThe LLM receives the redacted text and never sees the real PII.")
# Expected output:
# 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: 3Reversible redaction
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:
"""Reconstructs the original text using the 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}")
# Expected output:
# 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: TruePost-LLM Redaction: filter the model's outputs
Even if you redact PII before the LLM, the model can generate PII in its output — data memorized from training, data from the RAG context, or simply fabricated data that looks real.
@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:
"""Redacts PII from LLM outputs before sending to the user."""
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:
"""Scans and redacts PII from the LLM output."""
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",
)
# --- Demo ---
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()
# Expected output:
# 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..."Redaction in RAG pipelines
In a RAG pipeline, there are three points where you can redact PII:
class RAGPIIProtector:
"""Protects PII at each stage of the RAG pipeline."""
def __init__(self):
self.analyzer = AnalyzerEngine()
self.anonymizer = AnonymizerEngine()
def redact_at_index_time(self, document: str) -> dict:
"""Redacts PII when you index documents into the 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]:
"""Redacts PII from retrieved chunks before injecting into the 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:
"""Redacts PII from the user's query before searching."""
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}
# --- Demo ---
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]}...\"")
# Expected output:
# 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 of each redaction point
rag_redaction_tradeoffs = {
"index_time": {
"pros": [
"PII is never stored in the vector store",
"A single redaction for all queries",
"Reduces exposure risk in case of a vector store breach",
],
"cons": [
"Permanent loss of the original data (if you don't keep a backup)",
"Semantic search can be affected by placeholders",
"You can't re-process with better detectors without re-indexing",
],
"recommendation": "Ideal for permanent sensitive data (SSN, cards)",
},
"post_retrieval": {
"pros": [
"Original documents intact for re-processing",
"You can update the redaction logic without re-indexing",
"You only redact what's retrieved, not the whole corpus",
],
"cons": [
"PII stored in the vector store (risk in a breach)",
"Extra latency on every query (scan + redaction)",
"You need to guarantee that the redaction always runs",
],
"recommendation": "Ideal when you need the original documents for other uses",
},
"query_time": {
"pros": [
"Protects the user from sending their own PII to the system",
"Complements the other strategies",
],
"cons": [
"Can affect retrieval quality",
"Doesn't protect against PII in the documents",
],
"recommendation": "Always recommended as an additional layer",
},
}
print("RAG redaction trade-offs:\n")
for point, info in rag_redaction_tradeoffs.items():
print(f" {point.upper().replace('_', ' ')}:")
print(f" Recommendation: {info['recommendation']}")
print(f" Pros: {', '.join(info['pros'][:2])}")
print(f" Cons: {', '.join(info['cons'][:2])}")
print()Preserving context while protecting PII
The biggest challenge of pre-LLM redaction is preserving enough context so the model generates useful responses.
def redact_with_context_preservation(
text: str,
preserve_entity_types: bool = True,
) -> dict:
"""
Redacts PII while preserving the entity type
so the LLM understands the context.
"""
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("The LLM understands '<PERSON> from <LOCATION>'")
print("better than '[REDACTED] from [REDACTED]'")
# Expected output:
# 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 that allows reconstructing the original text."""
def __init__(self):
self.analyzer = AnalyzerEngine()
self._mapping_store: dict[str, dict[str, str]] = {}
def redact(self, text: str, request_id: str) -> str:
"""Redacts and stores mappings for reversal."""
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:
"""Reconstructs the original text."""
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):
"""Removes a request's mappings (for retention)."""
self._mapping_store.pop(request_id, None)
# --- Demo ---
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")
# Expected output:
# 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 complianceConnection to the project
The components from this capsule integrate into the PII Protection Layer:
PII Protection Layer
├── PIIScanner (Capsule 03)
├── PreLLMRedactor (THIS CAPSULE) ← Redaction before the LLM
├── PostLLMRedactor (THIS CAPSULE) ← Redaction after the LLM
├── Data Minimizer (Capsule 05)
├── Retention Scheduler (Capsule 06)
└── Audit Logger
The pipeline flow with redaction:
User Input
│
▼
PIIScanner → detects PII
│
▼
PreLLMRedactor → redacts PII from the input
│
▼
Data Minimizer → sends only what's needed
│
▼
LLM Processing
│
▼
PostLLMRedactor → redacts PII from the output
│
▼
[Optional] Reversible reconstruction
│
▼
Response to the user
Troubleshooting
Problem 1: "Redaction breaks the JSON structure"
If the text contains JSON and the values get redacted, the JSON can end up malformed.
Solution: Parse the JSON first, redact the individual values, and rebuild the 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 resultProblem 2: "The LLM generates worse responses with redacted texts"
Placeholders like <PERSON> confuse the model and produce lower-quality responses.
Solution: Use the SYNTHETIC strategy for pre-LLM redaction. The model receives fictional "realistic" data instead of placeholders, and the response is reconstructed with the real data using reversible redaction.
Problem 3: "Reversible redaction is a security risk"
The mappings from reversible redaction contain the original data. If they leak, the redaction was useless.
Solution: Store the mappings in memory (not in logs), encrypt the mappings at rest, and delete them immediately after reconstruction. Use the clear_mapping() function as part of the flow.
Problem 4: "Presidio doesn't detect certain types of PII in LLM outputs"
The model can generate PII in formats that Presidio doesn't recognize, such as "su número de teléfono es cinco-cinco-cinco, ciento veintitrés, cuarenta y cinco, sesenta y siete".
Solution: Add custom recognizers for verbalized formats, or use a normalization layer that converts verbalized text to a standard format before passing it through Presidio.
Exercises
Exercise 1: Redactor with a configurable strategy by entity type
Create a redactor that applies different strategies depending on the type of PII.
See solution
from presidio_anonymizer.entities import OperatorConfig
class SmartRedactor:
STRATEGY_MAP = {
"PERSON": OperatorConfig("replace", {"new_value": "[PERSON]"}),
"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)}")
# Expected output:
# Original: John Smith, john@test.com, 555-123-4567, SSN: 123-45-6789
# Redacted: [PERSON], redacted@example.com, *******4567, SSN: ***-**-****Exercise 2: Pre-LLM vs post-LLM comparator
Create a function that compares which PII is detected in the input vs the output of the LLM.
See solution
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"The LLM introduced new PII types: {new_in_output}"
if new_in_output
else "The LLM didn't introduce new PII"
),
}
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']}")Exercise 3: Redactor with an audit trail
Add an audit trail to the redactor that records each redaction with a timestamp and metadata.
See solution
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))Exercise 4: RAG pipeline with redaction at three points
Implement a mini RAG pipeline that redacts PII at indexing, retrieval, and query.
See solution
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']})")Summary
- 🔑 There are 5 redaction strategies: mask (placeholder), replace (fictional value), hash (deterministic), generalize (less specific), synthetic (realistic fictional data) — each with different trade-offs
- 🔑 Presidio Anonymizer implements the redaction strategies and works with the Analyzer's results — it supports operators configurable by entity type
- 🔑 Pre-LLM redaction is the most effective defense: if the data never reaches the model, it can't be leaked — but it reduces the available context
- 🔑 Post-LLM redaction is the last line of defense: it filters PII that the model generates through memorization, context leakage, or fabrication
- 🔑 Reversible redaction lets you reconstruct the original text using mappings — useful for personalizing responses after processing with redacted data
- 🔑 In RAG pipelines, you can redact at three points: indexing (permanent), post-retrieval (per query), and query (protects the user)
- 🔑 Preserving the entity type (
<PERSON>vs[REDACTED]) improves the quality of the LLM's responses because the model understands the context - 🔑 The reversible redaction mappings are sensitive data themselves — they must be encrypted, kept in memory, and deleted after use
Additional resources
- Presidio Anonymizer Documentation — Official Anonymizer documentation
- Presidio Operators — Guide to available operators and how to create custom operators
- Data Anonymization Techniques (ENISA) — ENISA's anonymization and pseudonymization techniques
- k-Anonymity, l-Diversity, t-Closeness — Formal anonymization models for datasets
- NIST De-identification Guidelines — NIST's guide for data de-identification
- Presidio Tutorial: Anonymize and Deanonymize — Reversible anonymization tutorial with Presidio
- GDPR Art. 4(5) — Pseudonymisation — Legal definition of pseudonymization under GDPR
- ARX Data Anonymization Tool — Open-source anonymization tool with formal privacy models
Created: March 2026 Version: 1.0