oskar/homelab-codex-ws
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

feat(eval): extend retrieval_eval.py + queries.yaml for hybrid mode (faza mailowa Krok 5, partial)

Browse source 
retrieval_eval.py now runs every existing query through hybrid_query
alongside flat/cascade, and gate criteria 1/3 (no degradation, negative
controls) check hybrid too -- this is the phase's key regression test: does
adding mail chunks to the HNSW index degrade the 7 existing paperless
queries? Criterion 4 (mail hit@3 in hybrid) is wired but skipped with an
explicit report note when queries.yaml's new `mail_queries` placeholder is
empty, rather than silently PASSing/FAILing on absent data.

queries.yaml: added an empty `mail_queries: []` placeholder + inline format
docs. The existing 7-query `queries:` list is untouched.

REMINDER for the operator (per plan §8, Krok 5): 3-5 mail queries are still
needed -- things you know are in last year's mail, plus each one's expected
gmail Message-ID -- before criterion 4 and the full plan §8 gate can PASS.
Until then this only re-verifies the paperless regression check.

docs/kb/modules/05-faza-mailowa-plan.md, §8.
This commit is contained in:
oskar 2026-07-22 19:04:02 +02:00
parent a0c9dbfa52
commit f0fcdfa835
2 changed files with 134 additions and 27 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
jobs/documents-ingest/eval/queries.yaml
View file

@ -71,3 +71,19 @@ queries:
note: >
Poprawnie na granicy "brak" w pilocie -- semantycznie sąsiednie dokumenty wspólnoty
mieszkaniowej, nie odpowiedź na zapytanie.
# Faza mailowa (docs/kb/modules/05-faza-mailowa-plan.md, §8, Krok 5) -- bramka jakościowa dla
# treści mailowej wprowadzonej w Etapie A (ostatnie 12 miesięcy, plan §7 Krok 4). PLACEHOLDER:
# operator ma dostarczyć 3-5 zapytań "wiem że to mam w mailach z ostatniego roku" +
# oczekiwany Message-ID (surowy, bez prefiksu "gmail:" -- envelope.id dla źródła gmail to
# bare Message-ID, np. "abc123@mail.gmail.com", inaczej niż "paperless:N" powyżej).
# Do czasu uzupełnienia ta lista jest pusta i retrieval_eval.py pomija kryterium hit@3 mailowe
# z jawną notatką w raporcie, zamiast fałszywie PASS/FAIL na braku danych.
#
# Format wpisu (identyczny co do pól z `queries:` powyżej):
# - id: "M1"
# text: "..."
# kind: hit
# expected_envelope: "<raw gmail Message-ID>"
# note: "..."
mail_queries: []

145
jobs/documents-ingest/eval/retrieval_eval.py
View file

@ -1,16 +1,23 @@
"""Retrieval quality gate -- module 5, phase 3, plan §6.2 (docs/kb/modules/05-faza3-plan.md).
"""Retrieval quality gate -- module 5, phase 3, plan §6.2 (docs/kb/modules/05-faza3-plan.md),
extended in faza mailowa Krok 5 (docs/kb/modules/05-faza-mailowa-plan.md, §8) to add the
`hybrid` track once mail content exists in `document_chunk` (faza mailowa Krok 2/4).
Read-only integration script (NOT collected by pytest -- it hits the live kb-postgres DB and
the live Ollama instance, exactly like the plan asked for a separate eval script rather than a
mocked test). Runs every query in `queries.yaml` through both the flat baseline
(`kb_retrieval.retrieval.flat_query`) and the cascade (`cascade_query`), for a sweep of N
values, and checks the plan's three gate criteria:
mocked test). Runs every query in `queries.yaml`'s `queries:` list through flat
(`kb_retrieval.retrieval.flat_query`), cascade (`cascade_query`, swept over N), and hybrid
(`hybrid_query`), and checks the gate criteria:
1. every query the flat path hits (top-1 dist < 0.45) must still be a hit in the cascade
(no degradation into the grey zone or a miss);
1. every query the flat path hits (top-1 dist < 0.45) must still be a hit in cascade AND in
hybrid (no degradation into the grey zone or a miss) -- faza mailowa's key risk: does
adding mail chunks to the HNSW index degrade existing paperless retrieval?
2. hit@3 (expected envelope among the top-3 *distinct* envelopes by best distance) for the
cascade must be >= hit@3 for the flat baseline, across the whole set;
3. both negative controls must stay above 0.55 in both tracks.
cascade must be >= hit@3 for the flat baseline, across the whole `queries:` set;
3. both negative controls must stay above 0.55 in flat, cascade, AND hybrid;
4. (faza mailowa, only when `mail_queries` in queries.yaml is non-empty) hit@3 in hybrid for
the operator-supplied mail queries must be >= 4/5 (or a proportional threshold for fewer
queries). An empty `mail_queries` list skips this criterion with an explicit note in the
report, rather than silently passing or failing on absent data.
`envelope`, `document_chunk`, and `document_summary` are only ever `SELECT`ed -- this script
writes nothing. Query embeddings go through Ollama on localhost/SOLARIA (bge-m3), same as
@ -25,6 +32,7 @@ from __future__ import annotations
import argparse
import asyncio
import json
import math
import os
import sys
from pathlib import Path
@ -45,10 +53,12 @@ from kb_retrieval.retrieval import ( # noqa: E402
DEFAULT_SUMMARY_MODEL,
cascade_query,
flat_query,
hybrid_query,
)
HIT_THRESHOLD = 0.45
NO_ANSWER_THRESHOLD = 0.55
MAIL_HIT3_MIN_FRACTION = 0.8 # plan §8: ">= 4/5 (lub 3/3-4/4 przy mniejszej liczbie)"
DEFAULT_QUERIES_PATH = Path(__file__).resolve().parent / "queries.yaml"
@ -58,6 +68,13 @@ def load_queries(path: Path) -> list[dict]:
return data["queries"]
def load_mail_queries(path: Path) -> list[dict]:
"""Placeholder section (plan §8, Krok 5) -- empty until the operator supplies 3-5 queries."""
with open(path, encoding="utf-8") as f:
data = yaml.safe_load(f)
return data.get("mail_queries") or []
def top1_dist(chunks: list[dict]) -> Optional[float]:
return min((c["dist"] for c in chunks), default=None)
@ -77,7 +94,7 @@ def hit_at_3(chunks: list[dict], expected_envelope: Optional[str]) -> Optional[b
return expected_envelope in {eid for eid, _ in top3}
async def run_query_both_tracks(
async def run_query_all_tracks(
conn: asyncpg.Connection,
session: aiohttp.ClientSession,
ollama_url: str,
@ -94,13 +111,18 @@ async def run_query_both_tracks(
conn, session, ollama_url, query["text"],
summary_model=summary_model, embed_model=embed_model, n=n, k=k,
)
return {"query": query, "flat": flat, "cascades": cascades}
hybrid = await hybrid_query(
conn, session, ollama_url, query["text"],
summary_model=summary_model, embed_model=embed_model, k=k,
)
return {"query": query, "flat": flat, "cascades": cascades, "hybrid": hybrid}
def summarize_query_result(result: dict) -> dict:
query = result["query"]
expected = query["expected_envelope"]
flat_chunks = result["flat"]["chunks"]
hybrid_chunks = result["hybrid"]["chunks"]
row = {
"id": query["id"],
@ -109,6 +131,8 @@ def summarize_query_result(result: dict) -> dict:
"expected_envelope": expected,
"flat_top1_dist": top1_dist(flat_chunks),
"flat_hit3": hit_at_3(flat_chunks, expected),
"hybrid_top1_dist": top1_dist(hybrid_chunks),
"hybrid_hit3": hit_at_3(hybrid_chunks, expected),
"cascade": {},
}
for n, cascade_result in result["cascades"].items():
@ -122,22 +146,28 @@ def summarize_query_result(result: dict) -> dict:
return row
def evaluate_gate(rows: list[dict], gate_n: int) -> dict:
"""Plan §6.2's three criteria, evaluated at one chosen N (the sweep is diagnostic, the
gate verdict is always for one specific configuration)."""
def evaluate_gate(rows: list[dict], gate_n: int, mail_rows: Optional[list[dict]] = None) -> dict:
"""Plan §6.2/§8's criteria, evaluated at one chosen N (the sweep is diagnostic, the gate
verdict is always for one specific configuration). `mail_rows` is None or empty when the
operator hasn't supplied `mail_queries` yet (plan §8 Krok 5 placeholder) -- criterion 4 is
then skipped with an explicit note, never silently PASSed or FAILed on absent data."""
reasons = []
# 1. no flat hit (dist < 0.45) may degrade under the cascade.
# 1. no flat hit (dist < 0.45) may degrade under cascade OR hybrid (faza mailowa's core
# regression risk: does the mail chunk mass added to HNSW push existing hits out?).
degraded = []
for row in rows:
flat_dist = row["flat_top1_dist"]
if flat_dist is not None and flat_dist < HIT_THRESHOLD:
cascade_dist = row["cascade"][gate_n]["top1_dist"]
if cascade_dist is None or cascade_dist >= HIT_THRESHOLD:
degraded.append(row["id"])
degraded.append(f"{row['id']} (cascade)")
hybrid_dist = row["hybrid_top1_dist"]
if hybrid_dist is None or hybrid_dist >= HIT_THRESHOLD:
degraded.append(f"{row['id']} (hybrid)")
criterion_1 = not degraded
if not criterion_1:
reasons.append(f"criterion 1 FAILED: flat hits degraded under cascade N={gate_n}: {degraded}")
reasons.append(f"criterion 1 FAILED: flat hits degraded: {degraded}")
# 2. hit@3 (cascade) >= hit@3 (flat), over queries that have an expected envelope.
scored = [row for row in rows if row["expected_envelope"] is not None]
@ -150,7 +180,7 @@ def evaluate_gate(rows: list[dict], gate_n: int) -> dict:
f"< flat={flat_hit3_count}/{len(scored)} (N={gate_n})"
)
# 3. negative controls stay above 0.55 in both tracks.
# 3. negative controls stay above 0.55 in flat, cascade, AND hybrid.
control_failures = []
for row in rows:
if row["kind"] not in ("negative_control", "negative_control_borderline"):
@ -160,28 +190,60 @@ def evaluate_gate(rows: list[dict], gate_n: int) -> dict:
cascade_dist = row["cascade"][gate_n]["top1_dist"]
if cascade_dist is not None and cascade_dist <= NO_ANSWER_THRESHOLD:
control_failures.append(f"{row['id']} cascade(N={gate_n})={cascade_dist:.4f}")
hybrid_dist = row["hybrid_top1_dist"]
if hybrid_dist is not None and hybrid_dist <= NO_ANSWER_THRESHOLD:
control_failures.append(f"{row['id']} hybrid={hybrid_dist:.4f}")
criterion_3 = not control_failures
if not criterion_3:
reasons.append(f"criterion 3 FAILED: negative control(s) crossed {NO_ANSWER_THRESHOLD}: {control_failures}")
passed = criterion_1 and criterion_2 and criterion_3
# 4. (faza mailowa, plan §8) mail queries' hit@3 in hybrid -- skipped entirely when the
# operator hasn't supplied queries yet (mail_rows empty), never PASSed/FAILed on no data.
mail_rows = mail_rows or []
if mail_rows:
mail_hit3_count = sum(1 for row in mail_rows if row["hybrid_hit3"])
required = math.ceil(len(mail_rows) * MAIL_HIT3_MIN_FRACTION)
criterion_4 = mail_hit3_count >= required
criterion_4_detail = {
"hit3": mail_hit3_count, "total": len(mail_rows), "required": required, "skipped": False,
}
if not criterion_4:
reasons.append(
f"criterion 4 FAILED: mail hit@3 (hybrid) = {mail_hit3_count}/{len(mail_rows)}, "
f"required >= {required}"
)
else:
criterion_4 = True # does not block PASS -- see note below
criterion_4_detail = {"hit3": 0, "total": 0, "required": 0, "skipped": True}
reasons.append(
"criterion 4 SKIPPED: mail_queries is empty in queries.yaml -- operator hasn't "
"supplied the 3-5 mail queries yet (plan §8, Krok 5). Gate PASS below reflects "
"only the paperless regression check, not full plan §8 completion."
)
passed = criterion_1 and criterion_2 and criterion_3 and criterion_4
return {
"gate_n": gate_n,
"passed": passed,
"criterion_1_no_degradation": criterion_1,
"criterion_2_hit3": {"cascade": cascade_hit3_count, "flat": flat_hit3_count, "total": len(scored)},
"criterion_3_negative_controls": criterion_3,
"criterion_4_mail_hit3": criterion_4_detail,
"reasons": reasons,
}
def print_report(rows: list[dict], n_values: list[int], gate_result: dict) -> None:
def print_report(
rows: list[dict], n_values: list[int], gate_result: dict, mail_rows: Optional[list[dict]] = None
) -> None:
print("=" * 100)
print("RETRIEVAL QUALITY GATE -- plan §6.2 (docs/kb/modules/05-faza3-plan.md)")
print("RETRIEVAL QUALITY GATE -- plan §6.2 (docs/kb/modules/05-faza3-plan.md), "
"+ hybrid/mail extension (05-faza-mailowa-plan.md §8)")
print("=" * 100)
header = f"{'id':<3} {'kind':<28} {'expected':<16} {'flat d1':>8} {'flat@3':>7}"
for n in n_values:
header += f" N={n:<3} d1 hit3"
header += " hybrid d1 hyb@3"
print(header)
for row in rows:
flat_d1 = f"{row['flat_top1_dist']:.4f}" if row["flat_top1_dist"] is not None else "--"
@ -192,26 +254,43 @@ def print_report(rows: list[dict], n_values: list[int], gate_result: dict) -> No
d1 = f"{c['top1_dist']:.4f}" if c["top1_dist"] is not None else "--"
h3 = "y" if c["hit3"] else ("--" if c["hit3"] is None else "n")
line += f" {d1:>8} {h3:>4}"
hyb_d1 = f"{row['hybrid_top1_dist']:.4f}" if row["hybrid_top1_dist"] is not None else "--"
hyb_h3 = "y" if row["hybrid_hit3"] else ("--" if row["hybrid_hit3"] is None else "n")
line += f" {hyb_d1:>9} {hyb_h3:>6}"
print(line)
print("-" * 100)
print(f"\nGate verdict at N={gate_result['gate_n']}, k={DEFAULT_K}:")
print(f" criterion 1 (no flat hit degrades) : {'PASS' if gate_result['criterion_1_no_degradation'] else 'FAIL'}")
print(f" criterion 1 (no flat hit degrades, cascade+hybrid): {'PASS' if gate_result['criterion_1_no_degradation'] else 'FAIL'}")
h3 = gate_result["criterion_2_hit3"]
print(f" criterion 2 (hit@3 cascade >= flat) : cascade={h3['cascade']}/{h3['total']} flat={h3['flat']}/{h3['total']} "
f"-> {'PASS' if gate_result['criterion_1_no_degradation'] and h3['cascade'] >= h3['flat'] else 'FAIL'}")
print(f" criterion 3 (negative controls > {NO_ANSWER_THRESHOLD}) : {'PASS' if gate_result['criterion_3_negative_controls'] else 'FAIL'}")
print(f" criterion 3 (negative controls > {NO_ANSWER_THRESHOLD}, flat+cascade+hybrid): {'PASS' if gate_result['criterion_3_negative_controls'] else 'FAIL'}")
c4 = gate_result["criterion_4_mail_hit3"]
if c4["skipped"]:
print(" criterion 4 (mail hit@3 in hybrid) : SKIPPED (mail_queries empty)")
else:
print(f" criterion 4 (mail hit@3 in hybrid) : {c4['hit3']}/{c4['total']} (required >= {c4['required']}) "
f"-> {'PASS' if gate_result['criterion_1_no_degradation'] else 'FAIL'}")
print(f"\n OVERALL: {'PASS' if gate_result['passed'] else 'FAIL'}")
for reason in gate_result["reasons"]:
print(f" - {reason}")
if mail_rows:
print("\nMail queries (hybrid only, plan §8 criterion 4):")
for row in mail_rows:
d1 = f"{row['hybrid_top1_dist']:.4f}" if row["hybrid_top1_dist"] is not None else "--"
h3 = "y" if row["hybrid_hit3"] else ("--" if row["hybrid_hit3"] is None else "n")
print(f" {row['id']:<4} {row['text']:<50} d1={d1:>8} hit3={h3}")
print("\nCost per query: flat = 1 embed + 1 SQL query. "
f"cascade = 1 embed (shared) + 2 SQL queries (stage1 top-N summaries, stage2 top-k chunks). "
"Cascade never costs an extra Ollama call, only one extra SQL round-trip.")
"hybrid = 1 embed (shared) + cascade's queries + 1 extra SQL query (mail branch).")
async def main_async(args: argparse.Namespace) -> dict:
queries = load_queries(Path(args.queries))
mail_queries = load_mail_queries(Path(args.queries))
n_values = [int(n) for n in args.n_sweep.split(",")]
if args.gate_n not in n_values:
n_values = sorted(set(n_values) | {args.gate_n})
@ -221,19 +300,31 @@ async def main_async(args: argparse.Namespace) -> dict:
async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=60)) as session:
results = []
for query in queries:
result = await run_query_both_tracks(
result = await run_query_all_tracks(
conn, session, args.ollama_url, query,
summary_model=args.summary_model, embed_model=args.embed_model,
n_values=n_values, k=args.k,
)
results.append(summarize_query_result(result))
mail_results = []
for query in mail_queries:
result = await run_query_all_tracks(
conn, session, args.ollama_url, query,
summary_model=args.summary_model, embed_model=args.embed_model,
n_values=n_values, k=args.k,
)
mail_results.append(summarize_query_result(result))
finally:
await conn.close()
gate_result = evaluate_gate(results, gate_n=args.gate_n)
print_report(results, n_values, gate_result)
gate_result = evaluate_gate(results, gate_n=args.gate_n, mail_rows=mail_results)
print_report(results, n_values, gate_result, mail_rows=mail_results)
output = {"rows": results, "gate": gate_result, "n_sweep": n_values, "k": args.k}
output = {
"rows": results, "mail_rows": mail_results, "gate": gate_result,
"n_sweep": n_values, "k": args.k,
}
if args.json_out:
Path(args.json_out).write_text(json.dumps(output, indent=2, default=str), encoding="utf-8")
print(f"\nFull JSON written to {args.json_out}")