68 lines
3.6 KiB
Python
68 lines
3.6 KiB
Python
"""4축 리포트: 품질(judge, 성공 실행 한정) · 프로세스 비용(meter) · 안정성 · 가성비. 실행 실패 ≠ 품질
|
|
패배 — 실패 arm 은 품질 pairwise 제외, 파일럿 1회에서 한 arm 실패 시 전체 품질 순위 판정 보류."""
|
|
from . import aggregate as agg
|
|
|
|
DISCLAIMER = ("이 파일럿은 ruler의 판별력, arm 격리, 실행 드라이버와 P1~P3의 잠정적 품질 신호를 검증한다. "
|
|
"Arm별 단일 실행이므로 통계적 우월성이나 일반적인 생산성 향상을 확정하지 않는다.")
|
|
|
|
|
|
def stability_axis(meters):
|
|
arms = list(meters)
|
|
if not arms:
|
|
return {"execution-success-rate": 0.0, "gate-block-total": 0}
|
|
ok = sum(1 for a in arms if meters[a].get("execution-failures", 0) == 0)
|
|
return {"execution-success-rate": ok / len(arms),
|
|
"gate-block-total": sum(meters[a].get("hook-blocks", 0) for a in arms)}
|
|
|
|
|
|
def quality_axis(judgments, run_id, arm_ids):
|
|
"""dedup 된 valid judgment 으로 pair별 집계(실패 arm 은 호출 전 이미 제외됨)."""
|
|
from itertools import combinations
|
|
out = {}
|
|
for a, b in combinations(arm_ids, 2):
|
|
pair_id = f"{a}-vs-{b}"
|
|
recs = [r for r in judgments if r.get("pair-id") == pair_id and r.get("status") == "valid"]
|
|
# judge-index 별 forward/reversed 를 실질 arm 관점으로 정규화 → stable 여부
|
|
by_ji = {}
|
|
for r in recs:
|
|
pj = r.get("pairwise-judgment") or {}
|
|
w = (pj.get("overall") or {}).get("winner", "tie")
|
|
by_ji.setdefault(r["judge-index"], {})[r["orientation"]] = agg.normalize(r["orientation"], w)
|
|
paired, stable_verdicts = [], []
|
|
for ji, o in by_ji.items():
|
|
if "forward" in o and "reversed" in o:
|
|
paired.append({"fwd": o["forward"], "rev": o["reversed"]})
|
|
if agg.stable(o["forward"], o["reversed"]):
|
|
stable_verdicts.append(o["forward"])
|
|
wins = stable_verdicts.count("first"); ties = stable_verdicts.count("tie")
|
|
out[pair_id] = {"wins-first": wins, "ties": ties, "wins-second": stable_verdicts.count("second"),
|
|
"stable-paired-votes": len(stable_verdicts), "unstable-paired-votes": len(paired) - len(stable_verdicts),
|
|
"preference-first": agg.preference_score(wins, ties, len(stable_verdicts)),
|
|
"panel-agreement": agg.panel_agreement(stable_verdicts),
|
|
"position-flip-consistency": agg.flip_consistency(paired),
|
|
"panel-verdict": agg.panel_verdict(stable_verdicts)}
|
|
return out
|
|
|
|
|
|
def ranking(quality, failed_arms):
|
|
if failed_arms:
|
|
return {"status": "held", "reason": f"arm {failed_arms} 실행 실패 — 파일럿 1회, 순위 판정 보류"}
|
|
return {"status": "decided", "pairs": quality}
|
|
|
|
|
|
def render_markdown(quality, process, stability, ranking_, calibrated):
|
|
L = ["# 🏁 Cascade Benchmark (P1+P2 / P3-A / P3-B-active)", ""]
|
|
if not calibrated:
|
|
L += ["> **UNCALIBRATED — 품질 판정에 사용 금지** (calibration 미통과 또는 미실행)", ""]
|
|
L += ["## 1. 품질(judge, 성공 실행 한정)", "```yaml", _y(quality), "```",
|
|
"## 2. 프로세스 비용(meter)", "```yaml", _y(process), "```",
|
|
"## 3. 안정성", "```yaml", _y(stability), "```",
|
|
"## 4. 순위/가성비", "```yaml", _y(ranking_), "```",
|
|
"", "---", f"> {DISCLAIMER}"]
|
|
return "\n".join(L) + "\n"
|
|
|
|
|
|
def _y(obj):
|
|
import yaml
|
|
return yaml.safe_dump(obj, allow_unicode=True, sort_keys=False).rstrip()
|