week_4: Module C (The Librarian) — C.2 cross-encoder reranker

application/cmd/cre_main.py

@@ -1073,10 +1073,10 @@ def run_librarian(
 
     For each knowledge-queue section: try the deterministic explicit-CRE fast
     path (C.0.5); on no/ambiguous reference, run the semantic retriever (C.1)
-    and log the top-K candidate shortlist. The cross-encoder rerank (C.2, W4),
-    decision/threshold routing (C.3-C.4, W5) and graph writes (W8) are not
-    built yet, so this is dry-run only: it never writes a link. ``--run_librarian``
-    without writes behaves identically and warns.
+    and rerank its top-K shortlist with the cross-encoder (C.2, W4), then log
+    the reranked candidates. Decision/threshold routing (C.3-C.4, W5) and graph
+    writes (W8) are not built yet, so this is dry-run only: it never writes a
+    link. ``--run_librarian`` without writes behaves identically and warns.
 
     Ops note: this is opt-in CLI only — it is not on the ``Procfile`` and is
     wired into neither the web app nor the background worker (that lands W8).
@@ -1089,6 +1089,10 @@ def run_librarian(
         build_retriever,
     )
     from application.utils.librarian.config_loader import load_config
+    from application.utils.librarian.cross_encoder import (
+        CrossEncoderReranker,
+        build_cross_encoder_score_fn,
+    )
     from application.utils.librarian.explicit_link_resolver import (
         ResolutionOutcome,
         resolve,
@@ -1141,6 +1145,16 @@ def run_librarian(
         connection=database.session.connection(),
     )
 
+    # C.2 reranker: reads each (section, candidate-CRE) pair together and
+    # re-sorts C.1's shortlist. Scored against the CRE's embeddings_content —
+    # the same text the hub vectors were built from.
+    cre_texts = database.get_embedding_contents_by_doc_type(defs.Credoctypes.CRE.value)
+    reranker = CrossEncoderReranker(
+        score_fn=build_cross_encoder_score_fn(cfg.crossencoder_model),
+        top_n=cfg.top_k_rerank,
+        cre_texts=cre_texts,
+    )
+
     source = FixtureKnowledgeSource(source_jsonl or _DEFAULT_LIBRARIAN_SOURCE)
     sections = explicit = semantic = rejected = 0
     for item in source.items():
@@ -1158,15 +1172,26 @@ def run_librarian(
             logger.info("[explicit] %s -> %s", section.chunk_id, resolution.cre_ids[0])
             continue
 
+        try:
+            audit = retriever.retrieve(section.text)
+            audit = reranker.rerank(section.text, audit)
+        # one bad section must not abort the batch
+        except Exception as exc:  # noqa: BLE001
+            rejected += 1
+            logger.warning(
+                "[semantic] %s skipped: retrieval/rerank failed: %s",
+                section.chunk_id,
+                exc,
+            )
+            continue
+
         semantic += 1
-        audit = retriever.retrieve(section.text)
-        top = ", ".join(
-            f"{c.cre_id}:{c.score_vector:.3f}" for c in audit.candidates[:5]
-        )
+        top = ", ".join(f"{c.cre_id}:{c.score_rerank:.3f}" for c in audit.reranked)
         logger.info(
-            "[semantic] %s -> %d candidates (top5: %s)",
+            "[semantic] %s -> %d candidates, reranked top%d: %s",
             section.chunk_id,
             len(audit.candidates),
+            len(audit.reranked),
             top or "<none>",
         )
 

application/database/db.py

@@ -2376,6 +2376,27 @@ def get_embeddings_by_doc_type(self, doc_type: str) -> Dict[str, List[float]]:
                     res[entry.node_id] = [float(e) for e in entry.embeddings.split(",")]
         return res
 
+    def get_embedding_contents_by_doc_type(self, doc_type: str) -> Dict[str, str]:
+        """Return ``{id -> embeddings_content}`` for a doc type.
+
+        ``embeddings_content`` is the text a vector was built from — the pair
+        text the Module C.2 cross-encoder scores a candidate against. Keyed by
+        cre_id for CREs, node_id otherwise, mirroring
+        ``get_embeddings_by_doc_type``. Rows with no content are skipped.
+        """
+        res: Dict[str, str] = {}
+        embeddings = (
+            self.session.query(Embeddings).filter(Embeddings.doc_type == doc_type).all()
+        )
+        for entry in embeddings:
+            if not entry.embeddings_content:
+                continue
+            if doc_type == cre_defs.Credoctypes.CRE.value:
+                res[entry.cre_id] = entry.embeddings_content
+            else:
+                res[entry.node_id] = entry.embeddings_content
+        return res
+
     def get_embeddings_by_doc_type_paginated(
         self, doc_type: str, page: int = 1, per_page: int = 100
     ) -> Tuple[Dict[str, List[float]], int, int]:

application/tests/librarian/config_loader_test.py

@@ -1,6 +1,7 @@
 import os
 import unittest
 from dataclasses import FrozenInstanceError
+from typing import ClassVar, Dict
 from unittest import mock
 
 from application.utils.librarian.config_loader import LibrarianConfig, load_config
@@ -27,7 +28,7 @@ def test_config_is_frozen(self):
 
 
 class TestConfigLoaderOverrides(unittest.TestCase):
-    OVERRIDES = {
+    OVERRIDES: ClassVar[Dict[str, str]] = {
         "CRE_LIBRARIAN_RETRIEVER_BACKEND": "pgvector",
         "CRE_LIBRARIAN_CROSSENCODER_MODEL": "cross-encoder/other",
         "CRE_LIBRARIAN_TOP_K_RETRIEVAL": "50",

application/tests/librarian/cross_encoder_test.py

@@ -0,0 +1,119 @@
+"""Tests for the C.2 cross-encoder reranker (Week 4).
+
+Hermetic: the cross-encoder score function and the CRE text map are injected as
+controlled values, so re-ordering, top-N truncation, tie stability, the audit
+shape, and the failure modes are all assertable without torch, an LLM, or a DB.
+"""
+
+import unittest
+
+from application.utils.librarian.cross_encoder import (
+    CrossEncoderReranker,
+    MissingCandidateTextError,
+    RerankerError,
+)
+from application.utils.librarian.schemas import CreCandidate, RetrievalAudit
+
+# The text each candidate CRE is scored against (its embeddings_content).
+CRE_TEXTS = {
+    "111-111": "text-a",
+    "222-222": "text-b",
+    "333-333": "text-c",
+}
+
+
+def make_audit(threshold=0.8):
+    """A C.1 shortlist in cosine order: a (0.90) > b (0.80) > c (0.70)."""
+    return RetrievalAudit(
+        retriever="in-memory-cosine/0.1.0",
+        candidates=[
+            CreCandidate(cre_id="111-111", score_vector=0.90),
+            CreCandidate(cre_id="222-222", score_vector=0.80),
+            CreCandidate(cre_id="333-333", score_vector=0.70),
+        ],
+        reranked=[],
+        threshold=threshold,
+    )
+
+
+# The cross-encoder disagrees with cosine: it likes b most, then c, then a.
+_PAIR_SCORES = {
+    ("q", "text-a"): 0.10,
+    ("q", "text-b"): 0.90,
+    ("q", "text-c"): 0.50,
+}
+
+
+def fake_score(pairs):
+    return [_PAIR_SCORES[(q, cand)] for q, cand in pairs]
+
+
+def make_reranker(top_n=5, cre_texts=None, score_fn=fake_score):
+    return CrossEncoderReranker(
+        score_fn=score_fn,
+        top_n=top_n,
+        cre_texts=CRE_TEXTS if cre_texts is None else cre_texts,
+    )
+
+
+class RerankTest(unittest.TestCase):
+    def test_reranks_by_cross_encoder_score_not_cosine(self) -> None:
+        audit = make_reranker().rerank("q", make_audit())
+        # Cosine order was a,b,c; the cross-encoder reorders to b,c,a.
+        self.assertEqual(
+            [c.cre_id for c in audit.reranked], ["222-222", "333-333", "111-111"]
+        )
+        scores = [c.score_rerank for c in audit.reranked]
+        self.assertEqual(scores, sorted(scores, reverse=True))
+        self.assertAlmostEqual(audit.reranked[0].score_rerank, 0.90)
+
+    def test_candidates_preserved_and_audit_shape(self) -> None:
+        audit = make_reranker().rerank("q", make_audit(threshold=0.8))
+        # candidates[] (the pre-rerank shortlist) is untouched, still cosine order.
+        self.assertEqual(
+            [c.cre_id for c in audit.candidates], ["111-111", "222-222", "333-333"]
+        )
+        # score_rerank is only set on reranked[], never on candidates[].
+        self.assertTrue(all(c.score_rerank is None for c in audit.candidates))
+        self.assertEqual(audit.retriever, "in-memory-cosine/0.1.0")
+        self.assertEqual(audit.threshold, 0.8)
+
+    def test_top_n_truncates_to_best(self) -> None:
+        audit = make_reranker(top_n=2).rerank("q", make_audit())
+        self.assertEqual([c.cre_id for c in audit.reranked], ["222-222", "333-333"])
+
+    def test_top_n_larger_than_shortlist_keeps_all(self) -> None:
+        audit = make_reranker(top_n=20).rerank("q", make_audit())
+        self.assertEqual(len(audit.reranked), 3)
+
+    def test_ties_preserve_cosine_order(self) -> None:
+        # Equal cross-encoder scores -> the stable sort keeps C.1's cosine order.
+        audit = make_reranker(score_fn=lambda pairs: [1.0] * len(pairs)).rerank(
+            "q", make_audit()
+        )
+        self.assertEqual(
+            [c.cre_id for c in audit.reranked], ["111-111", "222-222", "333-333"]
+        )
+
+    def test_empty_candidates_yields_empty_reranked(self) -> None:
+        empty = RetrievalAudit(retriever="r", candidates=[], reranked=[], threshold=0.8)
+        out = make_reranker().rerank("q", empty)
+        self.assertEqual(out.reranked, [])
+
+
+class FailureModeTest(unittest.TestCase):
+    def test_missing_cre_text_raises(self) -> None:
+        with self.assertRaises(MissingCandidateTextError):
+            make_reranker(cre_texts={"111-111": "text-a"}).rerank("q", make_audit())
+
+    def test_non_positive_top_n_rejected(self) -> None:
+        with self.assertRaises(RerankerError):
+            make_reranker(top_n=0)
+
+    def test_score_count_mismatch_rejected(self) -> None:
+        with self.assertRaises(RerankerError):
+            make_reranker(score_fn=lambda pairs: [0.5]).rerank("q", make_audit())
+
+
+if __name__ == "__main__":
+    unittest.main()

application/utils/librarian/__init__.py

@@ -15,7 +15,8 @@
              re-normalizing text) and ExplicitLinkResolver (fail-safe
              explicit-link resolution).
   W3 (C.1):  candidate retriever (in-memory + pgvector) + pipeline switch.
-Cross-encoder rerank (C.2, W4) onward is not built yet.
+  W4 (C.2):  cross-encoder reranker — re-sorts the C.1 shortlist, fills reranked[].
+Calibration + decision routing (C.3-C.4, W5-W6) onward is not built yet.
 
 Vendored RFC JSON schemas live under ``_rfc_schemas/``. They are pinned to
 upstream/owasp-graph @ 2b1437987768d5ed20fe9ee721ab9a898c4b84af (PR #734).

application/utils/librarian/candidate_retriever.py

@@ -157,7 +157,9 @@ def retrieve(self, text: str) -> RetrievalAudit:
         # Top-K by descending score. argsort is ascending, so take the tail
         # and reverse; cap at pool size when the hub is smaller than K.
         k = min(self._top_k, len(self._pool.cre_ids))
-        top_idx = np.argsort(scores)[-k:][::-1]
+        # Stable descending sort so tied cosine scores keep hub (index) order —
+        # deterministic across runs. argsort(-scores) descends; kind="stable".
+        top_idx = np.argsort(-scores, kind="stable")[:k]
 
         candidates: List[CreCandidate] = [
             CreCandidate(

application/utils/librarian/cross_encoder.py

@@ -0,0 +1,128 @@
+"""Module C.2 — cross-encoder reranker (Week 4). The careful re-reader.
+
+C.1 (the bi-encoder, W3) fingerprints the section and each CRE *separately* and
+cosine-ranks the whole hub — fast enough to scan every CRE, but it never reads a
+section and a candidate *together*, so the ordering inside the top-K shortlist is
+rough: the right CRE can sit at #7, not #1. C.2 fixes that ordering. It reads
+each ``(section text, candidate CRE text)`` *pair* together as one input, scores
+"do these two actually match?", re-sorts the shortlist by that score, and keeps
+the best N. Slow per pair, so it runs only over the K candidates C.1 already
+narrowed to — never the whole hub.
+
+Like C.1, the reranker is a thin dependency-injected seam over its model:
+
+  - ``score_fn(pairs) -> Sequence[float]`` — scores a batch of
+    ``(query_text, candidate_text)`` pairs, higher = better match. Prod wires a
+    pinned cross-encoder (``ms-marco-MiniLM-L-6-v2``); the harness and tests
+    inject a deterministic stub. C.2 never imports the model directly, so it
+    stays import-light and hermetically testable (mirrors C.1's ``embed_fn``).
+
+The text a candidate CRE is scored against is its ``embeddings_content`` — the
+same signal the hub vectors were built from — supplied as a ``{cre_id -> text}``
+map. The RFC is silent on ranking tech; it mandates only the
+``candidates[]``/``reranked[]`` audit trail. C.2 fills ``reranked[]`` (the slot
+C.1 deliberately left empty), populating ``score_rerank`` and re-ordering, while
+leaving ``candidates[]`` untouched so the pre-rerank shortlist stays auditable.
+"""
+
+from typing import Callable, List, Mapping, Sequence, Tuple
+
+from application.utils.librarian.schemas import RetrievalAudit
+
+# A function that scores a batch of (query_text, candidate_text) pairs.
+RerankFn = Callable[[Sequence[Tuple[str, str]]], Sequence[float]]
+
+# Identify the reranker in the RFC audit trail. Bumped when the model or the
+# scoring changes so a stored proposal is traceable to the code that ranked it.
+RERANKER_NAME = "cross-encoder-ms-marco-MiniLM-L-6-v2/0.1.0"
+DEFAULT_CROSSENCODER_MODEL = "cross-encoder/ms-marco-MiniLM-L-6-v2"
+
+
+class RerankerError(ValueError):
+    """Base class for reranker construction/usage failures."""
+
+
+class MissingCandidateTextError(RerankerError):
+    """A shortlisted CRE has no text to score against — cannot rerank fairly."""
+
+
+class CrossEncoderReranker:
+    """Re-score a C.1 shortlist by reading each (section, CRE) pair together.
+
+    ``top_n`` is ``CRE_LIBRARIAN_TOP_K_RERANK`` (default 5). ``rerank`` reads
+    ``audit.candidates`` (C.1's shortlist), scores every pair, re-sorts by the
+    cross-encoder score, keeps the best ``top_n``, and returns a copy of the
+    audit with ``reranked`` filled and ``candidates`` preserved.
+    """
+
+    def __init__(
+        self,
+        score_fn: RerankFn,
+        top_n: int,
+        *,
+        cre_texts: Mapping[str, str],
+    ) -> None:
+        if top_n <= 0:
+            raise RerankerError(f"top_n must be > 0, got {top_n}")
+        self._score_fn = score_fn
+        self._top_n = top_n
+        self._cre_texts = dict(cre_texts)
+
+    def rerank(self, text: str, audit: RetrievalAudit) -> RetrievalAudit:
+        """Return a copy of ``audit`` with ``reranked`` filled from ``candidates``.
+
+        Raises ``MissingCandidateTextError`` if a shortlisted CRE has no text to
+        score (a silent-quality trap otherwise) and ``RerankerError`` if the
+        model returns the wrong number of scores.
+        """
+        candidates = audit.candidates
+        if not candidates:
+            # Nothing to rerank (e.g. an empty hub upstream); keep the audit
+            # shape consistent — an explicit, empty reranked list.
+            return audit.model_copy(update={"reranked": []})
+
+        pairs: List[Tuple[str, str]] = []
+        for c in candidates:
+            cre_text = self._cre_texts.get(c.cre_id)
+            if not cre_text:
+                raise MissingCandidateTextError(
+                    f"no text for candidate CRE {c.cre_id!r}; the reranker needs "
+                    "the CRE's embeddings_content to score the pair"
+                )
+            pairs.append((text, cre_text))
+
+        scores = list(self._score_fn(pairs))
+        if len(scores) != len(candidates):
+            raise RerankerError(
+                f"score_fn returned {len(scores)} scores for {len(candidates)} "
+                "candidates; the reranker expects exactly one score per pair"
+            )
+
+        reranked = [
+            c.model_copy(update={"score_rerank": float(s)})
+            for c, s in zip(candidates, scores, strict=True)
+        ]
+        # Highest cross-encoder score first, then keep only the best top_n.
+        # Python's sort is stable, so ties preserve C.1's cosine order.
+        reranked.sort(key=lambda c: c.score_rerank, reverse=True)
+        return audit.model_copy(update={"reranked": reranked[: self._top_n]})
+
+
+def build_cross_encoder_score_fn(
+    model_name: str = DEFAULT_CROSSENCODER_MODEL,
+) -> RerankFn:
+    """Load a sentence-transformers CrossEncoder and adapt it to ``RerankFn``.
+
+    ``sentence_transformers`` (and torch) is imported lazily so this module —
+    and CI/tests that inject a stub score_fn — never need the heavy ML stack
+    loaded (mirrors C.1 keeping the embedding model out of module import).
+    """
+    from sentence_transformers import CrossEncoder  # lazy, heavy
+
+    model = CrossEncoder(model_name)
+
+    def score_fn(pairs: Sequence[Tuple[str, str]]) -> List[float]:
+        # CrossEncoder.predict takes a list of [a, b] pairs, returns an ndarray.
+        return [float(s) for s in model.predict([list(p) for p in pairs])]
+
+    return score_fn