hbyang
5669c73f10
성적서 '4. 모델 결과' 는 plagiarism_detection_performance 아래에 test_dataset / confusion_matrix / performance_metrics / threshold / interpretation 을 담은 JSON 을 캡처해 싣는다. 그 형식으로 출력한다. threshold 는 전 차수처럼 단일 수치가 아니라 세 조건(결합유사도 0.65 / 연속일치 35자 / 커버리지 0.30)의 조합이므로 객체로 둔다. 유사도만으로 판정하지 않는다는 점이 판정 기준의 핵심이라 수치 하나로 줄이면 오해를 준다. interpretation 문구는 실제 confusion matrix 에서 계산한다. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
220 lines
9.0 KiB
Python
220 lines
9.0 KiB
Python
"""성능지표 #4 정밀도 평가를 끝까지 실행한다.
|
|
|
|
build_plagiarism_testset.py 가 만든 index.jsonl 로 시험용 코퍼스와 인덱스를
|
|
세우고, pairs.jsonl 을 질의해 precision 을 계산한다. 운영 코퍼스와 인덱스는
|
|
건드리지 않는다.
|
|
|
|
python scripts/run_precision_eval.py --testset data/eval/testset_20260908
|
|
|
|
시험관 입회 시 이 명령 한 줄로 재현된다.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from app.core.config import get_settings # noqa: E402
|
|
from app.engine.detector import PlagiarismDetector # noqa: E402
|
|
from app.engine.provenance import CorpusStore, DocumentRecord, SegmentRecord # noqa: E402
|
|
|
|
|
|
def build_corpus(rows: list[dict], db_path: Path) -> None:
|
|
"""시험용 참조 코퍼스를 만든다. 작성자 하나를 문서 하나로 둔다."""
|
|
if db_path.exists():
|
|
db_path.unlink()
|
|
db_path.parent.mkdir(parents=True, exist_ok=True)
|
|
store = CorpusStore(db_path)
|
|
store.initialize()
|
|
|
|
by_author: dict[str, list[dict]] = defaultdict(list)
|
|
for row in rows:
|
|
by_author[row["author"]].append(row)
|
|
|
|
documents, segments = [], []
|
|
for index, (author, items) in enumerate(sorted(by_author.items()), start=1):
|
|
document_id = "testdoc:%04d" % index
|
|
documents.append(DocumentRecord(
|
|
document_id=document_id,
|
|
title="시험 참조 %04d" % index,
|
|
metadata={"provenance": "precision_testset", "author_group": author,
|
|
"human_verified": True, "ai_assistance": False},
|
|
))
|
|
for ordinal, item in enumerate(items, start=1):
|
|
segments.append(SegmentRecord(
|
|
segment_id=item["segment_id"],
|
|
document_id=document_id,
|
|
text=item["text"],
|
|
ordinal=str(ordinal),
|
|
))
|
|
store.upsert_documents(documents)
|
|
added, skipped = store.add_segments(segments)
|
|
print("시험 코퍼스: 문서 %d개 / 세그먼트 %d개 (중복 %d건 제외)"
|
|
% (len(documents), added, skipped))
|
|
|
|
|
|
def build_scorecard(settings, tp: int, fp: int, tn: int, fn: int,
|
|
precision: float, recall: float, f1: float) -> dict:
|
|
"""성적서 "4. 모델 결과" 게재 형식으로 정리한다.
|
|
|
|
전 차수 성적서(GERI `GERIR.CE.2511-02.009`) 13페이지와 같은 키 구성이라
|
|
이 출력을 그대로 캡처해 붙일 수 있다.
|
|
|
|
threshold 는 단일 수치가 아니라 세 조건의 조합이므로 객체로 둔다.
|
|
유사도만으로 판정하지 않는다는 점이 판정 기준의 핵심이다.
|
|
"""
|
|
predicted_positive = tp + fp
|
|
actual_positive = tp + fn
|
|
accuracy = (tp + tn) / (tp + fp + tn + fn) if (tp + fp + tn + fn) else 0.0
|
|
fpr = fp / (fp + tn) if (fp + tn) else 0.0
|
|
return {
|
|
"plagiarism_detection_performance": {
|
|
"model": "O2O Triple-Similarity Detector",
|
|
"engine_version": settings.engine_version,
|
|
"test_dataset": {
|
|
"total_samples": tp + fp + tn + fn,
|
|
"plagiarism_cases": actual_positive,
|
|
"non_plagiarism_cases": fp + tn,
|
|
},
|
|
"confusion_matrix": {
|
|
"true_positive": tp,
|
|
"false_positive": fp,
|
|
"true_negative": tn,
|
|
"false_negative": fn,
|
|
},
|
|
"performance_metrics": {
|
|
"precision": round(precision, 4),
|
|
"recall": round(recall, 4),
|
|
"f1_score": round(f1, 4),
|
|
"accuracy": round(accuracy, 4),
|
|
},
|
|
"threshold": {
|
|
"combined_similarity": settings.persistent_similarity_threshold,
|
|
"min_exact_span_chars": settings.persistent_min_exact_span,
|
|
"min_coverage": settings.persistent_min_coverage,
|
|
"require_exact_span_evidence": settings.require_exact_span_evidence,
|
|
},
|
|
"interpretation": {
|
|
"precision": "모델이 표절로 판단한 %d건 중 %d건(%.1f%%)이 실제 표절"
|
|
% (predicted_positive, tp, precision * 100),
|
|
"recall": "실제 표절 %d건 중 %d건(%.1f%%)을 정확히 탐지"
|
|
% (actual_positive, tp, recall * 100),
|
|
"false_positive_rate": "%.1f%%" % (fpr * 100),
|
|
},
|
|
}
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--testset", type=Path, required=True)
|
|
parser.add_argument("--work-dir", type=Path,
|
|
help="시험 코퍼스·인덱스를 둘 경로 (기본: <testset>/runtime)")
|
|
parser.add_argument("--skip-build", action="store_true",
|
|
help="이미 만든 코퍼스·인덱스를 재사용")
|
|
args = parser.parse_args()
|
|
|
|
work = args.work_dir or (args.testset / "runtime")
|
|
db_path = work / "corpus.sqlite3"
|
|
index_dir = work / "index"
|
|
|
|
index_rows = [json.loads(l) for l in
|
|
(args.testset / "index.jsonl").read_text(encoding="utf-8").splitlines() if l.strip()]
|
|
pairs = [json.loads(l) for l in
|
|
(args.testset / "pairs.jsonl").read_text(encoding="utf-8").splitlines() if l.strip()]
|
|
|
|
if not args.skip_build:
|
|
build_corpus(index_rows, db_path)
|
|
|
|
get_settings.cache_clear()
|
|
settings = get_settings().model_copy(update={
|
|
"corpus_db_path": str(db_path),
|
|
"persistent_index_dir": str(index_dir),
|
|
"use_persistent_index": True,
|
|
"use_llm_legal_judge": False,
|
|
})
|
|
|
|
if not args.skip_build:
|
|
from app.engine.persistent_index import PersistentCorpusIndex
|
|
index_dir.mkdir(parents=True, exist_ok=True)
|
|
stats = PersistentCorpusIndex(db_path, index_dir).sync()
|
|
print("시험 인덱스 구축 완료: %s (%s)" % (index_dir, stats))
|
|
|
|
detector = PlagiarismDetector(settings)
|
|
print("판정 기준: 유사도>=%.2f | 연속일치>=%d자 | 커버리지>=%.2f | 연속일치 필수=%s"
|
|
% (settings.persistent_similarity_threshold,
|
|
settings.persistent_min_exact_span,
|
|
settings.persistent_min_coverage,
|
|
settings.require_exact_span_evidence))
|
|
|
|
tp = fp = tn = fn = 0
|
|
buckets: dict[str, dict[str, int]] = defaultdict(lambda: {"tp": 0, "fp": 0, "tn": 0, "fn": 0})
|
|
for i, row in enumerate(pairs, start=1):
|
|
result = detector.detect(doc_id=row["pair_id"], text=row["derived_text"])
|
|
predicted = bool(result.is_infringement)
|
|
expected = bool(row["is_plagiarism"])
|
|
key = "tp" if (expected and predicted) else \
|
|
"fn" if expected else \
|
|
"fp" if predicted else "tn"
|
|
buckets[row.get("transformation", "unknown")][key] += 1
|
|
if key == "tp":
|
|
tp += 1
|
|
elif key == "fn":
|
|
fn += 1
|
|
elif key == "fp":
|
|
fp += 1
|
|
else:
|
|
tn += 1
|
|
if i % 100 == 0:
|
|
print(" 진행 %d/%d" % (i, len(pairs)))
|
|
|
|
precision = tp / (tp + fp) if (tp + fp) else 0.0
|
|
recall = tp / (tp + fn) if (tp + fn) else 0.0
|
|
f1 = 2 * precision * recall / (precision + recall) if (precision + recall) else 0.0
|
|
|
|
print("\n" + "=" * 62)
|
|
print("표절 여부 판별 정밀도 (precision) : %.4f [목표 0.97]" % precision)
|
|
print("재현율 (recall) : %.4f" % recall)
|
|
print("F1 : %.4f" % f1)
|
|
print("TP=%d FP=%d TN=%d FN=%d" % (tp, fp, tn, fn))
|
|
print("\n[변형 유형별]")
|
|
for name in sorted(buckets):
|
|
b = buckets[name]
|
|
total = sum(b.values())
|
|
p = b["tp"] / (b["tp"] + b["fp"]) if (b["tp"] + b["fp"]) else 0.0
|
|
r = b["tp"] / (b["tp"] + b["fn"]) if (b["tp"] + b["fn"]) else 0.0
|
|
print(" %-18s n=%4d P=%.3f R=%.3f TP=%d FP=%d TN=%d FN=%d"
|
|
% (name, total, p, r, b["tp"], b["fp"], b["tn"], b["fn"]))
|
|
scorecard = build_scorecard(settings, tp, fp, tn, fn, precision, recall, f1)
|
|
rendered = json.dumps(scorecard, ensure_ascii=False, indent=2)
|
|
print("\n" + "=" * 62)
|
|
print("4. 모델 결과 :")
|
|
print()
|
|
print(rendered)
|
|
print("\n최종 결과 : precision %.2f%% 달성" % (precision * 100))
|
|
(args.testset / "scorecard.json").write_text(rendered + "\n", encoding="utf-8")
|
|
|
|
result_path = args.testset / "result.json"
|
|
result_path.write_text(json.dumps({
|
|
"precision": precision, "recall": recall, "f1": f1,
|
|
"tp": tp, "fp": fp, "tn": tn, "fn": fn,
|
|
"by_transformation": dict(buckets),
|
|
"criteria": {
|
|
"similarity": settings.persistent_similarity_threshold,
|
|
"min_exact_span": settings.persistent_min_exact_span,
|
|
"min_coverage": settings.persistent_min_coverage,
|
|
"require_exact_span_evidence": settings.require_exact_span_evidence,
|
|
},
|
|
}, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
print("\n%s 에 결과 기록" % result_path)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|