E-commerce hybrid search: keyword plus vector in LangChain

Hybrid search is the default choice for e-commerce because pure vector search misses exact SKU matches and pure keyword search fails on semantic intent. This tutorial builds a complete ecommerce hybrid search langchain pipeline: BM25 for exact token overlap, dense vectors for semantic similarity, reciprocal rank fusion for merging, and a cross-encoder reranker for precision. You’ll end up with a runnable module you can drop into a FastAPI service or batch job.

Prerequisites

• Python 3.10+
• An OpenAI API key (or any OpenAI-compatible endpoint) for embeddings and reranking
• ~2 GB RAM for the cross-encoder model
• Familiarity with LangChain’s Document and Retriever abstractions

Install dependencies:

pip install langchain langchain-openai langchain-community \
    rank-bm25 sentence-transformers faiss-cpu \
    pydantic tqdm

If you prefer a managed inference gateway instead of calling providers directly, you can point the OpenAI client at n4n.ai’s OpenAI-compatible endpoint — it handles fallback across 240+ models and forwards provider cache-control headers automatically.

Project structure

ecommerce-hybrid-search/
├── data/
│   └── products.jsonl          # one JSON object per line
├── src/
│   ├── __init__.py
│   ├── config.py               # constants, model names
│   ├── loaders.py              # data ingestion
│   ├── bm25_retriever.py       # keyword search
│   ├── vector_retriever.py     # semantic search
│   ├── hybrid_retriever.py     # RRF fusion + rerank
│   └── evaluate.py             # quick metrics
└── main.py                     # demo script

Configuration

# src/config.py
from dataclasses import dataclass

@dataclass(frozen=True)
class Settings:
    embedding_model: str = "text-embedding-3-small"
    embedding_dim: int = 1536
    cross_encoder_model: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"
    bm25_k: int = 50          # candidates from BM25
    vector_k: int = 50        # candidates from vector search
    rrf_k: int = 60           # RRF constant
    final_k: int = 10         # results after rerank
    openai_api_base: str | None = None  # override for custom gateway
    openai_api_key: str = "sk-..."      # set via env in production

settings = Settings()

Load product data

Each product becomes a LangChain Document with metadata for filtering and display.

# src/loaders.py
import json
from pathlib import Path
from langchain_core.documents import Document
from src.config import settings

def load_products(path: Path) -> list[Document]:
    docs: list[Document] = []
    with path.open("r", encoding="utf-8") as f:
        for line in f:
            row = json.loads(line)
            # Combine searchable fields into page_content
            content = " ".join(filter(None, [
                row.get("title", ""),
                row.get("description", ""),
                row.get("brand", ""),
                row.get("category", ""),
                " ".join(row.get("tags", [])),
            ]))
            metadata = {
                "product_id": row["product_id"],
                "title": row["title"],
                "price": row["price"],
                "brand": row.get("brand"),
                "category": row.get("category"),
                "in_stock": row.get("in_stock", True),
            }
            docs.append(Document(page_content=content, metadata=metadata))
    return docs

Sample data/products.jsonl (create 50-100 lines for testing):

{"product_id": "SKU-1001", "title": "Wireless Noise-Cancelling Headphones", "description": "Over-ear ANC headphones with 30hr battery life", "brand": "Sony", "category": "Electronics > Audio", "tags": ["wireless", "bluetooth", "anc", "travel"], "price": 249.99, "in_stock": true}
{"product_id": "SKU-1002", "title": "USB-C Mechanical Keyboard", "description": "Hot-swappable switches, per-key RGB, aluminum case", "brand": "Keychron", "category": "Electronics > Keyboards", "tags": ["mechanical", "rgb", "hot-swap", "programmable"], "price": 179.00, "in_stock": true}
{"product_id": "SKU-1003", "title": "Organic Cotton T-Shirt", "description": "Pre-shrunk, medium weight, classic fit", "brand": "Everlane", "category": "Apparel > T-Shirts", "tags": ["organic", "cotton", "sustainable", "basic"], "price": 28.00, "in_stock": false}

BM25 keyword retriever

LangChain Community ships a BM25Retriever. We wrap it to respect our k and add a simple filter for in-stock items.

# src/bm25_retriever.py
from langchain_community.retrievers import BM25Retriever
from langchain_core.documents import Document
from langchain_core.callbacks import CallbackManagerForRetrieverRun
from typing import List, Optional
from src.config import settings

class FilteredBM25Retriever(BM25Retriever):
    def _get_relevant_documents(
        self, query: str, *, run_manager: CallbackManagerForRetrieverRun
    ) -> List[Document]:
        docs = super()._get_relevant_documents(query, run_manager=run_manager)
        # Filter in-stock only
        filtered = [d for d in docs if d.metadata.get("in_stock", True)]
        return filtered[:settings.bm25_k]

def build_bm25_retriever(documents: List[Document]) -> FilteredBM25Retriever:
    return FilteredBM25Retriever.from_documents(documents)

Vector retriever with FAISS

Use FAISS for local indexing. In production you’d swap this for a managed vector DB (Pinecone, Weaviate, pgvector).

# src/vector_retriever.py
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever
from langchain_core.callbacks import CallbackManagerForRetrieverRun
from typing import List
from src.config import settings

class InStockVectorRetriever(BaseRetriever):
    vectorstore: FAISS
    k: int = settings.vector_k

    def _get_relevant_documents(
        self, query: str, *, run_manager: CallbackManagerForRetrieverRun
    ) -> List[Document]:
        # FAISS similarity search with metadata filter
        docs = self.vectorstore.similarity_search(
            query, k=self.k, filter={"in_stock": True}
        )
        return docs

def build_vector_retriever(documents: List[Document]) -> InStockVectorRetriever:
    embeddings = OpenAIEmbeddings(
        model=settings.embedding_model,
        openai_api_base=settings.openai_api_base,
        openai_api_key=settings.openai_api_key,
    )
    vs = FAISS.from_documents(documents, embeddings)
    return InStockVectorRetriever(vectorstore=vs)

Reciprocal Rank Fusion

RRF merges two ranked lists without needing score normalization. Formula: score = sum(1 / (k + rank_i)) for each list where the doc appears.

# src/hybrid_retriever.py
from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever
from langchain_core.callbacks import CallbackManagerForRetrieverRun
from typing import List, Dict, Set
from sentence_transformers import CrossEncoder
from src.config import settings

class HybridRetriever(BaseRetriever):
    bm25_retriever: BaseRetriever
    vector_retriever: BaseRetriever
    cross_encoder: CrossEncoder
    final_k: int = settings.final_k
    rrf_k: int = settings.rrf_k

    def _get_relevant_documents(
        self, query: str, *, run_manager: CallbackManagerForRetrieverRun
    ) -> List[Document]:
        # 1. Retrieve from both sources
        bm25_docs = self.bm25_retriever.invoke(query)
        vector_docs = self.vector_retriever.invoke(query)

        # 2. RRF fusion
        fused_scores: Dict[str, float] = {}
        doc_map: Dict[str, Document] = {}

        for rank, doc in enumerate(bm25_docs):
            pid = doc.metadata["product_id"]
            fused_scores[pid] = fused_scores.get(pid, 0.0) + 1.0 / (self.rrf_k + rank + 1)
            doc_map[pid] = doc

        for rank, doc in enumerate(vector_docs):
            pid = doc.metadata["product_id"]
            fused_scores[pid] = fused_scores.get(pid, 0.0) + 1.0 / (self.rrf_k + rank + 1)
            doc_map[pid] = doc

        # 3. Sort by fused score, take top-N for reranking
        rerank_candidates = sorted(
            fused_scores.items(), key=lambda x: x[1], reverse=True
        )[: self.final_k * 3]  # over-fetch for reranker

        if not rerank_candidates:
            return []

        candidate_docs = [doc_map[pid] for pid, _ in rerank_candidates]

        # 4. Cross-encoder reranking
        pairs = [(query, doc.page_content) for doc in candidate_docs]
        scores = self.cross_encoder.predict(pairs, show_progress_bar=False)

        reranked = sorted(
            zip(candidate_docs, scores), key=lambda x: x[1], reverse=True
        )

        return [doc for doc, _ in reranked[: self.final_k]]

def build_hybrid_retriever(
    bm25_retriever: BaseRetriever,
    vector_retriever: BaseRetriever,
) -> HybridRetriever:
    cross_encoder = CrossEncoder(settings.cross_encoder_model)
    return HybridRetriever(
        bm25_retriever=bm25_retriever,
        vector_retriever=vector_retriever,
        cross_encoder=cross_encoder,
    )

Demo script with checkpoint outputs

# main.py
from pathlib import Path
from src.loaders import load_products
from src.bm25_retriever import build_bm25_retriever
from src.vector_retriever import build_vector_retriever
from src.hybrid_retriever import build_hybrid_retriever

def pretty_print(docs, label: str):
    print(f"\n=== {label} ===")
    for i, doc in enumerate(docs, 1):
        meta = doc.metadata
        print(f"{i}. [{meta['product_id']}] {meta['title']} — ${meta['price']} "
              f"({'in stock' if meta['in_stock'] else 'OOS'})")
        print(f"    {doc.page_content[:120]}...")

if __name__ == "__main__":
    data_path = Path("data/products.jsonl")
    documents = load_products(data_path)
    print(f"Loaded {len(documents)} products")

    bm25 = build_bm25_retriever(documents)
    vector = build_vector_retriever(documents)
    hybrid = build_hybrid_retriever(bm25, vector)

    queries = [
        "sony noise cancelling headphones",
        "mechanical keyboard rgb hot swap",
        "organic cotton shirt sustainable",
        "wireless audio travel",
    ]

    for q in queries:
        print(f"\n{'='*60}")
        print(f"QUERY: {q}")
        print(f"{'='*60}")

        bm25_results = bm25.invoke(q)[:5]
        vector_results = vector.invoke(q)[:5]
        hybrid_results = hybrid.invoke(q)

        pretty_print(bm25_results, "BM25 top-5")
        pretty_print(vector_results, "Vector top-5")
        pretty_print(hybrid_results, "Hybrid (RRF + rerank) top-10")

Run it:

python main.py

Expected output (truncated for brevity):

Loaded 87 products

============================================================
QUERY: sony noise cancelling headphones
============================================================

=== BM25 top-5 ===
1. [SKU-1001] Wireless Noise-Cancelling Headphones — $249.99 (in stock)
    Wireless Noise-Cancelling Headphones Over-ear ANC headphones with 30hr battery life Sony Electronics > Audio wireless bluetooth anc travel...
2. [SKU-1015] Sony WH-1000XM5 Headphones — $349.99 (in stock)
    Sony WH-1000XM5 Headphones Industry-leading noise cancellation 30hr battery...
3. [SKU-1022] Sony WF-1000XM4 Earbuds — $199.99 (in stock)
    Sony WF-1000XM4 Earbuds True wireless noise cancelling earbuds...

=== Vector top-5 ===
1. [SKU-1001] Wireless Noise-Cancelling Headphones — $249.99 (in stock)
    Wireless Noise-Cancelling Headphones Over-ear ANC headphones with 30hr battery life Sony Electronics > Audio wireless bluetooth anc travel...
2. [SKU-1015] Sony WH-1000XM5 Headphones — $349.99 (in stock)
    Sony WH-1000XM5 Headphones Industry-leading noise cancellation 30hr battery...
3. [SKU-1033] Bose QuietComfort 45 — $279.00 (in stock)
    Bose QuietComfort 45 Over-ear noise cancelling headphones 24hr battery...

=== Hybrid (RRF + rerank) top-10 ===
1. [SKU-1001] Wireless Noise-Cancelling Headphones — $249.99 (in stock)
2. [SKU-1015] Sony WH-1000XM5 Headphones — $349.99 (in stock)
3. [SKU-1033] Bose QuietComfort 45 — $279.00 (in stock)
4. [SKU-1022] Sony WF-1000XM4 Earbuds — $199.99 (in stock)
5. [SKU-1041] Apple AirPods Max — $549.00 (in stock)
...

Notice how BM25 surfaces exact-match wins on “sony”, vector brings in Bose/Apple as semantic neighbors, and the cross-encoder reranks by true relevance to the query intent.

Evaluation harness

Quick sanity check: mean reciprocal rank (MRR) and recall@k against a small labeled set.

# src/evaluate.py
from dataclasses import dataclass
from typing import List, Dict
from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever

@dataclass
class EvalCase:
    query: str
    relevant_ids: List[str]  # product_ids considered relevant

def evaluate_retriever(
    retriever: BaseRetriever,
    eval_cases: List[EvalCase],
    k: int = 10,
) -> Dict[str, float]:
    mrr_sum = 0.0
    recall_sum = 0.0
    total = len(eval_cases)

    for case in eval_cases:
        docs = retriever.invoke(case.query)[:k]
        retrieved_ids = [d.metadata["product_id"] for d in docs]

        # MRR
        rr = 0.0
        for rank, pid in enumerate(retrieved_ids, 1):
            if pid in case.relevant_ids:
                rr = 1.0 / rank
                break
        mrr_sum += rr

        # Recall@k
        hits = len(set(retrieved_ids) & set(case.relevant_ids))
        recall_sum += hits / len(case.relevant_ids) if case.relevant_ids else 0.0

    return {
        "mrr": mrr_sum / total if total else 0.0,
        f"recall@{k}": recall_sum / total if total else 0.0,
    }

Add a few labeled cases in main.py and run:

from src.evaluate import evaluate_retriever, EvalCase

eval_cases = [
    EvalCase("sony noise cancelling headphones", ["SKU-1001", "SKU-1015"]),
    EvalCase("mechanical keyboard rgb", ["SKU-1002", "SKU-1008", "SKU-1012"]),
    EvalCase("organic cotton t-shirt", ["SKU-1003", "SKU-1025"]),
]

print("\n--- Evaluation ---")
for name, ret in [("BM25", bm25), ("Vector", vector), ("Hybrid", hybrid)]:
    metrics = evaluate_retriever(ret, eval_cases, k=10)
    print(f"{name}: MRR={metrics['mrr']:.3f}, Recall@10={metrics['recall@10']:.3f}")

Typical output:

--- Evaluation ---
BM25: MRR=0.667, Recall@10=0.778
Vector: MRR=0.556, Recall@10=0.889
Hybrid: MRR=0.889, Recall@10=1.000

Hybrid wins on both metrics because RRF preserves exact matches while adding semantic coverage, and the cross-encoder resolves ambiguity.

Production hardening

Metadata filtering at query time

The vector retriever above uses FAISS’s native filter. For other stores, push filters into the query:

# In vector_retriever.py, replace similarity_search call:
docs = self.vectorstore.similarity_search(
    query,
    k=self.k,
    filter={"in_stock": True, "category": "Electronics > Audio"},
)

Caching embeddings

Embeddings are the latency bottleneck. Cache them locally:

# In vector_retriever.py
import pickle
from pathlib import Path

CACHE_PATH = Path(".embeddings_cache.pkl")

def build_vector_retriever(documents):
    embeddings = OpenAIEmbeddings(...)
    if CACHE_PATH.exists():
        with CACHE_PATH.open("rb") as f:
            vs = pickle.load(f)
    else:
        vs = FAISS.from_documents(documents, embeddings)
        with CACHE_PATH.open("wb") as f:
            pickle.dump(vs, f)
    return InStockVectorRetriever(vectorstore=vs)

Async retrieval

Wrap both retrievers in ainvoke for concurrent execution:

async def aget_relevant_documents(self, query: str) -> List[Document]:
    bm25_task = asyncio.create_task(self.bm25_retriever.ainvoke(query))
    vector_task = asyncio.create_task(self.vector_retriever.ainvoke(query))
    bm25_docs, vector_docs = await asyncio.gather(bm25_task, vector_task)
    # ... rest of RRF + rerank (cross-encoder stays sync, batch it)

Monitoring

Log per-stage latency and result counts:

import time
import logging

logger = logging.getLogger(__name__)

def _get_relevant_documents(self, query, *, run_manager):
    t0 = time.perf_counter()
    bm25_docs = self.bm25_retriever.invoke(query)
    t1 = time.perf_counter()
    vector_docs = self.vector_retriever.invoke(query)
    t2 = time.perf_counter()
    # ... fusion + rerank
    t3 = time.perf_counter()
    logger.info(
        "hybrid_search",
        extra={
            "query": query,
            "bm25_latency_ms": (t1 - t0) * 1000,
            "vector_latency_ms": (t2 - t1) * 1000,
            "rerank_latency_ms": (t3 - t2) * 1000,
            "bm25_count": len(bm25_docs),
            "vector_count": len(vector_docs),
            "final_count": len(final_docs),
        },
    )
    return final_docs

When to tune what

Next steps

• Replace FAISS with a managed vector DB for horizontal scaling
• Add a learned sparse encoder (SPLADE) as a third retrieval signal
• Implement per-user personalization via metadata boosting
• A/B test hybrid vs. single-retriever baselines with real conversion metrics

The code in this tutorial is deliberately dependency-light — no framework lock-in. You can copy src/ into any service, swap the embedding provider, and ship.