#!/usr/bin/env python3 """**결함을 찾는 지시문**으로 물으면 달라지나 — 임시 측정 도구 (2026-08-29). ## 왜 만들었나 지금까지 Grok 을 물은 것은 전부 `judge_still.md` — **둘 중 하나를 고르는** 지시문이었다. 거기서 Grok 은 이 주행 내내 `hard_violations` 를 한 건도 안 냈다(Gemini 는 7건). 그런데 같은 팩 안에 **결함을 찾는 것이 본업인** 지시문이 따로 있다 — `critique.md`. 그 글의 critical 목록에는 a duplicated fitting (two of something the space has one of) 가 그대로 적혀 있고, 이것이 바로 Gemini 가 S1sh4 후보 A 에서 잡은 「단 하나로 명시된 형광등이 여러 개」다. 그러니 물어야 할 것은 **모델이 못 보는 것인가, 시키는 일이 달랐던 것인가**다. ## 프로덕션과 같게 맞춘 것 조립은 `make_gemini_critique_fn` 을 그대로 옮겼다 — `"THE PROMPT:\\n"+prompt` → `ref_parts(labeled_refs)` → `"Photograph to examine:"` → 검사 대상 한 장. 지시문·스키마도 같은 리졸버에서 가져온다. ## 다른 것 — 여기서 밝혀 적는다 · **모델**: 프로덕션 critique 는 Gemini 자리다. 여기서는 Grok 도 같은 자리에 세워 본다(`--model=grok`). · **severity**: `critique.md` 는 「Rate every issue in `severity`」라고 쓰는데 프로덕션의 기본 갈래(`make_gemini_critique_fn`)는 `build_critique_schema(with_ref_gate=…)` 로 **severity 없는 스키마**를 준다. 이 도구는 `--severity` 로 고를 수 있게 두고 기본을 켜 둔다 — 지시문이 쓰라는 칸을 스키마가 안 주면 무엇을 재는지 흐려진다. ★참조가 하나라도 안 풀리면 **멈춘다.** 조용히 적게 보내면 재는 대상이 달라진다(전에 `ref_labels` 를 잘라 보내서 프레이밍 절을 지운 적이 있다). usage: critique_probe.py --tag=S1sh4 --label=A [--model=grok|gemini] [--effort=high] [--no-severity] [--ref-gate] --run (기본 dry) """ from __future__ import annotations import json import pathlib import sys import time sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[2])) sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent)) import _db # noqa: E402 ROOT = pathlib.Path(__file__).resolve().parents[3] OUT = ROOT / "artifact" / "20260829_critique_probe" / "calls.json" SEED = 20260829 def _asset_path(asset_id: str) -> pathlib.Path: """`` 로 남은 참조는 asset_id 로 실파일을 찾는다. ★접속 정보는 설정에서만 온다(`_db`) — 비밀번호를 도구에 안 적는다. """ got = _db.one_col( "SELECT file_path FROM image_asset WHERE id = :aid", {"aid": asset_id}) if not got: raise SystemExit(f"★asset {asset_id} 의 file_path 를 못 찾았다") p = pathlib.Path(str(got[0])) return p if p.is_absolute() else (ROOT / p) def _labeled_refs(refs: list) -> list: """records 의 refs 를 `(label, Path)` 로 — **하나라도 못 풀면 멈춘다**.""" out = [] for r in refs: raw = str(r.get("path") or "") if raw.startswith(" int: argv = [a for a in sys.argv[1:] if not a.startswith("--")] if len(argv) < 2: print(__doc__) return 2 project_id, episode_id = argv[0], argv[1] tag, label, model_key, effort = "", "A", "grok", "high" sysname = "critique" # ★기본이 dry (Codex NON-BLOCK) — 돌리려면 `--run` severity, ref_gate, dry = True, False, "--run" not in sys.argv for a in sys.argv[1:]: if a.startswith("--sys="): sysname = a.split("=", 1)[1].strip() elif a.startswith("--tag="): tag = a.split("=", 1)[1].strip() elif a.startswith("--label="): label = a.split("=", 1)[1].strip().upper() elif a.startswith("--model="): model_key = a.split("=", 1)[1].strip() elif a.startswith("--effort="): effort = a.split("=", 1)[1].strip() elif a == "--no-severity": severity = False elif a == "--ref-gate": ref_gate = True if not tag: raise SystemExit("--tag 이 필요하다") from app.core.config import settings from app.modules.pipeline.multiroll_gemini import ( STILL_JUDGE_PACK_VERSION, png_part, ref_index_part, ref_parts, resolve_judge_pack_version, resolve_judge_texts, ) from app.modules.pipeline.multiroll_select import build_critique_schema rdir = (pathlib.Path(settings.projects_dir) / project_id / "images" / episode_id / "scene" / "recipe") recs = json.loads((rdir / "records.json").read_text()) rec = recs.get(tag) if not isinstance(rec, dict): raise SystemExit(f"★{tag} 을 records 에서 못 찾았다 — 「없다」로 읽지 마라") img = rdir / f"{tag}_{label.lower()}.png" if not img.is_file(): raise SystemExit(f"★검사 대상 이미지가 없다: {img}") prompt = (rec.get("roll_prompts") or {}).get(label) or "" if not prompt: raise SystemExit(f"★{tag} {label} 의 roll 프롬프트가 비었다") refs = _labeled_refs(list(rec.get("refs") or [])) if sysname == "observe": # ★살아 있는 수정 갈래(GG46)가 **실제로 던지는** 글과 스키마. # 팩 버전도 그 갈래가 못박은 것을 그대로 가져온다 — 최신 팩을 # 집으면 「도는 것」이 아니라 「고쳐 둔 것」을 재게 된다. from app.modules.pipeline.multiroll_gemini import ( GG46_CRITIQUE_PACK_VERSION, ) from app.modules.pipeline.multiroll_select import ( build_gq_observe_schema, ) from app.modules.prompt_loader import load_prompt pack = resolve_judge_pack_version(GG46_CRITIQUE_PACK_VERSION) critique_sys = load_prompt( "multiroll_judge", "gq_observe_sys", version=pack) schema = build_gq_observe_schema() severity, ref_gate = True, False # 관찰 스키마 고정 계약 elif sysname == "critique": pack = resolve_judge_pack_version(STILL_JUDGE_PACK_VERSION) texts = resolve_judge_texts(2, "judge_still", STILL_JUDGE_PACK_VERSION) critique_sys = texts["critique_sys"] schema = build_critique_schema( with_severity=severity, with_ref_gate=ref_gate) else: raise SystemExit(f"--sys 는 critique|observe 중 하나다: {sysname!r}") parts = [{"type": "text", "text": "THE PROMPT:\n" + prompt}] parts += ref_parts(refs) parts.append({"type": "text", "text": "Photograph to examine:"}) parts.append(png_part(img)) if ref_gate: parts.append(ref_index_part(refs)) print(f"지시문 --sys={sysname} · 팩={pack} · {len(critique_sys):,}자") print(f"검사 대상 {tag} {label} · 프롬프트 {len(prompt):,}자 · " f"참조 {len(refs)}장") for lab, p in refs: print(f" · {lab[:64]}… → {p.name}") print(f"스키마 severity={severity} ref_gate={ref_gate} · " f"모델={model_key} effort={effort}") if dry: print("\n기본이 dry 라 호출 0 — 실제로 돌리려면 `--run`") return 0 t0 = time.time() usage: dict = {} if model_key == "grok": from app.modules.llm.openrouter_vlm_client import JSON_CLAUSE, _client model = str(getattr(settings, "grok_judge_model", "") or "").strip() if not model: raise SystemExit("GROK_JUDGE_MODEL 이 비었다 (fail-closed)") sys_prompt = critique_sys + JSON_CLAUSE.format( schema=json.dumps(schema, ensure_ascii=False, indent=2)) resp = _client().chat.completions.create( model=model, messages=[{"role": "system", "content": sys_prompt}, {"role": "user", "content": parts}], response_format={"type": "json_object"}, max_tokens=8000, extra_body={"seed": SEED, "reasoning": {"effort": effort}}, ) payload = json.loads(resp.choices[0].message.content or "{}") u = getattr(resp, "usage", None) if u is not None: det = getattr(u, "completion_tokens_details", None) usage = {"prompt_tokens": getattr(u, "prompt_tokens", None), "completion_tokens": getattr(u, "completion_tokens", None), "reasoning_tokens": getattr(det, "reasoning_tokens", None) if det is not None else None} else: from app.modules.llm.llm_client import call_structured model = "gemini-pro" payload = call_structured( "still_recipe_critique", critique_sys, parts, schema, project_config={"still_recipe_critique": {"model": model}}, schema_name="still_recipe_critique") dur = time.time() - t0 issues = list((payload or {}).get("issues") or (payload or {}).get("observations") or []) print(f"\n═══ {dur:.1f}초 · 지적 {len(issues)}건 · " f"추론토큰={usage.get('reasoning_tokens')} ═══") for i, it in enumerate(issues, 1): print(f"\n[{i}] severity={it.get('severity')} " f"needs_regeneration={it.get('needs_regeneration')} " f"unfixable={it.get('unfixable')}") print(f" {it.get('issue_ko')}") if it.get("fix_en"): print(f" fix: {str(it['fix_en'])[:300]}") OUT.parent.mkdir(parents=True, exist_ok=True) prev = json.loads(OUT.read_text()) if OUT.exists() else [] prev.append({ "tag": tag, "label": label, "sys": sysname, "model": model_key, "effort": effort, "severity_schema": severity, "ref_gate": ref_gate, "duration_s": round(dur, 1), "usage": usage, "payload": payload, }) OUT.write_text(json.dumps(prev, ensure_ascii=False, indent=1)) print(f"\n산출: {OUT}") return 0 if __name__ == "__main__": raise SystemExit(main())