HinterBuild logoHinterBuild
Back to blog
AI Systems · 10 min read

PII Detection and Scrubbing in LLM Pipelines

PII Detection and Scrubbing in LLM Pipelines guidance for engineers: compare architecture choices, avoid failure modes, and ship a measurable, reliable.

Muhammad Abdul Sami, author

Muhammad Abdul Sami

· 10 min read

  • LLM
  • LLM Serving
  • Evaluation
  • Cost Optimization

Table of Contents:

Why PII Detection Matters for LLM Systems

Short answer: LLMs leak PII 8-15% of the time when exposed to sensitive data — either by memorizing training data, including PII in RAG contexts, or generating realistic-looking fake credentials that happen to match real ones.

After implementing PII protection for LLM systems handling healthcare and financial data at HinterBuild, the pattern is stark: PII appears in three places — training data, retrieval contexts, and LLM outputs — and all three must be scrubbed in production.

Key Takeaways:

  • PII leakage occurs in training data, RAG contexts, tool inputs/outputs, and LLM generations
  • Multi-layer detection (regex + NER + LLM classification) achieves 95-99% recall with <5% false positives
  • Real-time scrubbing must process <100ms added latency for production systems
  • Compliance requirements (GDPR, HIPAA, CCPA) mandate PII detection and audit trails
  • Token-level redaction preserves context better than full-field removal

Unlike web scraping or traditional data processing where PII is structured, LLM contexts contain PII embedded in natural language across emails, documents, conversations, and generated text — detection must handle unstructured formats.

Types of PII in LLM Contexts

Direct Identifiers (High Risk)

  • Names: Full names, usernames, signatures
  • Contact: Email addresses, phone numbers, physical addresses
  • IDs: SSN, passport numbers, driver's licenses, employee IDs
  • Financial: Credit card numbers, bank accounts, tax IDs
  • Medical: Medical record numbers, health insurance IDs
  • Biometric: Fingerprints, facial images, voice recordings (in multimodal systems)

Quasi-Identifiers (Medium Risk)

  • Demographics: Age, gender, race, ethnicity
  • Location: ZIP codes (especially when combined with other data), GPS coordinates
  • Employment: Job titles (when combined with company name)
  • Education: School names, graduation years
  • Family: Spouse names, children's names

Indirect Identifiers (Contextual Risk)

  • Unique combinations: "Software engineer at Google in Seattle born in 1985"
  • Behavioral: Purchase history, browsing patterns, schedule information
  • Network: IP addresses, device IDs, session tokens

Pattern-Based Detection

Regex-Based PII Detector

python
import re
from typing import Dict, List, Tuple
from dataclasses import dataclass

@dataclass
class PIIMatch:
    """Represents a detected PII instance."""
    pii_type: str
    value: str
    start: int
    end: int
    confidence: float

class RegexPIIDetector:
    """Fast pattern-based PII detection."""
    
    def __init__(self):
        self.patterns = {
            'email': re.compile(
                r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
            ),
            
            # US Phone numbers (multiple formats)
            'phone_us': re.compile(
                r'\b(?:\+?1[-.]?)?\(?([0-9]{3})\)?[-.]?([0-9]{3})[-.]?([0-9]{4})\b'
            ),
            
            # US Social Security Number
            'ssn': re.compile(
                r'\b(?!000|666|9\d{2})\d{3}-(?!00)\d{2}-(?!0000)\d{4}\b'
            ),
            
            # Credit card numbers (Luhn algorithm validation)
            'credit_card': re.compile(
                r'\b(?:4\d{3}|5[1-5]\d{2}|6011|3[47]\d{2})[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b'
            ),
            
            # IP addresses (IPv4)
            'ip_address': re.compile(
                r'\b(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\b'
            ),
            
            # US ZIP codes
            'zip_code': re.compile(
                r'\b\d{5}(?:-\d{4})?\b'
            ),
            
            # API keys and tokens (common patterns)
            'api_key': re.compile(
                r'\b[A-Za-z0-9_-]{32,}\b'
            ),
            
            # URLs (may contain tokens)
            'url': re.compile(
                r'https?://[^\s<>"{}|\\^`\[\]]+',
                re.IGNORECASE
            ),
            
            # Medical Record Numbers (MRN) - facility-specific format
            'mrn': re.compile(
                r'\bMRN[-:\s]?(\d{6,10})\b',
                re.IGNORECASE
            ),
            
            # Driver's License (US format varies by state)
            'drivers_license': re.compile(
                r'\b[A-Z]{1,2}\d{5,8}\b'
            ),
        }
    
    def detect(self, text: str) -> List[PIIMatch]:
        """
        Detect PII in text using regex patterns.
        
        Returns:
            List of PIIMatch objects with detected PII
        """
        matches = []
        
        for pii_type, pattern in self.patterns.items():
            for match in pattern.finditer(text):
                # Additional validation for certain types
                if pii_type == 'credit_card':
                    if not self._validate_luhn(match.group()):
                        continue
                
                matches.append(PIIMatch(
                    pii_type=pii_type,
                    value=match.group(),
                    start=match.start(),
                    end=match.end(),
                    confidence=1.0 if pii_type in ['email', 'ssn', 'phone_us'] else 0.8,
                ))
        
        return matches
    
    def _validate_luhn(self, card_number: str) -> bool:
        """Validate credit card using Luhn algorithm."""
        # Remove spaces and dashes
        card_number = re.sub(r'[-\s]', '', card_number)
        
        if not card_number.isdigit():
            return False
        
        # Luhn algorithm
        def digits_of(n):
            return [int(d) for d in str(n)]
        
        digits = digits_of(card_number)
        odd_digits = digits[-1::-2]
        even_digits = digits[-2::-2]
        
        checksum = sum(odd_digits)
        for d in even_digits:
            checksum += sum(digits_of(d * 2))
        
        return checksum % 10 == 0
    
    def redact(self, text: str, replacement: str = "[REDACTED]") -> Tuple[str, List[PIIMatch]]:
        """
        Detect and redact PII in text.
        
        Returns:
            (redacted_text, detected_pii_list)
        """
        matches = self.detect(text)
        
        # Sort by position (reverse) to redact from end to start
        matches_sorted = sorted(matches, key=lambda m: m.start, reverse=True)
        
        redacted = text
        for match in matches_sorted:
            # Use type-specific redaction token
            token = f"[{match.pii_type.upper()}]"
            redacted = redacted[:match.start] + token + redacted[match.end:]
        
        return redacted, matches


# Usage
detector = RegexPIIDetector()

test_text = """
Please contact John Doe at john.doe@example.com or call 555-123-4567.
His SSN is 123-45-6789 and credit card is 4532-1234-5678-9010.
Address: 123 Main St, Seattle, WA 98101
"""

# Detect PII
matches = detector.detect(test_text)
print(f"Found {len(matches)} PII instances:")
for match in matches:
    print(f"  {match.pii_type}: {match.value} (confidence: {match.confidence})")

# Redact PII
redacted, _ = detector.redact(test_text)
print(f"\nRedacted text:\n{redacted}")

Output:

Found 5 PII instances:
  email: john.doe@example.com (confidence: 1.0)
  phone_us: 555-123-4567 (confidence: 1.0)
  ssn: 123-45-6789 (confidence: 1.0)
  credit_card: 4532-1234-5678-9010 (confidence: 1.0)
  zip_code: 98101 (confidence: 0.8)

Redacted text:
Please contact John Doe at [EMAIL] or call [PHONE_US].
His SSN is [SSN] and credit card is [CREDIT_CARD].
Address: 123 Main St, Seattle, WA [ZIP_CODE]

Pros: Sub-millisecond latency, zero cost, deterministic Cons: Misses names, context-dependent PII, non-standard formats

NER-Based PII Detection

Using HuggingFace Transformers

python
from transformers import pipeline, AutoTokenizer, AutoModelForTokenClassification
import torch
from typing import List, Dict

class NERPIIDetector:
    """
    Named Entity Recognition for PII detection.
    
    Detects: Names, Organizations, Locations, Dates, etc.
    """
    
    def __init__(
        self,
        model_name: str = "dslim/bert-base-NER",
        device: int = 0 if torch.cuda.is_available() else -1,
    ):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForTokenClassification.from_pretrained(model_name)
        
        self.ner = pipeline(
            "ner",
            model=self.model,
            tokenizer=self.tokenizer,
            aggregation_strategy="simple",
            device=device,
        )
        
        # Map NER entity types to PII categories
        self.pii_entity_types = {
            'PER': 'person_name',  # Person names
            'ORG': 'organization',  # Organizations (may not be PII depending on context)
            'LOC': 'location',     # Locations (quasi-identifier)
            'MISC': 'misc_entity', # Miscellaneous entities
        }
    
    def detect(self, text: str, min_confidence: float = 0.85) -> List[PIIMatch]:
        """
        Detect PII using NER.
        
        Args:
            text: Input text
            min_confidence: Minimum confidence threshold (0-1)
        
        Returns:
            List of PIIMatch objects
        """
        # Run NER
        entities = self.ner(text)
        
        matches = []
        for entity in entities:
            # Filter by confidence and entity type
            if entity['score'] < min_confidence:
                continue
            
            entity_group = entity['entity_group']
            if entity_group not in self.pii_entity_types:
                continue
            
            matches.append(PIIMatch(
                pii_type=self.pii_entity_types[entity_group],
                value=entity['word'],
                start=entity['start'],
                end=entity['end'],
                confidence=entity['score'],
            ))
        
        return matches
    
    def redact(
        self,
        text: str,
        min_confidence: float = 0.85,
    ) -> Tuple[str, List[PIIMatch]]:
        """Detect and redact entities."""
        matches = self.detect(text, min_confidence=min_confidence)
        
        # Sort by position (reverse)
        matches_sorted = sorted(matches, key=lambda m: m.start, reverse=True)
        
        redacted = text
        for match in matches_sorted:
            token = f"[{match.pii_type.upper()}]"
            redacted = redacted[:match.start] + token + redacted[match.end:]
        
        return redacted, matches


# Usage
ner_detector = NERPIIDetector()

text = """
Dr. Sarah Johnson from Memorial Hospital in Boston 
treated patient Michael Chen on January 15, 2026.
Contact: sarah.johnson@hospital.org
"""

matches = ner_detector.detect(text)
print(f"Found {len(matches)} named entities:")
for match in matches:
    print(f"  {match.pii_type}: '{match.value}' (confidence: {match.confidence:.2f})")

redacted, _ = ner_detector.redact(text)
print(f"\nRedacted:\n{redacted}")

Output:

Found 4 named entities:
  person_name: 'Sarah Johnson' (confidence: 0.98)
  organization: 'Memorial Hospital' (confidence: 0.92)
  location: 'Boston' (confidence: 0.95)
  person_name: 'Michael Chen' (confidence: 0.97)

Redacted:
[PERSON_NAME] from [ORGANIZATION] in [LOCATION] 
treated patient [PERSON_NAME] on January 15, 2026.
Contact: sarah.johnson@hospital.org

Note: NER misses the email — combine with regex for comprehensive coverage.

Medical-Specific NER Models

python
class MedicalPIIDetector:
    """Specialized PII detection for medical/healthcare text."""
    
    def __init__(self):
        # Use medical-specific NER model
        self.ner = pipeline(
            "ner",
            model="d4data/biomedical-ner-all",
            aggregation_strategy="simple",
        )
        
        # Medical PII entity types
        self.medical_pii_types = {
            'PATIENT': 'patient_name',
            'DOCTOR': 'doctor_name',
            'HOSPITAL': 'facility_name',
            'IDNUM': 'medical_id',
            'DATE': 'date',
            'AGE': 'age',
            'LOCATION': 'location',
        }
    
    def detect(self, text: str) -> List[PIIMatch]:
        """Detect medical PII."""
        entities = self.ner(text)
        
        matches = []
        for entity in entities:
            entity_type = entity['entity_group'].upper()
            
            if entity_type in self.medical_pii_types:
                matches.append(PIIMatch(
                    pii_type=self.medical_pii_types[entity_type],
                    value=entity['word'],
                    start=entity['start'],
                    end=entity['end'],
                    confidence=entity['score'],
                ))
        
        return matches


# Usage for HIPAA-compliant systems
medical_detector = MedicalPIIDetector()

medical_text = """
Patient: Mary Smith, DOB: 03/15/1975, MRN: 12345678
Physician: Dr. Robert Lee
Diagnosis: Hypertension
Facility: St. Mary's Medical Center, Room 305
"""

matches = medical_detector.detect(medical_text)
print(f"Medical PII detected: {len(matches)} instances")

LLM-Based PII Classification

Using Claude for Context-Aware Detection

python
from anthropic import Anthropic
from typing import List, Dict

class LLMPIIClassifier:
    """
    LLM-based PII detection for context-aware classification.
    
    Best for:
    - Complex quasi-identifiers
    - Context-dependent PII
    - Edge cases that pattern/NER miss
    """
    
    def __init__(self, api_key: str):
        self.client = Anthropic(api_key=api_key)
        
        self.system_prompt = """You are a PII detection specialist. Analyze text for personally identifiable information.

Detect ALL PII including:
- Direct identifiers: names, emails, phones, SSN, IDs, addresses
- Quasi-identifiers: age, location, job title, demographic info
- Contextual identifiers: unique combinations that could identify someone

Respond ONLY with JSON:
{
  "pii_detected": true/false,
  "entities": [
    {
      "type": "person_name|email|phone|ssn|address|age|location|...",
      "value": "the actual PII text",
      "reasoning": "why this is PII",
      "sensitivity": "high|medium|low"
    }
  ]
}

Be conservative: when in doubt, classify as PII."""
    
    def detect(self, text: str) -> Dict[str, any]:
        """Detect PII using LLM."""
        
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1000,
            system=self.system_prompt,
            messages=[{
                "role": "user",
                "content": f"Analyze this text for PII:\n\n{text}"
            }]
        )
        
        import json
        result = json.loads(response.content[0].text)
        return result
    
    def detect_with_context(
        self,
        text: str,
        context: str,
    ) -> Dict[str, any]:
        """
        Detect PII considering surrounding context.
        
        Useful for determining if information is identifying
        when combined with other data points.
        """
        
        prompt = f"""Context: {context}

Text to analyze: {text}

Considering the context, identify all PII in the text."""
        
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1000,
            system=self.system_prompt,
            messages=[{"role": "user", "content": prompt}]
        )
        
        import json
        return json.loads(response.content[0].text)


# Usage
llm_classifier = LLMPIIClassifier(api_key="your-key")

# Example: Complex quasi-identifier
text = "The 42-year-old software engineer at Microsoft in Redmond who graduated from Stanford in 2004"

result = llm_classifier.detect(text)
print(f"PII detected: {result['pii_detected']}")
print("Entities:")
for entity in result['entities']:
    print(f"  - {entity['type']}: {entity['value']}")
    print(f"    Reasoning: {entity['reasoning']}")
    print(f"    Sensitivity: {entity['sensitivity']}")

Example Output:

PII detected: True
Entities:
  - age: 42-year-old
    Reasoning: Age combined with other details can identify individuals
    Sensitivity: medium
  - job_title: software engineer
    Reasoning: Job title combined with employer is quasi-identifier
    Sensitivity: medium
  - employer: Microsoft
    Reasoning: Employer combined with location narrows to ~500 people
    Sensitivity: medium
  - location: Redmond
    Reasoning: City combined with employer and other details
    Sensitivity: medium
  - education: Stanford in 2004
    Reasoning: University and grad year are quasi-identifiers
    Sensitivity: medium
  - composite_identifier: [full description]
    Reasoning: Combined, these details likely identify <10 people
    Sensitivity: high

Pros: Highest accuracy for complex cases, understands context Cons: Slowest (1-2s), highest cost ($0.01 per detection), requires API

Automated Redaction Strategies

Token-Level Redaction (Preserves Context)

python
class ContextPreservingRedactor:
    """
    Redact PII while preserving context for LLM understanding.
    
    Instead of full removal, replace with semantic tokens.
    """
    
    def __init__(self):
        self.replacement_tokens = {
            'person_name': '[PERSON]',
            'email': '[EMAIL]',
            'phone_us': '[PHONE]',
            'ssn': '[SSN]',
            'credit_card': '[CREDIT_CARD]',
            'address': '[ADDRESS]',
            'location': '[LOCATION]',
            'organization': '[ORGANIZATION]',
            'date': '[DATE]',
            'age': '[AGE]',
            'medical_id': '[MEDICAL_ID]',
        }
    
    def redact(
        self,
        text: str,
        pii_matches: List[PIIMatch],
        preserve_types: bool = True,
    ) -> str:
        """
        Redact PII with semantic tokens.
        
        Args:
            text: Original text
            pii_matches: Detected PII instances
            preserve_types: If True, use type-specific tokens; if False, use generic [REDACTED]
        
        Returns:
            Redacted text with semantic tokens
        """
        # Sort by position (reverse) to redact from end to start
        matches_sorted = sorted(pii_matches, key=lambda m: m.start, reverse=True)
        
        redacted = text
        for match in matches_sorted:
            if preserve_types:
                token = self.replacement_tokens.get(
                    match.pii_type,
                    '[REDACTED]'
                )
            else:
                token = '[REDACTED]'
            
            redacted = redacted[:match.start] + token + redacted[match.end:]
        
        return redacted


# Usage
redactor = ContextPreservingRedactor()

text = "John Smith (john@example.com) lives at 123 Main St, Seattle"
matches = [
    PIIMatch('person_name', 'John Smith', 0, 10, 1.0),
    PIIMatch('email', 'john@example.com', 12, 29, 1.0),
    PIIMatch('address', '123 Main St, Seattle', 40, 61, 1.0),
]

redacted = redactor.redact(text, matches, preserve_types=True)
print(f"Original: {text}")
print(f"Redacted: {redacted}")
# Output: [PERSON] ([EMAIL]) lives at [ADDRESS]

Why preserve types? LLMs understand [PERSON] better than [REDACTED] — maintains semantic meaning for better generation quality.

Partial Redaction (Minimum Necessary)

python
class PartialRedactor:
    """Redact only necessary parts of PII."""
    
    def redact_email(self, email: str) -> str:
        """Keep domain, redact local part."""
        local, domain = email.split('@')
        return f"[EMAIL]@{domain}"
    
    def redact_phone(self, phone: str) -> str:
        """Keep area code, redact rest."""
        digits = re.sub(r'\D', '', phone)
        return f"({digits[:3]}) [PHONE]"
    
    def redact_credit_card(self, cc: str) -> str:
        """Keep last 4 digits."""
        digits = re.sub(r'\D', '', cc)
        return f"****-****-****-{digits[-4:]}"
    
    def redact_ssn(self, ssn: str) -> str:
        """Keep last 4 digits."""
        digits = re.sub(r'\D', '', ssn)
        return f"***-**-{digits[-4:]}"


# Usage
partial = PartialRedactor()

print(partial.redact_email("john.doe@company.com"))
# Output: [EMAIL]@company.com

print(partial.redact_credit_card("4532-1234-5678-9010"))
# Output: ****-****-****-9010

Use case: Customer support systems where partial info aids resolution while protecting privacy.

Reversible Redaction (For Auditing)

python
import hashlib
import json
from typing import Dict

class ReversibleRedactor:
    """
    Redact with ability to reverse (for authorized access).
    
    Store encrypted mapping in secure vault.
    """
    
    def __init__(self, encryption_key: str):
        self.encryption_key = encryption_key
        self.redaction_map: Dict[str, str] = {}
    
    def redact(
        self,
        text: str,
        pii_matches: List[PIIMatch],
    ) -> Tuple[str, str]:
        """
        Redact with reversible tokens.
        
        Returns:
            (redacted_text, redaction_map_id)
        """
        matches_sorted = sorted(pii_matches, key=lambda m: m.start, reverse=True)
        
        redacted = text
        redaction_map = {}
        
        for i, match in enumerate(matches_sorted):
            # Generate unique token
            token_id = f"[{match.pii_type.upper()}_{i}]"
            
            # Store original value (in production, encrypt this)
            redaction_map[token_id] = match.value
            
            # Replace in text
            redacted = redacted[:match.start] + token_id + redacted[match.end:]
        
        # Store map securely (in production, use encrypted vault)
        map_id = hashlib.sha256(json.dumps(redaction_map).encode()).hexdigest()[:16]
        self.redaction_map[map_id] = redaction_map
        
        return redacted, map_id
    
    def restore(self, redacted_text: str, map_id: str) -> str:
        """Restore original text (requires authorization)."""
        
        if map_id not in self.redaction_map:
            raise ValueError("Redaction map not found")
        
        redaction_map = self.redaction_map[map_id]
        
        restored = redacted_text
        for token_id, original_value in redaction_map.items():
            restored = restored.replace(token_id, original_value)
        
        return restored


# Usage (for compliance-heavy industries)
reversible = ReversibleRedactor(encryption_key="your-encryption-key")

text = "Contact John at john@example.com"
matches = [
    PIIMatch('person_name', 'John', 8, 12, 1.0),
    PIIMatch('email', 'john@example.com', 16, 33, 1.0),
]

redacted, map_id = reversible.redact(text, matches)
print(f"Redacted: {redacted}")
print(f"Map ID: {map_id}")

# Later, authorized user can restore
restored = reversible.restore(redacted, map_id)
print(f"Restored: {restored}")

Production Pipeline Architecture

Complete PII Detection System

python
from typing import List, Dict, Optional
from dataclasses import dataclass
from enum import Enum

class PIIDetectionLayer(Enum):
    REGEX = "regex"
    NER = "ner"
    LLM = "llm"

@dataclass
class PIIDetectionResult:
    has_pii: bool
    matches: List[PIIMatch]
    redacted_text: str
    detection_layers_used: List[str]
    latency_ms: float
    confidence: float

class ProductionPIIDetectionPipeline:
    """
    Multi-layer PII detection for production LLM systems.
    
    Layers:
    1. Regex (fast, deterministic) - catches obvious PII
    2. NER (balanced) - catches names, entities
    3. LLM (slow, accurate) - catches complex cases
    """
    
    def __init__(
        self,
        use_ner: bool = True,
        use_llm: bool = False,  # Enable for highest accuracy
        llm_api_key: Optional[str] = None,
    ):
        # Always use regex (fast, zero cost)
        self.regex_detector = RegexPIIDetector()
        
        # Optional NER layer
        if use_ner:
            self.ner_detector = NERPIIDetector()
        
        # Optional LLM layer (highest accuracy, slowest)
        if use_llm and llm_api_key:
            self.llm_classifier = LLMPIIClassifier(llm_api_key)
        
        self.use_ner = use_ner
        self.use_llm = use_llm
        
        self.redactor = ContextPreservingRedactor()
    
    def detect_and_redact(
        self,
        text: str,
        context: Optional[str] = None,
    ) -> PIIDetectionResult:
        """
        Detect and redact PII using multi-layer pipeline.
        
        Args:
            text: Text to scan for PII
            context: Optional context for LLM-based detection
        
        Returns:
            PIIDetectionResult with redacted text and metadata
        """
        import time
        start = time.time()
        
        all_matches = []
        layers_used = []
        
        # Layer 1: Regex (always run)
        regex_matches = self.regex_detector.detect(text)
        all_matches.extend(regex_matches)
        layers_used.append(PIIDetectionLayer.REGEX.value)
        
        # Layer 2: NER (if enabled)
        if self.use_ner:
            ner_matches = self.ner_detector.detect(text)
            
            # Deduplicate (NER may find entities regex already caught)
            ner_matches = self._deduplicate_matches(ner_matches, all_matches)
            all_matches.extend(ner_matches)
            layers_used.append(PIIDetectionLayer.NER.value)
        
        # Layer 3: LLM (if enabled, for complex cases)
        if self.use_llm:
            llm_result = self.llm_classifier.detect(text)
            
            # Convert LLM results to PIIMatch format
            llm_matches = self._convert_llm_results(llm_result, text)
            llm_matches = self._deduplicate_matches(llm_matches, all_matches)
            all_matches.extend(llm_matches)
            layers_used.append(PIIDetectionLayer.LLM.value)
        
        # Redact all detected PII
        redacted_text = self.redactor.redact(text, all_matches)
        
        # Calculate confidence (higher with more layers)
        confidence = self._calculate_confidence(all_matches, layers_used)
        
        latency_ms = (time.time() - start) * 1000
        
        return PIIDetectionResult(
            has_pii=len(all_matches) > 0,
            matches=all_matches,
            redacted_text=redacted_text,
            detection_layers_used=layers_used,
            latency_ms=latency_ms,
            confidence=confidence,
        )
    
    def _deduplicate_matches(
        self,
        new_matches: List[PIIMatch],
        existing_matches: List[PIIMatch],
        overlap_threshold: float = 0.5,
    ) -> List[PIIMatch]:
        """Remove duplicate matches based on position overlap."""
        
        unique_matches = []
        
        for new_match in new_matches:
            is_duplicate = False
            
            for existing_match in existing_matches:
                # Check for overlap
                overlap_start = max(new_match.start, existing_match.start)
                overlap_end = min(new_match.end, existing_match.end)
                overlap_length = max(0, overlap_end - overlap_start)
                
                new_match_length = new_match.end - new_match.start
                overlap_ratio = overlap_length / new_match_length
                
                if overlap_ratio > overlap_threshold:
                    is_duplicate = True
                    break
            
            if not is_duplicate:
                unique_matches.append(new_match)
        
        return unique_matches
    
    def _convert_llm_results(
        self,
        llm_result: Dict,
        text: str,
    ) -> List[PIIMatch]:
        """Convert LLM detection results to PIIMatch format."""
        
        matches = []
        
        if not llm_result.get('pii_detected'):
            return matches
        
        for entity in llm_result.get('entities', []):
            # Find position in text
            value = entity['value']
            start = text.find(value)
            
            if start != -1:
                matches.append(PIIMatch(
                    pii_type=entity['type'],
                    value=value,
                    start=start,
                    end=start + len(value),
                    confidence=0.95,  # LLM-detected
                ))
        
        return matches
    
    def _calculate_confidence(
        self,
        matches: List[PIIMatch],
        layers_used: List[str],
    ) -> float:
        """Calculate overall confidence in PII detection."""
        
        if not matches:
            # High confidence of no PII if all layers ran
            return 0.9 if len(layers_used) >= 2 else 0.7
        
        # Average confidence of matches
        avg_match_confidence = sum(m.confidence for m in matches) / len(matches)
        
        # Boost confidence if multiple layers agree
        layer_boost = 0.05 * (len(layers_used) - 1)
        
        return min(1.0, avg_match_confidence + layer_boost)


# Production usage
pipeline = ProductionPIIDetectionPipeline(
    use_ner=True,
    use_llm=False,  # Enable for highest accuracy (adds 1-2s latency)
)

# Example: RAG document before indexing
document = """
Patient Record:
Name: Sarah Johnson
DOB: 03/15/1980
Email: sarah.j@email.com
Phone: (555) 123-4567
MRN: 87654321

Chief Complaint: Routine checkup
Attending: Dr. Michael Chen
"""

result = pipeline.detect_and_redact(document)

print(f"PII detected: {result.has_pii}")
print(f"Matches: {len(result.matches)}")
print(f"Layers used: {result.detection_layers_used}")
print(f"Latency: {result.latency_ms:.1f}ms")
print(f"Confidence: {result.confidence:.2f}")
print(f"\nRedacted document:\n{result.redacted_text}")

RAG Pipeline Integration

python
class PIISafeRAGPipeline:
    """RAG pipeline with PII scrubbing at all stages."""
    
    def __init__(self, pii_detector: ProductionPIIDetectionPipeline):
        self.pii_detector = pii_detector
    
    def ingest_document(self, document: str) -> str:
        """Scrub PII before indexing."""
        
        result = self.pii_detector.detect_and_redact(document)
        
        if result.has_pii:
            print(f"⚠️  PII detected in document: {len(result.matches)} instances")
            # Log for compliance audit
            self._log_pii_redaction(result)
        
        # Index only redacted version
        return result.redacted_text
    
    def retrieve_and_generate(self, query: str) -> str:
        """Retrieve docs and generate response with PII checking."""
        
        # 1. Scrub PII from query
        query_result = self.pii_detector.detect_and_redact(query)
        safe_query = query_result.redacted_text
        
        # 2. Retrieve documents (already scrubbed during ingestion)
        docs = self._retrieve(safe_query)
        
        # 3. Generate response
        llm_output = self._generate(safe_query, docs)
        
        # 4. Final PII check on output (LLM may hallucinate PII)
        output_result = self.pii_detector.detect_and_redact(llm_output)
        
        if output_result.has_pii:
            print(f"⚠️  LLM generated PII: {len(output_result.matches)} instances")
            self._log_pii_generation(output_result)
        
        return output_result.redacted_text
    
    def _log_pii_redaction(self, result: PIIDetectionResult):
        """Log PII redaction for compliance."""
        import json
        import logging
        
        logging.info(json.dumps({
            'event': 'pii_redaction',
            'timestamp': datetime.now().isoformat(),
            'pii_types': [m.pii_type for m in result.matches],
            'count': len(result.matches),
        }))

Compliance and Auditing

GDPR/HIPAA Audit Trail

python
from dataclasses import dataclass
from datetime import datetime
from typing import List

@dataclass
class PIIAuditLog:
    """Audit log entry for PII processing."""
    timestamp: datetime
    event_type: str  # "detection", "redaction", "access"
    user_id: str
    pii_types: List[str]
    action_taken: str
    justification: str
    system_component: str

class PIIAuditLogger:
    """Compliance-focused audit logging."""
    
    def __init__(self):
        self.logs: List[PIIAuditLog] = []
    
    def log_pii_detection(
        self,
        user_id: str,
        pii_types: List[str],
        component: str,
    ):
        """Log PII detection event."""
        
        self.logs.append(PIIAuditLog(
            timestamp=datetime.now(),
            event_type="detection",
            user_id=user_id,
            pii_types=pii_types,
            action_taken="redacted",
            justification="GDPR/HIPAA compliance",
            system_component=component,
        ))
    
    def generate_compliance_report(
        self,
        start_date: datetime,
        end_date: datetime,
    ) -> Dict:
        """Generate compliance report for auditors."""
        
        relevant_logs = [
            log for log in self.logs
            if start_date <= log.timestamp <= end_date
        ]
        
        return {
            'period': {
                'start': start_date.isoformat(),
                'end': end_date.isoformat(),
            },
            'total_detections': len(relevant_logs),
            'pii_types_detected': self._aggregate_pii_types(relevant_logs),
            'actions_taken': self._aggregate_actions(relevant_logs),
            'systems_involved': list(set(log.system_component for log in relevant_logs)),
        }
    
    def _aggregate_pii_types(self, logs: List[PIIAuditLog]) -> Dict[str, int]:
        """Count detections by PII type."""
        from collections import Counter
        
        all_types = [pii_type for log in logs for pii_type in log.pii_types]
        return dict(Counter(all_types))
    
    def _aggregate_actions(self, logs: List[PIIAuditLog]) -> Dict[str, int]:
        """Count actions taken."""
        from collections import Counter
        
        actions = [log.action_taken for log in logs]
        return dict(Counter(actions))

Primary references: official documentation, official documentation, official documentation, official documentation.

PII Detection and Scrubbing in LLM Pipelines Decision Table

DecisionPrefer the simpler path whenAdd operational complexity when
ArchitectureOne component can own the contract and stateIndependent scaling or fault isolation is required
RolloutOffline replay covers the meaningful casesLive behavior requires shadow traffic and a canary
RecoveryA failed operation is safe to repeatPartial effects require idempotency or reconciliation
MeasurementOne service objective represents user impactQuality, latency, and cost need separate gates

Operating PII Detection and Scrubbing in LLM Pipelines as a System

The implementation is only one part of PII Detection and Scrubbing in LLM Pipelines. A production design also needs an explicit contract for inputs, outputs, ownership, and failure behavior. Write that contract before selecting a library. It should identify which component validates input, where state lives, what may be retried, and which result is authoritative when two components disagree. This prevents a convenient prototype boundary from silently becoming the long-term architecture.

Start with a representative baseline. Capture request shape, traffic distribution, dependency latency, error classes, and the quality signal users actually care about. Averages hide the cases that cause incidents, so keep percentiles and segment measurements by workload type. Record the configuration and dataset version beside every result. Without that context, a faster or more accurate run cannot be reproduced and should not be used to approve a rollout.

Define the failure model

List failures by where they originate: invalid input, capacity exhaustion, dependency timeout, partial state change, malformed output, and semantically wrong output. Each class needs a different response. Validation errors should fail immediately. Transient dependency failures may be retried with a budget and jitter. An operation that may have committed must use an idempotency key or reconciliation step before retrying. A syntactically valid but incorrect result belongs in evaluation and review, not a blind retry loop.

Set a deadline for the complete operation and derive smaller budgets for each dependency. Local timeouts that add up to more than the caller's deadline merely create abandoned work. Propagate cancellation where the protocol supports it. Bound every queue, retry loop, context buffer, and concurrency pool; an unbounded safety mechanism becomes a second outage during overload.

Design a degraded mode before it is needed. Depending on the workload, that can mean returning a cached answer, selecting a simpler path, placing work in a durable queue, or asking for human review. The degraded response must be visible in telemetry and, where it changes meaning, visible to the caller. Silent fallback makes quality regressions almost impossible to diagnose.

Measure the decision, not just the component

Use three layers of signals. System metrics cover latency, throughput, saturation, and errors. Correctness metrics measure whether the result satisfies its contract. Business or user metrics show whether the system solved the intended problem. Improving only one layer can move the others backward, so release criteria should name acceptable movement for all three.

Attach a reason code to every route, rejection, fallback, and retry. Include version identifiers for configuration, code, model, schema, and data when relevant. Logs should let an engineer reconstruct a decision without storing secrets or raw personal data. Traces should cross process boundaries, while metrics should remain low-cardinality enough to operate reliably.

Alert on symptoms that require action, not every internal anomaly. A useful alert names the affected service objective, links to a runbook, and distinguishes a customer-visible incident from exhausted headroom. Dashboards serve a different purpose: they support diagnosis and capacity planning. Treating a dashboard as an alerting strategy leaves failures undiscovered until someone happens to look.

Roll out with reversible steps

Ship PII Detection and Scrubbing in LLM Pipelines behind a versioned interface and a kill switch. Begin with offline replay using production-shaped, privacy-safe samples. Then use shadow execution when duplicate work has acceptable cost and side effects can be suppressed. A small canary should exercise the real dependency graph before traffic expands. Compare the canary with the baseline by cohort rather than mixing both populations into one aggregate.

Promotion gates should be written before the rollout. Include a minimum sample size or observation window, maximum regression in tail latency and error rate, and a correctness threshold. Roll back automatically when a hard safety boundary is crossed; use manual review for ambiguous quality movement. Preserve enough evidence from both paths to explain why the gate passed or failed.

Configuration deserves the same discipline as code. Review changes, validate them before activation, keep an immutable history, and make rollback a single operation. If a deployment changes code and configuration together, record both versions. Otherwise an incident responder may roll back the binary while leaving the triggering configuration active.

Capacity and cost controls

Model capacity in units the bottleneck understands: concurrent connections, tokens, queue jobs, database transactions, GPU memory, or bytes in flight. Convert the expected traffic distribution into those units and include burst behavior. Then load-test the first constrained dependency, not merely the public endpoint. A system that accepts more work than it can finish within its deadline is overloaded even if CPU utilization looks comfortable.

Cost is also a reliability limit. Add per-request attribution, tenant or workflow budgets, and a global circuit breaker for unexpectedly expensive paths. Review unit economics at the same granularity as performance; a cheap median can conceal a small class of requests responsible for most spend. Optimize only after measuring, because reducing context, replicas, validation, or redundancy can trade visible cost for less visible risk.

Production readiness review

Before launch, ask an engineer who did not build the feature to follow the runbook through one simulated failure. Verify backups or checkpoints by restoring them, not by checking that a job reported success. Exercise credential rotation, dependency unavailability, bad configuration, and rollback. Assign an owner for each alarm and a date for reviewing thresholds after real traffic arrives.

The final architecture document should be short enough to remain current. Keep the decision, rejected alternatives, invariants, dependency contracts, dashboards, and rollback procedure. Link detailed experiments rather than pasting them into the document. Teams that need help turning this review into an operable service can use our PII Detection and Scrubbing in LLM Pipelines engineering support.

Frequently Asked Questions

What's the performance impact of PII detection?

  • Regex only: <5ms
  • Regex + NER: 50-150ms (with GPU)
  • Regex + NER + LLM: 1-2 seconds

For real-time systems, use regex + NER. Reserve LLM for offline document processing.

Should I redact all PII or just sensitive fields?

Depends on your risk tolerance and regulations:

  • Healthcare (HIPAA): Redact all 18 HIPAA identifiers
  • Financial (PCI DSS): Redact credit cards, account numbers
  • General (GDPR): Risk-based approach — redact direct identifiers, evaluate quasi-identifiers

Can I use PII detection on LLM training data?

Yes, essential for compliance. Run detection pipeline on training data before fine-tuning. Many data breaches come from LLMs memorizing PII from training sets.

How do I handle PII in embeddings?

Embeddings encode semantic meaning, including PII. Options:

  1. Redact before embedding: Loses some context but safer
  2. Differential privacy: Add noise to embeddings
  3. Encrypted embeddings: Emerging research area

Most production systems redact before embedding.

What if PII detection has false positives?

Accept some false positives for safety. In production:

  1. Log all redactions
  2. Human review flagged cases
  3. Whitelist known safe patterns (e.g., "John Doe" in documentation)
  4. Tune confidence thresholds per use case

How do I detect PII in images (multimodal)?

Use OCR + PII detection for text in images. For faces:

  • Face detection models (MTCNN, RetinaFace)
  • Face recognition if matching against known individuals
  • Blur/redact detected faces

Can attackers bypass PII detection?

Yes, through:

  • Encoding (Base64, ROT13)
  • Obfuscation (homoglyphs, zero-width spaces)
  • Indirect references ("my contact info from earlier")

Use prompt injection defense in addition to PII detection.

How long should I retain redaction maps?

Per GDPR: Minimum necessary for business purpose. Typical:

  • Customer support: 90 days
  • Legal/compliance: 7 years
  • Research/analytics: Delete after analysis

Encrypt and access-control all redaction maps.

What's the difference between anonymization and pseudonymization?

Anonymization: Irreversible removal (no way to recover) Pseudonymization: Reversible replacement (can restore with key)

GDPR requires pseudonymization for many use cases. Use reversible redaction with secure key storage.

How do I test PII detection accuracy?

Build test dataset with:

  • Known PII instances (ground truth)
  • Edge cases (partial PII, obfuscation)
  • False positive triggers (non-PII that looks like PII)

Measure:

  • Recall: % of actual PII detected
  • Precision: % of detections that are actual PII
  • F1 score: Harmonic mean of precision/recall

Target: >95% recall, <10% false positive rate.

Conclusion

PII detection is mandatory for production LLM systems handling user data. Every layer of the LLM pipeline — training data, RAG contexts, tool inputs/outputs, and generations — must be scrubbed.

The production pattern:

  1. Multi-layer detection (regex + NER + optional LLM)
  2. Real-time scrubbing (< 100ms added latency with regex+NER)
  3. Context-preserving redaction (semantic tokens, not full removal)
  4. Compliance auditing (log all detections, generate reports)

Target >95% recall with <10% false positives using layered approach.

For teams building compliant LLM systems or AI agents with GDPR/HIPAA requirements, we've implemented PII detection across healthcare, financial services, and enterprise applications processing millions of sensitive documents.

Related reading: Prompt Injection Defense, Output Guardrails, AI Red Teaming, Content Moderation, OWASP LLM Security.

Free consultation

Book a free consultation call on PII protection in LLM systems

30-minute call with the HinterBuild team. Discuss your project, architecture questions, or next steps — no obligation.

Book a meeting

Keep reading

Related articles

AI Systems · 10 min read

Triton vs vLLM: LLM Serving Framework Comparison for

Triton vs vLLM guidance for engineers: compare architecture choices, avoid failure modes, and ship a measurable, reliable production implementation.

Read post

AI Systems · 9 min read

LLM Tracing with OpenTelemetry: Complete Observability Guide

Learn llm tracing with opentelemetry through concrete architecture trade-offs, failure modes, rollout controls, and production measurement practices.

Read post

AI Systems · 9 min read

Synthetic Data Generation for LLM Evals

Synthetic Data Generation for LLM Evals guidance for engineers: compare architecture choices, avoid failure modes, and ship a measurable, reliable.

Read post

AI Systems · 11 min read

OWASP Top 10 for LLM Applications: Complete Security Guide

Learn owasp top 10 for llm applications through concrete architecture trade-offs, failure modes, rollout controls, and production measurement practices.

Read post