recherche_semantique.py — v3 (chunks avec chevauchement)

[Mario Da Conceicao]

Version corrigée du 01/08/2026. Retenu : découpage en "chunks" avec chevauchement plutôt que par paragraphe brut — améliore la qualité de ce que MIRA retrouve dans ses spécialités. Refusé (pour l’instant) : stockage des vecteurs en NumPy séparé — inutile à notre échelle réelle (quelques dizaines de documents personnels, pas des milliers).

import json
import hashlib
from pathlib import Path
import math

class RechercheSemantique:
    def __init__(self, moteur_embeddings, taille_chunk=300, chevauchement=50):
        self.moteur = moteur_embeddings
        self.taille_chunk = taille_chunk       # en mots
        self.chevauchement = chevauchement     # en mots, entre deux chunks voisins

    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):
        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):
        return hashlib.md5(texte.encode("utf-8")).hexdigest()

    def _decouper_en_chunks(self, texte):
        """
        NOUVEAU : découpe par mots avec chevauchement, plutôt que par simple
        paragraphe. Donne de meilleurs résultats sur des documents techniques
        longs (une info à cheval sur deux paragraphes n'est plus coupée).
        """
        mots = texte.split()
        if not mots:
            return []
        chunks = []
        pas = max(1, self.taille_chunk - self.chevauchement)
        for debut in range(0, len(mots), pas):
            chunk = " ".join(mots[debut:debut + self.taille_chunk])
            if chunk.strip():
                chunks.append(chunk)
            if debut + self.taille_chunk >= len(mots):
                break
        return chunks

    def indexer_documents(self, dossier_documents, fichier_index):
        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

            for chunk in self._decouper_en_chunks(contenu):
                vecteur = self._normaliser(self.moteur.encoder(chunk))
                nouvel_index.append({
                    "fichier": fichier.name,
                    "texte": chunk,
                    "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