Compliance and Audit Trail
On this page
Compliance and Audit Trail
For AI applications in production, traceability is essential. A complete audit trail protects you during an inspection, a dispute, or an incident. This lesson shows you how to implement a robust audit system, closing out the course by bringing together everything you’ve built so far—logging, moderation, and data protection—into a single, defensible framework.
Why an Audit Trail?
Four distinct needs converge on the same infrastructure. First, regulatory compliance: the GDPR, the European AI Act, and industry standards all require you to demonstrate what your system does, and a demonstration without logs is worthless. Second, traceability in the strict sense—knowing who did what, when, with what result, even six months later. Debugging benefits daily from this, since understanding unexpected model behavior requires recovering the exact conditions of the call. Finally, accountability: documenting decisions made by the AI is what will allow you, when the time comes, to show that a contested recommendation rested on a controlled process.
Structure of an Audit Record
The dataclass below is as much a contract as it is a data structure. It groups identifiers, application context, the request, the response, the moderation verdict, and technical metadata. Two choices deserve your attention. The user identifier appears only in hashed form, never in plain text. And the prompt is not stored: only its hash and length are. This preserves your ability to prove that two calls concerned the same content without ever creating an archive of what your users wrote.
from dataclasses import dataclass, field, asdict
from datetime import datetime
from typing import Optional
import json
import uuid
import hashlib
@dataclass
class AuditRecord:
"""Audit record for an API call."""
# Identifiers
audit_id: str = field(default_factory=lambda: str(uuid.uuid4()))
timestamp: str = field(
default_factory=lambda: datetime.now().isoformat()
)
# Context
user_id_hash: str = "" # Never the plain ID
session_id: str = ""
application: str = ""
# Request (no sensitive data)
model: str = ""
prompt_hash: str = "" # Hash of the prompt, not the prompt itself
prompt_length: int = 0
# Response
response_id: str = ""
output_length: int = 0
input_tokens: int = 0
output_tokens: int = 0
# Moderation
input_flagged: bool = False
output_flagged: bool = False
# Metadata
duration_ms: int = 0
status: str = "success"
error: Optional[str] = None
def to_json(self) -> str:
return json.dumps(asdict(self), ensure_ascii=False)Complete Audit System
The client that leverages this structure chains input moderation, the API call, output moderation, and writes in every case. The important word is append-only: the JSONL file is opened in append mode, never rewritten, which is the condition for a log to have evidentiary value. Note the path of a flagged entry—the record is saved before the exception is raised, with the status blocked_input. A block therefore leaves a trace just like a success, and the final except block guarantees that no error, whatever its origin, slips under the radar.
import time
from openai import OpenAI
class AuditedClient:
"""OpenAI client with integrated audit trail."""
def __init__(self, application: str, audit_store: str = "audit.jsonl"):
self.client = OpenAI()
self.application = application
self.audit_store = audit_store
def _hash(self, text: str) -> str:
"""Truncated SHA-256 hash for auditing."""
return hashlib.sha256(text.encode()).hexdigest()[:16]
def _save_audit(self, record: AuditRecord):
"""Saves an audit record (append-only)."""
with open(self.audit_store, "a", encoding="utf-8") as f:
f.write(record.to_json() + "\n")
def create(self, user_id: str, session_id: str,
model: str, input: str, **kwargs) -> object:
"""API call with automatic auditing."""
record = AuditRecord(
user_id_hash=self._hash(user_id),
session_id=session_id,
application=self.application,
model=model,
prompt_hash=self._hash(input),
prompt_length=len(input),
)
start = time.monotonic()
try:
# Input moderation
mod = self.client.moderations.create(
model="omni-moderation-latest",
input=input
)
record.input_flagged = mod.results[0].flagged
if record.input_flagged:
record.status = "blocked_input"
self._save_audit(record)
raise ValueError("Inappropriate input content")
# API call
response = self.client.responses.create(
model=model,
input=input,
**kwargs
)
# Output moderation
mod_out = self.client.moderations.create(
model="omni-moderation-latest",
input=response.output_text
)
record.output_flagged = mod_out.results[0].flagged
# Fill metrics
record.response_id = response.id
record.output_length = len(response.output_text)
record.input_tokens = response.usage.input_tokens
record.output_tokens = response.usage.output_tokens
record.duration_ms = round((time.monotonic() - start) * 1000)
record.status = "success"
self._save_audit(record)
return response
except Exception as e:
record.duration_ms = round((time.monotonic() - start) * 1000)
record.status = "error"
record.error = type(e).__name__
self._save_audit(record)
raise
# Usage
audited = AuditedClient(application="support-chatbot")
response = audited.create(
user_id="[email protected]",
session_id="sess-abc123",
model="gpt-5.6-terra",
input="How do I return a product?"
)
print(response.output_text)In this example, the email address passed as user_id never reaches the file: it is hashed as soon as the record is constructed. A JSONL file works fine to get started and for a single instance; as soon as multiple processes write in parallel, direct this stream to a database or a log collector.
Audit Queries
A log you can’t query only serves to occupy disk space. Line-by-line reading of the JSONL format makes exploitation trivial, and the report below summarizes the covered period, the distribution between successes, errors, and blocks, token consumption, and average latency. The count of unique users relies on hashes, which proves incidentally that a pseudonymized log remains perfectly analyzable.
import json
def read_audit(file: str = "audit.jsonl") -> list[dict]:
"""Reads all audit records."""
records = []
try:
with open(file, "r", encoding="utf-8") as f:
for line in f:
if line.strip():
records.append(json.loads(line))
except FileNotFoundError:
pass
return records
def audit_report(file: str = "audit.jsonl") -> dict:
"""Generates an audit report."""
records = read_audit(file)
if not records:
return {"message": "No records"}
total = len(records)
errors = sum(1 for r in records if r["status"] == "error")
blocked = sum(1 for r in records if r["status"] == "blocked_input")
tokens_total = sum(
r.get("input_tokens", 0) + r.get("output_tokens", 0)
for r in records
)
durations = [r["duration_ms"] for r in records if r["duration_ms"] > 0]
return {
"period": {
"start": records[0]["timestamp"],
"end": records[-1]["timestamp"],
},
"total_calls": total,
"successes": total - errors - blocked,
"errors": errors,
"moderation_blocks": blocked,
"tokens_consumed": tokens_total,
"avg_latency_ms": round(sum(durations) / max(len(durations), 1)),
"unique_users": len(set(r["user_id_hash"] for r in records)),
}
print(json.dumps(audit_report(), indent=2, ensure_ascii=False))Retention and Archiving
Keeping data indefinitely contradicts the principle of storage limitation; deleting too soon deprives you of the proof you might need. Rotation solves this dilemma: beyond a duration you set—ninety days here—records leave the active file for a dated archive, while the current log is rewritten with only the recent ones. Schedule this function as a periodic task and align retention_days with the policy validated by your legal counsel, not a value chosen at random.
from datetime import datetime, timedelta
import shutil
import os
def archive_audit(file: str = "audit.jsonl",
retention_days: int = 90):
"""Archives and purges old records."""
records = read_audit(file)
limit = datetime.now() - timedelta(days=retention_days)
recent = []
archives = []
for record in records:
date_record = datetime.fromisoformat(record["timestamp"])
if date_record > limit:
recent.append(record)
else:
archives.append(record)
if archives:
# Archive old ones
archive_name = f"audit_archive_{datetime.now().strftime('%Y%m%d')}.jsonl"
with open(archive_name, "w", encoding="utf-8") as f:
for record in archives:
f.write(json.dumps(record, ensure_ascii=False) + "\n")
# Rewrite main file with recent ones
with open(file, "w", encoding="utf-8") as f:
for record in recent:
f.write(json.dumps(record, ensure_ascii=False) + "\n")
print(f"Archived {len(archives)} records to {archive_name}")
print(f"Kept {len(recent)} recent records")European AI Act: What Changes
The European AI Act (AI Act) imposes obligations based on the risk level of your application. The logic is graduated: the more your system weighs on people’s lives, the stricter the requirements. An FAQ chatbot, a summary, or a translation falls under minimal risk and calls only for basic transparency. As soon as there is conversation or sentiment analysis, you move to limited risk, with the obligation to inform the user they are interacting with an AI. The jump is abrupt for high risk—credit scoring, recruitment, medical diagnosis—where conformity assessment, risk management, training data documentation, complete traceability, human supervision, and demonstration of accuracy are added. Certain uses, such as social scoring or subliminal manipulation, are purely prohibited.
# Assess the risk level of your application
RISK_LEVELS = {
"minimal": {
"examples": ["FAQ chatbot", "text summary", "translation"],
"obligations": ["Basic transparency"]
},
"limited": {
"examples": ["customer service chatbot", "sentiment analysis"],
"obligations": ["Inform the user they are interacting with an AI"]
},
"high": {
"examples": ["credit scoring", "recruitment", "medical diagnosis"],
"obligations": [
"Conformity assessment",
"Risk management system",
"Documented training data",
"Logging and complete traceability",
"Mandatory human supervision",
"Demonstrated accuracy and robustness"
]
},
"unacceptable": {
"examples": ["social scoring", "subliminal manipulation"],
"obligations": ["PROHIBITED"]
}
}Honestly situate your project in this grid before writing the first line of code: an application classified as high risk imposes architectural constraints—exhaustive logging, human control point—that cannot be grafted on after the fact to a system designed without them.
Compliance Checklist
- Append-only audit trail with timestamp
- Hash of personal data in logs (never in plain text)
- Systematic moderation of inputs and outputs
- Up-to-date GDPR record of processing activities
- Defined retention and archiving policy
- AI Act risk level assessment performed
- Inform users they are interacting with an AI
- Human supervision for impactful decisions
Key Takeaways
- An audit trail is mandatory for regulatory compliance
- Log in append-only mode: never modify existing records
- Hash all personal data in audit logs
- The European AI Act classifies applications by risk level
- Implement automatic moderation AND human supervision
- Define a retention policy and archive regularly
Test Your Knowledge
Twenty-two API lessons: validate the mechanisms that hold production together.
1. Why migrate to the Responses API rather than staying on Chat Completions?
Answer: It’s OpenAI’s modern API: integrated conversation state management, native tools, and evolution focused on it—Chat Completions remains supported but legacy.
2. How do you choose between Sol, Terra, and Luna on the API side?
Answer: gpt-5.6-sol for demanding tasks, terra for cost/quality balance, luna for economic volume—and the reasoning level (none→max) is then adjusted per request.
3. What does Structured Output guarantee?
Answer: Output conforming to the provided JSON schema—fields, types, structure—exploitable by a program without defensive parsing or repair.
4. What do you do in the face of a 429 error?
Answer: Respect rate limits: retry with exponential backoff (and jitter), cap attempts, and plan for a circuit breaker—never an immediate retry loop.
5. Name two indispensable protections on the application security side.
Answer: Moderation of incoming/outgoing content via the dedicated API, and defense against prompt injection: isolate instructions from data, validate outputs, limit exposed capabilities.
Production, security, compliance: the last three blocks of the course make the difference between a prototype and a service—keep them as a checklist.