reach.retrieval

Lexical, dense embedding, and hybrid retrieval scorers for skill selection modeling.

Provide dense semantic, sparse lexical, and hybrid retrieval scorers.

Bm25Scorer

Bases: BaseModel

Score texts and skill descriptions using Lucene-variant BM25.

Source code in src/reach/retrieval.py

class Bm25Scorer(BaseModel):
    """Score texts and skill descriptions using Lucene-variant BM25."""

    model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True)

    documents: dict[str, tuple[str, ...]]
    k1: float = Field(default=K1, gt=0)
    b: float = Field(default=B, ge=0, le=1)

    _tables: _CorpusTables = PrivateAttr()

    @override
    def model_post_init(self, _context: object) -> None:
        """Derive cached corpus tables after initialization."""
        object.__setattr__(self, "_tables", _CorpusTables.of(self.documents))

    @classmethod
    def from_skills(
        cls,
        skills: Sequence[Skill],
        k1: float = K1,
        b: float = B,
    ) -> Bm25Scorer:
        """Instantiate a Bm25Scorer from a sequence of Skill models."""
        return cls(
            documents={s.name: tuple(tokenize(skill_text(s))) for s in skills},
            k1=k1,
            b=b,
        )

    @property
    def average_length(self) -> float:
        """Return the average document token length in the corpus."""
        return self._tables.average_length

    def idf(self, term: str) -> float:
        """Calculate the non-negative Lucene IDF for a term."""
        cached = self._tables.idf.get(term)
        if cached is not None:
            return cached
        df = self._tables.document_frequency[term]
        return _lucene_idf(len(self.documents), df)

    def score(self, query: Sequence[str], name: str) -> float:
        """Compute the Lucene BM25 score of a document against a tokenized query."""
        tokens = self.documents.get(name)
        if not tokens:
            return 0.0
        tables = self._tables
        counts = tables.term_frequency[name]
        idf = tables.idf
        norm = self.k1 * (1 - self.b + self.b * len(tokens) / tables.average_length)
        total = 0.0
        for term in query:
            tf = counts.get(term, 0)
            if tf:
                total += idf[term] * tf / (tf + norm)
        return total

    def contributions(self, query: Sequence[str], name: str) -> dict[str, float]:
        """Itemize BM25 score contributions for each matching query term."""
        tokens = self.documents.get(name)
        if not tokens:
            return {}
        tables = self._tables
        counts = tables.term_frequency[name]
        idf = tables.idf
        norm = self.k1 * (1 - self.b + self.b * len(tokens) / tables.average_length)
        asked = Counter(query)
        return {
            term: repeats * idf[term] * tf / (tf + norm)
            for term, repeats in asked.items()
            if (tf := counts.get(term, 0))
        }

    def rank_text(
        self,
        text: str,
        candidates: Sequence[Skill],
    ) -> list[tuple[str, float]]:
        """Rank candidate skills against query text, returning (name, score) pairs."""
        query = tokenize(text)
        if not query or not candidates:
            return sorted(((c.name, 0.0) for c in candidates), key=lambda pair: pair[0])

        matching_names = _matching_candidate_names(
            query,
            {c.name for c in candidates},
            self._tables.postings,
        )
        scored = [
            (c.name, self.score(query, c.name) if c.name in matching_names else 0.0)
            for c in candidates
        ]
        return sorted(scored, key=lambda pair: (-pair[1], pair[0]))

    def rank(
        self,
        target: Skill,
        candidates: Sequence[Skill],
    ) -> list[tuple[str, float]]:
        """Rank candidates against target skill vocabulary, excluding target itself."""
        return self.rank_text(
            skill_text(target),
            [c for c in candidates if c.name != target.name],
        )

average_length property

average_length: float

Return the average document token length in the corpus.

contributions

contributions(
    query: Sequence[str], name: str
) -> dict[str, float]

Itemize BM25 score contributions for each matching query term.

Source code in src/reach/retrieval.py

def contributions(self, query: Sequence[str], name: str) -> dict[str, float]:
    """Itemize BM25 score contributions for each matching query term."""
    tokens = self.documents.get(name)
    if not tokens:
        return {}
    tables = self._tables
    counts = tables.term_frequency[name]
    idf = tables.idf
    norm = self.k1 * (1 - self.b + self.b * len(tokens) / tables.average_length)
    asked = Counter(query)
    return {
        term: repeats * idf[term] * tf / (tf + norm)
        for term, repeats in asked.items()
        if (tf := counts.get(term, 0))
    }

from_skills classmethod

from_skills(
    skills: Sequence[Skill], k1: float = K1, b: float = B
) -> Bm25Scorer

Instantiate a Bm25Scorer from a sequence of Skill models.

Source code in src/reach/retrieval.py

@classmethod
def from_skills(
    cls,
    skills: Sequence[Skill],
    k1: float = K1,
    b: float = B,
) -> Bm25Scorer:
    """Instantiate a Bm25Scorer from a sequence of Skill models."""
    return cls(
        documents={s.name: tuple(tokenize(skill_text(s))) for s in skills},
        k1=k1,
        b=b,
    )

idf

idf(term: str) -> float

Calculate the non-negative Lucene IDF for a term.

Source code in src/reach/retrieval.py

def idf(self, term: str) -> float:
    """Calculate the non-negative Lucene IDF for a term."""
    cached = self._tables.idf.get(term)
    if cached is not None:
        return cached
    df = self._tables.document_frequency[term]
    return _lucene_idf(len(self.documents), df)

model_post_init

model_post_init(_context: object) -> None

Derive cached corpus tables after initialization.

Source code in src/reach/retrieval.py

@override
def model_post_init(self, _context: object) -> None:
    """Derive cached corpus tables after initialization."""
    object.__setattr__(self, "_tables", _CorpusTables.of(self.documents))

rank

rank(
    target: Skill, candidates: Sequence[Skill]
) -> list[tuple[str, float]]

Rank candidates against target skill vocabulary, excluding target itself.

Source code in src/reach/retrieval.py

def rank(
    self,
    target: Skill,
    candidates: Sequence[Skill],
) -> list[tuple[str, float]]:
    """Rank candidates against target skill vocabulary, excluding target itself."""
    return self.rank_text(
        skill_text(target),
        [c for c in candidates if c.name != target.name],
    )

rank_text

rank_text(
    text: str, candidates: Sequence[Skill]
) -> list[tuple[str, float]]

Rank candidate skills against query text, returning (name, score) pairs.

Source code in src/reach/retrieval.py

def rank_text(
    self,
    text: str,
    candidates: Sequence[Skill],
) -> list[tuple[str, float]]:
    """Rank candidate skills against query text, returning (name, score) pairs."""
    query = tokenize(text)
    if not query or not candidates:
        return sorted(((c.name, 0.0) for c in candidates), key=lambda pair: pair[0])

    matching_names = _matching_candidate_names(
        query,
        {c.name for c in candidates},
        self._tables.postings,
    )
    scored = [
        (c.name, self.score(query, c.name) if c.name in matching_names else 0.0)
        for c in candidates
    ]
    return sorted(scored, key=lambda pair: (-pair[1], pair[0]))

score

score(query: Sequence[str], name: str) -> float

Compute the Lucene BM25 score of a document against a tokenized query.

Source code in src/reach/retrieval.py

def score(self, query: Sequence[str], name: str) -> float:
    """Compute the Lucene BM25 score of a document against a tokenized query."""
    tokens = self.documents.get(name)
    if not tokens:
        return 0.0
    tables = self._tables
    counts = tables.term_frequency[name]
    idf = tables.idf
    norm = self.k1 * (1 - self.b + self.b * len(tokens) / tables.average_length)
    total = 0.0
    for term in query:
        tf = counts.get(term, 0)
        if tf:
            total += idf[term] * tf / (tf + norm)
    return total

DenseScorer

Bases: BaseModel

Score skills using dense semantic embeddings.

Source code in src/reach/retrieval.py

class DenseScorer(BaseModel):
    """Score skills using dense semantic embeddings."""

    model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True)

    vectors: dict[str, EmbeddingVector] = Field(default_factory=dict)
    model_name: str = DEFAULT_RETRIEVAL_MODEL
    mode: str = "cosine"

    _unit_vectors: dict[str, EmbeddingVector] = PrivateAttr(default_factory=dict)
    _text_vectors: dict[str, list[float]] = PrivateAttr(default_factory=dict)

    @override
    def model_post_init(self, _context: object) -> None:
        """Compute and cache normalized unit vectors after initialization."""
        object.__setattr__(
            self,
            "_unit_vectors",
            {name: _unit_vector(vec) for name, vec in self.vectors.items()},
        )
        object.__setattr__(self, "_text_vectors", {})

    @classmethod
    def from_skills(
        cls,
        skills: Sequence[Skill],
        model_name: str | None = None,
        mode: str = "cosine",
    ) -> DenseScorer:
        """Embed skill texts and instantiate a DenseScorer."""
        chosen_model = model_name or DEFAULT_RETRIEVAL_MODEL
        model = _load_model2vec_model(chosen_model)

        texts = [skill_text(s) for s in skills]
        embeddings = model.encode(texts)

        vectors: dict[str, EmbeddingVector] = {}
        for skill, emb in zip(skills, embeddings, strict=False):
            vectors[skill.name] = (
                emb.tolist() if hasattr(emb, "tolist") else [float(x) for x in emb]
            )

        return cls(vectors=vectors, model_name=chosen_model, mode=mode)

    def _get_or_compute_vector(self, skill: Skill) -> EmbeddingVector:
        """Retrieve pre-computed vector or embed on demand if available."""
        vec = self.vectors.get(skill.name)
        if vec:
            return vec
        try:
            model = _load_model2vec_model(self.model_name)
            emb = model.encode([skill_text(skill)])[0]
            return emb.tolist() if hasattr(emb, "tolist") else [float(x) for x in emb]
        except (RuntimeError, ValueError, TypeError, AttributeError):
            return []

    def _get_or_compute_unit_vector(self, skill: Skill) -> EmbeddingVector:
        """Retrieve pre-computed unit vector or embed on demand."""
        unit = self._unit_vectors.get(skill.name)
        if unit is not None:
            return unit
        vec = self._get_or_compute_vector(skill)
        if not vec:
            return []
        unit_vec = _unit_vector(vec)
        self._unit_vectors[skill.name] = unit_vec
        return unit_vec

    def _score_vector_pair(self, target_vec: EmbeddingVector, cand_vec: EmbeddingVector) -> float:
        """Calculate pairwise semantic score depending on configured projection mode."""
        if not cand_vec:
            return 0.0
        if self.mode == "directional":
            return directional_projection(target_vec, cand_vec)
        return cosine_similarity(target_vec, cand_vec)

    def rank(
        self,
        target: Skill,
        candidates: Sequence[Skill],
    ) -> list[ScoredPair]:
        """Rank candidate skills against target using semantic similarity."""
        if self.mode == "directional":
            target_vec = self._get_or_compute_vector(target)
            if not target_vec:
                return [(c.name, 0.0) for c in candidates if c.name != target.name]

            scored = [
                (c.name, directional_projection(target_vec, self._get_or_compute_vector(c)))
                for c in candidates
                if c.name != target.name
            ]
            return sorted(scored, key=lambda pair: (-pair[1], pair[0]))

        target_u = self._get_or_compute_unit_vector(target)
        if not target_u:
            return [(c.name, 0.0) for c in candidates if c.name != target.name]

        scored = [
            (
                c.name,
                sum(a * b for a, b in zip(target_u, cand_u, strict=False))
                if (cand_u := self._get_or_compute_unit_vector(c))
                else 0.0,
            )
            for c in candidates
            if c.name != target.name
        ]
        return sorted(scored, key=lambda pair: (-pair[1], pair[0]))

    def pairwise_similarity(
        self,
        skills: Sequence[Skill],
    ) -> list[PairwiseSimilarity]:
        """Calculate pairwise cosine similarity for all distinct skill pairs."""
        pairs: list[PairwiseSimilarity] = []

        skill_list = list(skills)
        unit_vecs = {s.name: self._get_or_compute_unit_vector(s) for s in skill_list}

        for i in range(len(skill_list)):
            s1 = skill_list[i]
            u1 = unit_vecs.get(s1.name)
            if not u1:
                continue
            for j in range(i + 1, len(skill_list)):
                s2 = skill_list[j]
                u2 = unit_vecs.get(s2.name)
                if u2:
                    sim = sum(a * b for a, b in zip(u1, u2, strict=False))
                    pairs.append((s1.name, s2.name, sim))

        return sorted(pairs, key=lambda p: (-p[2], p[0], p[1]))

    def _get_or_compute_text_vector(self, text: str) -> list[float]:
        """Retrieve or compute the dense embedding vector for a given text string."""
        if not text:
            return []
        if text in self.vectors:
            return self.vectors[text]
        cached = self._text_vectors.get(text)
        if cached is not None:
            return cached
        try:
            model = _load_model2vec_model(self.model_name)
            query_vec = model.encode([text])[0]
            vec_list = (
                query_vec.tolist()
                if hasattr(query_vec, "tolist")
                else [float(x) for x in query_vec]
            )
            self._text_vectors[text] = vec_list
            return vec_list
        except (RuntimeError, ValueError, TypeError, AttributeError):
            return []

    def rank_text(
        self,
        text: str,
        candidates: Sequence[Skill],
    ) -> list[tuple[str, float]]:
        """Rank candidate skills against query text using semantic similarity."""
        if not text or not candidates:
            return sorted(((c.name, 0.0) for c in candidates), key=lambda pair: pair[0])

        query_vec_list = self._get_or_compute_text_vector(text)
        if not query_vec_list:
            return sorted(((c.name, 0.0) for c in candidates), key=lambda pair: pair[0])

        if self.mode == "directional":
            scored = [
                (
                    c.name,
                    directional_projection(query_vec_list, self._get_or_compute_vector(c)),
                )
                for c in candidates
            ]
            return sorted(scored, key=lambda pair: (-pair[1], pair[0]))

        query_u = _unit_vector(query_vec_list)
        scored = [
            (
                c.name,
                sum(a * b for a, b in zip(query_u, cand_u, strict=False))
                if (cand_u := self._get_or_compute_unit_vector(c))
                else 0.0,
            )
            for c in candidates
        ]
        return sorted(scored, key=lambda pair: (-pair[1], pair[0]))

    def score_query(self, query: str, skill: Skill) -> float:
        """Calculate semantic similarity between a query text and a skill."""
        query_vec_list = self._get_or_compute_text_vector(query)
        skill_u = self._get_or_compute_unit_vector(skill)
        if not skill_u or not query_vec_list:
            return 0.0
        query_u = _unit_vector(query_vec_list)
        return sum(a * b for a, b in zip(query_u, skill_u, strict=False))

from_skills classmethod

from_skills(
    skills: Sequence[Skill],
    model_name: str | None = None,
    mode: str = "cosine",
) -> DenseScorer

Embed skill texts and instantiate a DenseScorer.

Source code in src/reach/retrieval.py

@classmethod
def from_skills(
    cls,
    skills: Sequence[Skill],
    model_name: str | None = None,
    mode: str = "cosine",
) -> DenseScorer:
    """Embed skill texts and instantiate a DenseScorer."""
    chosen_model = model_name or DEFAULT_RETRIEVAL_MODEL
    model = _load_model2vec_model(chosen_model)

    texts = [skill_text(s) for s in skills]
    embeddings = model.encode(texts)

    vectors: dict[str, EmbeddingVector] = {}
    for skill, emb in zip(skills, embeddings, strict=False):
        vectors[skill.name] = (
            emb.tolist() if hasattr(emb, "tolist") else [float(x) for x in emb]
        )

    return cls(vectors=vectors, model_name=chosen_model, mode=mode)

model_post_init

model_post_init(_context: object) -> None

Compute and cache normalized unit vectors after initialization.

Source code in src/reach/retrieval.py

@override
def model_post_init(self, _context: object) -> None:
    """Compute and cache normalized unit vectors after initialization."""
    object.__setattr__(
        self,
        "_unit_vectors",
        {name: _unit_vector(vec) for name, vec in self.vectors.items()},
    )
    object.__setattr__(self, "_text_vectors", {})

pairwise_similarity

pairwise_similarity(
    skills: Sequence[Skill],
) -> list[PairwiseSimilarity]

Calculate pairwise cosine similarity for all distinct skill pairs.

Source code in src/reach/retrieval.py

def pairwise_similarity(
    self,
    skills: Sequence[Skill],
) -> list[PairwiseSimilarity]:
    """Calculate pairwise cosine similarity for all distinct skill pairs."""
    pairs: list[PairwiseSimilarity] = []

    skill_list = list(skills)
    unit_vecs = {s.name: self._get_or_compute_unit_vector(s) for s in skill_list}

    for i in range(len(skill_list)):
        s1 = skill_list[i]
        u1 = unit_vecs.get(s1.name)
        if not u1:
            continue
        for j in range(i + 1, len(skill_list)):
            s2 = skill_list[j]
            u2 = unit_vecs.get(s2.name)
            if u2:
                sim = sum(a * b for a, b in zip(u1, u2, strict=False))
                pairs.append((s1.name, s2.name, sim))

    return sorted(pairs, key=lambda p: (-p[2], p[0], p[1]))

rank

rank(
    target: Skill, candidates: Sequence[Skill]
) -> list[ScoredPair]

Rank candidate skills against target using semantic similarity.

Source code in src/reach/retrieval.py

def rank(
    self,
    target: Skill,
    candidates: Sequence[Skill],
) -> list[ScoredPair]:
    """Rank candidate skills against target using semantic similarity."""
    if self.mode == "directional":
        target_vec = self._get_or_compute_vector(target)
        if not target_vec:
            return [(c.name, 0.0) for c in candidates if c.name != target.name]

        scored = [
            (c.name, directional_projection(target_vec, self._get_or_compute_vector(c)))
            for c in candidates
            if c.name != target.name
        ]
        return sorted(scored, key=lambda pair: (-pair[1], pair[0]))

    target_u = self._get_or_compute_unit_vector(target)
    if not target_u:
        return [(c.name, 0.0) for c in candidates if c.name != target.name]

    scored = [
        (
            c.name,
            sum(a * b for a, b in zip(target_u, cand_u, strict=False))
            if (cand_u := self._get_or_compute_unit_vector(c))
            else 0.0,
        )
        for c in candidates
        if c.name != target.name
    ]
    return sorted(scored, key=lambda pair: (-pair[1], pair[0]))

rank_text

rank_text(
    text: str, candidates: Sequence[Skill]
) -> list[tuple[str, float]]

Rank candidate skills against query text using semantic similarity.

Source code in src/reach/retrieval.py

def rank_text(
    self,
    text: str,
    candidates: Sequence[Skill],
) -> list[tuple[str, float]]:
    """Rank candidate skills against query text using semantic similarity."""
    if not text or not candidates:
        return sorted(((c.name, 0.0) for c in candidates), key=lambda pair: pair[0])

    query_vec_list = self._get_or_compute_text_vector(text)
    if not query_vec_list:
        return sorted(((c.name, 0.0) for c in candidates), key=lambda pair: pair[0])

    if self.mode == "directional":
        scored = [
            (
                c.name,
                directional_projection(query_vec_list, self._get_or_compute_vector(c)),
            )
            for c in candidates
        ]
        return sorted(scored, key=lambda pair: (-pair[1], pair[0]))

    query_u = _unit_vector(query_vec_list)
    scored = [
        (
            c.name,
            sum(a * b for a, b in zip(query_u, cand_u, strict=False))
            if (cand_u := self._get_or_compute_unit_vector(c))
            else 0.0,
        )
        for c in candidates
    ]
    return sorted(scored, key=lambda pair: (-pair[1], pair[0]))

score_query

score_query(query: str, skill: Skill) -> float

Calculate semantic similarity between a query text and a skill.

Source code in src/reach/retrieval.py

def score_query(self, query: str, skill: Skill) -> float:
    """Calculate semantic similarity between a query text and a skill."""
    query_vec_list = self._get_or_compute_text_vector(query)
    skill_u = self._get_or_compute_unit_vector(skill)
    if not skill_u or not query_vec_list:
        return 0.0
    query_u = _unit_vector(query_vec_list)
    return sum(a * b for a, b in zip(query_u, skill_u, strict=False))

HybridScorer

Bases: BaseModel

Fuse lexical BM25 and dense semantic rankings using Reciprocal Rank Fusion.

Source code in src/reach/retrieval.py

class HybridScorer(BaseModel):
    """Fuse lexical BM25 and dense semantic rankings using Reciprocal Rank Fusion."""

    model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True)

    lexical: Bm25Scorer
    semantic: Any
    rrf_k: int = Field(default=DEFAULT_RRF_K, gt=0)

    @classmethod
    def from_skills(
        cls,
        skills: Sequence[Skill],
        model_name: str | None = None,
        rrf_k: int = DEFAULT_RRF_K,
        k1: float = K1,
        b: float = B,
    ) -> HybridScorer:
        """Instantiate both lexical and dense semantic scorers over skills."""
        lexical = Bm25Scorer.from_skills(skills, k1=k1, b=b)
        semantic = DenseScorer.from_skills(skills, model_name=model_name)
        return cls(lexical=lexical, semantic=semantic, rrf_k=rrf_k)

    @classmethod
    def from_skills_and_vectors(
        cls,
        skills: Sequence[Skill],
        vectors: dict[str, list[float]],
        rrf_k: int = DEFAULT_RRF_K,
        k1: float = K1,
        b: float = B,
    ) -> HybridScorer:
        """Instantiate a HybridScorer using precomputed vectors."""
        lexical = Bm25Scorer.from_skills(skills, k1=k1, b=b)
        semantic = DenseScorer(vectors=vectors)
        return cls(lexical=lexical, semantic=semantic, rrf_k=rrf_k)

    def _fuse_rankings(
        self,
        bm25_ranked: Sequence[tuple[str, float]],
        dense_ranked: Sequence[tuple[str, float]],
    ) -> list[tuple[str, float]]:
        """Fuse positive-score lexical rankings with dense rankings via Reciprocal Rank Fusion."""
        lex_ranks = [name for name, score in bm25_ranked if score > 0.0]
        sem_ranks = [name for name, _ in dense_ranked]
        return compute_rrf([lex_ranks, sem_ranks], k=self.rrf_k)

    def rank_text(
        self,
        text: str,
        candidates: Sequence[Skill],
    ) -> list[tuple[str, float]]:
        """Rank candidate skills against query text using Reciprocal Rank Fusion."""
        if not candidates:
            return []
        return self._fuse_rankings(
            self.lexical.rank_text(text, candidates),
            self.semantic.rank_text(text, candidates),
        )

    def rank(
        self,
        target: Skill,
        candidates: Sequence[Skill],
    ) -> list[tuple[str, float]]:
        """Rank candidate skills using Reciprocal Rank Fusion."""
        pool = [c for c in candidates if c.name != target.name]
        if not pool:
            return []
        return self._fuse_rankings(
            self.lexical.rank(target, pool),
            self.semantic.rank(target, pool),
        )

from_skills classmethod

from_skills(
    skills: Sequence[Skill],
    model_name: str | None = None,
    rrf_k: int = DEFAULT_RRF_K,
    k1: float = K1,
    b: float = B,
) -> HybridScorer

Instantiate both lexical and dense semantic scorers over skills.

Source code in src/reach/retrieval.py

@classmethod
def from_skills(
    cls,
    skills: Sequence[Skill],
    model_name: str | None = None,
    rrf_k: int = DEFAULT_RRF_K,
    k1: float = K1,
    b: float = B,
) -> HybridScorer:
    """Instantiate both lexical and dense semantic scorers over skills."""
    lexical = Bm25Scorer.from_skills(skills, k1=k1, b=b)
    semantic = DenseScorer.from_skills(skills, model_name=model_name)
    return cls(lexical=lexical, semantic=semantic, rrf_k=rrf_k)

from_skills_and_vectors classmethod

from_skills_and_vectors(
    skills: Sequence[Skill],
    vectors: dict[str, list[float]],
    rrf_k: int = DEFAULT_RRF_K,
    k1: float = K1,
    b: float = B,
) -> HybridScorer

Instantiate a HybridScorer using precomputed vectors.

Source code in src/reach/retrieval.py

@classmethod
def from_skills_and_vectors(
    cls,
    skills: Sequence[Skill],
    vectors: dict[str, list[float]],
    rrf_k: int = DEFAULT_RRF_K,
    k1: float = K1,
    b: float = B,
) -> HybridScorer:
    """Instantiate a HybridScorer using precomputed vectors."""
    lexical = Bm25Scorer.from_skills(skills, k1=k1, b=b)
    semantic = DenseScorer(vectors=vectors)
    return cls(lexical=lexical, semantic=semantic, rrf_k=rrf_k)

rank

rank(
    target: Skill, candidates: Sequence[Skill]
) -> list[tuple[str, float]]

Rank candidate skills using Reciprocal Rank Fusion.

Source code in src/reach/retrieval.py

def rank(
    self,
    target: Skill,
    candidates: Sequence[Skill],
) -> list[tuple[str, float]]:
    """Rank candidate skills using Reciprocal Rank Fusion."""
    pool = [c for c in candidates if c.name != target.name]
    if not pool:
        return []
    return self._fuse_rankings(
        self.lexical.rank(target, pool),
        self.semantic.rank(target, pool),
    )

rank_text

rank_text(
    text: str, candidates: Sequence[Skill]
) -> list[tuple[str, float]]

Rank candidate skills against query text using Reciprocal Rank Fusion.

Source code in src/reach/retrieval.py

def rank_text(
    self,
    text: str,
    candidates: Sequence[Skill],
) -> list[tuple[str, float]]:
    """Rank candidate skills against query text using Reciprocal Rank Fusion."""
    if not candidates:
        return []
    return self._fuse_rankings(
        self.lexical.rank_text(text, candidates),
        self.semantic.rank_text(text, candidates),
    )

Scorer

Bases: Protocol

Protocol for scoring candidate skills against a target skill.

Source code in src/reach/retrieval.py

@runtime_checkable
class Scorer(Protocol):
    """Protocol for scoring candidate skills against a target skill."""

    def rank(
        self,
        target: Skill,
        candidates: Sequence[Skill],
    ) -> list[ScoredPair]:
        """Return candidate names paired with scores, strongest first."""
        ...

rank

rank(
    target: Skill, candidates: Sequence[Skill]
) -> list[ScoredPair]

Return candidate names paired with scores, strongest first.

Source code in src/reach/retrieval.py

def rank(
    self,
    target: Skill,
    candidates: Sequence[Skill],
) -> list[ScoredPair]:
    """Return candidate names paired with scores, strongest first."""
    ...

TextScorer

Bases: Protocol

Protocol for scoring candidate skills against arbitrary query text.

Source code in src/reach/retrieval.py

@runtime_checkable
class TextScorer(Protocol):
    """Protocol for scoring candidate skills against arbitrary query text."""

    def rank_text(
        self,
        text: str,
        candidates: Sequence[Skill],
    ) -> list[tuple[str, float]]:
        """Return candidate names paired with scores against query text, strongest first."""
        ...

rank_text

rank_text(
    text: str, candidates: Sequence[Skill]
) -> list[tuple[str, float]]

Return candidate names paired with scores against query text, strongest first.

Source code in src/reach/retrieval.py

def rank_text(
    self,
    text: str,
    candidates: Sequence[Skill],
) -> list[tuple[str, float]]:
    """Return candidate names paired with scores against query text, strongest first."""
    ...

build_scorer

build_scorer(
    name: str,
    skills: Sequence[Skill],
    config: RunConfig | None = None,
) -> Scorer

Build a Scorer instance according to the configured retrieval strategy.

Source code in src/reach/retrieval.py

def build_scorer(
    name: str,
    skills: Sequence[Skill],
    config: RunConfig | None = None,
) -> Scorer:
    """Build a Scorer instance according to the configured retrieval strategy."""
    scorer_type = name.lower().strip()
    k1 = config.retrieval.bm25_k1 if config else K1
    b = config.retrieval.bm25_b if config else B
    match scorer_type:
        case "bm25":
            return Bm25Scorer.from_skills(skills, k1=k1, b=b)
        case "dense":
            model_name = config.retrieval.model if config else DEFAULT_RETRIEVAL_MODEL
            return DenseScorer.from_skills(skills, model_name=model_name)
        case "hybrid":
            model_name = config.retrieval.model if config else DEFAULT_RETRIEVAL_MODEL
            rrf_k = config.retrieval.rrf_k if config else DEFAULT_RRF_K
            try:
                return HybridScorer.from_skills(
                    skills, model_name=model_name, rrf_k=rrf_k, k1=k1, b=b
                )
            except RuntimeError:
                return Bm25Scorer.from_skills(skills, k1=k1, b=b)
        case _:
            msg = f"unknown scorer: {name!r}; valid choices: 'bm25', 'dense', 'hybrid'"
            raise ValueError(msg)

classify_overlap_quadrant

classify_overlap_quadrant(
    lexical_ratio: float,
    semantic_similarity: float,
    lex_high: float = 0.5,
    sem_high: float = 0.75,
) -> str

Classify the relationship between lexical and semantic overlap into a diagnostic quadrant.

Source code in src/reach/retrieval.py

def classify_overlap_quadrant(
    lexical_ratio: float,
    semantic_similarity: float,
    lex_high: float = 0.5,
    sem_high: float = 0.75,
) -> str:
    """Classify the relationship between lexical and semantic overlap into a diagnostic quadrant."""
    match (lexical_ratio >= lex_high, semantic_similarity >= sem_high):
        case (True, True):
            return "Near-Duplicate"
        case (True, False):
            return "Boilerplate / Style"
        case (False, True):
            return "Latent Collision"
        case _:
            return "Distinct"

compute_rrf

compute_rrf(
    rankings: Sequence[Sequence[str]],
    k: int = DEFAULT_RRF_K,
) -> list[tuple[str, float]]

Fuse multiple ranked candidate name lists using Reciprocal Rank Fusion.

Source code in src/reach/retrieval.py

def compute_rrf(
    rankings: Sequence[Sequence[str]],
    k: int = DEFAULT_RRF_K,
) -> list[tuple[str, float]]:
    """Fuse multiple ranked candidate name lists using Reciprocal Rank Fusion."""
    scores: dict[str, float] = defaultdict(float)
    for ranking in rankings:
        for rank_idx, name in enumerate(ranking, start=1):
            scores[name] += 1.0 / (k + rank_idx)

    # Sort descending by fused score; break ties deterministically by alphabetical name
    return sorted(scores.items(), key=lambda pair: (-pair[1], pair[0]))

cosine_similarity

cosine_similarity(
    v1: Sequence[float], v2: Sequence[float]
) -> float

Calculate the cosine similarity between two numeric vectors.

Source code in src/reach/retrieval.py

def cosine_similarity(v1: Sequence[float], v2: Sequence[float]) -> float:
    """Calculate the cosine similarity between two numeric vectors."""
    dot = 0.0
    norm_a = 0.0
    norm_b = 0.0
    for a, b in zip(v1, v2, strict=False):
        dot += a * b
        norm_a += a * a
        norm_b += b * b

    if norm_a <= 0.0 or norm_b <= 0.0:
        return 0.0
    raw = dot / (math.sqrt(norm_a) * math.sqrt(norm_b))
    return max(-1.0, min(1.0, raw))

directional_projection

directional_projection(
    target: Sequence[float], candidate: Sequence[float]
) -> float

Calculate the directional projection of target onto candidate.

Source code in src/reach/retrieval.py

def directional_projection(target: Sequence[float], candidate: Sequence[float]) -> float:
    """Calculate the directional projection of target onto candidate."""
    dot = 0.0
    norm_target_sq = 0.0
    for t, c in zip(target, candidate, strict=False):
        dot += t * c
        norm_target_sq += t * t

    if norm_target_sq <= 0.0:
        return 0.0
    return dot / norm_target_sq

skill_text

skill_text(skill: Skill) -> str

Combine a skill's name and description for lexical indexing.

Source code in src/reach/retrieval.py

def skill_text(skill: Skill) -> str:
    """Combine a skill's name and description for lexical indexing."""
    return f"{skill.name} {skill.description}"

tokenize

tokenize(text: str) -> list[str]

Split text into lowercase alphanumeric and technical tokens.

Source code in src/reach/retrieval.py

def tokenize(text: str) -> list[str]:
    """Split text into lowercase alphanumeric and technical tokens."""
    return _TOKEN.findall(text.lower())