"""multiroll_gemini — multiroll_select 의 production 어댑터 (2026-07-13). nb2(GeminiImageClient) 생성 + Gemini 단독 VLM 판정/결함 검사(call_structured 멀티모달 parts) 어댑터를 still_recipe_service 와 plate_multiroll 이 공용한다. 판정·결함 계약 텍스트는 prompts/_base/multiroll_judge 팩(SOT). """ from __future__ import annotations import base64 import json import logging from pathlib import Path from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple logger = logging.getLogger(__name__) JUDGE_MODULE = "multiroll_judge" JUDGE_MODEL = "gemini-pro" # 결함 critique·참조 선택 판정 = Gemini # ── 후보 선정 판정 모델 (2026-08-06 사용자 지시) ───────────────────── # 선정 판정만 Claude Opus 로 옮긴다. critique(`make_gemini_critique_fn`)와 # 참조 사진 선택(`make_ref_pick_judge_fn`)은 Gemini 그대로다. # # 근거는 16샷 전수 육안 대조다 — 사용자 판정과 Gemini 판정이 일치한 것이 # 1건뿐이었고, 틀린 축이 ①방향(총구·시선이 상대를 향하는가) ②복잡 구조물 # 내부 공간(핸들 이중·운전석/조수석·좌석 회전) ③중요 엔티티의 정체(어느 # 나라 지폐인가·목걸이 펜던트가 그 물건인가)로 몰려 있었다. # # ★키가 없으면 alias 가 Router 에 등록되지 않는다(`_build_router`) — 그때는 # 기존 `gemini-pro` 로 남는다. 키 없이 배포해도 경로가 죽지 않는다. SELECT_JUDGE_MODEL = "claude-opus" # 두 번째 심판 — 복잡 구조물 기하 전담. 4모델 대조 실측(2026-08-06)이 근거다: # 핸들이 이중으로 겹친 이미지에 Opus·Fable 은 "1개"라 답했고 Sol 만 2,2 로 # 맞혔다. 대조군(명백히 1개)에서도 Sol 은 1,1 로 정확했다 — 과탐지가 아니다. # Gemini 는 그 대조군에서 2개, 손이 쥔 폰에 "닿은 것 없음"이라 답해 제외했다. SELECT_JUDGE_MODEL_2 = "gpt" # 하드 위반 1건당 정규화 점수에서 깎는 양. 처음엔 위반이 하나라도 있으면 # 탈락시켰는데 7샷 전부에서 모든 후보가 탈락해 필터가 무의미해졌다 — 두 # 심판이 각자 위반을 넉넉히 적기 때문이다. 개수 페널티로 바꾸니 승자가 모든 # 샷에서 위반 최소 후보와 일치했고, 계수를 0~0.40 으로 흔들어도 승자가 전부 # 그대로였다(저장된 판정으로 재계산, 유료 콜 0). SELECT_VIOLATION_PENALTY = 0.25 # 조건부 이중 — 첫 심판이 1위를 이 폭 이상으로 앞세우면 둘째를 부르지 않는다. # 208샷 실측(2026-08-06): Sol 이 판정을 뒤집은 것은 23건(11.2%)이고 그중 # 22건이 첫 심판의 정규화 점수차 0.30 **미만**에서 일어났다. 문턱 0.30 에서 # 둘째 호출의 48.6% 를 생략하면서 뒤집힘의 95.7% 를 그대로 잡는다. # # ★"첫 심판이 하드 위반을 못 봤는가"는 조건에 넣지 않았다 — 같은 놓침률에서 # 절감이 10%p 낮았다. 둘째 심판이 하드 위반을 남발하기 때문이다(한쪽만 잡은 # 위반 139건 대 17건, 내용은 "지붕 윤곽이 완전히 같지 않다" 급). 위반 유무는 # 판별력이 약하고 점수차가 진짜 신호다. # # ★표본 주의: 뒤집힘이 23건뿐이라 이 문턱은 과적합 위험이 있다. 0.30 은 # "1위를 2위보다 30% 이상 앞세웠다"는 해석 가능한 값이라 택했다. # 0.0 으로 내리면 항상 이중(기존 동작). SELECT_DUAL_MARGIN_SKIP = 0.30 # ── G+Q 판정 체계 (2026-08-10 사용자 확정 — 설계 SOT: # docs/superpowers/specs/2026-08-10-dual-judge-selection-redesign-design.md) ── # `multiroll_gq_judge_enabled` ON 일 때 선정·fix-rejudge 심판 = Gemini+Qwen # **동시**(margin-skip 없음). Qwen 은 Router 미등록(json_schema 미지원) — # 이 센티널을 `make_gemini_judge_fn._one` 이 qwen_vlm_client 로 디스패치한다. QWEN_JUDGE_MODEL = "qwen-vlm" # 합의 규칙: winner 일치 또는 "Qwen 눈의 격차"(Qwen 정규화 점수에서 Q 승자 − # G 승자) < 이 값 → Gemini 채택. 이상 → combine_select_verdicts 합산 합의. # 근거 = 255샷 실측(artifact/20260809_final_eval_gallery): winner 일치 68.6%, # 불일치 80건 중 격차<0.2 가 22건(Qwen 도 크게 반대하지 않는 구간) — 합쳐 # 77.3% 가 Gemini 채택 구간. 문턱을 낮추면 합산 경로가, 높이면 Gemini 단독 # 채택이 늘어난다. GQ_DISAGREE_MARGIN = 0.20 # 합의 정책 버전 — 지문 기여(ON 시). 문턱이 문자열에 접혀 있어 문턱 변경이 # 곧 정책 버전 변경 = 산출 무효화로 이어진다. GQ_SELECT_POLICY_VERSION = "gq_select_v1_margin020" # ── QK 판정 체계 (2026-08-12 사용자 확정 "qwen 3.8 max main -> kimi k3 # 보조, gemini 버리자") — 멀티롤 선정·수정 재판정의 판정 모델 전환: # 메인(취합·우선 자리)=Qwen 3.8 Max, 보조(관찰·교차 자리)=Kimi K3, # 경로=OpenRouter(재평가 실측=build_qk_openrouter_eval.py). 합의 수학은 # `_judge_gq` 를 그대로 쓴다(우선 심판=models[0], 격차=보조 눈) — 우선 # 채택 route 이름만 "qwen_priority" 다. 모델 슬롯은 settings 이 SOT. OPENROUTER_JUDGE_PREFIX = "openrouter:" # 합의 정책 버전 — QK ON 시 지문 기여. GQ 와 별도 문자열(같은 수학이라도 # 판정자가 다르면 산출이 다르다 — 정책 식별자를 공유하지 않는다). # v2 (2026-08-12 밤): 보조=Kimi(OpenRouter)→Gemini 반전 — OpenRouter 지연 # (Kimi 245s·Qwen 207s)+비JSON 50% 실측으로 사용자 재결정. 판정자 구성이 # 바뀌면 산출이 다르므로 정책 버전도 올린다(v1 산출 지문 무효화). # v3 (2026-08-12 밤, Codex 재확인 BLOCK): Gemini 보조 fallback 봉인 # (enable_fallback=False — safety 강등 시 GPT 가 판정했는데 Gemini 로 # 기록되는 오귀속 차단)은 **실행 의미 변경**이다 — v2(강등 허용) 산출이 # 봉인 계약 산출로 오독되지 않게 정책 버전을 가른다. v2 로 만들어진 # 카나리아 완료 샷은 다음 방문에서 drift/JIT 재검 대상이 되는 것이 # 정직한 계약이다. QK_SELECT_POLICY_VERSION = ( "qk_select_v3_qwen_main_gemini_assist_no_fallback_margin020") # ── G+G46 판정 체계 (2026-08-13 사용자 확정 "Gemini + Grok 4.6 을 50:50 # 으로") — 선정·fix 재판정 심판 = Gemini + grok-4.6(OpenRouter) **동시· # 동등**. 슬롯 계약과 합의 수학은 `_judge_gq` 재활용하되 **우선권 갈래가 # 없다**: winner 일치=채택(route=agree) / 불일치=`combine_select_verdicts` # 합산(route=combined — 격차 크기와 무관, gap 은 기록만). 동등 체계에서 # 격차 문턱의 "우선 채택"은 정의될 수 없어(우선 심판이 없다) 불일치 관례 # (combined 합산)를 전 구간에 쓴다. models[0]=Gemini 는 우선권이 아니라 # agree 시 verdict shape 제공 슬롯이다. 근거=23샷 파일럿(단독 판정 정·역 # split 이 셋 다 큼: gemini 5/23·grok 7/23 — 단독 우선권을 줄 근거 부재, # artifact/grok46judge23). GG46_SELECT_POLICY_VERSION = "gg46_select_v1_equal_combined_no_fallback" def _normalized_margin(res: Dict[str, Any]) -> float: """판정 결과의 1위·2위 정규화 점수 차. 후보가 하나면 1.0(=둘째 불필요).""" scores = sorted( ((v.get("score") or 0) for v in (res.get("verdicts") or [])), reverse=True) if len(scores) < 2: return 1.0 top = scores[0] or 1 return (scores[0] - scores[1]) / top def resolve_select_judge_models() -> List[str]: """후보 선정 판정 모델 목록 — 키가 있으면 이중, 없으면 기존 단독. G+Q ON 이면 [Gemini, Qwen] — 첫째가 우선 심판이다(`_judge_gq` 계약). ON+DASHSCOPE 키 없음 = fail-closed: 조용히 Gemini 단독으로 강등하면 "이중으로 판정했다"는 기록·지문이 거짓이 된다. 지문 계산도 이 함수를 지나므로 잘못된 구성은 지문 단계에서 이미 막힌다. """ from app.core.config import settings _on_flags = [ name for name, on in ( ("multiroll_qk_judge_enabled", settings.multiroll_qk_judge_enabled), ("multiroll_gq_judge_enabled", settings.multiroll_gq_judge_enabled), ("multiroll_gg46_judge_enabled", getattr(settings, "multiroll_gg46_judge_enabled", False)), ) if on ] if len(_on_flags) > 1: from app.core.errors import AppError raise AppError( code="multiroll.judge_flags_conflict", message=( f"{' 와 '.join(_on_flags)} 가 동시에 ON 이다 — 판정 체계는 " "하나만. 한쪽 플래그를 내려라 (fail-closed)" ), status_code=422, ) if settings.multiroll_qk_judge_enabled: # 2026-08-12 카나리아 실측 후 사용자 재결정("gemini 를 보조로 하고 # 기존 코드에서 QWEN 의 비중을 반대로"): OpenRouter 경유 판정이 # Kimi 평균 245s·Qwen 207s+비JSON 50% 로 운영 부적합 — # **메인=Qwen(DashScope 직결, 완주 실측 검증)** + # **보조=Gemini(기존 G+Q 의 우선 자리 모델을 보조 자리로 반전)**. # 기존 부품 그대로 재활용: _judge_gq 는 슬롯 순서 계약(models[0]= # 우선), 두 모델의 호출 경로 모두 기존 센티널/alias. Kimi 보조는 # 보류(OpenRouter 클라이언트·디스패치는 남겨 둠 — 재도입 시 지연 # 진단 선행). from app.core.errors import AppError from app.modules.llm.qwen_vlm_client import qwen_configured if not qwen_configured(): raise AppError( code="multiroll.qk_judge_unconfigured", message=( "multiroll_qk_judge_enabled=ON 인데 DASHSCOPE_API_KEY 가 " "비어 있다 — Qwen 메인 판정을 돌 수 없다. 키를 넣거나 " "플래그를 내려라 (fail-closed)" ), status_code=422, ) # 물리 모델 슬롯 공백 preflight (Codex 반전 리뷰 BLOCK-1): # qwen_configured 는 키만 본다 — qwen_vlm_model 이 공백이면 빈 # model 호출이 실패하고 _judge_gq 가 그것을 삼켜 Gemini 단독 # (single_*)으로 확정된다(이중 설정인데 단독 판정 — 이전 OpenRouter # 빈 슬롯 BLOCK 과 같은 부류). 첫 유료 호출 전에 막는다. if not (settings.qwen_vlm_model or "").strip() \ or not (settings.gemini_text_model or "").strip(): raise AppError( code="multiroll.qk_judge_empty_model_slot", message=( "qwen_vlm_model/gemini_text_model 중 빈 슬롯이 있다 — " "QK 이중 판정이 성립하지 않는다. 물리 모델 설정을 " "채워라 (fail-closed)" ), status_code=422, ) return [QWEN_JUDGE_MODEL, JUDGE_MODEL] if settings.multiroll_gq_judge_enabled: from app.core.errors import AppError from app.modules.llm.qwen_vlm_client import qwen_configured if not qwen_configured(): raise AppError( code="multiroll.gq_judge_unconfigured", message=( "multiroll_gq_judge_enabled=ON 인데 DASHSCOPE_API_KEY 가 " "비어 있다 — Qwen 없이 G+Q 판정을 돌 수 없다. 키를 " "넣거나 플래그를 내려라 (fail-closed)" ), status_code=422, ) return [JUDGE_MODEL, QWEN_JUDGE_MODEL] if getattr(settings, "multiroll_gg46_judge_enabled", False): # G+G46 (2026-08-13): Gemini + grok-4.6(OpenRouter) 동등 이중. # ON+OpenRouter 키 없음 = fail-closed (QK/GQ 관례 동형 — 조용한 # 단독 강등은 기록·지문을 거짓으로 만든다). 물리 슬롯 공백도 첫 # 유료 호출 전에 막는다(QK preflight 관례 동형). from app.core.errors import AppError from app.modules.llm.openrouter_vlm_client import ( openrouter_configured, ) if not openrouter_configured(): raise AppError( code="multiroll.gg46_judge_unconfigured", message=( "multiroll_gg46_judge_enabled=ON 인데 OPENROUTER_API_KEY " "가 비어 있다 — grok-4.6 판정을 돌 수 없다. 키를 넣거나 " "플래그를 내려라 (fail-closed)" ), status_code=422, ) if not (getattr(settings, "grok_judge_model", "") or "").strip() \ or not (settings.gemini_text_model or "").strip(): raise AppError( code="multiroll.gg46_judge_empty_model_slot", message=( "grok_judge_model/gemini_text_model 중 빈 슬롯이 있다 — " "G+G46 동등 이중 판정이 성립하지 않는다. 물리 모델 " "설정을 채워라 (fail-closed)" ), status_code=422, ) return [ JUDGE_MODEL, OPENROUTER_JUDGE_PREFIX + settings.grok_judge_model.strip(), ] if not settings.anthropic_api_key: return [JUDGE_MODEL] if not settings.multiroll_dual_select_judge_enabled: return [SELECT_JUDGE_MODEL] return [SELECT_JUDGE_MODEL, SELECT_JUDGE_MODEL_2] def _q_gap(q_res: Dict[str, Any], g_winner: str, q_winner: str) -> float: """"Qwen 눈의 격차" — Qwen 정규화 점수에서 Q 승자 − G 승자. Gemini 우선 체계라 재는 것은 "Qwen 이 Gemini 승자에 얼마나 반대하는가" 하나뿐이다. 전 후보 0점(top<=0)이면 격차를 잴 근거가 없다 — 0.0(=반대 근거 없음)으로 떨어져 Gemini 채택이 된다. """ scores = {v.get("label"): (v.get("score") or 0) for v in (q_res.get("verdicts") or [])} top = max(scores.values()) if scores else 0 if top <= 0: return 0.0 return (scores.get(q_winner, 0) - scores.get(g_winner, 0)) / top def _judge_gq(models: List[str], one, parts, labels: List[str], priority_route: str = "gemini_priority", equal_disagree_combined: bool = False) -> Dict[str, Any]: """우선+보조 동시 판정 — 2026-08-10 합의 규칙. margin-skip 없이 둘 다 부른다. 슬롯 계약: models[0]=우선(취합) 심판, models[1]=보조(관찰·교차) 심판. · winner 일치 → 우선 판정 채택 (route=agree) · 불일치 & 격차<`GQ_DISAGREE_MARGIN` → 우선 채택 (route=priority_route) · 불일치 & 격차≥문턱 → `combine_select_verdicts` 합산 (route=combined) · 한쪽 실패 → 남은 쪽 채택+경고 (route=single_<모델>) / 전건 실패 → raise priority_route (2026-08-12 QK 전환): 우선 채택 route 의 기록 이름 — G+Q(기본)="gemini_priority" byte-identical, QK="qwen_priority". equal_disagree_combined (2026-08-13 G+G46): True 면 **동등 이중** — 우선 채택 갈래를 걷어내고 불일치는 격차와 무관하게 전부 combined 합산이다(동등 체계에는 격차 문턱이 정할 "우선 심판"이 없다). 이때 models[0] 은 우선권이 아니라 agree 시 verdict shape 제공 슬롯이고, gap 은 기록으로만 남는다. False(default)=기존 수학 byte-identical. 반환 shape 는 단독 판정 호환(winner/ranking/verdicts…) — 채택 경로와 양쪽 승자·격차는 `gq` 키에 병기해 record·갤러리가 추적한다 (키 이름은 체계 무관 공용 기록 슬롯이다). """ out: Dict[str, Dict[str, Any]] = {} last_exc: Optional[BaseException] = None for m in models: try: out[m] = one(m, parts, f"_{m.replace('-', '')}") except Exception as exc: # noqa: BLE001 last_exc = exc logger.warning( "G+Q 판정: %s 실패 — 남은 심판으로 진행: %r", m, exc) if not out: raise RuntimeError("G+Q 선정 판정 전건 실패") from last_exc if len(out) == 1: m, res = next(iter(out.items())) return {**res, "gq": {"route": f"single_{m}", "models": [m]}} g_model, q_model = models[0], models[1] g, q = out[g_model], out[q_model] gw, qw = g.get("winner"), q.get("winner") winners = {g_model: gw, q_model: qw} if gw == qw: return {**g, "gq": {"route": "agree", "gap": 0.0, "per_model_winner": winners, "models": list(out)}} gap = _q_gap(q, str(gw), str(qw)) if not equal_disagree_combined and gap < GQ_DISAGREE_MARGIN: return {**g, "gq": {"route": priority_route, "gap": round(gap, 3), "per_model_winner": winners, "models": list(out)}} combined = combine_select_verdicts(out, labels) if equal_disagree_combined: # (Codex GG46 R1 BLOCK-1) combined 의 실소비자는 winner 필드가 # 아니라 **verdicts 점수**다 — 비flip 경로는 gemini_select 가 # verdicts 최고점으로 다시 고르고, flip 경로(judge_flip·fix # 재판정)는 combine_flip_verdicts 가 verdicts 점수를 합산한다. # 기존 shape 는 첫 심판(Gemini) 원점수를 실어, 동등 합산 승자가 # 하류에서 조용히 Gemini 우선으로 되돌아갔다(50:50 계약 불성립). # 동등 모드만 verdicts 점수를 합산값(adjusted×1000 — shape # 검증기가 int 요구)으로 교체해 소비자가 합산 승자를 그대로 # 뽑게 한다. 동점 미만 차이는 ranking(합산 순)이 가른다. # GQ/QK combined 는 기존 shape 그대로(byte-identical). _adj = (combined.get("dual") or {}).get("adjusted") or {} combined["verdicts"] = [ {**v, "score": int(round(_adj.get(v.get("label"), 0) * 1000))} for v in (combined.get("verdicts") or []) ] combined["gq"] = {"route": "combined", "gap": round(gap, 3), "per_model_winner": winners, "models": list(out)} return combined def combine_select_verdicts( per_model: Dict[str, Dict[str, Any]], labels: List[str], ) -> Dict[str, Any]: """이중 판정 합의 — 하드 위반 개수 페널티 + 모델별 정규화 점수 합. 점수를 그냥 더하면 안 된다. 척도가 모델마다 다르다(0~100 대 0~10) — 원 점수를 더하면 큰 척도 모델이 판정을 지배한다. 이 프로젝트에서 이미 겪은 함정이라 모델별 최고점으로 정규화한 뒤 더한다. 하드 위반은 **합집합**으로 센다. 한쪽이 못 보는 축을 다른 쪽이 채우는 것이 이중의 목적이므로 교집합을 쓰면 그 목적이 사라진다. 반환 shape 는 단독 판정과 호환된다(winner/ranking/verdicts) — 호출측 코드와 기록 계약을 그대로 둔다. """ norm: Dict[str, float] = {lab: 0.0 for lab in labels} vio: Dict[str, List[str]] = {} n_fail = 0 for model, res in per_model.items(): readings = res.get("readings") or [] verdicts = res.get("verdicts") or [] if res.get("all_candidates_fail"): n_fail += 1 top = max((v.get("score") or 0) for v in verdicts) if verdicts else 0 for v in verdicts: lab = v.get("label") if lab in norm: norm[lab] += (v.get("score") or 0) / (top or 1) for r in readings: lab = r.get("label") for hv in r.get("hard_violations") or []: if lab in norm: vio.setdefault(lab, []).append(f"[{model}] {hv}") adj = {lab: norm[lab] - SELECT_VIOLATION_PENALTY * len(vio.get(lab, [])) for lab in labels} ranking = sorted(labels, key=lambda lab: -adj[lab]) # verdicts 는 첫 심판 것을 싣되, 위반 합집합을 판정문에 덧붙여 기록이 # 두 심판을 모두 대변하게 한다. first = next(iter(per_model.values())) verdicts = [] for v in (first.get("verdicts") or []): lab = v.get("label") extra = vio.get(lab) or [] verdicts.append({ **v, "verdict_ko": (v.get("verdict_ko", "") + (" ★위반: " + " / ".join(extra) if extra else "")), }) return { "winner": ranking[0], "ranking": ranking, "verdicts": verdicts, # ★심판 **전원**이 선언했을 때만. 처음엔 `n_fail*2 >= n` 으로 썼는데 # 2명 중 1명만 선언해도 참이 되어(절반 이상 ≠ 과반) 254샷에서 # 45%(115건)가 "후보 전부 실패"로 찍혔다 — 단독 판정에서는 2% 였다. # 둘째 심판이 하드 위반을 남발하는 성향과 겹쳐 신호가 무의미해졌다. "all_candidates_fail": bool(per_model) and n_fail == len(per_model), "dual": { "models": list(per_model), "normalized": {k: round(v, 3) for k, v in norm.items()}, "adjusted": {k: round(v, 3) for k, v in adj.items()}, "violations": vio, "per_model_winner": {m: r.get("winner") for m, r in per_model.items()}, "agreed": len({r.get("winner") for r in per_model.values()}) == 1, }, } def _judge_dual(models: List[str], one, parts, labels: List[str], ) -> Dict[str, Any]: """조건부 이중 판정. 첫 심판을 먼저 부르고, 그가 1위를 `SELECT_DUAL_MARGIN_SKIP` 이상으로 앞세우면 둘째를 부르지 않는다 — 그 구간에서는 둘째가 판정을 거의 뒤집지 않는다는 것이 208샷 실측이다. 근소하면 둘째를 불러 합의한다. 한쪽이 죽으면 남은 쪽으로 진행한다. 둘 다 죽으면 원인을 달아 올린다. """ out: Dict[str, Dict[str, Any]] = {} last_exc: Optional[BaseException] = None for m in models: try: out[m] = one(m, parts, f"_{m.replace('-', '')}") except Exception as exc: # noqa: BLE001 last_exc = exc logger.warning( "이중 판정: %s 실패 — 남은 심판으로 진행: %r", m, exc) continue if len(out) == 1 and len(models) > 1: margin = _normalized_margin(out[m]) if margin >= SELECT_DUAL_MARGIN_SKIP: logger.info( "선정 판정: %s 가 1위를 %.2f 앞세워 둘째 심판 생략", m, margin) return out[m] if not out: # 원인을 삼키지 않는다 — 전건 실패의 진짜 이유(키·이미지 형식·쿼터)를 # 여기서 잃으면 호출측이 무엇을 고쳐야 하는지 알 수 없다. raise RuntimeError("이중 선정 판정 전건 실패") from last_exc if len(out) == 1: return next(iter(out.values())) return combine_select_verdicts(out, labels) def resolve_select_judge_model() -> str: """후보 선정 판정에 쓸 model alias — 키가 없으면 기존 Gemini 로 남는다. 지문(fingerprint)에도 이 함수의 반환값을 실어야 한다. 판정 모델이 바뀌면 선정 결과가 바뀌므로 산출이 무효화돼야 한다. """ from app.core.config import settings if settings.multiroll_qk_judge_enabled: # QK 메인(취합·우선 자리) — Qwen(DashScope) 센티널 (2026-08-12 반전) return QWEN_JUDGE_MODEL if settings.multiroll_gq_judge_enabled: return JUDGE_MODEL if getattr(settings, "multiroll_gg46_judge_enabled", False): # G+G46 동등 — 우선권이 없어 대표 alias 는 shape 제공 슬롯(Gemini). # GQ 와 alias 가 같지만 지문은 gg46_select_policy 스탬프+물리 쌍 # (resolve_select_judge_model_physical)이 가른다. return JUDGE_MODEL return SELECT_JUDGE_MODEL if settings.anthropic_api_key else JUDGE_MODEL def resolve_select_judge_model_physical() -> str: """선정 판정의 **물리** 모델 문자열 — 지문 스탬프용. alias 는 그대로인 채 뒤의 실제 모델만 바뀌는 경우(판 교체)를 지문이 잡아야 하므로 alias 와 따로 기록한다. 이중일 때는 두 모델을 모두 싣는다 — 한쪽만 바뀌어도 어느 롤이 뽑히는지가 바뀐다. """ from app.core.config import settings phys = { SELECT_JUDGE_MODEL: settings.anthropic_judge_model, SELECT_JUDGE_MODEL_2: settings.openai_model, JUDGE_MODEL: settings.gemini_text_model, QWEN_JUDGE_MODEL: settings.qwen_vlm_model, } return "+".join(phys.get(m, m) for m in resolve_select_judge_models()) _COUNT_WORDS = {1: "ONE", 2: "TWO", 3: "THREE", 4: "FOUR", 5: "FIVE"} JUDGE_PACK_VERSION_MAP = { "1": "1.202607132300", # v2 (E2E6 피드백 ⑥ 근본 대응, 2026-07-16): 판정·결함 계약에 엄격 # 우선순위 도입 — hard violation > 샷 텍스트 프레이밍·카메라축·크롭· # 순간 > 배치·관계 > 화면 내 보이는 범위의 정체성·소품 > pose/carried # (visible-within-frame 한정, 프레임 밖 부재 감점·와이드 가점 금지). # 실측 근거=S23sh1: 클로즈업 정답 롤 A 를 신체·carried 계약 우선으로 # 역선택(C7>B4>A3)+critique 가 가방 끈을 그려 넣어 악화. # ★judge_plate_ext 는 현재 runtime 소비처 0 (dormant — 세 소비자 전부 # judge_still 사용, Codex 배치 리뷰 MINOR 명시). v2 의 generic 화는 # 배선 시 효력. "2": "2.202607161610", # v3 (2026-07-21 E2E10 fix② Codex HIGH-4): fix_rejudge_header **단일 # 스템 팩** — [선정 원본 vs i2i 수정본] 2후보 재판정용 중립 헤더. # 기본 헤더('all candidates were generated from this')는 수정본이 # repair prompt 로 생성돼 거짓 — provenance 비노출 중립 계약으로 교체. # 기존 판정 계약(v2)은 무변경. "3": "3.202607211600", # v4 (2026-07-22 E2E11 fix③): gpt_composition_sys **단일 스템 팩** — # GPT 구도 전담 critique 계약. 기존 판정 계약(v2)·재판정 헤더(v3) # 무변경. "4": "4.202607221130", # v5 (2026-07-22 E2E13 fix⑤): v2 전체 스템 사본 + ①critique FIX # FEASIBILITY 절 — fix_en 은 현재 카메라·시점을 유지하는 국소 편집만, # 시점/카메라 이동·경사 반전이 필요한 결함은 unfixable=true 로 보고 # (L05B01 실측: 카메라 이동 fix 지시가 두 번째 철문 hard violation 을 # 생성해 반전 원본이 확정) ②judge_still/judge_plate_ext HARD 축 — # 프롬프트 structure facts 의 수직(오르내림) 방향과 반대로 렌더되거나 # 구조가 제공할 수 없는 vantage(최상층 위)에 선 후보=실격. # fix_rejudge_header(v3)·gpt_composition_sys(v4)는 별도 selector 유지. "5": "5.202607230000", # v6 (2026-08-06 사용자 확정 — 16샷 전수 육안 대조가 근거): v5 사본 + # ①judge_still 에 **세 축 서술 강제** — 후보마다 방향(시선·총구가 실제로 # 무엇에 꽂히는가)·복잡 구조물 내부 공간(조작 장치 개수·좌석 배치·반사의 # 광학적 가능성)·중요 엔티티 정체(지폐의 국적과 액면, 장신구 형태, 인종) # 를 점수 매기기 **전에** 문장으로 쓰게 한다. 점수만 받으면 안 본 축을 # 안 본 채로 넘어간다 — 총구가 상대를 전혀 겨누지 않은 후보에 A6:B14 를 # 준 것이 그 결과다. + `hard_violations` 배열과 `all_candidates_fail`. # ②critique — severity(critical/major/minor) 도입, "수정은 공짜가 아니다" # 실측 열거(발 하나 지우랬더니 시신 의상이 바뀜 등), fix_en 에 **보존 # 조항을 이름으로** 요구, 금지에는 그 자리를 무엇이 채우는지 함께 쓰게, # 소품의 액면·등급이 서사를 지탱할 때 그것을 명시하게. "6": "6.202608062200", # v7 (2026-08-07 심판 4종 대조가 근거): v6 사본 + 세 갈래. # ①judge_still 에 **PHYSICS** 네 번째 서술 축 — 공중에 뜬 몸·물체를 # 무엇이 받치는지(도움닫기·쥔 손·충격·딛고 선 면) 쓰게 하고, 아무것도 # 받치지 않으면 하드 위반. 같은 이미지에 Opus·Sol·Qwen 셋이 최고점을 # 주고 **셋 다 부유를 한 줄도 안 썼는데** Gemini 만 물리 불가로 걸었다. # v6 도 hard 예시 목록엔 적어 두었으나 목록에 있는 것과 보게 만드는 # 것은 다르다. "샷 텍스트가 공중이라 했다"는 면책이 아니라 지지를 # 찾아야 할 이유라고 못 박고, 걷는 자세를 눕힌 것은 도약이 아니라고 명시. # ②fix_rejudge_header 에 **수정 범위** 축 — 지시하지 않은 변경을 손실로 # 세게 하고 대등하면 덜 바꾼 쪽. 개악 실측 4건을 근거로 적었다. # ③fix_ref_label 신규 — 편집 호출에 동봉하는 참조의 라벨(편집 대상 아님). "7": "7.202608071100", # v8 (2026-08-10 G+Q 확정): **추가-스템 전용 팩** — gq_observe_sys(Qwen # 관찰 계약)·gq_compose_sys(Gemini 취합 계약) 둘만 담는다(v3 # fix_rejudge_header 단일 스템 팩과 같은 패턴). 판정·결함 계약(v7)은 # 무변경 — v7 디렉토리가 불변이어야 `judge_pack_content` 지문이 OFF # 상태에서 안 움직인다. "8": "8.202608101400", # v9 (2026-08-12 카메라·포즈 다각화 A안): v8 gq 스템 2종 사본 + 관찰 # 축에 STAGING(카메라 인지·자세 자연스러움 관찰 서술) 추가. GQ(v8 # 고정)는 미승격 유지 — QK 수정 흐름(QK_CRITIQUE_PACK_VERSION)의 # 도입판으로만 쓴다(새 체계라 물려받을 구 지문이 없다). "9": "9.202608121750", # v10 (2026-08-13 육안 8건 wave — 사용자 지시 #106): 3스템 팩. # ①gq_observe_sys — v9 사본 + LOOK 축 2종(착석=가구 설계 방향 정합, # 가림 연출된 얼굴의 노출·융합) + 심각도 경계 명문(참조 대비 가구· # 세트 드레싱 이동/소실은 최대 major — S39sh4 소파가 fix 를 발동시킨 # 실측이 근거). ②gq_compose_sys — v9 사본 + severity 를 관찰자 # rating 그대로 구조 필드로 전달(재평가 금지) + "critical 만 편집"이 # 코드 게이트로 강제됨을 명문. ③fix_rejudge_header — v7 사본 + # IDENTITY 자동 패배 축(인물이 다른 사람이 된 후보는 무엇을 고쳤든 # 탈락 — S39sh4 fix 인물 변형 채택이 근거). GQ(v8)·구 critique(v7) # 는 무접촉 — 지문 불변. "10": "10.202608131121", # v11 (2026-08-13 G+G46): 3스템 팩 — gq_observe_sys·fix_rejudge_header # 는 v10 사본 그대로, **gq_compose_sys 만** 2관찰자 취합 계약으로 개정 # (관찰 배열=두 관찰자 목록의 연결, 같은 결함 중복 가능 → 결함당 1건만 # 채택·중복은 issues 밖으로 — 실측 근거: 관찰 밀도 gemini 42 vs grok # 36, 수정-대상 일치 20/23 이라 중복이 상례다). QK(v10)·GQ(v8)·구 # critique(v7)는 무접촉 — 지문 불변. "11": "11.202608132045", # v12 (2026-08-14 표기 정책 개정 — 사용자 지시 "원어 텍스트 허용, 꼭 # 필요한 경우만"): **전체 팩** = v7 전 스템 사본 + v11 gq 3스템 사본 # 위에 텍스트 축만 개정. ①fix_tail — "No text ... anywhere" 절대문을 # "장면에 속한 실물 표기는 원본 그대로 보존, 자막·워터마크·오버레이만 # 금지"로 ②gq_observe_sys — 배제 목록의 no text 를 자막·오버레이 # 한정으로 좁히고(실물 표기는 깨진 글자·비원어·발명일 때만 결함) 화면 # 자막 굽기를 critical 열거에 추가 ③critique(legacy)도 동일 방향. # judge_still·compose·fix_rejudge_header 등 나머지는 바이트 사본. # still_recipe v20(생성측 표기 허용)과 정합 — 생성이 허용한 표기를 # 판정·수정이 도로 지우는 어긋남을 막는다. GQ(v8)·QK(v10) 무접촉. "12": "12.202608141305", # v13 (2026-08-19 참조 선별): **추가-스템 전용 팩**(v3·v4·v8 관례) — # `fix_missing_head`/`fix_missing_tail` 둘만 담는다. 참조 선별이 켜진 # 편집에서 "지금 사진에 아예 없는 것을 새로 넣어야 하는" 지적이 있을 # 때만 수정 지시문에 끼는 절이다(사용자 지시: 그때는 그 참조를 붙이고 # 지시문에 명시). 판정·결함·수정 본문 계약(v12)은 무접촉 — v12 # 디렉토리가 불변이어야 `judge_pack_content` 지문이 OFF 상태에서 안 # 움직인다. "13": "13.202608192332", } # ★전역 기본은 v6 에 둔다 (2026-08-07 Codex 리뷰 수용). # # v7 의 내용은 **스틸 전용**이다 — 인물의 물리적 지지, 편집 참조 라벨, # 수정 범위 축. 그런데 이 전역 하나를 올리면 구조물 씨드 # (`outdoor_structure_seed_step._config_hash` 의 `judge_pack`)와 배경 # 플레이트(`background_render_step` 의 `plate_multiroll_judge_pack`)까지 # 지문이 바뀌어 통째로 재실행 대상이 된다. 인물이 없는 산출에 인물 물리 # 축을 적용하려고 그 비용을 낼 이유가 없다. JUDGE_PACK_VERSION = "6" # 스틸(인물 샷) 전용 selector — 스틸 소비자만 명시로 이것을 쓴다. # v12 (2026-08-14): 표기 정책 개정(fix_tail) 승계 — judge_still 은 v7 # 바이트 사본이라 판정 계약 자체는 불변. STILL_JUDGE_PACK_VERSION = "12" # fix-rejudge 중립 헤더 스템 selector (E2E10 fix②) — 소비자·hash 공용 SOT FIX_REJUDGE_HEADER_PACK_VERSION = "3" # 스틸 전용 fix-rejudge 헤더 — v7 수정 범위 축(지시하지 않은 변경=손실) # + v10 IDENTITY 자동 패배 축(2026-08-13 #106: 인물이 다른 사람이 된 # 수정본은 무엇을 고쳤든 탈락 — S39sh4 실측). STILL_FIX_REJUDGE_HEADER_PACK_VERSION = "10" # GPT 구도 전담 critique 팩 selector (E2E11 fix③) — 소비자·hash 공용 SOT. # v4 (2026-07-22): gpt_composition_sys 단일 스템 — 구도·배치·스케일·시선축 # 위반만 보고(정체성/재질/조명 OUT OF SCOPE), Gemini critique 와 합산해 # 하나의 수정 프롬프트로 i2i. 판정 모델=GPT LVM(이미지 검증 분업). # v5 (E2E13 fix⑤ Codex BLOCKING-2): FIX FEASIBILITY 절 — 카메라 각/축 # 위반의 imperative fix 강제가 unfixable 방어를 우회하던 창 봉합(GPT # 경로도 시점 이동 fix 금지+unfixable=true 보고). GPT_COMPOSITION_PACK_VERSION = "5" GPT_COMPOSITION_MODEL = "gpt" # G+Q 수정 흐름 스템 팩 selector (2026-08-10) — Qwen 관찰·Gemini 취합 계약. # 스틸 지문에는 ON 일 때만 접힌다(`gq_critique_pack*` — OFF 무스탬프). GQ_CRITIQUE_PACK_VERSION = "8" # QK 수정 흐름 스템 팩 selector (2026-08-12) — Kimi 관찰·Qwen 취합 계약. # v9(STAGING 관찰 축 추가판)를 도입판으로 썼고, v10(2026-08-13 #106)에서 # 심각도 구조 전달+LOOK 축 2종을 얹었다. GQ(v8 고정)의 지문·산출은 # 1 byte 도 건드리지 않는다. QK_CRITIQUE_PACK_VERSION = "10" # G+G46 수정 흐름 스템 팩 selector (2026-08-13) — Gemini+grok-4.6 양쪽 # 관찰·Gemini 취합 계약. v11=v10 사본에 취합 스템만 2관찰자(중복 결함 # 1건 채택) 개정. v12(2026-08-14)=표기 정책 개정 관찰 스템 승계(자막· # 오버레이만 위반, 실물 원어 표기는 정당). 스틸 지문에는 ON 일 때만 # 접힌다(`gg46_critique_pack*` — OFF 무스탬프). GG46_CRITIQUE_PACK_VERSION = "12" # 참조 선별 — "사진에 아예 없는 것을 새로 넣어라" 절 스템 팩 selector # (2026-08-19). 추가-스템 전용 팩이라 기존 팩 지문에 접히지 않는다. FIX_MISSING_PACK_VERSION = "13" def load_fix_missing_texts( pack_version: str = FIX_MISSING_PACK_VERSION, ) -> Dict[str, str]: """참조 선별 ON 에서만 쓰는 '없는 것 새로 넣기' 절 (2026-08-19). 수정 지시문은 기본적으로 "여기 적힌 것만 고치고 나머지는 그대로"인데, 지적이 지금 사진에 **아예 없는** 인물·소품을 넣으라고 할 때가 있다 (시험 실측 S55sh1 인물 1명·S48sh9 스탠드). 그때만 이 절이 끼고 그 참조도 함께 붙는다 — 사용자 지시. """ from app.modules.prompt_loader import load_prompt resolved = resolve_judge_pack_version(pack_version) return { "missing_head": load_prompt( JUDGE_MODULE, "fix_missing_head", version=resolved).strip(), "missing_tail": load_prompt( JUDGE_MODULE, "fix_missing_tail", version=resolved).strip(), } def fix_ref_contract_sha() -> str: """참조 선별의 실질 지시문 지문 — 팩 밖 **코드 문자열**이라 따로 접는다. 선별이 실제로 무엇을 붙일지 정하는 문안은 둘이다: 번호가 무엇을 가리키는지 알려 주는 `_REF_INDEX_HEADER` 와, 결함 검사 스키마 안 선별 세 칸의 설명("참조를 붙이는 것은 공짜가 아니다 … 애매하면 아무 것도 적지 마라"). 이 둘은 팩이 아니라 코드에 있어 팩 내용 해시 (`fix_missing_pack_content`)가 못 덮는다 — 문안을 고치면 붙는 참조와 최종 그림이 달라지는데 모든 해시가 그대로여서, 반대 정책으로 만든 산출이 "같은 조건"으로 기록된다(2026-08-08 스테일 재사용 사고와 같은 부류). 스키마에서 **되읽어** 해시한다 — 상수를 따로 두고 손으로 맞추면 한쪽만 고쳐질 때 조용히 어긋난다. 대상은 선별 세 칸으로 한정하므로 severity·observation_index 등 다른 칸의 문안 개정에는 안 움직인다. """ import hashlib import json as _json from app.modules.pipeline.multiroll_select import build_critique_schema item = build_critique_schema( with_ref_gate=True)["properties"]["issues"]["items"] gate = { k: item["properties"][k] for k in ("needs_ref_indices", "adds_missing_entity", "missing_entity_name") } h = hashlib.sha256(_REF_INDEX_HEADER.encode("utf-8")) h.update(b"\x00") h.update(_json.dumps( gate, sort_keys=True, ensure_ascii=False).encode("utf-8")) return h.hexdigest()[:16] def load_fix_rejudge_header( pack_version: str = FIX_REJUDGE_HEADER_PACK_VERSION, ) -> str: """fix-rejudge 2후보 중립 판정 헤더 (Codex HIGH-4). 원본/수정본 provenance 를 노출하지 않고 '타깃 계약 충실도'로만 판정 — 헤더 문구=팩 v3 fix_rejudge_header 스템 SOT. """ from app.modules.prompt_loader import load_prompt return load_prompt( JUDGE_MODULE, "fix_rejudge_header", version=resolve_judge_pack_version(pack_version), ).strip() def resolve_judge_pack_version(selector: str = JUDGE_PACK_VERSION) -> str: try: return JUDGE_PACK_VERSION_MAP[selector] except KeyError: raise ValueError( f"unknown multiroll_judge pack selector {selector!r} " f"(known: {sorted(JUDGE_PACK_VERSION_MAP)})" ) def _pack_dir_content_hash(resolved: str) -> str: """판정 팩 내용 해시 — #77 에서 prompt_loader.pack_dir_content_hash 로 공용화(캐시·fail-closed 포함), 여기는 판정 모듈 위임만 남긴다.""" from app.modules.prompt_loader import pack_dir_content_hash return pack_dir_content_hash(JUDGE_MODULE, resolved) def judge_pack_content_hash(pack_version: str = JUDGE_PACK_VERSION) -> str: """판정 팩의 **내용** 지문 — 버전 문자열이 아니라 로드되는 bytes. 2026-08-08 확인: 판정 팩 v7 을 만들었는데 최종 스틸 311장 중 253장이 재판정 없이 재사용됐다. 판정·수정 텍스트는 생성 프롬프트에 들어가지 않아서, 버전 '문자열' 스탬프는 팩 내용이 바뀌어도 안 움직일 수 있다 — 이 해시는 **이 프로세스가 실제 로드하는 디렉토리의 bytes** 를 지문에 접는다. ★낡은 서버(옛 셀렉터를 든 메모리) 탐지책은 아니다(Codex HIGH-2): 낡은 프로세스는 옛 팩을 정직하게 해시해 옛 기록과 일치한다 — 그건 "그 프로세스가 실제로 옛 팩으로 판정한다"는 뜻이라 지문으로는 옳고, 스테일 서버 여부는 재기동 시각 확인으로 따로 잡아야 한다. """ return _pack_dir_content_hash(resolve_judge_pack_version(pack_version)) def atomic_write_bytes(out_path: Path, data: bytes) -> Path: """tmp 생성→replace — write 도중 crash 로 truncated 파일이 '존재'로 오인돼 skip 되는 창 제거 (Codex 2차 리뷰 B2).""" out_path.parent.mkdir(parents=True, exist_ok=True) tmp = out_path.with_name(out_path.name + ".tmp") tmp.write_bytes(data) tmp.replace(out_path) return out_path # 매직 바이트 → media type. 확장자·파일명이 아니라 **내용**으로 판정한다. # 2026-08-06 실측: 참조 중 일부가 `.png` 이름을 달고 실제로는 JPEG 였다. # Gemini 는 media type 불일치를 조용히 넘겼지만 Anthropic 은 400 으로 거부한다 # ("specified using the image/png media type, but the image appears to be # a image/jpeg image"). 선정 판정을 Claude 로 옮기는 순간 전 호출이 죽는다. _IMAGE_MAGIC: Tuple[Tuple[bytes, str], ...] = ( (b"\x89PNG\r\n\x1a\n", "image/png"), (b"\xff\xd8\xff", "image/jpeg"), (b"GIF87a", "image/gif"), (b"GIF89a", "image/gif"), ) def _sniff_media_type(data: bytes) -> str: """바이트 내용으로 media type 판정 — 못 알아보면 png 로 둔다(기존 동작).""" for magic, mime in _IMAGE_MAGIC: if data.startswith(magic): return mime if data[:4] == b"RIFF" and data[8:12] == b"WEBP": return "image/webp" return "image/png" def png_part(source: Any) -> Dict[str, Any]: """이미지 소스(Path|bytes) → data URL 멀티모달 part. media type 은 **내용**으로 정한다 (`_sniff_media_type`). 이름이 `.png` 라도 실제가 JPEG 이면 `image/jpeg` 로 나간다. """ data = ( Path(source).read_bytes() if isinstance(source, (str, Path)) else source ) b64 = base64.b64encode(data).decode("ascii") return { "type": "image_url", "image_url": { "url": f"data:{_sniff_media_type(data)};base64,{b64}", }, } def resolve_judge_texts( roll_count: int, judge_name: str = "judge_still", pack_version: str = JUDGE_PACK_VERSION, ) -> Dict[str, str]: """multiroll_judge 팩 로드 — 후보 수 단어({count_word})만 포맷. v2 부터 명시 selector 로만 해석(암묵 latest-pack 승격 금지 — 팩 디렉토리 추가만으로 전 소비자 판정 계약이 바뀌던 창 제거). """ from app.modules.pipeline.multiroll_select import roll_labels from app.modules.prompt_loader import load_prompt resolved = resolve_judge_pack_version(pack_version) word = _COUNT_WORDS.get(roll_count, str(roll_count)) labels = ", ".join(roll_labels(roll_count)) return { "judge_sys": load_prompt( JUDGE_MODULE, judge_name, count_word=word, label_list=labels, version=resolved, ), "critique_sys": load_prompt( JUDGE_MODULE, "critique", count_word_lower=word.lower(), version=resolved, ), "fix_head": load_prompt( JUDGE_MODULE, "fix_head", version=resolved).strip(), "fix_tail": load_prompt( JUDGE_MODULE, "fix_tail", version=resolved).strip(), "fix_label": load_prompt( JUDGE_MODULE, "fix_label", version=resolved).strip(), # 팩 v7 신규 — 편집 호출에 동봉하는 참조의 라벨. v7 미만 팩에는 # 스템이 없으므로 빈 문자열(기존 동작), v7 이상에서는 fail-closed. "fix_ref_label": _load_optional( resolved, "fix_ref_label", pack_version), } # 이 스템들이 처음 들어온 팩 — 그보다 앞선 팩에는 없는 것이 정상이다. _STEM_INTRODUCED_IN = {"fix_ref_label": "7"} def _selector_num(selector: str) -> int: """팩 selector → 정수. 해석 불가는 -1(=도입 팩 미만으로 취급). -1 로 떨어뜨리는 쪽이 안전하다: 알 수 없는 selector 에 fail-closed 를 걸면 그 팩의 정상 소비까지 막힌다. 반대로 필수 스템 결손을 놓치는 위험은 `resolve_judge_pack_version` 이 selector 자체를 먼저 검증해 좁혀 준다. """ try: return int(str(selector).strip()) except (TypeError, ValueError): return -1 def _load_optional(resolved: str, name: str, selector: str) -> str: """구 팩에는 없을 수 있는 스템. **도입 팩 이상에서는 fail-closed.** 소비자는 빈 문자열을 "그 기능 없음"으로 읽는다. 구 selector 로 내렸을 때 부팅이 깨지지 않게 하려는 것이지, 스템 결손·손상을 삼키려는 것이 아니다. 삼키면 어떻게 되는지가 이번 결함의 형태다(Codex 리뷰 수용): 팩이 v7 인데 `fix_ref_label` 이 없거나 읽히지 않으면 편집이 조용히 참조 0장으로 떨어지고, **바로 그것을 고치려던 유료 수정이 옛 동작으로 실행된다.** 기록에는 v7 이라 남는다. 그래서 도입 팩 이상에서는 막는다. """ from app.modules.prompt_loader import load_prompt # ★버전은 **숫자**로 비교한다 (2026-08-07 Codex 3차 리뷰). 문자열로 # 비교하면 `"10" >= "7"` 이 False 라, v10 이상에서 필수 스템이 결손돼도 # fail-closed 가 풀려 유료 편집이 참조 없이 실행된다 — 기록에는 새 팩으로 # 남는다. 이 프로젝트의 selector 는 십진 정수 문자열이다. required_from = _STEM_INTRODUCED_IN.get(name) required = required_from is not None and _selector_num( selector) >= _selector_num(required_from) try: text = load_prompt(JUDGE_MODULE, name, version=resolved) except (FileNotFoundError, KeyError) as exc: if required: raise ValueError( f"multiroll_judge 팩 {resolved} 에 필수 스템 {name!r} 이 없다 " f"— 조용한 강등 금지(fail-closed)") from exc return "" text = (text or "").strip() if required and not text: raise ValueError( f"multiroll_judge 팩 {resolved} 의 {name!r} 이 비어 있다 " f"— 조용한 강등 금지(fail-closed)") return text def ref_parts(labeled_refs: Sequence[Tuple[str, Any]]) -> List[Dict[str, Any]]: parts: List[Dict[str, Any]] = [] for label, src in labeled_refs: parts.append({"type": "text", "text": f"REFERENCE — {label}"}) parts.append(png_part(src)) return parts # 참조 선별 (2026-08-19) — 취합자가 지적마다 "어느 참조를 봐야 하는가"를 # 번호로 답하려면 번호가 무엇을 가리키는지 알아야 한다. 위 ref_parts 는 # 라벨만 붙이고 번호를 안 붙이므로(모든 소비자 공용이라 손대지 않는다) # 취합 호출에만 이 목록을 덧붙인다. 순서는 labeled_refs 순서 그대로이고, # 그 순서가 `needs_ref_indices` 의 계약이다. _REF_INDEX_HEADER = ( "REFERENCE INDEX — the reference photographs above, numbered in the " "order they were given. Use these numbers, and only these, in " "`needs_ref_indices`:" ) def ref_index_part( labeled_refs: Sequence[Tuple[str, Any]] ) -> Dict[str, Any]: lines = [f"{i}. {label}" for i, (label, _src) in enumerate(labeled_refs, 1)] body = "\n".join(lines) if lines else "(none were attached)" return {"type": "text", "text": f"{_REF_INDEX_HEADER}\n{body}"} def make_nb2_gen_fn( *, project_id: str, episode_id: Optional[str] = None, operation_type: str = "multiroll_roll", aspect_ratio: str = "16:9", sanitizer: Any = None, gemini_client: Any = None, context_extra: Optional[Dict[str, Any]] = None, ) -> Callable: """nb2 라벨드 참조 생성 gen_fn — moderation 시 sanitizer 1회 재시도. context_extra: set_context 에 병합할 추가 trace 필드(still_id 등) — llm_call_log 구조키 매칭(generation_call_id resolve)의 전제. still_safety_fallback_enabled=True 면 moderation 거부에 다단 후퇴 사다리(2026-08-16 사용자 지시): 원 모델 재시도 → 교차 백엔드 (Gemini↔Grok) → 연화 → 교차+연화. OFF(default)=기존 2회(원문→연화) byte-identical. 사다리가 삼키는 오류는 moderation 과 GrokPromptOverBudget(무호출 비호환)뿐 — 그 밖의 오류·예산 소진은 어느 단계든 즉시 전파(Codex BLOCK-3). """ if gemini_client is None: # 최종 스틸 생성 엔진 스위치 (2026-08-13 사용자 확정 "grok 2 로 # 가자"): 기본 nb2(Gemini) — byte-identical. "grok2" 면 OpenRouter # 경유 xAI Grok Imagine 2.0 (동일 gen_fn 계약: set_context/ # generate_image/labeled_references/기록·캡처). from app.core.config import settings as _sw_settings if getattr(_sw_settings, "still_image_backend", "nb2") == "grok2": from app.modules.llm.grok_image_client import GrokImageClient gemini_client = GrokImageClient() else: from app.modules.llm.gemini_image_client import GeminiImageClient gemini_client = GeminiImageClient() def _moderated(exc: Exception) -> bool: msg = str(exc).lower() return ( "moderation" in msg or "safety" in msg or "blocked" in msg or "content_policy" in msg ) def gen_fn(tag, prompt, labeled_refs, out_path: Path) -> Path: current = prompt for attempt in (1, 2): try: gemini_client.set_context( project_id=project_id, episode_id=episode_id, operation_type=operation_type, # Codex 8e70d4c0 H2: 호출별 롤/fix 태그를 구조키로 기록 # (예: still_S4sh1_ab_conti_a / ..._fix) — 최종 asset # 감사 링크가 selected roll/fix 를 exact 로 가리키는 # 전제. context_extra 가 같은 키를 주면 그쪽 우선. **{"multiroll_tag": tag, **(context_extra or {})}, ) png, _ms = gemini_client.generate_image( current, labeled_references=[ (label, Path(p).read_bytes() if isinstance(p, (str, Path)) else p) for label, p in labeled_refs ], aspect_ratio=aspect_ratio, ) return atomic_write_bytes(out_path, png) except Exception as exc: # noqa: BLE001 if attempt == 1 and _moderated(exc) and sanitizer is not None: sr = sanitizer.sanitize(current, str(exc), [], attempt) sanitized = sr.get("sanitized_prompt") or "" if sanitized: logger.warning( "multiroll_gemini %s: moderation — sanitize " "재시도", tag, ) current = sanitized continue raise raise RuntimeError("unreachable") def gen_fn_ladder(tag, prompt, labeled_refs, out_path: Path) -> Path: """SAFETY 다단 후퇴 — 단계: primary → 원 모델 재시도 → 교차 백엔드 → 연화(원 모델) → 연화(교차). 연화 저작은 1회만(단계 4 직전), 교차 클라이언트도 필요할 때 1회만 만든다. 비용 경계(Codex BLOCK-3): 사다리 **단계** 상한 5. 단계당 provider 재시도는 클라이언트 규칙 그대로다 — moderation 은 내부 재시도 없이 즉시 올라오고(정상 응답의 차단 사유), transient HTTP 만 최대 llm_max_retries. 사다리가 삼키는 오류는 moderation 과 GrokPromptOverBudget(무호출·결정론적 비호환) 둘뿐 — 그 밖의 오류와 ImageCallBudgetExceeded(batch abort 계약)는 어느 단계든 즉시 전파. """ from app.core.image_call_budget import ImageCallBudgetExceeded from app.modules.llm.grok_image_client import GrokPromptOverBudget refs = [ (label, Path(p).read_bytes() if isinstance(p, (str, Path)) else p) for label, p in labeled_refs ] def _cross_client() -> Any: # 교차 백엔드 = 기본 클라이언트와 다른 검열 스택. ★상속 관계 # (Grok ⊂ Gemini)라 Grok 판별을 먼저 한다. from app.modules.llm.gemini_image_client import GeminiImageClient from app.modules.llm.grok_image_client import GrokImageClient if isinstance(gemini_client, GrokImageClient): return GeminiImageClient() return GrokImageClient() # (stage, 교차 여부, 연화 여부) — stage 문자열은 llm_call_log # metadata(safety_ladder)로 남아 발화 흔적의 SOT 가 된다. stages = ( ("primary", False, False), ("same_model_retry", False, False), ("cross_backend", True, False), ("sanitized", False, True), ("cross_sanitized", True, True), ) cross: Any = None sanitized: Optional[str] = None last_exc: Optional[Exception] = None moderation_exc: Optional[Exception] = None for stage, use_cross, use_sanitized in stages: if use_sanitized and sanitized is None: if sanitizer is None: break sr = sanitizer.sanitize( prompt, str(moderation_exc or last_exc), [], 1) sanitized = (sr.get("sanitized_prompt") or "").strip() if not sanitized: break # 연화 실패 — 남은 단계가 전부 연화 의존 client = gemini_client if use_cross: if cross is None: cross = _cross_client() client = cross try: extra = {"multiroll_tag": tag, **(context_extra or {})} # (Codex BLOCK-1) set_context 는 update 라 이전 샷의 # 단계명이 공유 클라이언트에 잔류한다 — primary 도 항상 # 키를 실어 None 으로 명시 청소(projection 이 None 을 # 걸러 메타에는 안 남는다). extra["safety_ladder"] = ( stage if stage != "primary" else None) client.set_context( project_id=project_id, episode_id=episode_id, operation_type=operation_type, **extra) png, _ms = client.generate_image( sanitized if use_sanitized else prompt, labeled_references=refs, aspect_ratio=aspect_ratio) if stage != "primary": logger.warning( "multiroll_gemini %s: safety ladder %s 단계에서 " "회복", tag, stage) return atomic_write_bytes(out_path, png) except ImageCallBudgetExceeded: # (Codex BLOCK-3a) 예산 소진은 batch abort 계약 — 어느 # 단계든 즉시 전파, 유료 연화·후속 단계 진행 금지. raise except GrokPromptOverBudget as exc: # 무호출·결정론적 백엔드 비호환(8k 상한). (Codex R2 # BLOCK) moderation 으로 후퇴 국면에 들어와 있을 때만 # 이 단계를 건너뛴다 — moderation 이 한 번도 없었는데 # 삼키면 SAFETY 무관 입력(primary Grok+긴 프롬프트)에서 # 유료 교차가 발화해 OFF(즉시 실패)와 어긋나고, 기록의 # safety_ladder 도 SAFETY 회복이 아닌데 그렇게 읽힌다. if moderation_exc is None: raise last_exc = exc logger.warning( "multiroll_gemini %s: safety ladder %s 단계 백엔드 " "비호환(%s) — 다음 단계", tag, stage, str(exc)[:200]) except Exception as exc: # noqa: BLE001 if not _moderated(exc): # (Codex BLOCK-3b) moderation 밖 오류는 어느 단계든 # 즉시 전파 — 삼키면 실패 원인이 오독되고 provider # 내부 재시도(transient 최대 4 HTTP)와 곱해져 비용 # 창이 열린다. raise last_exc = exc moderation_exc = exc logger.warning( "multiroll_gemini %s: safety ladder %s 단계 " "moderation(%s) — 다음 단계", tag, stage, str(exc)[:200]) # 전 단계 소진 — 걷기 단위 실패 격리로 전파. 삼킨 오류는 # moderation/백엔드 비호환뿐이므로 원 사유(moderation)를 올린다. raise (moderation_exc or last_exc or RuntimeError("safety ladder: no attempt ran")) from app.core.config import settings as _lad_settings if bool(getattr(_lad_settings, "still_safety_fallback_enabled", False)): return gen_fn_ladder return gen_fn def make_gemini_judge_fn( *, judge_sys: str, judge_schema: Dict[str, Any], project_config: Optional[Dict[str, Any]] = None, step_tag: str = "multiroll_judge", opik_metadata: Optional[Dict[str, Any]] = None, prompt_header: str = ( "THE PROMPT (all candidates were generated from this):" ), ) -> Callable: """prompt_header (seed 품질 2R): 변형 롤 모드는 후보들이 서로 다른 저작 프롬프트로 생성되므로 기존 헤더 문구가 거짓 — 브리프용 헤더로 교체 가능. 미지정=기존 문구 byte-identical. ★모델: 후보 선정 판정이므로 `resolve_select_judge_models()` 를 쓴다 — G+Q ON 이면 **Gemini + Qwen 동시**(합의 규칙 `_judge_gq`), OFF 는 ANTHROPIC_API_KEY 가 있으면 Claude Opus + GPT-5.6 Sol 조건부 이중 (`_judge_dual`/`combine_select_verdicts`), 없으면 `gemini-pro` 단독.""" from app.modules.llm.llm_client import call_structured models = resolve_select_judge_models() base_pc = dict(project_config or {}) # QK 반전 구성(메인=Qwen 센티널이 앞) 여부 — Gemini 보조 호출 봉인에 # 쓴다 (Codex 반전 리뷰 BLOCK-2). 기존 GQ(Gemini 앞)는 기본 fallback # 유지 = byte-identical. _qk_reversed = bool(models) and models[0] == QWEN_JUDGE_MODEL # G+G46 동등 구성(Gemini 앞 + OpenRouter 판정 동승) — 같은 봉인을 # 태어날 때부터 적용한다: 동등 합의에서 GPT 강등 판정이 per_model_ # winner['gemini-pro']·물리 지문에 Gemini 로 남는 오귀속은 QK BLOCK-2 # 와 동류다(정책 문자열 no_fallback 이 이 계약을 잠근다). _gg46_equal = bool(models) and models[0] == JUDGE_MODEL and any( m.startswith(OPENROUTER_JUDGE_PREFIX) for m in models[1:]) def _one(model: str, parts, tag_suffix: str) -> Dict[str, Any]: tag = f"{step_tag}{tag_suffix}" if model == QWEN_JUDGE_MODEL: # Qwen 은 Router 미등록(json_schema 미지원) — 우회 클라이언트로. # 판정 sys·schema 는 Gemini 와 **공용**(같은 계약, 같은 후보 제시). from app.modules.llm.qwen_vlm_client import ask_qwen_structured return ask_qwen_structured( tag, judge_sys, parts, judge_schema, opik_metadata=opik_metadata, ) if model.startswith(OPENROUTER_JUDGE_PREFIX): # QK 판정(2026-08-12) — OpenRouter 경유. 판정 sys·schema 는 # 기존 심판과 **공용**(같은 계약, 같은 후보 제시). from app.modules.llm.openrouter_vlm_client import ( ask_openrouter_structured, ) return ask_openrouter_structured( tag, judge_sys, parts, judge_schema, model=model[len(OPENROUTER_JUDGE_PREFIX):], # 파일럿 검증 구성(2026-08-13): 판정류 긴 구조화 출력은 # max_tokens 미달 잘림이 실측 함정(qwen 4000 잘림 14건) — # grok-4.6 23샷 파일럿도 8000 으로 통과했다. 상한이 아니라 # 출력 예산이다(reasoning 은 별도). max_tokens=8000, opik_metadata=opik_metadata, ) # QK 반전의 Gemini 보조는 **Gemini 로 봉인**(enable_fallback=False — # Codex BLOCK-2): safety 강등 시 GPT Tier3 가 판정했는데 기록 # (per_model_winner['gemini-pro'])과 지문(물리 쌍)은 Gemini 로 남는 # 오귀속 차단. 실패는 _judge_gq 가 정직한 single_qwen-vlm route 로 # 처리한다. 기존 GQ 경로는 기본값 유지(byte-identical). return call_structured( tag, judge_sys, parts, judge_schema, project_config={**base_pc, tag: {"model": model}}, schema_name=step_tag, opik_metadata=opik_metadata, **({"enable_fallback": False} if ((_qk_reversed or _gg46_equal) and model == JUDGE_MODEL) else {}), ) def judge_fn(tag, prompt, labeled_refs, cand_paths, labels): parts: List[Dict[str, Any]] = [{ "type": "text", "text": prompt_header + "\n" + prompt, }] parts += ref_parts(labeled_refs) for lab, p in zip(labels, cand_paths): parts.append({"type": "text", "text": f"Candidate {lab}:"}) parts.append(png_part(p)) if len(models) > 1: # 모델 구성으로 가른다 — 슬롯 순서가 우선권 계약(models[0]= # 우선·취합)이다. Qwen 센티널이 **앞**이면 QK(Qwen 메인, # route=qwen_priority — 2026-08-12 반전), 뒤면 기존 G+Q # (Gemini 우선). OpenRouter 센티널도 QK 수학 공유(보류 중). # 아니면 기존 조건부 이중(margin-skip). 플래그 재조회보다 # 순수하다. if models[0] == QWEN_JUDGE_MODEL: return _judge_gq(models, _one, parts, list(labels), priority_route="qwen_priority") if QWEN_JUDGE_MODEL in models: return _judge_gq(models, _one, parts, list(labels)) if models[0] == JUDGE_MODEL and any( m.startswith(OPENROUTER_JUDGE_PREFIX) for m in models[1:]): # G+G46 (2026-08-13): Gemini 앞 + OpenRouter 판정 동승 = # 동등 이중 — 우선권 없음, 불일치=combined 합산. return _judge_gq(models, _one, parts, list(labels), equal_disagree_combined=True) if any(m.startswith(OPENROUTER_JUDGE_PREFIX) for m in models): return _judge_gq(models, _one, parts, list(labels), priority_route="qwen_priority") return _judge_dual(models, _one, parts, list(labels)) return call_structured( step_tag, judge_sys, parts, judge_schema, project_config={**base_pc, step_tag: {"model": models[0]}}, schema_name=step_tag, opik_metadata=opik_metadata, ) return judge_fn def make_ref_pick_judge_fn( *, project_config: Optional[Dict[str, Any]] = None, step_tag: str = "structure_seed_ref_pick", opik_metadata: Optional[Dict[str, Any]] = None, ) -> Callable: """참조 사진 대비 **상대 비교** 판정을 multiroll JudgeFn 계약으로 잇는다. ★판정에 브리프를 주지 않는다. 지금까지는 브리프 대비 절대 점수였는데, 참조 사진에 그 종류의 시설이 갖춘 것이 다 보여도 산출이 그것을 통째로 빠뜨린 채 통과했다(실측: 참조는 실제 점포인데 산출은 빈 띠에 글자만). 기준은 참조 사진 한 장이다. ★LLM 이 보는 칸은 승자와 사유 둘뿐이고, multiroll 이 요구하는 {winner, ranking, verdicts} 는 코드가 채운다. 점수 칸을 주면 문턱이 되살아나고, 칸을 늘릴수록 판정이 나빠진다. """ from app.modules.llm.llm_client import call_structured from app.modules.pipeline.search_grounded_ref import ( build_ref_pick_schema, load_ref_pick_system, ) pc = {**(project_config or {}), step_tag: {"model": JUDGE_MODEL}} def _as_judge(labs: Sequence[str], winner: str, reason: str) -> Dict[str, Any]: """승패를 1/0 으로만 옮긴다 — 정도를 매기지 않는다. 좌우를 바꿔 두 번 물을 때(judge_flip) 두 번 다 이긴 쪽만 2점이 되고, 한 번씩 이기면 동점이 돼 우선순위가 정한다. """ rest = [lab for lab in labs if lab != winner] return { "winner": winner, "ranking": [winner, *rest], "verdicts": [ {"label": lab, "score": 1 if lab == winner else 0, "verdict_ko": reason if lab == winner else ""} for lab in labs ], } def judge_fn(tag, prompt, labeled_refs, cand_paths, labels): labs = list(labels) if not labs: raise ValueError("ref_pick: 후보 라벨이 비었다") if len(labs) < 2: # 비교할 상대가 없다 — 유료 호출 없이 그 후보가 그대로 간다. return _as_judge(labs, labs[0], "후보가 하나뿐 — 비교 없음") parts: List[Dict[str, Any]] = ref_parts(labeled_refs) for lab, p in zip(labs, cand_paths): parts.append({"type": "text", "text": f"{lab}:"}) parts.append(png_part(p)) data = call_structured( step_tag, load_ref_pick_system(label_list=", ".join(labs)), parts, build_ref_pick_schema(labs), project_config=pc, schema_name=step_tag, opik_metadata=opik_metadata, ) winner = str(data.get("winner") or "") if winner not in labs: # 조용히 첫 후보로 떨어뜨리지 않는다 — 판정이 안 된 것을 판정된 # 것으로 기록하면 어느 경로가 이겼는지가 거짓이 된다. raise ValueError( f"ref_pick: 판정이 라벨 밖을 골랐다 {winner!r} (가능 {labs})") return _as_judge(labs, winner, str(data.get("reason_ko") or "")) return judge_fn def make_gemini_critique_fn( *, critique_sys: str, critique_schema: Dict[str, Any], project_config: Optional[Dict[str, Any]] = None, step_tag: str = "multiroll_critique", opik_metadata: Optional[Dict[str, Any]] = None, ref_gate: bool = False, ) -> Callable: from app.modules.llm.llm_client import call_structured pc = {**(project_config or {}), step_tag: {"model": JUDGE_MODEL}} def critique_fn(tag, prompt, labeled_refs, image_path): parts: List[Dict[str, Any]] = [ {"type": "text", "text": "THE PROMPT:\n" + prompt} ] parts += ref_parts(labeled_refs) parts.append({"type": "text", "text": "Photograph to examine:"}) parts.append(png_part(image_path)) if ref_gate: parts.append(ref_index_part(labeled_refs)) return call_structured( step_tag, critique_sys, parts, critique_schema, project_config=pc, schema_name=step_tag, opik_metadata=opik_metadata, ) return critique_fn def make_gq_critique_fn( *, critique_schema: Dict[str, Any], project_config: Optional[Dict[str, Any]] = None, step_tag: str = "multiroll_critique", opik_metadata: Optional[Dict[str, Any]] = None, pack_version: str = GQ_CRITIQUE_PACK_VERSION, ref_gate: bool = False, ) -> Callable: """G+Q 수정 흐름 critique — Qwen 관찰 → Gemini 취합 2단 (2026-08-10 확정). CritiqueFn 시그니처 동일 — 하류(`_critique_and_fix` 의 issue partition· fix 조립·fix-rejudge·nb2)는 무변경이다. 현행 단일 critique 는 관찰과 수정문 작성을 겸임하는데, 여기서는 역할을 가른다: 1. **Qwen 관찰**: 무엇이 잘못됐는가만(수정문 없음 — 실측상 관찰·셈이 강점, `matches_brief` 는 못 믿는 심판이라 관찰자 자리가 맞다). 2. **관찰 0건 = 즉시 무결 반환** — Gemini 호출 생략(유료 1콜 절약). "Qwen 단독 관찰" 합의의 의미: Qwen 이 못 본 결함은 이 흐름에서 고치지 않는다. 3. **Gemini 취합**: 관찰을 이미지와 대조해 기각할 수 있고, 채택분만 현행 critique 계약(issues: fix_en/unfixable/needs_regeneration)으로 옮긴다 — 하류 소비 계약 불변. 반환 dict 에 `qwen_observations`(관찰 원본)를 병기한다 — `record["critique"]` 에 접혀, 취합이 무엇을 기각했는지 대조로 추적된다. """ from app.modules.llm.llm_client import call_structured from app.modules.llm.qwen_vlm_client import ask_qwen_structured from app.modules.pipeline.multiroll_select import build_gq_observe_schema from app.modules.prompt_loader import load_prompt resolved = resolve_judge_pack_version(pack_version) observe_sys = load_prompt( JUDGE_MODULE, "gq_observe_sys", version=resolved) compose_sys = load_prompt( JUDGE_MODULE, "gq_compose_sys", version=resolved) observe_schema = build_gq_observe_schema() compose_tag = f"{step_tag}_compose" pc = {**(project_config or {}), compose_tag: {"model": JUDGE_MODEL}} def critique_fn(tag, prompt, labeled_refs, image_path): parts: List[Dict[str, Any]] = [ {"type": "text", "text": "THE PROMPT:\n" + prompt} ] parts += ref_parts(labeled_refs) parts.append({"type": "text", "text": "Photograph to examine:"}) parts.append(png_part(image_path)) obs = ask_qwen_structured( f"{step_tag}_observe", observe_sys, parts, observe_schema, opik_metadata=opik_metadata, ) observations = list(obs.get("observations") or []) if not observations: return {"issues": [], "qwen_observations": []} compose_parts = list(parts) if ref_gate: compose_parts.append(ref_index_part(labeled_refs)) compose_parts.append({ "type": "text", "text": ( "OBSERVATIONS (from a separate visual inspector — verify " "each against the photograph before adopting it):\n" + json.dumps(observations, ensure_ascii=False, indent=1) ), }) data = call_structured( compose_tag, compose_sys, compose_parts, critique_schema, project_config=pc, schema_name=compose_tag, opik_metadata=opik_metadata, ) out = dict(data) out["qwen_observations"] = observations return out return critique_fn def make_qk_critique_fn( *, critique_schema: Dict[str, Any], project_config: Optional[Dict[str, Any]] = None, opik_metadata: Optional[Dict[str, Any]] = None, step_tag: str = "multiroll_critique", pack_version: str = QK_CRITIQUE_PACK_VERSION, ref_gate: bool = False, ) -> Callable: """QK 수정 흐름 critique — Gemini 관찰 → Qwen 취합 2단. `make_gq_critique_fn` 과 같은 2단 계약(관찰 0건=즉시 무결 반환·취합이 관찰을 이미지 대조로 기각 가능·하류 소비 계약 불변) — **자리를 반전** 한다: 관찰(보조 자리)=Gemini(기존 G+Q 의 우선 모델), 취합(메인 자리)= Qwen(DashScope 직결). 2026-08-12 사용자 재결정 — 처음 설계(Kimi 관찰) 는 OpenRouter 지연 실측(관찰 평균 310s)으로 보류. 팩은 QK 도입판 v9(STAGING 관찰 축 포함)가 기본 — QK 는 새 체계라 물려 받을 구 지문이 없고, GQ(v8 고정)의 지문은 1 byte 도 건드리지 않는다. 반환 dict 에 `observer_observations`(관찰 원본)+`observer_model` 을 병기한다 — 관찰자가 Qwen 이 아니므로 GQ 의 `qwen_observations` 키를 재사용하지 않는다(기록 오독 방지). """ from app.modules.llm.llm_client import call_structured from app.modules.llm.qwen_vlm_client import ask_qwen_structured from app.modules.pipeline.multiroll_select import build_gq_observe_schema from app.modules.prompt_loader import load_prompt resolved = resolve_judge_pack_version(pack_version) observe_sys = load_prompt( JUDGE_MODULE, "gq_observe_sys", version=resolved) compose_sys = load_prompt( JUDGE_MODULE, "gq_compose_sys", version=resolved) observe_schema = build_gq_observe_schema() observe_tag = f"{step_tag}_observe" pc = {**(project_config or {}), observe_tag: {"model": JUDGE_MODEL}} def critique_fn(tag, prompt, labeled_refs, image_path): parts: List[Dict[str, Any]] = [ {"type": "text", "text": "THE PROMPT:\n" + prompt} ] parts += ref_parts(labeled_refs) parts.append({"type": "text", "text": "Photograph to examine:"}) parts.append(png_part(image_path)) # 관찰자도 Gemini 로 봉인 (Codex BLOCK-2): safety 강등으로 GPT 가 # 관찰했는데 observer_model='gemini-pro' 로 영속되는 오귀속 차단. # 실패는 삼키지 않고 전파한다(관찰자 대체 없음 — 기록 정직성). obs = call_structured( observe_tag, observe_sys, parts, observe_schema, project_config=pc, schema_name=observe_tag, opik_metadata=opik_metadata, enable_fallback=False, ) observations = list(obs.get("observations") or []) if not observations: return {"issues": [], "observer_observations": [], "observer_model": JUDGE_MODEL} compose_parts = list(parts) if ref_gate: compose_parts.append(ref_index_part(labeled_refs)) compose_parts.append({ "type": "text", "text": ( "OBSERVATIONS (from a separate visual inspector — verify " "each against the photograph before adopting it):\n" + json.dumps(observations, ensure_ascii=False, indent=1) ), }) data = ask_qwen_structured( f"{step_tag}_compose", compose_sys, compose_parts, critique_schema, opik_metadata=opik_metadata, ) out = dict(data) # ── severity 재결속 (2026-08-13 Codex R1 BLOCK-1) ──────────── # 심각도의 SOT 는 관찰자다. 취합(Qwen)이 major 를 critical 로 # 승격하면 유료 fix 가 도로 발동한다(S39sh4 부류 재발) — 산문 # 규칙이 아니라 코드가 지킨다: 각 이슈의 observation_index 로 # 관찰 원값을 되찾아 severity 를 덮어쓰고, 인덱스가 실제 관찰을 # 가리키지 않는 이슈(발명·번호 오류)는 fail-closed 로 버린다. # severity 미사용 스키마(구 경로 v7/GQ v8)는 이 블록을 안 탄다. issues = list(out.get("issues") or []) if any("severity" in i for i in issues if isinstance(i, dict)): bound: List[Dict[str, Any]] = [] dropped: List[Dict[str, Any]] = [] rebound = 0 for issue in issues: idx = issue.get("observation_index") if not isinstance(idx, int) or isinstance(idx, bool) \ or not (0 <= idx < len(observations)): dropped.append(issue) continue obs_sev = (observations[idx] or {}).get("severity") if obs_sev and issue.get("severity") != obs_sev: issue = {**issue, "severity": obs_sev} rebound += 1 bound.append(issue) out["issues"] = bound if dropped: out["compose_dropped_unbound"] = dropped if rebound: out["compose_severity_rebound_count"] = rebound out["observer_observations"] = observations out["observer_model"] = JUDGE_MODEL return out return critique_fn def make_gg46_critique_fn( *, critique_schema: Dict[str, Any], project_config: Optional[Dict[str, Any]] = None, opik_metadata: Optional[Dict[str, Any]] = None, step_tag: str = "multiroll_critique", pack_version: str = GG46_CRITIQUE_PACK_VERSION, ref_gate: bool = False, ) -> Callable: """G+G46 수정 흐름 critique — **양쪽 관찰**(Gemini+grok-4.6) → Gemini 취합 2단 (2026-08-13 사용자 확정 "수정사항을 취합해서"). GQ/QK 와 같은 2단 계약(관찰 0건=즉시 무결 반환·취합이 관찰을 이미지 대조로 기각 가능·하류 소비 계약 불변)에서 **관찰자가 둘**이다: 파일럿 실측(23샷)에서 두 관찰자의 강점이 달랐다 — grok 만 S64sh4 좌석 물리를, 관찰 밀도는 gemini 가 앞섰다. 병합=관찰 합집합이라 어느 한쪽 눈만으로는 못 잡던 결함이 취합 검증대에 오른다. 취합자=Gemini 인 이유(동등 원칙과의 관계 명문): 취합은 승자를 정하는 판정이 아니라 관찰을 사진과 대조해 기각·번역하는 역할이라 동등 계약 밖이다. grok-4.6 취합은 결함 샷마다 왕복 ~80-140s·~$0.04 를 더 내는데 파일럿에서 취합 품질 우위 근거가 없다. severity SOT 는 어차피 관찰자 (재결속 코드가 강제)라 취합자 선택이 심각도를 좌우하지 않는다. 계약 세부: - 병합 목록 = [Gemini 관찰..., grok 관찰...] (모델 슬롯 순서 고정) — compose 의 `observation_index` 와 severity 재결속이 **병합 목록** 기준으로 왕복한다. - 한쪽 관찰 실패 = 생존 관찰로 진행 + `observer_failed` 기록 (판정 `single_*` route 관례 동형 — 기록이 정직하면 강등이 사고가 아니다). 양쪽 실패 = 마지막 원인을 달아 전파. - 관찰자·취합자 Gemini 호출은 enable_fallback=False 봉인 (QK BLOCK-2 관례 — GPT 강등 판정이 Gemini 로 영속되는 오귀속 차단). - 반환 dict: `observer_observations`=병합 목록(인덱스 SOT), `observer_models`, `observer_counts`, (있을 때만) `observer_failed`. GQ(`qwen_observations`)·QK(`observer_model` 단수) 키와 구분된다. """ from app.modules.llm.llm_client import call_structured from app.modules.llm.openrouter_vlm_client import ( ask_openrouter_structured, ) from app.modules.pipeline.multiroll_select import build_gq_observe_schema from app.modules.prompt_loader import load_prompt from app.core.config import settings resolved = resolve_judge_pack_version(pack_version) observe_sys = load_prompt( JUDGE_MODULE, "gq_observe_sys", version=resolved) compose_sys = load_prompt( JUDGE_MODULE, "gq_compose_sys", version=resolved) observe_schema = build_gq_observe_schema() grok_model = (getattr(settings, "grok_judge_model", "") or "").strip() observer_models = [ JUDGE_MODEL, f"{OPENROUTER_JUDGE_PREFIX}{grok_model}"] observe_tag_g = f"{step_tag}_observe" observe_tag_x = f"{step_tag}_observe_grok46" compose_tag = f"{step_tag}_compose" pc = {**(project_config or {}), observe_tag_g: {"model": JUDGE_MODEL}, compose_tag: {"model": JUDGE_MODEL}} def critique_fn(tag, prompt, labeled_refs, image_path): parts: List[Dict[str, Any]] = [ {"type": "text", "text": "THE PROMPT:\n" + prompt} ] parts += ref_parts(labeled_refs) parts.append({"type": "text", "text": "Photograph to examine:"}) parts.append(png_part(image_path)) per_model: Dict[str, List[Dict[str, Any]]] = {} failed: List[str] = [] last_exc: Optional[BaseException] = None for model in observer_models: try: if model == JUDGE_MODEL: obs = call_structured( observe_tag_g, observe_sys, parts, observe_schema, project_config=pc, schema_name=observe_tag_g, opik_metadata=opik_metadata, enable_fallback=False, ) else: obs = ask_openrouter_structured( observe_tag_x, observe_sys, parts, observe_schema, model=model[len(OPENROUTER_JUDGE_PREFIX):], max_tokens=8000, opik_metadata=opik_metadata, ) per_model[model] = list(obs.get("observations") or []) except Exception as exc: # noqa: BLE001 last_exc = exc failed.append(model) logger.warning( "G+G46 관찰: %s 실패 — 남은 관찰자로 진행: %r", model, exc) if len(failed) == len(observer_models): raise RuntimeError("G+G46 관찰 전건 실패") from last_exc # 병합 순서=모델 슬롯 순서 고정 — 인덱스가 계약이다. observations: List[Dict[str, Any]] = [] for model in observer_models: observations.extend(per_model.get(model, [])) base_record = { "observer_observations": observations, "observer_models": list(observer_models), "observer_counts": { m: len(per_model.get(m, [])) for m in observer_models}, } if failed: base_record["observer_failed"] = failed if not observations: return {"issues": [], **base_record} compose_parts = list(parts) if ref_gate: compose_parts.append(ref_index_part(labeled_refs)) compose_parts.append({ "type": "text", "text": ( "OBSERVATIONS (concatenated from TWO independent visual " "inspectors — entries may report the same fault twice; " "verify each against the photograph before adopting it, " "and adopt each distinct fault at most once):\n" + json.dumps(observations, ensure_ascii=False, indent=1) ), }) data = call_structured( compose_tag, compose_sys, compose_parts, critique_schema, project_config=pc, schema_name=compose_tag, opik_metadata=opik_metadata, enable_fallback=False, ) out = dict(data) # ── severity 재결속 (QK Codex R1 BLOCK-1 관례 이식) ──────────── # 심각도의 SOT 는 관찰자다 — 취합이 승격/강등하거나 관찰에 없는 # 이슈를 발명하면 코드가 병합 목록 인덱스로 원값을 되찾아 덮어 # 쓰고, 미등록 인덱스는 fail-closed 로 버린다. issues = list(out.get("issues") or []) if any("severity" in i for i in issues if isinstance(i, dict)): bound: List[Dict[str, Any]] = [] dropped: List[Dict[str, Any]] = [] rebound = 0 for issue in issues: idx = issue.get("observation_index") if not isinstance(idx, int) or isinstance(idx, bool) \ or not (0 <= idx < len(observations)): dropped.append(issue) continue obs_sev = (observations[idx] or {}).get("severity") if obs_sev and issue.get("severity") != obs_sev: issue = {**issue, "severity": obs_sev} rebound += 1 bound.append(issue) out["issues"] = bound if dropped: out["compose_dropped_unbound"] = dropped if rebound: out["compose_severity_rebound_count"] = rebound out.update(base_record) return out return critique_fn def make_gpt_composition_critique_fn( *, critique_schema: Dict[str, Any], project_config: Optional[Dict[str, Any]] = None, step_tag: str = "multiroll_gpt_composition", opik_metadata: Optional[Dict[str, Any]] = None, pack_version: str = GPT_COMPOSITION_PACK_VERSION, ref_gate: bool = False, ) -> Callable: """GPT 구도 전담 critique 어댑터 (E2E11 fix③, 사용자 확정). CritiqueFn 시그니처 동일 — Gemini critique 와 별개로 선정본의 **구도 위반만** 검사(sys 계약이 정체성/재질/조명 OUT OF SCOPE 봉인). 반환 issues 는 Gemini issues 와 합산돼 하나의 수정 프롬프트로 i2i 된다. 모델=GPT LVM(모델 분업: 이미지 검증·구도=GPT). """ from app.modules.llm.llm_client import call_structured from app.modules.prompt_loader import load_prompt comp_sys = load_prompt( JUDGE_MODULE, "gpt_composition_sys", version=resolve_judge_pack_version(pack_version), ) pc = {**(project_config or {}), step_tag: {"model": GPT_COMPOSITION_MODEL}} def composition_fn(tag, prompt, labeled_refs, image_path): parts: List[Dict[str, Any]] = [ {"type": "text", "text": "THE GENERATION CONTRACT:\n" + prompt} ] parts += ref_parts(labeled_refs) parts.append( {"type": "text", "text": "Selected photograph to inspect:"}) parts.append(png_part(image_path)) # 참조 선별 — 이 갈래도 칸을 채워야 한다. 안 채우면 합산 목록에 # 칸 없는 지적이 섞여 `select_fix_refs` 가 통째로 거르기를 그만두고 # (fail-open), 지문에는 "선별 켜짐"으로 남아 조용히 무력해진다. if ref_gate: parts.append(ref_index_part(labeled_refs)) return call_structured( step_tag, comp_sys, parts, critique_schema, project_config=pc, schema_name=step_tag, opik_metadata=opik_metadata, ) return composition_fn