Spaces:
Runtime error
Runtime error
| """ | |
| rag/retriever.py | |
| ---------------- | |
| Hybrid retriever combining dense (FAISS) and lexical (BM25) search via | |
| Reciprocal Rank Fusion (Cormack et al., 2009). | |
| RRF formula: | |
| score(d) = Ξ£_{r β {dense, bm25}} 1 / (k_RRF + rank_r(d)) | |
| where k_RRF=60 (empirically optimal constant from the original paper). | |
| Chunks absent from a list receive rank = candidates + 1. | |
| Two additional constraints: | |
| - Source diversity: at most max_per_source chunks from the same regulatory | |
| source ("rbi" or "sebi") in the final top-k, to support cross-document | |
| synthesis queries. | |
| - Score floor: dense-only and bm25-only modes supported for ablation | |
| (pass mode="dense" | "bm25" | "hybrid"). | |
| """ | |
| from rag.bm25_index import BM25Index | |
| from rag.embeddings import BGEEmbedder | |
| from rag.index import FAISSIndex | |
| from rag.models import ChunkRecord, RetrievalResult | |
| class HybridRetriever: | |
| def __init__( | |
| self, | |
| faiss_index: FAISSIndex, | |
| bm25_index: BM25Index, | |
| embedder: BGEEmbedder, | |
| top_k: int = 5, | |
| candidates: int = 20, | |
| rrf_k: int = 60, | |
| max_per_source: int = 3, | |
| ) -> None: | |
| self._faiss = faiss_index | |
| self._bm25 = bm25_index | |
| self._embedder = embedder | |
| self.top_k = top_k | |
| self.candidates = candidates | |
| self.rrf_k = rrf_k | |
| self.max_per_source = max_per_source | |
| def retrieve( | |
| self, query: str, mode: str = "hybrid" | |
| ) -> list[RetrievalResult]: | |
| """ | |
| mode: "hybrid" | "dense" | "bm25" | |
| Used in Phase 3 ablation to isolate individual retriever contributions. | |
| """ | |
| absent_rank = self.candidates + 1 | |
| # ββ Dense retrieval βββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| dense_hits: list[tuple[ChunkRecord, float]] = [] | |
| if mode in ("hybrid", "dense"): | |
| qemb = self._embedder.encode_query(query) | |
| dense_hits = self._faiss.search(qemb, self.candidates) | |
| # ββ BM25 retrieval ββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| bm25_hits: list[tuple[ChunkRecord, float]] = [] | |
| if mode in ("hybrid", "bm25"): | |
| bm25_hits = self._bm25.search(query, self.candidates) | |
| # ββ Build rank & score lookup maps ββββββββββββββββββββββββββββββββββββ | |
| dense_rank = {c.chunk_id: r for r, (c, _) in enumerate(dense_hits, 1)} | |
| bm25_rank = {c.chunk_id: r for r, (c, _) in enumerate(bm25_hits, 1)} | |
| dense_score = {c.chunk_id: s for c, s in dense_hits} | |
| bm25_score = {c.chunk_id: s for c, s in bm25_hits} | |
| chunk_map = {c.chunk_id: c for c, _ in dense_hits + bm25_hits} | |
| # ββ RRF scoring βββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| rrf_scores: dict[str, float] = {} | |
| for cid in chunk_map: | |
| rd = dense_rank.get(cid, absent_rank) | |
| rb = bm25_rank.get(cid, absent_rank) | |
| if mode == "dense": | |
| rrf_scores[cid] = 1.0 / (self.rrf_k + rd) | |
| elif mode == "bm25": | |
| rrf_scores[cid] = 1.0 / (self.rrf_k + rb) | |
| else: | |
| rrf_scores[cid] = 1.0 / (self.rrf_k + rd) + 1.0 / (self.rrf_k + rb) | |
| ranked = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True) | |
| # ββ Source diversity cap + top-k selection ββββββββββββββββββββββββββββ | |
| results: list[RetrievalResult] = [] | |
| source_count: dict[str, int] = {} | |
| for cid, rrf in ranked: | |
| if len(results) >= self.top_k: | |
| break | |
| chunk = chunk_map[cid] | |
| n_from = source_count.get(chunk.source, 0) | |
| if n_from >= self.max_per_source: | |
| continue | |
| source_count[chunk.source] = n_from + 1 | |
| results.append(RetrievalResult( | |
| chunk = chunk, | |
| dense_score = dense_score.get(cid, 0.0), | |
| bm25_score = bm25_score.get(cid, 0.0), | |
| rrf_score = rrf, | |
| dense_rank = dense_rank.get(cid, absent_rank), | |
| bm25_rank = bm25_rank.get(cid, absent_rank), | |
| )) | |
| return results | |