Intermediate Examples

# pyright: strict
"""Example 29: BM25 IDF Term (co-16)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
import math  # => stdlib math -- log/sqrt for idf, cosine, and skip-pointer spacing
 
 
def bm25_idf(term_df: int, n_docs: int) -> float:  # => bM25's own RSJ idf: log((N - df + 0.5) / (df + 0.5))
    """BM25's own RSJ idf: log((N - df + 0.5) / (df + 0.5))."""
    return math.log((n_docs - term_df + 0.5) / (term_df + 0.5))  # => co-16: Robertson & Zaragoza (2009) eq. 3.3
 
 
def plain_idf(term_df: int, n_docs: int) -> float:  # => the Example 13 tf-idf idf, for comparison: log(N / df)
    """The Example 13 tf-idf idf, for comparison: log(N / df)."""
    return math.log(n_docs / term_df)  # => returns math.log(n_docs / term_df)
 
 
def main() -> None:  # => defines main
    n_docs: int = 10  # => a small corpus, N
    common_df: int = 10  # => a term present in EVERY document
    rare_df: int = 1  # => a term present in only one document
 
    bm25_common: float = bm25_idf(common_df, n_docs)  # => co-16: BM25's idf for the ubiquitous term
    plain_common: float = plain_idf(common_df, n_docs)  # => the plain tf-idf idf for the same term
    print(f"term in ALL {n_docs} docs: bm25_idf={bm25_common:.4f}  plain_idf={plain_common:.4f}")  # => shows term in ALL
 
    bm25_rare: float = bm25_idf(rare_df, n_docs)  # => bm25 rare = bm25_idf(rare_df, n_docs)
    plain_rare: float = plain_idf(rare_df, n_docs)  # => plain rare = plain_idf(rare_df, n_docs)
    print(f"term in only 1 doc:   bm25_idf={bm25_rare:.4f}  plain_idf={plain_rare:.4f}")  # => shows term in only 1 doc:   bm25_idf=
 
    assert math.isfinite(bm25_common), "BM25's idf must stay FINITE even for a term in every document"  # => BM25's idf must stay FINITE even for a term in every document
    assert bm25_common < 0, "BM25's idf for a term in EVERY doc goes NEGATIVE, unlike plain idf's 0"  # => BM25's idf for a term in EVERY doc goes NEGATIVE, unlike plain idf's 0
    assert not math.isclose(bm25_common, plain_common), "BM25's idf and plain idf must be DIFFERENT formulas"  # => BM25's idf and plain idf must be DIFFERENT formulas
    print(f"MATCH: bm25_idf({common_df}/{n_docs})={bm25_common:.4f} is finite and negative, unlike plain_idf's {plain_common:.4f}")  # => shows MATCH: bm25_idf(
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

term in ALL 10 docs: bm25_idf=-3.0445  plain_idf=0.0000
term in only 1 doc:   bm25_idf=1.8458  plain_idf=2.3026
MATCH: bm25_idf(10/10)=-3.0445 is finite and negative, unlike plain_idf's 0.0000

# pyright: strict
"""Example 30: BM25 Score: One Term (co-16)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
import math  # => stdlib math -- log/sqrt for idf, cosine, and skip-pointer spacing
 
def bm25_idf(term_df: int, n_docs: int) -> float:  # => bM25's own RSJ idf: log((N - df + 0.5) / (df + 0.5)) -- distinct from plain log(N/df)
    """BM25's own RSJ idf: log((N - df + 0.5) / (df + 0.5)) -- distinct from plain log(N/df)."""
    return math.log((n_docs - term_df + 0.5) / (term_df + 0.5))  # => returns math.log((n_docs - term_df + 0.5) / (term_df + 0.5))
 
 
def bm25_term_weight(  # => defines bm25 term weight
    tf: int, term_df: int, n_docs: int, dl: float, avgdl: float, *, k1: float = 1.2, b: float = 0.75  # => part of this step's computation, continued from the line above
) -> float:  # => part of this step's computation, continued from the line above
    """One query term's BM25 contribution: RSJ idf * saturating tf * length normalization."""
    idf: float = bm25_idf(term_df, n_docs)  # => idf = bm25_idf(term_df, n_docs)
    length_norm: float = (1 - b) + b * (dl / avgdl)  # => co-18: B -- 1.0 at average length, >1 for long docs
    return idf * (tf * (k1 + 1)) / (tf + k1 * length_norm)  # => co-17: the k1-saturated, B-normalized term score
 
 
def bm25_score(  # => defines bm25 score
    query_terms: list[str],  # => part of this step's computation, continued from the line above
    doc_tf: dict[str, int],  # => part of this step's computation, continued from the line above
    df: dict[str, int],  # => part of this step's computation, continued from the line above
    n_docs: int,  # => part of this step's computation, continued from the line above
    dl: float,  # => part of this step's computation, continued from the line above
    avgdl: float,  # => part of this step's computation, continued from the line above
    *,  # => part of this step's computation, continued from the line above
    k1: float = 1.2,  # => k1 = 1.2,
    b: float = 0.75,  # => b = 0.75,
) -> float:  # => part of this step's computation, continued from the line above
    """Sum bm25_term_weight over every query term the document actually contains."""
    total: float = 0.0  # => total = 0.0
    for term in query_terms:  # => iterates one item at a time
        if term in doc_tf:  # => true when term in doc_tf
            total += bm25_term_weight(doc_tf[term], df.get(term, 0), n_docs, dl, avgdl, k1=k1, b=b)  # => part of this step's computation, continued from the line above
    return total  # => returns total
 
 
def main() -> None:  # => defines main
    tf, term_df, n_docs = 3, 2, 5  # => this term occurs 3 times in the doc; 2 of 5 docs contain it
    dl, avgdl = 10.0, 8.0  # => this doc has 10 tokens; the corpus average is 8
    k1, b = 1.2, 0.75  # => co-19: the Lucene/Elasticsearch software defaults
 
    weight: float = bm25_term_weight(tf, term_df, n_docs, dl, avgdl, k1=k1, b=b)  # => co-16: the function under test
    print(f"bm25_term_weight(tf={tf}, df={term_df}, N={n_docs}, dl={dl}, avgdl={avgdl}) = {weight:.6f}")  # => shows bm25_term_weight(tf=
 
    # Hand computation, written out independently (not calling bm25_term_weight):
    hand_idf: float = math.log((n_docs - term_df + 0.5) / (term_df + 0.5))  # => hand idf = math.log((n_docs - term_df + 0.5) / (term_df + ...
    hand_B: float = (1 - b) + b * (dl / avgdl)  # => hand B = (1 - b) + b * (dl / avgdl)
    hand_weight: float = hand_idf * (tf * (k1 + 1)) / (tf + k1 * hand_B)  # => hand weight = hand_idf * (tf * (k1 + 1)) / (tf + k1 * hand_B)
    print(f"hand computation: idf={hand_idf:.6f} B={hand_B:.6f} weight={hand_weight:.6f}")  # => shows hand computation: idf=
 
    assert math.isclose(weight, hand_weight, rel_tol=1e-9), "bm25_term_weight must match the independent hand computation"  # => bm25_term_weight must match the independent hand computation
    print(f"MATCH: bm25_term_weight's {weight:.6f} equals the hand computation's {hand_weight:.6f}")  # => shows MATCH: bm25_term_weight's
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

bm25_term_weight(tf=3, df=2, N=5, dl=10.0, avgdl=8.0) = 0.501857
hand computation: idf=0.336472 B=1.187500 weight=0.501857
MATCH: bm25_term_weight's 0.501857 equals the hand computation's 0.501857

# pyright: strict
"""Example 31: BM25 Saturation Curve (co-17)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
import math  # => stdlib math -- log/sqrt for idf, cosine, and skip-pointer spacing
 
def bm25_idf(term_df: int, n_docs: int) -> float:  # => bM25's own RSJ idf: log((N - df + 0.5) / (df + 0.5)) -- distinct from plain log(N/df)
    """BM25's own RSJ idf: log((N - df + 0.5) / (df + 0.5)) -- distinct from plain log(N/df)."""
    return math.log((n_docs - term_df + 0.5) / (term_df + 0.5))  # => returns math.log((n_docs - term_df + 0.5) / (term_df + 0.5))
 
 
def bm25_term_weight(  # => defines bm25 term weight
    tf: int, term_df: int, n_docs: int, dl: float, avgdl: float, *, k1: float = 1.2, b: float = 0.75  # => part of this step's computation, continued from the line above
) -> float:  # => part of this step's computation, continued from the line above
    """One query term's BM25 contribution: RSJ idf * saturating tf * length normalization."""
    idf: float = bm25_idf(term_df, n_docs)  # => idf = bm25_idf(term_df, n_docs)
    length_norm: float = (1 - b) + b * (dl / avgdl)  # => co-18: B -- 1.0 at average length, >1 for long docs
    return idf * (tf * (k1 + 1)) / (tf + k1 * length_norm)  # => co-17: the k1-saturated, B-normalized term score
 
 
def bm25_score(  # => defines bm25 score
    query_terms: list[str],  # => part of this step's computation, continued from the line above
    doc_tf: dict[str, int],  # => part of this step's computation, continued from the line above
    df: dict[str, int],  # => part of this step's computation, continued from the line above
    n_docs: int,  # => part of this step's computation, continued from the line above
    dl: float,  # => part of this step's computation, continued from the line above
    avgdl: float,  # => part of this step's computation, continued from the line above
    *,  # => part of this step's computation, continued from the line above
    k1: float = 1.2,  # => k1 = 1.2,
    b: float = 0.75,  # => b = 0.75,
) -> float:  # => part of this step's computation, continued from the line above
    """Sum bm25_term_weight over every query term the document actually contains."""
    total: float = 0.0  # => total = 0.0
    for term in query_terms:  # => iterates one item at a time
        if term in doc_tf:  # => true when term in doc_tf
            total += bm25_term_weight(doc_tf[term], df.get(term, 0), n_docs, dl, avgdl, k1=k1, b=b)  # => part of this step's computation, continued from the line above
    return total  # => returns total
 
 
def main() -> None:  # => defines main
    term_df, n_docs = 3, 10  # => a moderately common term
    dl, avgdl = 8.0, 8.0  # => dl == avgdl -- length normalization is neutral (B == 1), isolating tf's effect
    k1: float = 1.2  # => the Lucene/Elasticsearch default
 
    scores: list[float] = [bm25_term_weight(tf, term_df, n_docs, dl, avgdl, k1=k1) for tf in range(1, 21)]  # => co-17: score at each tf
    deltas: list[float] = [scores[i] - scores[i - 1] for i in range(1, len(scores))]  # => the marginal gain of EACH extra occurrence
    for tf, score in zip(range(1, 21), scores):  # => iterates one item at a time
        marker: str = f" (+{scores[tf - 1] - scores[tf - 2]:.4f})" if tf > 1 else ""  # => marker = f" (+{scores[tf - 1] - scores[tf - 2]:.4f})" if...
        print(f"tf={tf:>2}: score={score:.4f}{marker}")  # => shows tf=
 
    for i in range(1, len(deltas)):  # => every adjacent PAIR of marginal gains
        assert deltas[i] <= deltas[i - 1] + 1e-9, f"marginal gain must be non-increasing (tf={i + 2})"  # => marginal gain must be non-increasing (tf={i + 2})
    assert deltas[0] > deltas[-1], "the FIRST extra occurrence must help strictly more than the LAST one shown"  # => the FIRST extra occurrence must help strictly more than the LAST one shown
    print(f"MATCH: marginal gain shrinks monotonically from {deltas[0]:.4f} (tf 1->2) to {deltas[-1]:.4f} (tf 19->20)")  # => shows MATCH: marginal gain shrinks monotonically from
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

tf= 1: score=0.7621
tf= 2: score=1.0479 (+0.2858)
tf= 3: score=1.1976 (+0.1497)
tf= 4: score=1.2898 (+0.0921)
tf= 5: score=1.3522 (+0.0624)
tf= 6: score=1.3973 (+0.0451)
tf= 7: score=1.4313 (+0.0341)
tf= 8: score=1.4580 (+0.0267)
tf= 9: score=1.4794 (+0.0214)
tf=10: score=1.4971 (+0.0176)
tf=11: score=1.5118 (+0.0147)
tf=12: score=1.5243 (+0.0125)
tf=13: score=1.5350 (+0.0107)
tf=14: score=1.5443 (+0.0093)
tf=15: score=1.5525 (+0.0082)
tf=16: score=1.5597 (+0.0072)
tf=17: score=1.5662 (+0.0064)
tf=18: score=1.5719 (+0.0058)
tf=19: score=1.5771 (+0.0052)
tf=20: score=1.5818 (+0.0047)
MATCH: marginal gain shrinks monotonically from 0.2858 (tf 1->2) to 0.0047 (tf 19->20)

# pyright: strict
"""Example 32: TF-IDF vs BM25 Saturation (co-17)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
import math  # => stdlib math -- log/sqrt for idf, cosine, and skip-pointer spacing
 
def bm25_idf(term_df: int, n_docs: int) -> float:  # => bM25's own RSJ idf: log((N - df + 0.5) / (df + 0.5)) -- distinct from plain log(N/df)
    """BM25's own RSJ idf: log((N - df + 0.5) / (df + 0.5)) -- distinct from plain log(N/df)."""
    return math.log((n_docs - term_df + 0.5) / (term_df + 0.5))  # => returns math.log((n_docs - term_df + 0.5) / (term_df + 0.5))
 
 
def bm25_term_weight(  # => defines bm25 term weight
    tf: int, term_df: int, n_docs: int, dl: float, avgdl: float, *, k1: float = 1.2, b: float = 0.75  # => part of this step's computation, continued from the line above
) -> float:  # => part of this step's computation, continued from the line above
    """One query term's BM25 contribution: RSJ idf * saturating tf * length normalization."""
    idf: float = bm25_idf(term_df, n_docs)  # => idf = bm25_idf(term_df, n_docs)
    length_norm: float = (1 - b) + b * (dl / avgdl)  # => co-18: B -- 1.0 at average length, >1 for long docs
    return idf * (tf * (k1 + 1)) / (tf + k1 * length_norm)  # => co-17: the k1-saturated, B-normalized term score
 
 
def bm25_score(  # => defines bm25 score
    query_terms: list[str],  # => part of this step's computation, continued from the line above
    doc_tf: dict[str, int],  # => part of this step's computation, continued from the line above
    df: dict[str, int],  # => part of this step's computation, continued from the line above
    n_docs: int,  # => part of this step's computation, continued from the line above
    dl: float,  # => part of this step's computation, continued from the line above
    avgdl: float,  # => part of this step's computation, continued from the line above
    *,  # => part of this step's computation, continued from the line above
    k1: float = 1.2,  # => k1 = 1.2,
    b: float = 0.75,  # => b = 0.75,
) -> float:  # => part of this step's computation, continued from the line above
    """Sum bm25_term_weight over every query term the document actually contains."""
    total: float = 0.0  # => total = 0.0
    for term in query_terms:  # => iterates one item at a time
        if term in doc_tf:  # => true when term in doc_tf
            total += bm25_term_weight(doc_tf[term], df.get(term, 0), n_docs, dl, avgdl, k1=k1, b=b)  # => part of this step's computation, continued from the line above
    return total  # => returns total
 
 
def tfidf_score(tf: int, term_df: int, n_docs: int) -> float:  # => plain tf-idf: tf * log(N / df) -- LINEAR in tf, no saturation
    """Plain tf-idf: tf * log(N / df) -- LINEAR in tf, no saturation."""
    return tf * math.log(n_docs / term_df)  # => returns tf * math.log(n_docs / term_df)
 
 
def main() -> None:  # => defines main
    term_df, n_docs = 3, 10  # => part of this step's computation, continued from the line above
    dl, avgdl = 8.0, 8.0  # => neutral length normalization, isolating the tf effect
 
    tfidf_scores: list[float] = [tfidf_score(tf, term_df, n_docs) for tf in (10, 20, 40, 80)]  # => co-14: linear growth
    bm25_scores: list[float] = [bm25_term_weight(tf, term_df, n_docs, dl, avgdl) for tf in (10, 20, 40, 80)]  # => co-17: saturating growth
    for tf, ts, bs in zip((10, 20, 40, 80), tfidf_scores, bm25_scores):  # => iterates one item at a time
        print(f"tf={tf:>3}: tf-idf={ts:8.4f}   bm25={bs:6.4f}")  # => shows tf=
 
    tfidf_growth_10_to_80: float = tfidf_scores[-1] - tfidf_scores[0]  # => tf-idf's growth over the SAME tf range
    bm25_growth_10_to_80: float = bm25_scores[-1] - bm25_scores[0]  # => BM25's growth over the SAME tf range
    print(f"growth from tf=10 to tf=80: tf-idf grew by {tfidf_growth_10_to_80:.4f}, bm25 grew by {bm25_growth_10_to_80:.4f}")  # => shows growth from tf=10 to tf=80: tf-idf grew by
 
    assert tfidf_growth_10_to_80 > bm25_growth_10_to_80 * 5, "tf-idf's growth must vastly outpace BM25's over a wide tf range"  # => tf-idf's growth must vastly outpace BM25's over a wide tf range
    assert bm25_scores[-1] < bm25_scores[0] * 1.5, "BM25 must be near-flat by tf=80, having started saturating much earlier"  # => BM25 must be near-flat by tf=80, having started saturating much earlier
    print("MATCH: tf-idf keeps climbing near-linearly while BM25 has essentially flattened out over the same range")  # => shows MATCH: tf-idf keeps climbing near-linearly while BM25 has essentially flattened out over the same range
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

tf= 10: tf-idf= 12.0397   bm25=1.4971
tf= 20: tf-idf= 24.0795   bm25=1.5818
tf= 40: tf-idf= 48.1589   bm25=1.6279
tf= 80: tf-idf= 96.3178   bm25=1.6519
growth from tf=10 to tf=80: tf-idf grew by 84.2781, bm25 grew by 0.1549
MATCH: tf-idf keeps climbing near-linearly while BM25 has essentially flattened out over the same range

# pyright: strict
"""Example 33: BM25 Length Normalization (co-18)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
import math  # => stdlib math -- log/sqrt for idf, cosine, and skip-pointer spacing
 
def bm25_idf(term_df: int, n_docs: int) -> float:  # => bM25's own RSJ idf: log((N - df + 0.5) / (df + 0.5)) -- distinct from plain log(N/df)
    """BM25's own RSJ idf: log((N - df + 0.5) / (df + 0.5)) -- distinct from plain log(N/df)."""
    return math.log((n_docs - term_df + 0.5) / (term_df + 0.5))  # => returns math.log((n_docs - term_df + 0.5) / (term_df + 0.5))
 
 
def bm25_term_weight(  # => defines bm25 term weight
    tf: int, term_df: int, n_docs: int, dl: float, avgdl: float, *, k1: float = 1.2, b: float = 0.75  # => part of this step's computation, continued from the line above
) -> float:  # => part of this step's computation, continued from the line above
    """One query term's BM25 contribution: RSJ idf * saturating tf * length normalization."""
    idf: float = bm25_idf(term_df, n_docs)  # => idf = bm25_idf(term_df, n_docs)
    length_norm: float = (1 - b) + b * (dl / avgdl)  # => co-18: B -- 1.0 at average length, >1 for long docs
    return idf * (tf * (k1 + 1)) / (tf + k1 * length_norm)  # => co-17: the k1-saturated, B-normalized term score
 
 
def bm25_score(  # => defines bm25 score
    query_terms: list[str],  # => part of this step's computation, continued from the line above
    doc_tf: dict[str, int],  # => part of this step's computation, continued from the line above
    df: dict[str, int],  # => part of this step's computation, continued from the line above
    n_docs: int,  # => part of this step's computation, continued from the line above
    dl: float,  # => part of this step's computation, continued from the line above
    avgdl: float,  # => part of this step's computation, continued from the line above
    *,  # => part of this step's computation, continued from the line above
    k1: float = 1.2,  # => k1 = 1.2,
    b: float = 0.75,  # => b = 0.75,
) -> float:  # => part of this step's computation, continued from the line above
    """Sum bm25_term_weight over every query term the document actually contains."""
    total: float = 0.0  # => total = 0.0
    for term in query_terms:  # => iterates one item at a time
        if term in doc_tf:  # => true when term in doc_tf
            total += bm25_term_weight(doc_tf[term], df.get(term, 0), n_docs, dl, avgdl, k1=k1, b=b)  # => part of this step's computation, continued from the line above
    return total  # => returns total
 
 
def main() -> None:  # => defines main
    term_df, n_docs = 3, 10  # => part of this step's computation, continued from the line above
    avgdl: float = 20.0  # => the corpus average document length
    tf: int = 4  # => the SAME term count in both documents below
 
    short_doc_len: float = 10.0  # => half the average length
    long_doc_len: float = 60.0  # => three times the average length
 
    short_score: float = bm25_term_weight(tf, term_df, n_docs, short_doc_len, avgdl)  # => co-18: short doc, same tf
    long_score: float = bm25_term_weight(tf, term_df, n_docs, long_doc_len, avgdl)  # => co-18: long doc, same tf
    print(f"short doc (dl={short_doc_len}, tf={tf}): score={short_score:.4f}")  # => shows short doc (dl=
    print(f"long doc  (dl={long_doc_len}, tf={tf}): score={long_score:.4f}")  # => shows long doc  (dl=
 
    b: float = 0.75  # => the default b
    short_B: float = (1 - b) + b * (short_doc_len / avgdl)  # => short B = (1 - b) + b * (short_doc_len / avgdl)
    long_B: float = (1 - b) + b * (long_doc_len / avgdl)  # => long B = (1 - b) + b * (long_doc_len / avgdl)
    print(f"B(short)={short_B:.4f}  B(long)={long_B:.4f}")  # => shows B(short)=
 
    assert long_B > short_B, "the longer document must have a LARGER B (length-norm factor)"  # => the longer document must have a LARGER B (length-norm factor)
    assert short_score > long_score, "with EQUAL tf, the SHORTER document must score HIGHER"  # => with EQUAL tf, the SHORTER document must score HIGHER
    print(f"MATCH: with identical tf={tf}, the short doc (B={short_B:.4f}) outscores the long doc (B={long_B:.4f})")  # => shows MATCH: with identical tf=
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

short doc (dl=10.0, tf=4): score=1.4120
long doc  (dl=60.0, tf=4): score=0.9581
B(short)=0.6250  B(long)=2.5000
MATCH: with identical tf=4, the short doc (B=0.6250) outscores the long doc (B=2.5000)

# pyright: strict
"""Example 34: BM25 b Sweep (co-18)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
import math  # => stdlib math -- log/sqrt for idf, cosine, and skip-pointer spacing
 
def bm25_idf(term_df: int, n_docs: int) -> float:  # => bM25's own RSJ idf: log((N - df + 0.5) / (df + 0.5)) -- distinct from plain log(N/df)
    """BM25's own RSJ idf: log((N - df + 0.5) / (df + 0.5)) -- distinct from plain log(N/df)."""
    return math.log((n_docs - term_df + 0.5) / (term_df + 0.5))  # => returns math.log((n_docs - term_df + 0.5) / (term_df + 0.5))
 
 
def bm25_term_weight(  # => defines bm25 term weight
    tf: int, term_df: int, n_docs: int, dl: float, avgdl: float, *, k1: float = 1.2, b: float = 0.75  # => part of this step's computation, continued from the line above
) -> float:  # => part of this step's computation, continued from the line above
    """One query term's BM25 contribution: RSJ idf * saturating tf * length normalization."""
    idf: float = bm25_idf(term_df, n_docs)  # => idf = bm25_idf(term_df, n_docs)
    length_norm: float = (1 - b) + b * (dl / avgdl)  # => co-18: B -- 1.0 at average length, >1 for long docs
    return idf * (tf * (k1 + 1)) / (tf + k1 * length_norm)  # => co-17: the k1-saturated, B-normalized term score
 
 
def bm25_score(  # => defines bm25 score
    query_terms: list[str],  # => part of this step's computation, continued from the line above
    doc_tf: dict[str, int],  # => part of this step's computation, continued from the line above
    df: dict[str, int],  # => part of this step's computation, continued from the line above
    n_docs: int,  # => part of this step's computation, continued from the line above
    dl: float,  # => part of this step's computation, continued from the line above
    avgdl: float,  # => part of this step's computation, continued from the line above
    *,  # => part of this step's computation, continued from the line above
    k1: float = 1.2,  # => k1 = 1.2,
    b: float = 0.75,  # => b = 0.75,
) -> float:  # => part of this step's computation, continued from the line above
    """Sum bm25_term_weight over every query term the document actually contains."""
    total: float = 0.0  # => total = 0.0
    for term in query_terms:  # => iterates one item at a time
        if term in doc_tf:  # => true when term in doc_tf
            total += bm25_term_weight(doc_tf[term], df.get(term, 0), n_docs, dl, avgdl, k1=k1, b=b)  # => part of this step's computation, continued from the line above
    return total  # => returns total
 
 
def main() -> None:  # => defines main
    term_df, n_docs = 3, 10  # => part of this step's computation, continued from the line above
    avgdl: float = 20.0  # => the corpus average -- short doc is well below it, long doc well above
 
    short_dl, short_tf = 5.0, 2  # => a SHORT, concentrated document: few words, term appears twice
    long_dl, long_tf = 50.0, 4  # => a LONG, diluted document: many words, term appears more often in raw count
 
    for b in (0.0, 0.5, 0.75, 1.0):  # => co-18: sweeping the length-normalization strength
        short_score: float = bm25_term_weight(short_tf, term_df, n_docs, short_dl, avgdl, b=b)  # => short score = bm25_term_weight(short_tf, term_df, n_docs, sho...
        long_score: float = bm25_term_weight(long_tf, term_df, n_docs, long_dl, avgdl, b=b)  # => long score = bm25_term_weight(long_tf, term_df, n_docs, long...
        winner: str = "short" if short_score > long_score else "long"  # => winner = "short" if short_score > long_score else "long"
        print(f"b={b:.2f}: short={short_score:.4f}  long={long_score:.4f}  winner={winner}")  # => shows b=
 
    score_at_b0_short: float = bm25_term_weight(short_tf, term_df, n_docs, short_dl, avgdl, b=0.0)  # => score at b0 short = bm25_term_weight(short_tf, term_df, n_docs, sho...
    score_at_b0_long: float = bm25_term_weight(long_tf, term_df, n_docs, long_dl, avgdl, b=0.0)  # => score at b0 long = bm25_term_weight(long_tf, term_df, n_docs, long...
    score_at_b1_short: float = bm25_term_weight(short_tf, term_df, n_docs, short_dl, avgdl, b=1.0)  # => score at b1 short = bm25_term_weight(short_tf, term_df, n_docs, sho...
    score_at_b1_long: float = bm25_term_weight(long_tf, term_df, n_docs, long_dl, avgdl, b=1.0)  # => score at b1 long = bm25_term_weight(long_tf, term_df, n_docs, long...
 
    assert score_at_b0_long > score_at_b0_short, "at b=0 (no length norm), the long doc's higher raw tf must win"  # => at b=0 (no length norm), the long doc's higher raw tf must win
    assert score_at_b1_short > score_at_b1_long, "at b=1 (full length norm), the short doc must overtake the long one"  # => at b=1 (full length norm), the short doc must overtake the long one
    print("MATCH: the ranking genuinely flips -- long doc wins at b=0, short doc wins at b=1")  # => shows MATCH: the ranking genuinely flips -- long doc wins at b=0, short doc wins at b=1
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

b=0.00: short=1.0479  long=1.2898  winner=long
b=0.50: short=1.2194  long=1.0995  winner=short
b=0.75: short=1.3281  long=1.0239  winner=short
b=1.00: short=1.4580  long=0.9581  winner=short
MATCH: the ranking genuinely flips -- long doc wins at b=0, short doc wins at b=1

# pyright: strict
"""Example 35: BM25 k1 Sweep (co-17)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
import math  # => stdlib math -- log/sqrt for idf, cosine, and skip-pointer spacing
 
def bm25_idf(term_df: int, n_docs: int) -> float:  # => bM25's own RSJ idf: log((N - df + 0.5) / (df + 0.5)) -- distinct from plain log(N/df)
    """BM25's own RSJ idf: log((N - df + 0.5) / (df + 0.5)) -- distinct from plain log(N/df)."""
    return math.log((n_docs - term_df + 0.5) / (term_df + 0.5))  # => returns math.log((n_docs - term_df + 0.5) / (term_df + 0.5))
 
 
def bm25_term_weight(  # => defines bm25 term weight
    tf: int, term_df: int, n_docs: int, dl: float, avgdl: float, *, k1: float = 1.2, b: float = 0.75  # => part of this step's computation, continued from the line above
) -> float:  # => part of this step's computation, continued from the line above
    """One query term's BM25 contribution: RSJ idf * saturating tf * length normalization."""
    idf: float = bm25_idf(term_df, n_docs)  # => idf = bm25_idf(term_df, n_docs)
    length_norm: float = (1 - b) + b * (dl / avgdl)  # => co-18: B -- 1.0 at average length, >1 for long docs
    return idf * (tf * (k1 + 1)) / (tf + k1 * length_norm)  # => co-17: the k1-saturated, B-normalized term score
 
 
def bm25_score(  # => defines bm25 score
    query_terms: list[str],  # => part of this step's computation, continued from the line above
    doc_tf: dict[str, int],  # => part of this step's computation, continued from the line above
    df: dict[str, int],  # => part of this step's computation, continued from the line above
    n_docs: int,  # => part of this step's computation, continued from the line above
    dl: float,  # => part of this step's computation, continued from the line above
    avgdl: float,  # => part of this step's computation, continued from the line above
    *,  # => part of this step's computation, continued from the line above
    k1: float = 1.2,  # => k1 = 1.2,
    b: float = 0.75,  # => b = 0.75,
) -> float:  # => part of this step's computation, continued from the line above
    """Sum bm25_term_weight over every query term the document actually contains."""
    total: float = 0.0  # => total = 0.0
    for term in query_terms:  # => iterates one item at a time
        if term in doc_tf:  # => true when term in doc_tf
            total += bm25_term_weight(doc_tf[term], df.get(term, 0), n_docs, dl, avgdl, k1=k1, b=b)  # => part of this step's computation, continued from the line above
    return total  # => returns total
 
 
def marginal_gain(term_df: int, n_docs: int, dl: float, avgdl: float, k1: float, tf_from: int, tf_to: int) -> float:  # => the score gained by moving from tf_from to tf_to occurrences, at a fixed k1
    """The score gained by moving from tf_from to tf_to occurrences, at a fixed k1."""
    return bm25_term_weight(tf_to, term_df, n_docs, dl, avgdl, k1=k1) - bm25_term_weight(tf_from, term_df, n_docs, dl, avgdl, k1=k1)  # => returns bm25_term_weight(tf_to, term_df, n_docs, dl, avgdl, k1=k1...
 
 
def main() -> None:  # => defines main
    term_df, n_docs = 3, 10  # => part of this step's computation, continued from the line above
    dl, avgdl = 8.0, 8.0  # => neutral length normalization
 
    for k1 in (0.5, 1.2, 2.0, 5.0):  # => co-17: small to large saturation constants
        early_gain: float = marginal_gain(term_df, n_docs, dl, avgdl, k1, 1, 2)  # => gain from the FIRST extra occurrence
        late_gain: float = marginal_gain(term_df, n_docs, dl, avgdl, k1, 9, 10)  # => gain from a MUCH later occurrence
        print(f"k1={k1:.1f}: early gain (tf 1->2)={early_gain:.4f}  late gain (tf 9->10)={late_gain:.4f}")  # => shows k1=
 
    low_k1_late_gain: float = marginal_gain(term_df, n_docs, dl, avgdl, 0.5, 9, 10)  # => a SMALL k1's late-stage gain
    high_k1_late_gain: float = marginal_gain(term_df, n_docs, dl, avgdl, 5.0, 9, 10)  # => a LARGE k1's late-stage gain
    assert high_k1_late_gain > low_k1_late_gain, "a LARGER k1 must still yield a bigger late-stage marginal gain (saturates later)"  # => a LARGER k1 must still yield a bigger late-stage marginal gain (saturates later)
    print(f"MATCH: at tf 9->10, k1=5.0's gain ({high_k1_late_gain:.4f}) exceeds k1=0.5's gain ({low_k1_late_gain:.4f}) -- k1 moves the saturation point")  # => shows MATCH: at tf 9->10, k1=5.0's gain (
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

k1=0.5: early gain (tf 1->2)=0.1524  late gain (tf 9->10)=0.0057
k1=1.2: early gain (tf 1->2)=0.2858  late gain (tf 9->10)=0.0176
k1=2.0: early gain (tf 1->2)=0.3811  late gain (tf 9->10)=0.0346
k1=5.0: early gain (tf 1->2)=0.5444  late gain (tf 9->10)=0.1089
MATCH: at tf 9->10, k1=5.0's gain (0.1089) exceeds k1=0.5's gain (0.0057) -- k1 moves the saturation point

# pyright: strict
"""Example 36: BM25 Defaults (co-19)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
import math  # => stdlib math -- log/sqrt for idf, cosine, and skip-pointer spacing
 
def bm25_idf(term_df: int, n_docs: int) -> float:  # => bM25's own RSJ idf: log((N - df + 0.5) / (df + 0.5)) -- distinct from plain log(N/df)
    """BM25's own RSJ idf: log((N - df + 0.5) / (df + 0.5)) -- distinct from plain log(N/df)."""
    return math.log((n_docs - term_df + 0.5) / (term_df + 0.5))  # => returns math.log((n_docs - term_df + 0.5) / (term_df + 0.5))
 
 
def bm25_term_weight(  # => defines bm25 term weight
    tf: int, term_df: int, n_docs: int, dl: float, avgdl: float, *, k1: float = 1.2, b: float = 0.75  # => part of this step's computation, continued from the line above
) -> float:  # => part of this step's computation, continued from the line above
    """One query term's BM25 contribution: RSJ idf * saturating tf * length normalization."""
    idf: float = bm25_idf(term_df, n_docs)  # => idf = bm25_idf(term_df, n_docs)
    length_norm: float = (1 - b) + b * (dl / avgdl)  # => co-18: B -- 1.0 at average length, >1 for long docs
    return idf * (tf * (k1 + 1)) / (tf + k1 * length_norm)  # => co-17: the k1-saturated, B-normalized term score
 
 
def bm25_score(  # => defines bm25 score
    query_terms: list[str],  # => part of this step's computation, continued from the line above
    doc_tf: dict[str, int],  # => part of this step's computation, continued from the line above
    df: dict[str, int],  # => part of this step's computation, continued from the line above
    n_docs: int,  # => part of this step's computation, continued from the line above
    dl: float,  # => part of this step's computation, continued from the line above
    avgdl: float,  # => part of this step's computation, continued from the line above
    *,  # => part of this step's computation, continued from the line above
    k1: float = 1.2,  # => k1 = 1.2,
    b: float = 0.75,  # => b = 0.75,
) -> float:  # => part of this step's computation, continued from the line above
    """Sum bm25_term_weight over every query term the document actually contains."""
    total: float = 0.0  # => total = 0.0
    for term in query_terms:  # => iterates one item at a time
        if term in doc_tf:  # => true when term in doc_tf
            total += bm25_term_weight(doc_tf[term], df.get(term, 0), n_docs, dl, avgdl, k1=k1, b=b)  # => part of this step's computation, continued from the line above
    return total  # => returns total
 
 
def main() -> None:  # => defines main
    default_k1: float = 1.2  # => co-19: the Lucene/Elasticsearch SOFTWARE default
    default_b: float = 0.75  # => co-19: the Lucene/Elasticsearch SOFTWARE default
 
    paper_k1_lower, paper_k1_upper = 1.2, 2.0  # => Robertson & Zaragoza (2009)'s own recommended OPEN interval
    paper_b_lower, paper_b_upper = 0.5, 0.8  # => Robertson & Zaragoza (2009)'s own recommended OPEN interval
 
    k1_in_papers_open_range: bool = paper_k1_lower < default_k1 < paper_k1_upper  # => STRICT inequality, as the paper states it
    b_in_papers_open_range: bool = paper_b_lower < default_b < paper_b_upper  # => STRICT inequality, as the paper states it
    print(f"default k1={default_k1}: inside paper's OPEN range (1.2, 2.0)? {k1_in_papers_open_range}")  # => shows default k1=
    print(f"default b={default_b}: inside paper's OPEN range (0.5, 0.8)? {b_in_papers_open_range}")  # => shows default b=
 
    term_df, n_docs, dl, avgdl, tf = 3, 10, 8.0, 8.0, 4  # => part of this step's computation, continued from the line above
    score: float = bm25_term_weight(tf, term_df, n_docs, dl, avgdl, k1=default_k1, b=default_b)  # => the software defaults, applied
    print(f"score with software defaults (k1={default_k1}, b={default_b}): {score:.4f}")  # => shows score with software defaults (k1=
 
    assert not k1_in_papers_open_range, "k1=1.2 sits exactly ON the paper's lower boundary -- NOT strictly inside its open interval"  # => k1=1.2 sits exactly ON the paper's lower boundary -- NOT strictly inside its open interval
    assert b_in_papers_open_range, "b=0.75 DOES sit strictly inside the paper's own recommended (0.5, 0.8) range"  # => b=0.75 DOES sit strictly inside the paper's own recommended (0.5, 0.8) range
    print("MATCH: k1=1.2 is the boundary value (not strictly inside the paper's range), while b=0.75 is strictly inside it")  # => shows MATCH: k1=1.2 is the boundary value (not strictly inside the paper's range), while b=0.75 is strictly inside it
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

default k1=1.2: inside paper's OPEN range (1.2, 2.0)? False
default b=0.75: inside paper's OPEN range (0.5, 0.8)? True
score with software defaults (k1=1.2, b=0.75): 1.2898
MATCH: k1=1.2 is the boundary value (not strictly inside the paper's range), while b=0.75 is strictly inside it

# pyright: strict
"""Example 37: BM25 Full Ranker (co-16)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
import math  # => stdlib math -- log/sqrt for idf, cosine, and skip-pointer spacing
 
def bm25_idf(term_df: int, n_docs: int) -> float:  # => bM25's own RSJ idf: log((N - df + 0.5) / (df + 0.5)) -- distinct from plain log(N/df)
    """BM25's own RSJ idf: log((N - df + 0.5) / (df + 0.5)) -- distinct from plain log(N/df)."""
    return math.log((n_docs - term_df + 0.5) / (term_df + 0.5))  # => returns math.log((n_docs - term_df + 0.5) / (term_df + 0.5))
 
 
def bm25_term_weight(  # => defines bm25 term weight
    tf: int, term_df: int, n_docs: int, dl: float, avgdl: float, *, k1: float = 1.2, b: float = 0.75  # => part of this step's computation, continued from the line above
) -> float:  # => part of this step's computation, continued from the line above
    """One query term's BM25 contribution: RSJ idf * saturating tf * length normalization."""
    idf: float = bm25_idf(term_df, n_docs)  # => idf = bm25_idf(term_df, n_docs)
    length_norm: float = (1 - b) + b * (dl / avgdl)  # => co-18: B -- 1.0 at average length, >1 for long docs
    return idf * (tf * (k1 + 1)) / (tf + k1 * length_norm)  # => co-17: the k1-saturated, B-normalized term score
 
 
def bm25_score(  # => defines bm25 score
    query_terms: list[str],  # => part of this step's computation, continued from the line above
    doc_tf: dict[str, int],  # => part of this step's computation, continued from the line above
    df: dict[str, int],  # => part of this step's computation, continued from the line above
    n_docs: int,  # => part of this step's computation, continued from the line above
    dl: float,  # => part of this step's computation, continued from the line above
    avgdl: float,  # => part of this step's computation, continued from the line above
    *,  # => part of this step's computation, continued from the line above
    k1: float = 1.2,  # => k1 = 1.2,
    b: float = 0.75,  # => b = 0.75,
) -> float:  # => part of this step's computation, continued from the line above
    """Sum bm25_term_weight over every query term the document actually contains."""
    total: float = 0.0  # => total = 0.0
    for term in query_terms:  # => iterates one item at a time
        if term in doc_tf:  # => true when term in doc_tf
            total += bm25_term_weight(doc_tf[term], df.get(term, 0), n_docs, dl, avgdl, k1=k1, b=b)  # => part of this step's computation, continued from the line above
    return total  # => returns total
 
 
def build_df(docs: dict[int, list[str]]) -> dict[str, int]:  # => defines build df
    df: dict[str, int] = {}  # => starts empty, populated by the loop below
    for tokens in docs.values():  # => iterates one item at a time
        for term in set(tokens):  # => iterates one item at a time
            df[term] = df.get(term, 0) + 1  # => counter pattern: 0 on first sight, then increments
    return df  # => returns df
 
 
def rank_bm25(docs: dict[int, list[str]], query_terms: list[str]) -> list[tuple[int, float]]:  # => defines rank bm25
    n_docs: int = len(docs)  # => this fixture's own size
    df: dict[str, int] = build_df(docs)  # => df = build_df(docs)
    total_len: int = sum(len(tokens) for tokens in docs.values())  # => total len = sum(len(tokens) for tokens in docs.values())
    avgdl: float = total_len / n_docs  # => co-18: the corpus-wide average document length
 
    scores: dict[int, float] = {}  # => starts empty, populated by the loop below
    for doc_id, tokens in docs.items():  # => iterates one item at a time
        doc_tf: dict[str, int] = {}  # => starts empty, populated by the loop below
        for t in tokens:  # => iterates one item at a time
            doc_tf[t] = doc_tf.get(t, 0) + 1  # => counter pattern: 0 on first sight, then increments
        scores[doc_id] = bm25_score(query_terms, doc_tf, df, n_docs, float(len(tokens)), avgdl)  # => scores = bm25_score(query_terms, doc_tf, df, n_docs, flo...
    return sorted(scores.items(), key=lambda kv: (-kv[1], kv[0]))  # => returns sorted(scores.items(), key=lambda kv: (-kv[1], kv[0]))
 
 
def main() -> None:  # => defines main
    docs: dict[int, list[str]] = {  # => docs = {
        0: ["search", "engine", "index", "search", "engine"],  # => strong match: BOTH terms, repeated
        1: ["search", "results", "page"],  # => weak match: only ONE term
        2: ["cooking", "recipe", "book"],  # => no match at all
        3: ["weather", "forecast", "today"],  # => filler, no match -- keeps df(search)/df(engine) LOW relative to N
        4: ["news", "update", "daily"],  # => filler, no match -- same reason: avoids a tiny-corpus idf sign flip
    }  # => opens/closes this multi-line literal
    query_terms: list[str] = ["search", "engine"]  # => query terms = ["search", "engine"]
    ranking: list[tuple[int, float]] = rank_bm25(docs, query_terms)  # => co-16: the full BM25 ranking
    for doc_id, score in ranking:  # => iterates one item at a time
        print(f"doc {doc_id}: score={score:.4f}")  # => shows doc
 
    # An INDEPENDENT reference pass: recompute doc 0's score from scratch, a different way.
    n_docs = len(docs)  # => n docs = len(docs)
    df_ref: dict[str, int] = build_df(docs)  # => df ref = build_df(docs)
    avgdl_ref: float = sum(len(t) for t in docs.values()) / n_docs  # => avgdl ref = sum(len(t) for t in docs.values()) / n_docs
    ref_score_doc0: float = 0.0  # => ref score doc0 = 0.0
    for term in query_terms:  # => iterates one item at a time
        tf_doc0: int = docs[0].count(term)  # => tf doc0 = docs[0].count(term)
        if tf_doc0 > 0:  # => true when tf_doc0 > 0
            idf_ref: float = math.log((n_docs - df_ref[term] + 0.5) / (df_ref[term] + 0.5))  # => idf ref = math.log((n_docs - df_ref[term] + 0.5) / (df_re...
            B_ref: float = (1 - 0.75) + 0.75 * (len(docs[0]) / avgdl_ref)  # => B ref = (1 - 0.75) + 0.75 * (len(docs[0]) / avgdl_ref)
            ref_score_doc0 += idf_ref * (tf_doc0 * 2.2) / (tf_doc0 + 1.2 * B_ref)  # => part of this step's computation, continued from the line above
 
    assert ranking[0][0] == 0, "doc 0 must rank first -- it matches both query terms, repeated"  # => doc 0 must rank first -- it matches both query terms, repeated
    assert ranking[1][0] == 1, "doc 1 must rank second -- it matches one query term, once"  # => doc 1 must rank second -- it matches one query term, once
    assert ranking[0][1] > ranking[1][1] > 0, "both matching docs must score strictly positive, doc 0 higher than doc 1"  # => both matching docs must score strictly positive, doc 0 higher than doc 1
    assert math.isclose(ranking[0][1], ref_score_doc0, rel_tol=1e-9), "doc 0's score must match the independent reference pass"  # => doc 0's score must match the independent reference pass
    print(f"MATCH: doc 0 ranks first ({ranking[0][1]:.4f}), matching the independent reference computation ({ref_score_doc0:.4f})")  # => shows MATCH: doc 0 ranks first (
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

doc 0: score=1.7426
doc 1: score=0.3535
doc 2: score=0.0000
doc 3: score=0.0000
doc 4: score=0.0000
MATCH: doc 0 ranks first (1.7426), matching the independent reference computation (1.7426)

# pyright: strict
"""Example 38: Top-K Heap (co-20)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
import heapq  # => stdlib binary heap -- backs the size-k top-k selection
import random  # => stdlib PRNG -- reproducible synthetic fixtures and trials
 
 
def topk_via_heap(scored: list[tuple[float, int]], k: int) -> list[tuple[float, int]]:  # => return the k highest-scoring (score, doc_id) pairs using a size-k min-heap
    """Return the k highest-scoring (score, doc_id) pairs using a size-k min-heap."""
    heap: list[tuple[float, int]] = []  # => co-20: a MIN-heap -- the smallest of the k kept sits at heap[0]
    for score, doc_id in scored:  # => iterates one item at a time
        if len(heap) < k:  # => true when len(heap) < k
            heapq.heappush(heap, (score, doc_id))  # => co-20: heap not full yet -- always keep this candidate
        elif score > heap[0][0]:  # => co-20: only replace the WORST kept item if this candidate beats it
            heapq.heapreplace(heap, (score, doc_id))  # => pop the min, push the new one, in one O(log k) step
    return sorted(heap, key=lambda sd: (-sd[0], sd[1]))  # => co-20: final k results, best score first
 
 
def topk_via_full_sort(scored: list[tuple[float, int]], k: int) -> list[tuple[float, int]]:  # => reference implementation: sort everything, take the first k
    """Reference implementation: sort everything, take the first k."""
    return sorted(scored, key=lambda sd: (-sd[0], sd[1]))[:k]  # => returns sorted(scored, key=lambda sd: (-sd[0], sd[1]))[:k]
 
 
def main() -> None:  # => defines main
    rng = random.Random(7)  # => fixed seed -- reproducible
    scored: list[tuple[float, int]] = [(rng.uniform(0, 100), doc_id) for doc_id in range(1000)]  # => 1000 (score, doc_id) pairs
 
    heap_result: list[tuple[float, int]] = topk_via_heap(scored, k=5)  # => co-20: the size-5 heap's own top-5
    sort_result: list[tuple[float, int]] = topk_via_full_sort(scored, k=5)  # => the reference: full sort, then slice
    print(f"heap top-5:  {[(round(s, 2), d) for s, d in heap_result]}")  # => shows heap top-5
    print(f"sort top-5:  {[(round(s, 2), d) for s, d in sort_result]}")  # => shows sort top-5
 
    assert heap_result == sort_result, "the heap's top-5 must be IDENTICAL to the full sort's top-5"  # => the heap's top-5 must be IDENTICAL to the full sort's top-5
    print("MATCH: the size-5 min-heap returns exactly the same top-5 as a full sort over all 1000 scores")  # => shows MATCH: the size-5 min-heap returns exactly the same top-5 as a full sort over all 1000 scores
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

heap top-5:  [(99.88, 558), (99.75, 733), (99.66, 905), (99.65, 607), (99.31, 143)]
sort top-5:  [(99.88, 558), (99.75, 733), (99.66, 905), (99.65, 607), (99.31, 143)]
MATCH: the size-5 min-heap returns exactly the same top-5 as a full sort over all 1000 scores

# pyright: strict
"""Example 39: Top-K vs Full Sort Timing (co-20)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
import heapq  # => stdlib binary heap -- backs the size-k top-k selection
import random  # => stdlib PRNG -- reproducible synthetic fixtures and trials
import time  # => stdlib timer -- wall-clock measurement, not a benchmark micro-op
from typing import Callable  # => from typing: Callable
 
 
def topk_via_heap(scored: list[tuple[float, int]], k: int) -> list[tuple[float, int]]:  # => defines topk via heap
    return heapq.nlargest(k, scored, key=lambda sd: sd[0])  # => co-20: heapq's own optimized top-k
 
 
def topk_via_full_sort(scored: list[tuple[float, int]], k: int) -> list[tuple[float, int]]:  # => defines topk via full sort
    return sorted(scored, key=lambda sd: -sd[0])[:k]  # => returns sorted(scored, key=lambda sd: -sd[0])[:k]
 
 
def main() -> None:  # => defines main
    rng = random.Random(11)  # => fixed seed -- reproducible
    n: int = 500_000  # => a large scored corpus
    k: int = 10  # => a SMALL k relative to n -- exactly where a heap should win
    scored: list[tuple[float, int]] = [(rng.uniform(0, 100), doc_id) for doc_id in range(n)]  # => scored = [(rng.uniform(0, 100), doc_id) for doc_id in ra...
 
    heap_result: list[tuple[float, int]] = topk_via_heap(scored, k)  # => correctness check first
    sort_result: list[tuple[float, int]] = topk_via_full_sort(scored, k)  # => sort result = topk_via_full_sort(scored, k)
    assert heap_result == sort_result, "heap and full-sort top-k must be IDENTICAL before comparing speed"  # => heap and full-sort top-k must be IDENTICAL before comparing speed
 
    heap_time: float = min(_time_call(lambda: topk_via_heap(scored, k)) for _ in range(3))  # => best of 3 runs
    sort_time: float = min(_time_call(lambda: topk_via_full_sort(scored, k)) for _ in range(3))  # => best of 3 runs
    print(f"n={n}, k={k}: heap={heap_time * 1000:.2f}ms  full_sort={sort_time * 1000:.2f}ms  ratio={sort_time / heap_time:.1f}x")  # => shows n=
 
    assert heap_time < sort_time, "the heap-based top-k must be FASTER than a full sort at this n and k"  # => the heap-based top-k must be FASTER than a full sort at this n and k
    print(f"MATCH: identical top-{k} results, and the heap is {sort_time / heap_time:.1f}x faster than a full sort")  # => shows MATCH: identical top
 
 
def _time_call(fn: Callable[[], object]) -> float:  # => defines  time call
    start: float = time.perf_counter()  # => start = time.perf_counter()
    fn()  # => part of this step's computation, continued from the line above
    return time.perf_counter() - start  # => returns time.perf_counter() - start
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

n=500000, k=10: heap=15.10ms  full_sort=100.76ms  ratio=6.7x
MATCH: identical top-10 results, and the heap is 6.7x faster than a full sort

# pyright: strict
"""Example 40: Precision/Recall Compute (co-21)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
 
def precision(retrieved: set[int], relevant: set[int]) -> float:  # => |relevant ∩ retrieved| / |retrieved| -- denominator is what you RETURNED
    """|relevant ∩ retrieved| / |retrieved| -- denominator is what you RETURNED."""
    if not retrieved:  # => true when not retrieved
        return 0.0  # => returns 0.0
    return len(relevant & retrieved) / len(retrieved)  # => co-21: how much of what you returned was actually relevant
 
 
def recall(retrieved: set[int], relevant: set[int]) -> float:  # => |relevant ∩ retrieved| / |relevant| -- denominator is what EXISTS to find
    """|relevant ∩ retrieved| / |relevant| -- denominator is what EXISTS to find."""
    if not relevant:  # => true when not relevant
        return 0.0  # => returns 0.0
    return len(relevant & retrieved) / len(relevant)  # => co-21: how much of what exists did you actually find
 
 
def main() -> None:  # => defines main
    relevant: set[int] = {1, 2, 3, 4, 5}  # => the 5 documents that are ACTUALLY relevant to this query
    retrieved: set[int] = {1, 2, 6, 7}  # => the 4 documents the system actually returned
 
    p: float = precision(retrieved, relevant)  # => co-21: 2 of the 4 retrieved were relevant
    r: float = recall(retrieved, relevant)  # => co-21: 2 of the 5 relevant were found
    print(f"relevant: {sorted(relevant)}")  # => shows relevant
    print(f"retrieved: {sorted(retrieved)}")  # => shows retrieved
    print(f"precision: {p:.4f}  recall: {r:.4f}")  # => shows precision
 
    hand_precision: float = 2 / 4  # => hand tally: {1,2} are the 2 relevant docs among the 4 retrieved
    hand_recall: float = 2 / 5  # => hand tally: {1,2} are the 2 found docs among the 5 relevant
    assert p == hand_precision, f"precision must equal the hand tally {hand_precision}"  # => precision must equal the hand tally {hand_precision}
    assert r == hand_recall, f"recall must equal the hand tally {hand_recall}"  # => recall must equal the hand tally {hand_recall}
    print(f"MATCH: precision={p} and recall={r} both equal their hand-tallied values")  # => shows MATCH: precision=
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

relevant: [1, 2, 3, 4, 5]
retrieved: [1, 2, 6, 7]
precision: 0.5000  recall: 0.4000
MATCH: precision=0.5 and recall=0.4 both equal their hand-tallied values

# pyright: strict
"""Example 41: Precision/Recall Direction (co-21)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
 
def precision(retrieved: set[int], relevant: set[int]) -> float:  # => defines precision
    return len(relevant & retrieved) / len(retrieved) if retrieved else 0.0  # => returns len(relevant & retrieved) / len(retrieved) if retrieved e...
 
 
def recall(retrieved: set[int], relevant: set[int]) -> float:  # => defines recall
    return len(relevant & retrieved) / len(relevant) if relevant else 0.0  # => returns len(relevant & retrieved) / len(relevant) if relevant els...
 
 
def main() -> None:  # => defines main
    relevant: set[int] = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}  # => 10 documents are truly relevant
 
    cautious_retrieval: set[int] = {1}  # => co-21: retrieves ONLY 1 doc -- but it IS relevant
    p_cautious: float = precision(cautious_retrieval, relevant)  # => 1/1 -- perfect precision
    r_cautious: float = recall(cautious_retrieval, relevant)  # => 1/10 -- terrible recall
    print(f"cautious (retrieves 1 doc): precision={p_cautious:.4f}  recall={r_cautious:.4f}")  # => shows cautious (retrieves 1 doc): precision=
 
    broad_retrieval: set[int] = set(range(1, 101))  # => co-21: retrieves EVERYTHING from 1 to 100
    p_broad: float = precision(broad_retrieval, relevant)  # => 10/100 -- terrible precision
    r_broad: float = recall(broad_retrieval, relevant)  # => 10/10 -- perfect recall
    print(f"broad (retrieves 100 docs): precision={p_broad:.4f}  recall={r_broad:.4f}")  # => shows broad (retrieves 100 docs): precision=
 
    assert p_cautious == 1.0 and r_cautious < 0.2, "the cautious retrieval must have PERFECT precision, POOR recall"  # => the cautious retrieval must have PERFECT precision, POOR recall
    assert r_broad == 1.0 and p_broad < 0.2, "the broad retrieval must have PERFECT recall, POOR precision"  # => the broad retrieval must have PERFECT recall, POOR precision
    print("MATCH: retrieving 1 doc maximizes precision at recall's expense; retrieving everything does the reverse")  # => shows MATCH: retrieving 1 doc maximizes precision at recall's expense; retrieving everything does the reverse
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

cautious (retrieves 1 doc): precision=1.0000  recall=0.1000
broad (retrieves 100 docs): precision=0.1000  recall=1.0000
MATCH: retrieving 1 doc maximizes precision at recall's expense; retrieving everything does the reverse

# pyright: strict
"""Example 42: F1 Harmonic Mean (co-21)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
 
def f1_score(precision: float, recall: float) -> float:  # => the harmonic mean of precision and recall
    """The harmonic mean of precision and recall."""
    if precision + recall == 0:  # => true when precision + recall == 0
        return 0.0  # => returns 0.0
    return 2 * precision * recall / (precision + recall)  # => co-21: harmonic, NOT arithmetic, mean
 
 
def main() -> None:  # => defines main
    balanced_p, balanced_r = 0.8, 0.8  # => a BALANCED system
    lopsided_p, lopsided_r = 1.0, 0.1  # => a LOPSIDED system -- perfect precision, terrible recall
 
    f1_balanced: float = f1_score(balanced_p, balanced_r)  # => f1 balanced = f1_score(balanced_p, balanced_r)
    f1_lopsided: float = f1_score(lopsided_p, lopsided_r)  # => f1 lopsided = f1_score(lopsided_p, lopsided_r)
    arithmetic_mean_lopsided: float = (lopsided_p + lopsided_r) / 2  # => an arithmetic mean, for contrast
    print(f"balanced (p={balanced_p}, r={balanced_r}): F1={f1_balanced:.4f}")  # => shows balanced (p=
    print(f"lopsided (p={lopsided_p}, r={lopsided_r}): F1={f1_lopsided:.4f}  (arithmetic mean would be {arithmetic_mean_lopsided:.4f})")  # => shows lopsided (p=
 
    assert lopsided_r < f1_lopsided < arithmetic_mean_lopsided, "F1 must sit BETWEEN recall and the arithmetic mean, closer to the SMALLER value"  # => F1 must sit BETWEEN recall and the arithmetic mean, closer to the SMALLER value
    assert f1_lopsided < 0.2, "F1 must stay near the smaller of the two lopsided values, not near their average"  # => F1 must stay near the smaller of the two lopsided values, not near their average
    print(f"MATCH: F1={f1_lopsided:.4f} stays close to the smaller value (recall={lopsided_r}), far below the arithmetic mean {arithmetic_mean_lopsided:.4f}")  # => shows MATCH: F1=
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

balanced (p=0.8, r=0.8): F1=0.8000
lopsided (p=1.0, r=0.1): F1=0.1818  (arithmetic mean would be 0.5500)
MATCH: F1=0.1818 stays close to the smaller value (recall=0.1), far below the arithmetic mean 0.5500

# pyright: strict
"""Example 43: Precision at K (co-22)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
 
def precision_at_k(ranked_results: list[int], relevant: set[int], k: int) -> float:  # => precision computed over only the top-k entries of a ranked result list
    """Precision computed over only the top-k entries of a ranked result list."""
    top_k: list[int] = ranked_results[:k]  # => co-22: only the FIRST k results count
    hits: int = sum(1 for doc_id in top_k if doc_id in relevant)  # => how many of the top k are relevant
    return hits / k if k > 0 else 0.0  # => returns hits / k if k > 0 else 0.0
 
 
def main() -> None:  # => defines main
    relevant: set[int] = {1, 3, 5, 7, 9, 11, 13}  # => 7 truly relevant documents, corpus-wide
    ranked: list[int] = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]  # => the system's ranked output, best-first
 
    p_at_5: float = precision_at_k(ranked, relevant, k=5)  # => co-22: precision over just the top 5
    p_at_10: float = precision_at_k(ranked, relevant, k=10)  # => co-22: precision over the top 10
    print(f"ranked (top 10): {ranked}")  # => shows ranked (top 10)
    print(f"precision@5:  {p_at_5:.4f}")  # => shows precision@5
    print(f"precision@10: {p_at_10:.4f}")  # => shows precision@10
 
    hand_top5_hits: int = sum(1 for d in ranked[:5] if d in relevant)  # => hand count: {1,3,5} in the top 5
    hand_top10_hits: int = sum(1 for d in ranked[:10] if d in relevant)  # => hand count: {1,3,5,7,9} in the top 10
    assert p_at_5 == hand_top5_hits / 5, "precision@5 must match a hand count of relevant docs in the first 5"  # => precision@5 must match a hand count of relevant docs in the first 5
    assert p_at_10 == hand_top10_hits / 10, "precision@10 must match a hand count of relevant docs in the first 10"  # => precision@10 must match a hand count of relevant docs in the first 10
    print(f"MATCH: precision@5 ({p_at_5}) and precision@10 ({p_at_10}) both equal their hand-counted values")  # => shows MATCH: precision@5 (
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

ranked (top 10): [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
precision@5:  0.6000
precision@10: 0.5000
MATCH: precision@5 (0.6) and precision@10 (0.5) both equal their hand-counted values

# pyright: strict
"""Example 44: Qrels Load (co-23)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
 
def load_qrels(raw: list[tuple[str, int]]) -> dict[str, set[int]]:  # => load a flat list of (query, relevant_doc_id) rows into query -> {relevant doc-ids}
    """Load a flat list of (query, relevant_doc_id) rows into query -> {relevant doc-ids}."""
    qrels: dict[str, set[int]] = {}  # => co-23: the ground-truth relevance judgments
    for query, doc_id in raw:  # => one labeled (query, doc) judgment at a time
        qrels.setdefault(query, set()).add(doc_id)  # => part of this step's computation, continued from the line above
    return qrels  # => returns qrels
 
 
def main() -> None:  # => defines main
    raw_judgments: list[tuple[str, int]] = [  # => flat rows, as a human annotator might produce them
        ("search engine", 1),  # => part of this step's computation, continued from the line above
        ("search engine", 3),  # => part of this step's computation, continued from the line above
        ("search engine", 7),  # => part of this step's computation, continued from the line above
        ("ranking algorithm", 2),  # => part of this step's computation, continued from the line above
        ("ranking algorithm", 4),  # => part of this step's computation, continued from the line above
    ]  # => opens/closes this multi-line literal
    qrels: dict[str, set[int]] = load_qrels(raw_judgments)  # => co-23: the loaded, query-grouped judgments
    for query in sorted(qrels):  # => iterates one item at a time
        print(f"{query!r}: {sorted(qrels[query])}")  # => prints this step's result
 
    assert qrels["search engine"] == {1, 3, 7}, "'search engine' must map to exactly its 3 labeled relevant docs"  # => 'search engine' must map to exactly its 3 labeled relevant docs
    assert qrels["ranking algorithm"] == {2, 4}, "'ranking algorithm' must map to exactly its 2 labeled relevant docs"  # => 'ranking algorithm' must map to exactly its 2 labeled relevant docs
    assert len(qrels) == 2, "there must be exactly 2 distinct queries in this judgment set"  # => there must be exactly 2 distinct queries in this judgment set
    print(f"MATCH: each query maps to exactly its labeled relevant docs, {len(qrels)} distinct queries total")  # => shows MATCH: each query maps to exactly its labeled relevant docs,
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

'ranking algorithm': [2, 4]
'search engine': [1, 3, 7]
MATCH: each query maps to exactly its labeled relevant docs, 2 distinct queries total

# pyright: strict
"""Example 45: Evaluate Two Configs (co-22, co-23)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
import math  # => stdlib math -- log/sqrt for idf, cosine, and skip-pointer spacing
 
def bm25_idf(term_df: int, n_docs: int) -> float:  # => bM25's own RSJ idf: log((N - df + 0.5) / (df + 0.5)) -- distinct from plain log(N/df)
    """BM25's own RSJ idf: log((N - df + 0.5) / (df + 0.5)) -- distinct from plain log(N/df)."""
    return math.log((n_docs - term_df + 0.5) / (term_df + 0.5))  # => returns math.log((n_docs - term_df + 0.5) / (term_df + 0.5))
 
 
def bm25_term_weight(  # => defines bm25 term weight
    tf: int, term_df: int, n_docs: int, dl: float, avgdl: float, *, k1: float = 1.2, b: float = 0.75  # => part of this step's computation, continued from the line above
) -> float:  # => part of this step's computation, continued from the line above
    """One query term's BM25 contribution: RSJ idf * saturating tf * length normalization."""
    idf: float = bm25_idf(term_df, n_docs)  # => idf = bm25_idf(term_df, n_docs)
    length_norm: float = (1 - b) + b * (dl / avgdl)  # => co-18: B -- 1.0 at average length, >1 for long docs
    return idf * (tf * (k1 + 1)) / (tf + k1 * length_norm)  # => co-17: the k1-saturated, B-normalized term score
 
 
def bm25_score(  # => defines bm25 score
    query_terms: list[str],  # => part of this step's computation, continued from the line above
    doc_tf: dict[str, int],  # => part of this step's computation, continued from the line above
    df: dict[str, int],  # => part of this step's computation, continued from the line above
    n_docs: int,  # => part of this step's computation, continued from the line above
    dl: float,  # => part of this step's computation, continued from the line above
    avgdl: float,  # => part of this step's computation, continued from the line above
    *,  # => part of this step's computation, continued from the line above
    k1: float = 1.2,  # => k1 = 1.2,
    b: float = 0.75,  # => b = 0.75,
) -> float:  # => part of this step's computation, continued from the line above
    """Sum bm25_term_weight over every query term the document actually contains."""
    total: float = 0.0  # => total = 0.0
    for term in query_terms:  # => iterates one item at a time
        if term in doc_tf:  # => true when term in doc_tf
            total += bm25_term_weight(doc_tf[term], df.get(term, 0), n_docs, dl, avgdl, k1=k1, b=b)  # => part of this step's computation, continued from the line above
    return total  # => returns total
 
 
def build_df(docs: dict[int, list[str]]) -> dict[str, int]:  # => defines build df
    df: dict[str, int] = {}  # => starts empty, populated by the loop below
    for tokens in docs.values():  # => iterates one item at a time
        for term in set(tokens):  # => iterates one item at a time
            df[term] = df.get(term, 0) + 1  # => counter pattern: 0 on first sight, then increments
    return df  # => returns df
 
 
def rank_tfidf(docs: dict[int, list[str]], query_terms: list[str]) -> list[int]:  # => defines rank tfidf
    n_docs: int = len(docs)  # => this fixture's own size
    df: dict[str, int] = build_df(docs)  # => df = build_df(docs)
    scores: dict[int, float] = {}  # => starts empty, populated by the loop below
    for doc_id, tokens in docs.items():  # => iterates one item at a time
        tf: dict[str, int] = {}  # => starts empty, populated by the loop below
        for t in tokens:  # => iterates one item at a time
            tf[t] = tf.get(t, 0) + 1  # => counter pattern: 0 on first sight, then increments
        scores[doc_id] = sum(tf[t] * math.log(n_docs / df[t]) for t in query_terms if t in tf)  # => scores = sum(tf[t] * math.log(n_docs / df[t]) for t in q...
    return [doc_id for doc_id, _ in sorted(scores.items(), key=lambda kv: (-kv[1], kv[0]))]  # => returns [doc_id for doc_id, _ in sorted(scores.items(), key=lambd...
 
 
def rank_bm25_docs(docs: dict[int, list[str]], query_terms: list[str]) -> list[int]:  # => defines rank bm25 docs
    n_docs: int = len(docs)  # => this fixture's own size
    df: dict[str, int] = build_df(docs)  # => df = build_df(docs)
    avgdl: float = sum(len(t) for t in docs.values()) / n_docs  # => avgdl = sum(len(t) for t in docs.values()) / n_docs
    scores: dict[int, float] = {}  # => starts empty, populated by the loop below
    for doc_id, tokens in docs.items():  # => iterates one item at a time
        tf: dict[str, int] = {}  # => starts empty, populated by the loop below
        for t in tokens:  # => iterates one item at a time
            tf[t] = tf.get(t, 0) + 1  # => counter pattern: 0 on first sight, then increments
        scores[doc_id] = bm25_score(query_terms, tf, df, n_docs, float(len(tokens)), avgdl)  # => scores = bm25_score(query_terms, tf, df, n_docs, float(l...
    return [doc_id for doc_id, _ in sorted(scores.items(), key=lambda kv: (-kv[1], kv[0]))]  # => returns [doc_id for doc_id, _ in sorted(scores.items(), key=lambd...
 
 
def precision_at_k(ranked: list[int], relevant: set[int], k: int) -> float:  # => defines precision at k
    top_k: list[int] = ranked[:k]  # => top k = ranked[:k]
    return sum(1 for d in top_k if d in relevant) / k if k > 0 else 0.0  # => returns sum(1 for d in top_k if d in relevant) / k if k > 0 else 0.0
 
 
def main() -> None:  # => defines main
    docs: dict[int, list[str]] = {  # => docs = {
        0: ["search", "engine"] * 5 + ["content"] * 40,  # => KEYWORD-STUFFED: 5x each term padded into a 50-token doc
        1: ["search", "engine", "index", "ranking"],  # => balanced, on-topic, each term ONCE
        2: ["cooking", "recipe"],  # => off-topic
        3: ["weather", "forecast", "today"],  # => filler, off-topic -- keeps df(search)/df(engine) LOW relative to N
        4: ["news", "update", "daily"],  # => filler, off-topic -- same reason: avoids a tiny-corpus idf sign flip
    }  # => opens/closes this multi-line literal
    query_terms: list[str] = ["search", "engine"]  # => query terms = ["search", "engine"]
    qrels: dict[int, bool] = {0: False, 1: True, 2: False, 3: False, 4: False}  # => co-23: doc 1 is the ONE truly relevant result
    relevant: set[int] = {doc_id for doc_id, is_rel in qrels.items() if is_rel}  # => relevant = {doc_id for doc_id, is_rel in qrels.items() if ...
 
    tfidf_ranking: list[int] = rank_tfidf(docs, query_terms)  # => co-22: tf-idf's ranked order
    bm25_ranking: list[int] = rank_bm25_docs(docs, query_terms)  # => co-22: BM25's ranked order
    p1_tfidf: float = precision_at_k(tfidf_ranking, relevant, k=1)  # => precision@1 for each config
    p1_bm25: float = precision_at_k(bm25_ranking, relevant, k=1)  # => p1 bm25 = precision_at_k(bm25_ranking, relevant, k=1)
    print(f"tf-idf ranking: {tfidf_ranking}  precision@1={p1_tfidf:.4f}")  # => shows tf-idf ranking
    print(f"bm25 ranking:   {bm25_ranking}   precision@1={p1_bm25:.4f}")  # => shows bm25 ranking
 
    assert tfidf_ranking[0] == 0, "tf-idf's unbounded tf lets the repetitive doc 0 rank first (a hand-tallied fact of this fixture)"  # => tf-idf's unbounded tf lets the repetitive doc 0 rank first (a hand-tallied fact of this fixture)
    assert bm25_ranking[0] == 1, "BM25's saturation lets the balanced, truly relevant doc 1 rank first instead"  # => BM25's saturation lets the balanced, truly relevant doc 1 rank first instead
    assert p1_bm25 > p1_tfidf, "BM25 must score HIGHER precision@1 than tf-idf on this fixture"  # => BM25 must score HIGHER precision@1 than tf-idf on this fixture
    print(f"MATCH: BM25's precision@1 ({p1_bm25}) beats tf-idf's ({p1_tfidf}) -- saturation avoided the keyword-stuffed doc")  # => shows MATCH: BM25's precision@1 (
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

tf-idf ranking: [0, 1, 2, 3, 4]  precision@1=0.0000
bm25 ranking:   [1, 0, 2, 3, 4]   precision@1=1.0000
MATCH: BM25's precision@1 (1.0) beats tf-idf's (0.0) -- saturation avoided the keyword-stuffed doc

# pyright: strict
"""Example 46: Average Precision (co-24)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
 
def average_precision(ranked: list[int], relevant: set[int]) -> float:  # => mean of precision@k, evaluated at every rank position where a relevant doc appears
    """Mean of precision@k, evaluated at every rank position where a relevant doc appears."""
    if not relevant:  # => true when not relevant
        return 0.0  # => returns 0.0
    precisions_at_hits: list[float] = []  # => co-24: one precision@k value per relevant hit, in rank order
    hits_so_far: int = 0  # => a running counter, starting at zero
    for k, doc_id in enumerate(ranked, start=1):  # => co-24: k is the 1-based rank position
        if doc_id in relevant:  # => true when doc_id in relevant
            hits_so_far += 1  # => advances hits_so_far
            precisions_at_hits.append(hits_so_far / k)  # => co-22: precision@k, evaluated ONLY at this hit
    return sum(precisions_at_hits) / len(relevant)  # => co-24: averaged over ALL relevant docs, not just the ones found
 
 
def main() -> None:  # => defines main
    relevant: set[int] = {2, 4, 6}  # => 3 truly relevant documents
    ranked: list[int] = [1, 2, 3, 4, 5, 6, 7]  # => relevant docs land at ranks 2, 4, and 6
 
    ap: float = average_precision(ranked, relevant)  # => co-24: this query's own Average Precision
    print(f"ranked: {ranked}")  # => shows ranked
    print(f"relevant: {sorted(relevant)} (found at ranks 2, 4, 6)")  # => shows relevant
    print(f"average precision: {ap:.4f}")  # => shows average precision
 
    hand_p_at_2: float = 1 / 2  # => rank 2: 1 hit out of 2 seen so far
    hand_p_at_4: float = 2 / 4  # => rank 4: 2 hits out of 4 seen so far
    hand_p_at_6: float = 3 / 6  # => rank 6: 3 hits out of 6 seen so far
    hand_ap: float = (hand_p_at_2 + hand_p_at_4 + hand_p_at_6) / 3  # => mean over the 3 relevant docs
    assert ap == hand_ap, f"AP must equal the hand computation {hand_ap}"  # => AP must equal the hand computation {hand_ap}
    print(f"MATCH: AP={ap} equals the hand computation ({hand_p_at_2}+{hand_p_at_4}+{hand_p_at_6})/3={hand_ap}")  # => shows MATCH: AP=
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

ranked: [1, 2, 3, 4, 5, 6, 7]
relevant: [2, 4, 6] (found at ranks 2, 4, 6)
average precision: 0.5000
MATCH: AP=0.5 equals the hand computation (0.5+0.5+0.5)/3=0.5

# pyright: strict
"""Example 47: MAP: Multi-Query (co-24)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
 
def average_precision(ranked: list[int], relevant: set[int]) -> float:  # => defines average precision
    if not relevant:  # => true when not relevant
        return 0.0  # => returns 0.0
    precisions_at_hits: list[float] = []  # => starts empty, populated by the loop below
    hits_so_far: int = 0  # => a running counter, starting at zero
    for k, doc_id in enumerate(ranked, start=1):  # => iterates one item at a time
        if doc_id in relevant:  # => true when doc_id in relevant
            hits_so_far += 1  # => advances hits_so_far
            precisions_at_hits.append(hits_so_far / k)  # => records this item, in order
    return sum(precisions_at_hits) / len(relevant)  # => returns sum(precisions_at_hits) / len(relevant)
 
 
def mean_average_precision(rankings: dict[str, list[int]], qrels: dict[str, set[int]]) -> float:  # => mAP: the mean of average_precision across every query in the set
    """MAP: the mean of average_precision across every query in the set."""
    aps: list[float] = [average_precision(rankings[q], qrels[q]) for q in rankings]  # => co-24: one AP per query
    return sum(aps) / len(aps) if aps else 0.0  # => returns sum(aps) / len(aps) if aps else 0.0
 
 
def main() -> None:  # => defines main
    rankings: dict[str, list[int]] = {  # => 3 queries, each with its own ranked result list
        "q1": [1, 2, 3, 4],  # => entry for 'q1'
        "q2": [5, 6, 7, 8],  # => entry for 'q2'
        "q3": [9, 10, 11, 12],  # => entry for 'q3'
    }  # => opens/closes this multi-line literal
    qrels: dict[str, set[int]] = {  # => 3 queries, each with its own relevant-doc set
        "q1": {1, 3},  # => AP for q1: hits at rank 1 (p=1.0) and rank 3 (p=2/3) -> (1.0+0.6667)/2
        "q2": {5, 6, 7, 8},  # => AP for q2: every result is relevant, all hits at consecutive ranks -> AP = 1.0
        "q3": {12},  # => AP for q3: the only relevant doc is LAST -> AP = 1/4
    }  # => opens/closes this multi-line literal
    map_score: float = mean_average_precision(rankings, qrels)  # => co-24: one number over all 3 queries
    per_query_ap: dict[str, float] = {q: average_precision(rankings[q], qrels[q]) for q in rankings}  # => the per-query breakdown
    for q, ap in per_query_ap.items():  # => iterates one item at a time
        print(f"{q}: AP={ap:.4f}")  # => prints this step's result
    print(f"MAP: {map_score:.4f}")  # => shows MAP
 
    hand_ap_q1: float = (1 / 1 + 2 / 3) / 2  # => hits at rank 1 and rank 3
    hand_ap_q2: float = 1.0  # => every result relevant, in order -- perfect AP
    hand_ap_q3: float = (1 / 4) / 1  # => the only relevant doc is the LAST of 4 results
    hand_map: float = (hand_ap_q1 + hand_ap_q2 + hand_ap_q3) / 3  # => hand map = (hand_ap_q1 + hand_ap_q2 + hand_ap_q3) / 3
    assert abs(map_score - hand_map) < 1e-9, f"MAP must equal the hand-computed mean {hand_map}"  # => MAP must equal the hand-computed mean {hand_map}
    print(f"MATCH: MAP={map_score:.4f} equals the hand-computed mean of the 3 per-query APs ({hand_map:.4f})")  # => shows MATCH: MAP=
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

q1: AP=0.8333
q2: AP=1.0000
q3: AP=0.2500
MAP: 0.6944
MATCH: MAP=0.6944 equals the hand-computed mean of the 3 per-query APs (0.6944)

# pyright: strict
"""Example 48: nDCG Compute (co-24)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
import math  # => stdlib math -- log/sqrt for idf, cosine, and skip-pointer spacing
 
 
def dcg(graded_relevances: list[float]) -> float:  # => discounted Cumulative Gain: sum of rel_i / log2(i + 1), i is the 1-based rank
    """Discounted Cumulative Gain: sum of rel_i / log2(i + 1), i is the 1-based rank."""
    return sum(rel / math.log2(i + 1) for i, rel in enumerate(graded_relevances, start=1))  # => co-24: later ranks discounted MORE
 
 
def ndcg(ranked_relevances: list[float], ideal_relevances: list[float]) -> float:  # => nDCG@len: DCG of the actual ranking, normalized by the IDEAL (best-possible) ranking's DCG
    """nDCG@len: DCG of the actual ranking, normalized by the IDEAL (best-possible) ranking's DCG."""
    ideal_dcg: float = dcg(sorted(ideal_relevances, reverse=True))  # => co-24: the best possible ordering of these SAME grades
    if ideal_dcg == 0:  # => true when ideal_dcg == 0
        return 0.0  # => returns 0.0
    return dcg(ranked_relevances) / ideal_dcg  # => co-24: normalized so a perfect ranking scores exactly 1.0
 
 
def main() -> None:  # => defines main
    grades: dict[int, float] = {1: 3.0, 2: 2.0, 3: 1.0, 4: 0.0}  # => graded relevance, 3=highly relevant, 0=irrelevant
 
    perfect_ranking: list[int] = [1, 2, 3, 4]  # => best grade first -- the IDEAL order
    shuffled_ranking: list[int] = [4, 1, 3, 2]  # => worst grade FIRST -- a genuinely bad order
 
    perfect_relevances: list[float] = [grades[d] for d in perfect_ranking]  # => perfect relevances = [grades[d] for d in perfect_ranking]
    shuffled_relevances: list[float] = [grades[d] for d in shuffled_ranking]  # => shuffled relevances = [grades[d] for d in shuffled_ranking]
    ideal_relevances: list[float] = list(grades.values())  # => the SAME grades, any order (sorted internally by dcg)
 
    ndcg_perfect: float = ndcg(perfect_relevances, ideal_relevances)  # => co-24: the perfect ranking's nDCG
    ndcg_shuffled: float = ndcg(shuffled_relevances, ideal_relevances)  # => co-24: the shuffled ranking's nDCG
    print(f"perfect ranking {perfect_ranking}: nDCG={ndcg_perfect:.6f}")  # => shows perfect ranking
    print(f"shuffled ranking {shuffled_ranking}: nDCG={ndcg_shuffled:.6f}")  # => shows shuffled ranking
 
    assert math.isclose(ndcg_perfect, 1.0, abs_tol=1e-9), "a PERFECT ranking must score nDCG == 1.0 exactly"  # => a PERFECT ranking must score nDCG == 1.0 exactly
    assert ndcg_shuffled < ndcg_perfect, "a SHUFFLED (worse) ranking must score strictly LESS than the perfect one"  # => a SHUFFLED (worse) ranking must score strictly LESS than the perfect one
    print(f"MATCH: the perfect ranking scores exactly {ndcg_perfect}, the shuffled ranking scores less ({ndcg_shuffled:.6f})")  # => shows MATCH: the perfect ranking scores exactly
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end