Data Privacy in AI Agents: GDPR, HIPAA, and PII Handling Best Practices

AI Agents and the Privacy Challenge

AI agents create unique privacy challenges that traditional software does not face. An agent might receive PII in user messages, retrieve sensitive data from databases, include personal information in LLM prompts sent to third-party APIs, and store conversation logs containing protected health information. Every one of these operations is a potential compliance violation under GDPR, HIPAA, CCPA, or other data protection regulations.

This post builds practical systems for classifying data, anonymizing PII, managing retention, and handling consent in AI agent applications.

Data Classification Pipeline

The first step in privacy compliance is knowing what data you have. A classification pipeline automatically labels data flowing through your agent:

from dataclasses import dataclass
from enum import Enum
from typing import Optional
import re

class DataSensitivity(Enum):
    PUBLIC = "public"
    INTERNAL = "internal"
    CONFIDENTIAL = "confidential"
    RESTRICTED = "restricted"  # PII, PHI, financial data

class PIIType(Enum):
    EMAIL = "email"
    PHONE = "phone"
    SSN = "ssn"
    NAME = "name"
    ADDRESS = "address"
    DOB = "date_of_birth"
    MEDICAL = "medical_record"
    FINANCIAL = "financial_account"

@dataclass
class ClassificationResult:
    sensitivity: DataSensitivity
    pii_types_found: list[PIIType]
    requires_anonymization: bool
    requires_consent: bool
    applicable_regulations: list[str]

class DataClassifier:
    PII_PATTERNS = {
        PIIType.EMAIL: r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}",
        PIIType.PHONE: r"\b(?:\+?1[-.]?)?\(?\d{3}\)?[-.]?\d{3}[-.]?\d{4}\b",
        PIIType.SSN: r"\b\d{3}-\d{2}-\d{4}\b",
        PIIType.DOB: r"\b(?:0[1-9]|1[0-2])/(?:0[1-9]|[12]\d|3[01])/(?:19|20)\d{2}\b",
    }

    MEDICAL_KEYWORDS = [
        "diagnosis", "prescription", "medication", "treatment",
        "blood pressure", "heart rate", "patient", "symptom",
        "allergies", "medical history", "lab results",
    ]

    FINANCIAL_KEYWORDS = [
        "account number", "routing number", "credit card",
        "bank account", "social security", "tax id", "ein",
    ]

    def classify(self, text: str) -> ClassificationResult:
        pii_types = []

        for pii_type, pattern in self.PII_PATTERNS.items():
            if re.search(pattern, text):
                pii_types.append(pii_type)

        lower_text = text.lower()
        if any(kw in lower_text for kw in self.MEDICAL_KEYWORDS):
            pii_types.append(PIIType.MEDICAL)
        if any(kw in lower_text for kw in self.FINANCIAL_KEYWORDS):
            pii_types.append(PIIType.FINANCIAL)

        regulations = []
        if pii_types:
            regulations.append("GDPR")
            regulations.append("CCPA")
        if PIIType.MEDICAL in pii_types:
            regulations.append("HIPAA")

        sensitivity = DataSensitivity.PUBLIC
        if pii_types:
            sensitivity = DataSensitivity.RESTRICTED
        elif any(kw in lower_text for kw in ["internal", "confidential"]):
            sensitivity = DataSensitivity.CONFIDENTIAL

        return ClassificationResult(
            sensitivity=sensitivity,
            pii_types_found=pii_types,
            requires_anonymization=bool(pii_types),
            requires_consent=PIIType.MEDICAL in pii_types,
            applicable_regulations=regulations,
        )

PII Anonymization Engine

When PII is detected, anonymize it before logging, sending to third-party APIs, or storing in conversation history:

import hashlib
from typing import Callable

class AnonymizationEngine:
    """Replace PII with anonymized tokens while preserving data utility."""

    def __init__(self, salt: str = "agent-privacy-salt"):
        self.salt = salt
        self._token_map: dict[str, str] = {}

    def anonymize(self, text: str, pii_types: list[PIIType]) -> str:
        anonymized = text

        for pii_type in pii_types:
            pattern = DataClassifier.PII_PATTERNS.get(pii_type)
            if pattern:
                anonymized = re.sub(
                    pattern,
                    lambda m: self._create_token(m.group(), pii_type),
                    anonymized,
                )

        return anonymized

    def _create_token(self, original: str, pii_type: PIIType) -> str:
        """Create a consistent pseudonymized token for a PII value."""
        hash_input = f"{self.salt}:{original}"
        hash_value = hashlib.sha256(hash_input.encode()).hexdigest()[:8]
        token = f"[{pii_type.value.upper()}_{hash_value}]"
        self._token_map[token] = original
        return token

    def deanonymize(self, text: str) -> str:
        """Reverse anonymization when authorized. Use with extreme caution."""
        result = text
        for token, original in self._token_map.items():
            result = result.replace(token, original)
        return result

# Usage example
classifier = DataClassifier()
anonymizer = AnonymizationEngine()

user_message = "My email is john@example.com and my SSN is 123-45-6789"
classification = classifier.classify(user_message)

if classification.requires_anonymization:
    safe_message = anonymizer.anonymize(user_message, classification.pii_types_found)
    # Result: "My email is [EMAIL_a1b2c3d4] and my SSN is [SSN_e5f6g7h8]"

Data Retention Policy Engine

GDPR requires data minimization and purpose limitation. Implement automated retention policies:

from datetime import datetime, timezone, timedelta

@dataclass
class RetentionPolicy:
    data_type: str
    retention_days: int
    purpose: str
    legal_basis: str

class RetentionManager:
    DEFAULT_POLICIES = [
        RetentionPolicy("conversation_logs", 90, "Customer support", "Legitimate interest"),
        RetentionPolicy("pii_data", 30, "Request processing", "Consent"),
        RetentionPolicy("analytics_data", 365, "Service improvement", "Legitimate interest"),
        RetentionPolicy("medical_data", 7, "Appointment scheduling", "Consent + legal obligation"),
    ]

    def __init__(self, db, policies: list[RetentionPolicy] | None = None):
        self.db = db
        self.policies = {p.data_type: p for p in (policies or self.DEFAULT_POLICIES)}

    async def enforce_retention(self) -> dict:
        """Run retention cleanup — schedule this as a daily cron job."""
        results = {}

        for data_type, policy in self.policies.items():
            cutoff = datetime.now(timezone.utc) - timedelta(days=policy.retention_days)

            deleted_count = await self.db.execute(
                f"DELETE FROM {data_type} WHERE created_at < $1 RETURNING id",
                cutoff,
            )

            results[data_type] = {
                "deleted": deleted_count,
                "policy_days": policy.retention_days,
                "cutoff_date": cutoff.isoformat(),
            }

        return results

    async def handle_deletion_request(self, user_id: str) -> dict:
        """GDPR Article 17: Right to erasure."""
        tables = ["conversation_logs", "pii_data", "analytics_data"]
        results = {}

        for table in tables:
            deleted = await self.db.execute(
                f"DELETE FROM {table} WHERE user_id = $1 RETURNING id",
                user_id,
            )
            results[table] = {"deleted": deleted}

        # Log the deletion for compliance audit trail
        await self.db.execute(
            "INSERT INTO deletion_log (user_id, deleted_at, tables_affected) "
            "VALUES ($1, $2, $3)",
            user_id,
            datetime.now(timezone.utc),
            list(results.keys()),
        )

        return results

Consent Management

Track and enforce user consent for data processing:

@dataclass
class ConsentRecord:
    user_id: str
    purpose: str
    granted: bool
    granted_at: Optional[datetime] = None
    revoked_at: Optional[datetime] = None

class ConsentManager:
    def __init__(self, db):
        self.db = db

    async def check_consent(self, user_id: str, purpose: str) -> bool:
        """Check if user has active consent for a specific purpose."""
        record = await self.db.fetchrow(
            "SELECT granted FROM consent_records "
            "WHERE user_id = $1 AND purpose = $2 AND revoked_at IS NULL",
            user_id, purpose,
        )
        return record["granted"] if record else False

    async def grant_consent(self, user_id: str, purpose: str) -> ConsentRecord:
        await self.db.execute(
            "INSERT INTO consent_records (user_id, purpose, granted, granted_at) "
            "VALUES ($1, $2, true, $3) "
            "ON CONFLICT (user_id, purpose) DO UPDATE SET "
            "granted = true, granted_at = $3, revoked_at = NULL",
            user_id, purpose, datetime.now(timezone.utc),
        )
        return ConsentRecord(user_id=user_id, purpose=purpose, granted=True)

    async def revoke_consent(self, user_id: str, purpose: str) -> None:
        await self.db.execute(
            "UPDATE consent_records SET revoked_at = $3, granted = false "
            "WHERE user_id = $1 AND purpose = $2",
            user_id, purpose, datetime.now(timezone.utc),
        )

class PrivacyAwareAgent:
    """Agent wrapper that enforces privacy policies."""

    def __init__(self, agent, classifier, anonymizer, consent_mgr):
        self.agent = agent
        self.classifier = classifier
        self.anonymizer = anonymizer
        self.consent = consent_mgr

    async def process_message(self, user_id: str, message: str) -> str:
        classification = self.classifier.classify(message)

        if classification.requires_consent:
            has_consent = await self.consent.check_consent(user_id, "data_processing")
            if not has_consent:
                return ("Your message contains sensitive information. "
                        "Please grant consent for data processing to continue.")

        # Anonymize before sending to LLM API
        safe_input = message
        if classification.requires_anonymization:
            safe_input = self.anonymizer.anonymize(
                message, classification.pii_types_found
            )

        response = await self._run_agent(safe_input)
        return response

    async def _run_agent(self, message: str) -> str:
        from agents import Runner
        result = await Runner.run(self.agent, message)
        return result.final_output

FAQ

Do I need to anonymize data sent to OpenAI or Anthropic APIs?

If you are processing PII under GDPR, sending it to a third-party API constitutes data transfer to a processor. You need a Data Processing Agreement (DPA) with the provider, and you should anonymize or pseudonymize data whenever the full PII is not required for the task. Both OpenAI and Anthropic offer DPAs and zero-data-retention API options. Use those options, and still anonymize when possible as a defense-in-depth measure.

How do I handle HIPAA compliance for healthcare AI agents?

HIPAA requires a Business Associate Agreement (BAA) with any service that processes Protected Health Information (PHI). Use an LLM provider that offers HIPAA-eligible services and sign a BAA. Encrypt PHI at rest and in transit. Log all access to PHI. Implement minimum necessary access — only retrieve and send the specific PHI fields needed for each task. Never store PHI in conversation logs without encryption and access controls.

What is the difference between anonymization and pseudonymization?

Anonymization permanently removes the ability to identify individuals — the process is irreversible. Pseudonymization replaces identifiers with tokens that can be reversed using a key. GDPR treats pseudonymized data as still personal data (requiring compliance), but anonymized data falls outside GDPR scope. The code in this post implements pseudonymization (reversible with the token map). For true anonymization, destroy the token map after processing and replace PII with generic placeholders instead of hashed tokens.

---

#DataPrivacy #GDPR #HIPAA #PII #Compliance #AgenticAI #LearnAI #AIEngineering