Module 6: Data Privacy & PII Protection

8. Project: PII Protection Layer

Project overview

This project closes Module 6 with the implementation of a complete PII Protection Layer — a personal data protection layer that integrates with your Sanitization Pipeline from Module 4. It's not a partial exercise — it's the production-ready artifact that demonstrates your AI system doesn't just resist attacks (M3), maintain data hygiene (M4), and secure credentials (M5), but protects your users' personal information.

In the previous capsules you built the individual pieces: PIIScanner (03), PreLLMRedactor and PostLLMRedactor (04), DataMinimizer (05), RetentionScheduler and DataEncryptor (06), and ConsentManager (07). Now everything consolidates into a single system that:

Scans PII in inputs and outputs with Presidio + custom recognizers
Redacts PII before sending to the LLM (pre-LLM) and after receiving the response (post-LLM)
Minimizes data sent to the LLM according to per-endpoint policies
Schedules retention with automatic deletion of expired data
Records auditing of every detection, redaction, and retention action

By the end you'll have a directory with runnable Python code and a FastAPI server that demonstrates the PII Protection Layer in action. This is the sixth artifact of the guide and integrates with the Sanitization Pipeline (M4) and the Injection Defense Pipeline (M3) in Module 8 (Secured AI System).

Project objective

Build a FastAPI PII protection middleware that processes every request through 5 layers (PII scanner → pre-LLM redactor → data minimizer → post-LLM redactor → audit logger), with centralized configuration, automatic retention, and performance metrics — all in a reusable package.

Technical specifications

Stack

Python >= 3.10
pydantic >= 2.0
fastapi >= 0.100
uvicorn >= 0.20
presidio-analyzer >= 2.2
presidio-anonymizer >= 2.2
spacy >= 3.5
cryptography >= 41.0

Deliverable structure

pii-protection-layer/
├── pii_layer/
│   ├── __init__.py
│   ├── scanner.py            # PIIScanner
│   ├── redactor.py           # PreLLMRedactor + PostLLMRedactor
│   ├── minimizer.py          # DataMinimizer
│   ├── retention.py          # RetentionScheduler
│   ├── audit.py              # PIIAuditLogger
│   ├── pipeline.py           # PIIProtectionPipeline (integration)
│   └── config.py             # Centralized configuration
├── app.py                    # FastAPI application
├── tests/
│   ├── test_scanner.py
│   ├── test_redactor.py
│   ├── test_minimizer.py
│   ├── test_retention.py
│   └── test_pipeline.py
├── requirements.txt
└── README.md

Required features

1. PIIScanner

✅ Presidio AnalyzerEngine with a spaCy model
✅ Custom recognizers (minimum 2: EMPLOYEE_ID and a type from your domain)
✅ Configurable: score_threshold, entities to scan, language
✅ Result with entities, severities, scan_time_ms

2. PreLLMRedactor

✅ Redaction with Presidio Anonymizer
✅ Operators configurable by entity type
✅ Reversible mode with mappings
✅ Result with redacted_text, redaction_count, mappings

3. PostLLMRedactor

✅ Scan of the LLM output for PII
✅ Automatic block if critical PII is detected (SSN, credit card)
✅ Redaction of non-critical PII in the output
✅ Result with action (pass/redacted/blocked)

4. DataMinimizer

✅ Field classification (required/useful/unnecessary/forbidden)
✅ Per-endpoint/task policies
✅ Generalization of sensitive data
✅ Result with minimized_data, reduction_percent

5. RetentionScheduler

✅ Policies by data type (llm_log, chat, redaction_mapping)
✅ Evaluation of records against policies
✅ Actions: delete, archive, anonymize
✅ Result with records_kept/deleted/archived/anonymized

6. PIIProtectionPipeline

✅ Integration of the 5 components into a sequential flow
✅ Audit logging of each stage
✅ Error handling per stage
✅ Timing metrics per component

Implementation code

Step 1: Project setup

mkdir pii-protection-layer && cd pii-protection-layer
mkdir pii_layer tests

Create requirements.txt:

pydantic>=2.0
fastapi>=0.100
uvicorn>=0.20
presidio-analyzer>=2.2
presidio-anonymizer>=2.2
spacy>=3.5
cryptography>=41.0
pytest>=7.0
pytest-asyncio>=0.21

pip install -r requirements.txt
python -m spacy download en_core_web_lg

Step 2: Centralized configuration

Create pii_layer/config.py:

from pydantic import BaseModel, Field
from typing import Optional


class ScannerConfig(BaseModel):
    score_threshold: float = 0.5
    languages: list[str] = Field(default_factory=lambda: ["en"])
    entities: Optional[list[str]] = None
    custom_recognizers_enabled: bool = True


class RedactorConfig(BaseModel):
    strategy: str = "mask"
    reversible: bool = False
    block_on_critical: bool = True
    critical_entities: list[str] = Field(
        default_factory=lambda: ["US_SSN", "CREDIT_CARD", "IBAN_CODE"]
    )
    fallback_message: str = (
        "I'm sorry, I can't provide that information "
        "for privacy reasons."
    )


class MinimizerConfig(BaseModel):
    forbidden_fields: list[str] = Field(
        default_factory=lambda: [
            "ssn", "credit_card", "password", "api_key",
            "social_security", "medical_record",
        ]
    )
    sensitive_fields: list[str] = Field(
        default_factory=lambda: [
            "email", "phone", "address", "date_of_birth",
            "full_name", "last_name",
        ]
    )
    include_sensitive: bool = False


class RetentionConfig(BaseModel):
    llm_log_days: int = 30
    chat_history_days: int = 90
    redaction_mapping_days: int = 0
    audit_log_days: int = 365
    user_cache_days: int = 7


class PIILayerConfig(BaseModel):
    scanner: ScannerConfig = Field(default_factory=ScannerConfig)
    redactor: RedactorConfig = Field(default_factory=RedactorConfig)
    minimizer: MinimizerConfig = Field(default_factory=MinimizerConfig)
    retention: RetentionConfig = Field(default_factory=RetentionConfig)
    enable_audit_log: bool = True
    enable_pre_llm_redaction: bool = True
    enable_post_llm_redaction: bool = True
    enable_minimization: bool = True

Step 3: PIIScanner

Create pii_layer/scanner.py:

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

from presidio_analyzer import (
    AnalyzerEngine,
    PatternRecognizer,
    Pattern,
    RecognizerRegistry,
)

from .config import ScannerConfig


class PIISeverity(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


ENTITY_SEVERITY = {
    "PERSON": PIISeverity.MEDIUM,
    "EMAIL_ADDRESS": PIISeverity.MEDIUM,
    "PHONE_NUMBER": PIISeverity.MEDIUM,
    "US_SSN": PIISeverity.CRITICAL,
    "CREDIT_CARD": PIISeverity.CRITICAL,
    "IBAN_CODE": PIISeverity.HIGH,
    "IP_ADDRESS": PIISeverity.LOW,
    "LOCATION": PIISeverity.LOW,
    "DATE_TIME": PIISeverity.LOW,
    "NRP": PIISeverity.MEDIUM,
    "URL": PIISeverity.LOW,
    "EMPLOYEE_ID": PIISeverity.MEDIUM,
    "SUPPORT_TICKET": PIISeverity.LOW,
}

SEVERITY_ORDER = {
    PIISeverity.LOW: 0, PIISeverity.MEDIUM: 1,
    PIISeverity.HIGH: 2, PIISeverity.CRITICAL: 3,
}


@dataclass
class PIIEntity:
    entity_type: str
    text: str
    start: int
    end: int
    score: float
    severity: PIISeverity


@dataclass
class ScanResult:
    text: str
    entities: list[PIIEntity] = field(default_factory=list)
    entity_count: int = 0
    max_severity: PIISeverity = PIISeverity.LOW
    scan_time_ms: float = 0.0

    @property
    def has_pii(self) -> bool:
        return self.entity_count > 0

    @property
    def has_critical(self) -> bool:
        return self.max_severity == PIISeverity.CRITICAL


def _create_custom_recognizers() -> list[PatternRecognizer]:
    return [
        PatternRecognizer(
            supported_entity="EMPLOYEE_ID",
            name="Employee ID",
            patterns=[Pattern("emp_id", r"\bEMP-\d{6}\b", 0.95)],
        ),
        PatternRecognizer(
            supported_entity="SUPPORT_TICKET",
            name="Support Ticket",
            patterns=[Pattern("ticket", r"\bTKT-\d{8}\b", 0.90)],
        ),
    ]


class PIIScanner:
    def __init__(self, config: Optional[ScannerConfig] = None):
        self.config = config or ScannerConfig()

        registry = RecognizerRegistry()
        registry.load_predefined_recognizers()

        if self.config.custom_recognizers_enabled:
            for rec in _create_custom_recognizers():
                registry.add_recognizer(rec)

        self.analyzer = AnalyzerEngine(registry=registry)

    def scan(
        self,
        text: str,
        language: Optional[str] = None,
        entities: Optional[list[str]] = None,
    ) -> ScanResult:
        start = time.perf_counter()

        lang = language or self.config.languages[0]
        ents = entities or self.config.entities

        results = self.analyzer.analyze(
            text=text,
            language=lang,
            entities=ents,
            score_threshold=self.config.score_threshold,
        )

        pii_entities = []
        max_sev = PIISeverity.LOW

        for r in results:
            sev = ENTITY_SEVERITY.get(r.entity_type, PIISeverity.MEDIUM)
            pii_entities.append(PIIEntity(
                entity_type=r.entity_type,
                text=text[r.start:r.end],
                start=r.start, end=r.end,
                score=r.score, severity=sev,
            ))
            if SEVERITY_ORDER[sev] > SEVERITY_ORDER[max_sev]:
                max_sev = sev

        elapsed = (time.perf_counter() - start) * 1000

        return ScanResult(
            text=text,
            entities=sorted(pii_entities, key=lambda e: e.start),
            entity_count=len(pii_entities),
            max_severity=max_sev,
            scan_time_ms=round(elapsed, 2),
        )

Step 4: Redactors

Create pii_layer/redactor.py:

from dataclasses import dataclass, field
from typing import Optional

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

from .config import RedactorConfig


@dataclass
class RedactionMapping:
    placeholder: str
    original: str
    entity_type: str


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


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


class PreLLMRedactor:
    def __init__(self, config: Optional[RedactorConfig] = None):
        self.config = config or RedactorConfig()
        self.analyzer = AnalyzerEngine()
        self.anonymizer = AnonymizerEngine()

    def redact(self, text: str, language: str = "en") -> PreLLMResult:
        results = self.analyzer.analyze(
            text=text, language=language,
            score_threshold=0.5,
        )

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

        mappings = []

        if self.config.reversible:
            counter = {}
            redacted = text
            for r in sorted(results, key=lambda x: x.end, reverse=True):
                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,
                ))
                redacted = redacted[:r.start] + placeholder + redacted[r.end:]

            return PreLLMResult(
                original_text=text, redacted_text=redacted,
                redaction_count=len(results),
                mappings=mappings, reversible=True,
            )

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

        return PreLLMResult(
            original_text=text,
            redacted_text=anonymized.text,
            redaction_count=len(results),
        )


class PostLLMRedactor:
    def __init__(self, config: Optional[RedactorConfig] = None):
        self.config = config or RedactorConfig()
        self.analyzer = AnalyzerEngine()
        self.anonymizer = AnonymizerEngine()

    def redact(self, llm_output: str, language: str = "en") -> PostLLMResult:
        results = self.analyzer.analyze(
            text=llm_output, language=language,
            score_threshold=0.5,
        )

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

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

        entities_info = [
            {
                "type": r.entity_type,
                "text": llm_output[r.start:r.end],
                "score": r.score,
            }
            for r in results
        ]

        if has_critical and self.config.block_on_critical:
            return PostLLMResult(
                original_output=llm_output,
                redacted_output=self.config.fallback_message,
                pii_found=len(results),
                entities_redacted=entities_info,
                action="blocked",
            )

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

        return PostLLMResult(
            original_output=llm_output,
            redacted_output=anonymized.text,
            pii_found=len(results),
            entities_redacted=entities_info,
            action="redacted",
        )

Step 5: DataMinimizer

Create pii_layer/minimizer.py:

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

from .config import MinimizerConfig


class MinAction(Enum):
    INCLUDED = "included"
    EXCLUDED = "excluded"
    BLOCKED = "blocked"
    GENERALIZED = "generalized"


@dataclass
class MinResult:
    original_field_count: int
    minimized_data: dict
    excluded_count: int = 0
    blocked_count: int = 0
    data_reduction_percent: float = 0.0


class DataMinimizer:
    def __init__(self, config: Optional[MinimizerConfig] = None):
        self.config = config or MinimizerConfig()

    def minimize(
        self,
        data: dict,
        allowed_fields: Optional[list[str]] = None,
    ) -> MinResult:
        minimized = {}
        excluded = 0
        blocked = 0

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

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

            if norm_key in self.config.forbidden_fields:
                blocked += 1
                continue

            if allowed_fields and key not in allowed_fields:
                excluded += 1
                continue

            if norm_key in self.config.sensitive_fields:
                if not self.config.include_sensitive:
                    excluded += 1
                    continue
                minimized[key] = self._generalize(key, value)
                continue

            minimized[key] = value

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

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

    def _generalize(self, key: str, value: Any) -> Any:
        if value is None:
            return None
        str_val = str(value)
        norm = key.lower()

        if norm == "email" and "@" in str_val:
            return f"***@{str_val.split('@')[1]}"
        if norm in ("phone", "telephone") and len(str_val) >= 4:
            return "***" + str_val[-4:]
        if norm in ("full_name",):
            parts = str_val.split()
            return parts[0] if parts else "[Name]"
        if norm in ("date_of_birth", "dob"):
            match = re.search(r"(19|20)\d{2}", str_val)
            if match:
                return f"{(int(match.group()) // 10) * 10}s"

        return value

Step 6: RetentionScheduler

Create pii_layer/retention.py:

from dataclasses import dataclass, field
from datetime import datetime, timezone, timedelta
from typing import Optional, Callable
from enum import Enum

from .config import RetentionConfig


class RetAction(Enum):
    KEEP = "keep"
    DELETE = "delete"
    ARCHIVE = "archive"
    ANONYMIZE = "anonymize"


@dataclass
class RetentionPolicy:
    data_type: str
    retention_days: int
    action: RetAction


@dataclass
class DataRecord:
    record_id: str
    data_type: str
    created_at: datetime
    content: Optional[str] = None

    @property
    def age_days(self) -> float:
        return (datetime.now(timezone.utc) - self.created_at).total_seconds() / 86400


@dataclass
class RetentionResult:
    records_checked: int
    kept: int = 0
    deleted: int = 0
    archived: int = 0
    anonymized: int = 0
    details: list[dict] = field(default_factory=list)


class RetentionScheduler:
    def __init__(self, config: Optional[RetentionConfig] = None):
        cfg = config or RetentionConfig()
        self.policies = {
            "llm_log": RetentionPolicy("llm_log", cfg.llm_log_days, RetAction.DELETE),
            "chat_message": RetentionPolicy("chat_message", cfg.chat_history_days, RetAction.ANONYMIZE),
            "redaction_mapping": RetentionPolicy("redaction_mapping", cfg.redaction_mapping_days, RetAction.DELETE),
            "audit_log": RetentionPolicy("audit_log", cfg.audit_log_days, RetAction.ARCHIVE),
            "user_cache": RetentionPolicy("user_cache", cfg.user_cache_days, RetAction.DELETE),
        }

    def evaluate(self, records: list[DataRecord]) -> RetentionResult:
        result = RetentionResult(records_checked=len(records))

        for record in records:
            policy = self.policies.get(record.data_type)
            if not policy:
                result.kept += 1
                continue

            if record.age_days <= policy.retention_days:
                result.kept += 1
                result.details.append({
                    "record_id": record.record_id,
                    "action": "keep",
                    "age_days": round(record.age_days, 1),
                })
            else:
                if policy.action == RetAction.DELETE:
                    record.content = None
                    result.deleted += 1
                elif policy.action == RetAction.ARCHIVE:
                    result.archived += 1
                elif policy.action == RetAction.ANONYMIZE:
                    if record.content:
                        import hashlib
                        record.content = hashlib.sha256(
                            record.content.encode()
                        ).hexdigest()[:16]
                    result.anonymized += 1

                result.details.append({
                    "record_id": record.record_id,
                    "action": policy.action.value,
                    "age_days": round(record.age_days, 1),
                })

        return result

Step 7: PIIAuditLogger

Create pii_layer/audit.py:

import json
import logging
import re
from datetime import datetime, timezone
from dataclasses import dataclass, field
from typing import Optional


@dataclass
class PIIAuditEntry:
    timestamp: str
    request_id: str
    stage: str
    action: str
    pii_count: int = 0
    details: dict = field(default_factory=dict)
    timing_ms: float = 0.0


PII_LOG_PATTERNS = {
    "email": re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"),
    "phone": re.compile(r"\b(?:\+\d{1,3}\s?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b"),
    "ssn": re.compile(r"\b\d{3}-\d{2}-\d{4}\b"),
    "credit_card": re.compile(r"\b(?:\d{4}[-\s]?){3}\d{4}\b"),
}


class PIIAuditLogger:
    def __init__(self, logger_name: str = "pii_audit"):
        self.logger = logging.getLogger(logger_name)
        if not self.logger.handlers:
            handler = logging.StreamHandler()
            handler.setFormatter(logging.Formatter(
                "%(asctime)s [PII_AUDIT] %(message)s"
            ))
            self.logger.addHandler(handler)
            self.logger.setLevel(logging.INFO)
        self.entries: list[PIIAuditEntry] = []

    def _sanitize_log(self, message: str) -> str:
        sanitized = message
        for pii_type, pattern in PII_LOG_PATTERNS.items():
            sanitized = pattern.sub(f"[{pii_type.upper()}_REDACTED]", sanitized)
        return sanitized

    def log(
        self,
        request_id: str,
        stage: str,
        action: str,
        pii_count: int = 0,
        details: Optional[dict] = None,
        timing_ms: float = 0.0,
    ):
        entry = PIIAuditEntry(
            timestamp=datetime.now(timezone.utc).isoformat(),
            request_id=request_id,
            stage=stage,
            action=action,
            pii_count=pii_count,
            details=details or {},
            timing_ms=round(timing_ms, 2),
        )
        self.entries.append(entry)

        log_data = {
            "request_id": request_id,
            "stage": stage,
            "action": action,
            "pii_count": pii_count,
            "timing_ms": round(timing_ms, 2),
        }
        self.logger.info(self._sanitize_log(json.dumps(log_data)))

    def get_stats(self) -> dict:
        if not self.entries:
            return {"total_entries": 0}

        pii_detected = sum(e.pii_count for e in self.entries if e.stage == "scan")
        blocked = sum(1 for e in self.entries if e.action == "blocked")

        return {
            "total_entries": len(self.entries),
            "total_pii_detected": pii_detected,
            "total_blocked": blocked,
            "avg_timing_ms": round(
                sum(e.timing_ms for e in self.entries) / len(self.entries), 2
            ),
        }

Step 8: Integrated pipeline

Create pii_layer/pipeline.py:

import time
import uuid
from dataclasses import dataclass, field
from typing import Optional, Callable

from .scanner import PIIScanner
from .redactor import PreLLMRedactor, PostLLMRedactor
from .minimizer import DataMinimizer
from .audit import PIIAuditLogger
from .config import PIILayerConfig


@dataclass
class PIIContext:
    request_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    user_id: str = "anonymous"
    endpoint: str = ""
    stages_passed: list[str] = field(default_factory=list)
    stages_failed: list[str] = field(default_factory=list)
    total_pii_detected: int = 0
    total_pii_redacted: int = 0
    timings: dict[str, float] = field(default_factory=dict)
    total_time_ms: float = 0.0


class PIIProtectionPipeline:
    def __init__(
        self,
        config: Optional[PIILayerConfig] = None,
        llm_caller: Optional[Callable] = None,
    ):
        self.config = config or PIILayerConfig()
        self.scanner = PIIScanner(self.config.scanner)
        self.pre_redactor = PreLLMRedactor(self.config.redactor)
        self.post_redactor = PostLLMRedactor(self.config.redactor)
        self.minimizer = DataMinimizer(self.config.minimizer)
        self.audit = PIIAuditLogger()
        self.llm_caller = llm_caller

    async def process(
        self,
        user_input: str,
        user_data: Optional[dict] = None,
        context: Optional[PIIContext] = None,
    ) -> dict:
        ctx = context or PIIContext()
        start = time.perf_counter()

        # Stage 1: Scan input for PII
        t = time.perf_counter()
        scan_result = self.scanner.scan(user_input)
        ctx.timings["scan"] = (time.perf_counter() - t) * 1000
        ctx.total_pii_detected = scan_result.entity_count
        ctx.stages_passed.append("scan")

        self.audit.log(
            ctx.request_id, "scan", "completed",
            pii_count=scan_result.entity_count,
            details={"max_severity": scan_result.max_severity.value},
            timing_ms=ctx.timings["scan"],
        )

        # Stage 2: Pre-LLM Redaction
        processed_input = user_input
        if self.config.enable_pre_llm_redaction and scan_result.has_pii:
            t = time.perf_counter()
            redact_result = self.pre_redactor.redact(user_input)
            ctx.timings["pre_redact"] = (time.perf_counter() - t) * 1000
            processed_input = redact_result.redacted_text
            ctx.total_pii_redacted += redact_result.redaction_count
            ctx.stages_passed.append("pre_redact")

            self.audit.log(
                ctx.request_id, "pre_redact", "redacted",
                pii_count=redact_result.redaction_count,
                timing_ms=ctx.timings["pre_redact"],
            )

        # Stage 3: Data Minimization
        if self.config.enable_minimization and user_data:
            t = time.perf_counter()
            min_result = self.minimizer.minimize(user_data)
            ctx.timings["minimize"] = (time.perf_counter() - t) * 1000
            ctx.stages_passed.append("minimize")

            self.audit.log(
                ctx.request_id, "minimize", "minimized",
                details={
                    "reduction": f"{min_result.data_reduction_percent}%",
                    "blocked": min_result.blocked_count,
                },
                timing_ms=ctx.timings["minimize"],
            )

        # Stage 4: LLM Call
        t = time.perf_counter()
        try:
            raw_output = await self._call_llm(processed_input)
            ctx.stages_passed.append("llm_call")
        except Exception as e:
            ctx.stages_failed.append("llm_call")
            ctx.total_time_ms = (time.perf_counter() - start) * 1000
            return {
                "answer": "Error processing your request. Please try again.",
                "pii_protected": True,
            }
        finally:
            ctx.timings["llm_call"] = (time.perf_counter() - t) * 1000

        # Stage 5: Post-LLM Redaction
        final_output = raw_output
        if self.config.enable_post_llm_redaction:
            t = time.perf_counter()
            post_result = self.post_redactor.redact(raw_output)
            ctx.timings["post_redact"] = (time.perf_counter() - t) * 1000
            final_output = post_result.redacted_output
            ctx.stages_passed.append("post_redact")

            if post_result.pii_found > 0:
                ctx.total_pii_redacted += post_result.pii_found
                self.audit.log(
                    ctx.request_id, "post_redact", post_result.action,
                    pii_count=post_result.pii_found,
                    timing_ms=ctx.timings["post_redact"],
                )

        ctx.total_time_ms = (time.perf_counter() - start) * 1000

        if self.config.enable_audit_log:
            self.audit.log(
                ctx.request_id, "pipeline", "completed",
                details={
                    "stages_passed": ctx.stages_passed,
                    "total_pii": ctx.total_pii_detected,
                    "total_redacted": ctx.total_pii_redacted,
                },
                timing_ms=ctx.total_time_ms,
            )

        return {
            "answer": final_output,
            "pii_protected": True,
            "pii_stats": {
                "detected": ctx.total_pii_detected,
                "redacted": ctx.total_pii_redacted,
            },
        }

    async def _call_llm(self, user_input: str) -> str:
        if self.llm_caller:
            return await self.llm_caller(user_input)

        from openai import AsyncOpenAI
        client = AsyncOpenAI()
        response = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "You are a helpful assistant. Answer concisely."},
                {"role": "user", "content": user_input},
            ],
            temperature=0.3,
        )
        return response.choices[0].message.content

Step 9: `pii_layer/init.py`

from .scanner import PIIScanner, ScanResult, PIIEntity, PIISeverity
from .redactor import PreLLMRedactor, PostLLMRedactor, PreLLMResult, PostLLMResult
from .minimizer import DataMinimizer, MinResult
from .retention import RetentionScheduler, RetentionResult, DataRecord
from .audit import PIIAuditLogger
from .pipeline import PIIProtectionPipeline, PIIContext
from .config import PIILayerConfig

Step 10: FastAPI Application

Create app.py:

import logging
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional

from pii_layer import (
    PIIProtectionPipeline, PIIContext, PIILayerConfig,
)

logging.basicConfig(level=logging.INFO)

app = FastAPI(title="PII Protected AI API", version="1.0")


class ChatRequest(BaseModel):
    message: str = Field(min_length=1, max_length=10000)
    user_id: str = "anonymous"
    user_data: Optional[dict] = None


class ChatResponse(BaseModel):
    answer: str
    pii_protected: bool = True
    pii_stats: Optional[dict] = None


config = PIILayerConfig()

pipeline = PIIProtectionPipeline(config=config)


@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    context = PIIContext(user_id=request.user_id, endpoint="/chat")
    try:
        result = await pipeline.process(
            user_input=request.message,
            user_data=request.user_data,
            context=context,
        )
        return ChatResponse(**result)
    except Exception:
        raise HTTPException(status_code=500, detail="Internal error.")


@app.get("/health")
async def health():
    return {"status": "ok", "pii_layer_version": "1.0"}


@app.get("/audit/stats")
async def audit_stats():
    return pipeline.audit.get_stats()

Step 11: Tests

Create tests/test_scanner.py:

from pii_layer.scanner import PIIScanner, PIISeverity
from pii_layer.config import ScannerConfig


def test_clean_text():
    scanner = PIIScanner()
    result = scanner.scan("The weather is nice today.")
    assert not result.has_pii


def test_email_detection():
    scanner = PIIScanner()
    result = scanner.scan("Contact me at test@example.com")
    assert result.has_pii
    assert any(e.entity_type == "EMAIL_ADDRESS" for e in result.entities)


def test_ssn_critical():
    scanner = PIIScanner()
    result = scanner.scan("SSN: 123-45-6789")
    assert result.has_critical
    assert result.max_severity == PIISeverity.CRITICAL


def test_custom_recognizer():
    scanner = PIIScanner()
    result = scanner.scan("Employee EMP-123456 reported.")
    assert any(e.entity_type == "EMPLOYEE_ID" for e in result.entities)


def test_score_threshold():
    strict = PIIScanner(ScannerConfig(score_threshold=0.9))
    lenient = PIIScanner(ScannerConfig(score_threshold=0.3))
    text = "Call Robert at 555-0100"
    strict_result = strict.scan(text)
    lenient_result = lenient.scan(text)
    assert lenient_result.entity_count >= strict_result.entity_count

Create tests/test_redactor.py:

from pii_layer.redactor import PreLLMRedactor, PostLLMRedactor
from pii_layer.config import RedactorConfig


def test_pre_llm_redaction():
    redactor = PreLLMRedactor()
    result = redactor.redact("My email is john@test.com")
    assert result.redaction_count > 0
    assert "john@test.com" not in result.redacted_text


def test_pre_llm_no_pii():
    redactor = PreLLMRedactor()
    result = redactor.redact("The sky is blue")
    assert result.redaction_count == 0
    assert result.redacted_text == "The sky is blue"


def test_reversible_redaction():
    redactor = PreLLMRedactor(RedactorConfig(reversible=True))
    result = redactor.redact("Contact John at john@test.com")
    assert result.reversible
    assert len(result.mappings) > 0


def test_post_llm_pass():
    redactor = PostLLMRedactor()
    result = redactor.redact("The product costs $49.99.")
    assert result.action == "pass"
    assert result.pii_found == 0


def test_post_llm_block_critical():
    redactor = PostLLMRedactor(RedactorConfig(block_on_critical=True))
    result = redactor.redact("Your SSN is 123-45-6789")
    assert result.action == "blocked"


def test_post_llm_redact_non_critical():
    redactor = PostLLMRedactor()
    result = redactor.redact("Contact support@acme.com for help")
    assert result.action in ("redacted", "pass")

Create tests/test_minimizer.py:

from pii_layer.minimizer import DataMinimizer
from pii_layer.config import MinimizerConfig


def test_blocks_forbidden_fields():
    minimizer = DataMinimizer()
    data = {"query": "hello", "ssn": "123-45-6789", "password": "secret"}
    result = minimizer.minimize(data)
    assert "ssn" not in result.minimized_data
    assert "password" not in result.minimized_data
    assert result.blocked_count == 2


def test_allowed_fields_filter():
    minimizer = DataMinimizer()
    data = {"query": "hello", "name": "John", "extra": "data"}
    result = minimizer.minimize(data, allowed_fields=["query"])
    assert "extra" not in result.minimized_data
    assert "query" in result.minimized_data


def test_reduction_percent():
    minimizer = DataMinimizer()
    data = {
        "query": "hi", "ssn": "123-45-6789",
        "credit_card": "4111111111111111",
    }
    result = minimizer.minimize(data)
    assert result.data_reduction_percent > 0


def test_excludes_sensitive_by_default():
    minimizer = DataMinimizer()
    data = {"query": "hello", "email": "test@test.com"}
    result = minimizer.minimize(data)
    assert "email" not in result.minimized_data

Create tests/test_retention.py:

from datetime import datetime, timezone, timedelta
from pii_layer.retention import RetentionScheduler, DataRecord


def test_keep_recent_records():
    scheduler = RetentionScheduler()
    now = datetime.now(timezone.utc)
    records = [DataRecord("r1", "llm_log", now - timedelta(days=5), "content")]
    result = scheduler.evaluate(records)
    assert result.kept == 1
    assert result.deleted == 0


def test_delete_expired_records():
    scheduler = RetentionScheduler()
    now = datetime.now(timezone.utc)
    records = [DataRecord("r1", "llm_log", now - timedelta(days=45), "content")]
    result = scheduler.evaluate(records)
    assert result.deleted == 1


def test_anonymize_old_chat():
    scheduler = RetentionScheduler()
    now = datetime.now(timezone.utc)
    record = DataRecord("r1", "chat_message", now - timedelta(days=120), "Hello world")
    result = scheduler.evaluate([record])
    assert result.anonymized == 1
    assert record.content != "Hello world"


def test_immediate_delete_redaction_mapping():
    scheduler = RetentionScheduler()
    now = datetime.now(timezone.utc)
    records = [DataRecord("r1", "redaction_mapping", now - timedelta(hours=1), "data")]
    result = scheduler.evaluate(records)
    assert result.deleted == 1

Create tests/test_pipeline.py:

import pytest
from pii_layer import PIIProtectionPipeline, PIIContext, PIILayerConfig


@pytest.fixture
def pipeline():
    config = PIILayerConfig()

    async def mock_llm(user_input: str) -> str:
        return "Your order is being processed and will arrive tomorrow."

    return PIIProtectionPipeline(config=config, llm_caller=mock_llm)


@pytest.mark.asyncio
async def test_pipeline_clean_input(pipeline):
    context = PIIContext()
    result = await pipeline.process("What is the weather today?", context=context)
    assert result["pii_protected"]
    assert "scan" in context.stages_passed


@pytest.mark.asyncio
async def test_pipeline_with_pii(pipeline):
    context = PIIContext()
    result = await pipeline.process(
        "My name is John Smith and my email is john@test.com",
        context=context,
    )
    assert result["pii_protected"]
    assert context.total_pii_detected > 0


@pytest.mark.asyncio
async def test_pipeline_llm_failure():
    async def failing_llm(user_input):
        raise Exception("API down")

    pipeline = PIIProtectionPipeline(llm_caller=failing_llm)
    context = PIIContext()
    result = await pipeline.process("Hello", context=context)
    assert "Error" in result["answer"] or "error" in result["answer"].lower()
    assert "llm_call" in context.stages_failed


@pytest.mark.asyncio
async def test_pipeline_with_user_data(pipeline):
    context = PIIContext()
    user_data = {
        "query": "order status",
        "order_id": "ORD-123",
        "ssn": "123-45-6789",
    }
    result = await pipeline.process(
        "What is my order status?",
        user_data=user_data,
        context=context,
    )
    assert result["pii_protected"]
    assert "minimize" in context.stages_passed

Running it

Run the tests

cd pii-protection-layer
pytest tests/ -v

# Expected output:
# tests/test_scanner.py::test_clean_text PASSED
# tests/test_scanner.py::test_email_detection PASSED
# tests/test_scanner.py::test_ssn_critical PASSED
# tests/test_scanner.py::test_custom_recognizer PASSED
# tests/test_scanner.py::test_score_threshold PASSED
# tests/test_redactor.py::test_pre_llm_redaction PASSED
# tests/test_redactor.py::test_pre_llm_no_pii PASSED
# tests/test_redactor.py::test_reversible_redaction PASSED
# tests/test_redactor.py::test_post_llm_pass PASSED
# tests/test_redactor.py::test_post_llm_block_critical PASSED
# tests/test_redactor.py::test_post_llm_redact_non_critical PASSED
# tests/test_minimizer.py::test_blocks_forbidden_fields PASSED
# tests/test_minimizer.py::test_allowed_fields_filter PASSED
# tests/test_minimizer.py::test_reduction_percent PASSED
# tests/test_minimizer.py::test_excludes_sensitive_by_default PASSED
# tests/test_retention.py::test_keep_recent_records PASSED
# tests/test_retention.py::test_delete_expired_records PASSED
# tests/test_retention.py::test_anonymize_old_chat PASSED
# tests/test_retention.py::test_immediate_delete_redaction_mapping PASSED
# tests/test_pipeline.py::test_pipeline_clean_input PASSED
# tests/test_pipeline.py::test_pipeline_with_pii PASSED
# tests/test_pipeline.py::test_pipeline_llm_failure PASSED
# tests/test_pipeline.py::test_pipeline_with_user_data PASSED
# All tests passed!

Run the server

uvicorn app:app --reload --port 8001

Test with curl

# Clean input (no PII)
curl -X POST http://localhost:8001/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "What is the weather today?", "user_id": "test-user"}'

# Input with PII (redacted before the LLM)
curl -X POST http://localhost:8001/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "My name is John Smith and my SSN is 123-45-6789. What is my order status?", "user_id": "test-user"}'

# Input with user_data (minimized)
curl -X POST http://localhost:8001/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "Order status?", "user_id": "test-user", "user_data": {"query": "order status", "order_id": "ORD-123", "ssn": "123-45-6789", "email": "john@test.com"}}'

# Audit stats
curl http://localhost:8001/audit/stats

# Health check
curl http://localhost:8001/health

Success criteria

Your project is complete when you can verify these points:

Evaluation rubric

Total: 100 points

Category	Points	Key criteria
PIIScanner	15	Presidio integration (4), custom recognizers (3), severities (3), configurable (3), scan_time_ms (2)
PreLLMRedactor	15	Presidio Anonymizer (4), configurable operators (3), reversible mode (4), mappings (4)
PostLLMRedactor	10	Output scan (3), block on critical (3), redact non-critical (2), fallback message (2)
DataMinimizer	10	Forbidden fields (3), sensitive exclusion (2), generalization (3), reduction metrics (2)
RetentionScheduler	10	Policies per type (3), delete/archive/anonymize (3), evaluation (2), immediate delete for mappings (2)
PIIAuditLogger	10	Log sanitization (3), audit entries (3), stats (2), PII pattern redaction (2)
Pipeline Integration	15	5 stages integrated (5), error handling (3), timing metrics (3), context tracking (4)
Tests	10	Scanner tests (2), redactor tests (2), minimizer tests (2), retention tests (2), pipeline tests (2)
Code Quality	5	Clean structure (2), typing (1), config centralized (1), no hardcoded values (1)

Grade distribution

Range	Grade
90-100	Excellent — production-ready PII Protection Layer
80-89	Very good — solid protection with minor improvements
70-79	Good — covers the basics but needs more robustness
60-69	Acceptable — missing components or depth
< 60	Needs revision — gaps in the protection

Common mistakes

1. Not caching Presidio's AnalyzerEngine

❌ Create a new AnalyzerEngine on every scan
✅ Create one at initialization and reuse it

Creating the engine includes loading the spaCy model (~200MB). Do it once in __init__ and reuse.

2. Logging PII in the audit logs

❌ logger.info(f"PII found: {entity.text}")
✅ logger.info(f"PII found: [{entity.entity_type}]")

The audit log records what type of PII was found, not what value it had. The value is exactly what you protect.

3. Not handling LLM failures with a fallback

❌ raise Exception("LLM call failed")
✅ return {"answer": "Error processing.", "pii_protected": True}

If the LLM API fails, return a safe fallback, not a 500 error.

4. Storing redaction mappings in persistent logs

The reversible redaction mappings contain the original data. If they're persisted in logs, the redaction was useless.

5. Not validating that the PostLLMRedactor actually runs

In production, a bug that disables the PostLLMRedactor exposes PII in outputs without anyone noticing. The tests must verify that PII in outputs is detected and acted upon.

6. Score threshold too high

A threshold of 0.9 misses many legitimate detections. Start with 0.5 in production and adjust based on false positive/negative metrics.

7. Not integrating the DataMinimizer with the pipeline

The minimizer is useless if the user's full data is sent to the LLM through another route (for example, in the system prompt or as unfiltered RAG context).

8. Ignoring PII in error messages

Error messages that include the user's input ("Error processing: My SSN is 123-45-6789") leak PII to the logs and potentially to the user.

Connection to the following modules

Your PII Protection Layer is the sixth artifact. As you progress:

Module	How it connects
Module 7: Security Testing	You test your PII layer with adversarial inputs, prompts that try to extract PII, and verify that the redaction works
Module 8: Integration	Your PII layer integrates with Injection Defense (M3) + Sanitization (M4) + Secrets (M5) in the Secured AI System

The integration in Module 8 combines all the pipelines:

Request → Sanitizer (M4) → Injection Detector (M3) → PII Scanner (M6) →
Pre-LLM Redactor (M6) → Data Minimizer (M6) → LLM →
Output Validator (M4) → Post-LLM Redactor (M6) → Content Filter (M4) →
Audit Logger (M4+M6) → Response

Summary

The PII Protection Layer is the central artifact of Module 6 — it integrates PII scanning, pre/post-LLM redaction, data minimization, retention, and audit logging into a reusable FastAPI middleware
6 components work in sequence: PIIScanner → PreLLMRedactor → DataMinimizer → LLM → PostLLMRedactor → PIIAuditLogger
Centralized configuration with Pydantic lets you tune the pipeline without changing code — thresholds, entities, retention periods, strategies
Error handling distinguishes between stages that can fail silently (scan, minimize) and stages that must fail with a fallback (LLM call, post-redaction)
Unit tests verify each component in isolation and the integrated pipeline
The pipeline integrates with the Sanitization Pipeline from Module 4 and consolidates in Module 8 as part of the Secured AI System
Presidio is the detection and redaction engine — it supports 30+ PII types, custom recognizers, and configurable operators

Project resources

Microsoft Presidio Documentation — Complete Presidio documentation
Presidio Anonymizer Operators — Operator reference for redaction
FastAPI Documentation — Web framework for the pipeline server
Pytest Documentation — Testing framework
OWASP LLM02: Sensitive Information Disclosure — The vulnerability the pipeline mitigates
Python cryptography Library — Encryption library for encryption at rest

Created: March 2026 Version: 1.0

← Previous7. Compliance Basics: GDPR, CCPA, and AI Next →1. Introduction: Security Testing & Auditing