"""판정 루프가 **정말로** 같은 것을 다시 사나 (2026-08-27, 감사 2-B). 감사표 2-B 는 판정·수정 루프를 $101/10일(전체의 26%)로 세고, 근거로 「네 단계 모두 system_prompt 가 0자 — 공통 브리프가 전부 user 에 있습니다」 · 「캐시 fix_rejudge 1.2% / 나머지 0.0%」 를 들었다. ★**두 근거가 다 의심스럽다.** ① `resolve_judge_texts(...)["judge_sys"]` 는 **6,801자**다. `openrouter_vlm_client.py:95` 가 `{"role":"system"}` 으로 싣는다. 「0자」는 **DB 로그를 잰 값**이지 나가는 payload 가 아니다. ② grok 은 `original_usage.cached_tokens` 를 보고하는데 **gemini 는 그 칸이 아예 없다.** 그러니 gemini 단계의 「0.0%」는 「캐시가 안 붙는다」가 아니라 **「재는 도구가 못 본다」**다. ★그래서 이 도구는 **나가는 payload(Opik span)** 에서만 잰다. 다음을 센다: role 별 글자 수 system 이 정말 비어 있나 cached_tokens 보고하는가 · 얼마나 적중하는가 이웃 호출 공통 접두 정순·역순 두 건의 **바이트 공통 접두**가 얼마인가 (캐시가 붙을 수 있는 최대치) ★**공통 접두**를 재는 것이지 「비슷한 정도」가 아니다. 캐시는 접두에만 붙는다 — 가운데가 같아도 소용없다. .venv/bin/python tools/prompt_measure/audit_judge_loop_reuse.py [페이지수] """ from __future__ import annotations import json import pathlib import sys from collections import defaultdict import requests sys.path.insert(0, "/Users/manta/Documents/Projects/TheRoad-I1/scratchpad") from _opik_env import opik_target # noqa: E402 OUT = pathlib.Path( "/Users/manta/Documents/Projects/TheRoad-I1/artifact/20260827_judge_loop") JUDGE_MARKS = ("judge", "rejudge", "critique", "observe", "compose") PAGE = 500 def _fetch(pages: int): base, ws, proj = opik_target() hdr = {"Comet-Workspace": ws, "Content-Type": "application/json"} out = [] for p in range(1, pages + 1): r = requests.get(f"{base}/v1/private/spans", headers=hdr, params={"workspace_name": ws, "project_name": proj, "size": PAGE, "page": p}, timeout=120) r.raise_for_status() got = r.json().get("content") or [] out += got print(f" page {p}: {len(got)}건 (누적 {len(out)})") if len(got) < PAGE: break return out def _op(span) -> str: for t in span.get("tags") or []: if str(t).startswith("op:"): return str(t)[3:] return "" def _msgs(span): inp = span.get("input") if isinstance(inp, list): return [m for m in inp if isinstance(m, dict)] if isinstance(inp, dict): return [m for m in (inp.get("messages") or []) if isinstance(m, dict)] return [] def _text_of(content) -> str: """이 메시지가 실어 보낸 **글자**. 이미지는 자리표로만 센다. ★이미지 base64 를 글자로 세면 수치가 통째로 이미지 크기가 된다 — 여기서 재려는 것은 **말**이지 그림이 아니다. """ if isinstance(content, str): return content parts = [] for p in content or []: if not isinstance(p, dict): continue if p.get("type") == "image_url": parts.append("") else: parts.append(str(p.get("text") or "")) return "\n".join(parts) def _wire(span) -> dict: sys_txt, usr_txt, imgs = [], [], 0 for m in _msgs(span): c = m.get("content") if isinstance(c, list): imgs += sum(1 for p in c if isinstance(p, dict) and p.get("type") == "image_url") (sys_txt if m.get("role") == "system" else usr_txt).append(_text_of(c)) return {"system": "\n".join(sys_txt), "user": "\n".join(usr_txt), "images": imgs} def _common_prefix(a: str, b: str) -> int: n = min(len(a), len(b)) i = 0 while i < n and a[i] == b[i]: i += 1 return i def main() -> None: pages = int(sys.argv[1]) if len(sys.argv) > 1 else 6 print(f"■ Opik span 을 {pages} 페이지까지 읽는다\n") spans = _fetch(pages) judged = [s for s in spans if any(k in _op(s).lower() for k in JUDGE_MARKS)] print(f"\n판정 계열 span {len(judged)} / 전체 {len(spans)}") if not judged: print(" ★판정 span 이 이 범위에 없다 — 페이지를 늘려야 한다." " 「0건」을 「없다」로 읽지 말 것.") return by_op = defaultdict(list) for s in judged: by_op[_op(s)].append(s) rows = [] print(f"\n{'단계':38s} {'건':>4s} {'sys':>7s} {'user':>8s} " f"{'img':>4s} {'캐시칸':>6s} {'적중':>7s}") print("─" * 82) for op in sorted(by_op, key=lambda k: -len(by_op[k])): ss = by_op[op] w = [_wire(s) for s in ss] sys_avg = sum(len(x["system"]) for x in w) // len(w) usr_avg = sum(len(x["user"]) for x in w) // len(w) img_avg = sum(x["images"] for x in w) / len(w) has_cache = sum( 1 for s in ss if any("cach" in k.lower() for k in (s.get("usage") or {}))) cached = sum(int((s.get("usage") or {}).get( "original_usage.cached_tokens") or 0) for s in ss) prompt = sum(int((s.get("usage") or {}).get("prompt_tokens") or 0) for s in ss) hit = f"{cached/prompt*100:5.1f}%" if prompt and has_cache else " — " print(f"{op[:38]:38s} {len(ss):4d} {sys_avg:7,d} {usr_avg:8,d} " f"{img_avg:4.1f} {has_cache:4d}/{len(ss):<2d} {hit:>7s}") rows.append({ "op": op, "n": len(ss), "sys_avg": sys_avg, "user_avg": usr_avg, "img_avg": img_avg, "cache_field_spans": has_cache, "cached_tokens": cached, "prompt_tokens": prompt, }) print("\n" + "─" * 82) print("■ 정순·역순 짝의 **공통 접두** — 캐시가 붙을 수 있는 최대치\n") pairs = [] for op in sorted(by_op): base_op = op.replace("_rev", "") if op == base_op or base_op not in by_op: continue f_list, r_list = by_op[base_op], by_op[op] for f, rv in zip(sorted(f_list, key=lambda s: s.get("start_time") or ""), sorted(r_list, key=lambda s: s.get("start_time") or "")): wf, wr = _wire(f), _wire(rv) sp = _common_prefix(wf["system"], wr["system"]) up = _common_prefix(wf["user"], wr["user"]) pairs.append({"op": base_op, "sys_prefix": sp, "user_prefix": up, "sys_len": len(wf["system"]), "user_len": len(wf["user"])}) if not pairs: print(" `_rev` 짝을 못 찾았다 — op 태그 이름을 확인할 것") else: by_base = defaultdict(list) for p in pairs: by_base[p["op"]].append(p) for base_op, ps in by_base.items(): sl = sum(p["sys_len"] for p in ps) // len(ps) ul = sum(p["user_len"] for p in ps) // len(ps) spf = sum(p["sys_prefix"] for p in ps) // len(ps) upf = sum(p["user_prefix"] for p in ps) // len(ps) tot, pre = sl + ul, spf + upf print(f" {base_op[:40]:40s} 짝 {len(ps):3d}") print(f" system {spf:,}/{sl:,} · user {upf:,}/{ul:,}" f" → 접두 {pre:,}/{tot:,} = " f"{(pre/tot*100 if tot else 0):.1f}%") OUT.mkdir(parents=True, exist_ok=True) (OUT / "reuse.json").write_text(json.dumps( {"by_op": rows, "pairs": pairs}, ensure_ascii=False, indent=2), encoding="utf-8") print(f"\n 기록 → {OUT / 'reuse.json'}") print("\n★「캐시칸」이 0/N 인 단계는 적중률을 **못 잰다.** 그 칸의 0% 를") print(" 「캐시가 안 붙는다」로 읽으면 안 된다 — gemini 가 그렇다.") if __name__ == "__main__": main()