games-development-ai/wg1/gen-worker/worker/agent_loop/calibrate.py

"""agent_loop/calibrate.py —— W-G1 player-panel 校准（"好玩 Claude-free 可信"的命门验证）。

目的：在【已知答案】样本上测 player(vision/text) 判定 vs 人类锚的吻合度，
重点 = 反样本（空心游戏）否决率——这是 text player 翻车、整个 W-G1 论点站不站得住的命门。

锚来源（见 labels.json.anchor_note）：
  · hollow 反样本 = 客观锚（真玩 remaining 零变化=空心，无主观成分）；
  · good 正样本   = opus-proxy 临时锚，创始人亲玩后覆盖 human.fun/verdict 即权威。

复用 studio 的 _judge_vision/_judge_text（与生产同一 player prompt/同款人格/同 max_tokens），
吃各游戏已存的 evidence（first-paint.png/after-play.png/verdict.json），不重开服务=便宜可复跑。
"""

import argparse
import json
import sys
from collections import defaultdict
from pathlib import Path

# 包内/直跑兼容导入（镜像 studio.py 的导入范式）。
try:
    from . import config, studio
    from .. import run, cost
except ImportError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
    import run, cost  # type: ignore
    from agent_loop import config, studio  # type: ignore

WORKER_ROOT = Path(__file__).resolve().parents[2]   # wg1/gen-worker
CALIB_DIR = WORKER_ROOT / "calibration"
GEN_DIR = run.GEN_DIR
BRIEFS_DIR = run.BRIEFS_DIR

# 生产同款人格（与 studio.run_player_panel 内联人格逐字一致，保证校准的是生产 player）。
PERSONA_VISION = "急性子休闲玩家,凭第一眼观感和反馈下判断"
PERSONA_TEXT = "细致的完整度审查者,只信运行数据、不脑补"


def _load_verdict(game_id):
    """读某游戏的真玩 verdict（喂文本 player 的运行数据；缺失返回 None）。"""
    p = GEN_DIR / game_id / "evidence" / "verdict.json"
    if p.exists():
        try:
            return json.loads(p.read_text(encoding="utf-8"))
        except Exception:
            return None
    return None


def _load_brief(brief_ref):
    """读题面文本（与生成时同一 brief，保证 player 看到的题面口径一致）。"""
    p = BRIEFS_DIR / f"{brief_ref}.json"
    return json.loads(p.read_text(encoding="utf-8"))["brief"]


def _judge_user_text(brief, verdict):
    """组装喂给 player 的证据文本（与 studio.run_player_panel 同口径）。"""
    summary = studio._play_summary(verdict)
    return (f"游戏题面:\n{brief}\n\n{summary}\n\n"
            "请基于以上证据(及截图,若有)给出你的玩家评判。")


def _num(x):
    """安全取数字（player 偶尔把分数写成字符串）。非数字返回 None。"""
    if isinstance(x, bool):
        return None
    if isinstance(x, (int, float)):
        return x
    try:
        return float(x)
    except (TypeError, ValueError):
        return None


def run_calibration(labels_path, stage):
    """跑校准：逐样本跑 vision+text player → 算每位 player 的吻合指标。返回报告 dict。"""
    labels = json.loads(Path(labels_path).read_text(encoding="utf-8"))
    samples = labels["samples"]
    vis_model = config.model_name_for("player_vision", stage)
    txt_model = config.model_name_for("player_text", stage)
    per_model = defaultdict(lambda: [0, 0])  # model -> [in, out]
    rows = []

    for s in samples:
        gid = s["game_id"]
        brief = _load_brief(s["brief_ref"])
        verdict = _load_verdict(gid)
        ut = _judge_user_text(brief, verdict)
        # 视觉位（吃截图）。U4/B9 后 _judge_* 返回 (in,out,cached) 三元组——校准只关心 in/out，cached 占位忽略
        #   （P2-2：校准成本有意取「全价上界」，不折缓存——校准是离线评估，账走保守上界即可，不必对齐生产逐款精账）。
        try:
            vj, (vi, vo, _vca) = studio._judge_vision(vis_model, PERSONA_VISION, ut, gid)
            per_model[vis_model][0] += vi
            per_model[vis_model][1] += vo
        except Exception as e:
            vj = {"error": str(e)[:180]}
        # 文本位（仅运行数据）
        try:
            tj, (ti, to, _tca) = studio._judge_text(txt_model, PERSONA_TEXT, ut)
            per_model[txt_model][0] += ti
            per_model[txt_model][1] += to
        except Exception as e:
            tj = {"error": str(e)[:180]}
        rows.append({"game_id": gid, "class": s["class"], "human": s["human"],
                     "vision": vj, "text": tj})

    # ── 指标：逐 player 算 好玩 MAE / verdict 吻合 / 反样本否决·假绿 / 正样本接受 ──
    metrics = {}
    for who, mname in (("vision", vis_model), ("text", txt_model)):
        fun_abs = []
        verdict_match = n_verdict = 0
        hollow_total = hollow_rejected = hollow_false_green = 0
        good_total = good_accepted = 0
        for r in rows:
            pj, human, cls = r[who], r["human"], r["class"]
            pf, pv = _num(pj.get("fun")), str(pj.get("verdict"))
            hf, hv = _num(human.get("fun")), str(human.get("verdict"))
            if pf is not None and hf is not None:
                fun_abs.append(abs(pf - hf))
            if pv in ("pass", "fix"):
                n_verdict += 1
                if pv == hv:
                    verdict_match += 1
            if cls == "hollow":
                hollow_total += 1
                # 正确否决 = 判 fix 且 好玩≤2（既看 verdict 也看分，防"判 fix 却给高分"的矛盾）
                if pv == "fix" and pf is not None and pf <= 2:
                    hollow_rejected += 1
                # 危险假绿 = 放过空心（判 pass 或 好玩≥3）
                if pv == "pass" or (pf is not None and pf >= 3):
                    hollow_false_green += 1
            elif cls == "good":
                good_total += 1
                if pv == "pass" and pf is not None and pf >= 3:
                    good_accepted += 1
        # 是否够格当 Claude-free 好玩门：必须 100% 否决反样本（零假绿）+ 不误杀正样本
        gatekeeper_ok = (hollow_total > 0 and hollow_false_green == 0
                         and (good_total == 0 or good_accepted == good_total))
        metrics[who] = {
            "model": mname,
            "fun_mae": round(sum(fun_abs) / len(fun_abs), 2) if fun_abs else None,
            "verdict_match": f"{verdict_match}/{n_verdict}",
            "hollow_rejected": f"{hollow_rejected}/{hollow_total}",
            "hollow_false_green": f"{hollow_false_green}/{hollow_total}",
            "good_accepted": f"{good_accepted}/{good_total}",
            "gatekeeper_ok": gatekeeper_ok,
        }

    # ── 成本（测而不闸；复用 cost.py 的 new-api quota 口径）──
    #   P2-2：此处 compute 不传 cached_tokens（默认 0）→ 成本是「全价上界」（不折缓存命中），与生产 run_studio 的
    #   精账（折 cache_ratio）有意不同——校准为离线评估，账走保守上界即可。
    try:
        pricing = cost.fetch_pricing()
        qpu, usd = cost.fetch_status()
        rws, total = {}, 0.0
        for name, (i, o) in per_model.items():
            c = cost.compute(name, i, o, pricing, qpu, usd)  # 不传 cached → 全价上界（P2-2）
            rws[name] = {"in": i, "out": o, "rmb": c["rmb"]}
            total += c["rmb"]
        cost_info = {"by_model": rws, "total_rmb": round(total, 5), "costNote": "fullPriceUpperBound"}
    except Exception as e:
        cost_info = {"error": str(e)[:180],
                     "by_model": {k: {"in": v[0], "out": v[1]} for k, v in per_model.items()}}

    report = {"stage": stage, "n_samples": len(samples),
              "anchor_note": labels.get("anchor_note", ""),
              "rows": rows, "metrics": metrics, "cost": cost_info}
    CALIB_DIR.mkdir(parents=True, exist_ok=True)
    (CALIB_DIR / "report.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
    return report


def main():
    ap = argparse.ArgumentParser(description="W-G1 player-panel 校准")
    ap.add_argument("--labels", default=str(CALIB_DIR / "labels.json"))
    ap.add_argument("--stage", default=None, help="models.yaml 阶段(默认 default_stage)")
    args = ap.parse_args()
    _cfg, default_stage = config.load_config()
    stage = args.stage or default_stage
    print(f"[calib] stage={stage} labels={args.labels}")
    rep = run_calibration(args.labels, stage)

    print("\n=== player 校准报告（逐样本）===")
    for r in rep["rows"]:
        h, v, t = r["human"], r["vision"], r["text"]
        flag = "🟥反样本" if r["class"] == "hollow" else "🟩正样本"
        print(f"\n[{r['game_id']}] {flag} 人类锚 fun={h.get('fun')} verdict={h.get('verdict')} ({h.get('anchor')})")
        print(f"  vision: fun={v.get('fun')} verdict={v.get('verdict')} note={str(v.get('note') or v.get('error'))[:64]}")
        print(f"  text  : fun={t.get('fun')} verdict={t.get('verdict')} note={str(t.get('note') or t.get('error'))[:64]}")
    print("\n--- 指标（命门=反样本假绿必须 0/N）---")
    for who, m in rep["metrics"].items():
        verdict_gate = "✅够格当好玩门" if m["gatekeeper_ok"] else "❌不够格"
        print(f"  {who}({m['model']}): fun_MAE={m['fun_mae']} verdict吻合={m['verdict_match']} "
              f"反样本否决={m['hollow_rejected']} 反样本假绿={m['hollow_false_green']} "
              f"正样本接受={m['good_accepted']}  → {verdict_gate}")
    c = rep.get("cost") or {}
    print(f"\n[calib] ¥={c.get('total_rmb', '?')}  report → wg1/gen-worker/calibration/report.json")


if __name__ == "__main__":
    main()