| """Cross-encoder reranking via BAAI/bge-reranker-v2-m3. |
| |
| Dense retrieval is great for recall but ranks by a single vector dot-product. A |
| cross-encoder reads (query, passage) **together** and scores true relevance — the single |
| biggest precision lever for cross-lingual search. We only score the top ~80 candidates, |
| so it's cheap. ``normalize=True`` squashes scores to 0..1 (sigmoid) for nice display. |
| """ |
| from __future__ import annotations |
|
|
| from typing import List, Optional |
|
|
| from app.config import Config, get_config |
|
|
|
|
| class Reranker: |
| def __init__(self, cfg: Optional[Config] = None): |
| self.cfg = cfg or get_config() |
| self._model = None |
|
|
| @property |
| def enabled(self) -> bool: |
| return bool(self.cfg.reranker["enabled"]) |
|
|
| def _load(self): |
| if self._model is not None: |
| return self._model |
| from FlagEmbedding import FlagReranker |
|
|
| r = self.cfg.reranker |
| self._model = FlagReranker(r["model"], use_fp16=bool(r["use_fp16"])) |
| return self._model |
|
|
| def scores(self, query: str, passages: List[str]) -> List[float]: |
| """Relevance score (0..1) for each passage against the query.""" |
| if not passages: |
| return [] |
| model = self._load() |
| pairs = [[query, p] for p in passages] |
| out = model.compute_score( |
| pairs, batch_size=self.cfg.reranker["batch_size"], normalize=True |
| ) |
| if isinstance(out, (int, float)): |
| return [float(out)] |
| return [float(x) for x in out] |
|
|