Fusion hybride : Combiner dense et sparse retrieval

La fusion hybride représente l'état de l'art du retrieval moderne. En combinant la compréhension sémantique du dense retrieval avec la précision lexicale du sparse retrieval, vous obtenez le meilleur des deux mondes. Ce guide explore les techniques de fusion, leurs implémentations et comment optimiser votre système hybride.

Pourquoi la fusion hybride ?

Chaque méthode de retrieval a ses forces et faiblesses :

La fusion hybride capture les cas où l'un ou l'autre échoue, améliorant le recall sans sacrifier la précision.

Benchmark BEIR : preuve par les chiffres

Sur le benchmark BEIR (diverse retrieval tasks), la fusion hybride surpasse systématiquement les approches isolées :

Techniques de fusion

1. Reciprocal Rank Fusion (RRF)

RRF est l'algorithme de fusion le plus populaire et le plus robuste. Il combine les rankings sans nécessiter de scores normalisés.

DEVELOPERpython
def reciprocal_rank_fusion(
    rankings: list[list[str]],
    k: int = 60
) -> list[tuple[str, float]]:
    """
    Reciprocal Rank Fusion

    rankings: Liste de classements (chaque classement = liste d'IDs ordonnés)
    k: Paramètre de lissage (60 par défaut, valeur standard)

    Formule: RRF_score(d) = Σ 1 / (k + rank(d))
    """
    fusion_scores = {}

    for ranking in rankings:
        for rank, doc_id in enumerate(ranking, start=1):
            if doc_id not in fusion_scores:
                fusion_scores[doc_id] = 0
            fusion_scores[doc_id] += 1 / (k + rank)

    # Trier par score décroissant
    sorted_results = sorted(
        fusion_scores.items(),
        key=lambda x: x[1],
        reverse=True
    )

    return sorted_results


# Exemple d'utilisation
dense_ranking = ["doc_a", "doc_c", "doc_b", "doc_d"]
sparse_ranking = ["doc_b", "doc_a", "doc_e", "doc_c"]

fused = reciprocal_rank_fusion([dense_ranking, sparse_ranking])
# [('doc_a', 0.032), ('doc_b', 0.032), ('doc_c', 0.031), ...]

Avantages de RRF :

• Ne nécessite pas de normalisation des scores
• Robuste aux outliers
• Paramètre k facile à tuner

2. Weighted Score Fusion

Combine les scores normalisés avec des poids configurables :

DEVELOPERpython
def weighted_score_fusion(
    dense_results: list[dict],
    sparse_results: list[dict],
    alpha: float = 0.5
) -> list[dict]:
    """
    Fusion pondérée des scores

    alpha: Poids du dense (0 = sparse only, 1 = dense only)

    Formule: score_final = alpha × dense_norm + (1-alpha) × sparse_norm
    """
    # Normaliser les scores (min-max)
    def normalize(results):
        if not results:
            return {}
        scores = [r["score"] for r in results]
        min_s, max_s = min(scores), max(scores)
        range_s = max_s - min_s if max_s != min_s else 1

        return {
            r["id"]: (r["score"] - min_s) / range_s
            for r in results
        }

    dense_norm = normalize(dense_results)
    sparse_norm = normalize(sparse_results)

    # Fusionner
    all_ids = set(dense_norm.keys()) | set(sparse_norm.keys())
    fused = []

    for doc_id in all_ids:
        d_score = dense_norm.get(doc_id, 0)
        s_score = sparse_norm.get(doc_id, 0)
        final_score = alpha * d_score + (1 - alpha) * s_score

        fused.append({
            "id": doc_id,
            "score": final_score,
            "dense_score": d_score,
            "sparse_score": s_score
        })

    return sorted(fused, key=lambda x: x["score"], reverse=True)

Comment choisir alpha ?

3. Convex Combination avec reranking

Approche en deux étapes : fusion puis reranking pour affiner :

DEVELOPERpython
from sentence_transformers import CrossEncoder

class HybridRetrieverWithRerank:
    def __init__(self, dense_retriever, sparse_retriever):
        self.dense = dense_retriever
        self.sparse = sparse_retriever
        self.reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

    def search(self, query: str, top_k: int = 5, rerank_k: int = 20):
        # Étape 1: Récupérer des candidats de chaque retriever
        dense_results = self.dense.search(query, top_k=rerank_k)
        sparse_results = self.sparse.search(query, top_k=rerank_k)

        # Étape 2: Fusion RRF
        dense_ids = [r["id"] for r in dense_results]
        sparse_ids = [r["id"] for r in sparse_results]
        fused = reciprocal_rank_fusion([dense_ids, sparse_ids])

        # Étape 3: Reranking des top candidats
        candidates = fused[:rerank_k]
        candidate_docs = self._get_documents([c[0] for c in candidates])

        pairs = [[query, doc["content"]] for doc in candidate_docs]
        rerank_scores = self.reranker.predict(pairs)

        # Combiner score RRF et rerank
        final_results = []
        for (doc_id, rrf_score), rerank_score, doc in zip(candidates, rerank_scores, candidate_docs):
            final_results.append({
                "id": doc_id,
                "content": doc["content"],
                "score": 0.3 * rrf_score + 0.7 * rerank_score
            })

        return sorted(final_results, key=lambda x: x["score"], reverse=True)[:top_k]

Implémentation avec les bases vectorielles

Qdrant : hybrid search natif

DEVELOPERpython
from qdrant_client import QdrantClient
from qdrant_client.models import (
    VectorParams, SparseVectorParams,
    PointStruct, SparseVector,
    SearchRequest, NamedVector, NamedSparseVector,
    Prefetch, FusionQuery, Fusion
)

client = QdrantClient("localhost", port=6333)

# Créer une collection hybride
client.create_collection(
    collection_name="hybrid_docs",
    vectors_config={
        "dense": VectorParams(size=1024, distance="Cosine")
    },
    sparse_vectors_config={
        "sparse": SparseVectorParams()
    }
)

# Indexer avec les deux types de vecteurs
def index_hybrid(doc_id: str, content: str, dense_emb, sparse_vec):
    client.upsert(
        collection_name="hybrid_docs",
        points=[PointStruct(
            id=doc_id,
            payload={"content": content},
            vector={
                "dense": dense_emb,
                "sparse": sparse_vec
            }
        )]
    )

# Recherche hybride avec RRF natif
def hybrid_search(query: str, top_k: int = 5):
    query_dense = encode_dense(query)
    query_sparse = encode_sparse(query)

    results = client.query_points(
        collection_name="hybrid_docs",
        prefetch=[
            Prefetch(
                query=query_dense,
                using="dense",
                limit=20
            ),
            Prefetch(
                query=query_sparse,
                using="sparse",
                limit=20
            )
        ],
        query=FusionQuery(fusion=Fusion.RRF),
        limit=top_k
    )

    return results

Elasticsearch : combined query

DEVELOPERpython
from elasticsearch import Elasticsearch

es = Elasticsearch()

def hybrid_search_es(query: str, query_embedding: list, top_k: int = 5):
    """
    Recherche hybride Elasticsearch avec kNN + BM25
    """
    response = es.search(
        index="hybrid_index",
        body={
            "size": top_k,
            "query": {
                "bool": {
                    "should": [
                        # Recherche BM25
                        {
                            "match": {
                                "content": {
                                    "query": query,
                                    "boost": 0.5
                                }
                            }
                        },
                        # Recherche kNN
                        {
                            "knn": {
                                "field": "embedding",
                                "query_vector": query_embedding,
                                "k": 20,
                                "num_candidates": 100,
                                "boost": 0.5
                            }
                        }
                    ]
                }
            }
        }
    )

    return response["hits"]["hits"]

Weaviate : hybrid alpha

DEVELOPERpython
import weaviate

client = weaviate.Client("http://localhost:8080")

def hybrid_search_weaviate(query: str, alpha: float = 0.5):
    """
    Weaviate hybrid search

    alpha: 0 = BM25 only, 1 = vector only
    """
    result = (
        client.query
        .get("Document", ["content", "title"])
        .with_hybrid(
            query=query,
            alpha=alpha,
            fusion_type="relativeScoreFusion"  # ou "rankedFusion"
        )
        .with_limit(5)
        .do()
    )

    return result["data"]["Get"]["Document"]

Stratégies avancées

Fusion conditionnelle

Adapter dynamiquement la stratégie selon le type de requête :

DEVELOPERpython
class AdaptiveHybridRetriever:
    def __init__(self, dense, sparse, classifier):
        self.dense = dense
        self.sparse = sparse
        self.classifier = classifier  # Classifie le type de requête

    def search(self, query: str, top_k: int = 5):
        # Classifier la requête
        query_type = self.classifier.predict(query)

        if query_type == "exact_match":
            # Codes, références → sparse dominant
            alpha = 0.2
        elif query_type == "semantic":
            # Questions naturelles → dense dominant
            alpha = 0.8
        else:
            # Hybride équilibré
            alpha = 0.5

        return self._hybrid_search(query, top_k, alpha)

    def _classify_query(self, query: str) -> str:
        """Heuristiques simples pour classifier"""
        # Détection de codes/références
        if re.search(r'[A-Z]{2,}\d+|#\d+|v\d+\.\d+', query):
            return "exact_match"

        # Requêtes très courtes → sparse
        if len(query.split()) <= 2:
            return "exact_match"

        # Questions → dense
        if query.lower().startswith(('comment', 'pourquoi', 'qu\'', 'quel')):
            return "semantic"

        return "balanced"

Multi-index fusion

Combiner plusieurs sources d'information :

DEVELOPERpython
def multi_source_fusion(
    query: str,
    retrievers: dict[str, Retriever],
    weights: dict[str, float],
    top_k: int = 5
):
    """
    Fusion de plusieurs sources

    retrievers = {
        "faq": faq_retriever,
        "docs": docs_retriever,
        "products": product_retriever
    }

    weights = {"faq": 1.5, "docs": 1.0, "products": 0.8}
    """
    all_rankings = []
    all_weights = []

    for source_name, retriever in retrievers.items():
        results = retriever.search(query, top_k=top_k * 2)
        ranking = [r["id"] for r in results]
        all_rankings.append(ranking)
        all_weights.append(weights.get(source_name, 1.0))

    # RRF pondéré
    fusion_scores = {}
    for ranking, weight in zip(all_rankings, all_weights):
        for rank, doc_id in enumerate(ranking, start=1):
            if doc_id not in fusion_scores:
                fusion_scores[doc_id] = 0
            fusion_scores[doc_id] += weight / (60 + rank)

    return sorted(fusion_scores.items(), key=lambda x: x[1], reverse=True)[:top_k]

Évaluation et tuning

A/B testing des paramètres

DEVELOPERpython
def evaluate_fusion_params(
    test_queries: list[dict],
    dense_retriever,
    sparse_retriever,
    param_grid: dict
):
    """
    Grid search sur les paramètres de fusion
    """
    results = []

    for alpha in param_grid.get("alpha", [0.3, 0.5, 0.7]):
        for k in param_grid.get("rrf_k", [20, 60, 100]):
            metrics = {
                "alpha": alpha,
                "rrf_k": k,
                "recall@5": [],
                "mrr": []
            }

            for test_case in test_queries:
                query = test_case["query"]
                relevant = test_case["relevant_docs"]

                # Exécuter la recherche
                dense_results = dense_retriever.search(query, top_k=20)
                sparse_results = sparse_retriever.search(query, top_k=20)

                # Fusion avec les paramètres
                fused = reciprocal_rank_fusion(
                    [[r["id"] for r in dense_results],
                     [r["id"] for r in sparse_results]],
                    k=k
                )

                # Calculer les métriques
                retrieved_ids = [doc_id for doc_id, _ in fused[:5]]
                hits = len(set(retrieved_ids) & set(relevant))
                metrics["recall@5"].append(hits / len(relevant))

                # MRR
                for i, doc_id in enumerate(retrieved_ids):
                    if doc_id in relevant:
                        metrics["mrr"].append(1 / (i + 1))
                        break
                else:
                    metrics["mrr"].append(0)

            metrics["recall@5"] = np.mean(metrics["recall@5"])
            metrics["mrr"] = np.mean(metrics["mrr"])
            results.append(metrics)

    return pd.DataFrame(results).sort_values("recall@5", ascending=False)

Monitoring en production

DEVELOPERpython
class HybridRetrieverWithMetrics:
    def __init__(self, dense, sparse, metrics_client):
        self.dense = dense
        self.sparse = sparse
        self.metrics = metrics_client

    def search(self, query: str, top_k: int = 5):
        start = time.time()

        # Recherches parallèles
        dense_results = self.dense.search(query, top_k=20)
        sparse_results = self.sparse.search(query, top_k=20)

        # Fusion
        fused = self._fuse(dense_results, sparse_results)

        # Métriques
        duration = time.time() - start
        self.metrics.record("retrieval_latency_ms", duration * 1000)
        self.metrics.record("dense_top1_in_final", dense_results[0]["id"] in [f["id"] for f in fused[:5]])
        self.metrics.record("sparse_top1_in_final", sparse_results[0]["id"] in [f["id"] for f in fused[:5]])

        # Analyse de divergence
        dense_set = set([r["id"] for r in dense_results[:5]])
        sparse_set = set([r["id"] for r in sparse_results[:5]])
        overlap = len(dense_set & sparse_set) / 5
        self.metrics.record("dense_sparse_overlap", overlap)

        return fused[:top_k]

Prochaines étapes

La fusion hybride est la base d'un retrieval robuste. Pour aller plus loin :

• Query Routing - Router vers la source optimale
• Ensemble Retrieval - Combiner plusieurs retrievers
• Fondamentaux du Retrieval - Vue d'ensemble

Fusion hybride automatique avec Ailog

Ailog implémente la fusion hybride de manière transparente :

• RRF natif optimisé pour vos contenus
• Alpha adaptatif basé sur l'analyse des requêtes
• Reranking automatique pour la précision maximale
• Monitoring intégré pour optimiser en continu

Testez gratuitement et bénéficiez d'un retrieval hybride sans configuration.