scripts/40_bix_regression.py

"""
BIX 현업 회귀 채점 하네스
==========================
현업이 전달한 평가셋으로 우리(로컬) 엔진을 직접 채점한다.

입력:
  - BIX_채점원장_302.xlsx (시트 '채점원장'): 302문항 골든셋 (질의/정답_menu_id)
  - BIX_회귀_통합.json: 확정실패 29건 + 도달성(reachability) 케이스

출력:
  - 302문항 strict Top-1 / Top-3 (현업 기준 67.5% / 86.1% 대비)
  - 확정실패 29건 통과율 (expected_menu_top1 정책 + forbidden_domain 위반 체크)
  - 도달성 케이스 Top-3 노출 여부

실행:
  .venv/Scripts/python.exe scripts/40_bix_regression.py            # HyDE off (빠름)
  .venv/Scripts/python.exe scripts/40_bix_regression.py --hyde     # HyDE on (프로덕션 동일)
  .venv/Scripts/python.exe scripts/40_bix_regression.py --only29   # 29건만 (빠름)
"""
import sys
import json
import argparse
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent.parent))

import pandas as pd

ROOT = Path(__file__).parent.parent
BIX_DIR = ROOT.parent / "BIX_메뉴검색_20260624"
LEDGER_XLSX = BIX_DIR / "BIX_채점원장_302.xlsx"
REG_JSON = BIX_DIR / "BIX_회귀_통합.json"


def _topk_ids(result: dict, k: int) -> list[str]:
    hits = result.get("search_results", [])[:k]
    return [h.get("menu_id", "") for h in hits]


def _topk_paths(result: dict, k: int) -> list[str]:
    hits = result.get("search_results", [])[:k]
    return [h.get("menu_path", "") for h in hits]


def run_302(run_agent, use_hyde: bool) -> dict:
    df = pd.read_excel(LEDGER_XLSX, sheet_name="채점원장")
    top1_hit = top3_hit = 0
    misses = []
    for _, row in df.iterrows():
        q = str(row["질의"]).strip()
        gold = str(row["정답_menu_id"]).strip()
        r = run_agent(query=q, top_n=3, use_hyde=use_hyde, use_reranker=False)
        ids = _topk_ids(r, 3)
        if ids and ids[0] == gold:
            top1_hit += 1
        if gold in ids:
            top3_hit += 1
        else:
            misses.append((row["eval_case_id"], q, gold, ids))
    n = len(df)
    return {
        "n": n, "top1": top1_hit, "top3": top3_hit,
        "top1_pct": top1_hit / n * 100, "top3_pct": top3_hit / n * 100,
        "misses": misses,
    }


def run_29(run_agent, use_hyde: bool) -> dict:
    data = json.loads(REG_JSON.read_text(encoding="utf-8"))
    cases = data["confirmed_failures"]
    passed = []
    failed = []
    for c in cases:
        q = c["query"]
        expected = set(c.get("expected_menu_ids", []))
        forbidden = c.get("forbidden_domains", []) or []
        r = run_agent(query=q, top_n=3, use_hyde=use_hyde, use_reranker=False)
        ids = _topk_ids(r, 3)
        paths = _topk_paths(r, 3)
        top1_ok = bool(ids) and ids[0] in expected
        # forbidden 도메인이 Top-1에 오면 위반
        top1_path = paths[0] if paths else ""
        forbidden_hit = any(top1_path.startswith(fd) for fd in forbidden)
        ok = top1_ok and not forbidden_hit
        rec = {
            "id": c["regression_case_id"], "query": q,
            "expected": list(expected), "top1": ids[0] if ids else "-",
            "top1_path": top1_path, "type": c.get("failure_type", ""),
        }
        (passed if ok else failed).append(rec)
    return {"n": len(cases), "passed": passed, "failed": failed}


def run_reach(run_agent, use_hyde: bool) -> dict:
    data = json.loads(REG_JSON.read_text(encoding="utf-8"))
    cases = data.get("reachability", [])
    results = []
    for c in cases:
        q = c["query"]
        target = c["menu_id"]
        r = run_agent(query=q, top_n=3, use_hyde=use_hyde, use_reranker=False)
        ids = _topk_ids(r, 3)
        in_top3 = target in ids
        rank = ids.index(target) + 1 if in_top3 else None
        results.append({
            "id": c["regression_case_id"], "query": q, "target": target,
            "menu_name": c.get("menu_name", ""), "in_top3": in_top3, "rank": rank,
        })
    return {"n": len(cases), "results": results}


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--hyde", action="store_true", help="HyDE on (프로덕션 동일)")
    ap.add_argument("--only29", action="store_true", help="29건+도달성만 (빠름)")
    args = ap.parse_args()

    from core.agent import run_agent

    print(f"\n{'='*60}")
    print(f"BIX 회귀 채점 (HyDE={'ON' if args.hyde else 'OFF'})")
    print(f"현업 기준(2026-06-18 라이브): Top-1 67.5% / Top-3 86.1%")
    print(f"{'='*60}\n")

    # ── 확정실패 29건 ──
    r29 = run_29(run_agent, args.hyde)
    print(f"[확정실패 29건] 통과 {len(r29['passed'])}/{r29['n']}")
    print("  ── 통과 ──")
    for p in r29["passed"]:
        print(f"   ✓ {p['id']:18} '{p['query']}' → {p['top1_path']}")
    print("  ── 여전히 실패 ──")
    for f in r29["failed"]:
        print(f"   ✗ {f['id']:18} '{f['query']}' → {f['top1_path']} (기대 {f['expected']})")
    print()

    # ── 도달성 ──
    rr = run_reach(run_agent, args.hyde)
    in3 = sum(1 for x in rr["results"] if x["in_top3"])
    print(f"[도달성] Top-3 노출 {in3}/{rr['n']}")
    for x in rr["results"]:
        mark = f"{x['rank']}위" if x["in_top3"] else "미노출"
        print(f"   {x['target']} {x['menu_name']:12} '{x['query']}' → {mark}")
    print()

    if args.only29:
        return

    # ── 302 전체 ──
    print("[302 전체] 채점 중... (시간 소요)")
    r302 = run_302(run_agent, args.hyde)
    print(f"\n[302 결과] Top-1 {r302['top1']}/{r302['n']} ({r302['top1_pct']:.1f}%)  "
          f"Top-3 {r302['top3']}/{r302['n']} ({r302['top3_pct']:.1f}%)")
    print(f"  (현업 기준 대비: Top-1 {r302['top1_pct']-67.5:+.1f}p / Top-3 {r302['top3_pct']-86.1:+.1f}p)")
    print(f"\n  Top-3 미포함 {len(r302['misses'])}건 (상위 15):")
    for cid, q, gold, ids in r302["misses"][:15]:
        print(f"   {cid} '{q}' 정답={gold} 우리Top3={ids}")


if __name__ == "__main__":
    main()