FromZero2AI
CoursesPlaygroundBlog
Support on Ko-fi
LLM Security & Production Hardening • Module A: Attack VectorsLesson 3: PII Detection & Data Privacy in LLM Pipelines
PreviousNext

Lesson 3: PII Detection & Data Privacy in LLM Pipelines

Detect and redact personally identifiable information before it enters the LLM context. Implement anonymization, synthetic data generation, and differential privacy techniques for AI workflows.

Built with AI for beginners. Free forever.

Support on Ko-fi•About•Blog•Privacy Policy•Terms of Service

LLM systems are PII magnets. Users paste emails, paste code with credentials, ask about their medical history. Without active detection and redaction, that data ends up in logs, training pipelines, and model responses — creating compliance and liability risks.

PII in LLM Pipelines: Where It Leaks

Input

User pastes email with names, SSNs, card numbers. Goes into prompt → logs → potentially model inputs.

RAG Corpus

Internal docs indexed without review. HR records, customer data, contracts end up as retrievable context.

Output

Model regurgitates PII from context. Logs output verbatim. Another user's data appears in a shared cache response.

Detection: Presidio (Microsoft)

# pip install presidio-analyzer presidio-anonymizer spacy
# python -m spacy download en_core_web_lg

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

analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()

# Entities to detect (extend as needed)
PII_ENTITIES = [
    "PERSON", "EMAIL_ADDRESS", "PHONE_NUMBER",
    "CREDIT_CARD", "US_SSN", "IP_ADDRESS",
    "LOCATION", "DATE_TIME", "MEDICAL_LICENSE",
]

def detect_pii(text: str) -> list[dict]:
    """Return list of detected PII entities with positions."""
    results = analyzer.analyze(text=text, entities=PII_ENTITIES, language="en")
    return [
        {
            "type": r.entity_type,
            "start": r.start,
            "end": r.end,
            "score": r.score,
            "value": text[r.start:r.end],
        }
        for r in results if r.score > 0.7
    ]

def redact_pii(text: str) -> str:
    """Replace PII with type placeholders: <PERSON>, <EMAIL_ADDRESS>, etc."""
    results = analyzer.analyze(text=text, entities=PII_ENTITIES, language="en")
    if not results:
        return text

    anonymized = anonymizer.anonymize(
        text=text,
        analyzer_results=results,
        operators={
            entity: OperatorConfig("replace", {"new_value": f"<{entity}>"})
            for entity in PII_ENTITIES
        },
    )
    return anonymized.text

# Example
text = "My name is John Smith and my SSN is 123-45-6789. Email me at john@example.com"
print(detect_pii(text))
# [{'type': 'PERSON', 'value': 'John Smith', ...}, {'type': 'US_SSN', 'value': '123-45-6789', ...}]

print(redact_pii(text))
# "My name is <PERSON> and my SSN is <US_SSN>. Email me at <EMAIL_ADDRESS>"

Reversible Tokenization (for audit trails)

import secrets
from dataclasses import dataclass, field

@dataclass
class PiiVault:
    """Replace PII with opaque tokens, restore on output."""
    _store: dict[str, str] = field(default_factory=dict)

    def tokenize(self, text: str) -> str:
        """Replace PII with tokens. Tokens are safe to send to LLM."""
        detected = detect_pii(text)
        for item in sorted(detected, key=lambda x: x["start"], reverse=True):
            token = f"__PII_{secrets.token_hex(4).upper()}__"
            self._store[token] = item["value"]
            text = text[:item["start"]] + token + text[item["end"]:]
        return text

    def detokenize(self, text: str) -> str:
        """Restore original PII values in the response."""
        for token, original in self._store.items():
            text = text.replace(token, original)
        return text

# Usage
vault = PiiVault()
user_input = "Schedule a call with Sarah Connor (sarah@example.com) for Monday"
safe_input = vault.tokenize(user_input)
# "Schedule a call with __PII_A3F1__ (__PII_B7C2__) for Monday"

response = call_llm(safe_input)
final_response = vault.detokenize(response)
# "I've scheduled a call with Sarah Connor (sarah@example.com) for Monday."

Anthropic API: Training Data Opt-Out

import anthropic

# Opt out of Anthropic using your API data for model training
# This is on by default for API users — but explicit is better
client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-haiku-4-5-20251001",
    max_tokens=500,
    messages=[{"role": "user", "content": user_message}],
    # anthropic_beta header if needed for specific features
)
# API usage is NOT used for training by default.
# Review Anthropic's usage policy at anthropic.com/privacy
Always scan before you log

Redact PII from request and response bodies before writing to your logging system. Logs are often the path of least resistance for data breaches — they're rarely encrypted at rest and frequently over-retained.

Classify before you index

Run PII detection on every document before adding it to your RAG vector store. A document tagged HIGH_PII should require explicit access control — not be retrievable for all users.

Lesson 4 extends to multi-tenant architectures — how to ensure one tenant's data can never appear in another tenant's LLM responses.