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

Posted on Oct 27 • Originally published at sovergate.com

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:

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..." } Enter fullscreen mode Exit fullscreen mode 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.

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.

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 Enter fullscreen mode Exit fullscreen mode 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.

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 Enter fullscreen mode Exit fullscreen mode 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:

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.

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 )) Enter fullscreen mode Exit fullscreen mode 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 Enter fullscreen mode Exit fullscreen mode 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 ", ) Enter fullscreen mode Exit fullscreen mode 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 ))) } } } Enter fullscreen mode Exit fullscreen mode 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 Enter fullscreen mode Exit fullscreen mode Two things that matter in practice:

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.

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

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.