#!/usr/bin/env python3 """grok-4.6 관찰 판정 파일럿 — 최종 스틸 3자(grok/qwen/gemini) 비교. 배경(2026-08-13 사용자 지시): OpenRouter 에 오늘 올라온 x-ai/grok-4.6 (vision)을 판정자로 실측 — 재실행 23샷 선정본을 관찰 계약으로 판정시키고 프로덕션 판정 쌍(Qwen 메인·Gemini 관찰)과 같은 입력·같은 계약으로 비교. 계약: - 관찰 문안 = 프로덕션 관찰 스템(multiroll_judge v10 gq_observe_sys.md) **원문 그대로**(SOT) + JSON 출력 절만 부가. 수정 지시는 관찰의 몫이 아니고, 프로덕션 수정 게이트=critical 만 fix — 비교도 그 축으로 본다. - 입력 = 선정본 스틸 + 선정 롤 프롬프트 전문(무절단). 참조 이미지는 세 모델 모두 미첨부(대칭) — 프로덕션 관찰과의 차이로 명시 기록. - 프로덕션 무접촉: records/CP/DB 무변경, 산출은 artifact/ 전용. - 재개 안전: results.json 에 있는 (모델,샷)은 건너뜀. 실패도 기록. 사용: .venv/bin/python grok46_judge_pilot.py # 판정 실행(재개 가능) .venv/bin/python grok46_judge_pilot.py --html # 갤러리만 재생성 """ from __future__ import annotations import argparse import base64 import html as H import json import re import time import urllib.request 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" OBSERVE_STEM = (ROOT.parent / "prompts" / "_base" / "multiroll_judge" / "10.202608131121" / "gq_observe_sys.md") OUT = ROOT.parent / "artifact" / "grok46judge23" JSON_TAIL = """ OUTPUT FORMAT: reply with ONLY a JSON object, no prose, no code fence: {"observations": [{"issue_ko": "", "severity": "critical|major|minor"}]} If nothing is wrong return {"observations": []}.""" # 선정 판정 = 프로덕션 judge_still 계약 원문(팩 v7 스템, STILL_JUDGE_PACK # _VERSION="7") — {count_word}/{label_list} 만 치환, JSON 절 부가. JUDGE_STEM = (ROOT.parent / "prompts" / "_base" / "multiroll_judge" / "7.202608071100" / "judge_still.md") JUDGE_JSON_TAIL = """ OUTPUT FORMAT: reply with ONLY a JSON object, no prose, no code fence: {"readings": {"A": {"direction": "...", "built_space": "...", "entities": "...", "physics": "..."}, "B": {"direction": "...", "built_space": "...", "entities": "...", "physics": "..."}}, "hard_violations": {"A": ["..."], "B": ["..."]}, "scores": {"A": 0, "B": 0}, "verdict_ko": {"A": "...", "B": "..."}, "winner": "A", "all_candidates_fail": false}""" MODELS = ("grok46", "qwen", "gemini") 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 post_json(url: str, body: dict, headers: dict, timeout: int = 600): req = urllib.request.Request( url, data=json.dumps(body).encode(), headers=headers, method="POST") with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read().decode()) def call_openai_style(url: str, key: str, model: str, sys_text: str, prompt: str, img_b64: str, max_tokens: int): body = { "model": model, "messages": [ {"role": "system", "content": sys_text}, {"role": "user", "content": [ {"type": "text", "text": "THE PROMPT (the candidate was generated from this):\n\n" + prompt}, {"type": "text", "text": "PHOTOGRAPH TO JUDGE:"}, {"type": "image_url", "image_url": {"url": "data:image/png;base64," + img_b64}}, ]}, ], "max_tokens": max_tokens, } d = post_json(url, body, { "Authorization": f"Bearer {key}", "Content-Type": "application/json"}) if d.get("error"): raise RuntimeError(json.dumps(d["error"], ensure_ascii=False)[:300]) msg = (d.get("choices") or [{}])[0].get("message", {}) return msg.get("content") or "", d.get("usage") def call_gemini(key: str, model: str, sys_text: str, prompt: str, img_b64: str): body = { "systemInstruction": {"parts": [{"text": sys_text}]}, "contents": [{"role": "user", "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_b64}}, ]}], "generationConfig": {"response_mime_type": "application/json", "maxOutputTokens": 4096}, } d = post_json( f"https://generativelanguage.googleapis.com/v1beta/models/{model}" f":generateContent?key={key}", body, {"Content-Type": "application/json"}) cands = d.get("candidates") or [] parts = (cands[0].get("content") or {}).get("parts") if cands else None text = "".join(p.get("text", "") for p in (parts or [])) return text, d.get("usageMetadata") def judge_one(model_key: str, sys_text: str, prompt: str, img_b64: str): if model_key == "grok46": return call_openai_style( "https://openrouter.ai/api/v1/chat/completions", env("OPENROUTER_API_KEY"), "x-ai/grok-4.6", sys_text, prompt, img_b64, max_tokens=8000) if model_key == "qwen": return call_openai_style( env("DASHSCOPE_BASE_URL").rstrip("/") + "/chat/completions", env("DASHSCOPE_API_KEY"), env("QWEN_VLM_MODEL") or "qwen3.8-max", sys_text, prompt, img_b64, max_tokens=4000) if model_key == "gemini": return call_gemini( env("GEMINI_API_KEY"), "gemini-3.1-pro-preview", sys_text, prompt, img_b64) raise ValueError(model_key) # ── 선정 판정 (2후보 A/B) ──────────────────────────────────────────── def _select_user_parts_openai(prompt: str, first_b64: str, second_b64: str): return [ {"type": "text", "text": "THE PROMPT (all candidates were generated from this):\n\n" + prompt}, {"type": "text", "text": "CANDIDATE A:"}, {"type": "image_url", "image_url": {"url": "data:image/png;base64," + first_b64}}, {"type": "text", "text": "CANDIDATE B:"}, {"type": "image_url", "image_url": {"url": "data:image/png;base64," + second_b64}}, ] def select_one(model_key: str, sys_text: str, prompt: str, first_b64: str, second_b64: str): """라벨 A=첫째 첨부, B=둘째 첨부 — 호출자가 순서로 정·역을 만든다.""" if model_key in ("grok46", "qwen"): url = ("https://openrouter.ai/api/v1/chat/completions" if model_key == "grok46" else env("DASHSCOPE_BASE_URL").rstrip("/") + "/chat/completions") key = (env("OPENROUTER_API_KEY") if model_key == "grok46" else env("DASHSCOPE_API_KEY")) model = ("x-ai/grok-4.6" if model_key == "grok46" else env("QWEN_VLM_MODEL") or "qwen3.8-max") body = { "model": model, "messages": [ {"role": "system", "content": sys_text}, {"role": "user", "content": _select_user_parts_openai( prompt, first_b64, second_b64)}, ], # 선정 출력(2후보×4축 readings+verdict)은 관찰보다 훨씬 길다 — # 1차 실측에서 qwen 4000 이 JSON 을 중간에 자름(비JSON 14건). "max_tokens": 8000, } if model_key == "qwen": # 2차 실측: 상한을 올려도 문자열 안 따옴표 파손 5건 잔존 — # DashScope JSON 모드로 문법 강제(프로덕션 스키마 강제 동류). body["response_format"] = {"type": "json_object"} d = post_json(url, body, { "Authorization": f"Bearer {key}", "Content-Type": "application/json"}) if d.get("error"): raise RuntimeError( json.dumps(d["error"], ensure_ascii=False)[:300]) msg = (d.get("choices") or [{}])[0].get("message", {}) return msg.get("content") or "", d.get("usage") if model_key == "gemini": body = { "systemInstruction": {"parts": [{"text": sys_text}]}, "contents": [{"role": "user", "parts": [ {"text": "THE PROMPT (all candidates were generated from " "this):\n\n" + prompt}, {"text": "CANDIDATE A:"}, {"inline_data": {"mime_type": "image/png", "data": first_b64}}, {"text": "CANDIDATE B:"}, {"inline_data": {"mime_type": "image/png", "data": second_b64}}, ]}], "generationConfig": {"response_mime_type": "application/json", "maxOutputTokens": 8192}, } d = post_json( "https://generativelanguage.googleapis.com/v1beta/models/" "gemini-3.1-pro-preview:generateContent?key=" + env("GEMINI_API_KEY"), body, {"Content-Type": "application/json"}) cands = d.get("candidates") or [] parts = ((cands[0].get("content") or {}).get("parts") if cands else None) return ("".join(p.get("text", "") for p in (parts or [])), d.get("usageMetadata")) raise ValueError(model_key) def run_select(scenes: set[int]) -> None: """선정 축 — 각 샷의 롤 a/b 를 정·역(첨부 순서=라벨) 2회 판정. pass1: 라벨 A=롤a·B=롤b / pass2: 라벨 A=롤b·B=롤a — 승자 라벨을 롤 정체로 되돌려 위치·라벨 편향을 상쇄(프로덕션 judge_flip 관례). 비교 기준 = records[tag].selected(프로덕션 선정, fix 이전 A/B). """ records = json.loads((RECIPE / "records.json").read_text("utf-8")) sys_text = (JUDGE_STEM.read_text(encoding="utf-8") .replace("{count_word}", "TWO") .replace("{label_list}", "A, B") + JUDGE_JSON_TAIL) OUT.mkdir(parents=True, exist_ok=True) (OUT / "prompt_judge_sys.txt").write_text(sys_text, encoding="utf-8") res_path = OUT / "results.json" results = (json.loads(res_path.read_text("utf-8")) if res_path.is_file() else {}) calls_path = OUT / "calls.json" calls = (json.loads(calls_path.read_text("utf-8")) if calls_path.is_file() else []) tags = sorted( (p.name[:-len("_sel.png")] for p in RECIPE.glob("S*_sel.png") if tag_key(p.name[:-len("_sel.png")])[0] in scenes), key=tag_key) jobs_of = {} 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 jobs_of[tag] = ( base64.b64encode(a_p.read_bytes()).decode(), base64.b64encode(b_p.read_bytes()).decode()) print(f"선정 축 대상 {len(jobs_of)}샷 × {len(MODELS)}모델 × 2회") from concurrent.futures import ThreadPoolExecutor for i, tag in enumerate([t for t in tags if t in jobs_of], 1): rec = records.get(tag) or {} prompt = rec.get("prompt") or "" a64, b64 = jobs_of[tag] node = results.setdefault(tag, {}) node.setdefault("prod_selected", rec.get("selected")) jobs = [] for mk in MODELS: cur = node.get(f"sel_{mk}") or {} for pno, (f64, s64, back) in ( (1, (a64, b64, {"A": "a", "B": "b"})), (2, (b64, a64, {"A": "b", "B": "a"}))): if isinstance(cur.get(f"pass{pno}"), dict) \ and "winner_roll" in cur[f"pass{pno}"]: continue # 재개 jobs.append((mk, pno, f64, s64, back)) if not jobs: continue def _one(job): mk, pno, f64, s64, back = job t0 = time.monotonic() recd = {"ts": datetime.now(timezone.utc).isoformat(), "tag": tag, "model": mk, "kind": f"select_pass{pno}", "prompt_chars": len(prompt)} try: text, usage = select_one(mk, sys_text, prompt, f64, s64) 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=len(jobs)) as ex: recs = list(ex.map(_one, jobs)) for recd in recs: mk = recd["model"] pno = recd["kind"][-1] slot = node.setdefault(f"sel_{mk}", {}) if "parsed" in recd: slot[f"pass{pno}"] = recd.pop("parsed") print(f"[{i}] {tag} {mk} p{pno} ok {recd['latency_s']}s " f"→ 롤 {slot[f'pass{pno}']['winner_roll']}", flush=True) else: slot[f"pass{pno}"] = {"error": recd["error"]} print(f"[{i}] {tag} {mk} p{pno} FAIL {recd['error']}", flush=True) calls.append(recd) for mk in MODELS: slot = node.get(f"sel_{mk}") or {} 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)) 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") def run(scenes: set[int]) -> None: records = json.loads((RECIPE / "records.json").read_text("utf-8")) sys_text = OBSERVE_STEM.read_text(encoding="utf-8") + JSON_TAIL OUT.mkdir(parents=True, exist_ok=True) (OUT / "prompt_sys.txt").write_text(sys_text, encoding="utf-8") res_path = OUT / "results.json" results = (json.loads(res_path.read_text("utf-8")) if res_path.is_file() else {}) calls_path = OUT / "calls.json" calls = (json.loads(calls_path.read_text("utf-8")) if calls_path.is_file() else []) tags = sorted( (p.name[:-len("_sel.png")] for p in RECIPE.glob("S*_sel.png") if tag_key(p.name[:-len("_sel.png")])[0] in scenes), key=tag_key) print(f"대상 {len(tags)}샷 × {len(MODELS)}모델") for i, tag in enumerate(tags, 1): rec = records.get(tag) or {} sel = rec.get("selected") or "" rp = rec.get("roll_prompts") or {} # 관찰 프롬프트 = 선정 롤 프롬프트 전문(프로덕션 # critique_selected_prompt_only 계약) — 무절단. prompt = rp.get(sel) or rec.get("prompt") or "" if not prompt: print(f"[{i}] {tag} 프롬프트 없음 — 건너뜀") continue img_b64 = base64.b64encode( (RECIPE / f"{tag}_sel.png").read_bytes()).decode() node = results.setdefault(tag, {"selected": sel}) todo = [mk for mk in MODELS if not (isinstance(node.get(mk), dict) and "observations" in node[mk])] if not todo: continue # 재개 — 샷 전체 판정됨 def _one(mk: str) -> dict: t0 = time.monotonic() recd = {"ts": datetime.now(timezone.utc).isoformat(), "tag": tag, "model": mk, "prompt_chars": len(prompt)} try: text, usage = judge_one(mk, sys_text, prompt, img_b64) recd["latency_s"] = round(time.monotonic() - t0, 1) recd["usage"] = usage parsed = lenient_json(text) obs = parsed.get("observations") if not isinstance(obs, list): raise ValueError("observations 배열 아님") recd["parsed"] = {"observations": obs, "latency_s": recd["latency_s"]} recd["n_obs"] = len(obs) except Exception as exc: # noqa: BLE001 — 실측 기록 recd["latency_s"] = round(time.monotonic() - t0, 1) recd["error"] = f"{type(exc).__name__}: {exc}"[:400] return recd # 샷당 3모델 병렬 — 걸린 시간=가장 느린 모델(직렬 ~7분/샷을 줄인다). # 파일 쓰기는 메인 스레드에서만(원자성 단순 유지). from concurrent.futures import ThreadPoolExecutor with ThreadPoolExecutor(max_workers=len(todo)) as ex: recs = list(ex.map(_one, todo)) for recd in recs: mk = recd["model"] if "parsed" in recd: node[mk] = recd.pop("parsed") print(f"[{i}/{len(tags)}] {tag} {mk} ok " f"{recd['latency_s']}s obs={recd['n_obs']}", flush=True) else: node[mk] = {"error": recd["error"]} print(f"[{i}/{len(tags)}] {tag} {mk} FAIL {recd['error']}", flush=True) calls.append(recd) 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") SEV_ORDER = {"critical": 0, "major": 1, "minor": 2} def build_html() -> None: results = json.loads((OUT / "results.json").read_text("utf-8")) tags = sorted(results, key=tag_key) (OUT / "img").mkdir(exist_ok=True) import shutil for tag in tags: for suffix in ("_sel", "_a", "_b"): src = RECIPE / f"{tag}{suffix}.png" dst = OUT / "img" / f"{tag}{suffix}.png" if src.is_file() and not dst.is_file(): shutil.copy(src, dst) def esc(x): return H.escape(str(x if x is not None else "")) stats = {m: {"obs": 0, "critical": 0, "major": 0, "minor": 0, "crit_shots": set(), "fail": 0, "lat": []} for m in MODELS} sel_stats = {m: {"match": 0, "diff": 0, "split": 0, "none": 0} for m in MODELS} rows = [] for tag in tags: node = results[tag] cols = [] for mk in MODELS: r = node.get(mk) or {} if "error" in r: stats[mk]["fail"] += 1 cols.append(f"실패: {esc(r['error'])}") continue obs = sorted( r.get("observations") or [], key=lambda o: SEV_ORDER.get(str(o.get("severity")), 9)) stats[mk]["obs"] += len(obs) stats[mk]["lat"].append(r.get("latency_s") or 0) for o in obs: sv = str(o.get("severity") or "") if sv in ("critical", "major", "minor"): stats[mk][sv] += 1 if sv == "critical": stats[mk]["crit_shots"].add(tag) 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
    ") # ── 선정 축 (있을 때만) ── sel_html = "" prod = str(node.get("prod_selected") or "").lower() if any(node.get(f"sel_{m}") for m in MODELS): pick_cells = [] for mk in MODELS: 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 "") sel_stats[mk]["match" if klass == "agree" else "diff"] += 1 elif pick == "split": klass, label = "split", "정·역 불일치" sel_stats[mk]["split"] += 1 else: klass, label = "err", "판정 실패" sel_stats[mk]["none"] += 1 det_v = [] for pno in ("pass1", "pass2"): p = s.get(pno) or {} if "winner_roll" in p: det_v.append( f"{pno}: 롤 {p['winner_roll']} · 점수 " f"{esc(p.get('scores'))} · " f"{esc((p.get('verdict_ko') or {}))}") elif "error" in p: det_v.append(f"{pno}: 실패 {esc(p['error'])}") pick_cells.append( f"
    {esc(label)}
    " "
    정·역 상세
    "
                    + esc("\n\n".join(det_v)) + "
    ") sel_html = ( "
    " f"
    " f"
    롤 a{' (프로덕션 선정)' if prod == 'a' else ''}" "
    " f"
    " f"
    롤 b{' (프로덕션 선정)' if prod == 'b' else ''}" "
    " "" + "".join(f"" for m in MODELS) + f"" + "".join(pick_cells) + "
    선정{m}
    프로덕션: 롤 {esc(prod) or '?'}
    ") rows.append( f"

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

    " f"
    " f"" + "".join(f"" for m in ("grok-4.6", "qwen3.8-max(프로덕션 메인)", "gemini-3.1-pro(프로덕션 관찰)")) + "" + "".join(cols) + "
    {m}
    " + sel_html) n = len(tags) crit_sets = {m: stats[m]["crit_shots"] for m in MODELS} def agree(a, b): return sum(1 for t in tags if (t in crit_sets[a]) == (t in crit_sets[b])) summary = ( "" "" "" + "".join( f"" f"" f"" f"" for m, s in ((m, stats[m]) for m in MODELS)) + "
    모델관찰 총계criticalmajorminor수정 대상 샷(critical≥1)실패중앙 지연
    {m}{s['obs']}{s['critical']}{s['major']}{s['minor']}{len(s['crit_shots'])}/{n}{s['fail']}{sorted(s['lat'])[len(s['lat'])//2] if s['lat'] else '-'}s" f"
    " + "

    샷 단위 수정-대상 일치(critical 유무 기준): " + " · ".join( f"{a}↔{b} {agree(a, b)}/{n}" for a, b in (("grok46", "qwen"), ("grok46", "gemini"), ("qwen", "gemini"))) + "

    계약: 프로덕션 관찰 스템(judge v10) 원문+JSON" " 절 · 입력=선정본+선정 롤 프롬프트 전문 · 참조 이미지는 3모델" " 공통 미첨부(프로덕션 관찰과의 차이) · 수정 게이트=critical 만" "(프로덕션 계약 동일)

    ") if any(any(results[t].get(f"sel_{m}") for m in MODELS) for t in tags): n_sel = sum(1 for t in tags if any(results[t].get(f"sel_{m}") for m in MODELS)) summary += ( "

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

    " "" "" "" + "".join( f"" f"" f"" f"" for m in MODELS) + "
    모델프로덕션과 일치프로덕션과 반대정·역 불일치(split)판정 실패
    {m}{sel_stats[m]['match']}/{n_sel}{sel_stats[m]['diff']}{sel_stats[m]['split']}{sel_stats[m]['none']}
    " "

    판정 계약=프로덕션 judge_still v7 원문(4축 " "readings·hard 위반·0-10 점수). 정·역=첨부 순서와 라벨을 함께 " "뒤집어 위치 편향 상쇄 — 두 회가 같은 롤이어야 수렴. 프로덕션 " "선정은 참조 첨부·G+Q 합의 등 조건이 달라 '기준'이 아니라 " "대조점이다.

    ") html = ( "\n" "grok-4.6 판정 파일럿 — 23샷 3자 비교\n\n

    grok-4.6 판정 파일럿 — 23샷 관찰 3자 비교

    \n" + summary + "\n".join(rows)) (OUT / "index.html").write_text(html, encoding="utf-8") print(f"갤러리: {OUT / 'index.html'}") def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--scenes", default="1,4,5,9,37,39,56,64,65") ap.add_argument("--html", action="store_true", help="갤러리만 재생성") ap.add_argument("--select", action="store_true", help="선정 축(2후보 a/b 정·역 2회)만 실행") args = ap.parse_args() scenes = {int(s) for s in args.scenes.split(",") if s.strip()} if args.select: run_select(scenes) elif not args.html: run(scenes) build_html() if __name__ == "__main__": main()