""" 평가셋 300개 기반 검색엔진 품질 평가 ===================================== 실행: .venv/Scripts/python.exe scripts/22_eval300.py .venv/Scripts/python.exe scripts/22_eval300.py --no-hyde .venv/Scripts/python.exe scripts/22_eval300.py --save-xlsx # 결과 엑셀 저장 """ import sys import re import json import argparse from pathlib import Path from collections import defaultdict sys.path.insert(0, str(Path(__file__).parent.parent)) from core.search_engine import MenuSearchEngine EVAL_PATH = Path(__file__).parent.parent / "data" / "eval_queries_300.json" RESULT_PATH = Path(__file__).parent.parent / "data" / "eval_results_300.json" XLSX_PATH = Path(__file__).parent.parent / "data" / "eval_results_300.xlsx" def normalize_path(path: str) -> str: if not path: return "" path = re.sub(r"\s*>\s*", ">", str(path)) path = re.sub(r"\(.*?\)", "", path) return path.strip().lower() def matches(pred: str, ans: str) -> bool: if not pred or not ans: return False if pred == ans: return True pred_end = pred.split(">")[-1].strip() ans_end = ans.split(">")[-1].strip() if pred_end and ans_end and pred_end == ans_end: pp = pred.split(">") ap = ans.split(">") return sum(1 for p in pp if p in ap) >= min(2, len(ap)) return False def save_xlsx(detail_rows: list[dict], summary: dict, xlsx_path: Path): """평가 결과를 엑셀로 저장 (상세 + 요약 시트)""" import pandas as pd from openpyxl import load_workbook from openpyxl.styles import PatternFill, Font, Alignment df = pd.DataFrame(detail_rows) df = df[["id", "rank", "source", "결과", "menu_id", "menu_name", "query", "정답경로", "예측1위", "예측2위", "예측3위", "예측4위", "예측5위", "mrr"]] with pd.ExcelWriter(xlsx_path, engine="openpyxl") as writer: df.to_excel(writer, index=False, sheet_name="상세결과") # 요약 시트 rows_summary = [ ["지표", "값"], ["평가셋 (n)", summary["n"]], ["HyDE", summary["hyde"]], ["Acc@1", f"{summary['acc1']:.1%}"], ["Acc@3", f"{summary['acc3']:.1%}"], ["Acc@5", f"{summary['acc5']:.1%}"], ["MRR@5", f"{summary['mrr']:.3f}"], [], ["[소스별]", "Acc@1", "Acc@3", "Acc@5", "n"], ] for src_row in summary.get("by_source", []): rows_summary.append([src_row["source"], f"{src_row['acc1']:.1%}", f"{src_row['acc3']:.1%}", f"{src_row['acc5']:.1%}", src_row["n"]]) rows_summary.append([]) rows_summary.append(["[순위구간별]", "Acc@1", "Acc@5", "n"]) for rk_row in summary.get("by_rank", []): rows_summary.append([rk_row["group"], f"{rk_row['acc1']:.1%}", f"{rk_row['acc5']:.1%}", rk_row["n"]]) ws_s = writer.book.create_sheet("요약") for r in rows_summary: ws_s.append(r) # 상세 시트 서식 ws = writer.sheets["상세결과"] col_widths = [5, 8, 10, 7, 14, 20, 38, 42, 42, 42, 42, 42, 42, 8] for i, w in enumerate(col_widths, 1): from openpyxl.utils import get_column_letter ws.column_dimensions[get_column_letter(i)].width = w green = PatternFill("solid", start_color="C6EFCE") yellow = PatternFill("solid", start_color="FFEB9C") red = PatternFill("solid", start_color="FFC7CE") for row in ws.iter_rows(min_row=2, max_row=ws.max_row): label = row[3].value # 결과 컬럼 fill = green if label == "O" else (yellow if label == "~" else red) row[3].fill = fill print(f"\n엑셀 저장: {xlsx_path}") def main(): parser = argparse.ArgumentParser() parser.add_argument("--no-hyde", action="store_true", help="HyDE 비활성화") parser.add_argument("--save-xlsx", action="store_true", help="결과 엑셀 저장") args = parser.parse_args() use_hyde = not args.no_hyde with open(EVAL_PATH, encoding="utf-8") as f: eval_set = json.load(f) print(f"평가셋: {len(eval_set)}개 | HyDE: {'ON' if use_hyde else 'OFF'}") print("검색엔진 초기화 중...") engine = MenuSearchEngine.get_instance() print("준비 완료\n") acc1, acc3, acc5, mrr_list = [], [], [], [] wrong_cases = [] detail_rows = [] by_source = defaultdict(lambda: {"acc1": [], "acc3": [], "acc5": []}) for item in eval_set: query = item["query"] menu_path = item["menu_path"] source = item["source"] hits = engine.search(query, top_n=5, threshold=0.0, use_hyde=use_hyde) paths = [normalize_path(h.get("menu_path", "")) for h in hits] names = [h.get("menu_name", "") for h in hits] ans = normalize_path(menu_path) hit1 = matches(paths[0], ans) if paths else False hit3 = any(matches(p, ans) for p in paths[:3]) hit5 = any(matches(p, ans) for p in paths[:5]) # MRR rr = 0.0 for k, p in enumerate(paths[:5], 1): if matches(p, ans): rr = 1.0 / k break mrr_list.append(rr) acc1.append(hit1) acc3.append(hit3) acc5.append(hit5) by_source[source]["acc1"].append(hit1) by_source[source]["acc3"].append(hit3) by_source[source]["acc5"].append(hit5) label = "O" if hit1 else ("~" if hit3 else "X") print(f" [{label}] [{item['id']:3d}] {query[:30]:<30} -> {paths[0] if paths else '(없음)'}") if not hit5: wrong_cases.append({"id": item["id"], "query": query, "expected": menu_path, "got": paths[0] if paths else ""}) # 상세 행 pad = lambda lst, n: lst + [""] * (n - len(lst)) p5 = pad(paths, 5) detail_rows.append({ "id": item["id"], "rank": item["rank"], "source": source, "결과": label, "menu_id": item["menu_id"], "menu_name": item["menu_name"], "query": query, "정답경로": menu_path, "예측1위": p5[0], "예측2위": p5[1], "예측3위": p5[2], "예측4위": p5[3], "예측5위": p5[4], "mrr": round(rr, 4), }) n = len(acc1) print(f"\n{'='*62}") print(f"== 평가 결과 (n={n}) HyDE={'ON' if use_hyde else 'OFF'} ==") print(f"{'='*62}") print(f" Acc@1 : {sum(acc1)}/{n} = {sum(acc1)/n:.1%}") print(f" Acc@3 : {sum(acc3)}/{n} = {sum(acc3)/n:.1%}") print(f" Acc@5 : {sum(acc5)}/{n} = {sum(acc5)/n:.1%}") print(f" MRR@5 : {sum(mrr_list)/n:.3f}") print(f"{'='*62}") # 소스별 분석 print(f"\n[소스별]") src_rows = [] for src in ["human", "llm"]: d = by_source[src] if not d["acc1"]: continue ns = len(d["acc1"]) print(f" {src:8s} (n={ns:3d}) Acc@1={sum(d['acc1'])/ns:.1%} " f"Acc@3={sum(d['acc3'])/ns:.1%} Acc@5={sum(d['acc5'])/ns:.1%}") src_rows.append({"source": src, "n": ns, "acc1": sum(d["acc1"])/ns, "acc3": sum(d["acc3"])/ns, "acc5": sum(d["acc5"])/ns}) # top200 vs 201+ 분석 by_rank = defaultdict(lambda: {"acc1": [], "acc5": []}) for item, a1, a5 in zip(eval_set, acc1, acc5): grp = "top200" if item["rank"] <= 200 else "201+" by_rank[grp]["acc1"].append(a1) by_rank[grp]["acc5"].append(a5) print(f"\n[순위 구간별]") rk_rows = [] for grp in ["top200", "201+"]: d = by_rank[grp] if not d["acc1"]: continue ng = len(d["acc1"]) print(f" {grp:8s} (n={ng:3d}) Acc@1={sum(d['acc1'])/ng:.1%} Acc@5={sum(d['acc5'])/ng:.1%}") rk_rows.append({"group": grp, "n": ng, "acc1": sum(d["acc1"])/ng, "acc5": sum(d["acc5"])/ng}) # Top5 완전 오답 print(f"\n[Top5 오답 {len(wrong_cases)}개]") for w in wrong_cases[:10]: print(f" [{w['id']:3d}] {w['query'][:28]:<28}") print(f" 정답: {w['expected']}") print(f" 예측: {w['got']}") # 결과 저장 summary = { "n": n, "hyde": "ON" if use_hyde else "OFF", "acc1": sum(acc1)/n, "acc3": sum(acc3)/n, "acc5": sum(acc5)/n, "mrr": sum(mrr_list)/n, "by_source": src_rows, "by_rank": rk_rows, } with open(RESULT_PATH, "w", encoding="utf-8") as f: json.dump({"summary": summary, "detail": detail_rows}, f, ensure_ascii=False, indent=2) print(f"\n결과 JSON 저장: {RESULT_PATH}") if args.save_xlsx: save_xlsx(detail_rows, summary, XLSX_PATH) if __name__ == "__main__": main()