feat: AI 생성 표본 제작 스크립트

보유 코퍼스 31,560건이 전부 human 라벨이라 이진 판별기를 학습할 수 없고
(train_ai_detector.py 가 exit 2 로 막는다), 조사 결과 공개된 한국어 AI 생성
판별 데이터셋도 확인되지 않았다. AI 쪽 절반을 직접 만든다.

build_ai_training_dataset.py 가 외부 호출을 금지하므로 생성은 별도 스크립트로
분리했다. 프롬프트는 이미 파생된 메타데이터만 쓰고 에피소드 본문은 어떤
경로로도 API 요청에 실리지 않는다. 주석만으로는 다음 사람이 META_COLUMNS 에
본문 컬럼을 한 줄 추가하는 것을 못 막으므로, assert_no_body_columns() 가
실행 시점에 차단하고 _build_prompt() 는 본문을 넘길 파라미터 자체가 없다.

- 길이 매칭: human 분포에서 목표 길이를 뽑아 지시하고 벗어난 결과는 버린다.
  분량이 다르면 판별기가 문체가 아니라 길이를 배운다.
- 생성기 다중화: --model 반복 지정 시 라운드로빈. 1개면 그 모델만 잡는
  판별기가 되므로 경고한다.
- 문체 6종을 결정적으로 배분해 한 모델의 한 문체 암기를 막는다.
- 생성문에도 human 과 같은 normalize_ocr 을 적용한다. 한쪽만 정규화하면
  오염 방향만 뒤집힐 뿐이다.
- append 전용 + prompt_id 스킵으로 재개 가능. dry-run 은 파일시스템도
  건드리지 않는다.

--base-url 로 사내 GPU 의 OpenAI 호환 서버를 쓰면 메타데이터조차 외부로
나가지 않는다.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

scripts/generate_ai_samples.py

@@ -0,0 +1,303 @@
+#!/usr/bin/env python3
+"""AI 생성 표본 제작 — 판별기 학습셋의 AI 쪽 절반을 만든다.
+
+왜 필요한가:
+  보유 코퍼스는 79권 31,560건이 전부 human 라벨이다. 이진 판별기는 한쪽
+  라벨만으로 학습할 수 없고(`train_ai_detector.py` 가 exit 2 로 막는다),
+  공개된 한국어 AI 생성 판별 데이터셋은 확인되지 않았다. 그래서 직접 만든다.
+
+⚠️ 원고를 외부로 내보내지 않는다:
+  프롬프트는 **이미 파생된 메타데이터만** 쓴다(`gpt_keyword`, `gpt_title`,
+  `gpt_persona_keyword`, `gpt_character_keyword`). 에피소드 본문은 어떤
+  경로로도 API 요청에 실리지 않으며, `_build_prompt` 가 본문 컬럼을 아예
+  받지 않는 구조로 그것을 강제한다. 이 성질을 깨는 수정을 하지 말 것.
+  (`build_ai_training_dataset.py` 가 외부 호출을 금지하는 것과 같은 이유다.)
+
+설계상 반드시 지켜야 하는 것:
+  1. **길이 매칭** — human 에피소드는 780~1,029자(p10~p90)에 몰려 있다. 분량이
+     다르면 판별기는 문체가 아니라 길이를 배운다(docs/AI_DETECTION.md 경고).
+     목표 길이를 human 분포에서 뽑아 지시하고, 벗어난 결과는 버린다.
+  2. **생성기 다중화** — 한 모델로만 만들면 그 모델만 잡는 판별기가 된다.
+     `--model` 을 반복 지정하면 라운드로빈으로 섞는다. 최소 2개를 권장한다.
+  3. **동일 정규화** — 생성문에도 human 과 똑같이 `normalize_ocr` 을 적용한다.
+     한쪽만 정규화하면 오염 방향만 뒤집힐 뿐이다.
+  4. **재개 가능** — append 전용 JSONL 이고, 재실행하면 이미 만든 prompt_id 를
+     건너뛴다. 중간에 죽어도 호출 비용을 다시 쓰지 않는다.
+
+로컬 GPU 사용:
+  --base-url 로 OpenAI 호환 서버(vLLM 등)를 가리키면 사내 GPU 의 오픈소스
+  모델로 생성할 수 있다. 원고 메타데이터조차 외부로 내보내지 않는 경로다.
+
+사용:
+  # 0) 비용 0 — 프롬프트만 확인
+  python scripts/generate_ai_samples.py --xlsx <파일> --limit 5 --dry-run
+
+  # 1) 실제 생성 (모델 2개 혼합)
+  python scripts/generate_ai_samples.py --xlsx <파일> \\
+      --model gpt-4o-mini --model gpt-4.1-mini \\
+      --limit 4000 --out data/training/ai_samples.jsonl
+
+출력 JSONL 1행 (build_ai_training_dataset.py --ai-jsonl 이 그대로 읽는다):
+  {"text": "...", "generator": "gpt-4o-mini", "book": "", "prompt_id": "...",
+   "target_chars": 950, "char_count": 963, "style": "...", "meta": {...}}
+"""
+
+from __future__ import annotations
+
+import argparse
+import contextlib
+import hashlib
+import json
+import logging
+import os
+import random
+import sys
+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("gen-ai-samples")
+
+from app.engine.ocr_normalize import normalize_ocr  # noqa: E402
+
+#: 프롬프트에 쓸 수 있는 컬럼. 본문 컬럼은 의도적으로 목록에 없다.
+META_COLUMNS = (
+    "gpt_keyword",
+    "gpt_title",
+    "gpt_persona_keyword",
+    "gpt_character_keyword",
+    "생애사건_category",
+)
+TEXT_COLUMN_BLOCKLIST = ("에피소드", "본문", "내용", "원고", "text", "content")
+
+SYSTEM_PROMPT = (
+    "당신은 한국어 자서전 원고를 쓰는 작가다. 주어진 소재로 자서전의 한 "
+    "에피소드를 쓴다. 제목·머리말·목록·마크다운을 쓰지 말고 본문 산문만 쓴다. "
+    "설명하거나 요약하지 말고, 장면과 감정을 가진 이야기로 쓴다."
+)
+
+#: 한 모델이 한 가지 문체로만 쓰면 판별기가 그 문체를 외운다. 지시를 흩는다.
+STYLE_VARIANTS = (
+    "담담하고 절제된 문체로, 감정을 직접 말하지 말고 장면으로 보여준다.",
+    "회상하는 노년의 목소리로, 긴 문장과 짧은 문장을 섞어 쓴다.",
+    "대화를 두세 번 넣고, 인물의 말버릇이 드러나게 쓴다.",
+    "구체적인 시간·장소·사물 이름을 여러 개 넣어 사실적으로 쓴다.",
+    "한 가지 감각(냄새·소리·촉감)을 중심으로 장면을 풀어 쓴다.",
+    "후회와 자기변명이 뒤섞인 솔직한 1인칭으로 쓴다.",
+)
+
+
+def _stable_int(key: str) -> int:
+    return int(hashlib.sha1(key.encode("utf-8")).hexdigest()[:12], 16)
+
+
+def assert_no_body_columns(columns: list[str]) -> list[str]:
+    """본문 컬럼이 프롬프트 소재에 섞이는 것을 코드로 막는다.
+
+    주석만으로는 다음 사람이 META_COLUMNS 에 본문 컬럼을 한 줄 추가하는 것을
+    못 막는다. 그 한 줄이 원고를 외부 API 로 내보내는 경로가 된다.
+    """
+    leaked = [c for c in columns
+              if any(bad in c.lower() for bad in TEXT_COLUMN_BLOCKLIST)]
+    if leaked:
+        raise SystemExit(
+            f"본문으로 보이는 컬럼이 프롬프트 소재에 있습니다: {leaked}. "
+            "원고는 외부로 내보내지 않습니다. META_COLUMNS 를 확인하세요."
+        )
+    return columns
+
+
+def _build_prompt(meta: dict[str, str], target_chars: int, style: str) -> str:
+    """파생 메타데이터만으로 프롬프트를 만든다. 본문은 인자로 들어오지 않는다."""
+    lines = [f"{k}: {v}" for k, v in meta.items() if v]
+    soil = "\n".join(lines) if lines else "(소재 없음 — 평범한 유년의 한 장면)"
+    return (
+        f"아래 소재로 자서전 에피소드를 한국어로 쓴다.\n\n{soil}\n\n"
+        f"문체 지시: {style}\n"
+        f"분량: 공백 포함 {target_chars}자 내외(±15%). 이 분량을 반드시 지킨다.\n"
+        f"본문만 출력한다."
+    )
+
+
+def load_human_length_band(frame, text_column: str) -> tuple[list[int], int, int]:
+    """human 분포에서 목표 길이 후보와 허용 구간을 뽑는다."""
+    lengths = sorted(
+        len(t) for t in frame[text_column].dropna().astype(str) if len(t) >= 120
+    )
+    if not lengths:
+        raise SystemExit(f"'{text_column}' 에서 길이를 잴 수 있는 행이 없습니다.")
+
+    def pct(p: float) -> int:
+        return lengths[min(len(lengths) - 1, int(len(lengths) * p / 100))]
+
+    return lengths, pct(5), pct(95)
+
+
+def load_done_ids(out_path: Path) -> set[str]:
+    """이미 만든 prompt_id. 재실행 시 호출 비용을 다시 쓰지 않기 위한 것."""
+    done: set[str] = set()
+    if not out_path.exists():
+        return done
+    for line in out_path.read_text(encoding="utf-8").splitlines():
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            done.add(json.loads(line)["prompt_id"])
+        except (json.JSONDecodeError, KeyError):
+            logger.warning("기존 출력에 깨진 행이 있어 건너뜁니다.")
+    return done
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(
+        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
+    )
+    ap.add_argument("--xlsx", type=Path, required=True, help="소재를 뽑을 에피소드 xlsx")
+    ap.add_argument("--text-column", default="에피소드",
+                    help="길이 분포 계산에만 쓴다. 본문은 API 로 보내지 않는다.")
+    ap.add_argument("--model", action="append", default=[],
+                    help="생성 모델 (반복 지정 = 혼합). 최소 2개 권장")
+    ap.add_argument("--base-url", default=None,
+                    help="OpenAI 호환 엔드포인트 (사내 vLLM 등)")
+    ap.add_argument("--limit", type=int, default=100, help="생성할 표본 수")
+    ap.add_argument("--temperature", type=float, default=1.0)
+    ap.add_argument("--seed", type=int, default=20260819)
+    ap.add_argument("--out", type=Path, default=Path("data/training/ai_samples.jsonl"))
+    ap.add_argument("--dry-run", action="store_true",
+                    help="프롬프트만 출력하고 API 를 호출하지 않는다 (비용 0)")
+    args = ap.parse_args()
+
+    import pandas as pd
+
+    frame = pd.read_excel(args.xlsx, sheet_name=0)
+    lengths, low_chars, high_chars = load_human_length_band(frame, args.text_column)
+    logger.info(
+        "human 길이 분포: 중앙값 %d자, 허용 구간 %d~%d자 (n=%d)",
+        lengths[len(lengths) // 2], low_chars, high_chars, len(lengths),
+    )
+
+    usable_meta = assert_no_body_columns([c for c in META_COLUMNS if c in frame.columns])
+    if not usable_meta:
+        raise SystemExit(
+            f"소재로 쓸 컬럼이 없습니다. 기대: {META_COLUMNS} / 실제: {list(frame.columns)}"
+        )
+    logger.info("프롬프트 소재 컬럼: %s", usable_meta)
+
+    models = args.model or ["gpt-4o-mini"]
+    if len(models) == 1 and not args.dry_run:
+        logger.warning(
+            "생성 모델이 1개입니다. 그 모델만 잡는 판별기가 됩니다. "
+            "--model 을 2개 이상 지정하세요."
+        )
+
+    done = load_done_ids(args.out)
+    if done:
+        logger.info("기존 출력 %d건을 건너뜁니다: %s", len(done), args.out)
+
+    rng = random.Random(args.seed)
+    # 소재 행을 결정적으로 섞는다. 같은 seed 면 같은 순서가 나온다.
+    order = sorted(range(len(frame)), key=lambda i: _stable_int(f"{args.seed}:{i}"))
+
+    client = None
+    if not args.dry_run:
+        from openai import OpenAI
+
+        api_key = os.environ.get("OPENAI_API_KEY")
+        if not api_key and not args.base_url:
+            raise SystemExit("OPENAI_API_KEY 가 없습니다. --dry-run 으로 먼저 확인하세요.")
+        client = OpenAI(api_key=api_key or "local", base_url=args.base_url, max_retries=2)
+
+    written = skipped_len = failed = 0
+    attempted = 0
+
+    # dry-run 은 파일시스템도 건드리지 않는다. 빈 출력 파일이 남으면 다음 실행이
+    # "이미 돌린 것"으로 오해된다.
+    if args.dry_run:
+        sink = contextlib.nullcontext(None)
+    else:
+        args.out.parent.mkdir(parents=True, exist_ok=True)
+        sink = args.out.open("a", encoding="utf-8")
+
+    with sink as sink:
+        for idx in order:
+            if written >= args.limit:
+                break
+            prompt_id = hashlib.sha1(f"{args.xlsx.name}:{idx}".encode()).hexdigest()[:16]
+            if prompt_id in done:
+                continue
+
+            meta = {c: str(frame[c].iloc[idx]) for c in usable_meta
+                    if isinstance(frame[c].iloc[idx], str) and frame[c].iloc[idx].strip()}
+            if not meta:
+                continue
+
+            target = rng.choice(lengths)
+            style = STYLE_VARIANTS[_stable_int(prompt_id) % len(STYLE_VARIANTS)]
+            model = models[attempted % len(models)]
+            prompt = _build_prompt(meta, target, style)
+            attempted += 1
+
+            if args.dry_run:
+                print(f"\n--- [{written + 1}] model={model} target={target}자 ---")
+                print(prompt)
+                written += 1
+                continue
+
+            try:
+                response = client.chat.completions.create(
+                    model=model,
+                    temperature=args.temperature,
+                    messages=[
+                        {"role": "system", "content": SYSTEM_PROMPT},
+                        {"role": "user", "content": prompt},
+                    ],
+                )
+                text = (response.choices[0].message.content or "").strip()
+            except Exception as exc:  # noqa: BLE001 — 어떤 실패든 기록하고 계속
+                failed += 1
+                logger.warning("생성 실패 (%s, %s): %s", model, prompt_id, exc)
+                continue
+
+            # human 과 같은 정규화를 거쳐야 양쪽 표기가 같은 형태가 된다.
+            text = normalize_ocr(text)
+            if not (low_chars <= len(text) <= high_chars):
+                # 길이가 어긋난 표본은 버린다. 남겨두면 판별기가 길이를 배운다.
+                skipped_len += 1
+                continue
+
+            sink.write(json.dumps({
+                "text": text,
+                "generator": model,
+                "book": "",
+                "prompt_id": prompt_id,
+                "target_chars": target,
+                "char_count": len(text),
+                "style": style,
+                "meta": meta,
+            }, ensure_ascii=False) + "\n")
+            sink.flush()  # 중간에 죽어도 여기까지는 남는다
+            written += 1
+            if written % 50 == 0:
+                logger.info("생성 %d/%d (길이미달 폐기 %d, 실패 %d)",
+                            written, args.limit, skipped_len, failed)
+
+    logger.info(
+        "완료: %d건 생성 / 길이 벗어나 폐기 %d건 / 호출 실패 %d건 → %s",
+        written, skipped_len, failed, args.out,
+    )
+    if args.dry_run:
+        logger.info("dry-run 이라 API 를 호출하지 않았습니다. 비용 0.")
+    elif skipped_len > written * 0.3:
+        logger.warning(
+            "폐기율이 %.0f%% 로 높습니다. 모델이 분량 지시를 안 따르고 있습니다. "
+            "프롬프트의 분량 문구를 조정하거나 다른 모델을 쓰세요.",
+            skipped_len / max(1, skipped_len + written) * 100,
+        )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_generate_ai_samples.py

@@ -0,0 +1,101 @@
+"""generate_ai_samples 단위테스트. API 를 호출하지 않는다."""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parent.parent
+_spec = importlib.util.spec_from_file_location(
+    "generate_ai_samples", ROOT / "scripts" / "generate_ai_samples.py"
+)
+gen = importlib.util.module_from_spec(_spec)
+sys.modules["generate_ai_samples"] = gen
+_spec.loader.exec_module(gen)
+
+
+class TestManuscriptNeverLeaves:
+    """이 프로젝트에서 가장 깨지면 안 되는 성질 — 원고는 외부로 나가지 않는다."""
+
+    def test_meta_columns_contain_no_body_column(self):
+        for column in gen.META_COLUMNS:
+            assert not any(bad in column.lower() for bad in gen.TEXT_COLUMN_BLOCKLIST)
+
+    def test_guard_rejects_body_column(self):
+        """META_COLUMNS 에 본문 컬럼을 넣는 수정은 실행 시점에 막혀야 한다."""
+        with pytest.raises(SystemExit) as err:
+            gen.assert_no_body_columns(["gpt_keyword", "에피소드"])
+        assert "에피소드" in str(err.value)
+
+    def test_guard_passes_clean_columns(self):
+        cols = ["gpt_keyword", "gpt_title"]
+        assert gen.assert_no_body_columns(cols) == cols
+
+    def test_prompt_contains_only_given_meta(self):
+        """프롬프트에 넣지 않은 문자열은 절대 나타나지 않는다."""
+        secret = "이것은 저자의 실제 원고 본문이다"
+        prompt = gen._build_prompt(
+            {"gpt_keyword": "습관", "gpt_title": "변화"}, 950, "담담하게"
+        )
+        assert secret not in prompt
+        assert "습관" in prompt and "변화" in prompt
+
+    def test_build_prompt_signature_has_no_text_param(self):
+        """본문을 넘길 자리 자체가 없어야 한다."""
+        import inspect
+
+        params = set(inspect.signature(gen._build_prompt).parameters)
+        assert params == {"meta", "target_chars", "style"}
+
+
+class TestLengthBand:
+    def test_band_from_distribution(self, tmp_path):
+        import pandas as pd
+
+        frame = pd.DataFrame({"에피소드": ["가" * n for n in range(200, 1200, 10)]})
+        lengths, low, high = gen.load_human_length_band(frame, "에피소드")
+        assert low < high
+        assert min(lengths) >= 120  # 120자 미만은 채점 불가라 제외된다
+        assert low >= min(lengths) and high <= max(lengths)
+
+    def test_rejects_empty_column(self):
+        import pandas as pd
+
+        frame = pd.DataFrame({"에피소드": ["짧음"] * 5})
+        with pytest.raises(SystemExit):
+            gen.load_human_length_band(frame, "에피소드")
+
+
+class TestResume:
+    def test_reads_done_ids(self, tmp_path):
+        out = tmp_path / "s.jsonl"
+        out.write_text(
+            json.dumps({"prompt_id": "aaa", "text": "x"}, ensure_ascii=False) + "\n"
+            + json.dumps({"prompt_id": "bbb", "text": "y"}, ensure_ascii=False) + "\n",
+            encoding="utf-8",
+        )
+        assert gen.load_done_ids(out) == {"aaa", "bbb"}
+
+    def test_missing_file_is_empty(self, tmp_path):
+        assert gen.load_done_ids(tmp_path / "nope.jsonl") == set()
+
+    def test_broken_line_does_not_crash(self, tmp_path):
+        """중간에 죽어 잘린 행이 있어도 재개는 되어야 한다."""
+        out = tmp_path / "s.jsonl"
+        out.write_text('{"prompt_id": "aaa"}\n{"prompt_id": "bro\n', encoding="utf-8")
+        assert gen.load_done_ids(out) == {"aaa"}
+
+
+class TestDiversity:
+    def test_style_variants_are_distinct(self):
+        assert len(set(gen.STYLE_VARIANTS)) == len(gen.STYLE_VARIANTS)
+        assert len(gen.STYLE_VARIANTS) >= 4
+
+    def test_style_assignment_is_deterministic(self):
+        a = gen._stable_int("abc") % len(gen.STYLE_VARIANTS)
+        b = gen._stable_int("abc") % len(gen.STYLE_VARIANTS)
+        assert a == b