418 lines
17 KiB
Python
418 lines
17 KiB
Python
"""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())
|