Advanced Examples

# pyright: strict
"""Example 57: NRT Refresh Model (co-27)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
from dataclasses import dataclass, field  # => from dataclasses: dataclass, field
 
 
@dataclass  # => part of this step's computation, continued from the line above
class NrtIndex:  # => part of this step's computation, continued from the line above
    """Docs added via add() sit in a PENDING buffer until refresh() commits them."""
 
    committed: dict[int, list[str]] = field(default_factory=lambda: {})  # => co-27: SEARCHABLE documents
    pending: dict[int, list[str]] = field(default_factory=lambda: {})  # => co-27: buffered, NOT yet searchable
 
    def add(self, doc_id: int, tokens: list[str]) -> None:  # => buffer a document -- it is NOT searchable until refresh() runs
        """Buffer a document -- it is NOT searchable until refresh() runs."""
        self.pending[doc_id] = tokens  # => co-27: goes into the buffer, not the committed index
 
    def refresh(self) -> None:  # => commit every pending document, making it searchable
        """Commit every pending document, making it searchable."""
        for doc_id, tokens in self.pending.items():  # => co-27: moves EVERY buffered doc into committed
            self.committed[doc_id] = tokens  # => self = tokens
        self.pending.clear()  # => the buffer is now empty -- everything has been committed
 
    def contains(self, doc_id: int) -> bool:  # => true only if doc_id is in the COMMITTED (searchable) index
        """True only if doc_id is in the COMMITTED (searchable) index."""
        return doc_id in self.committed  # => co-27: pending docs do NOT count as findable
 
 
def main() -> None:  # => defines main
    index = NrtIndex()  # => co-27: starts with nothing committed, nothing pending
    index.add(1, ["search", "engine"])  # => buffered -- NOT yet searchable
    print(f"immediately after add(1, ...): contains(1)={index.contains(1)}")  # => shows immediately after add(1, ...): contains(1)=
 
    assert not index.contains(1), "a document must be INVISIBLE to search before refresh() runs"  # => a document must be INVISIBLE to search before refresh() runs
 
    index.refresh()  # => co-27: commits the pending buffer
    print(f"after refresh(): contains(1)={index.contains(1)}")  # => shows after refresh(): contains(1)=
    assert index.contains(1), "the SAME document must be VISIBLE immediately after refresh()"  # => the SAME document must be VISIBLE immediately after refresh()
    print("MATCH: doc 1 was invisible pre-refresh and visible post-refresh -- genuine NRT behavior")  # => shows MATCH: doc 1 was invisible pre-refresh and visible post-refresh -- genuine NRT behavior
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

immediately after add(1, ...): contains(1)=False
after refresh(): contains(1)=True
MATCH: doc 1 was invisible pre-refresh and visible post-refresh -- genuine NRT behavior

# pyright: strict
"""Example 58: Typed Index Class (co-01, co-29)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
from dataclasses import dataclass, field  # => from dataclasses: dataclass, field
 
@dataclass  # => part of this step's computation, continued from the line above
class InvertedIndex:  # => part of this step's computation, continued from the line above
    """A typed, mutable inverted index: term -> {doc_id: term_frequency}."""
 
    postings: dict[str, dict[int, int]] = field(default_factory=lambda: {})  # => postings = field(default_factory=lambda: {})
    doc_lengths: dict[int, int] = field(default_factory=lambda: {})  # => doc lengths = field(default_factory=lambda: {})
 
    def add(self, doc_id: int, tokens: list[str]) -> None:  # => index one document's tokens -- updates postings AND doc_lengths
        """Index one document's tokens -- updates postings AND doc_lengths."""
        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
        for term, count in tf.items():  # => iterates one item at a time
            self.postings.setdefault(term, {})[doc_id] = count  # => part of this step's computation, continued from the line above
        self.doc_lengths[doc_id] = len(tokens)  # => self = len(tokens)
 
    def query(self, term: str) -> set[int]:  # => return every doc-id containing term
        """Return every doc-id containing term."""
        return set(self.postings.get(term, {}).keys())  # => returns set(self.postings.get(term, {}).keys())
 
 
def main() -> None:  # => defines main
    index = InvertedIndex()  # => co-01: an empty typed index
    index.add(0, ["search", "engine", "search"])  # => co-29: co-01's inverted index, built incrementally
    index.add(1, ["search", "results"])  # => part of this step's computation, continued from the line above
    index.add(2, ["cooking", "recipe"])  # => part of this step's computation, continued from the line above
 
    search_hits: set[int] = index.query("search")  # => co-01: docs 0 and 1 both contain "search"
    cooking_hits: set[int] = index.query("cooking")  # => only doc 2 contains "cooking"
    print(f"query('search'): {sorted(search_hits)}")  # => shows query('search')
    print(f"query('cooking'): {sorted(cooking_hits)}")  # => shows query('cooking')
    print(f"doc_lengths: {index.doc_lengths}")  # => shows doc_lengths
 
    assert search_hits == {0, 1}, "'search' must be found in exactly docs 0 and 1"  # => 'search' must be found in exactly docs 0 and 1
    assert cooking_hits == {2}, "'cooking' must be found in exactly doc 2"  # => 'cooking' must be found in exactly doc 2
    assert index.postings["search"][0] == 2, "doc 0's tf for 'search' must be 2, since it appears twice"  # => doc 0's tf for 'search' must be 2, since it appears twice
    assert index.doc_lengths == {0: 3, 1: 2, 2: 2}, "doc_lengths must record each document's own token count"  # => doc_lengths must record each document's own token count
    print("MATCH: the typed InvertedIndex class returns correct doc sets and per-document lengths")  # => shows MATCH: the typed InvertedIndex class returns correct doc sets and per-document lengths
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

query('search'): [0, 1]
query('cooking'): [2]
doc_lengths: {0: 3, 1: 2, 2: 2}
MATCH: the typed InvertedIndex class returns correct doc sets and per-document lengths

# pyright: strict
"""Example 59: Incremental Add (co-29)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
from dataclasses import dataclass, field  # => from dataclasses: dataclass, field
 
@dataclass  # => part of this step's computation, continued from the line above
class InvertedIndex:  # => part of this step's computation, continued from the line above
    """A typed, mutable inverted index: term -> {doc_id: term_frequency}."""
 
    postings: dict[str, dict[int, int]] = field(default_factory=lambda: {})  # => postings = field(default_factory=lambda: {})
    doc_lengths: dict[int, int] = field(default_factory=lambda: {})  # => doc lengths = field(default_factory=lambda: {})
 
    def add(self, doc_id: int, tokens: list[str]) -> None:  # => index one document's tokens -- updates postings AND doc_lengths
        """Index one document's tokens -- updates postings AND doc_lengths."""
        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
        for term, count in tf.items():  # => iterates one item at a time
            self.postings.setdefault(term, {})[doc_id] = count  # => part of this step's computation, continued from the line above
        self.doc_lengths[doc_id] = len(tokens)  # => self = len(tokens)
 
    def query(self, term: str) -> set[int]:  # => return every doc-id containing term
        """Return every doc-id containing term."""
        return set(self.postings.get(term, {}).keys())  # => returns set(self.postings.get(term, {}).keys())
 
 
def main() -> None:  # => defines main
    index = InvertedIndex()  # => co-01: an already "built" index, before the new document arrives
    index.add(0, ["search", "engine"])  # => part of this step's computation, continued from the line above
    index.add(1, ["cooking", "recipe"])  # => part of this step's computation, continued from the line above
 
    before: set[int] = index.query("ranking")  # => co-29: the term does not exist YET
    print(f"query('ranking') before incremental add: {sorted(before)}")  # => shows query('ranking') before incremental add
    assert before == set(), "'ranking' must be unfindable before any document containing it is added"  # => 'ranking' must be unfindable before any document containing it is added
 
    index.add(2, ["ranking", "algorithm"])  # => co-29: incremental add -- NO rebuild of docs 0 or 1
    after: set[int] = index.query("ranking")  # => co-29: the SAME index object, queried again
    print(f"query('ranking') after incremental add: {sorted(after)}")  # => shows query('ranking') after incremental add
 
    assert after == {2}, "'ranking' must be immediately findable in the newly added doc 2"  # => 'ranking' must be immediately findable in the newly added doc 2
    assert index.query("search") == {0}, "doc 0's postings must be UNCHANGED by the incremental add"  # => doc 0's postings must be UNCHANGED by the incremental add
    print("MATCH: doc 2 became findable immediately, and the pre-existing postings were untouched")  # => shows MATCH: doc 2 became findable immediately, and the pre-existing postings were untouched
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

query('ranking') before incremental add: []
query('ranking') after incremental add: [2]
MATCH: doc 2 became findable immediately, and the pre-existing postings were untouched

# pyright: strict
"""Example 60: Incremental Equals Rebuild (co-29)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
from dataclasses import dataclass, field  # => from dataclasses: dataclass, field
 
@dataclass  # => part of this step's computation, continued from the line above
class InvertedIndex:  # => part of this step's computation, continued from the line above
    """A typed, mutable inverted index: term -> {doc_id: term_frequency}."""
 
    postings: dict[str, dict[int, int]] = field(default_factory=lambda: {})  # => postings = field(default_factory=lambda: {})
    doc_lengths: dict[int, int] = field(default_factory=lambda: {})  # => doc lengths = field(default_factory=lambda: {})
 
    def add(self, doc_id: int, tokens: list[str]) -> None:  # => index one document's tokens -- updates postings AND doc_lengths
        """Index one document's tokens -- updates postings AND doc_lengths."""
        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
        for term, count in tf.items():  # => iterates one item at a time
            self.postings.setdefault(term, {})[doc_id] = count  # => part of this step's computation, continued from the line above
        self.doc_lengths[doc_id] = len(tokens)  # => self = len(tokens)
 
    def query(self, term: str) -> set[int]:  # => return every doc-id containing term
        """Return every doc-id containing term."""
        return set(self.postings.get(term, {}).keys())  # => returns set(self.postings.get(term, {}).keys())
 
 
def build_from_scratch(docs: dict[int, list[str]]) -> InvertedIndex:  # => a single-pass, from-scratch build -- the reference implementation
    """A single-pass, from-scratch build -- the reference implementation."""
    index = InvertedIndex()  # => co-29: starts empty, exactly like the incremental version did
    for doc_id, tokens in docs.items():  # => iterates one item at a time
        index.add(doc_id, tokens)  # => part of this step's computation, continued from the line above
    return index  # => returns index
 
 
def main() -> None:  # => defines main
    docs: dict[int, list[str]] = {  # => docs = {
        0: ["search", "engine", "search"],  # => doc 0's tokens, this fixture's row
        1: ["search", "results"],  # => doc 1's tokens, this fixture's row
        2: ["ranking", "algorithm"],  # => doc 2's tokens, this fixture's row
    }  # => opens/closes this multi-line literal
 
    incremental = InvertedIndex()  # => co-29: built ONE document at a time, in a different order
    incremental.add(2, docs[2])  # => deliberately OUT OF ORDER vs the dict's own iteration order
    incremental.add(0, docs[0])  # => part of this step's computation, continued from the line above
    incremental.add(1, docs[1])  # => part of this step's computation, continued from the line above
 
    rebuilt: InvertedIndex = build_from_scratch(docs)  # => co-29: the from-scratch reference build
    print(f"incremental postings['search']: {incremental.postings.get('search')}")  # => shows incremental postings['search']
    print(f"rebuilt postings['search']:     {rebuilt.postings.get('search')}")  # => shows rebuilt postings['search']
 
    all_terms: set[str] = set(incremental.postings) | set(rebuilt.postings)  # => every term EITHER index knows about
    for term in all_terms:  # => checks EVERY term, not just 'search'
        assert incremental.postings.get(term) == rebuilt.postings.get(term), f"{term!r}: incremental and rebuilt postings must match"  # => {term!r}: incremental and rebuilt postings must match
    assert incremental.doc_lengths == rebuilt.doc_lengths, "doc_lengths must also match between the two builds"  # => doc_lengths must also match between the two builds
    print(f"MATCH: incremental (built out of order) and from-scratch rebuild agree on all {len(all_terms)} terms")  # => shows MATCH: incremental (built out of order) and from-scratch rebuild agree on all
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

incremental postings['search']: {0: 2, 1: 1}
rebuilt postings['search']:     {0: 2, 1: 1}
MATCH: incremental (built out of order) and from-scratch rebuild agree on all 5 terms

# pyright: strict
"""Example 61: Persist JSON (co-30)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
import json  # => stdlib JSON -- postings persistence to/from disk
from pathlib import Path  # => from pathlib: Path
 
 
def save_postings_json(postings: dict[str, dict[int, int]], path: Path) -> None:  # => jSON object keys must be strings -- convert every int doc_id to str before dumping
    """JSON object keys must be strings -- convert every int doc_id to str before dumping."""
    json_safe: dict[str, dict[str, int]] = {  # => co-30: term -> {doc_id AS STRING -> tf}
        term: {str(doc_id): tf for doc_id, tf in doc_tfs.items()} for term, doc_tfs in postings.items()  # => part of this step's computation, continued from the line above
    }  # => opens/closes this multi-line literal
    path.write_text(json.dumps(json_safe), encoding="utf-8")  # => part of this step's computation, continued from the line above
 
 
def load_postings_json(path: Path) -> dict[str, dict[int, int]]:  # => reverse the string-key conversion, restoring int doc_ids
    """Reverse the string-key conversion, restoring int doc_ids."""
    json_safe: dict[str, dict[str, int]] = json.loads(path.read_text(encoding="utf-8"))  # => json safe = json.loads(path.read_text(encoding="utf-8"))
    return {term: {int(doc_id): tf for doc_id, tf in doc_tfs.items()} for term, doc_tfs in json_safe.items()}  # => co-30: str -> int, restored
 
 
def main() -> None:  # => defines main
    postings: dict[str, dict[int, int]] = {  # => a small in-memory index to persist
        "search": {0: 2, 1: 1},  # => entry for 'search'
        "engine": {0: 1},  # => entry for 'engine'
        "cooking": {2: 1},  # => entry for 'cooking'
    }  # => opens/closes this multi-line literal
    path = Path("postings.json")  # => co-30: a real file on disk, in this example's own directory
    save_postings_json(postings, path)  # => co-30: writes the JSON-safe form to disk
    reloaded: dict[str, dict[int, int]] = load_postings_json(path)  # => co-30: reads it back, restoring int keys
    print(f"original:  {postings}")  # => shows original
    print(f"reloaded:  {reloaded}")  # => shows reloaded
    print(f"file size: {path.stat().st_size} bytes")  # => shows file size
 
    assert reloaded == postings, "the reloaded postings must be IDENTICAL to the original in-memory postings"  # => the reloaded postings must be IDENTICAL to the original in-memory postings
    assert isinstance(next(iter(reloaded["search"])), int), "reloaded doc-ids must be int, not str, after loading"  # => reloaded doc-ids must be int, not str, after loading
    print(f"MATCH: postings round-tripped through JSON exactly, with doc-ids correctly restored to int")  # => shows MATCH: postings round-tripped through JSON exactly, with doc-ids correctly restored to int
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

original:  {'search': {0: 2, 1: 1}, 'engine': {0: 1}, 'cooking': {2: 1}}
reloaded:  {'search': {0: 2, 1: 1}, 'engine': {0: 1}, 'cooking': {2: 1}}
file size: 69 bytes
MATCH: postings round-tripped through JSON exactly, with doc-ids correctly restored to int

# pyright: strict
"""Example 62: Persist Binary (co-30)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
import json  # => stdlib JSON -- postings persistence to/from disk
import struct  # => stdlib fixed-width binary packing -- the compact postings encoding
from pathlib import Path  # => from pathlib: Path
 
 
def save_postings_binary(postings: dict[str, dict[int, int]], path: Path) -> None:  # => pack each term's postings as: term length, term bytes, count, then (doc_id, tf) pairs
    """Pack each term's postings as: term length, term bytes, count, then (doc_id, tf) pairs."""
    with path.open("wb") as f:  # => co-30: BINARY mode -- raw bytes, not text
        for term, doc_tfs in postings.items():  # => iterates one item at a time
            term_bytes: bytes = term.encode("utf-8")  # => the term itself, UTF-8 encoded
            f.write(struct.pack("<H", len(term_bytes)))  # => co-30: 2-byte unsigned length prefix
            f.write(term_bytes)  # => part of this step's computation, continued from the line above
            f.write(struct.pack("<H", len(doc_tfs)))  # => how many (doc_id, tf) pairs follow
            for doc_id, tf in sorted(doc_tfs.items()):  # => co-30: SORTED -- a stable, deterministic order
                f.write(struct.pack("<II", doc_id, tf))  # => co-30: two 4-byte unsigned ints, packed
 
 
def load_postings_binary(path: Path) -> dict[str, dict[int, int]]:  # => the exact inverse of save_postings_binary, reading the same fixed layout back
    """The exact inverse of save_postings_binary, reading the same fixed layout back."""
    postings: dict[str, dict[int, int]] = {}  # => co-30: rebuilt term -> {doc_id: tf} structure
    with path.open("rb") as f:  # => part of this step's computation, continued from the line above
        while True:  # => loops while the condition holds
            len_bytes: bytes = f.read(2)  # => the 2-byte term-length prefix, or EOF
            if not len_bytes:  # => true when not len_bytes
                break  # => part of this step's computation, continued from the line above
            (term_len,) = struct.unpack("<H", len_bytes)  # => part of this step's computation, continued from the line above
            term: str = f.read(term_len).decode("utf-8")  # => the term itself, decoded back to str
            (pair_count,) = struct.unpack("<H", f.read(2))  # => part of this step's computation, continued from the line above
            doc_tfs: dict[int, int] = {}  # => starts empty, populated by the loop below
            for _ in range(pair_count):  # => reads EXACTLY pair_count (doc_id, tf) pairs
                doc_id, tf = struct.unpack("<II", f.read(8))  # => part of this step's computation, continued from the line above
                doc_tfs[doc_id] = tf  # => stores this computed value under its key
            postings[term] = doc_tfs  # => stores this computed value under its key
    return postings  # => returns postings
 
 
def main() -> None:  # => defines main
    postings: dict[str, dict[int, int]] = {"search": {0: 2, 1: 1}, "engine": {0: 1}, "cooking": {2: 1}}  # => postings = {"search": {0: 2, 1: 1}, "engine": {0: 1}, "coo...
    binary_path = Path("postings.bin")  # => co-30: the compact binary file
    json_path = Path("postings_compare.json")  # => a JSON version of the SAME data, for a fair size comparison
 
    save_postings_binary(postings, binary_path)  # => co-30: writes the compact encoding
    json_path.write_text(json.dumps({t: {str(d): tf for d, tf in dt.items()} for t, dt in postings.items()}), encoding="utf-8")  # => part of this step's computation, continued from the line above
 
    reloaded: dict[str, dict[int, int]] = load_postings_binary(binary_path)  # => co-30: reads the binary file back
    binary_size: int = binary_path.stat().st_size  # => binary size = binary_path.stat().st_size
    json_size: int = json_path.stat().st_size  # => json size = json_path.stat().st_size
    print(f"binary size: {binary_size} bytes   json size: {json_size} bytes")  # => shows binary size
 
    assert reloaded == postings, "the binary round-trip must reproduce the EXACT original postings"  # => the binary round-trip must reproduce the EXACT original postings
    assert binary_size < json_size, "the compact binary encoding must be SMALLER than the JSON text encoding"  # => the compact binary encoding must be SMALLER than the JSON text encoding
    print(f"MATCH: binary round-trips exactly and is {json_size - binary_size} bytes smaller than JSON for the same data")  # => shows MATCH: binary round-trips exactly and is
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

binary size: 63 bytes   json size: 69 bytes
MATCH: binary round-trips exactly and is 6 bytes smaller than JSON for the same data

# pyright: strict
"""Example 63: Delta-Encode Postings (co-30)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
 
def delta_encode(sorted_doc_ids: list[int]) -> list[int]:  # => replace each doc-id with the GAP since the previous one (first id is its own gap)
    """Replace each doc-id with the GAP since the previous one (first id is its own gap)."""
    if not sorted_doc_ids:  # => true when not sorted_doc_ids
        return []  # => returns []
    deltas: list[int] = [sorted_doc_ids[0]]  # => co-30: the FIRST id is stored as-is, there is no "previous"
    for i in range(1, len(sorted_doc_ids)):  # => every SUBSEQUENT id, one gap at a time
        deltas.append(sorted_doc_ids[i] - sorted_doc_ids[i - 1])  # => records this item, in order
    return deltas  # => returns deltas
 
 
def delta_decode(deltas: list[int]) -> list[int]:  # => the exact inverse: running sum of the gaps reconstructs the original sorted ids
    """The exact inverse: running sum of the gaps reconstructs the original sorted ids."""
    if not deltas:  # => true when not deltas
        return []  # => returns []
    doc_ids: list[int] = [deltas[0]]  # => the first delta IS the first doc-id
    for gap in deltas[1:]:  # => every SUBSEQUENT gap, added cumulatively
        doc_ids.append(doc_ids[-1] + gap)  # => records this item, in order
    return doc_ids  # => returns doc_ids
 
 
def packed_size(values: list[int]) -> int:  # => a 1-byte-per-value encoding when every value fits in a byte, else 4 bytes -- a size proxy
    """A 1-byte-per-value encoding when every value fits in a byte, else 4 bytes -- a size proxy."""
    return sum(1 if v < 256 else 4 for v in values)  # => co-30: SMALL deltas pack into 1 byte, large raw ids need 4
 
 
def main() -> None:  # => defines main
    dense_doc_ids: list[int] = [1000, 1002, 1005, 1009, 1014, 1020]  # => CLUSTERED doc-ids -- small gaps between them
 
    deltas: list[int] = delta_encode(dense_doc_ids)  # => co-30: [1000, 2, 3, 4, 5, 6] -- mostly tiny numbers
    decoded: list[int] = delta_decode(deltas)  # => co-30: reconstructs the original list, exactly
    print(f"original doc-ids: {dense_doc_ids}")  # => shows original doc-ids
    print(f"deltas:           {deltas}")  # => shows deltas
    print(f"decoded:          {decoded}")  # => shows decoded
 
    raw_size: int = packed_size(dense_doc_ids)  # => how many bytes the RAW ids would need
    delta_size: int = packed_size(deltas)  # => how many bytes the DELTAS need
    print(f"raw size: {raw_size} bytes   delta size: {delta_size} bytes")  # => shows raw size
 
    assert decoded == dense_doc_ids, "delta_decode(delta_encode(x)) must reconstruct x exactly"  # => delta_decode(delta_encode(x)) must reconstruct x exactly
    assert delta_size < raw_size, "delta-encoded CLUSTERED ids must pack smaller than the raw ids"  # => delta-encoded CLUSTERED ids must pack smaller than the raw ids
    print(f"MATCH: exact round-trip, and delta encoding saves {raw_size - delta_size} bytes on this clustered posting list")  # => shows MATCH: exact round-trip, and delta encoding saves
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

original doc-ids: [1000, 1002, 1005, 1009, 1014, 1020]
deltas:           [1000, 2, 3, 4, 5, 6]
decoded:          [1000, 1002, 1005, 1009, 1014, 1020]
raw size: 24 bytes   delta size: 9 bytes
MATCH: exact round-trip, and delta encoding saves 15 bytes on this clustered posting list

# pyright: strict
"""Example 64: avgdl Incremental (co-18, co-29)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
from dataclasses import dataclass  # => from dataclasses: dataclass
 
 
@dataclass  # => part of this step's computation, continued from the line above
class RunningAvgdl:  # => part of this step's computation, continued from the line above
    """Tracks avgdl incrementally: O(1) per add, instead of re-scanning every document."""
 
    total_length: int = 0  # => co-18: sum of every document's length seen so far
    doc_count: int = 0  # => co-29: how many documents contribute to that sum
 
    def add(self, doc_length: int) -> None:  # => fold ONE new document's length into the running totals -- no rescan needed
        """Fold ONE new document's length into the running totals -- no rescan needed."""
        self.total_length += doc_length  # => co-29: O(1) update, not O(N) recomputation
        self.doc_count += 1  # => part of this step's computation, continued from the line above
 
    @property  # => part of this step's computation, continued from the line above
    def avgdl(self) -> float:  # => the current average document length, computed from the running totals
        """The current average document length, computed from the running totals."""
        return self.total_length / self.doc_count if self.doc_count else 0.0  # => co-18: BM25's own avgdl
 
 
def recompute_avgdl_from_scratch(doc_lengths: list[int]) -> float:  # => the reference: recompute avgdl by fully re-scanning every document length
    """The reference: recompute avgdl by fully re-scanning every document length."""
    return sum(doc_lengths) / len(doc_lengths) if doc_lengths else 0.0  # => returns sum(doc_lengths) / len(doc_lengths) if doc_lengths else 0.0
 
 
def main() -> None:  # => defines main
    running = RunningAvgdl()  # => co-18: starts with total_length=0, doc_count=0
    doc_lengths: list[int] = [10, 25, 8, 40, 15]  # => 5 documents, added ONE AT A TIME
 
    for length in doc_lengths:  # => co-29: incremental add, exactly as a live index would receive documents
        running.add(length)  # => part of this step's computation, continued from the line above
        print(f"after adding doc of length {length}: running avgdl={running.avgdl:.4f}")  # => shows after adding doc of length
 
    from_scratch: float = recompute_avgdl_from_scratch(doc_lengths)  # => co-18: the reference, recomputed fully
    print(f"final running avgdl:   {running.avgdl:.4f}")  # => shows final running avgdl
    print(f"from-scratch avgdl:    {from_scratch:.4f}")  # => shows from-scratch avgdl
 
    assert running.avgdl == from_scratch, "the incrementally-maintained avgdl must equal a full recomputation"  # => the incrementally-maintained avgdl must equal a full recomputation
    print(f"MATCH: the running avgdl ({running.avgdl:.4f}) exactly equals the from-scratch recomputation ({from_scratch:.4f})")  # => shows MATCH: the running avgdl (
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

after adding doc of length 10: running avgdl=10.0000
after adding doc of length 25: running avgdl=17.5000
after adding doc of length 8: running avgdl=14.3333
after adding doc of length 40: running avgdl=20.7500
after adding doc of length 15: running avgdl=19.6000
final running avgdl:   19.6000
from-scratch avgdl:    19.6000
MATCH: the running avgdl (19.6000) exactly equals the from-scratch recomputation (19.6000)

# pyright: strict
"""Example 65: BM25F: Fields (co-31)."""
 
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 bm25f_score(  # => defines bm25f score
    tf_by_field: dict[str, int], field_weights: dict[str, float], term_df: int, n_docs: int, k1: float = 1.2  # => part of this step's computation, continued from the line above
) -> float:  # => part of this step's computation, continued from the line above
    """BM25F: combine WEIGHTED tf across fields FIRST, saturate ONCE on the combined total."""
    weighted_tf: float = sum(tf_by_field.get(field, 0) * field_weights[field] for field in field_weights)  # => co-31: combine-first
    idf: float = math.log((n_docs - term_df + 0.5) / (term_df + 0.5))  # => co-16: the SAME RSJ idf, applied once
    return idf * (weighted_tf * (k1 + 1)) / (weighted_tf + k1)  # => co-31: ONE saturation curve over the combined tf
 
 
def main() -> None:  # => defines main
    field_weights: dict[str, float] = {"title": 3.0, "body": 1.0}  # => co-31: title matches count 3x as much as body matches
    term_df, n_docs = 3, 10  # => part of this step's computation, continued from the line above
 
    title_match: dict[str, int] = {"title": 1, "body": 0}  # => the term appears ONCE, in the title only
    body_match: dict[str, int] = {"title": 0, "body": 1}  # => the term appears ONCE, in the body only (equal RAW tf)
 
    title_score: float = bm25f_score(title_match, field_weights, term_df, n_docs)  # => co-31: weighted by title's 3.0
    body_score: float = bm25f_score(body_match, field_weights, term_df, n_docs)  # => co-31: weighted by body's 1.0
    print(f"title match (raw tf=1): score={title_score:.4f}")  # => shows title match (raw tf=1): score=
    print(f"body match  (raw tf=1): score={body_score:.4f}")  # => shows body match  (raw tf=1): score=
 
    assert title_score > body_score, "with EQUAL raw tf, the TITLE match must outrank the body match (higher field weight)"  # => with EQUAL raw tf, the TITLE match must outrank the body match (higher field weight)
    print(f"MATCH: an equal-tf title match ({title_score:.4f}) outranks the body match ({body_score:.4f}) due to field weighting")  # => shows MATCH: an equal-tf title match (
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

title match (raw tf=1): score=1.1976
body match  (raw tf=1): score=0.7621
MATCH: an equal-tf title match (1.1976) outranks the body match (0.7621) due to field weighting

# pyright: strict
"""Example 66: BM25F vs Naive (co-31)."""
 
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 bm25f_combine_first(tf_by_field: dict[str, int], field_weights: dict[str, float], term_df: int, n_docs: int, k1: float = 1.2) -> float:  # => the CORRECT BM25F: sum weighted tf across fields, saturate ONCE
    """The CORRECT BM25F: sum weighted tf across fields, saturate ONCE."""
    weighted_tf: float = sum(tf_by_field.get(field, 0) * field_weights[field] for field in field_weights)  # => weighted tf = sum(tf_by_field.get(field, 0) * field_weights[f...
    idf: float = math.log((n_docs - term_df + 0.5) / (term_df + 0.5))  # => idf = math.log((n_docs - term_df + 0.5) / (term_df + ...
    return idf * (weighted_tf * (k1 + 1)) / (weighted_tf + k1)  # => returns idf * (weighted_tf * (k1 + 1)) / (weighted_tf + k1)
 
 
def naive_saturate_then_sum(tf_by_field: dict[str, int], field_weights: dict[str, float], term_df: int, n_docs: int, k1: float = 1.2) -> float:  # => the BROKEN naive approach: saturate EACH field's weighted tf separately, then add
    """The BROKEN naive approach: saturate EACH field's weighted tf separately, then add."""
    idf: float = math.log((n_docs - term_df + 0.5) / (term_df + 0.5))  # => idf = math.log((n_docs - term_df + 0.5) / (term_df + ...
    total: float = 0.0  # => total = 0.0
    for field, weight in field_weights.items():  # => co-31: saturates PER FIELD -- the bug
        weighted_tf_field: float = tf_by_field.get(field, 0) * weight  # => weighted tf field = tf_by_field.get(field, 0) * weight
        if weighted_tf_field > 0:  # => true when weighted_tf_field > 0
            total += idf * (weighted_tf_field * (k1 + 1)) / (weighted_tf_field + k1)  # => part of this step's computation, continued from the line above
    return total  # => returns total
 
 
def main() -> None:  # => defines main
    field_weights: dict[str, float] = {"title": 1.0, "body": 1.0}  # => EQUAL weights, isolating the combine-order effect
    term_df, n_docs = 3, 10  # => part of this step's computation, continued from the line above
 
    all_in_one_field: dict[str, int] = {"title": 10, "body": 0}  # => tf=10, ALL in one field
    split_evenly: dict[str, int] = {"title": 5, "body": 5}  # => the SAME total tf=10, split 5+5 across two fields
 
    combine_first_one: float = bm25f_combine_first(all_in_one_field, field_weights, term_df, n_docs)  # => co-31: correct BM25F
    combine_first_split: float = bm25f_combine_first(split_evenly, field_weights, term_df, n_docs)  # => combine first split = bm25f_combine_first(split_evenly, field_weights...
    naive_one: float = naive_saturate_then_sum(all_in_one_field, field_weights, term_df, n_docs)  # => the BROKEN naive version
    naive_split: float = naive_saturate_then_sum(split_evenly, field_weights, term_df, n_docs)  # => naive split = naive_saturate_then_sum(split_evenly, field_wei...
    print(f"combine-first: all-in-one={combine_first_one:.4f}  split={combine_first_split:.4f}")  # => shows combine-first: all-in-one=
    print(f"naive:         all-in-one={naive_one:.4f}  split={naive_split:.4f}")  # => shows naive:         all-in-one=
 
    assert math.isclose(combine_first_one, combine_first_split, rel_tol=1e-9), "combine-first must give the SAME score regardless of split -- only total tf matters"  # => combine-first must give the SAME score regardless of split -- only total tf matters
    assert not math.isclose(naive_one, naive_split, rel_tol=1e-9), "the naive version must DISAGREE across splits -- it breaks the invariant"  # => the naive version must DISAGREE across splits -- it breaks the invariant
    assert naive_split > naive_one, "splitting tf evenly must INFLATE the naive score, since saturation resets per field"  # => splitting tf evenly must INFLATE the naive score, since saturation resets per field
    print(f"MATCH: combine-first is split-invariant ({combine_first_one:.4f} both ways); naive is NOT ({naive_one:.4f} vs {naive_split:.4f})")  # => shows MATCH: combine-first is split-invariant (
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

combine-first: all-in-one=1.4971  split=1.4971
naive:         all-in-one=1.4971  split=2.7044
MATCH: combine-first is split-invariant (1.4971 both ways); naive is NOT (1.4971 vs 2.7044)

# pyright: strict
"""Example 67: Fuzzy: Levenshtein (co-32)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
 
def levenshtein(a: str, b: str) -> int:  # => classic O(len(a) * len(b)) edit-distance dynamic-programming table
    """Classic O(len(a) * len(b)) edit-distance dynamic-programming table."""
    rows, cols = len(a) + 1, len(b) + 1  # => co-32: +1 for the "empty prefix" row/column
    dp: list[list[int]] = [[0] * cols for _ in range(rows)]  # => the DP table, all zeros initially
    for i in range(rows):  # => iterates one item at a time
        dp[i][0] = i  # => co-32: transforming a[:i] into "" costs i deletions
    for j in range(cols):  # => iterates one item at a time
        dp[0][j] = j  # => co-32: transforming "" into b[:j] costs j insertions
 
    for i in range(1, rows):  # => fills the table row by row
        for j in range(1, cols):  # => iterates one item at a time
            if a[i - 1] == b[j - 1]:  # => co-32: matching characters cost NOTHING extra
                dp[i][j] = dp[i - 1][j - 1]  # => dp = dp[i - 1][j - 1]
            else:  # => the fallback branch, when no prior condition matched
                dp[i][j] = 1 + min(  # => co-32: 1 + the cheapest of delete, insert, substitute
                    dp[i - 1][j],  # => delete a[i-1]
                    dp[i][j - 1],  # => insert b[j-1]
                    dp[i - 1][j - 1],  # => substitute a[i-1] for b[j-1]
                )  # => opens/closes this multi-line literal
    return dp[rows - 1][cols - 1]  # => co-32: the bottom-right cell holds the FULL-string edit distance
 
 
def main() -> None:  # => defines main
    distance: int = levenshtein("colour", "color")  # => co-32: British vs American spelling
    print(f"levenshtein('colour', 'color') = {distance}")  # => shows levenshtein('colour', 'color') =
 
    identical: int = levenshtein("search", "search")  # => the SAME string -- must be distance 0
    very_different: int = levenshtein("cat", "dog")  # => no shared characters at matching positions
    print(f"levenshtein('search', 'search') = {identical}")  # => shows levenshtein('search', 'search') =
    print(f"levenshtein('cat', 'dog') = {very_different}")  # => shows levenshtein('cat', 'dog') =
 
    assert distance == 1, "'colour' -> 'color' is exactly ONE deletion (the 'u') -- distance must be 1"  # => 'colour' -> 'color' is exactly ONE deletion (the 'u') -- distance must be 1
    assert identical == 0, "identical strings must have distance 0"  # => identical strings must have distance 0
    assert very_different == 3, "'cat' -> 'dog' shares no aligned characters -- distance must be 3 (full substitution)"  # => 'cat' -> 'dog' shares no aligned characters -- distance must be 3 (full substitution)
    print(f"MATCH: colour/color={distance}, identical={identical}, cat/dog={very_different} -- all match hand-verified edit distances")  # => shows MATCH: colour/color=
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

levenshtein('colour', 'color') = 1
levenshtein('search', 'search') = 0
levenshtein('cat', 'dog') = 3
MATCH: colour/color=1, identical=0, cat/dog=3 -- all match hand-verified edit distances

# pyright: strict
"""Example 68: Fuzzy: Damerau (co-32)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
 
def levenshtein(a: str, b: str) -> int:  # => defines levenshtein
    rows, cols = len(a) + 1, len(b) + 1  # => part of this step's computation, continued from the line above
    dp: list[list[int]] = [[0] * cols for _ in range(rows)]  # => dp = [[0] * cols for _ in range(rows)]
    for i in range(rows):  # => iterates one item at a time
        dp[i][0] = i  # => dp = i
    for j in range(cols):  # => iterates one item at a time
        dp[0][j] = j  # => dp = j
    for i in range(1, rows):  # => iterates one item at a time
        for j in range(1, cols):  # => iterates one item at a time
            if a[i - 1] == b[j - 1]:  # => true when a[i - 1] == b[j - 1]
                dp[i][j] = dp[i - 1][j - 1]  # => dp = dp[i - 1][j - 1]
            else:  # => the fallback branch, when no prior condition matched
                dp[i][j] = 1 + min(dp[i - 1][j], dp[i][j - 1], dp[i - 1][j - 1])  # => dp = 1 + min(dp[i - 1][j], dp[i][j - 1], dp[i - 1][j...
    return dp[rows - 1][cols - 1]  # => returns dp[rows - 1][cols - 1]
 
 
def damerau_levenshtein(a: str, b: str) -> int:  # => levenshtein PLUS a transposition operation: swapping two ADJACENT characters costs 1, not 2
    """Levenshtein PLUS a transposition operation: swapping two ADJACENT characters costs 1, not 2."""
    rows, cols = len(a) + 1, len(b) + 1  # => part of this step's computation, continued from the line above
    dp: list[list[int]] = [[0] * cols for _ in range(rows)]  # => dp = [[0] * cols for _ in range(rows)]
    for i in range(rows):  # => iterates one item at a time
        dp[i][0] = i  # => dp = i
    for j in range(cols):  # => iterates one item at a time
        dp[0][j] = j  # => dp = j
 
    for i in range(1, rows):  # => iterates one item at a time
        for j in range(1, cols):  # => iterates one item at a time
            cost: int = 0 if a[i - 1] == b[j - 1] else 1  # => 0 for a match, 1 for a substitution
            dp[i][j] = min(  # => dp = min(
                dp[i - 1][j] + 1,  # => delete
                dp[i][j - 1] + 1,  # => insert
                dp[i - 1][j - 1] + cost,  # => match or substitute
            )  # => opens/closes this multi-line literal
            if (  # => co-32: the EXTRA Damerau case -- an adjacent transposition
                i > 1  # => part of this step's computation, continued from the line above
                and j > 1  # => part of this step's computation, continued from the line above
                and a[i - 1] == b[j - 2]  # => part of this step's computation, continued from the line above
                and a[i - 2] == b[j - 1]  # => part of this step's computation, continued from the line above
            ):  # => part of this step's computation, continued from the line above
                dp[i][j] = min(dp[i][j], dp[i - 2][j - 2] + 1)  # => co-32: ONE swap, not two edits
    return dp[rows - 1][cols - 1]  # => returns dp[rows - 1][cols - 1]
 
 
def main() -> None:  # => defines main
    plain_distance: int = levenshtein("teh", "the")  # => co-32: plain Levenshtein sees 'teh'->'the' as TWO edits
    damerau_distance: int = damerau_levenshtein("teh", "the")  # => co-32: Damerau sees the SAME pair as ONE swap
    print(f"levenshtein('teh', 'the') = {plain_distance}")  # => shows levenshtein('teh', 'the') =
    print(f"damerau_levenshtein('teh', 'the') = {damerau_distance}")  # => shows damerau_levenshtein('teh', 'the') =
 
    assert plain_distance == 2, "plain Levenshtein must score 'teh'->'the' as distance 2 (substitute+substitute, or del+ins)"  # => plain Levenshtein must score 'teh'->'the' as distance 2 (substitute+substitute, or del+ins)
    assert damerau_distance == 1, "Damerau-Levenshtein must score the SAME pair as distance 1 (one adjacent swap)"  # => Damerau-Levenshtein must score the SAME pair as distance 1 (one adjacent swap)
    print(f"MATCH: the identical typo scores {plain_distance} under plain Levenshtein but {damerau_distance} under Damerau -- transpositions are cheap typos")  # => shows MATCH: the identical typo scores
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

levenshtein('teh', 'the') = 2
damerau_levenshtein('teh', 'the') = 1
MATCH: the identical typo scores 2 under plain Levenshtein but 1 under Damerau -- transpositions are cheap typos

# pyright: strict
"""Example 69: Spelling Correct (co-32)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
 
def levenshtein(a: str, b: str) -> int:  # => defines levenshtein
    rows, cols = len(a) + 1, len(b) + 1  # => part of this step's computation, continued from the line above
    dp: list[list[int]] = [[0] * cols for _ in range(rows)]  # => dp = [[0] * cols for _ in range(rows)]
    for i in range(rows):  # => iterates one item at a time
        dp[i][0] = i  # => dp = i
    for j in range(cols):  # => iterates one item at a time
        dp[0][j] = j  # => dp = j
    for i in range(1, rows):  # => iterates one item at a time
        for j in range(1, cols):  # => iterates one item at a time
            if a[i - 1] == b[j - 1]:  # => true when a[i - 1] == b[j - 1]
                dp[i][j] = dp[i - 1][j - 1]  # => dp = dp[i - 1][j - 1]
            else:  # => the fallback branch, when no prior condition matched
                dp[i][j] = 1 + min(dp[i - 1][j], dp[i][j - 1], dp[i - 1][j - 1])  # => dp = 1 + min(dp[i - 1][j], dp[i][j - 1], dp[i - 1][j...
    return dp[rows - 1][cols - 1]  # => returns dp[rows - 1][cols - 1]
 
 
def suggest_correction(misspelled: str, dictionary: list[str]) -> str:  # => return the dictionary term with the SMALLEST edit distance to misspelled
    """Return the dictionary term with the SMALLEST edit distance to misspelled."""
    best_term: str = dictionary[0]  # => co-32: starts with the first candidate as the running best
    best_distance: int = levenshtein(misspelled, best_term)  # => best distance = levenshtein(misspelled, best_term)
    for term in dictionary[1:]:  # => checks EVERY remaining candidate
        distance: int = levenshtein(misspelled, term)  # => distance = levenshtein(misspelled, term)
        if distance < best_distance:  # => co-32: strictly BETTER -- replaces the running best
            best_term, best_distance = term, distance  # => part of this step's computation, continued from the line above
    return best_term  # => returns best_term
 
 
def main() -> None:  # => defines main
    dictionary: list[str] = ["search", "engine", "ranking", "index", "query"]  # => the known-good vocabulary
    misspelled: str = "serch"  # => a typo of "search" -- missing the 'a'
 
    suggestion: str = suggest_correction(misspelled, dictionary)  # => co-32: the closest dictionary term
    print(f"misspelled: {misspelled!r}  suggestion: {suggestion!r}")  # => shows misspelled
 
    hand_distances: dict[str, int] = {term: levenshtein(misspelled, term) for term in dictionary}  # => an INDEPENDENT recount of every distance
    for term, dist in sorted(hand_distances.items(), key=lambda kv: kv[1]):  # => iterates one item at a time
        print(f"  distance to {term!r}: {dist}")  # => shows distance to
 
    hand_best: str = min(hand_distances, key=lambda t: hand_distances[t])  # => the minimum, computed a DIFFERENT way
    assert suggestion == hand_best, "suggest_correction's result must equal the independently-recomputed minimum"  # => suggest_correction's result must equal the independently-recomputed minimum
    assert suggestion == "search", "'serch' must correct to 'search', its true nearest dictionary neighbor"  # => 'serch' must correct to 'search', its true nearest dictionary neighbor
    print(f"MATCH: suggest_correction's {suggestion!r} equals the independently verified minimum-distance term")  # => shows MATCH: suggest_correction's
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

misspelled: 'serch'  suggestion: 'search'
  distance to 'search': 1
  distance to 'query': 4
  distance to 'index': 5
  distance to 'engine': 6
  distance to 'ranking': 7
MATCH: suggest_correction's 'search' equals the independently verified minimum-distance term

# pyright: strict
"""Example 70: Edge N-Gram Autocomplete (co-33)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
 
def edge_ngrams(term: str, min_len: int = 1) -> list[str]:  # => every PREFIX of term from length min_len up to the full term
    """Every PREFIX of term from length min_len up to the full term."""
    return [term[:i] for i in range(min_len, len(term) + 1)]  # => co-33: 's', 'se', 'sea', ..., the full term
 
 
def build_edge_ngram_index(docs: dict[int, str]) -> dict[str, set[int]]:  # => index every document's own single term by ALL of its edge n-grams
    """Index every document's own single term by ALL of its edge n-grams."""
    index: dict[str, set[int]] = {}  # => co-33: prefix -> {doc-ids whose term starts with this prefix}
    for doc_id, term in docs.items():  # => iterates one item at a time
        for prefix in edge_ngrams(term):  # => co-33: EVERY prefix of this document's term gets indexed
            index.setdefault(prefix, set()).add(doc_id)  # => part of this step's computation, continued from the line above
    return index  # => returns index
 
 
def main() -> None:  # => defines main
    docs: dict[int, str] = {0: "search", 1: "season", 2: "cooking"}  # => 3 single-term "documents" (e.g. autocomplete entries)
    index: dict[str, set[int]] = build_edge_ngram_index(docs)  # => co-33: this vocabulary's own edge n-gram index
 
    partial_query: str = "sea"  # => a user typing "sea" -- not yet a complete word
    hits: set[int] = index.get(partial_query, set())  # => co-33: everything whose FULL term starts with "sea"
    print(f"edge n-grams of 'search': {edge_ngrams('search')}")  # => shows edge n-grams of 'search'
    print(f"query {partial_query!r} matches docs: {sorted(hits)}")  # => shows query
 
    assert hits == {0, 1}, "'sea' must match BOTH 'search' (doc 0) and 'season' (doc 1) -- both start with it"  # => 'sea' must match BOTH 'search' (doc 0) and 'season' (doc 1) -- both start with it
    assert 2 not in hits, "'cooking' (doc 2) must NOT match -- it does not start with 'sea'"  # => 'cooking' (doc 2) must NOT match -- it does not start with 'sea'
    print(f"MATCH: typing 'sea' retrieves both prefix-matching terms, correctly excluding the unrelated one")  # => shows MATCH: typing 'sea' retrieves both prefix-matching terms, correctly excluding the unrelated one
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

edge n-grams of 'search': ['s', 'se', 'sea', 'sear', 'searc', 'search']
query 'sea' matches docs: [0, 1]
MATCH: typing 'sea' retrieves both prefix-matching terms, correctly excluding the unrelated one

# pyright: strict
"""Example 71: Char N-Gram Substring (co-33)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
 
def char_ngrams(term: str, n: int = 3) -> list[str]:  # => every contiguous n-character substring of term -- covers the WHOLE word, not just its start
    """Every contiguous n-character substring of term -- covers the WHOLE word, not just its start."""
    if len(term) < n:  # => true when len(term) < n
        return [term]  # => co-33: too short for a full n-gram -- the whole term is its own single gram
    return [term[i : i + n] for i in range(len(term) - n + 1)]  # => co-33: sliding window of width n
 
 
def build_char_ngram_index(docs: dict[int, str], n: int = 3) -> dict[str, set[int]]:  # => defines build char ngram index
    index: dict[str, set[int]] = {}  # => co-33: n-gram -> {doc-ids whose term contains this n-gram}
    for doc_id, term in docs.items():  # => iterates one item at a time
        for gram in char_ngrams(term, n):  # => co-33: EVERY trigram of this document's term
            index.setdefault(gram, set()).add(doc_id)  # => part of this step's computation, continued from the line above
    return index  # => returns index
 
 
def main() -> None:  # => defines main
    docs: dict[int, str] = {0: "search", 1: "research", 2: "cooking"}  # => "arch" is INTERIOR to both 0 and 1
    index: dict[str, set[int]] = build_char_ngram_index(docs, n=3)  # => co-33: this vocabulary's own trigram index
 
    interior_query: str = "arch"  # => co-33: appears in the MIDDLE of "search" and "research", not at the start
    query_grams: list[str] = char_ngrams(interior_query, n=3)  # => co-33: 'arc', 'rch' -- trigrams of the query itself
    hits: set[int] = set[int]().union(*(index.get(g, set()) for g in query_grams))  # => co-33: docs matching ANY query trigram
    print(f"trigrams of 'search': {char_ngrams('search', n=3)}")  # => shows trigrams of 'search'
    print(f"trigrams of query 'arch': {query_grams}")  # => shows trigrams of query 'arch'
    print(f"interior substring query {interior_query!r} matches docs: {sorted(hits)}")  # => shows interior substring query
 
    assert 0 in hits and 1 in hits, "'arch' is an INTERIOR substring of both 'search' and 'research' -- both must match"  # => 'arch' is an INTERIOR substring of both 'search' and 'research' -- both must match
    assert 2 not in hits, "'cooking' contains no trigram of 'arch' -- it must NOT match"  # => 'cooking' contains no trigram of 'arch' -- it must NOT match
    print(f"MATCH: an interior substring query found both containing documents via shared character n-grams")  # => shows MATCH: an interior substring query found both containing documents via shared character n-grams
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

trigrams of 'search': ['sea', 'ear', 'arc', 'rch']
trigrams of query 'arch': ['arc', 'rch']
interior substring query 'arch' matches docs: [0, 1]
MATCH: an interior substring query found both containing documents via shared character n-grams

# pyright: strict
"""Example 72: Synonym Expand (co-34)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
 
def expand_query(query_terms: list[str], synonyms: dict[str, set[str]]) -> set[str]:  # => expand every query term into itself PLUS all of its known synonyms
    """Expand every query term into itself PLUS all of its known synonyms."""
    expanded: set[str] = set()  # => co-34: the query terms, UNIONED with every synonym they map to
    for term in query_terms:  # => iterates one item at a time
        expanded.add(term)  # => the ORIGINAL term always stays in the expanded set
        expanded |= synonyms.get(term, set())  # => co-34: plus every synonym registered for it
    return expanded  # => returns expanded
 
 
def execute_or_query(index: dict[str, set[int]], terms: set[str]) -> set[int]:  # => boolean OR over an arbitrary set of terms -- any document matching any term qualifies
    """Boolean OR over an arbitrary set of terms -- any document matching any term qualifies."""
    hits: set[int] = set()  # => hits = set()
    for term in terms:  # => iterates one item at a time
        hits |= index.get(term, set())  # => co-03: boolean OR merge across query terms
    return hits  # => returns hits
 
 
def main() -> None:  # => defines main
    synonyms: dict[str, set[str]] = {"car": {"automobile", "vehicle"}}  # => co-34: "car" expands to include these
    index: dict[str, set[int]] = {  # => a small inverted index -- doc 1 uses ONLY the synonym, never "car" itself
        "car": {0},  # => entry for 'car'
        "automobile": {1},  # => entry for 'automobile'
        "bicycle": {2},  # => entry for 'bicycle'
    }  # => opens/closes this multi-line literal
 
    literal_hits: set[int] = execute_or_query(index, {"car"})  # => co-34: WITHOUT synonym expansion
    expanded_terms: set[str] = expand_query(["car"], synonyms)  # => co-34: "car" expanded to {"car", "automobile", "vehicle"}
    expanded_hits: set[int] = execute_or_query(index, expanded_terms)  # => co-34: WITH synonym expansion
    print(f"literal query 'car': {sorted(literal_hits)}")  # => shows literal query 'car'
    print(f"expanded terms: {sorted(expanded_terms)}")  # => shows expanded terms
    print(f"expanded query: {sorted(expanded_hits)}")  # => shows expanded query
 
    assert literal_hits == {0}, "the literal query must find only doc 0, which contains the exact word 'car'"  # => the literal query must find only doc 0, which contains the exact word 'car'
    assert 1 in expanded_hits, "doc 1 (which contains ONLY 'automobile') must be found once synonyms are expanded"  # => doc 1 (which contains ONLY 'automobile') must be found once synonyms are expanded
    assert expanded_hits == {0, 1}, "the expanded query must find exactly docs 0 and 1"  # => the expanded query must find exactly docs 0 and 1
    print(f"MATCH: synonym expansion found doc 1's 'automobile' even though the query literally said 'car'")  # => shows MATCH: synonym expansion found doc 1's 'automobile' even though the query literally said 'car'
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

literal query 'car': [0]
expanded terms: ['automobile', 'car', 'vehicle']
expanded query: [0, 1]
MATCH: synonym expansion found doc 1's 'automobile' even though the query literally said 'car'

# pyright: strict
"""Example 73: Semantic Embedding Cosine (co-35)."""
 
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 cosine_similarity(a: list[float], b: list[float]) -> float:  # => the Example 27 formula again: dot(a, b) / (|a| * |b|)
    """The Example 27 formula again: dot(a, b) / (|a| * |b|)."""
    dot: float = sum(x * y for x, y in zip(a, b))  # => co-27: the dot product
    norm_a: float = math.sqrt(sum(x * x for x in a))  # => co-27: |a|, the Euclidean length of a
    norm_b: float = math.sqrt(sum(y * y for y in b))  # => co-27: |b|, the Euclidean length of b
    if norm_a == 0 or norm_b == 0:  # => true when norm_a == 0 or norm_b == 0
        return 0.0  # => returns 0.0
    return dot / (norm_a * norm_b)  # => returns dot / (norm_a * norm_b)
 
 
def main() -> None:  # => defines main
    # TOY 3-dimensional embeddings, hand-picked to encode a "vehicle-ness" concept -- NOT from a real model.
    query_vec: list[float] = [0.9, 0.1, 0.0]  # => co-35: a query about "cars," represented as a toy vector
    doc_vectors: dict[int, list[float]] = {  # => doc vectors = {
        0: [0.85, 0.15, 0.05],  # => doc 0: "automobile" -- semantically close to the query, despite zero word overlap
        1: [0.05, 0.10, 0.95],  # => doc 1: "recipe" -- semantically distant
    }  # => opens/closes this multi-line literal
 
    scores: dict[int, float] = {  # => co-35: EVERY doc's cosine similarity to the query vector
        doc_id: cosine_similarity(query_vec, vec) for doc_id, vec in doc_vectors.items()  # => part of this step's computation, continued from the line above
    }  # => opens/closes this multi-line literal
    ranking: list[int] = sorted(scores, key=lambda d: -scores[d])  # => co-20: best (highest) similarity first
    for doc_id in ranking:  # => iterates one item at a time
        print(f"doc {doc_id}: cosine={scores[doc_id]:.4f}")  # => shows doc
 
    assert ranking[0] == 0, "the semantically NEAR document (doc 0) must rank first, by cosine similarity alone"  # => the semantically NEAR document (doc 0) must rank first, by cosine similarity alone
    assert scores[0] > scores[1], "doc 0's similarity must be strictly higher than doc 1's"  # => doc 0's similarity must be strictly higher than doc 1's
    print(f"MATCH: the nearest-meaning document ranks first ({scores[0]:.4f} vs {scores[1]:.4f}) using only vector cosine, no shared words")  # => shows MATCH: the nearest-meaning document 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: cosine=0.9963
doc 1: cosine=0.0635
MATCH: the nearest-meaning document ranks first (0.9963 vs 0.0635) using only vector cosine, no shared words

# pyright: strict
"""Example 74: ANN vs Exact kNN (co-35)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
import math  # => stdlib math -- log/sqrt for idf, cosine, and skip-pointer spacing
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 cosine_similarity(a: list[float], b: list[float]) -> float:  # => defines cosine similarity
    dot: float = sum(x * y for x, y in zip(a, b))  # => dot = sum(x * y for x, y in zip(a, b))
    norm_a: float = math.sqrt(sum(x * x for x in a))  # => norm a = math.sqrt(sum(x * x for x in a))
    norm_b: float = math.sqrt(sum(y * y for y in b))  # => norm b = math.sqrt(sum(y * y for y in b))
    return dot / (norm_a * norm_b) if norm_a and norm_b else 0.0  # => returns dot / (norm_a * norm_b) if norm_a and norm_b else 0.0
 
 
def exact_knn(query: list[float], vectors: dict[int, list[float]], k: int) -> list[int]:  # => brute-force: score EVERY vector, take the true top-k -- always correct, always O(N)
    """Brute-force: score EVERY vector, take the true top-k -- always correct, always O(N)."""
    scored: list[tuple[int, float]] = [(doc_id, cosine_similarity(query, v)) for doc_id, v in vectors.items()]  # => scored = [(doc_id, cosine_similarity(query, v)) for doc_...
    scored.sort(key=lambda dv: -dv[1])  # => co-35: full sort -- the GROUND TRUTH ranking
    return [doc_id for doc_id, _ in scored[:k]]  # => returns [doc_id for doc_id, _ in scored[:k]]
 
 
def toy_ann(query: list[float], vectors: dict[int, list[float]], k: int, sample_size: int, seed: int) -> list[int]:  # => a TOY approximate search: rank only a random SAMPLE, not the full vector set -- NOT production-grade
    """A TOY approximate search: rank only a random SAMPLE, not the full vector set -- NOT production-grade."""
    rng = random.Random(seed)  # => co-35: fixed seed -- reproducible sampling
    candidate_ids: list[int] = rng.sample(list(vectors), min(sample_size, len(vectors)))  # => co-35: a random SUBSET, not everything
    candidates: dict[int, list[float]] = {doc_id: vectors[doc_id] for doc_id in candidate_ids}  # => candidates = {doc_id: vectors[doc_id] for doc_id in candidat...
    return exact_knn(query, candidates, k)  # => exact search, but only over the sampled subset
 
 
def main() -> None:  # => defines main
    rng = random.Random(3)  # => fixed seed -- reproducible corpus
    n_docs: int = 200  # => a larger toy corpus, to make sampling's trade-off visible
    vectors: dict[int, list[float]] = {i: [rng.uniform(-1, 1) for _ in range(4)] for i in range(n_docs)}  # => vectors = {i: [rng.uniform(-1, 1) for _ in range(4)] for ...
    query: list[float] = [rng.uniform(-1, 1) for _ in range(4)]  # => query = [rng.uniform(-1, 1) for _ in range(4)]
 
    exact_top5: list[int] = exact_knn(query, vectors, k=5)  # => co-35: the GROUND TRUTH top 5
    ann_top5: list[int] = toy_ann(query, vectors, k=5, sample_size=20, seed=3)  # => co-35: an approximation over only 20 of 200
    overlap: set[int] = set(exact_top5) & set(ann_top5)  # => how many of the approximate results are ALSO in the true top 5
    print(f"exact top-5: {exact_top5}")  # => shows exact top-5
    print(f"toy ANN top-5 (sampled 20/200): {ann_top5}")  # => shows toy ANN top-5 (sampled 20/200)
    print(f"overlap with ground truth: {len(overlap)}/5")  # => shows overlap with ground truth
 
    assert 0 < len(overlap) < 5, "at this fixed seed the toy ANN must recover SOME but not ALL of the exact top-5 (a falsifiable recall bound, not a set-cardinality tautology)"  # => co-35: this fails if recall COLLAPSES to 0 or the sample ACCIDENTALLY matches perfectly -- both are real, checkable outcomes
 
    exact_time: float = min(_time_call(lambda: exact_knn(query, vectors, k=5)) for _ in range(3))  # => co-35: best of 3 runs -- exact_knn scores ALL 200 vectors
    ann_time: float = min(_time_call(lambda: toy_ann(query, vectors, k=5, sample_size=20, seed=3)) for _ in range(3))  # => co-35: best of 3 runs -- toy_ann scores only the sampled 20
    print(f"speed: exact={exact_time * 1000:.3f}ms  toy_ann={ann_time * 1000:.3f}ms  ratio={exact_time / ann_time:.1f}x")  # => shows the wall-clock trade-off the recall bound above does NOT capture
 
    assert ann_time < exact_time, "the toy ANN must be FASTER than exact search -- it scores 20 candidates, not 200"  # => the toy ANN must be FASTER than exact search -- it scores 20 candidates, not 200
    # NOTE: production systems use HNSW (Hierarchical Navigable Small World graphs), not random sampling --
    # this toy is built ONLY to demonstrate the recall/speed trade-off in isolation, in pure Python.
    if overlap:  # => gate the MATCH banner on the REAL recall bound above, not an unconditional print
        print(f"MATCH: the toy ANN recovered {len(overlap)}/5 of the true nearest neighbors from a 10x-smaller candidate pool, {exact_time / ann_time:.1f}x faster")  # => shows MATCH: the toy ANN recovered
    else:
        print("NO MATCH: the toy ANN recovered 0/5 of the true nearest neighbors -- recall collapsed at this sample size")  # => shows NO MATCH: recall collapsed at this sample size
 
 
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

exact top-5: [137, 160, 61, 22, 154]
toy ANN top-5 (sampled 20/200): [160, 154, 49, 148, 151]
overlap with ground truth: 2/5
speed: exact=0.164ms  toy_ann=0.028ms  ratio=6.0x
MATCH: the toy ANN recovered 2/5 of the true nearest neighbors from a 10x-smaller candidate pool, 6.0x faster

# pyright: strict
"""Example 75: Hybrid Lexical + Vector (co-35)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
 
def normalize(scores: dict[int, float]) -> dict[int, float]:  # => min-max normalize a score dict into [0, 1] so lexical and vector scores are comparable
    """Min-max normalize a score dict into [0, 1] so lexical and vector scores are comparable."""
    if not scores:  # => true when not scores
        return {}  # => returns {}
    lo, hi = min(scores.values()), max(scores.values())  # => co-35: the observed range, this corpus's own
    if hi == lo:  # => true when hi == lo
        return {doc_id: 1.0 for doc_id in scores}  # => returns {doc_id: 1.0 for doc_id in scores}
    return {doc_id: (s - lo) / (hi - lo) for doc_id, s in scores.items()}  # => co-35: rescaled into [0, 1]
 
 
def hybrid_rank(bm25_scores: dict[int, float], vector_scores: dict[int, float], alpha: float = 0.5) -> list[int]:  # => alpha blends the two normalized signals -- alpha=1 is pure lexical, alpha=0 is pure vector
    """alpha blends the two normalized signals -- alpha=1 is pure lexical, alpha=0 is pure vector."""
    norm_bm25: dict[int, float] = normalize(bm25_scores)  # => co-35: both signals rescaled to the SAME [0, 1] range
    norm_vector: dict[int, float] = normalize(vector_scores)  # => norm vector = normalize(vector_scores)
    combined: dict[int, float] = {  # => combined = {
        doc_id: alpha * norm_bm25.get(doc_id, 0.0) + (1 - alpha) * norm_vector.get(doc_id, 0.0)  # => co-35: the weighted blend
        for doc_id in set(bm25_scores) | set(vector_scores)  # => part of this step's computation, continued from the line above
    }  # => opens/closes this multi-line literal
    return sorted(combined, key=lambda d: -combined[d])  # => returns sorted(combined, key=lambda d: -combined[d])
 
 
def main() -> None:  # => defines main
    bm25_scores: dict[int, float] = {0: 5.0, 1: 1.0, 2: 0.2}  # => pure LEXICAL ranking: 0 > 1 > 2
    vector_scores: dict[int, float] = {0: 0.1, 1: 0.9, 2: 0.95}  # => pure VECTOR ranking: 2 > 1 > 0 -- DISAGREES with BM25
 
    pure_bm25_ranking: list[int] = sorted(bm25_scores, key=lambda d: -bm25_scores[d])  # => the lexical-only order, for comparison
    hybrid: list[int] = hybrid_rank(bm25_scores, vector_scores, alpha=0.5)  # => co-35: an EQUAL 50/50 blend
    print(f"pure BM25 ranking: {pure_bm25_ranking}")  # => shows pure BM25 ranking
    print(f"hybrid ranking (alpha=0.5): {hybrid}")  # => shows hybrid ranking (alpha=0.5)
 
    assert pure_bm25_ranking == [0, 1, 2], "pure BM25 must rank doc 0 first, given its own scores"  # => pure BM25 must rank doc 0 first, given its own scores
    assert hybrid != pure_bm25_ranking, "the hybrid ranking must DIFFER from pure BM25 when the two signals disagree"  # => the hybrid ranking must DIFFER from pure BM25 when the two signals disagree
    print(f"MATCH: blending in the vector signal changed the ranking from {pure_bm25_ranking} to {hybrid}")  # => shows MATCH: blending in the vector signal changed the ranking from
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

pure BM25 ranking: [0, 1, 2]
hybrid ranking (alpha=0.5): [1, 0, 2]
MATCH: blending in the vector signal changed the ranking from [0, 1, 2] to [1, 0, 2]

# pyright: strict
"""Example 76: Lexical Miss, Semantic Win (co-35)."""
 
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 cosine_similarity(a: list[float], b: list[float]) -> float:  # => defines cosine similarity
    dot: float = sum(x * y for x, y in zip(a, b))  # => dot = sum(x * y for x, y in zip(a, b))
    norm_a: float = math.sqrt(sum(x * x for x in a))  # => norm a = math.sqrt(sum(x * x for x in a))
    norm_b: float = math.sqrt(sum(y * y for y in b))  # => norm b = math.sqrt(sum(y * y for y in b))
    return dot / (norm_a * norm_b) if norm_a and norm_b else 0.0  # => returns dot / (norm_a * norm_b) if norm_a and norm_b else 0.0
 
 
def lexical_search(index: dict[str, set[int]], query_term: str) -> set[int]:  # => a plain co-04 boolean lookup -- exact term match only
    """A plain co-04 boolean lookup -- exact term match only."""
    return index.get(query_term, set())  # => co-04: no match unless the LITERAL term is present
 
 
def main() -> None:  # => defines main
    index: dict[str, set[int]] = {"automobile": {0}, "bicycle": {1}}  # => NO document literally contains "car"
 
    query_term: str = "car"  # => co-35: the user's literal query word
    query_vec: list[float] = [0.9, 0.1]  # => a toy embedding for "car"
    doc_vectors: dict[int, list[float]] = {  # => doc vectors = {
        0: [0.88, 0.12],  # => doc 0's "automobile" embedding -- semantically CLOSE to "car"
        1: [0.05, 0.98],  # => doc 1's "bicycle" embedding -- semantically FAR from "car"
    }  # => opens/closes this multi-line literal
 
    lexical_hits: set[int] = lexical_search(index, query_term)  # => co-04: BM25/boolean search over the literal term
    vector_scores: dict[int, float] = {doc_id: cosine_similarity(query_vec, v) for doc_id, v in doc_vectors.items()}  # => vector scores = {doc_id: cosine_similarity(query_vec, v) for do...
    semantic_ranking: list[int] = sorted(vector_scores, key=lambda d: -vector_scores[d])  # => co-35: ranked by embedding similarity
    print(f"lexical search for 'car': {sorted(lexical_hits)}")  # => shows lexical search for 'car'
    print(f"semantic ranking: {semantic_ranking} (scores: {vector_scores})")  # => shows semantic ranking
 
    assert lexical_hits == set(), "lexical (exact-term) search must find NOTHING -- 'car' appears in no document"  # => lexical (exact-term) search must find NOTHING -- 'car' appears in no document
    assert semantic_ranking[0] == 0, "semantic search must rank doc 0 ('automobile') first, despite zero word overlap"  # => semantic search must rank doc 0 ('automobile') first, despite zero word overlap
    print(f"MATCH: lexical search found {len(lexical_hits)} docs; semantic search correctly surfaced doc 0 anyway")  # => shows MATCH: lexical search found
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

lexical search for 'car': []
semantic ranking: [0, 1] (scores: {0: 0.9996907452347202, 1: 0.16093055829278619})
MATCH: lexical search found 0 docs; semantic search correctly surfaced doc 0 anyway

# pyright: strict
"""Example 77: PageRank Toy (co-36)."""
 
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
 
def pagerank(links: dict[str, list[str]], damping: float = 0.85, iterations: int = 100) -> dict[str, float]:  # => power-iteration PageRank, Brin & Page (1998)'s own random-surfer formulation
    """Power-iteration PageRank, Brin & Page (1998)'s own random-surfer formulation."""
    nodes: list[str] = list(links)  # => co-36: every node in the graph
    n: int = len(nodes)  # => this fixture's own size
    scores: dict[str, float] = {node: 1.0 / n for node in nodes}  # => co-36: uniform starting distribution
 
    for _ in range(iterations):  # => co-36: repeated power-iteration steps, toward convergence
        new_scores: dict[str, float] = {node: (1 - damping) / n for node in nodes}  # => the random-jump base term
        for node, outlinks in links.items():  # => iterates one item at a time
            if not outlinks:  # => co-36: a DANGLING node -- redistribute its mass to ALL nodes, or the total leaks below 1
                for target in nodes:  # => iterates one item at a time
                    new_scores[target] += damping * scores[node] / n  # => part of this step's computation, continued from the line above
                continue  # => part of this step's computation, continued from the line above
            share: float = scores[node] / len(outlinks)  # => co-36: this node's rank, split evenly among its outlinks
            for target in outlinks:  # => iterates one item at a time
                new_scores[target] += damping * share  # => co-36: each linked-to node receives its share
        scores = new_scores  # => scores = new_scores
    return scores  # => returns scores
 
 
def main() -> None:  # => defines main
    links: dict[str, list[str]] = {  # => co-36: A and B and D all point to C -- C should dominate
        "A": ["B", "C"],  # => entry for 'A'
        "B": ["C"],  # => entry for 'B'
        "C": [],  # => entry for 'C'
        "D": ["C"],  # => entry for 'D'
    }  # => opens/closes this multi-line literal
    scores: dict[str, float] = pagerank(links)  # => co-36: the converged PageRank distribution
    for node, score in sorted(scores.items(), key=lambda kv: -kv[1]):  # => iterates one item at a time
        print(f"{node}: {score:.4f}")  # => prints this step's result
 
    total: float = sum(scores.values())  # => co-36: PageRank's own probability-distribution invariant
    top_node: str = max(scores, key=lambda n: scores[n])  # => whichever node scored highest
    print(f"sum of scores: {total:.6f}")  # => shows sum of scores
 
    assert abs(total - 1.0) < 1e-6, "PageRank scores must sum to (approximately) 1.0 -- it's a probability distribution"  # => PageRank scores must sum to (approximately) 1.0 -- it's a probability distribution
    assert top_node == "C", "'C' is linked to by EVERY other node -- it must rank highest"  # => 'C' is linked to by EVERY other node -- it must rank highest
    print(f"MATCH: scores sum to {total:.6f} and the most-linked node 'C' ranks highest, as PageRank's own theory predicts")  # => shows MATCH: scores sum to
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

C: 0.5044
B: 0.2062
A: 0.1447
D: 0.1447
sum of scores: 1.000000
MATCH: scores sum to 1.000000 and the most-linked node 'C' ranks highest, as PageRank's own theory predicts

# pyright: strict
"""Example 78: PageRank + BM25 (co-36)."""
 
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 pagerank(links: dict[int, list[int]], damping: float = 0.85, iterations: int = 100) -> dict[int, float]:  # => defines pagerank
    nodes: list[int] = list(links)  # => nodes = list(links)
    n: int = len(nodes)  # => this fixture's own size
    scores: dict[int, float] = {node: 1.0 / n for node in nodes}  # => scores = {node: 1.0 / n for node in nodes}
    for _ in range(iterations):  # => iterates one item at a time
        new_scores: dict[int, float] = {node: (1 - damping) / n for node in nodes}  # => new scores = {node: (1 - damping) / n for node in nodes}
        for node, outlinks in links.items():  # => iterates one item at a time
            if not outlinks:  # => true when not outlinks
                continue  # => part of this step's computation, continued from the line above
            share: float = scores[node] / len(outlinks)  # => share = scores[node] / len(outlinks)
            for target in outlinks:  # => iterates one item at a time
                new_scores[target] += damping * share  # => part of this step's computation, continued from the line above
        scores = new_scores  # => scores = new_scores
    return scores  # => returns scores
 
 
def bm25_term_score(tf: int, term_df: int, n_docs: int, dl: float, avgdl: float, k1: float = 1.2, b: float = 0.75) -> float:  # => defines bm25 term score
    idf: float = math.log((n_docs - term_df + 0.5) / (term_df + 0.5))  # => idf = math.log((n_docs - term_df + 0.5) / (term_df + ...
    length_norm: float = (1 - b) + b * (dl / avgdl)  # => length norm = (1 - b) + b * (dl / avgdl)
    return idf * (tf * (k1 + 1)) / (tf + k1 * length_norm)  # => returns idf * (tf * (k1 + 1)) / (tf + k1 * length_norm)
 
 
def combined_rank(bm25_scores: dict[int, float], link_scores: dict[int, float], link_weight: float = 5.0) -> list[int]:  # => a simple linear combination: BM25 relevance PLUS a scaled link-authority score
    """A simple linear combination: BM25 relevance PLUS a scaled link-authority score."""
    combined: dict[int, float] = {  # => co-36: term relevance and link authority, added into ONE score
        doc_id: bm25_scores.get(doc_id, 0.0) + link_weight * link_scores.get(doc_id, 0.0) for doc_id in bm25_scores  # => part of this step's computation, continued from the line above
    }  # => opens/closes this multi-line literal
    return sorted(combined, key=lambda d: -combined[d])  # => returns sorted(combined, key=lambda d: -combined[d])
 
 
def main() -> None:  # => defines main
    # doc 0: a well-linked, MARGINALLY relevant page; doc 1: an orphan (unlinked), SLIGHTLY more relevant.
    bm25_scores: dict[int, float] = {0: 1.0, 1: 1.1}  # => doc 1 is the SLIGHTLY better term match on its own
    links: dict[int, list[int]] = {0: [], 1: [], 2: [0], 3: [0], 4: [0]}  # => co-36: three OTHER pages all link to doc 0
    link_scores: dict[int, float] = pagerank(links)  # => co-36: doc 0's link authority should dominate
 
    bm25_only_ranking: list[int] = sorted(bm25_scores, key=lambda d: -bm25_scores[d])  # => term relevance ALONE
    final_ranking: list[int] = combined_rank(bm25_scores, link_scores)  # => co-36: term relevance PLUS link authority
    print(f"link scores: {link_scores}")  # => shows link scores
    print(f"BM25-only ranking: {bm25_only_ranking}")  # => shows BM25-only ranking
    print(f"combined ranking:  {final_ranking}")  # => shows combined ranking
 
    assert bm25_only_ranking[0] == 1, "on TERM RELEVANCE alone, the marginally-better-matching orphan (doc 1) must rank first"  # => on TERM RELEVANCE alone, the marginally-better-matching orphan (doc 1) must rank first
    assert final_ranking[0] == 0, "once LINK AUTHORITY is folded in, the well-linked doc 0 must overtake the orphan"  # => once LINK AUTHORITY is folded in, the well-linked doc 0 must overtake the orphan
    print(f"MATCH: doc 1 wins on relevance alone, but the well-linked doc 0 wins once link authority is combined in")  # => shows MATCH: doc 1 wins on relevance alone, but the well-linked doc 0 wins once link authority is combined in
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end

link scores: {0: 0.10650000000000003, 1: 0.030000000000000006, 2: 0.030000000000000006, 3: 0.030000000000000006, 4: 0.030000000000000006}
BM25-only ranking: [1, 0]
combined ranking:  [0, 1]
MATCH: doc 1 wins on relevance alone, but the well-linked doc 0 wins once link authority is combined in

# pyright: strict
"""Example 79: Evaluate Hybrid (co-22, co-35)."""
 
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 cosine_similarity(a: list[float], b: list[float]) -> float:  # => defines cosine similarity
    dot: float = sum(x * y for x, y in zip(a, b))  # => dot = sum(x * y for x, y in zip(a, b))
    norm_a: float = math.sqrt(sum(x * x for x in a))  # => norm a = math.sqrt(sum(x * x for x in a))
    norm_b: float = math.sqrt(sum(y * y for y in b))  # => norm b = math.sqrt(sum(y * y for y in b))
    return dot / (norm_a * norm_b) if norm_a and norm_b else 0.0  # => returns dot / (norm_a * norm_b) if norm_a and norm_b else 0.0
 
 
def normalize(scores: dict[int, float]) -> dict[int, float]:  # => defines normalize
    if not scores:  # => true when not scores
        return {}  # => returns {}
    lo, hi = min(scores.values()), max(scores.values())  # => part of this step's computation, continued from the line above
    if hi == lo:  # => true when hi == lo
        return {doc_id: 1.0 for doc_id in scores}  # => returns {doc_id: 1.0 for doc_id in scores}
    return {doc_id: (s - lo) / (hi - lo) for doc_id, s in scores.items()}  # => returns {doc_id: (s - lo) / (hi - lo) for doc_id, s in scores.ite...
 
 
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
    relevant: set[int] = {2}  # => co-23: doc 2 (the "automobile" doc) is the ONE truly relevant result for this query
 
    lexical_scores: dict[int, float] = {0: 3.0, 1: 2.0, 2: 0.0}  # => co-16: doc 2 scores ZERO -- no literal query-term overlap
    vector_scores: dict[int, float] = {0: 0.1, 1: 0.2, 2: 0.95}  # => co-35: doc 2's embedding is by far the closest
 
    lexical_ranking: list[int] = sorted(lexical_scores, key=lambda d: -lexical_scores[d])  # => co-22: pure lexical order
    alpha: float = 0.3  # => co-35: 30% lexical, 70% vector -- enough weight to let the semantic signal win here
    combined: dict[int, float] = {  # => combined = {
        doc_id: alpha * normalize(lexical_scores).get(doc_id, 0.0) + (1 - alpha) * normalize(vector_scores).get(doc_id, 0.0)  # => part of this step's computation, continued from the line above
        for doc_id in lexical_scores  # => part of this step's computation, continued from the line above
    }  # => opens/closes this multi-line literal
    hybrid_ranking: list[int] = sorted(combined, key=lambda d: -combined[d])  # => co-35: the alpha-weighted hybrid blend
 
    p1_lexical: float = precision_at_k(lexical_ranking, relevant, k=1)  # => co-22: precision@1, lexical only
    p1_hybrid: float = precision_at_k(hybrid_ranking, relevant, k=1)  # => co-22: precision@1, hybrid
    print(f"lexical ranking: {lexical_ranking}  precision@1={p1_lexical:.4f}")  # => shows lexical ranking
    print(f"hybrid ranking:  {hybrid_ranking}  precision@1={p1_hybrid:.4f}")  # => shows hybrid ranking
 
    assert p1_lexical == 0.0, "lexical-only precision@1 must be 0 -- the relevant doc scored zero lexical relevance"  # => lexical-only precision@1 must be 0 -- the relevant doc scored zero lexical relevance
    assert p1_hybrid == 1.0, "hybrid precision@1 must be 1.0 -- the vector signal surfaces the truly relevant doc"  # => hybrid precision@1 must be 1.0 -- the vector signal surfaces the truly relevant doc
    print(f"MATCH: hybrid precision@1 ({p1_hybrid}) beats lexical-only ({p1_lexical}) on this synonym-dependent query")  # => shows MATCH: hybrid precision@1 (
 
 
if __name__ == "__main__":  # => entry point -- runs only when this file executes directly, not on import
    main()  # => runs the example end to end