File size: 2,824 Bytes
de1e3fc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | """Matching evaluation harness for threshold tuning (spec §9.6).
Given a labeled set of photo pairs (same-dog vs different-dog), report ROC/AUC and
precision/recall at candidate thresholds. Use this to tune REVIEW_THRESHOLD / STRONG_THRESHOLD
and to compare embedders. Works with whatever Embedder is configured (mock by default).
Input CSV format (header required):
image_a,image_b,same
/path/1.jpg,/path/2.jpg,1
/path/1.jpg,/path/3.jpg,0
Usage:
python -m scripts.eval_matching pairs.csv
"""
from __future__ import annotations
import csv
import sys
import numpy as np
from app.ml import get_embedder
def _load_pairs(path: str) -> list[tuple[str, str, int]]:
out = []
with open(path, newline="", encoding="utf-8") as fh:
for row in csv.DictReader(fh):
out.append((row["image_a"], row["image_b"], int(row["same"])))
return out
def _cosine(a: np.ndarray, b: np.ndarray) -> float:
return float(np.dot(a, b)) # embeddings are L2-normalized
def evaluate(pairs: list[tuple[str, str, int]]) -> dict:
embedder = get_embedder()
scores: list[float] = []
labels: list[int] = []
for a, b, same in pairs:
va, vb = embedder.embed([a, b])
scores.append(_cosine(va, vb))
labels.append(same)
scores_arr = np.array(scores)
labels_arr = np.array(labels)
# Sweep thresholds; compute precision/recall and a trapezoidal ROC-AUC.
thresholds = np.linspace(0, 1, 101)
rows = []
tpr_list, fpr_list = [], []
pos = max(int(labels_arr.sum()), 1)
neg = max(int((1 - labels_arr).sum()), 1)
for t in thresholds:
pred = scores_arr >= t
tp = int(((pred == 1) & (labels_arr == 1)).sum())
fp = int(((pred == 1) & (labels_arr == 0)).sum())
fn = int(((pred == 0) & (labels_arr == 1)).sum())
precision = tp / (tp + fp) if (tp + fp) else 1.0
recall = tp / (tp + fn) if (tp + fn) else 0.0
rows.append((float(t), precision, recall))
tpr_list.append(tp / pos)
fpr_list.append(fp / neg)
# AUC via trapezoid over sorted FPR.
order = np.argsort(fpr_list)
auc = float(np.trapz(np.array(tpr_list)[order], np.array(fpr_list)[order]))
return {"auc": abs(auc), "embedder": embedder.name, "rows": rows}
def main() -> None:
if len(sys.argv) < 2:
print(__doc__)
sys.exit(1)
pairs = _load_pairs(sys.argv[1])
result = evaluate(pairs)
print(f"Embedder: {result['embedder']} ROC-AUC: {result['auc']:.3f}")
print(f"{'thresh':>7} {'precision':>10} {'recall':>8}")
for t, p, r in result["rows"][::10]: # every 0.1
print(f"{t:7.2f} {p:10.3f} {r:8.3f}")
print("\nTune REVIEW_THRESHOLD / STRONG_THRESHOLD from these curves; record findings in DECISIONS.md.")
if __name__ == "__main__":
main()
|