hybrid-search-implementation

name	hybrid-search-implementation
description	Combine vector and keyword search for improved retrieval. Use when implementing RAG systems, building search engines, or when neither approach alone provides sufficient recall.

Hybrid Search Implementation

Patterns for combining vector similarity and keyword-based search.

When to Use This Skill

• Building RAG systems with improved recall
• Combining semantic understanding with exact matching
• Handling queries with specific terms (names, codes)
• Improving search for domain-specific vocabulary
• When pure vector search misses keyword matches

Core Concepts

1. Hybrid Search Architecture

Query → ┬─► Vector Search ──► Candidates ─┐
        │                                  │
        └─► Keyword Search ─► Candidates ─┴─► Fusion ─► Results

2. Fusion Methods

Method	Description	Best For
RRF	Reciprocal Rank Fusion	General purpose
Linear	Weighted sum of scores	Tunable balance
Cross-encoder	Rerank with neural model	Highest quality
Cascade	Filter then rerank	Efficiency

Templates

Template 1: Reciprocal Rank Fusion

from typing import List, Dict, Tuple
from collections import defaultdict

def reciprocal_rank_fusion(
    result_lists: List[List[Tuple[str, float]]],
    k: int = 60,
    weights: List[float] = None
) -> List[Tuple[str, float]]:
    """
    Combine multiple ranked lists using RRF.

    Args:
        result_lists: List of (doc_id, score) tuples per search method
        k: RRF constant (higher = more weight to lower ranks)
        weights: Optional weights per result list

    Returns:
        Fused ranking as (doc_id, score) tuples
    """
    if weights is None:
        weights = [1.0] * len(result_lists)

    scores = defaultdict(float)

    for result_list, weight in zip(result_lists, weights):
        for rank, (doc_id, _) in enumerate(result_list):
            # RRF formula: 1 / (k + rank)
            scores[doc_id] += weight * (1.0 / (k + rank + 1))

    # Sort by fused score
    return sorted(scores.items(), key=lambda x: x[1], reverse=True)


def linear_combination(
    vector_results: List[Tuple[str, float]],
    keyword_results: List[Tuple[str, float]],
    alpha: float = 0.5
) -> List[Tuple[str, float]]:
    """
    Combine results with linear interpolation.

    Args:
        vector_results: (doc_id, similarity_score) from vector search
        keyword_results: (doc_id, bm25_score) from keyword search
        alpha: Weight for vector search (1-alpha for keyword)
    """
    # Normalize scores to [0, 1]
    def normalize(results):
        if not results:
            return {}
        scores = [s for _, s in results]
        min_s, max_s = min(scores), max(scores)
        range_s = max_s - min_s if max_s != min_s else 1
        return {doc_id: (score - min_s) / range_s for doc_id, score in results}

    vector_scores = normalize(vector_results)
    keyword_scores = normalize(keyword_results)

    # Combine
    all_docs = set(vector_scores.keys()) | set(keyword_scores.keys())
    combined = {}

    for doc_id in all_docs:
        v_score = vector_scores.get(doc_id, 0)
        k_score = keyword_scores.get(doc_id, 0)
        combined[doc_id] = alpha * v_score + (1 - alpha) * k_score

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

Template 2: PostgreSQL Hybrid Search

import asyncpg
from typing import List, Dict, Optional
import numpy as np

class PostgresHybridSearch:
    """Hybrid search with pgvector and full-text search."""

    def __init__(self, pool: asyncpg.Pool):
        self.pool = pool

    async def setup_schema(self):
        """Create tables and indexes."""
        async with self.pool.acquire() as conn:
            await conn.execute("""
                CREATE EXTENSION IF NOT EXISTS vector;

                CREATE TABLE IF NOT EXISTS documents (
                    id TEXT PRIMARY KEY,
                    content TEXT NOT NULL,
                    embedding vector(1536),
                    metadata JSONB DEFAULT '{}',
                    ts_content tsvector GENERATED ALWAYS AS (
                        to_tsvector('english', content)
                    ) STORED
                );

                -- Vector index (HNSW)
                CREATE INDEX IF NOT EXISTS documents_embedding_idx
                ON documents USING hnsw (embedding vector_cosine_ops);

                -- Full-text index (GIN)
                CREATE INDEX IF NOT EXISTS documents_fts_idx
                ON documents USING gin (ts_content);
            """)

    async def hybrid_search(
        self,
        query: str,
        query_embedding: List[float],
        limit: int = 10,
        vector_weight: float = 0.5,
        filter_metadata: [] = 
    ) -> []:
        
          .pool.acquire()  conn:
            
            where_clause = 
            params = [query_embedding, query, limit * ]

             filter_metadata:
                 key, value  filter_metadata.items():
                    params.append(value)
                    where_clause += 

            results =  conn.fetch(, *params, vector_weight)

             [(row)  row  results]

      () -> []:
        
         sentence_transformers  CrossEncoder

        
        candidates =  .hybrid_search(
            query, query_embedding, limit=rerank_candidates
        )

          candidates:
             []

        
        model = CrossEncoder()

        pairs = [(query, c[])  c  candidates]
        scores = model.predict(pairs)

         candidate, score  (candidates, scores):
            candidate[] = (score)

        
        reranked = (candidates, key= x: x[], reverse=)
         reranked[:limit]

Template 3: Elasticsearch Hybrid Search

from elasticsearch import Elasticsearch
from typing import List, Dict, Optional

class ElasticsearchHybridSearch:
    """Hybrid search with Elasticsearch and dense vectors."""

    def __init__(
        self,
        es_client: Elasticsearch,
        index_name: str = "documents"
    ):
        self.es = es_client
        self.index_name = index_name

    def create_index(self, vector_dims: int = 1536):
        """Create index with dense vector and text fields."""
        mapping = {
            "mappings": {
                "properties": {
                    "content": {
                        "type": "text",
                        "analyzer": "english"
                    },
                    "embedding": {
                        "type": "dense_vector",
                        "dims": vector_dims,
                        "index": True,
                        "similarity": "cosine"
                    },
                    "metadata": {
                        "type": "object",
                        "enabled": True
                    }
                }
            }
        }
        .es.indices.create(index=.index_name, body=mapping, ignore=)

     () -> []:
        
        
        search_body = {
            : limit,
            : {
                : {
                    : [
                        
                        {
                            : {
                                : {: {}},
                                : {
                                    : ,
                                    : {: query_embedding}
                                }
                            }
                        },
                        
                        {
                            : {
                                : {
                                    : query,
                                    : boost_text
                                }
                            }
                        }
                    ],
                    : 
                }
            }
        }

        
         :
            search_body[][][] = 

        response = .es.search(index=.index_name, body=search_body)

         [
            {
                : hit[],
                : hit[][],
                : hit[].get(, {}),
                : hit[]
            }
             hit  response[][]
        ]

     () -> []:
        
        search_body = {
            : limit,
            : [
                {
                    : {
                        : {
                            : query
                        }
                    }
                },
                {
                    : {
                        : {
                            : ,
                            : query_embedding,
                            : window_size,
                            : window_size * 
                        }
                    }
                }
            ],
            : {
                : {
                    : window_size,
                    : 
                }
            }
        }

        response = .es.search(index=.index_name, body=search_body)

         [
            {
                : hit[],
                : hit[][],
                : hit[]
            }
             hit  response[][]
        ]

Template 4: Custom Hybrid RAG Pipeline

from typing import List, Dict, Optional, Callable
from dataclasses import dataclass

@dataclass
class SearchResult:
    id: str
    content: str
    score: float
    source: str  # "vector", "keyword", "hybrid"
    metadata: Dict = None


class HybridRAGPipeline:
    """Complete hybrid search pipeline for RAG."""

    def __init__(
        self,
        vector_store,
        keyword_store,
        embedder,
        reranker=None,
        fusion_method: str = "rrf",
        vector_weight: float = 0.5
    ):
        self.vector_store = vector_store
        self.keyword_store = keyword_store
        self.embedder = embedder
        self.reranker = reranker
        self.fusion_method = fusion_method
        self.vector_weight = vector_weight

    async def search(
        self,
        query: str,
        top_k: int = 10,
        filter: Optional[Dict] = None,
        use_rerank:  = 
    ) -> [SearchResult]:
        

        
        query_embedding = .embedder.embed(query)

        
        vector_results, keyword_results =  asyncio.gather(
            ._vector_search(query_embedding, top_k * , ),
            ._keyword_search(query, top_k * , )
        )

        
         .fusion_method == :
            fused = ._rrf_fusion(vector_results, keyword_results)
        :
            fused = ._linear_fusion(vector_results, keyword_results)

        
         use_rerank  .reranker:
            fused =  ._rerank(query, fused[:top_k * ])

         fused[:top_k]

      () -> [SearchResult]:
        results =  .vector_store.search(embedding, limit, )
         [
            SearchResult(
                =r[],
                content=r[],
                score=r[],
                source=,
                metadata=r.get()
            )
             r  results
        ]

      () -> [SearchResult]:
        results =  .keyword_store.search(query, limit, )
         [
            SearchResult(
                =r[],
                content=r[],
                score=r[],
                source=,
                metadata=r.get()
            )
             r  results
        ]

     () -> [SearchResult]:
        
        k = 
        scores = {}
        content_map = {}

         rank, result  (vector_results):
            scores[result.] = scores.get(result., ) +  / (k + rank + )
            content_map[result.] = result

         rank, result  (keyword_results):
            scores[result.] = scores.get(result., ) +  / (k + rank + )
             result.   content_map:
                content_map[result.] = result

        sorted_ids = (scores.keys(), key= x: scores[x], reverse=)

         [
            SearchResult(
                =doc_id,
                content=content_map[doc_id].content,
                score=scores[doc_id],
                source=,
                metadata=content_map[doc_id].metadata
            )
             doc_id  sorted_ids
        ]

      () -> [SearchResult]:
        
          results:
             results

        pairs = [(query, r.content)  r  results]
        scores = .reranker.predict(pairs)

         result, score  (results, scores):
            result.score = (score)

         (results, key= x: x.score, reverse=)

Best Practices

Do's

• Tune weights empirically - Test on your data
• Use RRF for simplicity - Works well without tuning
• Add reranking - Significant quality improvement
• Log both scores - Helps with debugging
• A/B test - Measure real user impact

Don'ts

• Don't assume one size fits all - Different queries need different weights
• Don't skip keyword search - Handles exact matches better
• Don't over-fetch - Balance recall vs latency
• Don't ignore edge cases - Empty results, single word queries

Resources

• RRF Paper
• Vespa Hybrid Search
• Cohere Rerank