recherche_semantique.py — v2 (incrémental, normalisé)

[Mario Da Conceicao]

Version corrigée du 01/08/2026, après relecture critique. Retenu : indexation incrémentale (hash), normalisation des vecteurs. Refusé : nltk/spaCy — dépendances trop lourdes pour un système minimal, aucun besoin réel à ce stade.

import json
import hashlib
from pathlib import Path
import math

class RechercheSemantique:
    def __init__(self, moteur_embeddings):
        self.moteur = moteur_embeddings

    def _cosinus(self, v1, v2):
        produit = sum(a * b for a, b in zip(v1, v2))
        norme1 = math.sqrt(sum(a * a for a in v1))
        norme2 = math.sqrt(sum(b * b for b in v2))
        if norme1 == 0 or norme2 == 0:
            return 0
        return produit / (norme1 * norme2)

    def _normaliser(self, vecteur):
        """NOUVEAU : normalise le vecteur pour une similarité cosinus correcte."""
        norme = math.sqrt(sum(x * x for x in vecteur))
        return [x / norme for x in vecteur] if norme > 0 else vecteur

    def _hash_contenu(self, texte):
        """NOUVEAU : sert à détecter si un document a changé depuis le dernier index."""
        return hashlib.md5(texte.encode("utf-8")).hexdigest()

    def indexer_documents(self, dossier_documents, fichier_index):
        """
        MODIFIÉ : indexation incrémentale — un document dont le contenu n'a
        pas changé (même hash) n'est pas retraité, pour ne pas tout
        recalculer à chaque fois.
        """
        index_existant = {}
        if Path(fichier_index).exists():
            with open(fichier_index, "r", encoding="utf-8") as f:
                for entree in json.load(f):
                    index_existant.setdefault(entree["fichier"], []).append(entree)

        nouvel_index = []
        for fichier in Path(dossier_documents).glob("*.txt"):
            contenu = fichier.read_text(encoding="utf-8", errors="replace")
            hash_fichier = self._hash_contenu(contenu)

            deja_indexe = index_existant.get(fichier.name)
            if deja_indexe and deja_indexe[0].get("hash") == hash_fichier:
                nouvel_index.extend(deja_indexe)
                continue

            paragraphes = [p.strip() for p in contenu.split("\n\n") if p.strip()]
            for paragraphe in paragraphes:
                vecteur = self._normaliser(self.moteur.encoder(paragraphe))
                nouvel_index.append({
                    "fichier": fichier.name,
                    "texte": paragraphe,
                    "vecteur": vecteur,
                    "hash": hash_fichier,
                })

        with open(fichier_index, "w", encoding="utf-8") as f:
            json.dump(nouvel_index, f)
        return len(nouvel_index)

    def chercher(self, question, fichier_index, top_n=3):
        with open(fichier_index, "r", encoding="utf-8") as f:
            index = json.load(f)

        vecteur_question = self._normaliser(self.moteur.encoder(question))
        resultats = [(self._cosinus(vecteur_question, e["vecteur"]), e) for e in index]
        resultats.sort(key=lambda x: x[0], reverse=True)
        return [r[1] for r in resultats[:top_n]]

[Auteur]

Messages