RAG et automatisation Python : concevoir une chaîne traçable

Ajouter des documents au contexte d’un modèle ne transforme pas automatiquement sa réponse en vérité. Un système RAG peut retrouver le mauvais passage, ignorer une version récente, mélanger deux périmètres ou générer une conclusion absente des sources. La valeur de l’architecture tient donc moins au mot « RAG » qu’à la capacité de tester chaque étape et de refuser une réponse lorsque les preuves sont insuffisantes.

Réponse en bref

Séparez la chaîne en huit contrats : corpus, ingestion, segmentation, index, recherche, paquet de contexte, génération et validation. Versionnez les sources, journalisez les identifiants plutôt que les contenus sensibles, évaluez la recherche indépendamment de la réponse et prévoyez une abstention. Python peut automatiser ces étapes avec un environnement isolé, des logs structurés et une base SQLite pour un prototype. Le RAG fournit un contexte récupéré ; il ne garantit ni vérité, ni exhaustivité, ni conformité de la réponse.

Ce que signifie RAG

Le papier de Lewis et ses coauteurs présenté à NeurIPS 2020 décrit des modèles qui combinent une mémoire paramétrique et une mémoire non paramétrique consultée par un mécanisme de recherche. Cette famille d’approches permet d’apporter des passages externes au processus de génération et d’envisager une mise à jour des connaissances sans réentraîner tout le modèle.

Dans une application, le terme recouvre souvent une chaîne plus large : collecte de documents, conversion, découpage, représentation, recherche, assemblage du contexte, appel du modèle, citations et contrôles. Deux systèmes appelés « RAG » peuvent donc avoir des niveaux de qualité et de traçabilité très différents.

La documentation OpenAI sur la recherche décrit les vector stores et les opérations de recherche sémantique. L’outil File Search peut effectuer la recherche dans des fichiers attachés à des vector stores pour une réponse. Ces capacités gèrent une partie de l’infrastructure ; elles ne décident pas quelles sources l’organisation peut utiliser, quelle version fait foi, ni comment valider une affirmation métier.

Commencer par la décision, pas par la base vectorielle

Un RAG est pertinent lorsque la réponse dépend d’un corpus identifiable et changeant, et lorsque les utilisateurs ont besoin de retrouver les passages qui fondent la réponse. Il n’est pas nécessaire pour toute automatisation.

Besoin	Option initiale	Pourquoi
retrouver un document exact par identifiant	index classique ou base relationnelle	déterministe et simple
appliquer une règle stable	code et tests	la règle ne doit pas être réinterprétée
rechercher une formulation proche dans un corpus	recherche lexicale ou sémantique	besoin de rappel documentaire
synthétiser plusieurs passages autorisés	RAG avec validation	génération utile, mais preuve nécessaire
prendre une décision réglementée	workflow métier avec autorité humaine	la génération ne doit pas porter seule la décision

Si une requête SQL ou une règle explicite suffit, ajoutez-la avant le modèle. Une architecture plus simple est souvent plus facile à tester, à sécuriser et à expliquer.

Le contrat de preuve en huit étapes

1. Corpus

Pour chaque source, conservez un propriétaire, une origine, une date, une version, un périmètre d’autorisation et une règle de retrait. Un fichier accessible n’est pas nécessairement autorisé à l’indexation. Un document ancien n’est pas nécessairement faux, mais sa validité doit être explicite.

Sortie attendue : registre de sources avec identifiant stable et empreinte.

2. Ingestion

L’ingestion vérifie format, taille, encodage, type réel, duplication et métadonnées. Elle doit pouvoir échouer sans laisser un document partiellement actif. Conservez l’empreinte cryptographique du fichier ou du texte normalisé pour détecter une version réellement différente.

Sortie attendue : événement d’ingestion avec état accepté, rejeté ou à revoir.

3. Segmentation

Le découpage ne doit pas supprimer le titre, la section, la date ou l’identité de la source. Une taille fixe peut couper une exception de la règle qu’elle nuance. Testez plusieurs stratégies sur des questions réelles et gardez la relation entre fragment et document.

Sortie attendue : fragments adressables, chacun relié à une source et une version.

4. Index

L’index peut être lexical, vectoriel ou hybride. Documentez le modèle de représentation, ses paramètres, la date d’indexation et les filtres de métadonnées. Un changement de modèle ou de découpage crée une nouvelle condition expérimentale.

Sortie attendue : version d’index reproductible.

5. Recherche

La recherche reçoit une question, applique les filtres et retourne des fragments avec scores et identifiants. Évaluez d’abord si les passages utiles sont présents parmi les résultats. Une réponse finale fluide ne doit pas masquer un mauvais rappel.

Sortie attendue : liste ordonnée et explicable de fragments candidats.

6. Paquet de contexte

Le paquet de contexte fixe les passages envoyés, leur ordre, leur budget et leurs références. Il marque les zones de confiance et les conflits. Ne concaténez pas silencieusement des versions contradictoires.

Sortie attendue : objet immuable contenant question, sources et règle d’abstention.

7. Génération

La consigne demande une réponse limitée aux preuves fournies, avec références et abstention en cas d’insuffisance. Cette consigne réduit un risque ; elle ne garantit pas que le modèle la respectera. Le résultat reste une proposition à contrôler.

Sortie attendue : réponse, références déclarées, modèle et paramètres nécessaires à l’audit.

8. Validation

Vérifiez que les références existent, que chaque affirmation importante est soutenue et qu’aucune donnée interdite n’est sortie. Selon l’enjeu, le workflow exige une validation humaine ou refuse automatiquement la diffusion.

Sortie attendue : état validé, abstention, rejet ou revue humaine.

Séparer les tests de recherche et de génération

Couche	Question de test	Mesure possible
ingestion	le bon contenu et la bonne version sont-ils actifs ?	taux d’erreur, doublons, retraits appliqués
recherche	le passage de référence est-il retrouvé ?	rappel à k, rang du passage, taux sans résultat
contexte	les passages envoyés sont-ils pertinents et autorisés ?	précision du paquet, conflits signalés
réponse	les affirmations suivent-elles les passages ?	fidélité évaluée avec grille
citations	les références pointent-elles vers le bon extrait ?	précision des citations
abstention	le système refuse-t-il quand la preuve manque ?	vrais et faux refus sur cas négatifs
exploitation	le service reste-t-il observable et réversible ?	latence, erreurs, coût, version et rollback

Créez un jeu d’évaluation avant de régler les paramètres. Il doit contenir des questions avec réponse, des questions ambiguës, des documents obsolètes, des conflits et des questions sans réponse. Sans cas négatifs, le système apprend seulement à toujours produire quelque chose.

Un prototype local centré sur le paquet de preuve

Le code suivant n’appelle aucun modèle et n’effectue pas de recherche vectorielle. Il construit un socle volontairement simple : documents versionnés dans SQLite, recherche lexicale transparente, seuil d’abstention, paquet de contexte et journal sans question brute. Ce baseline permet de tester le contrat avant d’ajouter un fournisseur ou un index sémantique.

Préparer l’environnement

La documentation Python décrit venv comme un moyen de créer des environnements légers et isolés. Un environnement est recréable et ne doit pas être versionné.

python3 -m venv .venv
.venv/bin/python -m pip install --upgrade pip

Le prototype utilise uniquement la bibliothèque standard ; aucune installation supplémentaire n’est requise pour l’exécuter.

Code

from __future__ import annotations

import hashlib
import json
import logging
import math
import re
import sqlite3
from dataclasses import asdict, dataclass
from typing import Iterable


logging.basicConfig(level=logging.INFO, format="%(message)s")
LOGGER = logging.getLogger("evidence-retrieval")

SCHEMA = """
CREATE TABLE IF NOT EXISTS documents (
    document_id TEXT PRIMARY KEY,
    title TEXT NOT NULL,
    source_url TEXT NOT NULL,
    version TEXT NOT NULL,
    content TEXT NOT NULL,
    content_sha256 TEXT NOT NULL,
    active INTEGER NOT NULL CHECK (active IN (0, 1))
);
"""


@dataclass(frozen=True)
class Evidence:
    document_id: str
    title: str
    source_url: str
    version: str
    content_sha256: str
    score: float
    excerpt: str


def terms(text: str) -> set[str]:
    """Baseline lisible : mots normalisés de trois caractères ou plus."""
    return set(re.findall(r"[a-z0-9à-ÿ]{3,}", text.casefold()))


def fingerprint(content: str) -> str:
    normalized = " ".join(content.split())
    return hashlib.sha256(normalized.encode("utf-8")).hexdigest()


def connect() -> sqlite3.Connection:
    database = sqlite3.connect(":memory:")
    database.row_factory = sqlite3.Row
    database.executescript(SCHEMA)
    return database


def upsert_documents(
    database: sqlite3.Connection,
    documents: Iterable[dict[str, str]],
) -> None:
    with database:
        for item in documents:
            database.execute(
                """
                INSERT INTO documents (
                    document_id, title, source_url, version,
                    content, content_sha256, active
                ) VALUES (?, ?, ?, ?, ?, ?, 1)
                ON CONFLICT(document_id) DO UPDATE SET
                    title = excluded.title,
                    source_url = excluded.source_url,
                    version = excluded.version,
                    content = excluded.content,
                    content_sha256 = excluded.content_sha256,
                    active = 1
                """,
                (
                    item["document_id"],
                    item["title"],
                    item["source_url"],
                    item["version"],
                    item["content"],
                    fingerprint(item["content"]),
                ),
            )


def retrieve(
    database: sqlite3.Connection,
    question: str,
    *,
    limit: int = 3,
    minimum_score: float = 0.34,
) -> list[Evidence]:
    query_terms = terms(question)
    if not query_terms:
        return []

    candidates: list[Evidence] = []
    rows = database.execute(
        "SELECT * FROM documents WHERE active = 1"
    ).fetchall()

    for row in rows:
        document_terms = terms(f"{row['title']} {row['content']}")
        overlap = query_terms & document_terms
        # Le score favorise la couverture de la question et pénalise
        # légèrement les correspondances limitées dans un grand vocabulaire.
        coverage = len(overlap) / len(query_terms)
        dilution = len(overlap) / math.sqrt(max(len(document_terms), 1))
        score = round(0.85 * coverage + 0.15 * dilution, 4)
        if score < minimum_score:
            continue

        candidates.append(
            Evidence(
                document_id=row["document_id"],
                title=row["title"],
                source_url=row["source_url"],
                version=row["version"],
                content_sha256=row["content_sha256"],
                score=score,
                excerpt=row["content"][:500],
            )
        )

    candidates.sort(key=lambda item: (-item.score, item.document_id))
    return candidates[:limit]


def build_evidence_packet(
    question: str,
    evidence: list[Evidence],
) -> dict[str, object]:
    question_hash = hashlib.sha256(question.encode("utf-8")).hexdigest()
    status = "ready_for_review" if evidence else "abstain_no_evidence"
    packet: dict[str, object] = {
        "status": status,
        "question_sha256": question_hash,
        "instruction": (
            "Répondre uniquement depuis les extraits fournis, citer les "
            "document_id et signaler toute information insuffisante."
        ),
        "evidence": [asdict(item) for item in evidence],
    }
    LOGGER.info(
        json.dumps(
            {
                "event": "evidence_packet_built",
                "question_sha256": question_hash,
                "status": status,
                "document_ids": [item.document_id for item in evidence],
            },
            ensure_ascii=False,
            sort_keys=True,
        )
    )
    return packet


if __name__ == "__main__":
    db = connect()
    upsert_documents(
        db,
        [
            {
                "document_id": "procedure-conges-v3",
                "title": "Procédure de demande de congés",
                "source_url": "https://intranet.example/procedures/conges",
                "version": "3.0",
                "content": (
                    "Une demande de congés est transmise au responsable "
                    "pour validation avant son enregistrement définitif."
                ),
            },
            {
                "document_id": "charte-achats-v2",
                "title": "Charte des achats",
                "source_url": "https://intranet.example/politiques/achats",
                "version": "2.0",
                "content": (
                    "Toute commande suit le circuit d'approbation défini "
                    "par le service achats."
                ),
            },
        ],
    )
    results = retrieve(db, "Comment faire valider une demande de congés ?")
    packet = build_evidence_packet(
        "Comment faire valider une demande de congés ?",
        results,
    )
    print(json.dumps(packet, ensure_ascii=False, indent=2))

Ce que ce code permet de vérifier

• les requêtes SQL sont paramétrées ;
• une version et une empreinte accompagnent chaque document ;
• les documents désactivés ne sont pas recherchés ;
• le score est lisible et peut être testé ;
• aucun résultat sous le seuil ne passe dans le paquet ;
• le log contient l’empreinte de la question, pas sa valeur brute ;
• le paquet expose les références exactes envoyées à l’étape suivante ;
• l’absence de résultat produit abstain_no_evidence.

Le seuil 0.34, le poids des scores et la longueur de l’extrait sont des paramètres pédagogiques, pas des valeurs universelles. Ils doivent être réglés sur un jeu d’évaluation. Le code garde les extraits dans le paquet et les affiche ; il ne convient donc pas à un corpus sensible sans contrôle d’accès, chiffrement, politique de logs et revue de l’interface.

Remplacer le baseline par une recherche gérée

Une fois les contrats et tests établis, un vector store ou File Search peut remplacer l’étape de recherche. Conservez toutefois l’interface logique :

question + filtres
  → résultats avec identifiants, scores et versions
  → paquet de contexte
  → génération
  → validation des références et des affirmations

Ne liez pas tout le système à la forme actuelle d’une réponse fournisseur. Créez un adaptateur qui transforme le résultat externe en votre objet Evidence. Vous pourrez ainsi comparer recherche lexicale, sémantique ou hybride avec le même jeu de tests.

La documentation OpenAI indique que les vector stores alimentent la recherche sémantique et que les fichiers y sont traités pour être recherchés. Vérifiez les formats, limites, coûts, règles de données et options actuels dans la documentation au moment du développement. Ne les figez pas dans un article ou une architecture sans date.

Journaliser sans créer une nouvelle fuite

Le module logging de Python fournit une infrastructure flexible, mais le choix des champs reste à votre charge. Une trace utile n’a pas besoin de contenir la question complète, la réponse complète ou les documents.

Journalisez de préférence :

• identifiant de requête ;
• empreinte ou identifiant de la question ;
• version du corpus et de l’index ;
• identifiants des fragments récupérés ;
• modèle et configuration pertinents ;
• décision de validation ;
• durée, erreur et coût lorsque disponibles.

Évitez par défaut : secrets, jetons, données personnelles, contenu documentaire complet et réponse brute. Définissez une durée de conservation et des droits d’accès. Une empreinte réduit l’exposition mais peut encore servir à corréler des événements ; elle n’annule pas toutes les obligations.

Exploiter SQLite avec des limites claires

Le module sqlite3 de Python fournit une interface DB-API vers SQLite. Il convient bien à un prototype local, à un outil mono-processus ou à une file de tests limitée. Il ne devient pas automatiquement la bonne base pour un service distribué ou fortement concurrent.

Pour un prototype :

• utilisez des contraintes et des transactions ;
• paramétrez les requêtes ;
• séparez métadonnées et contenu lorsque les droits diffèrent ;
• testez sauvegarde, restauration et retrait d’un document ;
• n’exposez jamais directement le fichier de base au Web ;
• migrez l’architecture lorsque le volume, la concurrence ou les exigences l’imposent.

Le guide Créer un moniteur de citations IA en Python montre un autre usage local de SQLite avec déduplication et dénominateurs visibles.

Les scénarios de test indispensables

1. source absente : la bonne sortie est l’abstention ;
2. source obsolète : la version active doit gagner ou le conflit doit être visible ;
3. documents contradictoires : la réponse ne doit pas choisir silencieusement ;
4. question hors périmètre : aucun passage simplement proche ne doit devenir preuve ;
5. injection dans un document : le texte documentaire ne doit pas modifier les règles système ;
6. retrait d’une source : elle disparaît de l’index et des réponses futures ;
7. citation inexistante : la validation rejette l’identifiant ;
8. affirmation non soutenue : une citation voisine ne suffit pas ;
9. donnée sensible : le log et l’interface appliquent la politique prévue ;
10. panne fournisseur : le workflow échoue proprement sans inventer de réponse.

Exécutez ces cas à chaque changement important de modèle, d’index, de segmentation ou de consigne. Un test réussi une fois ne garantit pas les versions futures.

Critères de passage du prototype au pilote

• le corpus et ses droits sont inventoriés ;
• chaque source possède une version et une procédure de retrait ;
• un jeu de questions avec cas négatifs existe ;
• le rappel de recherche est mesuré séparément ;
• la règle d’abstention est testée ;
• les citations sont validées contre les fragments réellement fournis ;
• les logs ne stockent pas de secrets ni de contenu inutile ;
• un responsable humain est défini pour les réponses à impact ;
• les limites de coût, latence et indisponibilité sont acceptées ;
• le rollback vers la version précédente de l’index est documenté.

Limites

Le prototype est pédagogique. Sa recherche par mots ne traite ni synonymes complexes, ni multilinguisme, ni compréhension sémantique. Il utilise une base en mémoire et des URL fictives. Il ne chiffre pas les données, ne gère pas les accès, n’appelle aucun modèle et ne valide aucune réponse générée. Les documentations de fournisseurs évoluent : vérifiez leurs paramètres, politiques et limites actuels avant toute intégration. Même avec une recherche excellente, un système génératif peut produire une affirmation incorrecte ; une source retrouvée n’est jamais une garantie de vérité.

Articles liés

• Créer un moniteur de citations IA en Python
• Prompt ou workflow : quand une instruction ne suffit plus
• Protéger les données sensibles lors de l’usage d’un assistant IA

Prochaine étape

Constituez dix questions avec réponse, cinq questions sans réponse et deux conflits de version. Faites passer le baseline lexical, mesurez la recherche, puis seulement comparez une solution vectorielle ou File Search avec les mêmes cas.