AI Engineer Portal

Your personal operating system for career transition.

Private mode

Exercise

Build a hybrid retrieval pipeline

Build a Hybrid Retrieval Pipeline

Pure vector search misses documents with specific entity names, codes, or technical identifiers. Pure keyword search misses paraphrases and synonyms. Hybrid search combines both and consistently outperforms either alone across real-world query distributions.

The standard approach is Reciprocal Rank Fusion (RRF): each method ranks candidate documents independently, then RRF merges the rankings using score(doc) = sum(1 / (k + rank)) across all lists.

What you are building

Create a HybridRetriever class that:

Stores documents in two parallel indices: a vector index (cosine similarity) and a keyword index (TF-IDF term scoring).
Retrieves candidates from both indices independently for a given query, returning the top-N from each.
Merges ranked lists using Reciprocal Rank Fusion.
Returns the top-k results by final RRF score with source attribution (which method(s) found each document).
Supports metadata filtering applied after retrieval but before merging.

Why this matters

Every serious production RAG system uses hybrid retrieval. Understanding RRF means you can tune the constant k (controls how much top positions are rewarded) and extend to three or more retrieval methods.

Constraints

Use only the Python standard library. Implement TF-IDF scoring directly.
Type-hint everything. Return list[RetrievedDoc] with score and source_methods fields.
The embedding function should be injectable (for testing with mocks).

Rag Systems / medium / Step 2 of 6

Practice stage

General drill

Hint

Keep the solution explicit and reviewable.

Next drill

Add reranking to retrieval results

Success criteria

Make the solution explicit, debuggable, and easy to explain.

Review checklist

Review where the boundary is, what gets validated, and what would be hard to debug later.

from __future__ import annotations

import math
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Callable, Optional

@dataclass
class Document:
    id: str
    text: str
    metadata: dict = field(default_factory=dict)

@dataclass
class RetrievedDoc:
    id: str
    text: str
    metadata: dict
    rrf_score: float
    source_methods: list[str]  # e.g. ["vector", "keyword"]

def cosine_similarity(a: list[float], b: list[float]) -> float:
    # TODO: implement
    raise NotImplementedError

class KeywordIndex:
    """TF-IDF style keyword index."""

def __init__(self):
        self._docs: dict[str, Document] = {}
        self._term_df: dict[str, int] = defaultdict(int)

def add(self, doc: Document) -> None:
        # TODO: store doc, update term document frequencies
        raise NotImplementedError

def search(self, query: str, top_k: int = 20) -> list[tuple[str, float]]:
        """Return (doc_id, score) pairs sorted by TF-IDF score descending."""
        # TODO: tokenize query, score each doc by sum of TF-IDF for query terms
        raise NotImplementedError

class VectorIndex:
    def __init__(self, embed_fn: Callable[[str], list[float]]):
        self.embed_fn = embed_fn
        self._docs: dict[str, Document] = {}
        self._embeddings: dict[str, list[float]] = {}

def add(self, doc: Document) -> None:
        # TODO: store doc and its embedding
        raise NotImplementedError

def search(self, query: str, top_k: int = 20) -> list[tuple[str, float]]:
        """Return (doc_id, score) pairs sorted by cosine similarity descending."""
        # TODO: embed query, compute cosine similarity against all stored embeddings
        raise NotImplementedError

class HybridRetriever:
    def __init__(self, embed_fn: Callable[[str], list[float]]):
        self.vector_index = VectorIndex(embed_fn)
        self.keyword_index = KeywordIndex()
        self._docs: dict[str, Document] = {}

def add_documents(self, docs: list[Document]) -> None:
        # TODO: add to both indices
        raise NotImplementedError

def retrieve(
        self,
        query: str,
        top_k: int = 5,
        candidate_k: int = 20,
        rrf_k: int = 60,
        where: Optional[dict] = None,
    ) -> list[RetrievedDoc]:
        """Retrieve using vector + keyword search, merge with RRF."""
        # TODO: get candidates from both indices
        # TODO: apply metadata filter if provided
        # TODO: apply RRF fusion
        # TODO: return top_k RetrievedDoc objects
        raise NotImplementedError

Practice

Generate a variation

Generate a new exercise variation to deepen understanding or practice a related concept.

Attempt history

Recent submissions

Before you submit, decide what a strong answer should make obvious to the reviewer.

No attempts yet.