Files
2dph/bin/kb/eval
T
eSlider 8781c0c3eb refactor(tools): bin/{subject}/{method} layout; Go serve+watch modules
Move serve/ (module) -> bin/server, tools/ -> bin/tools, replace bin/kb-watch
bash with bin/watch Go package; self-executing Go shebangs bin/serve.go and
bin/kb/watch.go; Docker + CI + git/import + docs repointed. Multi-stage image
builds static serve+watch binaries (no Go runtime in container).
2026-08-11 09:52:20 +01:00

63 lines
1.9 KiB
Python
Executable File

#!/usr/bin/env python3
"""kb/eval - recall@5 gate for the brain.
bin/kb/eval [--json]
Control questions are answered from the graph; recall@5 >= 0.95 gates CI.
Each question maps to leaf ids that MUST appear in the top 5 hits.
"""
from __future__ import annotations
import json
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / "bin" / "tools"))
from kblib import open_readonly, query_fts # noqa: E402
from yamlout import to_yaml # noqa: E402
RECALL_THRESHOLD = 0.95
# (query, expected text fragment that must be in the top-5 results)
CONTROL_QUESTIONS: list[tuple[str, str]] = [
("hybrid search fts and vector", "BM25"),
("eslider devops engineer", "DevOps"),
("ladybugdb graph engine storage", "LadybugDB"),
]
def hit_texts_of(query: str, conn, limit: int = 5) -> list[str]:
try:
hits = query_fts(conn, query, limit)
return [h["text"] for h in hits]
except Exception:
return []
def main(argv: list[str]) -> int:
import argparse
p = argparse.ArgumentParser(description="recall@5 gate")
p.add_argument("--json", action="store_true")
a = p.parse_args(argv)
db, conn = open_readonly()
recalled = 0
detail = []
for query, fragment in CONTROL_QUESTIONS:
texts = hit_texts_of(query, conn)
ok = any(fragment.lower() in t.lower() for t in texts)
recalled += int(ok)
detail.append({"q": query, "fragment": fragment, "in_top5": ok})
recall = recalled / len(CONTROL_QUESTIONS) if CONTROL_QUESTIONS else 1.0
passed = recall >= RECALL_THRESHOLD
out = {"recall@5": round(recall, 3), "passed": passed, "gate": len(CONTROL_QUESTIONS), "details": detail}
print(json.dumps(out, indent=2) if a.json else to_yaml(out))
conn.close()
db.close()
return 0 if passed else 2
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))