hbyang
a530d2139f
AI 생성 표본이 없어 정확도를 측정할 수 없는 상태에서, 규칙 기반 휴리스틱을 실제로 쓸 수 있게 만든다. 점수의 의미를 "AI일 확률"에서 "등록 코퍼스의 인간 저작물 대비 문체 이례도"로 재정의해, 라벨 없이도 참인 진술이 되도록 했다. 덤으로 high 배지 비율이 정의상 고정되어 검토 부하가 예측 가능해진다. - ai_detector: low_cut/high_cut 주입 지원. 둘 다 주어질 때만 적용하고, 적용되면 model_version 에 +corpus-percentile 을 붙여 컷 출처를 드러낸다. is_stub 은 여전히 true — 컷을 맞췄을 뿐 학습된 모델이 아니다. - calibrate_cuts/percentile/score_text_heuristic 순수 함수 추가. - scripts/calibrate_ai_detector_cuts.py: 코퍼스 표본의 점수 분포에서 백분위 컷을 산출하고 .env 두 줄을 출력한다. - 규칙 점수가 상단에서 clip 되어 p90=p98 이 되면 그 컷은 high 배지를 영원히 0건으로 만든다. 이 경우 .env 를 출력하지 않고 exit 3 으로 중단하며 saturated_share/distinct_scores 진단을 남긴다. 유효 표본 100건 미만은 exit 2. - kiwipiepy 유무가 점수를 바꾸므로 실제 사용 여부를 리포트에 기록하고 경고한다. 기본값은 그대로 비활성(AI_DETECTOR_ALLOW_HEURISTIC=false)이라 켜지 않으면 동작이 바뀌지 않는다. 컷 미설정 시에는 자리표시자임을 note 로 경고한다. 테스트 180건 통과 (신규 13건). Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1156 lines
43 KiB
Python
1156 lines
43 KiB
Python
"""한국어 AI 생성 의심도 — KatFishNet 계열 언어특징 기반 분류기.
|
||
|
||
⚠️ 본 모듈이 산출하는 값은 **AI 생성 여부의 확정 판정이 아니라, 사람 검토를
|
||
우선 배정하기 위한 "의심도"** 다. 자서전 도메인은 대필·윤문·편집자 개입이
|
||
일상적이라 정제된 문체가 AI로 오판되기 쉽고, 학습에 쓰이지 않은 생성 모델에는
|
||
일반화가 잘 되지 않으며, 가벼운 수정만으로도 회피된다. 저자에게 통보되는
|
||
판정 근거로 쓰지 말 것. 자세한 한계는 docs/AI_DETECTION.md 참조.
|
||
|
||
설계 원칙:
|
||
1) **결정적** — 같은 입력이면 항상 같은 출력. 난수·해시 더미 없음.
|
||
2) **설명 가능** — 점수의 근거가 되는 언어특징을 그대로 노출한다. 검토자가
|
||
"왜 의심되는가"를 확인할 수 없으면 이 기능은 쓸 수 없다.
|
||
3) **정직한 미가용** — 학습된 모델이 없으면 점수를 지어내지 않는다.
|
||
available=False 로 명시하거나, 미검증 휴리스틱임을 is_stub/model_version/
|
||
note 로 드러낸다.
|
||
|
||
특징 계열 (KatFishNet 계열 한국어 표지):
|
||
· 띄어쓰기 — 어절 길이 분포, 공백 비율
|
||
· 문장/쉼표 — 문장 길이 변동계수(burstiness), 쉼표 밀도, 종결어미 분포
|
||
· 품사 — 품사 다양도, 품사 n-gram 엔트로피/반복, 조사·어미 다양도
|
||
· 반복 — 문자 3-gram·어절 bigram 반복률, hapax 비율
|
||
· 길이 — 문자/어절/형태소 수
|
||
|
||
품사 특징은 kiwipiepy 가 있을 때만 산출된다. 미설치/오류 시에도 **특징 벡터의
|
||
길이와 순서는 변하지 않으며**(0.0 채움 + pos_available=0.0), 모델 아티팩트의
|
||
requires_pos 메타와 대조해 경고를 남긴다.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import math
|
||
import os
|
||
import re
|
||
import unicodedata
|
||
from collections import Counter
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
from typing import Literal, Sequence
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 상수
|
||
# ---------------------------------------------------------------------------
|
||
|
||
#: 이 길이 미만이면 점수를 내지 않는다. 짧은 글은 언어특징 통계가 무의미하다.
|
||
HARD_MIN_CHARS = 120
|
||
|
||
#: 이 길이 미만이면 점수는 내되 신뢰도 경고를 붙인다.
|
||
MIN_RELIABLE_CHARS = 300
|
||
|
||
#: 세그먼트(구간) 채점 시 목표 길이. 문단을 이 길이 이상으로 병합해 채점한다.
|
||
SEGMENT_TARGET_CHARS = 600
|
||
|
||
#: 기본 모델 경로. config.py 는 담당 범위 밖이라 환경변수로 읽는다.
|
||
#: 통합 시 Settings.ai_detector_model_path 로 승격할 것 (docs/AI_DETECTION.md).
|
||
ENV_MODEL_PATH = "AI_DETECTOR_MODEL_PATH"
|
||
DEFAULT_MODEL_PATH = "./data/models/ai_detector.joblib"
|
||
|
||
#: 잠정 의심도 구간. 학습 시 target FPR 기준으로 재산출되어 아티팩트에 저장되며,
|
||
#: 아티팩트 값이 있으면 그쪽이 우선한다. 아래는 아무 근거 없는 자리표시자이므로,
|
||
#: 휴리스틱 모드로 운영할 때는 scripts/calibrate_ai_detector_cuts.py 로 산출한
|
||
#: 백분위 컷을 설정으로 주입할 것.
|
||
DEFAULT_LOW_CUT = 0.40
|
||
DEFAULT_HIGH_CUT = 0.70
|
||
|
||
#: 백분위 캘리브레이션 기본값. 등록 코퍼스(전부 인간 저작) 대비 상위 10%를
|
||
#: medium, 상위 2%를 high 로 본다. "AI 확률"이 아니라 "인간 저작 대비 이례도"다.
|
||
DEFAULT_LOW_PERCENTILE = 90.0
|
||
DEFAULT_HIGH_PERCENTILE = 98.0
|
||
|
||
Provenance = Literal["human", "ai", "mixed", "edited", "unknown"]
|
||
SuspicionLevel = Literal["low", "medium", "high"]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 특징 정의 — 순서가 곧 모델 입력 벡터의 순서다. 절대 중간 삽입/삭제 금지.
|
||
# 변경이 필요하면 FEATURE_SET_VERSION 을 올리고 모델을 재학습할 것.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
FEATURE_SET_VERSION = "kf-ko-v1"
|
||
|
||
_BASE_FEATURES: tuple[str, ...] = (
|
||
# 길이
|
||
"char_count",
|
||
"eojeol_count",
|
||
"sentence_count",
|
||
"paragraph_count",
|
||
# 띄어쓰기
|
||
"space_ratio",
|
||
"mean_eojeol_len",
|
||
"std_eojeol_len",
|
||
"cv_eojeol_len",
|
||
"long_eojeol_ratio",
|
||
"short_eojeol_ratio",
|
||
# 문장
|
||
"mean_sentence_len",
|
||
"std_sentence_len",
|
||
"cv_sentence_len",
|
||
"max_sentence_len",
|
||
"min_sentence_len",
|
||
"mean_paragraph_len",
|
||
# 쉼표·문장부호
|
||
"comma_per_sentence",
|
||
"comma_ratio",
|
||
"punct_ratio",
|
||
"punct_diversity",
|
||
"ellipsis_ratio",
|
||
"quote_ratio",
|
||
"exclaim_question_ratio",
|
||
# 종결
|
||
"end_da_ratio",
|
||
"end_yo_ratio",
|
||
"end_noun_ratio",
|
||
# 문자 구성
|
||
"hangul_ratio",
|
||
"digit_ratio",
|
||
"latin_ratio",
|
||
"newline_ratio",
|
||
# 반복·다양도
|
||
"eojeol_ttr",
|
||
"hapax_ratio",
|
||
"top1_token_share",
|
||
"top5_token_share",
|
||
"char3gram_repeat_ratio",
|
||
"word_bigram_repeat_ratio",
|
||
"distinct_word_bigram_ratio",
|
||
)
|
||
|
||
_POS_FEATURES: tuple[str, ...] = (
|
||
"pos_available",
|
||
"morph_count",
|
||
"mean_morphs_per_eojeol",
|
||
"content_word_ratio",
|
||
"pos_ttr",
|
||
"pos_bigram_entropy",
|
||
"pos_trigram_repeat_ratio",
|
||
"josa_ratio",
|
||
"josa_diversity",
|
||
"eomi_ratio",
|
||
"eomi_diversity",
|
||
"noun_ratio",
|
||
"verb_ratio",
|
||
"adj_ratio",
|
||
"adverb_ratio",
|
||
"conj_adverb_ratio",
|
||
"dependent_noun_ratio",
|
||
)
|
||
|
||
FEATURE_NAMES: tuple[str, ...] = _BASE_FEATURES + _POS_FEATURES
|
||
|
||
#: 길이 계열 특징. 문체가 아니라 '분량'을 학습할 위험이 있어 학습 시 제외할 수
|
||
#: 있다(train_ai_detector.py --drop-length-features). human 에피소드와 AI 생성물의
|
||
#: 길이 분포가 다르면, 모델이 문체 대신 길이만 보고 맞히는 착시가 생긴다.
|
||
LENGTH_FEATURES: tuple[str, ...] = (
|
||
"char_count",
|
||
"eojeol_count",
|
||
"sentence_count",
|
||
"paragraph_count",
|
||
"morph_count",
|
||
"max_sentence_len",
|
||
"min_sentence_len",
|
||
"mean_paragraph_len",
|
||
)
|
||
|
||
#: 사람이 읽을 수 있는 특징 설명 (검토 콘솔·문서용).
|
||
FEATURE_LABELS_KO: dict[str, str] = {
|
||
"cv_sentence_len": "문장 길이 변동계수(낮을수록 균일 = AI 경향)",
|
||
"cv_eojeol_len": "어절 길이 변동계수",
|
||
"comma_per_sentence": "문장당 쉼표 수",
|
||
"eojeol_ttr": "어절 다양도(TTR)",
|
||
"hapax_ratio": "1회 등장 어절 비율",
|
||
"pos_bigram_entropy": "품사 bigram 엔트로피(낮을수록 정형적)",
|
||
"pos_trigram_repeat_ratio": "품사 trigram 반복률",
|
||
"eomi_diversity": "어미 다양도",
|
||
"josa_diversity": "조사 다양도",
|
||
"char3gram_repeat_ratio": "문자 3-gram 반복률",
|
||
"word_bigram_repeat_ratio": "어절 bigram 반복률",
|
||
"end_da_ratio": "'-다' 종결 비율",
|
||
"end_yo_ratio": "'-요/-습니다' 종결 비율",
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 형태소 분석기 (선택적)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_kiwi_instance = None
|
||
_kiwi_failed = False
|
||
|
||
|
||
def _get_kiwi():
|
||
"""kiwipiepy 인스턴스. 미설치/초기화 실패 시 None 을 돌려주고 폴백한다."""
|
||
global _kiwi_instance, _kiwi_failed
|
||
if _kiwi_instance is not None:
|
||
return _kiwi_instance
|
||
if _kiwi_failed:
|
||
return None
|
||
try:
|
||
from kiwipiepy import Kiwi
|
||
|
||
_kiwi_instance = Kiwi()
|
||
return _kiwi_instance
|
||
except Exception as exc: # 미설치, 모델 파일 손상, 메모리 부족 등 전부 포함
|
||
logger.warning("kiwipiepy unavailable — POS features disabled: %s", exc)
|
||
_kiwi_failed = True
|
||
return None
|
||
|
||
|
||
def reset_kiwi_cache() -> None:
|
||
"""테스트용 — 형태소 분석기 캐시 초기화."""
|
||
global _kiwi_instance, _kiwi_failed
|
||
_kiwi_instance = None
|
||
_kiwi_failed = False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 텍스트 분해
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_SENT_SPLIT = re.compile(r"(?<=[.!?。…])\s+|\n{1,}")
|
||
_PARA_SPLIT = re.compile(r"\n\s*\n")
|
||
_HANGUL = re.compile(r"[가-힣]")
|
||
_LATIN = re.compile(r"[A-Za-z]")
|
||
_DIGIT = re.compile(r"[0-9]")
|
||
_PUNCT = re.compile(r"[.,!?;:…·\"'“”‘’()\[\]{}—\-~/]")
|
||
|
||
|
||
def normalize_text(text: str) -> str:
|
||
"""비교 가능한 형태로 정규화. 원문 손실을 최소화한다(공백만 정돈)."""
|
||
if not text:
|
||
return ""
|
||
t = unicodedata.normalize("NFKC", text)
|
||
t = t.replace("\r\n", "\n").replace("\r", "\n")
|
||
t = re.sub(r"[ \t ]+", " ", t)
|
||
t = re.sub(r"\n{3,}", "\n\n", t)
|
||
return t.strip()
|
||
|
||
|
||
def split_sentences(text: str) -> list[str]:
|
||
return [s.strip() for s in _SENT_SPLIT.split(text) if s and s.strip()]
|
||
|
||
|
||
def split_paragraphs(text: str) -> list[str]:
|
||
parts = [p.strip() for p in _PARA_SPLIT.split(text) if p and p.strip()]
|
||
if parts:
|
||
return parts
|
||
return [text.strip()] if text.strip() else []
|
||
|
||
|
||
def _mean(xs: Sequence[float]) -> float:
|
||
return sum(xs) / len(xs) if xs else 0.0
|
||
|
||
|
||
def _std(xs: Sequence[float]) -> float:
|
||
if len(xs) < 2:
|
||
return 0.0
|
||
m = _mean(xs)
|
||
return math.sqrt(sum((x - m) ** 2 for x in xs) / len(xs))
|
||
|
||
|
||
def _safe_div(a: float, b: float) -> float:
|
||
return a / b if b else 0.0
|
||
|
||
|
||
def _entropy(counts: Sequence[int]) -> float:
|
||
total = sum(counts)
|
||
if total <= 0:
|
||
return 0.0
|
||
h = 0.0
|
||
for c in counts:
|
||
if c <= 0:
|
||
continue
|
||
p = c / total
|
||
h -= p * math.log(p, 2)
|
||
return h
|
||
|
||
|
||
def _repeat_ratio(items: Sequence) -> float:
|
||
"""전체 중 '두 번 이상 등장한 항목이 차지하는 비율'."""
|
||
if not items:
|
||
return 0.0
|
||
cnt = Counter(items)
|
||
repeated = sum(c for c in cnt.values() if c > 1)
|
||
return repeated / len(items)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 특징 추출
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def extract_features(text: str, use_pos: bool = True) -> dict[str, float]:
|
||
"""raw 한국어 텍스트 → 결정적 언어특징 사전.
|
||
|
||
반환 키는 항상 FEATURE_NAMES 전체를 포함한다(값이 0.0 일지언정 누락 없음).
|
||
kiwipiepy 가 없으면 품사 특징은 0.0, pos_available=0.0 이 된다.
|
||
"""
|
||
feats: dict[str, float] = {name: 0.0 for name in FEATURE_NAMES}
|
||
|
||
norm = normalize_text(text)
|
||
if not norm:
|
||
return feats
|
||
|
||
chars = len(norm)
|
||
eojeols = norm.split()
|
||
sentences = split_sentences(norm)
|
||
paragraphs = split_paragraphs(norm)
|
||
eojeol_lens = [len(e) for e in eojeols]
|
||
sent_lens = [len(s) for s in sentences]
|
||
|
||
feats["char_count"] = float(chars)
|
||
feats["eojeol_count"] = float(len(eojeols))
|
||
feats["sentence_count"] = float(len(sentences))
|
||
feats["paragraph_count"] = float(len(paragraphs))
|
||
|
||
# --- 띄어쓰기 ---
|
||
feats["space_ratio"] = _safe_div(norm.count(" "), chars)
|
||
mean_e = _mean(eojeol_lens)
|
||
std_e = _std(eojeol_lens)
|
||
feats["mean_eojeol_len"] = mean_e
|
||
feats["std_eojeol_len"] = std_e
|
||
feats["cv_eojeol_len"] = _safe_div(std_e, mean_e)
|
||
feats["long_eojeol_ratio"] = _safe_div(
|
||
sum(1 for n in eojeol_lens if n >= 8), len(eojeol_lens)
|
||
)
|
||
feats["short_eojeol_ratio"] = _safe_div(
|
||
sum(1 for n in eojeol_lens if n <= 2), len(eojeol_lens)
|
||
)
|
||
|
||
# --- 문장 ---
|
||
mean_s = _mean(sent_lens)
|
||
std_s = _std(sent_lens)
|
||
feats["mean_sentence_len"] = mean_s
|
||
feats["std_sentence_len"] = std_s
|
||
# burstiness 대용. AI 생성문은 문장 길이가 균일해 이 값이 낮은 경향.
|
||
feats["cv_sentence_len"] = _safe_div(std_s, mean_s)
|
||
feats["max_sentence_len"] = float(max(sent_lens)) if sent_lens else 0.0
|
||
feats["min_sentence_len"] = float(min(sent_lens)) if sent_lens else 0.0
|
||
feats["mean_paragraph_len"] = _mean([len(p) for p in paragraphs])
|
||
|
||
# --- 쉼표·문장부호 ---
|
||
commas = norm.count(",")
|
||
puncts = _PUNCT.findall(norm)
|
||
feats["comma_per_sentence"] = _safe_div(commas, len(sentences))
|
||
feats["comma_ratio"] = _safe_div(commas, chars)
|
||
feats["punct_ratio"] = _safe_div(len(puncts), chars)
|
||
feats["punct_diversity"] = _safe_div(len(set(puncts)), len(puncts))
|
||
feats["ellipsis_ratio"] = _safe_div(
|
||
norm.count("…") + len(re.findall(r"\.\.\.", norm)), max(1, len(sentences))
|
||
)
|
||
feats["quote_ratio"] = _safe_div(
|
||
len(re.findall(r"[\"“”'‘’]", norm)), chars
|
||
)
|
||
feats["exclaim_question_ratio"] = _safe_div(
|
||
norm.count("!") + norm.count("?"), max(1, len(sentences))
|
||
)
|
||
|
||
# --- 종결 형태 ---
|
||
if sentences:
|
||
da = yo = noun_end = 0
|
||
for s in sentences:
|
||
body = s.rstrip(".!?…\"'“”’ ")
|
||
if not body:
|
||
continue
|
||
if body.endswith(("습니다", "합니다", "요")):
|
||
yo += 1
|
||
elif body.endswith("다"):
|
||
da += 1
|
||
elif _HANGUL.search(body[-1:]):
|
||
noun_end += 1
|
||
feats["end_da_ratio"] = _safe_div(da, len(sentences))
|
||
feats["end_yo_ratio"] = _safe_div(yo, len(sentences))
|
||
feats["end_noun_ratio"] = _safe_div(noun_end, len(sentences))
|
||
|
||
# --- 문자 구성 ---
|
||
feats["hangul_ratio"] = _safe_div(len(_HANGUL.findall(norm)), chars)
|
||
feats["digit_ratio"] = _safe_div(len(_DIGIT.findall(norm)), chars)
|
||
feats["latin_ratio"] = _safe_div(len(_LATIN.findall(norm)), chars)
|
||
feats["newline_ratio"] = _safe_div(norm.count("\n"), chars)
|
||
|
||
# --- 반복·다양도 ---
|
||
if eojeols:
|
||
cnt = Counter(eojeols)
|
||
feats["eojeol_ttr"] = _safe_div(len(cnt), len(eojeols))
|
||
feats["hapax_ratio"] = _safe_div(
|
||
sum(1 for c in cnt.values() if c == 1), len(cnt)
|
||
)
|
||
ordered = cnt.most_common(5)
|
||
feats["top1_token_share"] = _safe_div(ordered[0][1], len(eojeols))
|
||
feats["top5_token_share"] = _safe_div(
|
||
sum(c for _, c in ordered), len(eojeols)
|
||
)
|
||
bigrams = [f"{a}␟{b}" for a, b in zip(eojeols, eojeols[1:])]
|
||
feats["word_bigram_repeat_ratio"] = _repeat_ratio(bigrams)
|
||
feats["distinct_word_bigram_ratio"] = _safe_div(
|
||
len(set(bigrams)), len(bigrams)
|
||
)
|
||
|
||
compact = re.sub(r"\s+", "", norm)
|
||
if len(compact) >= 3:
|
||
trigrams = [compact[i : i + 3] for i in range(len(compact) - 2)]
|
||
feats["char3gram_repeat_ratio"] = _repeat_ratio(trigrams)
|
||
|
||
# --- 품사 ---
|
||
if use_pos:
|
||
pos_feats = _extract_pos_features(norm, len(eojeols))
|
||
if pos_feats:
|
||
feats.update(pos_feats)
|
||
|
||
return feats
|
||
|
||
|
||
def _extract_pos_features(norm: str, eojeol_count: int) -> dict[str, float] | None:
|
||
"""kiwipiepy 기반 품사 특징. 사용 불가면 None (호출자가 0.0 유지)."""
|
||
kiwi = _get_kiwi()
|
||
if kiwi is None:
|
||
return None
|
||
try:
|
||
tokens = kiwi.tokenize(norm)
|
||
except Exception as exc:
|
||
logger.warning("kiwi tokenize failed — POS features skipped: %s", exc)
|
||
return None
|
||
|
||
if not tokens:
|
||
return None
|
||
|
||
tags = [t.tag for t in tokens]
|
||
n = len(tags)
|
||
out: dict[str, float] = {
|
||
"pos_available": 1.0,
|
||
"morph_count": float(n),
|
||
"mean_morphs_per_eojeol": _safe_div(n, eojeol_count),
|
||
}
|
||
|
||
tag_counts = Counter(tags)
|
||
out["pos_ttr"] = _safe_div(len(tag_counts), n)
|
||
|
||
bigrams = [f"{a}_{b}" for a, b in zip(tags, tags[1:])]
|
||
out["pos_bigram_entropy"] = _entropy(list(Counter(bigrams).values()))
|
||
trigrams = [f"{a}_{b}_{c}" for a, b, c in zip(tags, tags[1:], tags[2:])]
|
||
out["pos_trigram_repeat_ratio"] = _repeat_ratio(trigrams)
|
||
|
||
def _ratio(prefixes: tuple[str, ...]) -> float:
|
||
return _safe_div(
|
||
sum(c for tag, c in tag_counts.items() if tag.startswith(prefixes)), n
|
||
)
|
||
|
||
def _diversity(prefixes: tuple[str, ...]) -> float:
|
||
"""해당 계열 안에서 서로 다른 표층형이 얼마나 다양한가."""
|
||
forms = {t.form for t in tokens if t.tag.startswith(prefixes)}
|
||
total = sum(1 for t in tokens if t.tag.startswith(prefixes))
|
||
return _safe_div(len(forms), total)
|
||
|
||
content = ("NNG", "NNP", "VV", "VA", "MAG")
|
||
out["content_word_ratio"] = _ratio(content)
|
||
out["josa_ratio"] = _ratio(("JK", "JX", "JC"))
|
||
out["josa_diversity"] = _diversity(("JK", "JX", "JC"))
|
||
out["eomi_ratio"] = _ratio(("EF", "EC", "EP", "ETN", "ETM"))
|
||
out["eomi_diversity"] = _diversity(("EF", "EC", "EP", "ETN", "ETM"))
|
||
out["noun_ratio"] = _ratio(("NNG", "NNP"))
|
||
out["verb_ratio"] = _ratio(("VV",))
|
||
out["adj_ratio"] = _ratio(("VA",))
|
||
out["adverb_ratio"] = _ratio(("MAG",))
|
||
out["conj_adverb_ratio"] = _ratio(("MAJ",))
|
||
out["dependent_noun_ratio"] = _ratio(("NNB",))
|
||
return out
|
||
|
||
|
||
def features_to_vector(
|
||
feats: dict[str, float], zeroed: Sequence[str] = ()
|
||
) -> list[float]:
|
||
"""FEATURE_NAMES 순서 그대로의 수치 벡터. 모델 입출력의 유일한 계약.
|
||
|
||
zeroed 에 든 특징은 0.0 으로 눌러 모델이 무시하도록 한다. 학습과 추론에서
|
||
**반드시 같은 목록**을 써야 하므로, 목록은 아티팩트에 저장되고 로드 시
|
||
복원된다. 벡터 길이는 어떤 경우에도 변하지 않는다.
|
||
"""
|
||
blocked = set(zeroed)
|
||
return [
|
||
0.0 if name in blocked else float(feats.get(name, 0.0))
|
||
for name in FEATURE_NAMES
|
||
]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 결과 객체
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@dataclass(frozen=True)
|
||
class SegmentScore:
|
||
"""구간별 의심도. 문서 전체가 아니라 어느 부분이 의심되는지 보여준다."""
|
||
|
||
index: int
|
||
start: int
|
||
end: int
|
||
char_count: int
|
||
score: float | None
|
||
suspicion_level: SuspicionLevel | None
|
||
scored: bool
|
||
note: str = ""
|
||
preview: str = ""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class AiDetectionResult:
|
||
"""AI 생성 의심도 결과.
|
||
|
||
⚠️ score 는 'AI가 썼을 확률'이 아니라 '사람 검토 우선순위'다. 확정 판정이
|
||
아니며, 저자 통보·계약 조치의 단독 근거로 사용해서는 안 된다.
|
||
"""
|
||
|
||
available: bool
|
||
score: float | None
|
||
suspicion_level: SuspicionLevel | None
|
||
provenance: Provenance
|
||
is_stub: bool
|
||
model_version: str
|
||
note: str
|
||
feature_set_version: str = FEATURE_SET_VERSION
|
||
pos_available: bool = False
|
||
char_count: int = 0
|
||
warnings: list[str] = field(default_factory=list)
|
||
segments: list[SegmentScore] = field(default_factory=list)
|
||
features: dict[str, float] = field(default_factory=dict)
|
||
top_contributions: list[tuple[str, float]] = field(default_factory=list)
|
||
|
||
def to_dict(self) -> dict:
|
||
"""API/로그 직렬화용. schemas.py 매핑은 docs/AI_DETECTION.md 참조."""
|
||
return {
|
||
"available": self.available,
|
||
"score": self.score,
|
||
"suspicion_level": self.suspicion_level,
|
||
"provenance": self.provenance,
|
||
"is_stub": self.is_stub,
|
||
"model_version": self.model_version,
|
||
"note": self.note,
|
||
"feature_set_version": self.feature_set_version,
|
||
"pos_available": self.pos_available,
|
||
"char_count": self.char_count,
|
||
"warnings": list(self.warnings),
|
||
"segments": [
|
||
{
|
||
"index": s.index,
|
||
"start": s.start,
|
||
"end": s.end,
|
||
"char_count": s.char_count,
|
||
"score": s.score,
|
||
"suspicion_level": s.suspicion_level,
|
||
"scored": s.scored,
|
||
"note": s.note,
|
||
"preview": s.preview,
|
||
}
|
||
for s in self.segments
|
||
],
|
||
"top_contributions": [
|
||
{"feature": name, "contribution": round(val, 4)}
|
||
for name, val in self.top_contributions
|
||
],
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 휴리스틱 baseline (모델 없을 때, 명시적 opt-in)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
#: 미검증 참조 구간. 79권 인간 저작 코퍼스로 캘리브레이션하기 전까지는
|
||
#: 문헌상 경향을 반영한 자리표시자일 뿐이다. 절대 판정 근거가 아니다.
|
||
#: (feature, human_typical, ai_typical) — ai 쪽에 가까울수록 가점.
|
||
_HEURISTIC_RULES: tuple[tuple[str, float, float], ...] = (
|
||
("cv_sentence_len", 0.75, 0.35), # AI: 문장 길이 균일
|
||
("cv_eojeol_len", 0.62, 0.45), # AI: 어절 길이도 균일
|
||
("hapax_ratio", 0.78, 0.60), # AI: 어휘 재사용 잦음
|
||
("comma_per_sentence", 0.60, 1.40), # AI: 쉼표 과다 사용
|
||
("char3gram_repeat_ratio", 0.28, 0.42), # AI: 표현 반복
|
||
("word_bigram_repeat_ratio", 0.05, 0.14),
|
||
("mean_sentence_len", 38.0, 58.0), # AI: 만연체 경향
|
||
)
|
||
|
||
|
||
def _heuristic_score(feats: dict[str, float]) -> float:
|
||
"""규칙 기반 baseline 점수 (0~1). 결정적이며 근거를 추적할 수 있다.
|
||
|
||
각 규칙을 human_typical↔ai_typical 사이의 선형 위치로 환산해 평균한다.
|
||
학습 모델이 준비되면 즉시 대체될 임시 계층이다.
|
||
"""
|
||
positions: list[float] = []
|
||
for name, human_val, ai_val in _HEURISTIC_RULES:
|
||
val = feats.get(name)
|
||
if val is None:
|
||
continue
|
||
span = ai_val - human_val
|
||
if span == 0:
|
||
continue
|
||
pos = (val - human_val) / span
|
||
positions.append(max(0.0, min(1.0, pos)))
|
||
if not positions:
|
||
return 0.0
|
||
return max(0.0, min(1.0, sum(positions) / len(positions)))
|
||
|
||
|
||
def percentile(values: Sequence[float], pct: float) -> float:
|
||
"""선형 보간 백분위. numpy 없이도 동작하도록 직접 구현한다."""
|
||
if not values:
|
||
return 0.0
|
||
ordered = sorted(values)
|
||
if len(ordered) == 1:
|
||
return float(ordered[0])
|
||
pos = (len(ordered) - 1) * max(0.0, min(100.0, pct)) / 100.0
|
||
low = int(math.floor(pos))
|
||
high = int(math.ceil(pos))
|
||
if low == high:
|
||
return float(ordered[low])
|
||
return float(ordered[low] + (ordered[high] - ordered[low]) * (pos - low))
|
||
|
||
|
||
def calibrate_cuts(
|
||
human_scores: Sequence[float],
|
||
low_percentile: float = DEFAULT_LOW_PERCENTILE,
|
||
high_percentile: float = DEFAULT_HIGH_PERCENTILE,
|
||
) -> tuple[float, float]:
|
||
"""인간 저작 코퍼스 점수 분포 → (low_cut, high_cut).
|
||
|
||
라벨이 전혀 필요 없다. "AI가 쓴 글의 점수는 얼마인가"가 아니라 "우리 코퍼스의
|
||
인간 저작물 중 상위 몇 %인가"로 컷을 정의하기 때문이다. 그래서 정확도를
|
||
측정하지 않고도 참인 진술("등록 자서전 대비 상위 2% 이례적 문체")이 되고,
|
||
'high' 배지 비율이 정의상 고정되어 검토 부하도 예측 가능해진다.
|
||
|
||
두 백분위가 같은 값으로 뭉개지면(분포가 평평한 경우) high 를 살짝 올려
|
||
medium 구간이 사라지지 않게 한다.
|
||
"""
|
||
if not human_scores:
|
||
raise ValueError("점수 표본이 비어 있어 컷을 산출할 수 없습니다.")
|
||
if not 0.0 <= low_percentile < high_percentile <= 100.0:
|
||
raise ValueError(
|
||
f"백분위는 0 <= low < high <= 100 이어야 합니다: {low_percentile}, {high_percentile}"
|
||
)
|
||
low = percentile(human_scores, low_percentile)
|
||
high = percentile(human_scores, high_percentile)
|
||
if high <= low:
|
||
high = min(1.0, low + 0.01)
|
||
return round(low, 4), round(high, 4)
|
||
|
||
|
||
def score_text_heuristic(text: str, use_pos: bool = True) -> float | None:
|
||
"""캘리브레이션용 단일 텍스트 점수. 너무 짧으면 None(표본에서 제외)."""
|
||
norm = normalize_text(text)
|
||
if len(norm) < HARD_MIN_CHARS:
|
||
return None
|
||
return _heuristic_score(extract_features(norm, use_pos=use_pos))
|
||
|
||
|
||
def _heuristic_contributions(feats: dict[str, float]) -> list[tuple[str, float]]:
|
||
"""휴리스틱에서 어느 특징이 점수를 끌어올렸는지."""
|
||
out: list[tuple[str, float]] = []
|
||
for name, human_val, ai_val in _HEURISTIC_RULES:
|
||
val = feats.get(name)
|
||
if val is None:
|
||
continue
|
||
span = ai_val - human_val
|
||
if span == 0:
|
||
continue
|
||
pos = max(0.0, min(1.0, (val - human_val) / span))
|
||
out.append((name, pos - 0.5))
|
||
out.sort(key=lambda kv: abs(kv[1]), reverse=True)
|
||
return out[:6]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 모델 아티팩트
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@dataclass
|
||
class ModelArtifact:
|
||
"""학습 산출물 + 재현에 필요한 메타데이터."""
|
||
|
||
estimator: object
|
||
feature_names: tuple[str, ...]
|
||
feature_set_version: str
|
||
model_version: str
|
||
requires_pos: bool
|
||
low_cut: float
|
||
high_cut: float
|
||
metrics: dict
|
||
zeroed_features: tuple[str, ...] = ()
|
||
trained_at: str = ""
|
||
sklearn_version: str = ""
|
||
notes: str = ""
|
||
|
||
|
||
def load_artifact(path: str | Path) -> ModelArtifact | None:
|
||
"""joblib 아티팩트 로드. 실패는 예외가 아니라 None (서비스는 계속 떠야 한다)."""
|
||
p = Path(path)
|
||
if not p.exists():
|
||
logger.info("AI detector model not found at %s — running unavailable", p)
|
||
return None
|
||
try:
|
||
import joblib
|
||
except Exception as exc:
|
||
logger.warning("joblib unavailable — cannot load AI detector model: %s", exc)
|
||
return None
|
||
|
||
try:
|
||
payload = joblib.load(p)
|
||
except Exception as exc:
|
||
logger.error("Failed to load AI detector artifact %s: %s", p, exc)
|
||
return None
|
||
|
||
if not isinstance(payload, dict) or "estimator" not in payload:
|
||
logger.error("Malformed AI detector artifact at %s (missing 'estimator')", p)
|
||
return None
|
||
|
||
names = tuple(payload.get("feature_names") or ())
|
||
if names and names != FEATURE_NAMES:
|
||
logger.error(
|
||
"AI detector artifact feature mismatch (artifact=%d, code=%d). "
|
||
"Retrain required — refusing to load.",
|
||
len(names),
|
||
len(FEATURE_NAMES),
|
||
)
|
||
return None
|
||
|
||
return ModelArtifact(
|
||
estimator=payload["estimator"],
|
||
feature_names=names or FEATURE_NAMES,
|
||
feature_set_version=payload.get("feature_set_version", "unknown"),
|
||
model_version=payload.get("model_version", "unknown"),
|
||
requires_pos=bool(payload.get("requires_pos", False)),
|
||
low_cut=float(payload.get("low_cut", DEFAULT_LOW_CUT)),
|
||
high_cut=float(payload.get("high_cut", DEFAULT_HIGH_CUT)),
|
||
metrics=payload.get("metrics", {}),
|
||
zeroed_features=tuple(payload.get("zeroed_features") or ()),
|
||
trained_at=payload.get("trained_at", ""),
|
||
sklearn_version=payload.get("sklearn_version", ""),
|
||
notes=payload.get("notes", ""),
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 탐지기
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class AiGenerationDetector:
|
||
"""언어특징 기반 AI 생성 의심도 산출기.
|
||
|
||
동작 모드 (결과의 is_stub/model_version/note 로 항상 구분 가능):
|
||
· trained — 학습 아티팩트 로드 성공. is_stub=False.
|
||
· heuristic — 아티팩트 없음 + allow_heuristic=True. is_stub=True,
|
||
model_version="heuristic-baseline-v1".
|
||
· unavailable— 아티팩트 없음 + allow_heuristic=False(기본). available=False,
|
||
score=None.
|
||
"""
|
||
|
||
HEURISTIC_VERSION = "heuristic-baseline-v1"
|
||
|
||
def __init__(
|
||
self,
|
||
model_path: str | Path | None = None,
|
||
allow_heuristic: bool = False,
|
||
use_pos: bool = True,
|
||
low_cut: float | None = None,
|
||
high_cut: float | None = None,
|
||
):
|
||
self.model_path = str(
|
||
model_path or os.environ.get(ENV_MODEL_PATH) or DEFAULT_MODEL_PATH
|
||
)
|
||
self.allow_heuristic = allow_heuristic
|
||
self.use_pos = use_pos
|
||
# 휴리스틱 모드에서 쓸 백분위 컷. 둘 다 주어졌을 때만 적용한다.
|
||
self.low_cut = low_cut
|
||
self.high_cut = high_cut
|
||
self.artifact = load_artifact(self.model_path)
|
||
|
||
# -- 모드 --------------------------------------------------------------
|
||
|
||
@property
|
||
def mode(self) -> Literal["trained", "heuristic", "unavailable"]:
|
||
if self.artifact is not None:
|
||
return "trained"
|
||
return "heuristic" if self.allow_heuristic else "unavailable"
|
||
|
||
@property
|
||
def model_version(self) -> str:
|
||
if self.artifact is not None:
|
||
return self.artifact.model_version
|
||
if not self.allow_heuristic:
|
||
return "unavailable"
|
||
# 컷이 코퍼스로 캘리브레이션되었는지를 버전 문자열에 드러낸다.
|
||
return (
|
||
f"{self.HEURISTIC_VERSION}+corpus-percentile"
|
||
if self.cuts_calibrated else self.HEURISTIC_VERSION
|
||
)
|
||
|
||
@property
|
||
def cuts_calibrated(self) -> bool:
|
||
"""컷이 자리표시자가 아니라 실제 코퍼스 분포에서 나왔는지."""
|
||
if self.artifact is not None:
|
||
return True
|
||
return self.low_cut is not None and self.high_cut is not None
|
||
|
||
def _cuts(self) -> tuple[float, float]:
|
||
if self.artifact is not None:
|
||
return self.artifact.low_cut, self.artifact.high_cut
|
||
if self.low_cut is not None and self.high_cut is not None:
|
||
return self.low_cut, self.high_cut
|
||
return DEFAULT_LOW_CUT, DEFAULT_HIGH_CUT
|
||
|
||
def _level(self, score: float) -> SuspicionLevel:
|
||
low, high = self._cuts()
|
||
if score < low:
|
||
return "low"
|
||
if score < high:
|
||
return "medium"
|
||
return "high"
|
||
|
||
# -- 점수 --------------------------------------------------------------
|
||
|
||
def _score_features(self, feats: dict[str, float]) -> float | None:
|
||
"""특징 → 0~1 점수. 모드에 따라 학습 모델 또는 휴리스틱."""
|
||
if self.artifact is not None:
|
||
vec = features_to_vector(feats, self.artifact.zeroed_features)
|
||
try:
|
||
proba = self.artifact.estimator.predict_proba([vec])
|
||
return float(proba[0][1])
|
||
except Exception as exc:
|
||
logger.error("AI detector inference failed: %s", exc)
|
||
return None
|
||
if self.allow_heuristic:
|
||
return _heuristic_score(feats)
|
||
return None
|
||
|
||
def _contributions(self, feats: dict[str, float]) -> list[tuple[str, float]]:
|
||
"""설명용 상위 기여 특징.
|
||
|
||
선형 모델이면 coef × 표준화값을, 그 외에는 휴리스틱 규칙 위치를 쓴다.
|
||
기여도를 못 뽑는 모델(HGB 등)이면 빈 리스트 — 지어내지 않는다.
|
||
"""
|
||
if self.artifact is None:
|
||
return _heuristic_contributions(feats) if self.allow_heuristic else []
|
||
est = self.artifact.estimator
|
||
try:
|
||
coefs, scaler = _linear_parts(est)
|
||
if coefs is None:
|
||
return []
|
||
vec = features_to_vector(feats, self.artifact.zeroed_features)
|
||
if scaler is not None:
|
||
mean = getattr(scaler, "mean_", None)
|
||
scale = getattr(scaler, "scale_", None)
|
||
if mean is not None and scale is not None:
|
||
vec = [
|
||
(v - float(m)) / (float(s) if s else 1.0)
|
||
for v, m, s in zip(vec, mean, scale)
|
||
]
|
||
pairs = [
|
||
(name, float(c) * float(v))
|
||
for name, c, v in zip(FEATURE_NAMES, coefs, vec)
|
||
]
|
||
pairs.sort(key=lambda kv: abs(kv[1]), reverse=True)
|
||
return pairs[:6]
|
||
except Exception as exc:
|
||
logger.debug("Contribution extraction skipped: %s", exc)
|
||
return []
|
||
|
||
# -- 공개 API ----------------------------------------------------------
|
||
|
||
def detect(self, text: str, with_segments: bool = True) -> AiDetectionResult:
|
||
"""문서 1건의 AI 생성 의심도.
|
||
|
||
짧은 텍스트는 통계가 불안정하므로 HARD_MIN_CHARS 미만이면 채점을
|
||
거부하고, MIN_RELIABLE_CHARS 미만이면 경고를 붙인다.
|
||
"""
|
||
norm = normalize_text(text)
|
||
char_count = len(norm)
|
||
warnings: list[str] = []
|
||
|
||
if self.mode == "unavailable":
|
||
return AiDetectionResult(
|
||
available=False,
|
||
score=None,
|
||
suspicion_level=None,
|
||
provenance="unknown",
|
||
is_stub=False,
|
||
model_version="unavailable",
|
||
note=(
|
||
"AI 생성 판별 모델이 학습되지 않아 점수를 산출하지 않습니다. "
|
||
"scripts/train_ai_detector.py 로 학습 후 "
|
||
f"{ENV_MODEL_PATH} 를 지정하세요."
|
||
),
|
||
char_count=char_count,
|
||
warnings=["model_not_trained"],
|
||
)
|
||
|
||
if char_count < HARD_MIN_CHARS:
|
||
return AiDetectionResult(
|
||
available=False,
|
||
score=None,
|
||
suspicion_level=None,
|
||
provenance="unknown",
|
||
is_stub=self.artifact is None,
|
||
model_version=self.model_version,
|
||
note=(
|
||
f"텍스트가 너무 짧아({char_count}자) 언어특징 통계가 "
|
||
f"무의미합니다. 최소 {HARD_MIN_CHARS}자 필요."
|
||
),
|
||
char_count=char_count,
|
||
warnings=["text_too_short"],
|
||
)
|
||
|
||
if char_count < MIN_RELIABLE_CHARS:
|
||
warnings.append("short_text_low_confidence")
|
||
|
||
feats = extract_features(norm, use_pos=self.use_pos)
|
||
pos_ok = feats.get("pos_available", 0.0) >= 1.0
|
||
if not pos_ok:
|
||
warnings.append("pos_features_unavailable")
|
||
if self.artifact is not None and self.artifact.requires_pos and not pos_ok:
|
||
warnings.append("pos_required_by_model_but_missing")
|
||
logger.warning(
|
||
"Model %s was trained with POS features but kiwipiepy is "
|
||
"unavailable — score reliability degraded.",
|
||
self.artifact.model_version,
|
||
)
|
||
|
||
score = self._score_features(feats)
|
||
if score is None:
|
||
return AiDetectionResult(
|
||
available=False,
|
||
score=None,
|
||
suspicion_level=None,
|
||
provenance="unknown",
|
||
is_stub=self.artifact is None,
|
||
model_version=self.model_version,
|
||
note="점수 산출에 실패했습니다(추론 오류). 로그를 확인하세요.",
|
||
pos_available=pos_ok,
|
||
char_count=char_count,
|
||
warnings=warnings + ["inference_failed"],
|
||
features=feats,
|
||
)
|
||
|
||
segments = self._score_segments(norm) if with_segments else []
|
||
provenance = self._infer_provenance(score, segments)
|
||
|
||
if self.artifact is None and self.cuts_calibrated:
|
||
note = (
|
||
"학습 모델이 아니라 언어특징 규칙 점수입니다. 'AI일 확률'이 아니라 "
|
||
"등록 코퍼스의 인간 저작물 대비 문체 이례도이며, 검토 우선순위 "
|
||
"정렬용입니다. 판정 근거로 사용할 수 없습니다."
|
||
)
|
||
elif self.artifact is None:
|
||
note = (
|
||
"⚠️ 미검증 휴리스틱 baseline 입니다. 학습된 모델도 아니고 컷도 "
|
||
"캘리브레이션되지 않았습니다(자리표시자). 판정 근거로 사용할 수 "
|
||
"없습니다. scripts/calibrate_ai_detector_cuts.py 로 컷을 산출하세요."
|
||
)
|
||
else:
|
||
note = (
|
||
"AI 생성 '의심도'이며 확정 판정이 아닙니다. 대필·윤문된 원고는 "
|
||
"높게 나올 수 있으므로 반드시 사람 검토를 거치세요."
|
||
)
|
||
|
||
return AiDetectionResult(
|
||
available=True,
|
||
score=round(score, 4),
|
||
suspicion_level=self._level(score),
|
||
provenance=provenance,
|
||
is_stub=self.artifact is None,
|
||
model_version=self.model_version,
|
||
note=note,
|
||
pos_available=pos_ok,
|
||
char_count=char_count,
|
||
warnings=warnings,
|
||
segments=segments,
|
||
features=feats,
|
||
top_contributions=self._contributions(feats),
|
||
)
|
||
|
||
# -- 구간 채점 ---------------------------------------------------------
|
||
|
||
def _build_segments(self, norm: str) -> list[tuple[int, int, str]]:
|
||
"""문단을 SEGMENT_TARGET_CHARS 이상으로 병합. (start, end, text) 반환."""
|
||
segments: list[tuple[int, int, str]] = []
|
||
cursor = 0
|
||
buf_start: int | None = None
|
||
buf: list[str] = []
|
||
buf_len = 0
|
||
|
||
for para in split_paragraphs(norm):
|
||
idx = norm.find(para, cursor)
|
||
if idx < 0:
|
||
idx = cursor
|
||
cursor = idx + len(para)
|
||
if buf_start is None:
|
||
buf_start = idx
|
||
buf.append(para)
|
||
buf_len += len(para)
|
||
if buf_len >= SEGMENT_TARGET_CHARS:
|
||
segments.append((buf_start, cursor, "\n\n".join(buf)))
|
||
buf, buf_len, buf_start = [], 0, None
|
||
|
||
if buf and buf_start is not None:
|
||
if segments and buf_len < HARD_MIN_CHARS:
|
||
# 마지막 자투리는 직전 구간에 흡수 (단독 채점 불가 길이)
|
||
s, _, prev = segments[-1]
|
||
segments[-1] = (s, cursor, prev + "\n\n" + "\n\n".join(buf))
|
||
else:
|
||
segments.append((buf_start, cursor, "\n\n".join(buf)))
|
||
return segments
|
||
|
||
def _score_segments(self, norm: str) -> list[SegmentScore]:
|
||
out: list[SegmentScore] = []
|
||
for i, (start, end, seg_text) in enumerate(self._build_segments(norm)):
|
||
n = len(seg_text)
|
||
preview = seg_text[:60].replace("\n", " ")
|
||
if n < HARD_MIN_CHARS:
|
||
out.append(
|
||
SegmentScore(
|
||
index=i, start=start, end=end, char_count=n,
|
||
score=None, suspicion_level=None, scored=False,
|
||
note=f"{HARD_MIN_CHARS}자 미만 — 채점 제외",
|
||
preview=preview,
|
||
)
|
||
)
|
||
continue
|
||
feats = extract_features(seg_text, use_pos=self.use_pos)
|
||
s = self._score_features(feats)
|
||
if s is None:
|
||
out.append(
|
||
SegmentScore(
|
||
index=i, start=start, end=end, char_count=n,
|
||
score=None, suspicion_level=None, scored=False,
|
||
note="추론 실패", preview=preview,
|
||
)
|
||
)
|
||
continue
|
||
out.append(
|
||
SegmentScore(
|
||
index=i, start=start, end=end, char_count=n,
|
||
score=round(s, 4), suspicion_level=self._level(s),
|
||
scored=True,
|
||
note="" if n >= MIN_RELIABLE_CHARS else "짧은 구간 — 신뢰도 낮음",
|
||
preview=preview,
|
||
)
|
||
)
|
||
return out
|
||
|
||
# -- provenance --------------------------------------------------------
|
||
|
||
def _infer_provenance(
|
||
self, doc_score: float, segments: list[SegmentScore]
|
||
) -> Provenance:
|
||
"""작성 경로 **추정**. 확정이 아니며 검토자 참고용이다.
|
||
|
||
규칙 (docs/AI_DETECTION.md 와 동일하게 유지할 것):
|
||
human — 문서·구간 모두 low
|
||
ai — 문서 high 이고 low 구간이 없음
|
||
mixed — high 구간과 low 구간이 공존 (부분 삽입 의심)
|
||
edited — 전 구간이 medium 에 몰림 (AI 초안 + 사람 윤문, 또는 그 역)
|
||
unknown— 채점된 구간이 없어 판단 불가
|
||
"""
|
||
low, high = self._cuts()
|
||
scored = [s for s in segments if s.scored and s.score is not None]
|
||
|
||
if not scored:
|
||
if doc_score >= high:
|
||
return "ai"
|
||
if doc_score < low:
|
||
return "human"
|
||
return "edited"
|
||
|
||
levels = [s.suspicion_level for s in scored]
|
||
has_high = "high" in levels
|
||
has_low = "low" in levels
|
||
|
||
if has_high and has_low:
|
||
return "mixed"
|
||
if doc_score >= high and not has_low:
|
||
return "ai"
|
||
if doc_score < low and not has_high:
|
||
return "human"
|
||
if all(lv == "medium" for lv in levels):
|
||
return "edited"
|
||
return "mixed" if has_high else "edited"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 싱글턴 접근자
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_detector_cache: dict[tuple, AiGenerationDetector] = {}
|
||
|
||
|
||
def get_ai_detector(
|
||
model_path: str | Path | None = None,
|
||
allow_heuristic: bool | None = None,
|
||
use_pos: bool = True,
|
||
low_cut: float | None = None,
|
||
high_cut: float | None = None,
|
||
) -> AiGenerationDetector:
|
||
"""탐지기 인스턴스(프로세스 캐시).
|
||
|
||
allow_heuristic 을 생략하면 환경변수 AI_DETECTOR_ALLOW_HEURISTIC 을 따르고,
|
||
그것도 없으면 False(= 모델 없으면 unavailable) 다. 기본값을 False 로 두는
|
||
이유는, 미검증 점수가 조용히 운영에 노출되는 상황을 막기 위해서다.
|
||
|
||
low_cut/high_cut 은 휴리스틱 모드에서만 쓰이며, 둘 다 주어져야 적용된다.
|
||
"""
|
||
if allow_heuristic is None:
|
||
allow_heuristic = os.environ.get(
|
||
"AI_DETECTOR_ALLOW_HEURISTIC", ""
|
||
).strip().lower() in {"1", "true", "yes"}
|
||
key = (str(model_path or ""), bool(allow_heuristic), bool(use_pos), low_cut, high_cut)
|
||
if key not in _detector_cache:
|
||
_detector_cache[key] = AiGenerationDetector(
|
||
model_path=model_path, allow_heuristic=allow_heuristic, use_pos=use_pos,
|
||
low_cut=low_cut, high_cut=high_cut,
|
||
)
|
||
return _detector_cache[key]
|
||
|
||
|
||
def reset_detector_cache() -> None:
|
||
"""테스트/재로딩용."""
|
||
_detector_cache.clear()
|
||
|
||
|
||
def _linear_parts(estimator) -> tuple[Sequence[float] | None, object | None]:
|
||
"""추정기에서 (선형 계수, 스케일러) 추출. 없으면 (None, None).
|
||
|
||
Pipeline / CalibratedClassifierCV 를 한 겹씩 벗겨 본다.
|
||
"""
|
||
scaler = None
|
||
est = estimator
|
||
|
||
steps = getattr(est, "steps", None)
|
||
if steps:
|
||
for _, step in steps:
|
||
if hasattr(step, "mean_") and hasattr(step, "scale_"):
|
||
scaler = step
|
||
est = steps[-1][1]
|
||
|
||
calibrated = getattr(est, "calibrated_classifiers_", None)
|
||
if calibrated:
|
||
inner = getattr(calibrated[0], "estimator", None)
|
||
if inner is not None:
|
||
inner_steps = getattr(inner, "steps", None)
|
||
if inner_steps:
|
||
for _, step in inner_steps:
|
||
if hasattr(step, "mean_") and hasattr(step, "scale_"):
|
||
scaler = step
|
||
inner = inner_steps[-1][1]
|
||
est = inner
|
||
|
||
coef = getattr(est, "coef_", None)
|
||
if coef is None:
|
||
return None, scaler
|
||
return list(coef[0]), scaler
|