Hybrid Search Strategies

Semantic search understands "cozy blanket for winter" but fails on "SKU-ABC123". Keyword search nails exact matches but misses "warm throw" when you search "blanket". The solution? Hybrid search - combining both for the best of both worlds.

When Each Search Type Wins

Implementing Hybrid Search

Option 1: Weaviate (Built-in Hybrid)

Weaviate has native hybrid search support - the easiest option.

import weaviate
from weaviate.classes.query import HybridFusion

client = weaviate.connect_to_weaviate_cloud(
    cluster_url="your-url",
    auth_credentials=weaviate.auth.AuthApiKey("your-key")
)

products = client.collections.get("Products")

def hybrid_search_weaviate(query: str, alpha: float = 0.7, limit: int = 10):
    """
    Hybrid search using Weaviate's built-in support.

    alpha: 0 = pure keyword, 1 = pure vector, 0.5-0.7 typically best for e-commerce
    """
    response = products.query.hybrid(
        query=query,
        alpha=alpha,
        fusion_type=HybridFusion.RELATIVE_SCORE,  # or RANKED
        limit=limit,
        return_metadata=["score", "explain_score"]
    )

    return [
        {
            "name": obj.properties["name"],
            "score": obj.metadata.score,
            "explain": obj.metadata.explain_score
        }
        for obj in response.objects
    ]


# Example: Mixed query benefits from hybrid
results = hybrid_search_weaviate("Nike running shoes waterproof", alpha=0.6)

Option 2: Manual Hybrid with Pinecone + Elasticsearch

When your vector DB doesn't have built-in hybrid, combine two systems.

from pinecone import Pinecone
from elasticsearch import Elasticsearch
from openai import OpenAI
from typing import List, Dict

class HybridSearchEngine:
    def __init__(
        self,
        pinecone_index,
        es_client: Elasticsearch,
        openai_client: OpenAI,
        es_index: str = "products"
    ):
        self.vector_index = pinecone_index
        self.es = es_client
        self.client = openai_client
        self.es_index = es_index

    def search(
        self,
        query: str,
        alpha: float = 0.6,
        top_k: int = 20,
        final_k: int = 10
    ) -> List[Dict]:
        """
        Hybrid search combining vector and keyword results.

        Args:
            query: Search query
            alpha: Weight for semantic (1=pure semantic, 0=pure keyword)
            top_k: How many to retrieve from each system
            final_k: How many final results to return
        """

        # Get results from both systems
        vector_results = self._vector_search(query, top_k)
        keyword_results = self._keyword_search(query, top_k)

        # Combine using weighted scoring
        combined = self._combine_results(
            vector_results,
            keyword_results,
            alpha
        )

        return combined[:final_k]

    def _vector_search(self, query: str, top_k: int) -> List[Dict]:
        """Semantic search using Pinecone"""

        embedding = self.client.embeddings.create(
            model="text-embedding-3-small",
            input=query
        ).data[0].embedding

        results = self.vector_index.query(
            vector=embedding,
            top_k=top_k,
            include_metadata=True
        )

        return [
            {
                "id": m.id,
                "score": m.score,
                "metadata": m.metadata,
                "source": "vector"
            }
            for m in results.matches
        ]

    def _keyword_search(self, query: str, top_k: int) -> List[Dict]:
        """Keyword search using Elasticsearch BM25"""

        response = self.es.search(
            index=self.es_index,
            body={
                "query": {
                    "multi_match": {
                        "query": query,
                        "fields": ["name^3", "description", "brand^2", "category"],
                        "type": "best_fields",
                        "fuzziness": "AUTO"
                    }
                },
                "size": top_k
            }
        )

        max_score = response["hits"]["max_score"] or 1

        return [
            {
                "id": hit["_id"],
                "score": hit["_score"] / max_score,  # Normalize to 0-1
                "metadata": hit["_source"],
                "source": "keyword"
            }
            for hit in response["hits"]["hits"]
        ]

    def _combine_results(
        self,
        vector_results: List[Dict],
        keyword_results: List[Dict],
        alpha: float
    ) -> List[Dict]:
        """Combine results using weighted scoring"""

        # Build lookup by ID
        combined = {}

        for result in vector_results:
            combined[result["id"]] = {
                "id": result["id"],
                "metadata": result["metadata"],
                "vector_score": result["score"],
                "keyword_score": 0,
                "sources": ["vector"]
            }

        for result in keyword_results:
            if result["id"] in combined:
                combined[result["id"]]["keyword_score"] = result["score"]
                combined[result["id"]]["sources"].append("keyword")
            else:
                combined[result["id"]] = {
                    "id": result["id"],
                    "metadata": result["metadata"],
                    "vector_score": 0,
                    "keyword_score": result["score"],
                    "sources": ["keyword"]
                }

        # Calculate final score
        for item in combined.values():
            item["final_score"] = (
                alpha * item["vector_score"] +
                (1 - alpha) * item["keyword_score"]
            )

        # Sort by final score
        return sorted(combined.values(), key=lambda x: x["final_score"], reverse=True)

Reciprocal Rank Fusion (RRF)

RRF is elegant - it combines rankings without needing to normalize scores.

def reciprocal_rank_fusion(
    result_lists: List[List[Dict]],
    k: int = 60  # RRF constant, typically 60
) -> List[Dict]:
    """
    Combine multiple ranked result lists using RRF.

    RRF score = sum(1 / (k + rank)) for each list where the item appears

    This avoids score normalization issues since it only uses rank positions.
    """

    rrf_scores = {}

    for results in result_lists:
        for rank, result in enumerate(results, 1):
            doc_id = result["id"]

            if doc_id not in rrf_scores:
                rrf_scores[doc_id] = {
                    "id": doc_id,
                    "metadata": result.get("metadata", {}),
                    "rrf_score": 0,
                    "appearances": 0
                }

            # RRF formula
            rrf_scores[doc_id]["rrf_score"] += 1 / (k + rank)
            rrf_scores[doc_id]["appearances"] += 1

    # Sort by RRF score
    return sorted(rrf_scores.values(), key=lambda x: x["rrf_score"], reverse=True)


# Usage
vector_results = vector_search(query)
keyword_results = keyword_search(query)

fused = reciprocal_rank_fusion([vector_results, keyword_results])

Query Classification for Adaptive Search

Instead of fixed alpha, adapt based on query type.

def classify_query_type(query: str) -> Dict:
    """Classify query to determine optimal search strategy"""

    # Check for SKU/product code patterns
    import re

    sku_patterns = [
        r'^[A-Z]{2,4}-?\d{4,}

Boosting and Filtering

Combine hybrid search with business logic.

def search_with_boosting(
    query: str,
    boost_in_stock: float = 1.5,
    boost_high_rating: float = 1.2,
    recency_decay: bool = True
) -> List[Dict]:
    """Hybrid search with business boosting"""

    # Get base hybrid results
    results = hybrid_engine.search(query, alpha=0.6, final_k=50)

    # Apply boosts
    for result in results:
        metadata = result["metadata"]
        boost = 1.0

        # Boost in-stock items
        if metadata.get("in_stock", False):
            boost *= boost_in_stock

        # Boost high-rated items
        if metadata.get("rating", 0) >= 4.5:
            boost *= boost_high_rating

        # Apply recency decay for time-sensitive products
        if recency_decay and metadata.get("created_at"):
            days_old = (datetime.now() - metadata["created_at"]).days
            recency_factor = 1 / (1 + days_old / 365)  # Decay over a year
            boost *= (0.5 + 0.5 * recency_factor)  # Min 0.5x, max 1x

        result["boosted_score"] = result["final_score"] * boost

    # Re-sort by boosted score
    return sorted(results, key=lambda x: x["boosted_score"], reverse=True)[:10]

Measuring Hybrid Search Quality

def evaluate_search_quality(
    test_queries: List[Dict],  # [{"query": "...", "relevant_ids": [...]}]
    search_function,
    k: int = 10
) -> Dict:
    """Evaluate search quality with standard IR metrics"""

    results = {
        "precision_at_k": [],
        "recall_at_k": [],
        "mrr": [],  # Mean Reciprocal Rank
        "ndcg": []  # Normalized Discounted Cumulative Gain
    }

    for test in test_queries:
        query = test["query"]
        relevant = set(test["relevant_ids"])

        # Get search results
        search_results = search_function(query, top_k=k)
        result_ids = [r["id"] for r in search_results]

        # Precision@K
        relevant_in_results = len(set(result_ids) & relevant)
        precision = relevant_in_results / k
        results["precision_at_k"].append(precision)

        # Recall@K
        recall = relevant_in_results / len(relevant) if relevant else 0
        results["recall_at_k"].append(recall)

        # MRR (first relevant result)
        for i, rid in enumerate(result_ids, 1):
            if rid in relevant:
                results["mrr"].append(1 / i)
                break
        else:
            results["mrr"].append(0)

    # Average metrics
    return {
        metric: sum(values) / len(values)
        for metric, values in results.items()
    }


# Compare different alpha values
for alpha in [0.0, 0.3, 0.5, 0.7, 1.0]:
    def search_fn(q, top_k):
        return hybrid_engine.search(q, alpha=alpha, final_k=top_k)

    metrics = evaluate_search_quality(test_queries, search_fn)
    print(f"Alpha {alpha}: Precision@10={metrics['precision_at_k']:.3f}, MRR={metrics['mrr']:.3f}")

Key Takeaways


  

    
Hybrid beats pure semantic or keyword - Especially for e-commerce with mixed query types
    
RRF for score combination - Elegant solution that avoids normalization issues
    
Adaptive alpha based on query type - SKUs need keyword, descriptions need semantic
    
Measure and tune - Use precision/recall to find optimal settings for your catalog
  


Next up: RAG Agents Assessment - Test your knowledge with a comprehensive quiz!
,  # SKU-12345
        r'^B0[A-Z0-9]{8}

Boosting and Filtering

Combine hybrid search with business logic.

___CODEBLOCK_4___

Measuring Hybrid Search Quality

___CODEBLOCK_5___

Key Takeaways


  

    
Hybrid beats pure semantic or keyword - Especially for e-commerce with mixed query types
    
RRF for score combination - Elegant solution that avoids normalization issues
    
Adaptive alpha based on query type - SKUs need keyword, descriptions need semantic
    
Measure and tune - Use precision/recall to find optimal settings for your catalog
  


Next up: RAG Agents Assessment - Test your knowledge with a comprehensive quiz!
,        # Amazon ASIN
        r'^\d{12,13}

Boosting and Filtering

Combine hybrid search with business logic.

___CODEBLOCK_4___

Measuring Hybrid Search Quality

___CODEBLOCK_5___

Key Takeaways


  

    
Hybrid beats pure semantic or keyword - Especially for e-commerce with mixed query types
    
RRF for score combination - Elegant solution that avoids normalization issues
    
Adaptive alpha based on query type - SKUs need keyword, descriptions need semantic
    
Measure and tune - Use precision/recall to find optimal settings for your catalog
  


Next up: RAG Agents Assessment - Test your knowledge with a comprehensive quiz!
,            # UPC/EAN
    ]

    for pattern in sku_patterns:
        if re.match(pattern, query.strip(), re.IGNORECASE):
            return {"type": "sku", "alpha": 0.1}  # Almost pure keyword

    # Check for brand + model pattern
    brand_model_pattern = r'^[A-Z][a-z]+\s+[A-Z]?[a-z]*\s*\d+.*

Boosting and Filtering

Combine hybrid search with business logic.

___CODEBLOCK_4___

Measuring Hybrid Search Quality

___CODEBLOCK_5___

Key Takeaways


  

    
Hybrid beats pure semantic or keyword - Especially for e-commerce with mixed query types
    
RRF for score combination - Elegant solution that avoids normalization issues
    
Adaptive alpha based on query type - SKUs need keyword, descriptions need semantic
    
Measure and tune - Use precision/recall to find optimal settings for your catalog
  


Next up: RAG Agents Assessment - Test your knowledge with a comprehensive quiz!

    if re.match(brand_model_pattern, query.strip()):
        return {"type": "brand_model", "alpha": 0.3}

    # Check for natural language (multiple words, descriptive)
    words = query.split()
    if len(words) >= 4:
        return {"type": "descriptive", "alpha": 0.8}  # More semantic

    # Default balanced
    return {"type": "mixed", "alpha": 0.5}


class AdaptiveHybridSearch:
    def __init__(self, search_engine: HybridSearchEngine):
        self.engine = search_engine

    def search(self, query: str, top_k: int = 10) -> List[Dict]:
        """Search with automatically determined alpha"""

        classification = classify_query_type(query)
        alpha = classification["alpha"]

        print(f"Query type: {classification['type']}, using alpha={alpha}")

        return self.engine.search(query, alpha=alpha, final_k=top_k)


# Examples
engine = AdaptiveHybridSearch(hybrid_engine)

# SKU query - will use alpha=0.1 (mostly keyword)
engine.search("SKU-78901")

# Natural language - will use alpha=0.8 (mostly semantic)
engine.search("comfortable office chair for someone with back pain")

# Mixed - will use alpha=0.5
engine.search("Sony headphones")

Boosting and Filtering

Combine hybrid search with business logic.

Measuring Hybrid Search Quality

Key Takeaways

Next up: RAG Agents Assessment - Test your knowledge with a comprehensive quiz!