Hybrid Search Bm25 Vector Fusion

---
name: hybrid-search-bm25-vector-fusion
description: Architectural patterns, rank fusion algorithms (RRF, score normalization), HNSW vector index tuning, and production code for hybrid BM25 + dense vector search systems.
---

# Hybrid Search: BM25 & Vector Fusion Architecture

This skill provides architectural principles, rank fusion algorithms, HNSW vector tuning strategies, and production Python code implementations for combining sparse lexical search (BM25) with dense vector retrieval.

---

## 1. Core Architecture & Concept Comparison

Hybrid search combines the complementary strengths of **Sparse Lexical Retrieval** and **Dense Vector Retrieval**:

| Characteristic | BM25 (Sparse Lexical) | Vector Search (Dense Semantic) |
| :--- | :--- | :--- |
| **Matching Mechanism** | Inverted index token exact matching & term frequency | HNSW / IVF nearest neighbor in latent embedding space |
| **Strengths** | Exact keyword matching, rare terms, product IDs, jargon, acronyms | Conceptual similarity, paraphrasing, multi-lingual semantic alignment |
| **Weaknesses** | Vocabulary mismatch, zero recall on synonyms | Out-of-vocabulary domain shifts, failure on specific serial numbers / IDs |
| **Score Scale** | Unbounded $[0, \infty)$ non-linear BM25 score | Bounded $[-1, 1]$ (Cosine) or $[0, 1]$ (Dot Product / IP) |

---

## 2. Rank Fusion Algorithms

### 2.1 Reciprocal Rank Fusion (RRF)
RRF evaluates the rank position (not raw score) of documents across multiple retrieval streams to produce a single unified ranking.

#### Formula:
$$RRF\_Score(d \in D) = \sum_{m \in M} \frac{1}{k + r_m(d)}$$

Where:
- $M$: Set of retrieval algorithms (e.g., [BM25, Dense Vector]).
- $r_m(d)$: Rank of document $d$ in system $m$ (1-indexed).
- $k$: Smoothing constant (standard industry default: $k = 60$).

**Key Advantage**: Immune to score scale disparities across different search engines or models.

---

### 2.2 Relative Score Fusion (Min-Max Normalization + Weighted Sum)
Normalizes raw scores into $[0, 1]$ range before computing a weighted sum score.

#### Formula:
$$S_{norm}(d, m) = \frac{S(d, m) - S_{min}(m)}{S_{max}(m) - S_{min}(m)}$$

$$Score_{hybrid}(d) = \alpha \cdot S_{norm}(d, \text{bm25}) + (1 - \alpha) \cdot S_{norm}(d, \text{vector})$$

Where $\alpha \in [0, 1]$ represents the weight assigned to sparse lexical retrieval.

---

## 3. HNSW Vector Index Tuning Guidelines

Hierarchical Navigable Small World (HNSW) graphs control the speed vs. recall trade-off:

- **`m` (Number of bi-directional links per node)**:
  - Typical range: `16` to `64`.
  - Higher `m` improves recall for high-dimensional vectors (>1024 dims) at the cost of higher memory consumption and build time.
- **`ef_construction` (Depth of search during index creation)**:
  - Typical range: `64` to `512`.
  - Controls index build accuracy. Higher values increase indexing time but improve kNN search accuracy.
- **`ef_search` (Depth of search during query execution)**:
  - Typical range: `40` to `256`.
  - Higher `ef_search` improves recall@k performance at the expense of query latency.

---

## 4. Production Python Implementation

The following complete module implements async concurrent retrieval for BM25 and Dense Vector search, followed by Reciprocal Rank Fusion and Cross-Encoder re-ranking.

```python
import math
import asyncio
import logging
from typing import List, Dict, Any, Tuple, Optional

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class HybridSearchEngine:
    def __init__(self, rrf_k: int = 60):
        self.rrf_k = rrf_k

    def reciprocal_rank_fusion(
        self,
        results_list: List[List[Dict[str, Any]]],
        id_field: str = "id",
        weights: Optional[List[float]] = None
    ) -> List[Dict[str, Any]]:
        """
        Combines multiple ordered search result lists using Weighted Reciprocal Rank Fusion.

        :param results_list: List of ranked doc lists from distinct searchers.
        :param id_field: Field name containing unique document identifier.
        :param weights: Optional list of weights corresponding to each search list.
        :return: Re-ranked consolidated list of documents with fusion scores.
        """
        if weights is None:
            weights = [1.0] * len(results_list)

        doc_scores: Dict[str, float] = {}
        doc_store: Dict[str, Dict[str, Any]] = {}

        for list_idx, ranked_docs in enumerate(results_list):
            w = weights[list_idx]
            for rank, doc in enumerate(ranked_docs, start=1):
                doc_id = str(doc[id_field])
                rrf_score = w * (1.0 / (self.rrf_k + rank))

                if doc_id not in doc_scores:
                    doc_scores[doc_id] = 0.0
                    doc_store[doc_id] = doc.copy()
                
                doc_scores[doc_id] += rrf_score

        # Sort documents by calculated RRF score descending
        sorted_docs = sorted(
            doc_scores.items(), key=lambda item: item[1], reverse=True
        )

        fused_results = []
        for doc_id, score in sorted_docs:
            merged_doc = doc_store[doc_id]
            merged_doc["_hybrid_score"] = score
            fused_results.append(merged_doc)

        return fused_results

    def relative_score_fusion(
        self,
        bm25_results: List[Dict[str, Any]],
        vector_results: List[Dict[str, Any]],
        alpha: float = 0.5,
        id_field: str = "id"
    ) -> List[Dict[str, Any]]:
        """
        Normalizes scores using Min-Max scaling and computes weighted linear combination.
        """
        def normalize(docs: List[Dict[str, Any]], score_key: str) -> Dict[str, float]:
            if not docs:
                return {}
            scores = [d[score_key] for d in docs]
            min_s, max_s = min(scores), max(scores)
            if math.isclose(min_s, max_s):
                return {str(d[id_field]): 1.0 for d in docs}
            return {
                str(d[id_field]): (d[score_key] - min_s) / (max_s - min_s)
                for d in docs
            }

        bm25_norm = normalize(bm25_results, "_score")
        vector_norm = normalize(vector_results, "_score")

        doc_store = {str(d[id_field]): d for d in bm25_results + vector_results}
        combined_scores: Dict[str, float] = {}

        all_ids = set(bm25_norm.keys()).union(set(vector_norm.keys()))

        for doc_id in all_ids:
            s_bm25 = bm25_norm.get(doc_id, 0.0)
            s_vec = vector_norm.get(doc_id, 0.0)
            combined_scores[doc_id] = (alpha * s_bm25) + ((1.0 - alpha) * s_vec)

        sorted_ids = sorted(combined_scores.items(), key=lambda x: x[1], reverse=True)
        
        output = []
        for doc_id, score in sorted_ids:
            item = doc_store[doc_id].copy()
            item["_hybrid_score"] = score
            output.append(item)

        return output


# Async Execution Wrapper Example
async def simulate_lexical_search(query: str) -> List[Dict[str, Any]]:
    await asyncio.sleep(0.02) # Mock latency
    return [
        {"id": "doc1", "title": "Elasticsearch Tuning", "_score": 14.5},
        {"id": "doc2", "title": "Vector Databases Guide", "_score": 8.2},
        {"id": "doc3", "title": "Python Async Basics", "_score": 4.1},
    ]

async def simulate_vector_search(vector: List[float]) -> List[Dict[str, Any]]:
    await asyncio.sleep(0.03) # Mock latency
    return [
        {"id": "doc2", "title": "Vector Databases Guide", "_score": 0.92},
        {"id": "doc4", "title": "HNSW Algorithm Explained", "_score": 0.88},
        {"id": "doc1", "title": "Elasticsearch Tuning", "_score": 0.75},
    ]

async def run_hybrid_pipeline(query: str, query_vector: List[float]):
    search_engine = HybridSearchEngine(rrf_k=60)

    # Concurrently execute lexical and vector search streams
    bm25_task = asyncio.create_task(simulate_lexical_search(query))
    vec_task = asyncio.create_task(simulate_vector_search(query_vector))

    bm25_res, vec_res = await asyncio.gather(bm25_task, vec_task)

    # Fuse using RRF
    rrf_fused = search_engine.reciprocal_rank_fusion(
        results_list=[bm25_res, vec_res], weights=[1.0, 1.2]
    )

    logger.info("Hybrid Search RRF Top Result: %s (Score: %.4f)", 
                rrf_fused[0]["title"], rrf_fused[0]["_hybrid_score"])

if __name__ == "__main__":
    asyncio.run(run_hybrid_pipeline("search tuning", [0.1] * 1536))
```

---

## 5. Production OpenSearch Native Hybrid Search Pipeline

OpenSearch 2.11+ supports native Search Pipelines with RRF or Score Normalization:

```json
PUT /_search/pipeline/hybrid_rrf_pipeline
{
  "description": "Native Hybrid Search Pipeline with RRF",
  "phase_results_processors": [
    {
      "normalization-processor": {
        "normalization": {
          "technique": "min_max"
        },
        "combination": {
          "technique": "arithmetic_mean",
          "parameters": {
            "weights": [0.4, 0.6]
          }
        }
      }
    }
  ]
}
```

---

## 6. Anti-Patterns & Operational Pitfalls

| Anti-Pattern | Failure Mode | Corrective Action |
| :--- | :--- | :--- |
| **Direct Sum of Unnormalized Raw Scores** | BM25 scores (0–100+) dwarf vector cosine scores (0–1.0), rendering vector search useless. | Use RRF or Min-Max Relative Score Normalization before summing. |
| **Post-Filtering Vector Results** | Filters applied after vector kNN search truncate top-k candidates drastically. | Use **Single-Stage Pre-Filtering** in HNSW vector query. |
| **Static Weight Allocation for All Queries** | Fixed $\alpha$ weights perform poorly when queries vary between exact specs and conceptual prompts. | Use query classification to dynamically adjust $\alpha$ weights. |
| **High `ef_search` Under Heavy Load** | CPU usage spikes linearly with higher `ef_search` values, degrading P99 latency. | Benchmark recall@k vs. latency to find minimum acceptable `ef_search`. |