"""AI 생성 판별기 CPU 학습 CLI (sklearn, 외부 API·GPU 불필요). 입력: build_ai_training_dataset.py 산출 JSONL {"text","label","source_group","split", ...} 출력: joblib 아티팩트 + metrics JSON 모델: --model logreg : StandardScaler + LogisticRegression (기본, 설명 가능) --model hgb : HistGradientBoostingClassifier (비선형, 설명력 낮음) 두 경우 모두 CalibratedClassifierCV 로 확률 보정한다. 보정하지 않은 점수를 '의심도 %'로 UI 에 띄우면 검토자가 값을 과신하게 된다. 평가: AUROC / AUPRC / confusion matrix / FPR·TPR / precision·recall + **목표 FPR 기준 임계값**. 자서전 도메인에서는 인간 저작을 AI로 오판하는 비용이 압도적으로 크므로, F1 최적점이 아니라 FPR 상한으로 컷을 잡는다. 실패 조건 (조용히 넘어가지 않고 종료 코드 2): · label 필드 없음 · 단일 클래스만 존재 · 그룹 수가 CV fold 수보다 적음 · 학습/평가 표본 부족 사용: python scripts/train_ai_detector.py \ --data data/training/ai_dataset.jsonl \ --model logreg --target-fpr 0.05 \ --out data/models/ai_detector.joblib """ from __future__ import annotations import argparse import json import logging import sys from collections import Counter from pathlib import Path ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT)) logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s") logger = logging.getLogger("train-ai-detector") from app.engine.ai_detector import ( # noqa: E402 FEATURE_NAMES, FEATURE_SET_VERSION, LENGTH_FEATURES, extract_features, features_to_vector, ) # --------------------------------------------------------------------------- # 데이터 # --------------------------------------------------------------------------- def load_dataset(path: Path) -> list[dict]: rows: list[dict] = [] with path.open(encoding="utf-8") as fh: for i, line in enumerate(fh, start=1): line = line.strip() if not line: continue try: rows.append(json.loads(line)) except json.JSONDecodeError: logger.warning("%s:%d JSON 파싱 실패 — 건너뜀", path.name, i) return rows def validate(rows: list[dict]) -> tuple[list[dict], str | None]: """학습 가능 여부 검증. 문제가 있으면 (rows, 사유) 를 돌려준다.""" if not rows: return rows, "데이터가 0건입니다." missing = [i for i, r in enumerate(rows) if "label" not in r] if missing: return rows, f"label 필드가 없는 행 {len(missing)}건 (예: index {missing[:3]})" usable = [] for r in rows: text = (r.get("text") or "").strip() if not text: continue try: label = int(r["label"]) except (TypeError, ValueError): return rows, f"label 을 정수로 해석할 수 없습니다: {r['label']!r}" if label not in (0, 1): return rows, f"label 은 0/1 이어야 합니다: {label}" r["label"] = label usable.append(r) if not usable: return usable, "본문이 있는 행이 없습니다." counts = Counter(r["label"] for r in usable) if len(counts) < 2: return usable, ( f"단일 클래스만 존재합니다({dict(counts)}). AI 샘플과 human 샘플이 " "모두 필요합니다." ) if min(counts.values()) < 10: return usable, f"소수 클래스 표본이 너무 적습니다: {dict(counts)} (최소 10건)" groups = {r.get("source_group") or "" for r in usable} if len(groups) < 4: return usable, ( f"source_group 이 {len(groups)}개뿐입니다. group 분할 검증이 " "불가능하며 성능이 부풀려집니다. 최소 4개 필요." ) return usable, None def featurize( rows: list[dict], use_pos: bool, zeroed: tuple[str, ...] = () ) -> tuple[list[list[float]], list[int], list[str]]: X: list[list[float]] = [] y: list[int] = [] groups: list[str] = [] for i, r in enumerate(rows): feats = extract_features(r["text"], use_pos=use_pos) X.append(features_to_vector(feats, zeroed)) y.append(int(r["label"])) groups.append(str(r.get("source_group") or f"row:{i}")) if (i + 1) % 500 == 0: logger.info("특징 추출 %d/%d", i + 1, len(rows)) return X, y, groups # --------------------------------------------------------------------------- # 평가 # --------------------------------------------------------------------------- def threshold_at_fpr(y_true, scores, target_fpr: float) -> tuple[float, dict]: """목표 FPR 이하를 만족하는 최소 임계값과 그 지점의 지표.""" from sklearn.metrics import roc_curve fpr, tpr, thr = roc_curve(y_true, scores) chosen = None for f, t, th in zip(fpr, tpr, thr): if f <= target_fpr: chosen = (float(f), float(t), float(th)) if chosen is None: chosen = (float(fpr[0]), float(tpr[0]), float(thr[0])) return chosen[2], {"fpr": chosen[0], "tpr": chosen[1], "threshold": chosen[2]} def evaluate(y_true, scores, threshold: float) -> dict: from sklearn.metrics import ( average_precision_score, confusion_matrix, roc_auc_score, ) preds = [1 if s >= threshold else 0 for s in scores] tn, fp, fn, tp = confusion_matrix(y_true, preds, labels=[0, 1]).ravel() 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 return { "threshold": round(float(threshold), 4), "auroc": round(float(roc_auc_score(y_true, scores)), 4), "auprc": round(float(average_precision_score(y_true, scores)), 4), "confusion_matrix": {"tn": int(tn), "fp": int(fp), "fn": int(fn), "tp": int(tp)}, "fpr": round(fp / (fp + tn), 4) if (fp + tn) else 0.0, "tpr": round(recall, 4), "precision": round(precision, 4), "recall": round(recall, 4), "f1": round(f1, 4), "n": len(y_true), "positives": int(sum(y_true)), } # --------------------------------------------------------------------------- # 학습 # --------------------------------------------------------------------------- def build_estimator(kind: str, seed: int): from sklearn.ensemble import HistGradientBoostingClassifier from sklearn.linear_model import LogisticRegression from sklearn.pipeline import Pipeline from sklearn.preprocessing import StandardScaler if kind == "logreg": return Pipeline([ ("scaler", StandardScaler()), ("clf", LogisticRegression( max_iter=2000, class_weight="balanced", random_state=seed )), ]) if kind == "hgb": return HistGradientBoostingClassifier( max_iter=300, learning_rate=0.06, max_depth=6, random_state=seed ) raise ValueError(f"unknown model: {kind}") def main() -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--data", type=Path, required=True) ap.add_argument("--model", choices=["logreg", "hgb"], default="logreg") ap.add_argument("--out", type=Path, default=Path("data/models/ai_detector.joblib")) ap.add_argument("--metrics-out", type=Path, default=None) ap.add_argument("--target-fpr", type=float, default=0.05, help="이 FPR 이하가 되도록 임계값을 잡는다 (인간 오판 억제)") ap.add_argument("--high-fpr", type=float, default=0.01, help="'높음' 배지용 더 보수적인 FPR") ap.add_argument("--cv-folds", type=int, default=5) ap.add_argument("--seed", type=int, default=20260810) ap.add_argument("--no-pos", action="store_true", help="품사 특징 사용 안 함") ap.add_argument( "--drop-length-features", action="store_true", help="길이 계열 특징을 0으로 눌러 제외. human/AI 표본의 분량 분포가 다르면 " "모델이 문체 대신 길이를 학습하므로, 실데이터에서는 켜고 한 번 " "돌려 성능 차이를 반드시 비교할 것.", ) ap.add_argument("--trained-at", default="", help="아티팩트에 기록할 학습 시각 문자열") args = ap.parse_args() try: import numpy as np import sklearn from sklearn.calibration import CalibratedClassifierCV except ImportError as exc: logger.error("scikit-learn/numpy 가 필요합니다: %s", exc) return 2 if not args.data.exists(): logger.error("데이터 없음: %s", args.data) return 2 rows = load_dataset(args.data) rows, reason = validate(rows) if reason: logger.error("학습 불가: %s", reason) return 2 use_pos = not args.no_pos zeroed: tuple[str, ...] = LENGTH_FEATURES if args.drop_length_features else () logger.info( "레코드 %d건, 품사특징=%s, 제외특징=%s", len(rows), use_pos, list(zeroed) or "없음", ) # split 필드가 있으면 그대로 신뢰 (빌더가 group 단위로 만든 것). # train=모델 적합, val=임계값 선택, test=최종 보고로 역할을 섞지 않는다. has_split = any(r.get("split") for r in rows) if has_split: train_rows = [r for r in rows if r.get("split") == "train"] val_rows = [r for r in rows if r.get("split") == "val"] test_rows = [r for r in rows if r.get("split") == "test"] if not train_rows or not val_rows or not test_rows: logger.error("train/val/test 중 빈 split이 있습니다. 빌더 분할을 확인하세요.") return 2 train_groups = {r.get("source_group") for r in train_rows} val_groups = {r.get("source_group") for r in val_rows} test_groups = {r.get("source_group") for r in test_rows} overlap = (train_groups & val_groups) | (train_groups & test_groups) | (val_groups & test_groups) if overlap: logger.error( "train/val/test 그룹 중복 %d건 — 누출입니다. 중단합니다: %s", len(overlap), list(overlap)[:5], ) return 2 else: from sklearn.model_selection import GroupShuffleSplit all_groups = [str(r.get("source_group") or "") for r in rows] gss = GroupShuffleSplit(n_splits=1, test_size=0.2, random_state=args.seed) fit_idx, te_idx = next(gss.split(rows, [r["label"] for r in rows], all_groups)) fit_rows = [rows[i] for i in fit_idx] fit_groups = [all_groups[i] for i in fit_idx] val_split = GroupShuffleSplit(n_splits=1, test_size=0.2, random_state=args.seed + 1) tr_local, va_local = next(val_split.split( fit_rows, [r["label"] for r in fit_rows], fit_groups )) train_rows = [fit_rows[i] for i in tr_local] val_rows = [fit_rows[i] for i in va_local] test_rows = [rows[i] for i in te_idx] logger.warning("split 필드가 없어 GroupShuffleSplit 으로 train/val/test를 나눴습니다.") logger.info("train=%d / val=%d / test=%d", len(train_rows), len(val_rows), len(test_rows)) for name, subset in (("train", train_rows), ("val", val_rows), ("test", test_rows)): counts = Counter(r["label"] for r in subset) if len(counts) < 2: logger.error("%s 에 단일 클래스만 있습니다: %s", name, dict(counts)) return 2 logger.info("%s 라벨 분포: %s", name, dict(counts)) X_tr, y_tr, g_tr = featurize(train_rows, use_pos, zeroed) X_val, y_val, _ = featurize(val_rows, use_pos, zeroed) X_te, y_te, _ = featurize(test_rows, use_pos, zeroed) n_groups = len(set(g_tr)) folds = min(args.cv_folds, n_groups) if folds < 2: logger.error("train 그룹이 %d개뿐이라 교차검증 보정이 불가합니다.", n_groups) return 2 if folds < args.cv_folds: logger.warning("그룹 수가 적어 CV fold 를 %d → %d 로 줄입니다.", args.cv_folds, folds) base = build_estimator(args.model, args.seed) try: from sklearn.model_selection import StratifiedGroupKFold cv = StratifiedGroupKFold(n_splits=folds, shuffle=True, random_state=args.seed) cv_split = list(cv.split(np.array(X_tr), np.array(y_tr), groups=g_tr)) calibrated = CalibratedClassifierCV(base, method="sigmoid", cv=cv_split) except Exception as exc: logger.warning("StratifiedGroupKFold 사용 불가(%s) — 일반 CV 로 대체", exc) calibrated = CalibratedClassifierCV(base, method="sigmoid", cv=folds) logger.info("학습 시작 (model=%s, folds=%d)", args.model, folds) calibrated.fit(np.array(X_tr), np.array(y_tr)) scores_te = calibrated.predict_proba(np.array(X_te))[:, 1] scores_tr = calibrated.predict_proba(np.array(X_tr))[:, 1] scores_val = calibrated.predict_proba(np.array(X_val))[:, 1] # 임계값은 모델이 보지 않은 val에서 잡고 test는 최종 보고에만 사용한다. low_cut, low_info = threshold_at_fpr(y_val, scores_val, args.target_fpr) high_cut, high_info = threshold_at_fpr(y_val, scores_val, args.high_fpr) val_negatives = y_val.count(0) if val_negatives < 100: logger.warning( "val human 표본이 %d건뿐이라 FPR %.3f 컷 추정이 불안정합니다(최소 100, 권장 300).", val_negatives, args.high_fpr, ) if high_cut < low_cut: high_cut = low_cut if high_cut <= low_cut: logger.warning( "low_cut 과 high_cut 이 같습니다(%.4f). 'medium' 배지가 사라져 " "모든 결과가 low/high 로만 갈립니다. 표본이 너무 쉽게 분리되거나 " "표본 수가 부족하다는 신호이니, 실데이터에서 이 경고가 뜨면 " "--target-fpr/--high-fpr 을 벌리고 데이터를 재점검하세요.", low_cut, ) metrics = { "model": args.model, "feature_set_version": FEATURE_SET_VERSION, "use_pos": use_pos, "zeroed_features": list(zeroed), "n_train": len(y_tr), "n_val": len(y_val), "n_test": len(y_te), "n_train_groups": n_groups, "cv_folds": folds, "cuts": { "low_cut": round(float(low_cut), 4), "high_cut": round(float(high_cut), 4), "target_fpr": args.target_fpr, "high_fpr": args.high_fpr, "validation_low_point": low_info, "validation_high_point": high_info, }, "test": evaluate(y_te, scores_te, low_cut), "test_at_high_cut": evaluate(y_te, scores_te, high_cut), "validation": evaluate(y_val, scores_val, low_cut), "train": evaluate(y_tr, scores_tr, low_cut), } logger.info("test AUROC=%.4f AUPRC=%.4f FPR=%.4f TPR=%.4f", metrics["test"]["auroc"], metrics["test"]["auprc"], metrics["test"]["fpr"], metrics["test"]["tpr"]) logger.info("confusion(test): %s", metrics["test"]["confusion_matrix"]) suffix = "-nolen" if zeroed else "" model_version = ( f"{args.model}{suffix}-{FEATURE_SET_VERSION}" f"-auroc{metrics['test']['auroc']:.3f}" ) payload = { "estimator": calibrated, "feature_names": tuple(FEATURE_NAMES), "feature_set_version": FEATURE_SET_VERSION, "model_version": model_version, "requires_pos": use_pos, "low_cut": float(low_cut), "high_cut": float(high_cut), "metrics": metrics, "zeroed_features": tuple(zeroed), "trained_at": args.trained_at, "sklearn_version": sklearn.__version__, "notes": ( "AI 생성 '의심도' 모델. 확정 판정이 아니며 사람 검토 보조용. " "임계값은 target FPR 기준으로 산출됨." ), } import joblib args.out.parent.mkdir(parents=True, exist_ok=True) joblib.dump(payload, args.out) metrics_path = args.metrics_out or args.out.with_suffix(".metrics.json") metrics_path.write_text( json.dumps(metrics, ensure_ascii=False, indent=2), encoding="utf-8" ) logger.info("아티팩트 저장: %s (version=%s)", args.out, model_version) logger.info("지표 저장: %s", metrics_path) if metrics["test"]["auroc"] < 0.7: logger.warning( "AUROC %.3f — 실용 수준이 아닙니다. 이 모델을 운영에 붙이지 마세요.", metrics["test"]["auroc"], ) return 0 if __name__ == "__main__": raise SystemExit(main())