rag/eval/run_batch.py

# -*- coding: utf-8 -*-
"""배치 평가: 골든셋으로 Context Recall/Precision (기계 채점, LLM 비용 0).

검색 대상: 각 질의회신의 **기준서 컬렉션만**(질의회신 컬렉션 제외).
  정답(expected_ref_keys)은 100% 기준서 문단이므로, 질의회신을 검색 대상에 섞으면
  유사 질의회신이 상위를 독식해 정답을 밀어냄(실측 recall 3.7%→26%로 7배 왜곡).
  → 질의회신 컬렉션을 원천 제외해 self-leakage도 동시 차단(자기 자신이 결과에 안 나옴).

지표(RAGAS 근사) — recall을 3단으로 병기(정직성):
  · 문단 recall (exact)   = 정확한 문단까지 회수한 비율        — 가장 엄격
  · 문단 recall (인접완화) = exact 1.0 + 인접 문단 0.5점        — 실용 하한
  · 호 recall             = 올바른 기준서(제NNNN호) 회수 비율   — 실질 성능 상한
  · 문단 precision(exact) = 검색 상위 k 중 정확 문단 비율
인접 판정(자의성 배제): 같은 호 · 같은 상위경로/접두 · 같은 꼬리 · 마지막 정수 차이 ±1.
  (예: 제1102호 B45 ↔ B46 인접 0.5점 / B45 ↔ B53(차이8) 오답 / 5.7.5 ↔ 5.8.1 오답)

결과: 전체 평균 + 게시판별 + top-k(5/10). batch_YYYYMMDD.json + summary.md.
사용: python3 -m rag.eval.run_batch [--sample 30] [--per-coll 50] [--date YYYYMMDD]
"""
import argparse
import json
import re
import time
import unicodedata
from collections import defaultdict

from rag import common as C
from rag.search import Index

GOLD = C.ROOT / "eval" / "goldenset.jsonl"
RESULTS = C.ROOT / "eval" / "results"

# 게시판 → 검색 대상 기준서 컬렉션 (질의회신 컬렉션은 검색 안 함)
STD_COLL = {"016001": "kifrs_standards", "016002": "kifrs_standards",
            "016005": "kifrs_standards", "016003": "kgaap_standards",
            "016006": "kgaap_standards"}

# 호(kifrs: 제1102호) + 장(kgaap 일반기업기준: 제5장) 모두 — 두 체계 혼재(질의회신 정답에 섞임)
_HO = re.compile(r"(제\d+[호장])")
_LABEL = re.compile(r"문단\s*(.+)$")
_INT = re.compile(r"\d+")


def ho_of(ref_key):
    """ref_key/section_key에서 기준서 단위(제NNNN호/제N장) 추출. kifrs=호, kgaap=장."""
    m = _HO.match(ref_key)
    return m.group(1) if m else None


def parse_para(ref_key):
    """문단키를 (호, 상위경로prefix, 마지막정수, 꼬리)로 분해. 용어섹션/파싱불가 → None."""
    hom = _HO.match(ref_key)
    if not hom:
        return None
    lm = _LABEL.search(ref_key)     # '문단 ' 뒤 라벨. '용어의 정의' 등은 매칭 안 됨
    if not lm:
        return None
    label = lm.group(1).strip()
    ints = list(_INT.finditer(label))
    if not ints:
        return None
    last = ints[-1]
    return (hom.group(1), label[:last.start()], int(last.group()), label[last.end():])


def adjacent(a, b, n=1):
    """두 문단키가 인접한가: 같은 호·상위경로·꼬리 & 마지막 정수 차이 0<Δ≤n."""
    pa, pb = parse_para(a), parse_para(b)
    if not pa or not pb:
        return False
    return (pa[0] == pb[0] and pa[1] == pb[1] and pa[3] == pb[3]
            and pa[2] != pb[2] and abs(pa[2] - pb[2]) <= n)


def got_keys(metas):
    """검색 결과 메타에서 회수한 ref_key/section_key 집합."""
    got = set()
    for m in metas:
        if m.get("ref_key"):
            got.add(m["ref_key"])
        if m.get("section_key"):
            got.add(m["section_key"])
    return got


# 표기 정규화: NFKC로 원문자·전각을 표준형으로 '풀어씀'(제거 아님) + 공백만 제거.
#   'B96⑷'→'B96(4)', 전각'（）'→'()'. 따라서 'B96⑷'↔'B96(4)'는 같게, 'B96⑷'↔'B96⑴'은
#   **다르게** 취급(하위항목 ⑴⑵⑷ 구분 보존). ← 원문자를 '제거'하면 다른 하위문단이 합쳐져
#   recall이 부당하게 부풀려짐(검증: 제거식은 기준서 고유키 3,131그룹 충돌 → false-positive).
#   NFKC식은 충돌 0. 실제 골든셋엔 괄호숫자↔원문자 같은 진짜 표기차가 거의 없어 효과 +0.0%p.
_WS = re.compile(r"\s+")


def _nk(k):
    return _WS.sub("", unicodedata.normalize("NFKC", k))


def score(expected, got):
    """정답 대비 3단 점수 반환: (exact_recall, relaxed_recall, ho_recall, exact_hit수).

    exact는 표기 정규화(_nk) 후 매칭 — 원문자·괄호·공백 차이는 같은 문단으로 인정.
    """
    if not expected:
        return 0.0, 0.0, 0.0, 0
    gotnorm = {_nk(g) for g in got}
    def hit(e):
        return _nk(e) in gotnorm
    exact_hits = sum(1 for e in expected if hit(e))
    relaxed = 0.0
    for e in expected:
        if hit(e):
            relaxed += 1.0
        elif any(adjacent(e, g) for g in got):
            relaxed += 0.5
    exp_ho = {ho_of(e) for e in expected if ho_of(e)}
    got_ho = {ho_of(g) for g in got if ho_of(g)}
    ho_recall = len(exp_ho & got_ho) / len(exp_ho) if exp_ho else 0.0
    return exact_hits / len(expected), relaxed / len(expected), ho_recall, exact_hits


def evaluate(index, golden, ks=(5, 10), per_coll=50, progress_every=50):
    agg = {k: defaultdict(lambda: {"exact": [], "relaxed": [], "ho": [], "prec": []})
           for k in ks}
    self_excluded = 0
    kmax = max(ks)
    n = len(golden)
    dump = []          # 원시 검색결과 — 채점 로직 바뀌면 재실행 없이 rescore.py로 재계산
    t_start = time.time()
    for qi, g in enumerate(golden, 1):
        scoll = STD_COLL[g["board"]]
        tq = time.time()
        # 기준서 컬렉션만 검색(질의회신 원천 제외). 넉넉히 kmax+3.
        hits = index.retrieve_routed(g["question"], [scoll], k=kmax + 3,
                                     min_standards=0, per_coll=per_coll)
        dt = time.time() - tq
        if qi == 1 or qi % progress_every == 0 or qi == n:
            elapsed = time.time() - t_start
            eta = elapsed / qi * (n - qi)
            print(f"  [{qi}/{n}] 쿼리 {dt:.1f}s · 누적 {elapsed:.0f}s · 예상잔여 {eta:.0f}s",
                  flush=True)
        # 안전망: 혹시 남을 self(doc_no 일치) 제외 — 기준서만 검색이라 사실상 0
        before = len(hits)
        hits = [h for h in hits if h.get("doc_no") != g["doc_no"]]
        self_excluded += before - len(hits)
        exp = g["expected_ref_keys"]
        for k in ks:
            got = got_keys([h["meta"] for h in hits[:k]])
            ex, rel, ho, hit = score(exp, got)
            for scope in (g["board"], "ALL"):
                agg[k][scope]["exact"].append(ex)
                agg[k][scope]["relaxed"].append(rel)
                agg[k][scope]["ho"].append(ho)
                agg[k][scope]["prec"].append(hit / k)
        dump.append({"id": g["id"], "board": g["board"], "expected": exp,
                     "hits": [{"ref_key": h["meta"].get("ref_key", ""),
                               "section_key": h["meta"].get("section_key", "")}
                              for h in hits[:kmax]]})
    return agg, self_excluded, dump


def summarize(agg, ks):
    def avg(x):
        return sum(x) / len(x) if x else 0.0
    out = {}
    for k in ks:
        out[k] = {b: {"exact": round(avg(v["exact"]), 4),
                      "relaxed": round(avg(v["relaxed"]), 4),
                      "ho": round(avg(v["ho"]), 4),
                      "precision": round(avg(v["prec"]), 4),
                      "n": len(v["exact"])}
                  for b, v in agg[k].items()}
    return out


def write_markdown(summary, ks, meta, path):
    a5, a10 = summary[5]["ALL"], summary[10]["ALL"]
    L = ["# 배치 평가 — Context Recall/Precision (기계 채점, LLM 비용 0)", "",
         f"- 골든셋 **{meta['n']}건** (질의회신 질문 → 정답=인용 기준서 문단, 조인 성공분만).",
         f"- 검색 대상: **각 질의회신의 기준서 컬렉션만**(질의회신 컬렉션 제외) · "
         f"per_coll={meta['per_coll']} · dense+리랭킹(bge-reranker-v2-m3, fp16).",
         f"- **self-leakage 차단**: 질의회신 컬렉션을 검색 대상에서 원천 제외 → 자기 글이 "
         f"결과에 안 나옴(추가 doc_no 필터로 제외된 잔여 {meta['self_excluded']}건).",
         f"- 소요 {meta['elapsed_s']}s ({meta['elapsed_s']/max(meta['n'],1):.1f}s/건).", "",
         "## 성능 (전체 평균)", "",
         "| 지표 | top-5 | top-10 | 무엇을 재나 / 무엇을 못 재나 |",
         "|---|---|---|---|",
         f"| 문단 recall — exact | {a5['exact']:.3f} | {a10['exact']:.3f} | "
         "정확한 문단까지 회수. **가장 엄격**(표기 정규화 NFKC 적용하나 현재 데이터엔 효과 +0.0%p) |",
         f"| 문단 recall — 인접완화 | {a5['relaxed']:.3f} | {a10['relaxed']:.3f} | "
         "정답 문단은 1.0, **±1 인접 문단은 0.5**점. exact가 놓치는 '옆 문단' 회수를 반영(실용 하한) |",
         f"| 호 recall — 기준서 단위 | {a5['ho']:.3f} | {a10['ho']:.3f} | "
         "올바른 **기준서(제NNNN호)**를 찾았는가. **실질 성능에 가까움**. 단, 문단 정밀도는 못 잼 |",
         f"| 문단 precision — exact | {a5['precision']:.3f} | {a10['precision']:.3f} | "
         "상위 k 중 정확 문단 비율 |", "",
         "### 인접완화 채점 기준 (자의적 후함 배제)",
         "- 인접 = **같은 호 · 같은 상위경로(예: `5.7.`)/접두(예: `B`,`AG`) · 같은 꼬리 · "
         "마지막 정수 차이 정확히 ±1**. 이때만 0.5점.",
         "- 예: `제1102호 B45`↔`B46` 인접(0.5) / `B45`↔`B53`(차이 8) 오답 / "
         "`5.7.5`↔`5.8.1`(상위경로 다름) 오답 / `3A`↔`3`(꼬리 다름) 오답.",
         "- exact를 **대체하지 않고 별도 병기**. 인접완화 ≥ exact는 구조상 보장(exact 점수를 낮추지 않음).", "",
         "### 왜 문단 exact가 낮은가 (정직한 해석)",
         "- 질의회신 1건당 인용 문단 **평균 2.1개·최대 14개** — 여러 문단을 모두 top-k에 넣어야 만점.",
         "- 인용이 **번호를 축약/범위로** 표기하는 경우가 있어 레코드 단위 exact와 어긋남.",
         "- 임베딩·리랭킹은 **올바른 기준서(호)는 잘 찾지만**(위 호 recall) 인용된 정확한 문단 "
         "번호까지 pinpoint하는 것은 본질적으로 어려움 → exact는 하한, 호는 상한, 인접완화는 그 사이.",
         "- 수치를 부풀리지 않음: 세 지표를 모두 공개하고 각자의 한계를 명시.", "",
         "### 문단 exact 실패 원인 분류 (dump 1,956건 자동 분류)",
         "- 표기 정규화(NFKC)로 원문자·괄호·전각 차이는 exact에 이미 흡수 → 남은 실패는 "
         "**진짜 검색 실패 100%**. 골든셋엔 `괄호숫자↔원문자` 같은 진짜 표기차가 거의 없어 "
         "**정규화 효과 +0.0%p**(원문자 '제거'식은 `B96⑴/⑵/⑷`를 합쳐 +3.4%p 부풀렸으나 "
         "false-positive라 폐기 → 문단 exact 27%는 채점 아티팩트가 아닌 실제 검색 성능).",
         "- 실패 내역: **62.6% 호는 맞고 정확 문단만 놓침**(문단 pinpoint 한계) · "
         "30.0% 기준서(호)조차 못 찾음(질의회신이 여러 기준서에 광범위 인용) · 7.4% ±1 인접까지 감.",
         "- **답변 품질 영향은 제한적**: 실패의 63%는 올바른 기준서를 이미 회수(호 recall "
         f"{a10['ho']:.1%})했고 인접 문단을 근거로 제시 → 사용자가 정답 문단 부근 원문을 받게 됨.", "",
         "## 게시판별 (top-10)", "",
         "| 게시판 | 문단 exact | 인접완화 | 호 recall | n |", "|---|---|---|---|---|"]
    for b in ("016001", "016002", "016003", "016005", "016006"):
        if b in summary[10]:
            s = summary[10][b]
            L.append(f"| {b} | {s['exact']:.3f} | {s['relaxed']:.3f} | {s['ho']:.3f} | {s['n']} |")
    path.write_text("\n".join(L) + "\n", encoding="utf-8")


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--sample", type=int, default=0, help="상위 N건만(0=전체)")
    ap.add_argument("--per-coll", type=int, default=50)
    ap.add_argument("--date", default="test")
    args = ap.parse_args()

    golden = [json.loads(l) for l in GOLD.open(encoding="utf-8")]
    if args.sample:
        golden = golden[:args.sample]
    prog = 5 if len(golden) <= 60 else 50
    print(f"평가 대상 {len(golden)}건, Index 로드...", flush=True)
    idx = Index()
    ks = (5, 10)
    t0 = time.time()
    agg, self_excluded, dump = evaluate(idx, golden, ks=ks, per_coll=args.per_coll,
                                        progress_every=prog)
    summary = summarize(agg, ks)
    meta = {"n": len(golden), "self_excluded": self_excluded,
            "per_coll": args.per_coll, "elapsed_s": round(time.time() - t0, 1)}

    RESULTS.mkdir(parents=True, exist_ok=True)
    (RESULTS / f"batch_{args.date}.json").write_text(
        json.dumps({"meta": meta, "summary": summary}, ensure_ascii=False, indent=2),
        encoding="utf-8")
    # 원시 검색결과 저장 → 채점 로직 변경 시 rescore.py로 재실행 없이 재계산(원시 로그, gitignore)
    with (RESULTS / f"retrieval_{args.date}.jsonl").open("w", encoding="utf-8") as f:
        for d in dump:
            f.write(json.dumps(d, ensure_ascii=False) + "\n")
    write_markdown(summary, ks, meta, RESULTS / "summary.md")

    a5, a10 = summary[5]["ALL"], summary[10]["ALL"]
    print(f"\n=== 결과 (n={meta['n']}, {meta['elapsed_s']}s, self제외 {self_excluded}) ===")
    print(f"  문단 recall  exact  top5={a5['exact']:.3f} top10={a10['exact']:.3f}")
    print(f"  문단 recall  인접완화 top5={a5['relaxed']:.3f} top10={a10['relaxed']:.3f}")
    print(f"  호   recall         top5={a5['ho']:.3f} top10={a10['ho']:.3f}")
    print(f"  문단 precision exact top5={a5['precision']:.3f} top10={a10['precision']:.3f}")


if __name__ == "__main__":
    main()