Module 6: Data Privacy & PII Protection

5. Data Minimization

Next →

Overview

In capsules 03 and 04 you learned to detect and redact PII in the data that flows through your AI system. But there's a more fundamental question you should ask yourself before redacting: do you need to send that data to the LLM in the first place?

Data minimization is the principle of sending only the minimum amount of data necessary for the LLM to do its task. If a user asks "what's the status of my order?", the LLM doesn't need their SSN, their medical history, or their full address — it needs the order number and its status. Every additional piece of data you send to the model is a piece of data that can leak.

This principle comes directly from GDPR (Art. 5(1)(c): "data shall be adequate, relevant and limited to what is necessary") and is one of the most effective defenses against LLM02 because it reduces the attack surface: if the data never reaches the model, the model can't reveal it.

In this capsule you build a Data Minimizer that filters, classifies, and reduces the data before sending it to the LLM. This component integrates between the PII Scanner/Redactor and the model call in the PII Protection Layer.

The need-to-know principle

Data minimization applied to AI is equivalent to the "need-to-know" principle in information security: each component of the system only accesses the data it needs to do its job.

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


class AccessLevel(Enum):
    FULL = "full"
    PARTIAL = "partial"
    MINIMAL = "minimal"
    NONE = "none"


@dataclass
class DataAccessPolicy:
    """Defines what data each component can see."""
    component: str
    access_level: AccessLevel
    allowed_fields: list[str]
    denied_fields: list[str]
    rationale: str


access_policies = [
    DataAccessPolicy(
        component="LLM (general query)",
        access_level=AccessLevel.MINIMAL,
        allowed_fields=["query", "product_info", "order_status"],
        denied_fields=["ssn", "credit_card", "full_address", "medical_history"],
        rationale="The LLM only needs the question's context and product data",
    ),
    DataAccessPolicy(
        component="LLM (personalized response)",
        access_level=AccessLevel.PARTIAL,
        allowed_fields=["first_name", "query", "order_history", "preferences"],
        denied_fields=["ssn", "credit_card", "full_address", "email", "phone"],
        rationale="Personalization requires name and preferences, not financial data",
    ),
    DataAccessPolicy(
        component="RAG retrieval",
        access_level=AccessLevel.PARTIAL,
        allowed_fields=["query_embedding", "document_metadata"],
        denied_fields=["raw_documents_with_pii"],
        rationale="The retriever searches by semantic similarity, it doesn't need PII",
    ),
    DataAccessPolicy(
        component="Audit log",
        access_level=AccessLevel.MINIMAL,
        allowed_fields=["request_id", "timestamp", "action", "redaction_count"],
        denied_fields=["original_text", "pii_values", "user_data"],
        rationale="Logs record events, not personal data",
    ),
]

print("Data access policies:\n")
for policy in access_policies:
    print(f"  {policy.component} [{policy.access_level.value}]")
    print(f"    Allowed: {', '.join(policy.allowed_fields[:3])}")
    print(f"    Denied:  {', '.join(policy.denied_fields[:3])}")
    print(f"    Rationale: {policy.rationale[:60]}...")
    print()

Data classification for LLM access

Before minimizing, you need to classify each data field by how necessary it is for the LLM's task.

from enum import Enum
from typing import Optional


class DataNecessity(Enum):
    REQUIRED = "required"
    USEFUL = "useful"
    UNNECESSARY = "unnecessary"
    FORBIDDEN = "forbidden"


@dataclass
class FieldClassification:
    field_name: str
    necessity: DataNecessity
    sensitivity: str
    minimization_action: str


def classify_fields_for_task(
    user_data: dict,
    task_type: str,
) -> list[FieldClassification]:
    """Classifies each field by how necessary it is for the task."""

    task_requirements = {
        "order_status": {
            "required": ["order_id", "query"],
            "useful": ["first_name", "order_date"],
            "unnecessary": ["email", "phone", "address"],
            "forbidden": ["ssn", "credit_card", "password"],
        },
        "product_recommendation": {
            "required": ["query", "preferences"],
            "useful": ["purchase_history", "first_name"],
            "unnecessary": ["full_name", "address", "phone"],
            "forbidden": ["ssn", "credit_card", "date_of_birth"],
        },
        "support_ticket": {
            "required": ["query", "ticket_id", "product_name"],
            "useful": ["first_name", "purchase_date"],
            "unnecessary": ["address", "payment_method"],
            "forbidden": ["ssn", "credit_card", "medical_info"],
        },
    }

    requirements = task_requirements.get(task_type, {})
    classifications = []

    sensitivity_map = {
        "ssn": "critical",
        "credit_card": "critical",
        "password": "critical",
        "email": "high",
        "phone": "high",
        "full_name": "medium",
        "first_name": "low",
        "address": "high",
        "date_of_birth": "medium",
        "order_id": "low",
        "query": "low",
        "preferences": "low",
    }

    action_map = {
        DataNecessity.REQUIRED: "include_as_is",
        DataNecessity.USEFUL: "include_redacted",
        DataNecessity.UNNECESSARY: "exclude",
        DataNecessity.FORBIDDEN: "block",
    }

    for field_name in user_data.keys():
        if field_name in requirements.get("required", []):
            necessity = DataNecessity.REQUIRED
        elif field_name in requirements.get("useful", []):
            necessity = DataNecessity.USEFUL
        elif field_name in requirements.get("forbidden", []):
            necessity = DataNecessity.FORBIDDEN
        else:
            necessity = DataNecessity.UNNECESSARY

        classifications.append(FieldClassification(
            field_name=field_name,
            necessity=necessity,
            sensitivity=sensitivity_map.get(field_name, "unknown"),
            minimization_action=action_map[necessity],
        ))

    return classifications


# --- Demo ---

user_data = {
    "query": "What's the status of my order?",
    "order_id": "ORD-12345",
    "first_name": "María",
    "full_name": "María García López",
    "email": "maria@empresa.com",
    "phone": "555-123-4567",
    "ssn": "123-45-6789",
    "credit_card": "4111-1111-1111-1111",
    "address": "Calle Reforma 123, CDMX",
}

classifications = classify_fields_for_task(user_data, "order_status")

print("Field classification for 'order_status':\n")
for c in classifications:
    icon = {
        DataNecessity.REQUIRED: "✅",
        DataNecessity.USEFUL: "🔶",
        DataNecessity.UNNECESSARY: "❌",
        DataNecessity.FORBIDDEN: "🚫",
    }[c.necessity]
    print(f"  {icon} {c.field_name}: {c.necessity.value} → {c.minimization_action}")

# Expected output:
#   ✅ query: required → include_as_is
#   ✅ order_id: required → include_as_is
#   🔶 first_name: useful → include_redacted
#   ❌ full_name: unnecessary → exclude
#   ❌ email: unnecessary → exclude
#   ❌ phone: unnecessary → exclude
#   🚫 ssn: forbidden → block
#   🚫 credit_card: forbidden → block
#   ❌ address: unnecessary → exclude

Data Minimizer: the complete class

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


class MinimizationAction(Enum):
    INCLUDED = "included"
    REDACTED = "redacted"
    EXCLUDED = "excluded"
    BLOCKED = "blocked"
    GENERALIZED = "generalized"


@dataclass
class MinimizedField:
    field_name: str
    action: MinimizationAction
    original_value: Optional[str] = None
    minimized_value: Optional[str] = None


@dataclass
class MinimizationResult:
    original_field_count: int
    minimized_data: dict
    actions: list[MinimizedField] = field(default_factory=list)
    excluded_count: int = 0
    blocked_count: int = 0
    data_reduction_percent: float = 0.0


class DataMinimizer:
    """Minimizes the data sent to the LLM based on the task's need."""

    FORBIDDEN_FIELDS = {
        "ssn", "social_security", "credit_card", "card_number",
        "password", "secret", "api_key", "token",
        "medical_record", "health_data", "biometric",
    }

    SENSITIVE_FIELDS = {
        "email", "phone", "telephone", "address", "full_address",
        "date_of_birth", "dob", "full_name", "last_name",
        "passport", "license_number", "account_number",
    }

    def __init__(
        self,
        task_allowed_fields: Optional[list[str]] = None,
        include_sensitive: bool = False,
        generalize_dates: bool = True,
        generalize_locations: bool = True,
    ):
        self.task_allowed_fields = task_allowed_fields
        self.include_sensitive = include_sensitive
        self.generalize_dates = generalize_dates
        self.generalize_locations = generalize_locations

    def minimize(
        self,
        data: dict,
        task_type: Optional[str] = None,
    ) -> MinimizationResult:
        """Minimizes a data dict according to the configured policies."""
        minimized = {}
        actions = []
        excluded = 0
        blocked = 0

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

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

            if normalized_key in self.FORBIDDEN_FIELDS:
                actions.append(MinimizedField(
                    field_name=key,
                    action=MinimizationAction.BLOCKED,
                ))
                blocked += 1
                continue

            if self.task_allowed_fields and key not in self.task_allowed_fields:
                actions.append(MinimizedField(
                    field_name=key,
                    action=MinimizationAction.EXCLUDED,
                ))
                excluded += 1
                continue

            if normalized_key in self.SENSITIVE_FIELDS:
                if not self.include_sensitive:
                    actions.append(MinimizedField(
                        field_name=key,
                        action=MinimizationAction.EXCLUDED,
                    ))
                    excluded += 1
                    continue

                generalized = self._generalize(key, value)
                if generalized != value:
                    minimized[key] = generalized
                    actions.append(MinimizedField(
                        field_name=key,
                        action=MinimizationAction.GENERALIZED,
                        minimized_value=str(generalized),
                    ))
                    continue

            minimized[key] = value
            actions.append(MinimizedField(
                field_name=key,
                action=MinimizationAction.INCLUDED,
            ))

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

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

    def _generalize(self, key: str, value: Any) -> Any:
        """Generalizes a value to reduce its specificity."""
        if value is None:
            return None

        str_value = str(value)
        normalized_key = key.lower()

        if normalized_key in ("email",) and "@" in str_value:
            domain = str_value.split("@")[1]
            return f"***@{domain}"

        if normalized_key in ("phone", "telephone"):
            if len(str_value) >= 4:
                return "***" + str_value[-4:]
            return "***"

        if normalized_key in ("address", "full_address"):
            parts = str_value.split(",")
            if len(parts) >= 2:
                return parts[-1].strip()
            return "[Location]"

        if normalized_key in ("date_of_birth", "dob"):
            import re
            year_match = re.search(r"(19|20)\d{2}", str_value)
            if year_match:
                year = int(year_match.group())
                decade = (year // 10) * 10
                return f"{decade}s"
            return "[Date]"

        if normalized_key in ("full_name",):
            parts = str_value.split()
            if parts:
                return parts[0]
            return "[Name]"

        return value


# --- Demo ---

user_data = {
    "query": "What's the status of my order?",
    "order_id": "ORD-12345",
    "first_name": "María",
    "full_name": "María García López",
    "email": "maria@empresa.com",
    "phone": "555-123-4567",
    "ssn": "123-45-6789",
    "credit_card": "4111-1111-1111-1111",
    "address": "Calle Reforma 123, CDMX, México",
    "date_of_birth": "03/15/1990",
    "order_status": "shipped",
}

minimizer = DataMinimizer(
    task_allowed_fields=[
        "query", "order_id", "first_name", "order_status",
    ],
)

result = minimizer.minimize(user_data)

print("Data Minimization Results:\n")
print(f"  Original fields: {result.original_field_count}")
print(f"  Minimized fields: {len(result.minimized_data)}")
print(f"  Excluded: {result.excluded_count}")
print(f"  Blocked: {result.blocked_count}")
print(f"  Data reduction: {result.data_reduction_percent}%")
print(f"\n  Minimized data:")
for key, value in result.minimized_data.items():
    print(f"    {key}: {value}")
print(f"\n  Actions:")
for action in result.actions:
    icon = {
        MinimizationAction.INCLUDED: "✅",
        MinimizationAction.REDACTED: "🔶",
        MinimizationAction.EXCLUDED: "❌",
        MinimizationAction.BLOCKED: "🚫",
        MinimizationAction.GENERALIZED: "📐",
    }[action.action]
    print(f"    {icon} {action.field_name}: {action.action.value}")

# Expected output:
#   Original fields: 11
#   Minimized fields: 4
#   Excluded: 5
#   Blocked: 2
#   Data reduction: ~70%
#
#   Minimized data:
#     query: What's the status of my order?
#     order_id: ORD-12345
#     first_name: María
#     order_status: shipped

Smart context truncation

When the context for the LLM is a long text (document, chat history, RAG results), you need to truncate smartly to maximize relevance within the token budget.

from dataclasses import dataclass


@dataclass
class TruncationResult:
    original_length: int
    truncated_length: int
    estimated_tokens: int
    strategy_used: str
    text: str


class ContextTruncator:
    """Truncates context smartly for the LLM."""

    def __init__(
        self,
        max_tokens: int = 2000,
        chars_per_token: int = 4,
    ):
        self.max_tokens = max_tokens
        self.max_chars = max_tokens * chars_per_token
        self.chars_per_token = chars_per_token

    def truncate(
        self,
        text: str,
        strategy: str = "smart",
    ) -> TruncationResult:
        """Truncates text according to the strategy."""
        if len(text) <= self.max_chars:
            return TruncationResult(
                original_length=len(text),
                truncated_length=len(text),
                estimated_tokens=len(text) // self.chars_per_token,
                strategy_used="none",
                text=text,
            )

        if strategy == "start":
            truncated = text[:self.max_chars]
        elif strategy == "end":
            truncated = text[-self.max_chars:]
        elif strategy == "smart":
            truncated = self._smart_truncate(text)
        elif strategy == "middle_out":
            truncated = self._middle_out(text)
        else:
            truncated = text[:self.max_chars]

        return TruncationResult(
            original_length=len(text),
            truncated_length=len(truncated),
            estimated_tokens=len(truncated) // self.chars_per_token,
            strategy_used=strategy,
            text=truncated,
        )

    def _smart_truncate(self, text: str) -> str:
        """Keeps the start and end, truncates the middle."""
        keep_start = self.max_chars // 3
        keep_end = self.max_chars // 3
        start = text[:keep_start]
        end = text[-keep_end:]
        return f"{start}\n\n[... {len(text) - keep_start - keep_end} chars omitted ...]\n\n{end}"

    def _middle_out(self, text: str) -> str:
        """Prioritizes the middle content (useful for documents with headers/footers)."""
        total_to_remove = len(text) - self.max_chars
        remove_start = total_to_remove // 2
        remove_end = total_to_remove - remove_start
        return text[remove_start:len(text) - remove_end]


# --- Demo ---

truncator = ContextTruncator(max_tokens=100)

long_text = "Important intro. " + "Middle content. " * 200 + "Critical conclusion."

for strategy in ["start", "end", "smart", "middle_out"]:
    result = truncator.truncate(long_text, strategy=strategy)
    print(f"Strategy: {strategy}")
    print(f"  Original: {result.original_length} chars")
    print(f"  Truncated: {result.truncated_length} chars")
    print(f"  Tokens: ~{result.estimated_tokens}")
    print(f"  Preview: \"{result.text[:50]}...\"")
    print()

Prompt engineering for privacy

You can design your prompts so the LLM generates useful responses without needing sensitive data.

privacy_prompt_patterns = {
    "reference_by_id": {
        "bad": (
            "User María García (maria@test.com, SSN: 123-45-6789) "
            "asks about their order #12345"
        ),
        "good": (
            "User [ID: USR-789] asks about order #12345. "
            "Order status: shipped, estimated delivery: tomorrow."
        ),
        "principle": "Reference by ID, not by personal data",
    },
    "provide_answer_not_data": {
        "bad": (
            "Customer data:\n"
            "Name: Juan López\nEmail: juan@test.com\n"
            "History: 15 purchases, total $5,432\n"
            "Question: Am I eligible for the VIP discount?"
        ),
        "good": (
            "A customer with 15 purchases and a historical total >$5,000 "
            "asks whether they're eligible for the VIP discount. "
            "Policy: VIP requires 10+ purchases and >$3,000."
        ),
        "principle": "Provide the answer, not the data to compute it",
    },
    "aggregate_not_individual": {
        "bad": (
            "Department employees:\n"
            "- Ana Ruiz, $85,000\n- Pedro Soto, $92,000\n"
            "- Carmen Vega, $78,000\n"
            "Question: What's the average salary?"
        ),
        "good": (
            "Department with 3 employees. "
            "Salary range: $78,000-$92,000. Average: $85,000. "
            "Question: summarize the salary information."
        ),
        "principle": "Send aggregates, not individual data",
    },
    "separate_context_from_query": {
        "bad": (
            "My name is Carlos Méndez, I'm 42, I live in "
            "Guadalajara, my email is carlos@test.com. "
            "What restaurants do you recommend?"
        ),
        "good": (
            "A user in Guadalajara is looking for restaurant "
            "recommendations. Preferences: Mexican cuisine, mid-price "
            "range."
        ),
        "principle": "Extract only what's relevant from the user's query",
    },
}

print("Prompt engineering patterns for privacy:\n")
for pattern, info in privacy_prompt_patterns.items():
    print(f"  {pattern.upper().replace('_', ' ')}")
    print(f"    Principle: {info['principle']}")
    print(f"    ❌ Bad: \"{info['bad'][:60]}...\"")
    print(f"    ✅ Good: \"{info['good'][:60]}...\"")
    print()

Selective field inclusion per endpoint

from typing import Optional


class EndpointMinimizer:
    """Minimizes data based on the API endpoint."""

    ENDPOINT_FIELDS = {
        "/chat": {
            "include": ["query", "session_id", "language"],
            "context_fields": ["first_name"],
            "max_context_chars": 500,
        },
        "/search": {
            "include": ["query"],
            "context_fields": [],
            "max_context_chars": 200,
        },
        "/support": {
            "include": ["query", "ticket_id", "product_name", "error_code"],
            "context_fields": ["first_name", "purchase_date"],
            "max_context_chars": 1000,
        },
        "/recommendation": {
            "include": ["query", "preferences", "category"],
            "context_fields": ["purchase_history_summary"],
            "max_context_chars": 800,
        },
    }

    def minimize_for_endpoint(
        self,
        data: dict,
        endpoint: str,
    ) -> dict:
        """Filters data according to the endpoint's configuration."""
        config = self.ENDPOINT_FIELDS.get(endpoint)
        if not config:
            return {"query": data.get("query", "")}

        result = {}
        for field in config["include"]:
            if field in data:
                result[field] = data[field]

        for field in config["context_fields"]:
            if field in data:
                value = str(data[field])
                max_chars = config["max_context_chars"]
                result[field] = value[:max_chars]

        return result


# --- Demo ---

endpoint_minimizer = EndpointMinimizer()

full_data = {
    "query": "Do you have laptops on sale?",
    "session_id": "sess-123",
    "first_name": "María",
    "full_name": "María García López",
    "email": "maria@test.com",
    "phone": "555-1234",
    "preferences": "electronics, budget",
    "purchase_history_summary": "5 purchases in the last 6 months",
    "language": "es",
}

for endpoint in ["/chat", "/search", "/recommendation"]:
    minimized = endpoint_minimizer.minimize_for_endpoint(full_data, endpoint)
    print(f"  {endpoint}: {list(minimized.keys())}")

# Expected output:
#   /chat: ['query', 'session_id', 'language', 'first_name']
#   /search: ['query']
#   /recommendation: ['query', 'preferences', 'purchase_history_summary']

RAG context minimization

@dataclass
class RAGContextMinimizer:
    """Minimizes the RAG context before sending it to the LLM."""
    max_chunks: int = 3
    max_chars_per_chunk: int = 500
    max_total_chars: int = 2000
    remove_metadata: bool = True

    def minimize_context(
        self,
        chunks: list[dict],
        query: str,
    ) -> dict:
        """Reduces the RAG context to the minimum necessary."""
        selected = chunks[:self.max_chunks]

        minimized_chunks = []
        total_chars = 0

        for chunk in selected:
            text = chunk.get("content", chunk.get("text", ""))

            if len(text) > self.max_chars_per_chunk:
                text = text[:self.max_chars_per_chunk] + "..."

            if total_chars + len(text) > self.max_total_chars:
                remaining = self.max_total_chars - total_chars
                if remaining > 100:
                    text = text[:remaining] + "..."
                else:
                    break

            minimized_chunk = {"content": text}
            if not self.remove_metadata:
                minimized_chunk["source"] = chunk.get("source", "unknown")

            minimized_chunks.append(minimized_chunk)
            total_chars += len(text)

        return {
            "chunks": minimized_chunks,
            "original_count": len(chunks),
            "selected_count": len(minimized_chunks),
            "total_chars": total_chars,
            "estimated_tokens": total_chars // 4,
        }


# --- Demo ---

chunks = [
    {"content": "Product A is a laptop with 16GB RAM. " * 20, "source": "catalog.pdf", "score": 0.95},
    {"content": "Product B is a tablet with 8GB RAM. " * 15, "source": "catalog.pdf", "score": 0.87},
    {"content": "Warranty covers 2 years. " * 10, "source": "warranty.pdf", "score": 0.82},
    {"content": "Return policy is 30 days. " * 10, "source": "returns.pdf", "score": 0.75},
    {"content": "Shipping takes 3-5 days. " * 10, "source": "shipping.pdf", "score": 0.60},
]

minimizer = RAGContextMinimizer(max_chunks=3, max_total_chars=1000)
result = minimizer.minimize_context(chunks, "laptop specs")

print(f"RAG Context Minimization:")
print(f"  Original chunks: {result['original_count']}")
print(f"  Selected chunks: {result['selected_count']}")
print(f"  Total chars: {result['total_chars']}")
print(f"  Estimated tokens: {result['estimated_tokens']}")

Connection to the project

The Data Minimizer integrates into the PII Protection Layer between the scanner/redactor and the LLM call:

User Input + Context
  │
  ▼
PIIScanner → detects PII
  │
  ▼
PreLLMRedactor → redacts PII
  │
  ▼
DataMinimizer → reduces to the minimum        ← THIS COMPONENT
  │
  ▼
LLM Processing (with minimal data)
  │
  ▼
PostLLMRedactor → filters PII from the output

Troubleshooting

Problem 1: "Minimization reduces the context so much that the LLM gives generic responses"

If the LLM doesn't have enough context, it generates vague responses like "I don't have enough information."

Solution: Calibrate the task_allowed_fields by task type. Start with more fields and reduce gradually until you find the minimum that keeps the quality. Measure response quality with and without minimization.

Problem 2: "I don't know which fields are necessary for each task"

Solution: Start with a permissive policy (include everything except FORBIDDEN), monitor which fields the LLM actually uses in its responses, and gradually exclude the ones that never appear in the output.

Problem 3: "Date generalization loses too much information"

Converting "03/15/1990" to "1990s" can be too aggressive for some tasks.

Solution: Adjust the generalization granularity by task. For age verification, "adult" may be enough. For cohort analysis, the exact year may be necessary.

Exercises

Exercise 1: Minimizer with reduction metrics

Extend the DataMinimizer to compute detailed data-reduction metrics.

See solution 
def calculate_minimization_metrics(
    original: dict, minimized: dict,
) -> dict:
    original_chars = sum(len(str(v)) for v in original.values())
    minimized_chars = sum(len(str(v)) for v in minimized.values())

    return {
        "original_fields": len(original),
        "minimized_fields": len(minimized),
        "field_reduction": f"{(1 - len(minimized)/len(original))*100:.0f}%",
        "original_chars": original_chars,
        "minimized_chars": minimized_chars,
        "char_reduction": f"{(1 - minimized_chars/original_chars)*100:.0f}%",
        "original_tokens_est": original_chars // 4,
        "minimized_tokens_est": minimized_chars // 4,
        "token_savings": (original_chars - minimized_chars) // 4,
    }


data = {
    "query": "order status", "order_id": "ORD-123",
    "name": "John Smith", "email": "john@test.com",
    "ssn": "123-45-6789", "address": "123 Main St, NYC",
}
minimized = {"query": "order status", "order_id": "ORD-123"}

metrics = calculate_minimization_metrics(data, minimized)
for k, v in metrics.items():
    print(f"  {k}: {v}")

Exercise 2: Configurable policy engine

Create a policy engine that reads the minimization rules from a configuration dictionary.

See solution 
MINIMIZATION_POLICIES = {
    "strict": {
        "forbidden": ["ssn", "credit_card", "password", "medical"],
        "exclude": ["email", "phone", "address", "full_name", "dob"],
        "include_all_else": False,
        "allowed": ["query", "session_id"],
    },
    "moderate": {
        "forbidden": ["ssn", "credit_card", "password"],
        "exclude": ["address", "medical"],
        "include_all_else": True,
        "allowed": [],
    },
}


def apply_policy(data: dict, policy_name: str) -> dict:
    policy = MINIMIZATION_POLICIES.get(policy_name)
    if not policy:
        raise ValueError(f"Unknown policy: {policy_name}")

    result = {}
    for key, value in data.items():
        key_lower = key.lower()
        if key_lower in policy["forbidden"]:
            continue
        if key_lower in policy["exclude"]:
            continue
        if policy["include_all_else"] or key_lower in policy["allowed"]:
            result[key] = value
    return result


data = {
    "query": "help", "email": "test@test.com",
    "ssn": "123-45-6789", "name": "John",
}
print("Strict:", apply_policy(data, "strict"))
print("Moderate:", apply_policy(data, "moderate"))

# Output:
# Strict: {'query': 'help'}
# Moderate: {'query': 'help', 'email': 'test@test.com', 'name': 'John'}

Exercise 3: Context builder that assembles minimized prompts

Create a function that builds a prompt for the LLM using only the minimized data.

See solution 
def build_minimized_prompt(
    minimized_data: dict,
    system_template: str = "You are an assistant. Respond based only on the provided context.",
) -> list[dict]:
    context_parts = []
    query = minimized_data.pop("query", "")

    for key, value in minimized_data.items():
        context_parts.append(f"{key}: {value}")

    context_str = "\n".join(context_parts) if context_parts else "No additional context."

    return [
        {"role": "system", "content": system_template},
        {"role": "user", "content": f"Context:\n{context_str}\n\nQuestion: {query}"},
    ]


minimized = {
    "query": "What's the status of my order?",
    "order_id": "ORD-12345",
    "order_status": "shipped",
    "first_name": "María",
}

messages = build_minimized_prompt(minimized)
for msg in messages:
    print(f"  [{msg['role']}]: {msg['content'][:80]}...")

Exercise 4: Detector of unnecessary data in existing prompts

Create a function that analyzes an existing prompt and identifies data that could be removed.

See solution 
from presidio_analyzer import AnalyzerEngine


def audit_prompt_data(prompt: str) -> dict:
    """Identifies potentially unnecessary data in a prompt."""
    analyzer = AnalyzerEngine()
    results = analyzer.analyze(text=prompt, language="en", score_threshold=0.4)

    unnecessary = []
    for r in results:
        entity_text = prompt[r.start:r.end]
        unnecessary.append({
            "type": r.entity_type,
            "text": entity_text,
            "score": r.score,
            "recommendation": (
                "REMOVE" if r.entity_type in ("US_SSN", "CREDIT_CARD")
                else "CONSIDER_REMOVING" if r.entity_type in ("EMAIL_ADDRESS", "PHONE_NUMBER")
                else "REVIEW"
            ),
        })

    return {
        "prompt_length": len(prompt),
        "pii_found": len(unnecessary),
        "items": unnecessary,
        "estimated_reduction": f"{sum(len(i['text']) for i in unnecessary)} chars removable",
    }


prompt = (
    "User John Smith (john@test.com, SSN: 123-45-6789) "
    "asks about product pricing. His account is ACC-123."
)
audit = audit_prompt_data(prompt)
print(f"PII in prompt: {audit['pii_found']}")
for item in audit['items']:
    print(f"  [{item['recommendation']}] {item['type']}: \"{item['text']}\"")

Summary

  • 🔑 Data minimization is sending only the strictly necessary data to the LLM — if the data doesn't reach the model, it can't be leaked
  • 🔑 The principle comes from GDPR Art. 5(1)(c) and is one of the most effective defenses against LLM02 because it reduces the attack surface
  • 🔑 Classifying fields by necessity (REQUIRED, USEFUL, UNNECESSARY, FORBIDDEN) lets you automate which data is sent to the model per task
  • 🔑 The DataMinimizer filters, excludes, and generalizes data automatically according to policies configurable per endpoint and task
  • 🔑 Smart truncation of context maximizes relevance within the token budget — the "smart" strategy preserves the start and end
  • 🔑 Prompt engineering for privacy lets you get useful responses without sending personal data: reference by ID, aggregates instead of individuals
  • 🔑 Per-endpoint minimization allows different policies: /search only needs the query, /support needs the ticket and product
  • 🔑 The trade-off is privacy vs quality: less data = safer but potentially less personalized responses

Additional resources

  1. GDPR Art. 5 — Data Minimisation — Official text of the data minimization principle
  2. NIST Privacy Framework — Data Minimization — NIST's framework, which includes minimization practices
  3. Data Minimization in Machine Learning (ACM) — Paper on data minimization in ML
  4. OpenAI Token Counter (tiktoken) — OpenAI's official library for counting tokens
  5. OWASP Data Minimization — The minimization principle in the context of web security
  6. Microsoft Privacy Principles — Privacy principles including minimization
  7. ICO Guide to Data Minimisation — The UK regulator's guide on minimization

Created: March 2026 Version: 1.0

← Previous4. PII Redaction: Before and After the LLMNext →6. Retention Policies and Encryption