Add request logging with PII redaction pipeline

Introduction

In production, LLM gateways handle prompts that frequently contain user emails, phone numbers, and other sensitive identifiers — data that must never appear in raw log storage. Without automatic PII redaction baked into the logging path, every debugging session becomes a compliance risk under GDPR, SOC 2, and internal data policies. By the end of this lesson, you will implement a structured logging middleware that intercepts each gateway request and response, applies regex-based PII detection and redaction, and emits clean structured JSON entries to your log aggregation pipeline — giving you full observability without exposing personal data.

Key Terminology

• PIIPattern — A Python dataclass that bundles a human-readable category name, a compiled re.Pattern, and a typed replacement string (e.g., [EMAIL_REDACTED]); each entry in the PII_PATTERNS list defines one redaction rule independently so the core detection loop never needs to change when new categories are added.
• redact_pii — A function that iterates every PIIPattern, performs a global regex substitution on the input text, records the names of matched categories, and returns both the cleaned string and the list of detected PII type names as a tuple.
• GatewayLogEntry — A Pydantic model that captures the full request-response lifecycle — including request_id, tenant_id, prompt_redacted, response_redacted, latency_ms, and pii_types_detected — and serializes to a single-line JSON string via to_json_line() for ingestion by Fluentd or Loki.
• AuditLoggingMiddleware — A Starlette BaseHTTPMiddleware subclass that intercepts every gateway request, applies redact_pii to both the raw prompt body and the downstream response body, assembles a GatewayLogEntry, and emits it via the gateway.audit logger before returning an unmodified response to the caller.
• pii_types_detected — The field in GatewayLogEntry that records the category names (e.g., "email", "ssn") of every PII type found in either the prompt or the response, enabling compliance dashboards to track data exposure trends without storing the raw sensitive values.
• structured log sink — The stdout-based emission path where audit_logger.info(entry.to_json_line()) writes one deterministic JSON object per request; because the schema is fixed by GatewayLogEntry, the cluster log aggregator (Fluentd or Loki) can index every field without additional parsing configuration.

Concepts

The Four-Stage Logging Pipeline

Audit logging for an LLM gateway is not a single write operation — it is an ordered pipeline with four distinct responsibilities: middleware capture, PII detection, redaction, and log emission. Each stage does exactly one thing, and that separation is what makes the system both testable and maintainable.

The middleware intercepts the raw HTTP stream before any application logic runs, so PII is caught at the earliest possible chokepoint. Detection is then a read-only scan — no data is persisted or forwarded yet. Redaction transforms the text in memory, replacing sensitive values with typed placeholders. Only after both passes completes does the cleaned entry reach the log sink. Collapsing any two stages — detecting and writing simultaneously, for instance — makes it impossible to test or extend either without touching both.

Redacting Before Writing: The Compliance Boundary

The central constraint is that PII must be stripped before any write. Once a raw prompt containing an email address reaches a log store, a network socket, or a database row, the compliance exposure exists regardless of who reads it later. GDPR's data-minimization principle and SOC 2's confidentiality criteria both require that you never log data you do not need — and you never need the raw PII value, only the knowledge that some was present.

This is why AuditLoggingMiddleware calls redact_pii on the request body before forwarding via call_next, and again on the response body before emitting the log entry (see Code Walkthrough). The gateway still forwards the original, unredacted content to the upstream model — redaction applies only to the audit trail. The middleware then reconstructs the Response from response_body so the caller receives an unmodified reply.

Extensible Detection Without Logic Changes

The PIIPattern dataclass decouples pattern definition from detection logic. The redact_pii function does not know how many patterns exist or what they match — it only knows how to iterate a list and call .sub. This means adding coverage for a new PII category (say, passport numbers) requires one new PIIPattern entry in PII_PATTERNS and zero changes to the detection loop.

The typed placeholder strings — [EMAIL_REDACTED], [SSN_REDACTED], and so on — preserve the category signal in the log. A log consumer reading "pii_types_detected": ["email", "ssn"] knows what was present and can trigger downstream alerts or metrics without ever seeing raw values. The placeholders are structurally distinct from legitimate prompt content, so they cannot be misread as model output during log analysis.

Schema-Driven Observability

GatewayLogEntry doubles as both a compliance artifact and an observability instrument. Its fields answer different questions for different consumers: request_id enables distributed tracing; tenant_id (read from the X-Tenant-ID header) enables per-customer cost attribution; latency_ms feeds SLO dashboards; pii_types_detected feeds compliance reporting. Because to_json_line() produces a deterministic, schema-fixed JSON object, Fluentd and Loki can index every field without custom parsing rules — the schema is the contract between the gateway and the log aggregation pipeline.

Code Walkthrough

Now that you understand the four-stage pipeline — middleware capture, PII detection, redaction, and log sink — here is how each stage translates into working Python code.

Building the PII detector

The PIIPattern dataclass bundles a human-readable name, a compiled regular expression, and the typed placeholder that replaces any match. The PII_PATTERNS list covers the five most common sensitive data categories found in LLM prompts: email addresses, US phone numbers, SSNs, credit card numbers, and IPv4 addresses. The redact_pii function iterates over every pattern, checks for a match, records the category name, performs a global substitution, and returns both the cleaned text and the list of detected types. Adding a new PII category means appending one PIIPattern entry — the core redaction logic needs no changes.

Code snippetpython
1import re
2from dataclasses import dataclass
3
4@dataclass
5class PIIPattern:
6    name: str
7    pattern: re.Pattern
8    replacement: str
9
10PII_PATTERNS = [
11    PIIPattern("email",       re.compile(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}"), "[EMAIL_REDACTED]"),
12    PIIPattern("phone_us",    re.compile(r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b"),                   "[PHONE_REDACTED]"),
13    PIIPattern("ssn",         re.compile(r"\b\d{3}-\d{2}-\d{4}\b"),                           "[SSN_REDACTED]"),
14    PIIPattern("credit_card", re.compile(r"\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b"),        "[CC_REDACTED]"),
15    PIIPattern("ip_address",  re.compile(r"\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b"),         "[IP_REDACTED]"),
16]
17
18def redact_pii(text: str) -> tuple[str, list[str]]:
19    detected_types: list[str] = []
20    redacted = text
21    for pii in PII_PATTERNS:
22        if pii.pattern.search(redacted):
23            detected_types.append(pii.name)
24            redacted = pii.pattern.sub(pii.replacement, redacted)
25    return redacted, detected_types

Defining the log schema and middleware

GatewayLogEntry is a Pydantic model that captures the full request-response lifecycle: a unique request ID for distributed tracing, a tenant ID for multi-tenant attribution, the redacted prompt and response, token counts for cost tracking, latency in milliseconds, HTTP status, and the list of detected PII categories. Its to_json_line method produces a single-line JSON string that Fluentd or Loki can index without additional parsing configuration.

AuditLoggingMiddleware wires all four stages together. For each incoming request it generates a UUID, decodes the raw body, runs redact_pii on the prompt text, then calls the next handler in the middleware chain. After the downstream response arrives, it reads the response body, runs redact_pii a second time on the response content, assembles a GatewayLogEntry, and emits it via audit_logger.info. That log line travels to stdout where the cluster log aggregator — Fluentd or Loki — picks it up and indexes it as structured JSON.

Code snippetpython
1import time
2import uuid
3import logging
4from datetime import datetime, timezone
5
6from pydantic import BaseModel
7from starlette.middleware.base import BaseHTTPMiddleware
8from starlette.requests import Request
9from starlette.responses import Response
10
11class GatewayLogEntry(BaseModel):
12    request_id: str
13    tenant_id: str
14    model: str
15    prompt_redacted: str
16    response_redacted: str
17    input_tokens: int
18    output_tokens: int
19    latency_ms: float
20    status_code: int
21    pii_types_detected: list[str]
22    timestamp: datetime
23
24    def to_json_line(self) -> str:
25        return self.model_dump_json()
26
27audit_logger = logging.getLogger("gateway.audit")
28
29class AuditLoggingMiddleware(BaseHTTPMiddleware):
30    async def dispatch(self, request: Request, call_next):
31        request_id = str(uuid.uuid4())
32        tenant_id = request.headers.get("X-Tenant-ID", "unknown")
33        start_time = time.monotonic()
34
35        body = await request.body()
36        prompt_text = body.decode("utf-8", errors="replace")
37        prompt_redacted, prompt_pii = redact_pii(prompt_text)
38
39        response: Response = await call_next(request)
40        latency_ms = (time.monotonic() - start_time) * 1000
41
42        response_body = b""
43        async for chunk in response.body_iterator:
44            response_body += chunk
45        response_text = response_body.decode("utf-8", errors="replace")
46        response_redacted, response_pii = redact_pii(response_text)
47
48        entry = GatewayLogEntry(
49            request_id=request_id,
50            tenant_id=tenant_id,
51            model=request.headers.get("X-Model", "unknown"),
52            prompt_redacted=prompt_redacted,
53            response_redacted=response_redacted,
54            input_tokens=0,
55            output_tokens=0,
56            latency_ms=round(latency_ms, 2),
57            status_code=response.status_code,
58            pii_types_detected=list(set(prompt_pii + response_pii)),
59            timestamp=datetime.now(timezone.utc),
60        )
61        audit_logger.info(entry.to_json_line())
62        return Response(
63            content=response_body,
64            status_code=response.status_code,
65            headers=dict(response.headers),
66            media_type=response.media_type,
67        )

Verify by sending a test request whose body contains a real email address and checking that the gateway.audit log output shows [EMAIL_REDACTED] in the prompt_redacted field and that the pii_types_detected array includes "email".

Do's and Don'ts

Now that you have a working middleware that redacts request and response bodies before writing audit entries, here are the operational practices that keep the pipeline correct under load and safe to extend.

Do's

1. ✓Do compile regex patterns inside PIIPattern at import time — the PII_PATTERNS list stores re.compile(...) results so each pattern object is compiled once and reused on every redact_pii call; compiling inside the function on each request adds avoidable latency under gateway-level throughput.
2. ✓Do call redact_pii twice in AuditLoggingMiddleware — once on the request body and once on the buffered response body — LLM responses regularly echo back user-submitted PII, so redacting only the prompt side leaves response_redacted unclean and breaks the SOC 2 / GDPR compliance guarantee.
3. ✓Do emit audit entries through the dedicated gateway.audit logger using to_json_line() — a named logger decoupled from the root application logger lets Fluentd or Loki route structured JSON audit lines to a separate compliance-controlled sink without parsing configuration, and without mingling them with debug output.

Don'ts

1. ✗Don't store prompt_text or response_text in any log field or variable that persists beyond the redaction step — the only values that may enter a GatewayLogEntry are prompt_redacted and response_redacted; persisting the raw decoded strings anywhere invalidates the entire compliance guarantee the middleware exists to enforce.
2. ✗Don't add new PII categories by editing redact_pii's core loop — the function is intentionally open for extension by appending a new PIIPattern to PII_PATTERNS; modifying the loop risks corrupting the detected_types accumulation and the pii_types_detected field that compliance auditors query.
3. ✗Don't return the original response object after consuming its body_iterator in AuditLoggingMiddleware — iterating response.body_iterator to buffer the body for redaction exhausts the stream, so callers receive an empty body unless the middleware reconstructs a new Response with content=response_body.