o2o-plagiarism-ai/scripts/eval_rouge.py

#!/usr/bin/env python3
"""요약 성능(ROUGE) 평가 하니스 — 계획서 성능지표 No.7.

데이터(요약 정답셋)가 들어오면 즉시 측정할 수 있도록 평가환경을 선구축한다.

입력 형식 (JSONL, 한 줄당 한 건):
    {"text": "원문 ...", "reference": "사람 작성 요약 정답 ..."}
  또는 이미 시스템 요약이 있는 경우:
    {"system": "엔진 요약 ...", "reference": "정답 요약 ..."}
  참조 요약이 여러 개면 (계획서 수식이 Σ_S∈refs 로 다중 참조를 전제한다):
    {"text": "원문 ...", "references": ["정답 요약 1 ...", "정답 요약 2 ..."]}

동작:
  - "text" 만 있으면 자체 Summarizer 로 system 요약을 생성한 뒤 reference 와 비교
  - "system" 이 있으면 그대로 사용
  - 정답셋 파일이 없으면 내장 dry-run 샘플로 파이프라인 동작만 검증

사용:
    python -m scripts.eval_rouge                          # dry-run (내장 샘플)
    python -m scripts.eval_rouge data/eval/summary.jsonl  # 정답셋 평가
    python -m scripts.eval_rouge data/eval/summary.jsonl --mode char --ratio 0.3
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

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

from app.engine.rouge import evaluate_pairs  # noqa: E402
from app.engine.summarizer import get_summarizer  # noqa: E402

# 정답셋 도착 전 파이프라인 검증용 내장 샘플 (원문/정답 요약)
_DRY_RUN = [
    {
        "text": (
            "홍길동은 조선시대 의적이었다. 그는 활빈당을 만들어 탐관오리의 재물을 빼앗았다. "
            "빼앗은 재물은 가난한 백성들에게 나누어 주었다. 조정에서는 그를 잡으려 했으나 실패했다. "
            "결국 홍길동은 율도국으로 떠나 새 나라를 세웠다."
        ),
        "reference": "홍길동은 활빈당을 만들어 탐관오리의 재물을 빼앗아 백성에게 나눠주고 율도국을 세웠다.",
    },
    {
        "text": (
            "어린 왕자는 자신의 작은 별을 떠나 여러 행성을 여행했다. 여행 중 다양한 어른들을 만났다. "
            "지구에서 여우를 만나 관계의 의미를 배웠다. 그는 자신의 장미가 소중함을 깨달았다."
        ),
        "reference": "어린 왕자는 여러 행성을 여행하며 여우를 통해 자신의 장미가 소중함을 깨닫는다.",
    },
]


def _load(path: str | None) -> tuple[list[dict], bool]:
    if path is None:
        return _DRY_RUN, True
    p = Path(path)
    if not p.exists():
        print(f"[warn] 파일 없음: {path} → 내장 dry-run 샘플로 진행", file=sys.stderr)
        return _DRY_RUN, True
    rows = [json.loads(line) for line in p.read_text(encoding="utf-8").splitlines() if line.strip()]
    return rows, False


def _run_iaa(rows: list[dict], mode: str) -> None:
    """사람 상한(human ceiling) 측정.

    왜 필요한가:
      사람 둘이 같은 글을 요약해도 표현 선택이 달라 ROUGE 는 100 이 안 나온다.
      그 상한이 목표(65)보다 낮으면 **어떤 시스템도 목표를 달성할 수 없다.**
      정답셋 300건을 만들기 전에 파일럿 20건으로 이 값을 먼저 재고, 낮으면
      참조 요약 규격(길이·원문 표현 활용도)을 조정해야 한다.

    한 작성자를 system, 다른 작성자를 reference 로 놓고 양방향으로 채점한다.
    """
    pairs: list[tuple[str, list[str]]] = []
    for row in rows:
        refs = row.get("references") or []
        if isinstance(refs, str) or len(refs) < 2:
            continue
        clean = [r for r in refs if r and r.strip()]
        # 양방향 — 누가 system 이냐에 따라 값이 달라지므로 둘 다 넣고 평균한다.
        for i, sys_text in enumerate(clean):
            others = [r for j, r in enumerate(clean) if j != i]
            pairs.append((sys_text, others))

    if not pairs:
        print("참조가 2개 이상인 행이 없습니다. `references` 에 작성자별 요약을 "
              "2개 이상 넣으세요.", file=sys.stderr)
        sys.exit(1)

    scores = evaluate_pairs(pairs, mode=mode)
    print(f"\n=== 사람 상한(IAA) — {len(pairs)}개 조합 / 토큰={mode} ===")
    print(f"{'metric':<10}{'precision':>12}{'recall':>12}{'f1':>12}")
    for metric in ("rouge1", "rouge2", "rougeL"):
        s = scores[metric]
        print(f"{metric:<10}{s['precision']:>12.4f}{s['recall']:>12.4f}{s['f1']:>12.4f}")

    ceiling = scores["rouge1"]["recall"]
    target = 0.65
    print(f"\n사람 상한 ROUGE-1 recall = {ceiling:.4f} / 목표 {target}")
    if ceiling < target:
        print(
            "⚠️ 상한이 목표보다 낮습니다. 이 규격으로는 어떤 시스템도 목표를 "
            "달성할 수 없습니다. 참조 요약을 더 길게 하거나 원문 표현을 더 살리는 "
            "방향으로 규격을 조정한 뒤 파일럿을 다시 도세요."
        )
    elif ceiling < target * 1.15:
        print(
            f"⚠️ 여유가 {(ceiling - target) * 100:.1f}%p 뿐입니다. 시스템이 사람 "
            "수준에 근접해야 달성됩니다. 규격을 조금 더 완화하는 편이 안전합니다."
        )
    else:
        print("여유가 충분합니다. 이 규격으로 본 구축을 진행해도 됩니다.")


def main() -> None:
    ap = argparse.ArgumentParser(description="요약 ROUGE 평가 (성능지표 No.7)")
    ap.add_argument("dataset", nargs="?", default=None, help="JSONL 정답셋 (없으면 dry-run)")
    ap.add_argument("--mode", choices=["lemma", "char"], default="lemma", help="토큰화 방식")
    ap.add_argument("--ratio", type=float, default=0.3, help="자체 요약 길이 비율")
    ap.add_argument("--iaa", action="store_true",
                    help="사람 상한 측정 — 참조 2개 이상인 행에서 작성자끼리 채점")
    args = ap.parse_args()

    rows, is_dry = _load(args.dataset)

    if args.iaa:
        return _run_iaa(rows, args.mode)

    summarizer = get_summarizer()

    pairs: list[tuple[str, list[str]]] = []
    multi_ref_rows = 0
    for row in rows:
        # references(복수) 우선, 없으면 reference(단수). 둘 다 같은 경로로 처리된다.
        raw = row.get("references") or row.get("reference", "")
        reference = [raw] if isinstance(raw, str) else [r for r in raw if r and r.strip()]
        if not reference:
            continue
        if len(reference) > 1:
            multi_ref_rows += 1
        if "system" in row and row["system"]:
            system = row["system"]
        else:
            system = summarizer.summarize(row.get("text", ""), ratio=args.ratio).final
        pairs.append((system, reference))

    if not pairs:
        print("평가할 (system, reference) 페어가 없습니다.", file=sys.stderr)
        sys.exit(1)

    scores = evaluate_pairs(pairs, mode=args.mode)

    banner = "DRY-RUN (내장 샘플)" if is_dry else f"{args.dataset} ({len(pairs)}건)"
    print(f"\n=== 요약 ROUGE 평가 — {banner} / 토큰={args.mode} ===")
    print(f"{'metric':<10}{'precision':>12}{'recall':>12}{'f1':>12}")
    for metric in ("rouge1", "rouge2", "rougeL"):
        s = scores[metric]
        print(f"{metric:<10}{s['precision']:>12.4f}{s['recall']:>12.4f}{s['f1']:>12.4f}")

    # 계획서 p.24 수식의 분모가 '참조 n-gram 수'이므로 목표와 대조하는 값은 recall 이다.
    # F1 로 대조하면 우리에게 불리한 자체 기준으로 채점하게 된다.
    target = 0.65
    r1 = scores["rouge1"]["recall"]
    status = "달성" if r1 >= target else "미달"
    print(f"\n목표(No.7) ROUGE 0.65 대비 ROUGE-1 recall = {r1:.4f} → {status}")
    print(f"  (참고: ROUGE-1 F1 = {scores['rouge1']['f1']:.4f} — 계획서 지표 아님)")
    if multi_ref_rows:
        print(f"  다중 참조 사용: {multi_ref_rows}/{len(pairs)}건")
    if is_dry:
        print("\n※ 이는 파이프라인 검증용 dry-run 수치입니다. 컴북스 요약 정답셋 수령 후 본 평가로 정식 측정.")


if __name__ == "__main__":
    main()