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 retenti
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:
- Scrub personal data before anything is written or sent
- Keep logging off the request path, so it never adds latency or raises
- Make the logs tamper-evident with a hash chain
- 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("[email protected]")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.
Originally published by Dev.to AI. Aggregated on AIWithGhost for educational purposes β full credit and traffic to the original publisher.