"""지목 샷 선정 재판정 — 심판 4종을 같은 조건으로 나란히 세운다. 묻는 것은 둘이다. ① **생성 문제인가 선택 문제인가** — 후보 중에 더 나은 것이 실재하는데 못 골랐던 것이라면 판정 팩 v6 로 뒤집혀야 한다. 후보 전체가 나빴다면 안 뒤집힌다. ② **어느 심판을 믿을 것인가** — 같은 팩·같은 후보·같은 참조를 주고 넷을 동시에 물어 사용자 육안 판정과 대조한다. 이미지 생성 0 — 판정 콜만 든다. 심판 claude-opus 현행 1심 (SELECT_JUDGE_MODEL) gpt 현행 2심, 복잡 구조물 기하 전담 (SELECT_JUDGE_MODEL_2) gemini-pro 구 단독 심판 (JUDGE_MODEL) — C 실행이 쓴 심판 qwen3.8-max DashScope 직접 — 프로젝트 Router 는 json_schema 를 보내는데 Qwen 이 지원하지 않는다(`qwen_vlm_pilot` 배선 재사용) 넷 다 **팩 v6 `judge_still` 과 v6 스키마**를 그대로 받는다. 달라지는 것은 모델뿐이므로 비교는 심판만 재고 있다. 프로덕션 합의(`combine_select_verdicts`: 정규화 점수 합 − 하드 위반 개수 페널티)도 저장된 판정으로 함께 계산한다 — 추가 콜 0. 사용 .venv/bin/python reselect_v6.py [--stems S104sh7,...] [--all] [--judges claude-opus,gpt,gemini-pro,qwen] """ from __future__ import annotations import argparse import json import sys import threading import traceback from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from typing import Any, Dict, List, Tuple sys.path.insert(0, str(Path(__file__).resolve().parent)) from app.modules.pipeline.multiroll_gemini import ( # noqa: E402 JUDGE_MODEL, JUDGE_PACK_VERSION, SELECT_JUDGE_MODEL, SELECT_JUDGE_MODEL_2, combine_select_verdicts, png_part, ref_parts, resolve_judge_pack_version, resolve_judge_texts, ) from app.modules.pipeline.multiroll_select import ( # noqa: E402 build_judge_schema, roll_labels, ) PROJ = "e716bafb-24bb-42b7-aea0-fdb383844ee8" EPI = "d6a9aa85-b75e-400c-980c-4ee7e876a15b" ROOT = Path(__file__).resolve().parent.parent RECIPE = ROOT / f"projects/{PROJ}/images/{EPI}/scene/recipe" QWEN = "qwen" ROUTER_JUDGES = [SELECT_JUDGE_MODEL, SELECT_JUDGE_MODEL_2, JUDGE_MODEL] ALL_JUDGES = ROUTER_JUDGES + [QWEN] # 사용자 육안 지목 중 "더 나은 후보가 실재했는지"를 물어야 하는 샷. DEFAULT_STEMS = [ "S104sh7", # 2026-08-07 지목 — B 가 훨씬 좋은데 A 를 골랐다 "S62sh4", # 지폐가 미국 달러 (한국 지폐 후보 실재) "S42sh4", # 폰이 손 없이 뜬다 (손 보이는 후보 실재) "S13sh3", # 핸들 림 이중 / 인물이 조수석 "S15sh5", # 룸미러 반사가 광학적으로 불가능 "S18sh5", # 좌석이 돌아가 있다 "S88sh6", # 총구·시선 방향 (점수 A6:B14 로 정반대였다) "S76sh5", # 순찰정 ] # 사용자가 육안으로 밝힌 정답 — 아는 것만. 나머지는 "현재 선정이 틀렸다"까지만 # 알므로 정답 라벨을 지어내지 않는다(그러면 대조가 거짓이 된다). USER_TRUTH = {"S104sh7": "B"} def build_size_index() -> Dict[int, List[Path]]: """`` 로 남은 인라인 참조를 크기 일치로 되찾기 위한 색인.""" idx: Dict[int, List[Path]] = {} for p in (ROOT / f"projects/{PROJ}").rglob("*.png"): idx.setdefault(p.stat().st_size, []).append(p) return idx def resolve_refs(entries, size_idx) -> Tuple[List[Tuple[str, Path]], List]: """기록의 refs → [(label, Path)]. 복원 실패는 건너뛴다. 참조를 비운 채 판정하면 원 판정과 다른 계약이 되어 대조가 무의미해진다. """ out, unresolved = [], [] for e in entries: label, path = e.get("label", ""), str(e.get("path", "")) if path.startswith(" List[Dict[str, Any]]: """`make_gemini_judge_fn.judge_fn` 과 동일한 제시 순서.""" parts: List[Dict[str, Any]] = [{ "type": "text", "text": "THE PROMPT (all candidates were generated from this):\n" + prompt, }] parts += ref_parts(refs) for lab, p in zip(labels, cands): parts.append({"type": "text", "text": f"Candidate {lab}:"}) parts.append(png_part(p)) return parts def judge_router(model: str, judge_sys: str, schema, parts) -> Dict[str, Any]: """프로젝트 Router 심판 — 모델만 강제하고 나머지는 프로덕션과 동일.""" from app.modules.llm.llm_client import call_structured tag = "still_recipe_select_rejudge" return call_structured( tag, judge_sys, parts, schema, project_config={tag: {"model": model}}, schema_name=tag, ) def judge_qwen(judge_sys: str, schema, parts) -> Dict[str, Any]: """Qwen — json_schema 미지원이라 DashScope 를 직접 부른다.""" from qwen_vlm_pilot import ask_qwen payload, meta = ask_qwen(judge_sys, parts, schema, max_retry=1) payload["_qwen_meta"] = meta return payload def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("out", type=Path) ap.add_argument("--stems", default="") ap.add_argument("--all", action="store_true") ap.add_argument("--judges", default=",".join(ALL_JUDGES)) ap.add_argument("--workers", type=int, default=4) args = ap.parse_args() judges = [j.strip() for j in args.judges.split(",") if j.strip()] records = json.loads((RECIPE / "records.json").read_text("utf-8")) size_idx = build_size_index() if args.all: stems = [k for k, v in records.items() if "::" not in k and isinstance(v, dict) and len(v.get("verdicts") or []) >= 2] elif args.stems: stems = [s.strip() for s in args.stems.split(",") if s.strip()] else: stems = list(DEFAULT_STEMS) pack = resolve_judge_pack_version(JUDGE_PACK_VERSION) print(f"판정 팩 = v{JUDGE_PACK_VERSION} ({pack})", flush=True) print(f"심판 = {judges}", flush=True) jobs, skipped = [], [] for stem in stems: rec = records.get(stem) if not isinstance(rec, dict): skipped.append({"stem": stem, "why": "기록 없음"}) continue n = len(rec.get("verdicts") or []) if n < 2: skipped.append({"stem": stem, "why": f"후보 {n}장 — 고를 것이 없다"}) continue labels = roll_labels(n) cands = [RECIPE / f"{stem}_{lab.lower()}.png" for lab in labels] missing = [p.name for p in cands if not p.exists()] if missing: skipped.append({"stem": stem, "why": "후보 이미지 결손", "detail": missing}) continue refs, unresolved = resolve_refs(rec.get("refs") or [], size_idx) if unresolved: skipped.append({"stem": stem, "why": "참조 복원 실패", "detail": unresolved}) continue jobs.append((stem, rec, labels, cands, refs)) print(f"재판정 대상 {len(jobs)}샷 × 심판 {len(judges)} " f"= {len(jobs) * len(judges)}콜 · 건너뜀 {len(skipped)}", flush=True) for s in skipped: print(f" 건너뜀 {s['stem']}: {s['why']}", flush=True) tasks = [(stem, rec, labels, cands, refs, j) for (stem, rec, labels, cands, refs) in jobs for j in judges] results: Dict[str, Dict[str, Any]] = {} lock, done = threading.Lock(), [0] def run(task): stem, rec, labels, cands, refs, model = task texts = resolve_judge_texts(len(labels), judge_name="judge_still") schema = build_judge_schema(labels) parts = build_parts(rec.get("prompt", ""), refs, labels, cands) if model == QWEN: res = judge_qwen(texts["judge_sys"], schema, parts) else: res = judge_router(model, texts["judge_sys"], schema, parts) return stem, model, res with ThreadPoolExecutor(max_workers=args.workers) as ex: futs = {ex.submit(run, t): (t[0], t[5]) for t in tasks} for f in as_completed(futs): stem, model = futs[f] slot = results.setdefault(stem, {"per_judge": {}}) try: _s, _m, res = f.result() slot["per_judge"][model] = res except Exception as exc: # noqa: BLE001 slot["per_judge"][model] = { "error": repr(exc), "trace": traceback.format_exc()[-1200:]} with lock: done[0] += 1 print(f" [{done[0]}/{len(tasks)}] {stem} · {model}", flush=True) # 기존 선정·사용자 정답과 대조 + 프로덕션 합의 재계산(추가 콜 0) for stem, slot in results.items(): rec = records[stem] labels = roll_labels(len(rec.get("verdicts") or [])) slot["labels"] = labels slot["old_selected"] = rec.get("selected") slot["old_ranking"] = rec.get("ranking") slot["old_verdicts"] = rec.get("verdicts") slot["user_truth"] = USER_TRUTH.get(stem) ok = {m: r for m, r in slot["per_judge"].items() if isinstance(r, dict) and "error" not in r and r.get("winner")} slot["winners"] = {m: r.get("winner") for m, r in ok.items()} slot["agreed"] = len(set(slot["winners"].values())) == 1 if ok else None dual = {m: ok[m] for m in (SELECT_JUDGE_MODEL, SELECT_JUDGE_MODEL_2) if m in ok} if len(dual) == 2: slot["production_combined"] = combine_select_verdicts(dual, labels) args.out.parent.mkdir(parents=True, exist_ok=True) args.out.write_text(json.dumps( {"pack": pack, "judges": judges, "results": results, "skipped": skipped}, ensure_ascii=False, indent=1), "utf-8") print("\n" + "=" * 66) hdr = f"{'샷':10s} {'구':4s}" for j in judges: hdr += f" {j[:11]:>11s}" hdr += f" {'합의':>5s} {'사용자':>5s}" print(hdr) for stem in sorted(results): s = results[stem] row = f"{stem:10s} {str(s.get('old_selected')):4s}" for j in judges: w = s["winners"].get(j) row += f" {(w or '실패'):>11s}" pc = (s.get("production_combined") or {}).get("winner") row += f" {(pc or '-'):>5s} {(s.get('user_truth') or '-'):>5s}" print(row) print(f"\n기록 → {args.out}") if __name__ == "__main__": main()