Ai Redacting Data

Redacting PII and Sensitive Data with DSPy

Strip personal information and sensitive data from text before it reaches an LM — or before it leaves your system.

Step 1 - Understand What to Redact

Before writing code, answer three questions:

1. What PII types? Names, emails, phones, SSNs, credit cards, addresses, dates of birth, IP addresses, medical record numbers (MRNs), or all of the above. 2. Replacement strategy? See the table in Step 3. 3. Compliance requirement? GDPR (EU personal data), HIPAA (US health data), PCI-DSS (payment data), or internal policy.

The answers drive which pipeline path you need.

---

Step 2 - Set Up DSPy

import dspy
import re
from dataclasses import dataclass, field
from typing import Literal

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

---

Step 3 - Replacement Strategies

---

Step 4 - Regex First for Structured Patterns

Regex is fast, deterministic, and never sends PII to an external API. Always run it before the LM pass.

# Patterns for structured PII
PATTERNS = {
    "EMAIL":       re.compile(r'\b[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}\b'),
    "PHONE":       re.compile(r'\b(\+?1[-.\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'),
    "IP_ADDRESS":  re.compile(r'\b\d{1,3}(?:\.\d{1,3}){3}\b'),
    "DATE_OF_BIRTH": re.compile(r'\b(?:DOB|Date of Birth|born)[:\s]+\d{1,2}[/\-]\d{1,2}[/\-]\d{2,4}\b', re.IGNORECASE),
    "ZIP_CODE":    re.compile(r'\b\d{5}(?:-\d{4})?\b'),
}

@dataclass
class PIIMatch:
    pii_type: str
    value: str
    start: int
    end: int

def regex_detect(text: str) -> list[PIIMatch]:
    matches = []
    for pii_type, pattern in PATTERNS.items():
        for m in pattern.finditer(text):
            matches.append(PIIMatch(pii_type=pii_type, value=m.group(), start=m.start(), end=m.end()))
    return matches

---

Step 5 - LM Signature for Contextual PII

Use the LM only for PII that requires reading context — names, addresses, and other free-form entities.

class DetectContextualPII(dspy.Signature):
    """Identify personal information in text that requires context to detect.
    Return a JSON list of objects with fields: pii_type, value.
    PII types to detect - PERSON_NAME, ADDRESS, MEDICAL_RECORD_NUMBER, ORG_NAME (when linked to a person).
    Do not flag generic words that happen to resemble names."""

    text: str = dspy.InputField(desc="Text to scan for personal information")
    pii_entities: list[dict] = dspy.OutputField(
        desc='JSON list - [{"pii_type": "PERSON_NAME", "value": "Jane Doe"}, ...]'
    )

detect_pii = dspy.Predict(DetectContextualPII)

---

Step 6 - Full Redaction Module

class PIIRedactor(dspy.Module):
    def __init__(self, strategy: Literal["placeholder", "indexed", "blank"] = "placeholder"):
        self.strategy = strategy
        self.detect = dspy.Predict(DetectContextualPII)

    def _make_replacement(self, pii_type: str, entity_index: dict) -> str:
        if self.strategy == "indexed":
            key = pii_type
            n = entity_index.get(key, 0) + 1
            entity_index[key] = n
            return f"[{pii_type}_{n}]"
        elif self.strategy == "blank":
            return "████"
        else:
            return f"[{pii_type}]"

    def forward(self, text: str) -> dspy.Prediction:
        entity_index: dict[str, int] = {}
        seen: dict[str, str] = {}  # value → replacement (for consistency)

        # Pass 1 - regex for structured patterns
        regex_hits = regex_detect(text)

        # Pass 2 - LM for contextual PII (only send text with structured PII pre-masked)
        pre_masked = text
        for hit in sorted(regex_hits, key=lambda h: h.start, reverse=True):
            pre_masked = pre_masked[:hit.start] + f"[{hit.pii_type}]" + pre_masked[hit.end:]

        lm_result = self.detect(text=pre_masked)
        lm_entities = lm_result.pii_entities or []

        # Build replacement map from LM entities
        for entity in lm_entities:
            val = entity.get("value", "")
            pii_type = entity.get("pii_type", "PII")
            if val and val not in seen:
                seen[val] = self._make_replacement(pii_type, entity_index)

        # Apply LM replacements to original text
        redacted = text
        for val, replacement in sorted(seen.items(), key=lambda kv: len(kv[0]), reverse=True):
            redacted = redacted.replace(val, replacement)

        # Apply regex replacements
        for hit in sorted(regex_hits, key=lambda h: h.start, reverse=True):
            if hit.value not in seen:
                seen[hit.value] = self._make_replacement(hit.pii_type, entity_index)

        # Re-apply to get a clean final pass
        final = text
        for val, replacement in sorted(seen.items(), key=lambda kv: len(kv[0]), reverse=True):
            final = final.replace(val, replacement)

        return dspy.Prediction(
            redacted_text=final,
            entities_found=seen,
        )

---

Step 7 - Validate Redaction Quality

Do not use dspy.Assert or dspy.Suggest here — they are deprecated. Use dspy.Refine with a reward function.

class ValidateRedaction(dspy.Signature):
    """Check whether any PII survived redaction. Return True if clean, False if PII remains."""

    original_text: str = dspy.InputField()
    redacted_text: str = dspy.InputField()
    is_clean: bool = dspy.OutputField(desc="True if no PII remains, False otherwise")
    leaked_examples: list[str] = dspy.OutputField(desc="Examples of PII that leaked through, empty list if clean")

def redaction_reward(example, prediction, trace=None) -> float:
    validator = dspy.Predict(ValidateRedaction)
    result = validator(
        original_text=example.text,
        redacted_text=prediction.redacted_text,
    )
    return 1.0 if result.is_clean else 0.0

---

Step 8 - GDPR and HIPAA Compliance Patterns

GDPR - Right to erasure

# Store the entity map so you can reverse-map or fully erase later
redactor = PIIRedactor(strategy="indexed")
result = redactor(text=document)
# Persist result.entities_found keyed by document ID
# On erasure request - delete the mapping; ciphertext becomes permanently anonymized

HIPAA - Safe Harbor de-identification

HIPAA Safe Harbor requires removing 18 PHI identifiers. Add these patterns:

HIPAA_PATTERNS = {
    "MRN":        re.compile(r'\bMRN[:\s#]+\w+\b', re.IGNORECASE),
    "NPI":        re.compile(r'\bNPI[:\s#]+\d{10}\b', re.IGNORECASE),
    "DEVICE_ID":  re.compile(r'\b(?:device|serial)[:\s#]+[A-Z0-9\-]{6,}\b', re.IGNORECASE),
    "URL":        re.compile(r'https?://\S+'),
    "ACCOUNT":    re.compile(r'\baccount[:\s#]+\w+\b', re.IGNORECASE),
}
PATTERNS.update(HIPAA_PATTERNS)

---

Step 9 - When NOT to Use AI Redaction

• Structured fields with known formats - regex alone is sufficient and faster (emails, SSNs, credit cards).
• Already-tokenized data - if PII was never collected as free text, there is nothing to redact.
• When you can avoid collecting PII in the first place - the best redaction is prevention.
• High-stakes legal documents without human review - LM redaction can miss things; always add a human-in-the-loop audit step for compliance filings.

---

Key Patterns

# Quick usage
redactor = PIIRedactor(strategy="indexed")
result = redactor(text="Call Jane Doe at 555-123-4567 or jane@example.com")
print(result.redacted_text)
# "Call [PERSON_NAME_1] at [PHONE_1] or [EMAIL]"
print(result.entities_found)
# {"Jane Doe": "[PERSON_NAME_1]", "555-123-4567": "[PHONE_1]", "jane@example.com": "[EMAIL]"}

---

Gotchas

• The LM sees the PII you are trying to hide - sending raw text to an external LM for detection defeats the purpose if the PII itself is sensitive. Run regex first and send only the pre-masked text to the LM, or use a locally hosted model.

• Common words misidentified as names - Claude flags "Will" (a verb), "Mark" (a noun), "Faith" (a concept) as PERSON_NAME. Prompt the signature to exclude words that are clearly not names in context, and validate detections against a stoplist.

• Inconsistent placeholders break co-reference - without a seen mapping dict, the same person can appear as [PERSON_1] in paragraph 1 and [PERSON_2] in paragraph 3. Always deduplicate entity values before assigning replacements.

• Non-English and transliterated names are missed - Claude's contextual PII detection is weakest on names from languages with different romanization conventions (e.g., Chinese pinyin, Arabic transliteration). Add language-specific name lists or a multilingual NER model for those cases.

• Using `dspy.Assert`/`dspy.Suggest` for validation is outdated - those APIs are removed in DSPy 2.5+. Use dspy.Refine with a reward function as shown in Step 7.

---

Cross-References

Install any skill: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill <name>

• /ai-parsing-data - extract structured fields from text (complementary pattern)
• /ai-checking-outputs - validate that outputs meet quality criteria
• /dspy-refine - iterative refinement with a reward function for validation loops
• /dspy-retrieval - if you need to redact before indexing documents
• Install `/ai-do` if you do not have it — it routes any AI problem to the right skill and is the fastest way to work: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill ai-do

Additional Resources

See examples.md for worked examples: 1. Customer support email redactor 2. Medical record de-identifier (HIPAA Safe Harbor) 3. Pre-LLM sanitizer for third-party API calls

ai-redacting-data - Examples

Example 1 - Customer Support Email Redactor

Mask names, email addresses, and account numbers before routing tickets to a shared inbox or logging system.

import dspy
import re
from dataclasses import dataclass
from typing import Literal

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

PATTERNS = {
    "EMAIL":   re.compile(r'\b[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}\b'),
    "PHONE":   re.compile(r'\b(\+?1[-.\s]?)?(\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4})\b'),
    "ACCOUNT": re.compile(r'\baccount[:\s#]+\w+\b', re.IGNORECASE),
    "ORDER":   re.compile(r'\border[:\s#]+[A-Z0-9\-]+\b', re.IGNORECASE),
}

class DetectSupportPII(dspy.Signature):
    """Find names and personal identifiers in a customer support email.
    Return JSON list with pii_type (PERSON_NAME or ORG_NAME) and value."""

    text: str = dspy.InputField()
    pii_entities: list[dict] = dspy.OutputField()

class SupportEmailRedactor(dspy.Module):
    def __init__(self):
        self.detect = dspy.Predict(DetectSupportPII)
        self._seen: dict[str, str] = {}
        self._counters: dict[str, int] = {}

    def _placeholder(self, pii_type: str, value: str) -> str:
        if value in self._seen:
            return self._seen[value]
        n = self._counters.get(pii_type, 0) + 1
        self._counters[pii_type] = n
        label = f"[{pii_type}_{n}]"
        self._seen[value] = label
        return label

    def forward(self, email_text: str) -> dspy.Prediction:
        self._seen.clear()
        self._counters.clear()

        # Regex pass
        pre_masked = email_text
        regex_replacements = []
        for pii_type, pattern in PATTERNS.items():
            for m in pattern.finditer(email_text):
                label = self._placeholder(pii_type, m.group())
                regex_replacements.append((m.group(), label))

        for value, label in regex_replacements:
            pre_masked = pre_masked.replace(value, label)

        # LM pass on pre-masked text
        result = self.detect(text=pre_masked)
        for entity in (result.pii_entities or []):
            val = entity.get("value", "")
            pii_type = entity.get("pii_type", "PII")
            if val:
                self._placeholder(pii_type, val)

        # Final replacement on original
        final = email_text
        for value, label in sorted(self._seen.items(), key=lambda kv: len(kv[0]), reverse=True):
            final = final.replace(value, label)

        return dspy.Prediction(redacted=final, entity_map=dict(self._seen))


# Usage
redactor = SupportEmailRedactor()

email = """
Hi Support,

My name is Sarah Chen and I'm having trouble with my account #AC-88421.
I placed order #ORD-2024-99012 last Tuesday and it still hasn't shipped.
Please reach me at sarah.chen@gmail.com or 415-555-0192.

Thanks,
Sarah
"""

result = redactor(email_text=email.strip())
print(result.redacted)
# Hi Support,
#
# My name is [PERSON_NAME_1] and I'm having trouble with my account [ACCOUNT_1].
# I placed [ORDER_1] last Tuesday and it still hasn't shipped.
# Please reach me at [EMAIL_1] or [PHONE_1].
#
# Thanks,
# [PERSON_NAME_1]

print(result.entity_map)
# {"Sarah Chen": "[PERSON_NAME_1]", "AC-88421": "[ACCOUNT_1]", ...}

---

Example 2 - Medical Record De-Identifier (HIPAA Safe Harbor)

Remove all 18 PHI categories required by HIPAA Safe Harbor before storing or sharing medical notes.

import dspy
import re

lm = dspy.LM("openai/gpt-4o-mini")  # or use a local model to keep PHI off external APIs
dspy.configure(lm=lm)

# HIPAA Safe Harbor PHI patterns
HIPAA_PATTERNS = {
    "NAME":       None,  # handled by LM
    "DATE":       re.compile(r'\b(?:Jan(?:uary)?|Feb(?:ruary)?|Mar(?:ch)?|Apr(?:il)?|May|Jun(?:e)?|'
                             r'Jul(?:y)?|Aug(?:ust)?|Sep(?:tember)?|Oct(?:ober)?|Nov(?:ember)?|'
                             r'Dec(?:ember)?)\s+\d{1,2},?\s+\d{4}\b|\b\d{1,2}[/\-]\d{1,2}[/\-]\d{2,4}\b',
                             re.IGNORECASE),
    "PHONE":      re.compile(r'\b(\+?1[-.\s]?)?(\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4})\b'),
    "FAX":        re.compile(r'\bfax[:\s]+[\d\-\(\)\s]+', re.IGNORECASE),
    "EMAIL":      re.compile(r'\b[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}\b'),
    "SSN":        re.compile(r'\b\d{3}-\d{2}-\d{4}\b'),
    "MRN":        re.compile(r'\bMRN[:\s#]+\w+\b', re.IGNORECASE),
    "ACCOUNT":    re.compile(r'\baccount[:\s#]+\w+\b', re.IGNORECASE),
    "ZIP":        re.compile(r'\b\d{5}(?:-\d{4})?\b'),
    "IP_ADDRESS": re.compile(r'\b\d{1,3}(?:\.\d{1,3}){3}\b'),
    "URL":        re.compile(r'https?://\S+'),
    "NPI":        re.compile(r'\bNPI[:\s#]+\d{10}\b', re.IGNORECASE),
    "DOB":        re.compile(r'\b(?:DOB|Date of Birth|born)[:\s]+\d{1,2}[/\-]\d{1,2}[/\-]\d{2,4}\b',
                             re.IGNORECASE),
    "AGE_OVER_89": re.compile(r'\b(?:age[:\s]+)?(?:9\d|1[0-9]{2})\s*(?:years?\s*old|y/?o)\b',
                               re.IGNORECASE),
}

class DetectHIPAAPHI(dspy.Signature):
    """Identify HIPAA Protected Health Information requiring contextual understanding.
    Focus on - patient names, provider names, geographic subdivisions smaller than state,
    device identifiers, and certificate or license numbers.
    Do not flag medical conditions or treatment descriptions."""

    text: str = dspy.InputField(desc="Clinical note with structured PHI already masked")
    phi_entities: list[dict] = dspy.OutputField(
        desc='List of {"phi_type": "PATIENT_NAME", "value": "..."}'
    )

class HIPAADeidentifier(dspy.Module):
    def __init__(self):
        self.detect_phi = dspy.Predict(DetectHIPAAPHI)

    def forward(self, clinical_note: str) -> dspy.Prediction:
        seen: dict[str, str] = {}
        counters: dict[str, int] = {}

        def label(phi_type: str, value: str) -> str:
            if value in seen:
                return seen[value]
            n = counters.get(phi_type, 0) + 1
            counters[phi_type] = n
            tag = f"[{phi_type}]" if n == 1 else f"[{phi_type}_{n}]"
            seen[value] = tag
            return tag

        # Regex pass
        pre_masked = clinical_note
        for phi_type, pattern in HIPAA_PATTERNS.items():
            if pattern is None:
                continue
            for m in pattern.finditer(clinical_note):
                tag = label(phi_type, m.group())
                pre_masked = pre_masked.replace(m.group(), tag)

        # LM pass
        lm_result = self.detect_phi(text=pre_masked)
        for entity in (lm_result.phi_entities or []):
            val = entity.get("value", "")
            phi_type = entity.get("phi_type", "PHI")
            if val:
                label(phi_type, val)

        # Final pass on original text
        final = clinical_note
        for value, tag in sorted(seen.items(), key=lambda kv: len(kv[0]), reverse=True):
            final = final.replace(value, tag)

        return dspy.Prediction(deidentified=final, phi_removed=len(seen))


# Usage
deidentifier = HIPAADeidentifier()

note = """
Patient: Maria Gonzalez  DOB: 03/14/1962  MRN: 8821044
Provider: Dr. James Whitfield, NPI: 1234567890
Visit date: April 3, 2025

Chief complaint: Patient presents with chest pain since 02/28/2025.
Contact: (602) 555-7734  |  maria.gonzalez@yahoo.com
Address: 4521 W. Camelback Rd, Phoenix, AZ 85031
"""

result = deidentifier(clinical_note=note.strip())
print(result.deidentified)
print(f"PHI entities removed - {result.phi_removed}")

---

Example 3 - Pre-LLM Sanitizer for Third-Party API Calls

Redact sensitive data before sending to an external AI API, then restore context in the response.

import dspy
import re
import hashlib

lm = dspy.LM("openai/gpt-4o-mini")  # internal/trusted model for detection
dspy.configure(lm=lm)

PATTERNS = {
    "EMAIL":       re.compile(r'\b[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}\b'),
    "PHONE":       re.compile(r'\b(\+?1[-.\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'),
    "IP_ADDRESS":  re.compile(r'\b\d{1,3}(?:\.\d{1,3}){3}\b'),
}

class DetectNames(dspy.Signature):
    """Find person names in text. Return JSON list of name strings only."""
    text: str = dspy.InputField()
    names: list[str] = dspy.OutputField()

class PreLLMSanitizer(dspy.Module):
    """Redact PII, call an external LM, return response with placeholders intact."""

    def __init__(self):
        self.detect_names = dspy.Predict(DetectNames)
        # External LM (untrusted with PII)
        self.external_lm = dspy.LM("openai/gpt-4o")  # or any external provider
        self.external_task = dspy.ChainOfThought("sanitized_text -> summary")

    def _hash_token(self, value: str) -> str:
        return "[" + hashlib.sha256(value.encode()).hexdigest()[:8].upper() + "]"

    def forward(self, text: str, task: str = "summarize") -> dspy.Prediction:
        token_map: dict[str, str] = {}  # token → original value (for optional restore)
        seen: dict[str, str] = {}       # original value → token

        # Step 1 - regex redaction
        sanitized = text
        for pii_type, pattern in PATTERNS.items():
            for m in pattern.finditer(text):
                if m.group() not in seen:
                    token = self._hash_token(m.group())
                    seen[m.group()] = token
                    token_map[token] = m.group()

        # Step 2 - name detection on pre-sanitized text
        pre = text
        for val, tok in sorted(seen.items(), key=lambda kv: len(kv[0]), reverse=True):
            pre = pre.replace(val, tok)
        name_result = self.detect_names(text=pre)
        for name in (name_result.names or []):
            if name and name not in seen:
                token = self._hash_token(name)
                seen[name] = token
                token_map[token] = name

        # Step 3 - build fully sanitized text
        for val, tok in sorted(seen.items(), key=lambda kv: len(kv[0]), reverse=True):
            sanitized = sanitized.replace(val, tok)

        # Step 4 - call external LM with sanitized text
        with dspy.context(lm=self.external_lm):
            ext_result = self.external_task(sanitized_text=sanitized)

        return dspy.Prediction(
            sanitized_input=sanitized,
            external_response=ext_result.summary,
            token_map=token_map,  # keep for optional de-anonymization
        )


# Usage
sanitizer = PreLLMSanitizer()

doc = """
John Martinez submitted a support ticket about a billing issue.
His account was charged $299 twice on 2025-04-01.
Email: john.martinez@company.com | Phone: 312-555-8820
CC ending in 4532 was charged.
"""

result = sanitizer(text=doc.strip())
print("Sanitized input sent to external LM:")
print(result.sanitized_input)
print("\nExternal LM response (contains only tokens, no PII):")
print(result.external_response)
# All PII replaced with hash tokens - safe to log, cache, or display