Initial commit: ABAP indexing pipeline (ingest, parser, summarize, index, query, wiki)

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

eval/run_eval.py

@@ -0,0 +1,108 @@
+"""Stage 6 — 평가. 질문셋으로 검색 recall / 조각 적중 / trace unit 포함률을 측정한다.
+
+python -m eval.run_eval  (인덱스 DB 적재 상태에서, API 서버 없이 tools 직접 호출)
+결과는 eval/results/<date>.json 에 저장 — 프롬프트/사전 변경 전후 비교용.
+
+질문 유형 (eval/questions.jsonl):
+- find    : {"question", "expected_programs"}                      → search_programs recall@5/@10
+- logic   : {"question", "expected": [{"program","line_from","line_to"}]}
+            → search_logic 상위 조각 중 프로그램이 같고 줄 범위가 겹치면 적중. hit@10, MRR
+- trace   : {"program", "symbol", "expected_units"}                 → trace_variable unit 포함률
+- explain : {"program", "expected_keywords"}                        → 요약 텍스트 키워드 적중률
+"""
+from __future__ import annotations
+
+import json
+from datetime import date
+from pathlib import Path
+
+from query import tools
+
+HERE = Path(__file__).resolve().parent
+
+
+def _flatten_units(writes: list[dict], acc: set[str]) -> None:
+    for w in writes:
+        acc.add(w.get("unit", ""))
+        if w.get("callee"):
+            acc.add(w["callee"].upper())
+        _flatten_units(w.get("callee_writes", []), acc)
+
+
+def _overlaps(chunk: dict, exp: dict) -> bool:
+    if chunk["program"] != exp["program"].upper():
+        return False
+    lo, hi = int(exp.get("line_from", 0)), int(exp.get("line_to", 10**9))
+    return chunk["line_start"] <= hi and chunk["line_end"] >= lo
+
+
+def eval_logic(q: dict) -> dict:
+    res = tools.search_logic(q["question"], top_k=10)
+    ranked = [c for g in res["programs"] for c in g["chunks"]]
+    ranked.sort(key=lambda c: -c["score"])
+    expected = q["expected"]
+    hits, first_rank = 0, None
+    for exp in expected:
+        for i, c in enumerate(ranked[:10], 1):
+            if _overlaps(c, exp):
+                hits += 1
+                first_rank = i if first_rank is None else min(first_rank, i)
+                break
+    return {
+        "hit@10": hits / len(expected) if expected else 0.0,
+        "mrr": (1.0 / first_rank) if first_rank else 0.0,
+        "found": [f"{c['program']} {c['include']} L{c['line_start']}-{c['line_end']}" for c in ranked[:10]],
+    }
+
+
+def main() -> None:
+    questions = [json.loads(l) for l in (HERE / "questions.jsonl").read_text(encoding="utf-8").splitlines() if l.strip()]
+    results = []
+    for q in questions:
+        r: dict = {"id": q["id"], "type": q["type"], "question": q.get("question", "")}
+        try:
+            if q["type"] == "find":
+                found = [x["program"] for x in tools.search_programs(q["question"], top_k=10)]
+                expected = set(q["expected_programs"])
+                r["recall@5"] = len(expected & set(found[:5])) / len(expected)
+                r["recall@10"] = len(expected & set(found[:10])) / len(expected)
+                r["found"] = found[:10]
+            elif q["type"] == "logic":
+                r.update(eval_logic(q))
+            elif q["type"] == "trace":
+                t = tools.trace_variable(q["program"], q["symbol"])
+                units: set[str] = set()
+                _flatten_units(t["writes"], units)
+                expected = set(u.upper() for u in q["expected_units"])
+                r["unit_hit_rate"] = len(expected & units) / len(expected)
+                r["units"] = sorted(units)
+            elif q["type"] == "explain":
+                s = tools.get_program_summary(q["program"])
+                text = json.dumps(s, ensure_ascii=False).upper()
+                hits = [k for k in q["expected_keywords"] if k.upper() in text]
+                r["keyword_hit_rate"] = len(hits) / len(q["expected_keywords"])
+        except Exception as e:  # noqa: BLE001
+            r["error"] = f"{type(e).__name__}: {e}"
+        results.append(r)
+
+    # 유형별 평균
+    agg: dict[str, dict[str, float]] = {}
+    for r in results:
+        for k, v in r.items():
+            if isinstance(v, (int, float)) and k not in ("id",):
+                a = agg.setdefault(r["type"], {}).setdefault(k, [0.0, 0])
+                a[0] += v
+                a[1] += 1
+    summary = {t: {k: round(v[0] / v[1], 3) for k, v in m.items()} for t, m in agg.items()}
+
+    out_dir = HERE / "results"
+    out_dir.mkdir(exist_ok=True)
+    out_path = out_dir / f"{date.today().isoformat()}.json"
+    out_path.write_text(json.dumps({"summary": summary, "results": results}, ensure_ascii=False, indent=2),
+                        encoding="utf-8")
+    print(json.dumps({"summary": summary, "results": results}, ensure_ascii=False, indent=2))
+    print(f"저장: {out_path}")
+
+
+if __name__ == "__main__":
+    main()