#!/usr/bin/env python3 """#21 — grok 판정 한 콜이 110초인 이유를 **effort 로** 줄일 수 있나 (pilot). ## 왜 이걸 재나 PR #43·#44 는 콜의 **개수와 직렬 여부**를 줄였다(E2E 127→60분). 그런데 **한 콜은 106.0 → 109.6초로 그대로**다. 이제 그게 가장 큰 단일 덩어리다 — 판정 6콜 10.6분 대 롤 12장 3.1분. 무료로 잰 상관: 판정 시간 ↔ **출력 토큰** +0.996(초당 53토큰). grok-4.6 은 completion 15,790토큰인데 실제 답은 1,715자 — **시간의 대부분이 답이 아니라 추론 출력**이다. 그래서 추론량을 줄이면 시간이 준다는 가설이다. ## ★모델이 실제로 무엇을 받는지 먼저 확인했다 `GET /api/v1/models` 에서 `x-ai/grok-4.6` 직접 조회: reasoning.mandatory : true ← 끌 수 없다 reasoning.default_enabled : true reasoning.default_effort : "high" ← 지금 여기서 돌고 있다 reasoning.supported_efforts: ["xhigh","high","medium","low"] 즉 `none`·`minimal`·`max` 는 **보내면 안 된다**. 내릴 수 있는 것은 `medium` · `low` 둘뿐이다. 이 도구는 그 둘만 태운다. `exclude=true` 는 **시간 레버가 아니다** — 공식 문서상 추론은 그대로 하고 응답에서 trace 만 뺀다. 그래서 이 pilot 의 arm 에 넣지 않는다. ## 이 단계에서 재는 것 — **시간과 잡음뿐** 품질 판정은 다음 단계다. 여기서 볼 것은 하나: **low 가 현행보다 중앙값 30% 이상 빠른가.** 안 빠르면 품질 위험을 감수할 이유가 없으니 멈춘다. arm = omitted(현행 = provider default high) / medium / low 쌍 = 방금 완주한 주행의 후보 6쌍 반복 = arm 당 2회 → 6 × 3 × 2 = 36콜 ## ★섞이지 않게 하는 것들 - **동시에 안 던진다.** 동시 호출은 provider 큐·경합이 섞여 per-call latency 비교를 오염시킨다(Codex 지적, 수용). 하나씩 순차로. - **arm 순서를 샷마다 회전한다.** 늘 같은 순서면 뒤 arm 이 워밍업 이득을 본다. - **프로덕션과 같은 역순 입력**을 쓴다 — Grok 슬롯은 역순을 본다 (`_judge_cross_model_order`). 정순으로 재면 다른 것을 잰다. - 산출은 **읽어서 잇는다.** 부분 재실행이 앞선 판정을 날리면 그 뒤 집계가 적은 표본으로 초록이 된다. ## 곁: seed `supported_parameters` 에 `seed` 가 있다. 이미지 생성에는 seed 가 없지만 **판정 모델에는 있다.** 있으면 같은 arm 반복이 재현되어 「원래 흔들리는 것」과 「arm 이 바꾼 것」을 가르기 쉬워진다. 그래서 보내 보고 **실제로 재현되는지 이 pilot 이 확인한다** — provider 가 무시할 수도 있으므로 「보냈으니 고정됐다」로 읽지 않는다. usage: grok_reasoning_effort_pilot.py [--reps N] --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])) ROOT = pathlib.Path(__file__).resolve().parents[3] OUT = ROOT / "artifact" / "20260829_grok_effort_pilot" / "calls.json" #: 모델이 실제로 받는 값만. 조회로 확인했다 — 지어내지 않았다. #: 기본 arm. `--arms=` 로 바꾼다 — 모델이 받는 값만. ARMS = ["omitted", "medium", "low"] ALLOWED_ARMS = {"omitted", "xhigh", "high", "medium", "low"} SEED = 20260829 # ── 판정 지시문 임시 변형 (2026-08-29 사용자 지시) ──────────────── # # ★팩은 **한 글자도 안 건드린다.** 여기서 문자열로 바꿔 태워 보고, 효과가 # 서면 그때 새 팩 버전을 만든다. 프로덕션 판정 계약을 실험으로 흔들지 # 않는다. # # 고치는 자리와 이유: # # 현행 1번 절 끝은 `hard_violations` 를 정의한 뒤 **한쪽으로만 경고한다** — # 「한 심판이 139건을 올렸고 다른 심판은 17건이었다, 그게 진짜를 덮는다」. # 실측에서 Grok 은 추론을 많이 할수록(high 3,207 · medium 4,857 · # xhigh 7,486 토큰) 하드위반을 **0건**으로 냈고, low(595)에서만 채웠다. # 경고를 무겁게 읽을수록 칸이 비는 모양이다. 그리고 medium·xhigh 는 # `all_candidates_fail=True` 를 내면서 **이유를 한 줄도 안 적었다** — # 프로덕션에서 그 값이 참이면 재촬영 표시가 붙는데 무엇을 고칠지 모른다. # # 그래서 둘만 고친다: # ① 한쪽 경고를 **양쪽으로** — 덜 올리는 것도 같은 크기의 잘못이다 # ② `all_candidates_fail` 을 내면 **이유를 그 칸에 적게** 한다 # 경계의 정의(실격 결함만, 사소한 차이는 verdict 로)는 **그대로 둔다**. _JUDGE_ANECDOTE = """say it in that candidate's verdict instead. Measured on this pipeline, one judge filed 139 such minor differences as hard violations against another judge's 17, which drowns the real ones.""" _JUDGE_BALANCED = """say it in that candidate's verdict instead. That boundary cuts both ways. A fault of the kinds listed above is a hard violation even when the rest of the frame is good, and you must file it. Leaving one out hides it from the repair stage just as surely as burying the real ones under minor differences. If you set `all_candidates_fail`, every candidate must carry at least one entry in its `hard_violations` saying why it cannot be used.""" PROMPT_VARIANTS = {"orig": None, "v2": (_JUDGE_ANECDOTE, _JUDGE_BALANCED)} def apply_prompt_variant(judge_sys: str, variant: str) -> str: """★못 찾으면 **막는다** — 조용히 원문으로 돌면 두 arm 이 같아진다.""" sub = PROMPT_VARIANTS.get(variant) if sub is None: if variant not in PROMPT_VARIANTS: raise SystemExit(f"모르는 변형: {variant!r} " f"(있는 것: {sorted(PROMPT_VARIANTS)})") return judge_sys old, new = sub if judge_sys.count(old) != 1: raise SystemExit( f"변형 {variant!r} 의 대상 문구를 {judge_sys.count(old)}건 찾았다 — " "팩이 바뀌었다. 조용히 원문으로 돌면 두 arm 이 같아져 " "「효과 없음」이라는 거짓 결론이 남는다") return judge_sys.replace(old, new) def _pairs(project_id: str, episode_id: str): """방금 완주한 주행의 후보 쌍 — records 에서 **선정 판정 재료 그대로**.""" from app.core.config import settings rec = (pathlib.Path(settings.projects_dir) / project_id / "images" / episode_id / "scene" / "recipe" / "records.json") if not rec.exists(): raise SystemExit(f"★내 조회로는 records 를 못 찾았다: {rec}") data = json.loads(rec.read_text()) out = [] for tag, r in data.items(): if not isinstance(r, dict) or not r.get("roll_prompts"): continue d = rec.parent a, b = d / f"{tag}_a.png", d / f"{tag}_b.png" if not (a.is_file() and b.is_file()): continue rp = r["roll_prompts"] out.append({ "tag": tag, "prompt": rp.get("A") or next(iter(rp.values())), "cands": [a, b], "labels": ["A", "B"], # ★칸 이름을 조립부에서 그대로 — `run_multiroll_select` 는 # 승자를 `selected` 에 쓴다. `judge.winner` 로 찾았다가 전부 # None 이 나왔고, 그건 「없다」가 아니라 못 찾은 것이었다. "prod_winner": r.get("selected"), "prod_route": ((r.get("cross_model_order") or {}).get("route")), "prod_slot_winner": ( (r.get("cross_model_order") or {}).get("slot_winner")), }) return sorted(out, key=lambda x: x["tag"]) def main() -> int: argv = [a for a in sys.argv[1:] if not a.startswith("--")] # ★기본이 dry (Codex NON-BLOCK) — 돌리려면 `--run` dry = "--run" not in sys.argv reps = 2 only = "" variant = "orig" arms = list(ARMS) for a in sys.argv[1:]: if a.startswith("--reps="): reps = int(a.split("=", 1)[1]) elif a.startswith("--tag="): only = a.split("=", 1)[1].strip() elif a.startswith("--prompt="): variant = a.split("=", 1)[1].strip() elif a.startswith("--arms="): arms = [x.strip() for x in a.split("=", 1)[1].split(",") if x.strip()] # ★모델이 안 받는 값은 400 이 나거나 **조용히 무시**된다 — 무시되면 # 「올렸는데 안 변한다」는 거짓 결론이 남는다. 조회로 확인한 목록 # 밖은 여기서 막는다. bad = [a for a in arms if a not in ALLOWED_ARMS] if bad: raise SystemExit(f"이 모델이 안 받는 effort: {bad} " f"(받는 것: {sorted(ALLOWED_ARMS)})") if len(argv) < 2: print(__doc__) return 2 project_id, episode_id = argv[0], argv[1] # ★이름·모듈을 조립부에서 그대로 가져온다 — 손으로 지으면 한 칸 # 어긋나도 ImportError 로 늦게 드러난다(실제로 한 번 그랬다). from app.core.config import settings from app.modules.pipeline.multiroll_gemini import ( JUDGE_PACK_VERSION, STILL_JUDGE_PACK_VERSION, png_part, resolve_judge_texts, ) from app.modules.pipeline.multiroll_select import ( build_judge_schema, flip_display_to_canonical, normalize_flip_verdict, roll_labels, ) model = str(getattr(settings, "grok_judge_model", "") or "").strip() if not model: raise SystemExit("GROK_JUDGE_MODEL 이 비었다 (fail-closed)") pairs = _pairs(project_id, episode_id) texts = resolve_judge_texts(2, "judge_still", STILL_JUDGE_PACK_VERSION) schema = build_judge_schema(roll_labels(2), with_physics=True) if only: pairs = [p for p in pairs if p["tag"] == only] if not pairs: raise SystemExit(f"★{only} 을 못 찾았다 — 「없다」로 읽지 마라") jobs = [] for i, p in enumerate(pairs): # ★arm 순서를 샷마다 회전 — 늘 같은 순서면 뒤 arm 이 워밍업 이득 order = arms[i % len(arms):] + arms[:i % len(arms)] for rep in range(reps): for arm in order: jobs.append((p, arm, rep)) print(f"모델 {model} · 팩 judge={STILL_JUDGE_PACK_VERSION}" f"(전역 {JUDGE_PACK_VERSION})") print(f"쌍 {len(pairs)}개 × arm {len(arms)} {arms} × 반복 {reps} = " f"**{len(jobs)}콜** (순차, 동시 던지지 않음)") for p in pairs: print(f" {p['tag']:9s} 프롬프트 {len(p['prompt']):,}자 · " f"프로덕션 승자 {p['prod_winner']} · {p['prod_route']} " f"{p['prod_slot_winner']}") if dry: print("\n기본이 dry 라 호출 0 — 실제로 돌리려면 `--run`") print(f"arm: {arms} (모델이 받는 값만 — none/minimal/max 는 제외)") return 0 from app.modules.llm.openrouter_vlm_client import _client, JSON_CLAUSE OUT.parent.mkdir(parents=True, exist_ok=True) prev = {} if OUT.exists(): for r in json.loads(OUT.read_text()): prev[r["key"]] = r print(f" 기존 {len(prev)}콜을 읽어 잇는다") client = _client() judge_sys = apply_prompt_variant(texts["judge_sys"], variant) if variant != "orig": print(f" ★지시문 변형 {variant!r} 적용 — " f"{len(texts['judge_sys']):,}자 → {len(judge_sys):,}자") sys_prompt = judge_sys + JSON_CLAUSE.format( schema=json.dumps(schema, ensure_ascii=False, indent=2)) for n, (p, arm, rep) in enumerate(jobs, 1): key = f"{p['tag']}|{arm}|{variant}|{rep}" if key in prev: continue # ★프로덕션과 같은 **역순** 입력 — Grok 슬롯은 역순을 본다 rev = list(reversed(p["cands"])) parts = [{"type": "text", "text": texts.get("judge_prompt_header") or "THE PROMPT (all candidates were generated from this):"}, {"type": "text", "text": p["prompt"]}] for disp, src in zip(p["labels"], rev): parts.append({"type": "text", "text": f"Candidate {disp}:"}) parts.append(png_part(src)) extra: dict = {"seed": SEED} if arm != "omitted": extra["reasoning"] = {"effort": arm} t0 = time.time() err = None payload = None usage = {} try: 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=extra, ) txt = resp.choices[0].message.content or "" u = getattr(resp, "usage", None) if u is not None: usage = { "prompt_tokens": getattr(u, "prompt_tokens", None), "completion_tokens": getattr(u, "completion_tokens", None), } det = getattr(u, "completion_tokens_details", None) if det is not None: usage["reasoning_tokens"] = getattr( det, "reasoning_tokens", None) payload = json.loads(txt) except Exception as exc: # noqa: BLE001 err = repr(exc)[:300] dur = time.time() - t0 canon = None if payload: try: canon = normalize_flip_verdict( payload, flip_display_to_canonical(p["labels"]), p["labels"]) except Exception as exc: # noqa: BLE001 err = f"역매핑 실패: {exc!r}"[:200] row = { "key": key, "tag": p["tag"], "arm": arm, "rep": rep, "prompt_variant": variant, "seed": SEED, "duration_s": round(dur, 1), "usage": usage, "error": err, "winner": (canon or {}).get("winner"), "ranking": (canon or {}).get("ranking"), "all_candidates_fail": (canon or {}).get("all_candidates_fail"), "hard_by_label": { str(r.get("label")): list(r.get("hard_violations") or []) for r in ((canon or {}).get("readings") or []) }, "answer_chars": len(json.dumps(payload, ensure_ascii=False)) if payload else 0, } prev[key] = row OUT.write_text(json.dumps(list(prev.values()), ensure_ascii=False, indent=1)) ct = (usage or {}).get("completion_tokens") rt = (usage or {}).get("reasoning_tokens") print(f"[{n}/{len(jobs)}] {p['tag']:9s} {arm:8s} r{rep} " f"{dur:6.1f}초 완성토큰={ct} 추론토큰={rt} " f"승자={row['winner']}" + (f" ✗ {err[:60]}" if err else "")) _report(list(prev.values()), arms) print(f"\n산출: {OUT}") return 0 def _report(rows, arms=None) -> None: import statistics as st ok = [r for r in rows if not r["error"]] print(f"\n═══ 시간 (성공 {len(ok)}/{len(rows)}) ═══") print(f"{'arm':10s} {'n':>3s} {'중앙':>8s} {'최소':>7s} {'최대':>7s} " f"{'완성토큰 중앙':>14s}") base = None for arm in (arms or ARMS): b = [r for r in ok if r["arm"] == arm] if not b: continue d = sorted(r["duration_s"] for r in b) med = st.median(d) cts = [r["usage"].get("completion_tokens") for r in b if (r["usage"] or {}).get("completion_tokens")] print(f"{arm:10s} {len(b):>3d} {med:>7.1f}초 {d[0]:>6.1f}초 " f"{d[-1]:>6.1f}초 {(st.median(cts) if cts else 0):>13,.0f}") if arm == "omitted": base = med if base: print(f"\n★기준(omitted, provider default high) 중앙 {base:.1f}초") for arm in (arms or ARMS)[1:]: b = [r["duration_s"] for r in ok if r["arm"] == arm] if not b: continue cut = (base - st.median(b)) / base * 100 verdict = ("계속 — 30% 이상" if cut >= 30 else "★멈춤 — 30% 미만이면 품질 위험을 감수할 이유가 없다") print(f" {arm:8s} {cut:+.0f}% {verdict}") # ★seed 가 실제로 재현을 만드나 — 보냈다고 고정됐다고 읽지 않는다 print("\n═══ seed 가 재현을 만드나 (같은 arm 두 반복) ═══") same = diff = 0 for tag in sorted({r["tag"] for r in ok}): for arm in (arms or ARMS): reps = [r for r in ok if r["tag"] == tag and r["arm"] == arm] if len(reps) < 2: continue ws = {r["winner"] for r in reps} ds = [r["duration_s"] for r in reps] if len(ws) == 1: same += 1 else: diff += 1 print(f" {tag} {arm}: 승자가 갈렸다 {ws} · {ds}") print(f" 같은 승자 {same}쌍 · 갈린 쌍 {diff}") if diff == 0 and same: print(" ★모든 반복이 같다 — seed 가 먹었을 수도, 판정이 원래 " "확고했을 수도 있다. 이것만으로 「고정됐다」고 쓰지 않는다") if __name__ == "__main__": raise SystemExit(main())