#!/usr/bin/env python3 """판정 3자 비교 — grok-4.6 vs gemini-3.1-pro vs gemini-3.7-flash. 배경(2026-08-18 사용자 지시): 08-13 파일럿(artifact/grok46judge23)에 최신 Gemini flash 를 한 자리 더해 다시 본다. **기존 세 모델은 다시 실행하지 않는다** — 그 결과를 그대로 읽어 비교 표에만 담고, 새로 호출하는 것은 gemini-3.7-flash 하나뿐이다. 계약(원본과 같아야 비교가 성립한다): - 지시문 = 원본이 보낸 그대로. artifact/grok46judge23/prompt_sys.txt(관찰), prompt_judge_sys.txt(선정)를 **파일에서 읽어** 쓴다. 다시 조립하지 않는다. - 입력 = 같은 레시피 디렉토리의 같은 PNG·같은 롤 프롬프트 전문(무절단). - 참조 이미지 미첨부(원본과 동일), 수정 게이트 = critical 만. - 선정 축 = 정·역 2회(첨부 순서와 라벨을 함께 뒤집어 위치 편향 상쇄). - maxOutputTokens 는 원본 gemini 와 같은 값(관찰 4096 / 선정 8192). 프로덕션 무접촉: 읽기만 한다. records/체크포인트/DB 를 쓰지 않고, 산출은 새 artifact 디렉토리에만 만든다. 기존 grok46judge23 은 건드리지 않는다. 사용: .venv/bin/python judge3_flash_pilot.py # 새 모델 실행 + 갤러리 .venv/bin/python judge3_flash_pilot.py --html # 갤러리만 재생성 """ from __future__ import annotations import argparse import base64 import html as H import json import re import time import urllib.error import urllib.request from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone from pathlib import Path ROOT = Path(__file__).resolve().parent PROJ = "c7e3b2e7-c545-4516-93b2-62a51a74d794" EPI = "7c902020-4451-4967-9eb6-1e53c2b9b717" RECIPE = ROOT.parent / "projects" / PROJ / "images" / EPI / "scene" / "recipe" SRC = ROOT.parent / "artifact" / "grok46judge23" # 읽기 전용 OUT = ROOT.parent / "artifact" / "20260818_judge3_flash" # 새 산출 IMG_BASE = "../grok46judge23/img" # 이미지 재사용 NEW_KEY = "gflash" NEW_MODEL = "gemini-3.7-flash" # 표에 세울 세 자리 (사용자 지시) + 참고로만 남기는 qwen COLS = ("grok46", "gemini", NEW_KEY) LABEL = { "grok46": "grok-4.6", "gemini": "gemini-3.1-pro", NEW_KEY: "gemini-3.7-flash (신규)", "qwen": "qwen3.8-max", } MAX_WORKERS = 4 # 주행 중이라 넉넉히 잡지 않는다 RETRY_STATUS = (429, 500, 502, 503, 504) def env(key: str) -> str: for line in (ROOT / ".env").read_text(encoding="utf-8").splitlines(): if line.startswith(key + "="): return line.split("=", 1)[1].strip() return "" def tag_key(tag: str): m = re.match(r"S(\d+)sh(\d+)$", tag) return (int(m.group(1)), int(m.group(2))) def lenient_json(text: str) -> dict: t = (text or "").strip() t = re.sub(r"^```(?:json)?\s*|\s*```$", "", t) m = re.search(r"\{.*\}", t, re.S) if not m: raise ValueError(f"JSON 없음: {t[:200]!r}") return json.loads(m.group(0)) def call_gemini(model: str, sys_text: str, parts: list, max_out: int): """전송 계층 오류(429/5xx)만 한 번 다시 보낸다. 주행과 같은 시각에 도니 몰림으로 생긴 거절을 모델 능력으로 세지 않기 위한 것이다. 응답 내용 오류(JSON 파손 등)는 다시 보내지 않는다. """ body = { "systemInstruction": {"parts": [{"text": sys_text}]}, "contents": [{"role": "user", "parts": parts}], "generationConfig": {"response_mime_type": "application/json", "maxOutputTokens": max_out}, } url = (f"https://generativelanguage.googleapis.com/v1beta/models/" f"{model}:generateContent?key={env('GEMINI_API_KEY')}") last = None for attempt in range(2): req = urllib.request.Request( url, data=json.dumps(body).encode(), headers={"Content-Type": "application/json"}, method="POST") try: with urllib.request.urlopen(req, timeout=600) as r: d = json.loads(r.read().decode()) break except urllib.error.HTTPError as exc: last = exc if exc.code in RETRY_STATUS and attempt == 0: time.sleep(20) continue detail = "" try: detail = exc.read().decode()[:300] except Exception: # noqa: BLE001 pass raise RuntimeError(f"HTTP {exc.code}: {detail}") from exc else: raise RuntimeError(f"재시도 후에도 실패: {last}") cands = d.get("candidates") or [] if not cands: raise RuntimeError(f"후보 없음: {json.dumps(d, ensure_ascii=False)[:250]}") c = cands[0] finish = c.get("finishReason") text = "".join(p.get("text", "") for p in ((c.get("content") or {}).get("parts") or [])) if finish and finish not in ("STOP",): raise RuntimeError(f"finishReason={finish} (출력 {len(text)}자)") return text, d.get("usageMetadata") # ── 관찰 축 ────────────────────────────────────────────────────────── def run_observe(results: dict, tags: list[str], calls: list) -> None: sys_text = (SRC / "prompt_sys.txt").read_text(encoding="utf-8") records = json.loads((RECIPE / "records.json").read_text("utf-8")) todo = [t for t in tags if not (isinstance((results.get(t) or {}).get(NEW_KEY), dict) and "observations" in results[t][NEW_KEY])] print(f"관찰 축 — 새로 부를 샷 {len(todo)}/{len(tags)}", flush=True) if not todo: return def _one(tag: str) -> dict: rec = records.get(tag) or {} sel = rec.get("selected") or "" prompt = ((rec.get("roll_prompts") or {}).get(sel) or rec.get("prompt") or "") img = base64.b64encode( (RECIPE / f"{tag}_sel.png").read_bytes()).decode() parts = [ {"text": "THE PROMPT (the candidate was generated from this):" "\n\n" + prompt}, {"text": "PHOTOGRAPH TO JUDGE:"}, {"inline_data": {"mime_type": "image/png", "data": img}}, ] t0 = time.monotonic() recd = {"ts": datetime.now(timezone.utc).isoformat(), "tag": tag, "model": NEW_KEY, "kind": "observe", "prompt_chars": len(prompt)} try: text, usage = call_gemini(NEW_MODEL, sys_text, parts, 4096) recd["latency_s"] = round(time.monotonic() - t0, 1) recd["usage"] = usage obs = lenient_json(text).get("observations") if not isinstance(obs, list): raise ValueError("observations 배열 아님") recd["parsed"] = {"observations": obs, "latency_s": recd["latency_s"]} except Exception as exc: # noqa: BLE001 — 실측 기록 recd["latency_s"] = round(time.monotonic() - t0, 1) recd["error"] = f"{type(exc).__name__}: {exc}"[:400] return recd with ThreadPoolExecutor(max_workers=MAX_WORKERS) as ex: for i, recd in enumerate(ex.map(_one, todo), 1): tag = recd["tag"] node = results.setdefault(tag, {}) if "parsed" in recd: node[NEW_KEY] = recd.pop("parsed") print(f"[{i}/{len(todo)}] {tag} 관찰 ok " f"{recd['latency_s']}s " f"obs={len(node[NEW_KEY]['observations'])}", flush=True) else: node[NEW_KEY] = {"error": recd["error"]} print(f"[{i}/{len(todo)}] {tag} 관찰 FAIL {recd['error']}", flush=True) calls.append(recd) # ── 선정 축 ────────────────────────────────────────────────────────── def run_select(results: dict, tags: list[str], calls: list) -> None: sys_text = (SRC / "prompt_judge_sys.txt").read_text(encoding="utf-8") records = json.loads((RECIPE / "records.json").read_text("utf-8")) jobs = [] for tag in tags: a_p, b_p = RECIPE / f"{tag}_a.png", RECIPE / f"{tag}_b.png" if not (a_p.is_file() and b_p.is_file()): print(f"{tag}: 롤 a/b 파일 부재 — 선정 축 건너뜀") continue cur = (results.get(tag) or {}).get(f"sel_{NEW_KEY}") or {} for pno in (1, 2): p = cur.get(f"pass{pno}") if isinstance(p, dict) and "winner_roll" in p: continue jobs.append((tag, pno)) print(f"선정 축 — 새로 부를 왕복 {len(jobs)} (샷 {len(tags)} × 정·역 2)", flush=True) if not jobs: return cache: dict[str, tuple[str, str]] = {} def b64(tag: str): if tag not in cache: cache[tag] = ( base64.b64encode((RECIPE / f"{tag}_a.png").read_bytes()).decode(), base64.b64encode((RECIPE / f"{tag}_b.png").read_bytes()).decode()) return cache[tag] def _one(job) -> dict: tag, pno = job a64, b64_ = b64(tag) first, second, back = ((a64, b64_, {"A": "a", "B": "b"}) if pno == 1 else (b64_, a64, {"A": "b", "B": "a"})) prompt = (records.get(tag) or {}).get("prompt") or "" parts = [ {"text": "THE PROMPT (all candidates were generated from this):" "\n\n" + prompt}, {"text": "CANDIDATE A:"}, {"inline_data": {"mime_type": "image/png", "data": first}}, {"text": "CANDIDATE B:"}, {"inline_data": {"mime_type": "image/png", "data": second}}, ] t0 = time.monotonic() recd = {"ts": datetime.now(timezone.utc).isoformat(), "tag": tag, "model": NEW_KEY, "kind": f"select_pass{pno}", "prompt_chars": len(prompt)} try: text, usage = call_gemini(NEW_MODEL, sys_text, parts, 8192) recd["latency_s"] = round(time.monotonic() - t0, 1) recd["usage"] = usage parsed = lenient_json(text) w = str(parsed.get("winner") or "").strip().upper() if w not in back: raise ValueError(f"winner 라벨 불명: {w!r}") recd["parsed"] = { "winner_label": w, "winner_roll": back[w], "scores": parsed.get("scores"), "hard_violations": parsed.get("hard_violations"), "verdict_ko": parsed.get("verdict_ko"), "all_candidates_fail": parsed.get("all_candidates_fail"), "latency_s": recd["latency_s"]} except Exception as exc: # noqa: BLE001 — 실측 기록 recd["latency_s"] = round(time.monotonic() - t0, 1) recd["error"] = f"{type(exc).__name__}: {exc}"[:400] return recd with ThreadPoolExecutor(max_workers=MAX_WORKERS) as ex: for i, recd in enumerate(ex.map(_one, jobs), 1): tag = recd["tag"] pno = recd["kind"][-1] slot = results.setdefault(tag, {}).setdefault( f"sel_{NEW_KEY}", {}) if "parsed" in recd: slot[f"pass{pno}"] = recd.pop("parsed") print(f"[{i}/{len(jobs)}] {tag} 선정 p{pno} ok " f"{recd['latency_s']}s → 롤 " f"{slot[f'pass{pno}']['winner_roll']}", flush=True) else: slot[f"pass{pno}"] = {"error": recd["error"]} print(f"[{i}/{len(jobs)}] {tag} 선정 p{pno} FAIL " f"{recd['error']}", flush=True) calls.append(recd) for tag in tags: slot = (results.get(tag) or {}).get(f"sel_{NEW_KEY}") if not slot: continue r1 = (slot.get("pass1") or {}).get("winner_roll") r2 = (slot.get("pass2") or {}).get("winner_roll") slot["pick"] = (r1 if r1 and r1 == r2 else ("split" if r1 and r2 else None)) # ── 갤러리 ─────────────────────────────────────────────────────────── SEV_ORDER = {"critical": 0, "major": 1, "minor": 2} def esc(x): return H.escape(str(x if x is not None else "")) def build_html(results: dict) -> None: tags = sorted(results, key=tag_key) all_keys = COLS + ("qwen",) stats = {m: {"obs": 0, "critical": 0, "major": 0, "minor": 0, "crit_shots": set(), "fail": 0, "lat": []} for m in all_keys} sel_stats = {m: {"match": 0, "diff": 0, "split": 0, "none": 0} for m in all_keys} # 통계는 네 모델 모두 모은다 (qwen 은 참고 줄로만 보여 준다) for tag in tags: node = results[tag] for mk in all_keys: r = node.get(mk) or {} if not r or "error" in r: if r: stats[mk]["fail"] += 1 continue obs = r.get("observations") or [] stats[mk]["obs"] += len(obs) if r.get("latency_s") is not None: stats[mk]["lat"].append(r["latency_s"]) for o in obs: sv = str(o.get("severity") or "") if sv in SEV_ORDER: stats[mk][sv] += 1 if sv == "critical": stats[mk]["crit_shots"].add(tag) prod_of = {t: str((results[t].get("prod_selected") or "")).lower() for t in tags} for tag in tags: for mk in all_keys: s = results[tag].get(f"sel_{mk}") if not s: continue pick = s.get("pick") prod = prod_of[tag] if pick in ("a", "b"): sel_stats[mk]["match" if prod and pick == prod else "diff"] += 1 elif pick == "split": sel_stats[mk]["split"] += 1 else: sel_stats[mk]["none"] += 1 n = len(tags) n_sel = sum(1 for t in tags if any(results[t].get(f"sel_{m}") for m in all_keys)) def med(v): return f"{sorted(v)[len(v)//2]}s" if v else "-" def sumrow(mk, dim=False): s = stats[mk] cls = " class='dim'" if dim else "" return (f"{esc(LABEL[mk])}" f"{' 참고' if dim else ''}" f"{s['obs']}{s['critical']}" f"{s['major']}{s['minor']}" f"{len(s['crit_shots'])}/{n}{s['fail']}" f"{med(s['lat'])}") def selrow(mk, dim=False): s = sel_stats[mk] cls = " class='dim'" if dim else "" return (f"{esc(LABEL[mk])}" f"{' 참고' if dim else ''}" f"{s['match']}/{n_sel}{s['diff']}" f"{s['split']}{s['none']}") def agree(a, b): return sum(1 for t in tags if (t in stats[a]["crit_shots"]) == (t in stats[b]["crit_shots"])) def sel_agree(a, b): ok = [t for t in tags if (results[t].get(f"sel_{a}") or {}).get("pick") in ("a", "b") and (results[t].get(f"sel_{b}") or {}).get("pick") in ("a", "b")] same = sum(1 for t in ok if results[t][f"sel_{a}"]["pick"] == results[t][f"sel_{b}"]["pick"]) return f"{same}/{len(ok)}" pairs = (("grok46", "gemini"), ("grok46", NEW_KEY), ("gemini", NEW_KEY)) summary = ( "

관찰 축 — 결함 찾기

" "" "" "" + "".join(sumrow(m) for m in COLS) + sumrow("qwen", True) + "
모델관찰 총계criticalmajorminor수정 대상 샷(critical≥1)실패중앙 지연
" "

샷 단위 수정-대상 일치(critical 유무 기준): " + " · ".join(f"{LABEL[a].split(' ')[0]}↔{LABEL[b].split(' ')[0]} " f"{agree(a, b)}/{n}" for a, b in pairs) + "

" "

선정 축 — 2롤(a/b) 승자 재판정 (정·역 2회 수렴 기준)

" "" "" "" + "".join(selrow(m) for m in COLS) + selrow("qwen", True) + "
모델프로덕션과 일치프로덕션과 반대정·역 불일치(split)판정 실패
" "

두 모델이 모두 수렴한 샷에서 서로 같은 롤을 고른 비율: " + " · ".join(f"{LABEL[a].split(' ')[0]}↔{LABEL[b].split(' ')[0]} " f"{sel_agree(a, b)}" for a, b in pairs) + "

" "

계약: 08-13 파일럿과 같은 지시문 파일·같은 PNG·" "같은 롤 프롬프트 전문. 관찰=프로덕션 관찰 스템(judge v10) 원문" "+JSON 절, 선정=judge_still v7 원문(4축 readings·hard 위반·0-10 " "점수). 참조 이미지는 전 모델 공통 미첨부(프로덕션 관찰과의 차이). " "수정 게이트=critical 만. 정·역=첨부 순서와 라벨을 함께 뒤집어 위치 " "편향 상쇄 — 두 회가 같은 롤이어야 수렴.

" "

이번에 새로 호출한 것은 gemini-3.7-flash 하나" f"({datetime.now().strftime('%Y-%m-%d')}). grok-4.6·gemini-3.1-pro" "·qwen3.8-max 수치는 08-13 파일럿 결과를 다시 실행하지 않고 그대로 " "옮긴 것이다. 프로덕션 선정은 참조 첨부·합의 조건이 달라 '정답'이 " "아니라 대조점이다.

") rows = [] for tag in tags: node = results[tag] cols = [] for mk in COLS: r = node.get(mk) or {} if not r: cols.append("기록 없음") continue if "error" in r: cols.append(f"실패: {esc(r['error'])}") continue obs = sorted(r.get("observations") or [], key=lambda o: SEV_ORDER.get(str(o.get("severity")), 9)) li = "".join( f"
  • " f"[{esc(o.get('severity'))}] {esc(o.get('issue_ko'))}" f"
    • " for o in obs) or "
  • 이슈 없음
    • " fix = "수정 대상" if any( o.get("severity") == "critical" for o in obs) else "통과" cols.append( f"
    " f"{fix}
    " f"
    {esc(r.get('latency_s'))}s
    ") prod = prod_of[tag] sel_html = "" if any(node.get(f"sel_{m}") for m in COLS): cells = [] for mk in COLS: s = node.get(f"sel_{mk}") or {} pick = s.get("pick") if pick in ("a", "b"): klass = "agree" if prod and pick == prod else "diff" label = f"롤 {pick}" + (" =프로덕션" if klass == "agree" else " ≠프로덕션" if prod else "") elif pick == "split": klass, label = "split", "정·역 불일치" elif s: klass, label = "err", "판정 실패" else: klass, label = "err", "기록 없음" det = [] for pno in ("pass1", "pass2"): p = s.get(pno) or {} if "winner_roll" in p: det.append(f"{pno}: 롤 {p['winner_roll']} · 점수 " f"{esc(p.get('scores'))} · " f"{esc(p.get('verdict_ko') or {})}") elif "error" in p: det.append(f"{pno}: 실패 {esc(p['error'])}") cells.append( f"
    {esc(label)}
    " "
    정·역 상세
    "
                    + esc("\n\n".join(det)) + "
    ") sel_html = ( "
    " f"
    " f"
    롤 a" f"{' (프로덕션 선정)' if prod == 'a' else ''}
    " "
    " f"
    " f"
    롤 b" f"{' (프로덕션 선정)' if prod == 'b' else ''}
    " "
    " "" + "".join(f"" for m in COLS) + f"" + "".join(cells) + "
    선정{esc(LABEL[m])}
    프로덕션: 롤 {esc(prod) or '?'}
    ") rows.append( f"

    {tag} 선정 " f"{esc(node.get('selected'))}

    " f"
    " f"" "" + "".join(f"" for m in COLS) + "" + "".join(cols) + "
    {esc(LABEL[m])}
    " + sel_html) html = ( '\n' "판정 3자 비교 — grok-4.6 vs gemini-3.1-pro vs " "gemini-3.7-flash\n\n" "

    판정 3자 비교 — grok-4.6 · gemini-3.1-pro · gemini-3.7-flash

    " f"\n

    사랑했지만 1화 {n}샷 · 08-13 파일럿과 같은 입력·" "같은 지시문

    \n" + summary + "\n".join(rows)) OUT.mkdir(parents=True, exist_ok=True) (OUT / "index.html").write_text(html, encoding="utf-8") print(f"갤러리: {OUT / 'index.html'}") def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--html", action="store_true", help="갤러리만 재생성") args = ap.parse_args() OUT.mkdir(parents=True, exist_ok=True) res_path = OUT / "results.json" if res_path.is_file(): results = json.loads(res_path.read_text("utf-8")) else: # 기존 결과를 그대로 복사해 온다 (원본은 읽기만 한다) results = json.loads((SRC / "results.json").read_text("utf-8")) tags = sorted(results, key=tag_key) if not args.html: calls_path = OUT / "calls.json" calls = (json.loads(calls_path.read_text("utf-8")) if calls_path.is_file() else []) try: run_observe(results, tags, calls) run_select(results, tags, calls) finally: calls_path.write_text( json.dumps(calls, ensure_ascii=False, indent=1), "utf-8") res_path.write_text( json.dumps(results, ensure_ascii=False, indent=1), "utf-8") build_html(results) if __name__ == "__main__": main()