How to Log LLM Calls Under GDPR (Python and TypeScript)

If your app sends user text to an LLM, your logs probably contain personal data: names, emails, support tickets, sometimes health questions. Under GDPR those logs are personal data too. They need a legal basis, a retention period and EU-compatible storage.

If the LLM feature is high-risk under the EU AI Act (hiring, credit scoring, healthcare), you also have to keep logs: Article 12 requires automatic logging, and deployers must keep the logs for at least six months (Article 26(6)).

So you need logs, and you need them not to be a liability. This post covers the pattern I use:

1. Scrub personal data before anything is written or sent
2. Keep logging off the request path, so it never adds latency or raises
3. Make the logs tamper-evident with a hash chain
4. Store them in the EU and delete them on schedule

What a log entry should look like

After scrubbing, an entry still tells you what the system did. It just doesn't tell you who it did it to:

{
  "system_id": "cv-screening",
  "timestamp": "2026-10-27T09:14:02.113Z",
  "model": "gpt-4o-2024-08-06",
  "prompt_scrubbed": "Summarise the application from [NAME_REDACTED_3f9a1c2b7d4e] ([EMAIL_REDACTED_7b2213e0bee9]) for the backend role.",
  "response_scrubbed": "The candidate has 6 years of Python experience...",
  "prompt_tokens": 847,
  "completion_tokens": 203,
  "latency_ms": 1243,
  "finish_reason": "stop",
  "pii_detections": [
    { "category": "name", "placeholder": "[NAME_REDACTED_3f9a1c2b7d4e]" },
    { "category": "email", "placeholder": "[EMAIL_REDACTED_7b2213e0bee9]" }
  ],
  "chained_hash": "a3f9c2847e1d..."
}

Note system_id: log per use case ("cv-screening", "support-triage"), not per vendor. The same model summarising marketing copy and summarising customer records are very different risks.

Step 1: scrub before you log

Logging raw text and redacting later doesn't help. The raw data already existed in your log store, and that's processing. Scrub in memory, before anything is written or sent.

Regex for structured PII

One combined pattern, one pass. Order matters: at the same position, the first alternative wins, so card numbers are tried before phone numbers.

import hashlib
import hmac
import re

PATTERNS = {
    "email": r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}",
    "iban": r"\b[A-Z]{2}\d{2}(?:\s?[A-Z0-9]{4}){2,7}(?:\s?[A-Z0-9]{1,4})?\b",
    "card": r"\b(?:\d{4}[\s-]?){3}\d{4}\b",
    "phone": r"\+?\d{1,3}[\s.-]?\(?\d{2,4}\)?[\s.-]?\d{3,4}[\s.-]?\d{3,4}",
    "ip": r"\b(?:\d{1,3}\.){3}\d{1,3}\b",
}
PII = re.compile("|".join(f"(?P<{name}>{rx})" for name, rx in PATTERNS.items()))


def token(category: str, value: str, n: int, key: bytes | None) -> str:
    if key is None:
        return f"[{category.upper()}_REDACTED_{n}]"
    digest = hmac.new(key, value.encode(), hashlib.sha256).hexdigest()[:12]
    return f"[{category.upper()}_REDACTED_{digest}]"


def scrub(text: str, key: bytes | None = None) -> tuple[str, list[dict]]:
    detections = []

    def replace(match: re.Match) -> str:
        category = match.lastgroup
        n = sum(d["category"] == category for d in detections) + 1
        placeholder = token(category, match.group(), n, key)
        detections.append({"category": category, "placeholder": placeholder})
        return placeholder

    return PII.sub(replace, text), detections

These patterns are a starting point. Tune them for the countries you serve. They over-redact some long numbers, which is the safe direction for logs.

NER for names

Regex can't find names. A local NER model can, and it runs inside your infrastructure:

import spacy

nlp = spacy.load("en_core_web_lg")  # xx_ent_wiki_sm for multilingual text


def scrub_names(text: str, key: bytes | None = None) -> tuple[str, list[dict]]:
    detections, parts, last = [], [], 0
    for ent in nlp(text).ents:
        if ent.label_ != "PERSON":
            continue
        placeholder = token("name", ent.text, len(detections) + 1, key)
        detections.append({"category": "name", "placeholder": placeholder})
        parts += [text[last:ent.start_char], placeholder]
        last = ent.end_char
    parts.append(text[last:])
    return "".join(parts), detections


def scrub_all(text: str, key: bytes | None = None) -> tuple[str, list[dict]]:
    text, regex_hits = scrub(text, key)
    text, name_hits = scrub_names(text, key)
    return text, regex_hits + name_hits

Placeholders or tokens?

Without a key you get positional placeholders ([EMAIL_REDACTED_1]). They're safe, but [EMAIL_REDACTED_1] in one log and in another log aren't necessarily the same person.

With a key you get HMAC tokens. The same email always becomes the same token, so you can still count unique users or follow one case across logs without storing anyone's email. Two rules:

• Use HMAC, not a plain hash. sha256("[e-mail]") can be reversed by hashing a list of known emails. Without the key, an HMAC can't.
• Keep the key in your infrastructure. If it never reaches your log storage, nobody with access to the logs can reverse the tokens.

Why not encrypt instead? Encrypted PII is still personal data under GDPR, so the log store would still hold recoverable personal data. A one-way token gives you the analytics without that.

Step 2: keep logging off the request path

Two rules: logging must never add latency, and it must never raise. A common first attempt is to start a thread per call and scrub on the request path. That adds the NER latency to every request and creates unbounded threads under load.

Instead, the request path does one thing: puts the raw data on an in-memory queue. A single background worker scrubs, sends and retries.

import queue
import threading
import time
from datetime import datetime, timezone

import requests


class LLMLogger:
    def __init__(self, ingest_url: str, api_key: str, system_id: str, pii_key: bytes | None = None):
        self.ingest_url = ingest_url
        self.headers = {"Authorization": f"Bearer {api_key}"}
        self.system_id = system_id
        self.pii_key = pii_key
        self.pending: queue.Queue = queue.Queue(maxsize=10_000)
        threading.Thread(target=self._worker, daemon=True).start()

    def log(self, prompt: str, response: str, **meta) -> None:
        """Runs on the request path: constant time, never blocks, never raises."""
        try:
            meta["timestamp"] = datetime.now(timezone.utc).isoformat()
            self.pending.put_nowait((prompt, response, meta))
        except Exception:
            pass  # a full queue drops the entry rather than slowing the app down

    def _worker(self) -> None:
        while True:
            prompt, response, meta = self.pending.get()
            try:
                prompt_s, prompt_d = scrub_all(prompt, self.pii_key)
                response_s, response_d = scrub_all(response, self.pii_key)
                entry = {
                    "system_id": self.system_id,
                    "prompt_scrubbed": prompt_s,
                    "response_scrubbed": response_s,
                    "pii_detections": prompt_d + response_d,
                    **meta,
                }
                self._send(entry)
            except Exception:
                pass

    def _send(self, entry: dict) -> None:
        for attempt in range(5):
            try:
                r = requests.post(self.ingest_url, json=entry, headers=self.headers, timeout=2)
                if r.status_code < 500:
                    return
            except requests.RequestException:
                pass
            time.sleep(min(2**attempt, 30))

The raw prompt only ever lives in your process's memory. After five failed attempts the entry is dropped. In production, put a disk-backed queue in between so entries survive restarts and longer outages.

Usage. Your OpenAI call doesn't change, and the logger only sees the result:

import os
import time

from openai import OpenAI

client = OpenAI()
logger = LLMLogger(
    ingest_url="https://logs.example.eu/ingest",
    api_key=os.environ["LOG_API_KEY"],
    system_id="cv-screening",
    pii_key=os.environ["PII_HMAC_KEY"].encode(),
)


def call_llm(user_prompt: str) -> str:
    start = time.perf_counter()
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": user_prompt}],
    )
    content = response.choices[0].message.content or ""

    logger.log(
        user_prompt,
        content,
        model=response.model,
        prompt_tokens=response.usage.prompt_tokens,
        completion_tokens=response.usage.completion_tokens,
        latency_ms=round((time.perf_counter() - start) * 1000),
        finish_reason=response.choices[0].finish_reason,
    )
    return content  # the user never waits for logging

Streaming

Two details people miss: by default, streams don't report token usage, and if the client disconnects mid-stream, code after the loop never runs. Ask for usage, and log in finally:

def call_llm_streaming(user_prompt: str):
    start = time.perf_counter()
    chunks, usage, model, finish_reason = [], None, None, None
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": user_prompt}],
        stream=True,
        stream_options={"include_usage": True},  # the last chunk carries token counts
    )
    try:
        for chunk in stream:
            model = chunk.model
            if chunk.usage:
                usage = chunk.usage
            if chunk.choices:
                choice = chunk.choices[0]
                finish_reason = choice.finish_reason or finish_reason
                if choice.delta.content:
                    chunks.append(choice.delta.content)
                    yield choice.delta.content
    finally:
        # Runs even if the client disconnects and the generator is closed
        stream.close()
        logger.log(
            user_prompt,
            "".join(chunks),
            model=model,
            prompt_tokens=usage.prompt_tokens if usage else None,
            completion_tokens=usage.completion_tokens if usage else None,
            latency_ms=round((time.perf_counter() - start) * 1000),
            finish_reason=finish_reason or "interrupted",
        )

TypeScript

The same pattern in Node: log() only pushes onto a queue, and setImmediate drains it after the current tick. scrubAll is a port of the Python scrubber above.

import { scrubAll } from "./scrub"

type Meta = Record<string, unknown>

export class LLMLogger {
  private pending: { prompt: string; response: string; meta: Meta }[] = []
  private draining = false

  constructor(
    private cfg: { ingestUrl: string; apiKey: string; systemId: string; piiKey?: string },
  ) {}

  log(prompt: string, response: string, meta: Meta): void {
    try {
      if (this.pending.length >= 10_000) return // drop rather than grow without bound
      this.pending.push({ prompt, response, meta: { ...meta, timestamp: new Date().toISOString() } })
      if (!this.draining) {
        this.draining = true
        setImmediate(() => void this.drain())
      }
    } catch {
      // never affects the app
    }
  }

  private async drain(): Promise<void> {
    try {
      while (this.pending.length > 0) {
        const { prompt, response, meta } = this.pending.shift()!
        try {
          const [promptScrubbed, promptHits] = scrubAll(prompt, this.cfg.piiKey)
          const [responseScrubbed, responseHits] = scrubAll(response, this.cfg.piiKey)
          await this.send({
            systemId: this.cfg.systemId,
            promptScrubbed,
            responseScrubbed,
            piiDetections: [...promptHits, ...responseHits],
            ...meta,
          })
        } catch {
          // drop this entry, keep draining
        }
      }
    } finally {
      this.draining = false
    }
  }

  private async send(entry: Meta): Promise<void> {
    for (let attempt = 0; attempt < 5; attempt++) {
      try {
        const res = await fetch(this.cfg.ingestUrl, {
          method: "POST",
          headers: { "Content-Type": "application/json", Authorization: `Bearer ${this.cfg.apiKey}` },
          body: JSON.stringify(entry),
          signal: AbortSignal.timeout(2000),
        })
        if (res.status < 500) return
      } catch {
        // network error or timeout: retry
      }
      await new Promise((resolve) => setTimeout(resolve, Math.min(2 ** attempt * 1000, 30_000)))
    }
  }
}

Step 3: make the logs tamper-evident

An audit log is only evidence if you can show it wasn't edited. A hash chain does that: each entry's hash includes the previous entry's hash, so changing any entry breaks every hash after it.

import hashlib
import json


def chained_hash(prev_hash: str, entry: dict) -> str:
    content = json.dumps(entry, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256((prev_hash + content).encode()).hexdigest()


def verify(entries: list[dict]) -> bool:
    prev = "genesis"
    for e in entries:
        body = {k: v for k, v in e.items() if k != "chained_hash"}
        if chained_hash(prev, body) != e["chained_hash"]:
            return False
        prev = e["chained_hash"]
    return True

Two things that matter in practice:

• Compute the chain where logs are stored, not in the app. With several app instances, each one has its own "previous hash" and the chain forks. Compute it on the ingest side, with one writer per chain (for example inside a database transaction).
• Anchor the latest hash somewhere else. Someone who can rewrite every entry can also rewrite the whole chain. Publishing the latest hash periodically (in a report, an email, a separate system) makes that detectable.

Step 4: where the logs live, and for how long

Location. Many EU legal teams block US-based logging and observability tools, even with an EU region. The reason is the US CLOUD Act: a US-incorporated provider can be compelled to hand over data it controls wherever it's stored. Transfers to US companies certified under the EU–US Data Privacy Framework are lawful today, but the framework is under legal challenge, and the CJEU struck down both of its predecessors. The option with the least legal risk is a provider with no US parent and EU-only infrastructure, or self-hosting in the EU.

Retention. GDPR says to keep data no longer than necessary (Article 5(1)(e)). The AI Act sets a floor for high-risk systems. They don't conflict: scrub, keep what the law requires, delete automatically after that.

Purpose	Typical period
High-risk AI under the AI Act (Art. 19, Art. 26(6))	At least 6 months
Sector rules (e.g. financial services)	Per sector, often 5–10 years
Debugging	30–90 days

Document whatever you choose in your Record of Processing Activities, and implement deletion. Don't just write the period down.

Common mistakes

• Logging prompts verbatim. One prompt with a name and a medical question makes the whole log store sensitive.
• Logging on the request path. Logging adds 200 ms, someone removes it to fix performance, and now there are no logs at all.
• Assuming prompts are clean. "We tell users not to paste personal data." They do anyway, and your system prompt may pull in user attributes from your database.
• No retention period. Two years of logs with no legal basis for keeping them past six months.
• No DPA with the LLM provider. Article 28 GDPR requires one. OpenAI, Anthropic and Mistral all offer one. Sign it.

Checklist

• [ ] Legal basis documented for sending user data to the LLM
• [ ] DPA signed with the LLM provider
• [ ] PII scrubbed in memory, before any write or send
• [ ] HMAC key kept inside your infrastructure, never shipped with the logs
• [ ] Logging off the request path: queue, background worker, 2-second timeout, retries
• [ ] Hash chain computed on the storage side, latest hash anchored elsewhere
• [ ] Logs stored in the EU, retention defined and deletion automated
• [ ] DPIA done if the use case is high-risk

---

Disclosure: I'm building Sovergate, which packages this pattern as a Python and TypeScript SDK with tamper-evident storage in Germany and AI Act Article 12 reports. Everything above works without it. This isn't legal advice: check your setup with your DPO.