Audit trail RAG : Tracer les requetes et reponses

Un audit trail complet est essentiel pour la conformite, le debugging et l'amelioration continue de votre systeme RAG. Ce guide montre comment implementer un logging exhaustif.

Pourquoi un audit trail ?

Quoi logger ?

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

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

    # Requete
    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

    # Reponse
    response: str
    response_timestamp: datetime

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

    # Erreurs
    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 fichier JSONL
        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,
            "user_id": audit_log.user_id,
            "query": audit_log.query,
            "query_timestamp": audit_log.query_timestamp.isoformat(),
            "retrieved_docs": audit_log.retrieved_doc_ids,
            "retrieval_scores": audit_log.retrieval_scores,
            "retrieval_latency_ms": audit_log.retrieval_latency_ms,
            "model": audit_log.model_used,
            "prompt_tokens": audit_log.prompt_tokens,
            "completion_tokens": audit_log.completion_tokens,
            "generation_latency_ms": audit_log.generation_latency_ms,
            "response": audit_log.response,
            "response_timestamp": audit_log.response_timestamp.isoformat(),
            "ip_address": audit_log.ip_address,
            "error": audit_log.error
        }

        self.logger.info(json.dumps(log_entry))

# Integration dans le pipeline RAG
class AuditedRAGPipeline:
    def __init__(self, retriever, generator, audit_logger: RAGAuditLogger):
        self.retriever = retriever
        self.generator = generator
        self.audit = audit_logger

    def query(self, query: str, session_id: str, user_id: str = None) -> str:
        request_id = str(uuid4())
        query_time = datetime.now()

        try:
            # Retrieval
            retrieval_start = datetime.now()
            docs = self.retriever.search(query, k=5)
            retrieval_latency = (datetime.now() - retrieval_start).total_seconds() * 1000

            # Generation
            gen_start = datetime.now()
            response = self.generator.generate(query, docs)
            gen_latency = (datetime.now() - gen_start).total_seconds() * 1000

            # Log
            self.audit.log_request(RAGAuditLog(
                request_id=request_id,
                session_id=session_id,
                user_id=user_id,
                query=query,
                query_timestamp=query_time,
                retrieved_doc_ids=[d.id for d in docs],
                retrieval_scores=[d.score for d in docs],
                retrieval_latency_ms=retrieval_latency,
                model_used=self.generator.model_name,
                prompt_tokens=response.usage.prompt_tokens,
                completion_tokens=response.usage.completion_tokens,
                generation_latency_ms=gen_latency,
                response=response.text,
                response_timestamp=datetime.now(),
                ip_address=None,
                user_agent=None,
                feedback_score=None,
                error=None,
                error_type=None
            ))

            return response.text

        except Exception as e:
            # Log l'erreur aussi
            self.audit.log_request(RAGAuditLog(
                request_id=request_id,
                session_id=session_id,
                user_id=user_id,
                query=query,
                query_timestamp=query_time,
                retrieved_doc_ids=[],
                retrieval_scores=[],
                retrieval_latency_ms=0,
                model_used="",
                prompt_tokens=0,
                completion_tokens=0,
                generation_latency_ms=0,
                response="",
                response_timestamp=datetime.now(),
                ip_address=None,
                user_agent=None,
                feedback_score=None,
                error=str(e),
                error_type=type(e).__name__
            ))
            raise

Stockage et retention

DEVELOPERpython
# Pour les gros volumes: utiliser un data lake
import boto3

class S3AuditLogger:
    def __init__(self, bucket: str, prefix: str = "rag-audit/"):
        self.s3 = boto3.client("s3")
        self.bucket = bucket
        self.prefix = prefix

    def log_request(self, audit_log: RAGAuditLog):
        # Partitionner par date
        date_str = audit_log.query_timestamp.strftime("%Y/%m/%d")
        key = f"{self.prefix}{date_str}/{audit_log.request_id}.json"

        self.s3.put_object(
            Bucket=self.bucket,
            Key=key,
            Body=json.dumps(audit_log.__dict__, default=str),
            ContentType="application/json"
        )

# Politique de retention
retention_policy = {
    "logs_raw": "90 days",
    "logs_aggregated": "1 year",
    "pii_removed": "5 years"
}

Analyse des logs

DEVELOPERpython
import pandas as pd

def analyze_rag_performance(log_file: str) -> Dict:
    logs = []
    with open(log_file) as f:
        for line in f:
            logs.append(json.loads(line))

    df = pd.DataFrame(logs)

    return {
        "total_requests": len(df),
        "avg_retrieval_latency_ms": df["retrieval_latency_ms"].mean(),
        "avg_generation_latency_ms": df["generation_latency_ms"].mean(),
        "p95_total_latency_ms": (df["retrieval_latency_ms"] + df["generation_latency_ms"]).quantile(0.95),
        "error_rate": df["error"].notna().mean(),
        "avg_docs_retrieved": df["retrieved_docs"].apply(len).mean(),
        "top_queries": df["query"].value_counts().head(10).to_dict()
    }

Integration avec Ailog

Ailog inclut un audit trail complet :

• Logs temps reel : Chaque requete tracee
• Dashboard analytics : Visualisation des patterns
• Export : JSONL, CSV pour compliance
• Retention configurable : Selon vos besoins RGPD

Voir les logs sur Ailog

Guides connexes

• RGPD et chatbots
• Cloud souverain
• Monitoring RAG