হাইব্রিড রিট্রিভাল ও এরর বিশ্লেষণ

from evidencedesk.retrieval.catalog import Scope, load_catalog
from evidencedesk.retrieval.labels import load_labels
from evidencedesk.retrieval.report import evaluate
from evidencedesk.retrieval.searchers import DenseSearcher

labels = load_labels()
cat = load_catalog("paragraph")  # the Beginner chunks, plus three new documents
dense = DenseSearcher(cat.chunks)
rows = evaluate(dense, cat, labels, Scope(), k=3, floor=0.3)  # the Beginner baseline: current only, cosine >= 0.3
print(f"{len(rows)} questions, {sum(r.failed for r in rows)} fail")
for r in rows:
    if r.failed:
        f = r.findings[0]
        print(f"{r.id:<5}{f.stage:<10}{f.label:<17}{f.detail}")

24 questions, 5 fail
rc03 context   missed_evidence  retrieved at rank 1 but dropped before the prompt
rc06 retrieval missed_evidence  answer not in the top 3; rank 1 is kc-privacy-v1:c0
rc08 retrieval missed_evidence  answer not in the top 3; rank 1 is kc-sizing-v1:c1
rc15 context   irrelevant_match 3 chunk(s) in the prompt; the corpus has no answer
h01  retrieval missed_evidence  answer not in the top 3; rank 1 is kc-torque-v1:c0

from evidencedesk.retrieval.candidates import ranked
from evidencedesk.retrieval.report import run_trace
from evidencedesk.retrieval.taxonomy import Trace, diagnose

rc01 = next(x for x in labels if x.id == "rc01")
cut = load_catalog("fixed", size=60, overlap=0)  # windows far too small: the evidence is cut in two
hit = ranked([(cat.by_id["kc-returns-v3:c0"], 0.55)], "dense")
cases = {
    "ingestion": run_trace(rc01, DenseSearcher(cut.chunks), cut, Scope(), 3),
    "generation": Trace(tuple(cat.chunks), hit, hit, "Return within 60 days.", ("kc-returns-v3:c0",)),
    "citation": Trace(tuple(cat.chunks), hit, hit, "Return within 30 days of delivery.", ("kc-returns-v3:c1",)),
}
for name, trace in cases.items():
    print(f"{name:<11}", [(f.stage, f.label) for f in diagnose(rc01, trace)])
lost = sum(not any(x.grade_of(c) == 2 for c in cut.chunks) for x in labels if x.answerable)
print(f"with windows of 60 characters, {lost} of 21 answers are whole in no chunk")

ingestion   [('ingestion', 'missed_evidence')]
generation  [('generation', 'unsupported_answer')]
citation    [('citation', 'unsupported_answer')]
with windows of 60 characters, 16 of 21 answers are whole in no chunk

# file: evidencedesk/retrieval/taxonomy.py (excerpt)
STAGES = ("ingestion", "retrieval", "context", "generation", "citation")


LABELS = ("missed_evidence", "irrelevant_match", "stale_document", "duplicate_chunks", "unsupported_answer")


@dataclass(frozen=True)
class Finding:
    stage: str  # one of STAGES: where the fix belongs
    label: str  # one of LABELS: what the symptom is
    detail: str


@dataclass(frozen=True)
class Trace:
    """What one run did, step by step. Built by the pipeline, read by diagnose()."""

    indexed: tuple[Chunk, ...]  # every chunk the index holds
    retrieved: RankedList  # what search returned
    context: RankedList  # what survived into the prompt (score floor, de-duplication, token budget ...)
    answer: str | None = None  # the model's claim, if a model was called
    cited: tuple[str, ...] = ()  # chunk ids the model cited

{"id": "rc09", "question": "Are refunds paid as store credit?", "tags": ["stale"], "answerable": true,
 "gold": [{"section": "kc-returns-v3#How refunds are paid", "evidence": "Refunds go back to the original payment method within 5 business days", "grade": 2}],
 "reason": "The current policy says original payment method. The superseded v2 says store credit: retrieving it gives a confident wrong yes."}

from dataclasses import replace
from evidencedesk.retrieval.catalog import Catalog
from evidencedesk.retrieval.report import label_counts

open_ = label_counts(evaluate(dense, cat, labels, None, k=3, floor=0.3))  # no filter at all
shut = label_counts(evaluate(dense, cat, labels, Scope(), k=3, floor=0.3))
print(f"{'stage':<11}{'label':<17}{'no filter':>10}{'filtered':>10}")
for key in sorted(open_.keys() | shut.keys()):
    print(f"{key[0]:<11}{key[1]:<17}{open_[key]:>10}{shut[key]:>10}")
twin = replace(cat.by_id["kc-returns-v3:c0"], id="kc-returns-v3:c0-copy")  # a re-import that kept both copies
dup = Catalog(cat.chunks + [twin], cat.audience)
t = run_trace(rc01, DenseSearcher(dup.chunks), dup, Scope(), 3)
print([c.chunk.id for c in t.retrieved])
print([(f.stage, f.label) for f in diagnose(rc01, t)])

stage      label             no filter  filtered
context    irrelevant_match          1         1
context    missed_evidence           1         1
retrieval  irrelevant_match          6         4
retrieval  missed_evidence           3         3
retrieval  stale_document            7         0
['kc-returns-v3:c0', 'kc-returns-v3:c0-copy', 'kc-warranty-v2:c0']
[('context', 'duplicate_chunks')]

from evidencedesk.retrieval.searchers import SparseSearcher

sparse = SparseSearcher(cat.chunks)
bm = sparse.bm25
print(f"{bm.n} chunks, average length {bm.avg_len:.1f} words")
for term in ("rd2-24", "return", "bike"):
    print(f"{term:<8}in {bm.df[term]:>2} chunks  idf {bm.idf(term):.2f}")
for q in ("RD2-24", "How do I get my money back?"):
    print(q)
    for s in (sparse, dense):
        print(f"  {s.name:<7}", [(c.chunk.id.removeprefix("kc-"), round(c.score, 2)) for c in s.search(q, 3, cat.allowed(Scope()))])
n, hand = bm.length["kc-recalls-v1:c0"], 3.56 * 1 / (1 + 1.2 * (1 - 0.75 + 0.75 * bm.length["kc-recalls-v1:c0"] / bm.avg_len))
print(f"by hand for RD2-24: chunk of {n} words, 3.56 * 1 / (1 + 1.2 * (0.25 + 0.75 * {n} / {bm.avg_len:.1f})) = {hand:.2f}")

52 chunks, average length 21.0 words
rd2-24  in  1 chunks  idf 3.56
return  in  2 chunks  idf 3.05
bike    in 12 chunks  idf 1.44
RD2-24
  sparse  [('recalls-v1:c0', 1.5)]
  dense   [('recalls-v1:c0', 0.29), ('troubleshoot-v1:c1', 0.22), ('troubleshoot-v1:c0', 0.19)]
How do I get my money back?
  sparse  [('returns-v3:c2', 1.72)]
  dense   [('privacy-v1:c0', 0.22), ('shipping-v1:c1', 0.22), ('returns-v3-bn:c1', 0.18)]
by hand for RD2-24: chunk of 25 words, 3.56 * 1 / (1 + 1.2 * (0.25 + 0.75 * 25 / 21.0)) = 1.50

from evidencedesk.retrieval.report import hit_at

TAGS = ["exact_term", "paraphrase", "bangla", "structured"]
by = {s.name: evaluate(s, cat, labels, Scope(), k=5) for s in (sparse, dense)}
print(f"{'tag':<12}{'n':>3}{'sparse@1':>10}{'sparse@3':>10}{'dense@1':>9}{'dense@3':>9}")
for tag in TAGS + [None]:
    cells = [hit_at(by[s], k, tag)[0] for s in ("sparse", "dense") for k in (1, 3)]
    print(f"{tag or 'all':<12}{hit_at(by['sparse'], 1, tag)[1]:>3}{cells[0]:>10}{cells[1]:>10}{cells[2]:>9}{cells[3]:>9}")
rank = {s: {r.id: r.rank for r in rows} for s, rows in by.items()}
for a, b in (("sparse", "dense"), ("dense", "sparse")):
    only = [q for q in rank[a] if rank[a][q] and rank[a][q] <= 3 and not (rank[b][q] and rank[b][q] <= 3)]
    print(f"top 3 has the answer for {a} but not {b}:", only)

tag           n  sparse@1  sparse@3  dense@1  dense@3
exact_term    4         3         4        2        4
paraphrase    6         3         4        4        4
bangla        3         2         3        3        3
structured    6         4         6        3        5
all          21        15        19       14       18
top 3 has the answer for sparse but not dense: ['rc06', 'h01']
top 3 has the answer for dense but not sparse: ['rc05']

# file: evidencedesk/retrieval/fusion.py (excerpt)
K = 60  # the paper's constant: fixed in a pilot, "near-optimal, but the choice was not critical"


def rrf(lists: Sequence[RankedList], k: int = K, top: int | None = None, source: str = "hybrid") -> RankedList:
    """score(d) = sum over the lists that contain d of 1 / (k + rank of d in that list). Scores are ignored.

    Exact fractions make ties real ties; ties go to the better best-rank, then to the chunk id, so output is stable.
    """
    if k < 0:
        raise ValueError("k must not be negative")
    total: dict[str, Fraction] = {}
    first: dict[str, Candidate] = {}
    parts: dict[str, list[Contribution]] = {}
    for lst in lists:
        for rank, cand in enumerate(lst, 1):  # the position in the list, so a hand-built list works too
            cid = cand.chunk.id
            total[cid] = total.get(cid, Fraction(0)) + Fraction(1, k + rank)
            first.setdefault(cid, cand)
            parts.setdefault(cid, []).append(Contribution(cand.source, rank, cand.score))
    best = {cid: min(p.rank for p in ps) for cid, ps in parts.items()}
    order = sorted(total, key=lambda cid: (-total[cid], best[cid], cid))
    fused = [Candidate(first[cid].chunk, float(total[cid]), n, source, tuple(parts[cid])) for n, cid in enumerate(order, 1)]
    return fused[:top]

import json
from fractions import Fraction
from evidencedesk.retrieval.candidates import ids, ranked
from evidencedesk.retrieval.fusion import rrf
from evidencedesk.retrieval.labels import DATA

class Doc:  # the smallest stand-in for a chunk: fusion only reads .id
    def __init__(self, id): self.id = id

for ex in json.loads((DATA / "rrf_examples.json").read_text(encoding="utf-8"))[:2]:
    lists = [ranked([(Doc(i), 0.0) for i in names], "x") for names in ex["lists"]]
    got = [(c.chunk.id, c.score) for c in rrf(lists, ex["k"])]
    hand = [(i, float(sum(Fraction(p.strip()) for p in expr.split("+")))) for i, expr in ex["expected"]]
    print(ex["id"], "k =", ex["k"], "same as by hand:", all(a[0] == b[0] and abs(a[1] - b[1]) < 1e-12 for a, b in zip(got, hand)))
    print("   ", [(i, round(s, 5)) for i, s in got])

two-lists k = 60 same as by hand: True
    [('d1', 0.03252), ('d3', 0.03227), ('d2', 0.01613), ('d4', 0.01587)]
agreement-beats-one-first-place k = 60 same as by hand: True
    [('q', 0.03226), ('p', 0.01639), ('r', 0.01639)]

from evidencedesk.retrieval.searchers import HybridSearcher

hybrid = HybridSearcher(sparse, dense)
allowed = cat.allowed(Scope())
for q in ("RD2-24", "Is a bike returnable after a month?", "What torque should the stem bolts have on a size M?"):
    lab = next(x for x in labels if x.question == q)
    fused = hybrid.search(q, 20, allowed)
    top = next(c for c in fused if lab.grade_of(c.chunk) == 2)
    where = {p.source: p.rank for p in top.parts}
    print(f"{q[:34]:<36}answer: sparse rank {where.get('sparse', '-')}, dense rank {where.get('dense', '-')}, hybrid rank {top.rank}")
top = lambda q: round(dense.search(q, 1, allowed)[0].score, 3)
print("best dense score: unanswerable 'electric scooters'", top("Do you sell electric scooters?"),
      "| answerable 'RD2-24'", top("RD2-24"), "| 'money back'", top("How do I get my money back?"))

RD2-24                              answer: sparse rank 1, dense rank 1, hybrid rank 1
Is a bike returnable after a month  answer: sparse rank 4, dense rank 1, hybrid rank 2
What torque should the stem bolts   answer: sparse rank 3, dense rank 12, hybrid rank 5
best dense score: unanswerable 'electric scooters' 0.456 | answerable 'RD2-24' 0.295 | 'money back' 0.224

# file: evidencedesk/retrieval/catalog.py (excerpt)
@dataclass(frozen=True)
class Scope:
    """What a request is allowed to see. The default is a public customer asking about current policy."""

    statuses: frozenset[str] = frozenset({"current"})
    audiences: frozenset[str] = frozenset({"public"})
    langs: frozenset[str] | None = None  # None = any language

# file: evidencedesk/retrieval/searchers.py (excerpt)
class DenseSearcher:
    name = "dense"

    def __init__(self, chunks: list[Chunk], model: Model | None = None) -> None:
        self.chunks, self.model = {c.id: c for c in chunks}, model or Model.toy()
        self.index, _ = VectorIndex.build([c.to_passage() for c in chunks], self.model, lambda: "fixed-clock")

    def search(self, query: str, k: int = 5, allowed: set[str] | None = None) -> RankedList:
        """Filter first, then rank only the allowed vectors. Zero or negative cosine is not a match."""
        self.index.meta.check(self.model)
        rows = [n for n, cid in enumerate(self.index.ids) if allowed is None or cid in allowed]
        if unusable_reason(query) or not rows:
            return []
        hits = top_k(self.model.embed_batch([query])[0], self.index.matrix[rows], [self.index.ids[n] for n in rows], k)
        return ranked([(self.chunks[h.passage_id], h.score) for h in hits if h.score > 0], self.name)

from evidencedesk.retrieval.searchers import post_filter

q = "Can I return a bike after 40 days?"
everyone = cat.allowed(Scope(audiences=frozenset({"public", "staff"}), statuses=frozenset({"current", "superseded"})))
short = lambda found: [c.chunk.id.removeprefix("kc-") for c in found]
print("no filter  ", short(hybrid.search(q, 3, everyone)))
print("post-filter", short(post_filter(hybrid, q, 3, allowed)))
print("pre-filter ", short(hybrid.search(q, 3, allowed)))
leak = "goodwill refund store manager"
print(leak, "->", short(hybrid.search(leak, 3, everyone))[:2], "| post-filter:", short(post_filter(hybrid, leak, 3, allowed)))
print(len(everyone), "chunks in the index,", len(allowed), "visible to a public customer asking for current policy")

no filter   ['returns-v2:c0', 'returns-v3:c0', 'staff-goodwill-v1:c0']
post-filter ['returns-v3:c0']
pre-filter  ['returns-v3:c0', 'returns-v3:c2', 'battery-v1:c1']
goodwill refund store manager -> ['staff-goodwill-v1:c0', 'staff-goodwill-v1:c1'] | post-filter: []
52 chunks in the index, 48 visible to a public customer asking for current policy

def context_chars(searcher, c):  # characters the model would read: the top 3 chunks
    sizes = [sum(len(x.chunk.text) for x in searcher.search(l.question, 3, c.allowed(Scope()))) for l in labels]
    return sum(sizes) / len(sizes)

print(f"{'strategy':<10}{'chunks':>7}{'whole':>7}{'hit@1':>7}{'hit@3':>7}{'hit@3 tables/code/long':>24}{'context chars':>15}")
for name in ("fixed", "section", "paragraph", "structure"):
    c = load_catalog(name)
    h = HybridSearcher(SparseSearcher(c.chunks), DenseSearcher(c.chunks))
    whole = sum(any(x.grade_of(ch) == 2 for ch in c.chunks) for x in labels if x.answerable)
    rows = evaluate(h, c, labels, Scope(), k=5)
    print(f"{name:<10}{len(c.chunks):>7}{whole:>5}/21{hit_at(rows, 1)[0]:>7}{hit_at(rows, 3)[0]:>7}"
          f"{hit_at(rows, 3, 'structured')[0]:>16}/{hit_at(rows, 3, 'structured')[1]}{context_chars(h, c):>15.0f}")

strategy   chunks  whole  hit@1  hit@3  hit@3 tables/code/long  context chars
fixed          46   20/21     16     19               5/6            562
section        44   21/21     18     20               6/6            517
paragraph      52   21/21     16     19               5/6            382
structure      54   21/21     18     19               5/6            386

s = load_catalog("structure")
for x in s.chunks:
    if x.id in ("kc-torque-v1:c1", "kc-torque-v1:c8"):
        print(f"{x.id.removeprefix('kc-torque-v1:'):<3}{x.kind:<6}{x.location.label:<24}", repr(x.text[:58]))
flat = next(x for x in cat.chunks if x.doc_id == "kc-torque-v1" and "Stem bolts" in x.text)
print("paragraph chunker:", repr(flat.text[:70]))

c1 table Bolt torque, lines 16-18 '| Part | Size M | Size L | Note |\n|---|---|---|---|\n| Stem'
c8 code  Check a serial number, lines 37-40 '```bash\nkestrel-recall check --serial RD2-24-00417\nkestrel'
paragraph chunker: '| Part | Size M | Size L | Note | |---|---|---|---| | Stem bolts | 5 |'

# file: evidencedesk/retrieval/pipeline.py (excerpt)
class Reranker(Protocol):
    def rerank(self, query: str, found: RankedList) -> RankedList: ...  # same candidates, new scores and order


@dataclass
class TwoStage:
    first: Searcher
    reranker: Reranker
    candidates: int = 20  # how many the first stage fetches
    final: int = 3  # how many reach the context
    name: str = "two_stage"
    scored: int = field(default=0, init=False)  # candidates the second stage had to score: its cost, counted not timed

    def search(self, query: str, k: int | None = None, allowed: set[str] | None = None) -> RankedList:
        broad = self.first.search(query, self.candidates, allowed)
        self.scored += len(broad)
        return self.reranker.rerank(query, broad)[: k or self.final]

from evidencedesk.retrieval.pipeline import OverlapReranker, TwoStage

base = evaluate(hybrid, cat, labels, Scope(), k=3)
print(f"first stage alone: hit@1 {hit_at(base, 1)[0]}, hit@3 {hit_at(base, 3)[0]} of 21")
print("candidates  answer among candidates  after rerank hit@1  hit@3  chunks scored per question")
for n in (3, 10, 30):
    two = TwoStage(hybrid, OverlapReranker(), candidates=n, final=3)
    rows = evaluate(two, cat, labels, Scope(), k=3)
    pool = hit_at(evaluate(hybrid, cat, labels, Scope(), k=n), n)[0]
    print(f"{n:>10}{pool:>20}/21{hit_at(rows, 1)[0]:>19}{hit_at(rows, 3)[0]:>7}{two.scored / len(labels):>16.1f}")

first stage alone: hit@1 16, hit@3 19 of 21
candidates  answer among candidates  after rerank hit@1  hit@3  chunks scored per question
         3                  19/21                 17     19             3.0
        10                  20/21                 18     20            10.0
        30                  20/21                 18     20            21.7

from evidencedesk.retrieval.report import label_counts

print(f"{'strategy':<9}{'hit@1':>6}{'hit@3':>6}{'hit@5':>6}{'answer outranked':>18}{'no answer in 5':>16}{'junk for unanswerable':>23}")
for s in (sparse, dense, hybrid):
    rows = evaluate(s, cat, labels, Scope(), k=5)
    c = label_counts(rows)
    junk = sum(r.returned > 0 for r in rows if not r.answerable)
    print(f"{s.name:<9}{hit_at(rows, 1)[0]:>6}{hit_at(rows, 3)[0]:>6}{hit_at(rows, 5)[0]:>6}"
          f"{c[('retrieval', 'irrelevant_match')]:>18}{c[('retrieval', 'missed_evidence')]:>16}{junk:>20}/3")
missed = {s.name: [r.id for r in evaluate(s, cat, labels, Scope(), k=5) if r.answerable and r.rank is None] for s in (sparse, dense, hybrid)}
print("no answer in the top 5:", missed)

strategy  hit@1 hit@3 hit@5  answer outranked  no answer in 5  junk for unanswerable
sparse       15    19    20                 5               1                   1/3
dense        14    18    19                 5               2                   3/3
hybrid       16    19    20                 4               1                   3/3
no answer in the top 5: {'sparse': ['rc08'], 'dense': ['rc08', 'h01'], 'hybrid': ['rc08']}

python -m pytest tests/test_retrieval_fusion.py tests/test_retrieval_labels.py tests/test_retrieval_taxonomy.py \
  tests/test_retrieval_filters.py tests/test_retrieval_structure.py tests/test_retrieval_pipeline.py 2>&1 | tail -1 | sed 's/ in [0-9.]*s//'

40 passed

top = {x.id: (dense.search(x.question, 1, allowed) or [None])[0] for x in labels}
best = min(((sum((c is not None and c.score >= t) != x.answerable for x in labels for c in [top[x.id]]), t)
            for t in [n / 100 for n in range(20, 80)]))
print(f"best single dense threshold: {best[1]:.2f} -> {best[0]} of {len(labels)} wrong answer/abstain decisions")

best single dense threshold: 0.28 -> 2 of 24 wrong answer/abstain decisions