Reranker

nlql.rerank —— 两段式检索的第二段：可插拔 Reranker 协议。召回过取候选后，对每个 (query, passage) 联合精排。内置 FakeReranker 与 CrossEncoderReranker。

rerank

Second-stage rerankers: refine coarse vector recall with precise (query, passage) scoring.

__all__ module-attribute

__all__ = ['Reranker', 'FakeReranker', 'CrossEncoderReranker']

FakeReranker

Deterministic offline reranker: scores by query/passage token overlap.

A stand-in for a cross-encoder — it re-scores each candidate by how many query tokens the passage contains (independent of the embedding), so tests can assert that reranking reorders results. No model, no network.

rerank

rerank(query: str, units: Sequence[Unit]) -> list[Unit]

源代码位于： src/nlql/rerank/base.py

def rerank(self, query: str, units: Sequence[Unit]) -> list[Unit]:
    query_tokens = set(query.lower().split())
    denom = len(query_tokens) or 1
    scored = list(units)
    for unit in scored:
        passage_tokens = set(unit.content.lower().split())
        unit.scores["rerank"] = len(query_tokens & passage_tokens) / denom
    scored.sort(key=lambda u: u.scores.get("rerank", 0.0), reverse=True)
    return scored

Reranker

Bases: Protocol

Re-scores and reorders candidate units against a query.

rerank

rerank(query: str, units: Sequence[Unit]) -> list[Unit]

Return units reordered best-first; should set unit.scores['rerank'].

源代码位于： src/nlql/rerank/base.py

def rerank(self, query: str, units: Sequence[Unit]) -> list[Unit]:
    """Return ``units`` reordered best-first; should set ``unit.scores['rerank']``."""
    ...

CrossEncoderReranker

CrossEncoderReranker(model: str = 'cross-encoder/ms-marco-MiniLM-L-6-v2')

Reranks candidates with a sentence-transformers CrossEncoder model.

源代码位于： src/nlql/rerank/cross_encoder.py

def __init__(self, model: str = "cross-encoder/ms-marco-MiniLM-L-6-v2") -> None:
    self._model_name = model
    self._model: Any = None

rerank

rerank(query: str, units: Sequence[Unit]) -> list[Unit]

源代码位于： src/nlql/rerank/cross_encoder.py

def rerank(self, query: str, units: Sequence[Unit]) -> list[Unit]:
    items = list(units)
    if not items:
        return items
    self._ensure_loaded()
    scores = self._model.predict([(query, u.content) for u in items])
    for unit, score in zip(items, scores, strict=True):
        unit.scores["rerank"] = float(score)
    items.sort(key=lambda u: u.scores.get("rerank", 0.0), reverse=True)
    return items