RAG Audit Trail: Tracing Queries and Responses

A complete audit trail is essential for compliance, debugging and continuous improvement of your RAG system. This guide shows how to implement exhaustive logging.

Why Audit Trail?

What to Log?

DEVELOPERpython
from dataclasses import dataclass
from datetime import datetime
from typing import List, Dict, Optional

@dataclass
class RAGAuditLog:
    # Identifiers
    request_id: str
    session_id: str
    user_id: Optional[str]

    # Request
    query: str
    query_timestamp: datetime

    # Retrieval
    retrieved_doc_ids: List[str]
    retrieval_scores: List[float]
    retrieval_latency_ms: float

    # Generation
    model_used: str
    prompt_tokens: int
    completion_tokens: int
    generation_latency_ms: float

    # Response
    response: str
    response_timestamp: datetime

    # Metadata
    ip_address: Optional[str]
    user_agent: Optional[str]
    feedback_score: Optional[int]

    # Errors
    error: Optional[str]
    error_type: Optional[str]

Implementation

DEVELOPERpython
import json
import logging
from uuid import uuid4
from datetime import datetime

class RAGAuditLogger:
    def __init__(self, log_path: str = "rag_audit.jsonl"):
        self.log_path = log_path
        self.logger = logging.getLogger("rag_audit")
        handler = logging.FileHandler(log_path)
        handler.setFormatter(logging.Formatter("%(message)s"))
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)

    def log_request(self, audit_log: RAGAuditLog):
        log_entry = {
            "request_id": audit_log.request_id,
            "session_id": audit_log.session_id,
            "query": audit_log.query,
            "query_timestamp": audit_log.query_timestamp.isoformat(),
            "retrieved_docs": audit_log.retrieved_doc_ids,
            "model": audit_log.model_used,
            "response": audit_log.response,
            "error": audit_log.error
        }
        self.logger.info(json.dumps(log_entry))

Storage and Retention

DEVELOPERpython
retention_policy = {
    "logs_raw": "90 days",
    "logs_aggregated": "1 year",
    "pii_removed": "5 years"
}

Log Analysis

DEVELOPERpython
import pandas as pd

def analyze_rag_performance(log_file: str) -> Dict:
    logs = [json.loads(line) for line in open(log_file)]
    df = pd.DataFrame(logs)

    return {
        "total_requests": len(df),
        "avg_retrieval_latency_ms": df["retrieval_latency_ms"].mean(),
        "error_rate": df["error"].notna().mean(),
        "top_queries": df["query"].value_counts().head(10).to_dict()
    }

Integration with Ailog

Ailog includes a complete audit trail:

• Real-time logs: Every request traced
• Analytics dashboard: Pattern visualization
• Export: JSONL, CSV for compliance
• Configurable retention: Per your GDPR needs

View logs on Ailog

Related Guides

• GDPR and Chatbots
• Sovereign Cloud
• RAG Monitoring