Python Implementation

Here's a complete hybrid search implementation:

import numpy as np
from typing import List, Dict, Tuple
from rank_bm25 import BM25Okapi
from sentence_transformers import SentenceTransformer
import chromadb

class HybridSearch:
    """
    Hybrid search combining BM25 and vector search.
    """
    
    def __init__(self, embedding_model: str = "nomic-embed-text"):
        # Initialize components
        self.bm25 = None
        self.vector_store = chromadb.Client()
        self.collection = self.vector_store.create_collection("docs")
        self.embedding_model = SentenceTransformer(embedding_model)
        self.documents = []
    
    def add_documents(self, documents: List[str]):
        """Add documents to both indexes."""
        self.documents = documents
        
        # BM25 index
        tokenized_docs = [doc.lower().split() for doc in documents]
        self.bm25 = BM25Okapi(tokenized_docs)
        
        # Vector index
        embeddings = self.embedding_model.encode(documents)
        self.collection.add(
            documents=documents,
            embeddings=embeddings.tolist(),
            ids=[f"doc_{i}" for i in range(len(documents))]
        )
    
    def bm25_search(self, query: str, k: int = 10) -> List[Tuple[int, float]]:
        """BM25 lexical search."""
        tokenized_query = query.lower().split()
        scores = self.bm25.get_scores(tokenized_query)
        
        # Get top-k indices
        top_k_idx = np.argsort(scores)[::-1][:k]
        return [(idx, scores[idx]) for idx in top_k_idx]
    
    def vector_search(self, query: str, k: int = 10) -> List[Tuple[int, float]]:
        """Vector semantic search."""
        query_embedding = self.embedding_model.encode([query])
        
        results = self.collection.query(
            query_embeddings=query_embedding.tolist(),
            n_results=k
        )
        
        # Extract indices and distances
        indices = [int(id.split('_')[1]) for id in results['ids'][0]]
        distances = [1 - d for d in results['distances'][0]]  # Convert to similarity
        
        return list(zip(indices, distances))
    
    def rrf_score(self, rank: int, k: int = 60) -> float:
        """Reciprocal Rank Fusion score."""
        return 1.0 / (k + rank)
    
    def hybrid_search(
        self, 
        query: str, 
        k: int = 10,
        bm25_weight: float = 0.5,
        vector_weight: float = 0.5
    ) -> List[Dict]:
        """
        Hybrid search combining BM25 and vector search.
        """
        # Get results from both methods
        bm25_results = self.bm25_search(query, k=k*2)
        vector_results = self.vector_search(query, k=k*2)
        
        # Calculate RRF scores
        doc_scores = {}
        
        for rank, (idx, _) in enumerate(bm25_results, 1):
            doc_scores[idx] = doc_scores.get(idx, 0) + self.rrf_score(rank)
        
        for rank, (idx, _) in enumerate(vector_results, 1):
            doc_scores[idx] = doc_scores.get(idx, 0) + self.rrf_score(rank)
        
        # Sort by combined score
        ranked_docs = sorted(doc_scores.items(), key=lambda x: x[1], reverse=True)
        
        # Return top-k with scores
        results = []
        for idx, score in ranked_docs[:k]:
            results.append({
                'document': self.documents[idx],
                'score': score,
                'index': idx
            })
        
        return results

# Example usage
search = HybridSearch()

documents = [
    "Python exception handling with try/except blocks",
    "Error handling best practices in Python",
    "Java exception handling tutorial",
    "Python debugging techniques",
    "How to catch exceptions in Python"
]

search.add_documents(documents)

# Hybrid search
results = search.hybrid_search("Python exception handling", k=3)

for r in results:
    print(f"Score: {r['score']:.4f} | {r['document']}")

Expected Output

Score: 0.0325 | Python exception handling with try/except blocks
Score: 0.0323 | How to catch exceptions in Python
Score: 0.0320 | Error handling best practices in Python