Défis de la recherche dans les systèmes RAG

Dans le développement de systèmes RAG (Retrieval-Augmented Generation), la recherche initiale présente souvent des limites. Les résultats retournés peuvent sembler pertinents de par la présence de mots-clés, mais ne répondent pas véritablement à l'intention de l'utilisateur. Les problèmes courants incluent :

• La dépendance excessive aux mots-clés : Le document contient les termes de la requête mais manque de connexion sémantique réelle.
• L'absence de compréhension multimodale : Difficulté à établir des liens sémantiques entre le texte et les images.
• L'imprécision des algorithmes de similarité : Les mesures traditionnelles (comme la similarité cosinus) ne captent pas toujours la pertinence profonde.

Présentation du moteur d'évaluation sémantique

Un moteur d'évaluation sémantique basé sur des modèles de vision-langage comme Qwen2.5-VL peut adresser ces défis. Sa valeur réside dans sa capacité à :

• Traiter simultanément du texte, des images et des contenus mixtes.
• Produire un score de probabilité quantifié (entre 0 et 1) indiquant le degré de pertinence.
• Fournir une évaluation interprétable, utile pour le débogage.

Intégrer ce moteur dans un pipeline RAG s'articule autour de trois étapes fondamentales.

Étape 1 : Préparation des données d'entrée

La première étape consiste à structurer la requête utilisateur et les documents candidats sous un format compatible avec le moteur multimodal. Une requête peut contenir du texte, une image ou les deux. Les documents également.

import requests
from typing import Dict, List, Optional

# Structure pour une requête multimodale
requete_utilisateur: Dict = {
    "contenu_textuel": "Documents techniques sur les architectures de réseaux de neurones convolutifs",
    "reference_image": None,  # Chemin vers une image ou base64, le cas échéant
    "contexte_supplementaire": "Je cherche des articles académiques récents."
}

# Liste de documents candidats issus d'une recherche initiale
candidats_documentaires: List[Dict] = [
    {"contenu_textuel": "Une introduction détaillée aux réseaux de neurones convolutifs (CNN) et leurs applications en vision par ordinateur..."},
    {"contenu_textuel": "Guide pratique pour l'entraînement de modèles de détection d'objets basés sur des transformateurs..."},
    {"contenu_textuel": "Comparaison des performances de différents optimiseurs pour l'apprentissage profond..."}
]

Étape 2 : Interrogation du moteur d'évaluation

Une classe encapsule les appels à l'API du moteur d'évaluation. Elle calcule un score pour chaque paire (requête, document).

class EvaluateurSémantiqueMultimodal:
    def __init__(self, endpoint_api: str = "http://localhost:8000/evaluer"):
        self.endpoint = endpoint_api

    def calculer_pertinence(self, requete: Dict, document: Dict) -> float:
        """Évalue la pertinence d'un document par rapport à une requête."""
        charge_utile = {
            "requete": requete,
            "document": document
        }
        try:
            reponse = requests.post(self.endpoint, json=charge_utile, timeout=15)
            reponse.raise_for_status()
            donnees = reponse.json()
            return donnees.get("score_pertinence", 0.0)
        except requests.exceptions.RequestException as err:
            print(f"Erreur lors de l'appel API : {err}")
            return 0.0

    def classer_documents(self, requete: Dict, documents: List[Dict], nb_max: int = 3) -> List[tuple]:
        """Classe les documents par score de pertinence décroissant."""
        resultats_avec_scores = []
        for doc in documents:
            score = self.calculer_pertinence(requete, doc)
            resultats_avec_scores.append((score, doc))
        resultats_avec_scores.sort(key=lambda x: x[0], reverse=True)
        return resultats_avec_scores[:nb_max]

# Instanciation de l'évaluateur et classement des candidats
evaluateur = EvaluateurSémantiqueMultimodal()
classement_final = evaluateur.classer_documents(requete_utilisateur, candidats_documentaires)

Étape 3 : Intégration dans le pipeline RAG

Le classement obtenu remplace les résultats de la recherche initiale avant la phase de génération. Cela garantit que le LLM travaille sur les documents les plus pertinents.

def pipeline_rag_ameliore(requete_textuelle: str, fonction_recherche_initiale) -> List[Dict]:
    """Pipeline RAG intégrant l'étape de réévaluation sémantique."""
    # Recherche initiale
    documents_bruts = fonction_recherche_initiale(requete_textuelle)
    
    # Réévaluation et reclassement
    requete_pour_evaluateur = {"contenu_textuel": requete_textuelle, "reference_image": None}
    documents_classees = evaluateur.classer_documents(requete_pour_evaluateur, documents_bruts)
    
    # Extraction des documents pour le prompt de génération
    documents_pour_generation = [doc for _, doc in documents_classees]
    
    return documents_pour_generation

# Supposons une fonction de recherche existante
# resultats_finaux = pipeline_rag_ameliore("Explication des CNN", recherche_vecteur)

Analyse des performances et applications

L'intégration de cette étape de réévaluation a montré des améliorations mesurables sur des benchmarks internes.

Cette approche est particulièrement bénéfique pour les requêtes complexes ou multidimensionnelles. Par exemple, une requête comme "Schéma de câblage pour un contrôleur moteur pas-à-pas" bénéficiera d'une meilleure compréhension de l'image jointe et du texte, menant à des résultats plus ciblés.

Optimisations pour la production

Pour un déploiement en production, il est crucial d'optimiser la latence et les coûts. Les stratégies incluent :

• L'évaluation par lots : Envoyer plusieurs documents dans une seule requête API pour réduire la surcharge réseau.
• La mise en cache : Utiliser des mécanismes de cache (comme lru_cache) pour stocker les scores de paires requête/document récurrentes.
• Le filtrage par seuil : Appliquer un score de pertinence minimum pour ne reclassifier que les documents les plus prometteurs.

from functools import lru_cache

class EvaluateurAvecCache(EvaluateurSémantiqueMultimodal):
    def __init__(self, endpoint_api: str, taille_cache: int = 500):
        super().__init__(endpoint_api)
        self._calculer_pertinence_avec_cache = lru_cache(maxsize=taille_cache)(self._calculer_pertinence_basique)
    
    def _calculer_pertinence_basique(self, requete_hashable: str, document_hashable: str) -> float:
        """Version hashable pour le cache, appelant l'API."""
        # Logique de conversion en hashable et appel à l'API réelle
        pass
    
    def calculer_pertinence(self, requete: Dict, document: Dict) -> float:
        # Convertir en clés de cache (ex: tuple de strings)
        requete_key = str(sorted(requete.items()))
        doc_key = str(sorted(document.items()))
        return self._calculer_pertinence_avec_cache(requete_key, doc_key)

Cette méthode de réévaluation sémantique post-recherche permet d'améliorer significativement la qualité des résultats fournis au modèle de génération, menant à des réponses plus précises et contextuellement adaptées.