Audit trail RAG : Tracer les requetes et reponses

Ein vollständiger Audit-Trail ist essenziell für Compliance, Debugging und die kontinuierliche Verbesserung Ihres RAG-Systems. Dieser Leitfaden zeigt, wie man ein umfassendes Logging implementiert.

Pourquoi un audit trail ?

Quoi logger ?

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

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

    # Anfrage
    query: str
    query_timestamp: datetime

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

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

    # Antwort
    response: str
    response_timestamp: datetime

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

    # Fehler
    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 für JSONL-Datei
        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 in die RAG-Pipeline
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

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

            # Protokollierung
            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:
            # Fehler ebenfalls protokollieren
            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
# Für große Volumen: Data Lake verwenden
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):
        # Nach Datum partitionieren
        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"
        )

# Aufbewahrungsrichtlinie
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 beinhaltet einen vollständigen Audit-Trail:

• 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