Hybride Fusion: Dense und Sparse Retrieval kombinieren

Die hybride Fusion stellt den Stand der Technik im modernen Retrieval dar. Durch die Kombination des semantischen Verständnisses von Dense Retrieval mit der lexikalischen Präzision von Sparse Retrieval erhalten Sie das Beste aus beiden Welten. Dieser Leitfaden untersucht Fusionstechniken, ihre Implementierungen und wie Sie Ihr hybrides System optimieren.

Warum hybride Fusion?

Jede Retrieval-Methode hat ihre Stärken und Schwächen:

Die hybride Fusion erfasst die Fälle, in denen die eine oder andere Methode versagt, und verbessert den Recall, ohne die Präzision zu opfern.

Benchmark BEIR: Der Beweis in Zahlen

Im BEIR-Benchmark (verschiedene Retrieval-Aufgaben) übertrifft die hybride Fusion systematisch die einzelnen Ansätze:

Fusionstechniken

1. Reciprocal Rank Fusion (RRF)

RRF ist der beliebteste und robusteste Fusionsalgorithmus. Er kombiniert Rankings, ohne normalisierte Scores zu erfordern.

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

    rankings: Liste von Rankings (jedes Ranking = geordnete Liste von IDs)
    k: Glättungsparameter (Standard 60)

    Formel: 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)

    # Nach absteigendem Score sortieren
    sorted_results = sorted(
        fusion_scores.items(),
        key=lambda x: x[1],
        reverse=True
    )

    return sorted_results


# Anwendungsbeispiel
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), ...]

Vorteile von RRF:

• Erfordert keine Score-Normalisierung
• Robust gegenüber Ausreißern
• Parameter k leicht zu tunen

2. Weighted Score Fusion

Kombiniert normalisierte Scores mit konfigurierbaren Gewichten:

DEVELOPERpython
def weighted_score_fusion(
    dense_results: list[dict],
    sparse_results: list[dict],
    alpha: float = 0.5
) -> list[dict]:
    """
    Gewichtete Fusion der Scores

    alpha: Gewicht von Dense (0 = nur Sparse, 1 = nur Dense)

    Formel: score_final = alpha × dense_norm + (1-alpha) × sparse_norm
    """
    # Scores normalisieren (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)

    # Fusionieren
    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)

Wie wählt man Alpha?

3. Konvexe Kombination mit Reranking

Zweistufiger Ansatz: Fusion, dann Reranking zur Verfeinerung:

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):
        # Schritt 1: Kandidaten von jedem Retriever abrufen
        dense_results = self.dense.search(query, top_k=rerank_k)
        sparse_results = self.sparse.search(query, top_k=rerank_k)

        # Schritt 2: RRF-Fusion
        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])

        # Schritt 3: Reranking der Top-Kandidaten
        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)

        # RRF- und Rerank-Score kombinieren
        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]

Implementierung mit Vektordatenbanken

Qdrant: Nativer Hybrid Search

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)

# Eine hybride Collection erstellen
client.create_collection(
    collection_name="hybrid_docs",
    vectors_config={
        "dense": VectorParams(size=1024, distance="Cosine")
    },
    sparse_vectors_config={
        "sparse": SparseVectorParams()
    }
)

# Mit beiden Vektortypen indexieren
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
            }
        )]
    )

# Hybride Suche mit nativem RRF
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: Kombinierte Abfrage

DEVELOPERpython
from elasticsearch import Elasticsearch

es = Elasticsearch()

def hybrid_search_es(query: str, query_embedding: list, top_k: int = 5):
    """
    Hybride Elasticsearch-Suche mit kNN + BM25
    """
    response = es.search(
        index="hybrid_index",
        body={
            "size": top_k,
            "query": {
                "bool": {
                    "should": [
                        # BM25-Suche
                        {
                            "match": {
                                "content": {
                                    "query": query,
                                    "boost": 0.5
                                }
                            }
                        },
                        # kNN-Suche
                        {
                            "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"  # oder "rankedFusion"
        )
        .with_limit(5)
        .do()
    )

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

Fortgeschrittene Strategien

Bedingte Fusion

Die Strategie dynamisch an den Anfragetyp anpassen:

DEVELOPERpython
class AdaptiveHybridRetriever:
    def __init__(self, dense, sparse, classifier):
        self.dense = dense
        self.sparse = sparse
        self.classifier = classifier  # Klassifiziert den Anfragetyp

    def search(self, query: str, top_k: int = 5):
        # Anfrage klassifizieren
        query_type = self.classifier.predict(query)

        if query_type == "exact_match":
            # Codes, Referenzen → Sparse dominant
            alpha = 0.2
        elif query_type == "semantic":
            # Natürliche Fragen → Dense dominant
            alpha = 0.8
        else:
            # Ausgewogen hybrid
            alpha = 0.5

        return self._hybrid_search(query, top_k, alpha)

    def _classify_query(self, query: str) -> str:
        """Einfache Heuristiken zur Klassifizierung"""
        # Erkennung von Codes/Referenzen
        if re.search(r'[A-Z]{2,}\d+|#\d+|v\d+\.\d+', query):
            return "exact_match"

        # Sehr kurze Anfragen → sparse
        if len(query.split()) <= 2:
            return "exact_match"

        # Fragen → dense
        if query.lower().startswith(('wie', 'warum', 'was', 'welche')):
            return "semantic"

        return "balanced"

Multi-Index-Fusion

Mehrere Informationsquellen kombinieren:

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

    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))

    # Gewichtetes RRF
    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]

Bewertung und Tuning

A/B-Tests der Parameter

DEVELOPERpython
def evaluate_fusion_params(
    test_queries: list[dict],
    dense_retriever,
    sparse_retriever,
    param_grid: dict
):
    """
    Grid Search über die Fusionsparameter
    """
    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"]

                # Suche ausführen
                dense_results = dense_retriever.search(query, top_k=20)
                sparse_results = sparse_retriever.search(query, top_k=20)

                # Fusion mit den Parametern
                fused = reciprocal_rank_fusion(
                    [[r["id"] for r in dense_results],
                     [r["id"] for r in sparse_results]],
                    k=k
                )

                # Metriken berechnen
                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 in der Produktion

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()

        # Parallele Suchen
        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)

        # Metriken
        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]])

        # Divergenzanalyse
        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]

Nächste Schritte

Die hybride Fusion ist die Grundlage eines robusten Retrieval. Um tiefer einzusteigen:

• Query Routing - Anfragen zur optimalen Quelle leiten
• Ensemble Retrieval - Mehrere Retriever kombinieren
• Retrieval-Grundlagen - Überblick

Automatische hybride Fusion mit Ailog

Ailog implementiert die hybride Fusion transparent:

• Natives RRF optimiert für Ihre Inhalte
• Adaptives Alpha basierend auf Anfrageanalyse
• Automatisches Reranking für maximale Präzision
• Integriertes Monitoring zur kontinuierlichen Optimierung

Kostenlos testen und von hybridem Retrieval ohne Konfiguration profitieren.