882ca3f357
裁定三落地:accepted = 机械预筛 ∧ 独立模型玩法判定,九门 pass 降为「未见明显死」预筛。 - 波2a 玩法判定器(cheap_verify.py):读 brief+run-summary+全程截图(含新增局中连拍 midplay-1..N,play.cdp.cjs 只读并发拍、不进任何门判据),布尔裁 broken/hollow/off_brief, fail-closed;判定真相层落 evidence/judge.json + verdict.json 写回 accepted/judge 段; 金标夹具 fixtures/judge-golden/ + judge_golden.py 校准跑批 - 波2b 拆残笼:check 词法 tokenizer 替换 stripCode 正则(字符串内 // 假阴根修,292 在册 源零差异);prompt v1.6.3 外科清洗(两层奖励→纯设计语/删盲驱动器围设计/删熔断恐吓), registry 热取==内置逐字节一致(cheap_roles 内置回退同步);edit_file 空白归一+近似片段提示 - 判定器补修:空输出有界重试一次(glm 思考吃光 max_tokens 时 content 空)+重试放大公式 反缩 bug 修(旧 min(×2,6000) 大 base 反缩)+finishReason 落盘;imagesSeen 模型自报+ 盲检微扰重掷、两掷均盲=仪器故障 degraded(闲鱼中转静默丢图实锤,不再错杀 hollow) - 判定档切 MiniMax-M3(创始人 2026-07-10 拍板,裁定三①边界随 v2 plan 修订): generation.yaml judge.model+max_tokens=202752(512K 网关实探 400 拒,取实探上限); 3 例盲案 M3 重判全 accept、¥0.026/局 - 三路消费对齐:CLI 预筛语义修(cheap_studio 喂 verdict.pass 而非 finished,违裁定三的 放行已纠)/Service 路 apply_gameplay_judge 显式 prefilter 参数/result_out 权威改读 accepted(无键回落预筛,旧产物兼容)+trace.gameplayJudge additive 透传; tier2 gate_judge/genconfig 同步判定配置与消费 测试:test_gameplay_judge 33 项新增,全仓 pytest 480 绿;真判定闭环 accept ¥0.061。 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
970 lines
59 KiB
Python
970 lines
59 KiB
Python
"""
|
||
cheap_verify.py — 便宜档「丰富度」LLM 验证 agent（非阻塞·只报告）。
|
||
|
||
架构红线（创始人多轮强调，绝不可破）：**项目代码只做机械确定的事**；玩法/美术/音乐的「丰富」——
|
||
无论生成还是校验——都需大模型能力，**绝不写成代码校验**（code-presence/静态扫描/正则/断言一律禁）。
|
||
故本模块只做两件 100% 机械确定的事，「丰富不丰富」的判断本身完全交 LLM judge：
|
||
① 读产物 L3 源码（确定性文件 I/O）+ 拼 judge prompt（确定性字符串）；
|
||
② 解析 LLM 返回的 JSON 成结构化评分（确定性解析 + degraded 兜底）。
|
||
本模块**不含任何丰富度的代码断言**——11 条丰富度清单只是【喂给 LLM 的评分尺】，命中与否全由 judge 读码后裁；
|
||
评分按层标 L2/L3/L4 聚合成三分组小计（L2 内容丰富 / L3 留存结构 / L4 传播钩子，不设单一总分），规范见质量模型 SoT §3.3 计分制。
|
||
（注：sim-business-game-design.md §10 写了「4 条可机检 code-presence」子集，那与本红线冲突——本模块刻意不照它写代码校验。）
|
||
|
||
非阻塞铁律：LLM 调失败/超时/解析失败 → 返回 {score:None, degraded:True, reason:...}，
|
||
**绝不抛、绝不阻断生成、绝不进 verdict / 不改达标判定**。接线方（cheap_studio）再包一层 try/except 双保险。
|
||
|
||
被 cheap_studio.run_studio 收口段调用（九门 play 之后、写 run-summary 之前），结果 additive 写 summary["richness"]。
|
||
"""
|
||
|
||
import asyncio
|
||
import base64
|
||
import json
|
||
import time
|
||
from pathlib import Path
|
||
|
||
import cheap_run # 仅 stdlib 依赖（读产物 src 文件）；不 import agentscope/_bootstrap，故纯函数可在无 agentscope 环境单测。
|
||
|
||
# ── 通用底座 v2 = 11 条丰富度清单（judge 的评分 rubric）──
|
||
# 这是【喂给 LLM judge 的打分尺】，不是代码校验：丰富不丰富全由 judge 读源码后逐条裁。
|
||
# 规范来源 = 质量模型 SoT《游戏质量与爆火能力·设计》§3.3 计分制（每条 0/1 + 层标 L2/L3/L4，产三分组小计、不设单一总分）。
|
||
# v2 = 现行 8 条（与 .agents/skills/sim-business-game-design.md §10 同源）+ 新增 3 条（首 3 分钟脚本 / 炫耀时刻 / 同款钩子）。
|
||
# 每项 = (标准名, 层标, 含义)；judge 按标准名逐条回命中与否 + 理由，命中数按层聚合成三分组小计。
|
||
# ★锚定纪律（§3.3 第 3 条）：原 8 条判据文本与顺序一字不改（评分尺变更须金标复验、原 8 条判定漂移 ≤±1），
|
||
# 新增 3 条追加于末尾；层标只是聚合归属，不改原条目的判据本身。
|
||
# 分层含义：L2 内容丰富（有料耐玩）/ L3 留存结构（想再玩的结构前提）/ L4 传播钩子（想传出去的结构前提）。
|
||
RICHNESS_CHECKLIST = [
|
||
# 原 8 条（文本/顺序锁定，仅补层标）
|
||
("即时反馈", "L2", "每个主要操作有「飘字 +N + 音效 + 粒子」即时回报"),
|
||
("可见成长", "L2", "有数字/规模肉眼可见地变大（分数飙升/店铺扩张/等级上涨）"),
|
||
("下一个解锁", "L3", "任意时刻玩家眼前都有「再攒一点就解锁 X」的钩子"),
|
||
("30秒爽点", "L2", "开局 30 秒内有第一次升级/解锁"),
|
||
("数值滚雪球", "L2", "成长有「越来越快」的暴富段，而非平淡线性"),
|
||
("情感锚", "L2", "有萌角色/拥有物让玩家「想养大它」"),
|
||
("放置回归", "L3", "离线回来有惊喜（离线收益弹窗等）"),
|
||
("音反馈", "L2", "收益/升级/解锁有声音（哪怕程序化）"),
|
||
# 新增 3 条（v2 草案，随首个品类件落地验证；判据文本源自 §3.3 第 1 条）
|
||
("首3分钟脚本", "L3", "首 3 分钟脚本成立：0–10s 零阅读上手 / 10–60s 首次升级 / 1–3min 露出 2–3 个后续锁"),
|
||
("炫耀时刻", "L4", "有可截图分享的结算/成就画面（打完有结算展示场景，不是直接黑屏）"),
|
||
("同款钩子", "L4", "有清晰的品类原型 + 主题标签元数据，可供 remix 预填做同款"),
|
||
# 第 12 条（2026-07-03 增补，质量 SoT §4 规范一「通用底座 12 条口径」；创始人授权默认项拍板采纳）：
|
||
# 源自便宜档主 prompt 自检第 ⑨ 条（策划知识包 v2 否决项判定句式），升为评分尺条目。
|
||
# 锚定纪律：追加于末尾、前 11 条一字不动；L2 分母随之 6→7，档位观测线百分比不变、分母随升；
|
||
# 落地与金标复验同批（复验漂移超线按规范三复采样定性再判回退）。
|
||
("核心操作非无脑", "L2", "每次主操作含真实决策或技巧含量、判错有真代价（如补错货压库存），不是点哪都加分/点了就自动结算"),
|
||
]
|
||
|
||
_MAX = len(RICHNESS_CHECKLIST) # = 12（2026-07-03 ⑨升第 12 条；此前 v2 = 11）
|
||
_LAYERS = ("L2", "L3", "L4") # 三分组固定顺序（消费面按分组小计各自取阈，见 §3.6 档位表）
|
||
|
||
# ── 品类扩展 rubric（W-GENRE 品类件④；质量模型 SoT §4 规范二）──
|
||
# 数据源唯一 = fixtures/genre-rubrics/<genre>.json（items 键，条目 {name,layer,meaning,positive,negative}），
|
||
# 经 load_genre_checklist 加载。2026-07-02 rubric 挂点归一：原内联 GENRE_CHECKLISTS 字典退役，
|
||
# TRPG 条目一字不动迁入 trpg.json——品类条目自此只有 fixture 一路，不再留内联第二源。
|
||
# 每品类另加 ≥4 条品类特有维度：同一字段形状 (标准名, 层标, 含义)、同样标层，
|
||
# **分母独立小计**（与通用底座分开、不混合平均——评分输出落 richness.genre.groups，通用 groups 不变）。
|
||
# 这仍是【喂给 LLM judge 的评分尺】数据、不是代码校验（红线：不进九门/不进 verdict/不进脚手架）；
|
||
# 设计侧同源对齐 = .agents/skills/<genre>-game-design.md §10 品类自检。
|
||
# 锚定纪律（规范三）：品类条目文本变更须金标复验（金标正例 + 薄反例，分组小计漂移 ≤±1 即回退）。
|
||
_GENRE_RUBRICS_DIR = Path(__file__).resolve().parent / "fixtures" / "genre-rubrics"
|
||
|
||
# 品类脚手架目录名 → 品类 key（run_studio 据 scaffold_template 透传 genre；None/未登记 → 不评品类扩展）。
|
||
# 非遗品类键收敛 = "heritage"（与 rubric 文件名/散文 skill heritage-game-design.md 同名）：
|
||
# 模板目录 _template-feiyi 是资产目录名、金标 spec 的 _genre="heritage-craft" 是金标名（driver 承重不动），
|
||
# 三者同一品类——映射依据详见 fixtures/genre-rubrics/heritage.json 的 _note。
|
||
GENRE_BY_TEMPLATE = {
|
||
"_template-trpg": "trpg",
|
||
"_template-puzzle": "puzzle",
|
||
"_template-feiyi": "heritage",
|
||
# 2026-07-04 数据面补全(策划知识包 v2):剧情/经营两品类 fixture 落地补线,
|
||
# 条目源见各 fixture _note(narrative=skill §10 表逐字迁入;sim-business=质量 SoT §4 规范二名目展开)。
|
||
"_template-story": "narrative",
|
||
"_template-shop": "sim-business",
|
||
}
|
||
|
||
|
||
def load_genre_checklist(genre) -> list:
|
||
"""读品类 rubric fixture（品类条目唯一数据源）→ [(标准名, 层标, 判据)*]；genre 空/文件缺/坏条目 → []（绝不抛）。
|
||
|
||
条目形状随 fixture（name/layer/meaning，正反例 positive/negative 只作金标锚材料、不进 judge prompt
|
||
——prompt 里逐条只给一句可判定判据，与通用底座条目同形）。
|
||
"""
|
||
if not genre or not isinstance(genre, str):
|
||
return []
|
||
try:
|
||
p = _GENRE_RUBRICS_DIR / (genre + ".json")
|
||
if not p.is_file():
|
||
return []
|
||
data = json.loads(p.read_text(encoding="utf-8"))
|
||
out = []
|
||
for it in (data.get("items") or []):
|
||
if not isinstance(it, dict):
|
||
continue
|
||
name, layer, meaning = it.get("name"), it.get("layer"), it.get("meaning")
|
||
if isinstance(name, str) and name and layer in _LAYERS and isinstance(meaning, str) and meaning:
|
||
out.append((name, layer, meaning))
|
||
return out
|
||
except Exception: # noqa: BLE001 fixture 坏 → 扩展为空，评分退化为纯通用底座，绝不阻断
|
||
return []
|
||
|
||
|
||
def _layer_maxes(checklist=RICHNESS_CHECKLIST) -> dict:
|
||
"""各层条目数（分母）。通用底座 12 条口径 = L2:7 / L3:3 / L4:2（2026-07-03 第 12 条入 L2）。"""
|
||
m = {ly: 0 for ly in _LAYERS}
|
||
for _name, layer, _meaning in checklist:
|
||
m[layer] = m.get(layer, 0) + 1
|
||
return m
|
||
|
||
|
||
def _group_subtotals(hits: list, checklist=RICHNESS_CHECKLIST) -> dict:
|
||
"""把逐条命中按层标聚合成三分组小计 {L2:{score,max}, L3:..., L4:...}。
|
||
|
||
这是 §3.3 计分制的落库主形状——三层各自对得上自己的消费点（品类件验收/批次观测/回流分析），
|
||
**不做单一总分平均**（单一总分诱导 Goodhart，见 §3.3 第 1 条）。
|
||
"""
|
||
maxes = _layer_maxes(checklist)
|
||
scores = {ly: 0 for ly in _LAYERS}
|
||
for h in hits:
|
||
ly = h.get("layer")
|
||
if h.get("hit") and ly in scores:
|
||
scores[ly] += 1
|
||
return {ly: {"score": scores[ly], "max": maxes.get(ly, 0)} for ly in _LAYERS}
|
||
|
||
# 只读 L3 游戏本体（玩法/画面/数值/资产）；跳过 L1 plumbing（host-config/game.js/index.html——
|
||
# 固定脚手架、与「游戏丰富不丰富」无关，喂给 judge 只会稀释信号）。
|
||
_L3_SOURCE_FILES = ("game-logic.js", "core.js", "render.js", "balance.js", "assets.js")
|
||
_MAX_SRC_BYTES = 24000 # 喂 judge 的源码总量上限（控 token / 成本；超限按字节截断）。
|
||
_JUDGE_MAX_TOKENS = 4000 # judge 输出小（11 条短理由 + 一句点评），4000 足够、M3 无 thinking。
|
||
|
||
# judge 的 system / user 框定——只评分、不改码、要看源码真实现、输出严格 JSON。
|
||
# 计数用 len(RICHNESS_CHECKLIST) 动态拼(2026-07-03 ⑨升第 12 条时改:此前「11 条」写死在文案里,
|
||
# 清单变更即漂;动态取数后条数永远与评分尺同源。评分尺变更本身仍受锚定纪律约束=金标复验)。
|
||
_JUDGE_SYSTEM = (
|
||
"你是轻量小游戏的【丰富度评审 agent】。任务 = 读一款便宜档 LittleJS 小游戏的源码，"
|
||
f"只评判它「作为一款游戏够不够丰富、好玩、耐玩、想传出去」，逐条给出 {len(RICHNESS_CHECKLIST)} 条丰富度清单的命中与否 + 一句中文理由，"
|
||
"最后输出严格 JSON。你只评分、不修改代码、不阻断发布——这是非阻塞的质量信号。"
|
||
f"这 {len(RICHNESS_CHECKLIST)} 条分三层：L2 内容丰富（有料耐玩）、L3 留存结构（想再玩的结构前提）、L4 传播钩子（想让别人看/做同款的前提）；"
|
||
"评判要看源码里**真实实现**的玩法、数值成长、解锁阶梯、即时反馈、音效、结算画面与品类元数据，别被空壳或注释骗。"
|
||
)
|
||
|
||
# 品类扩展评分时的 system 变体（W-GENRE 件④）：仅把量词放宽到「通用 N 条 + 品类扩展若干条」，
|
||
# 其余判据措辞一字不动（锚定纪律：无品类路走上面原版 _JUDGE_SYSTEM）。
|
||
_JUDGE_SYSTEM_GENRE = (
|
||
"你是轻量小游戏的【丰富度评审 agent】。任务 = 读一款便宜档 LittleJS 小游戏的源码，"
|
||
f"只评判它「作为一款游戏够不够丰富、好玩、耐玩、想传出去」，逐条给出丰富度清单（通用 {len(RICHNESS_CHECKLIST)} 条 + 品类扩展若干条）的命中与否 + 一句中文理由，"
|
||
"最后输出严格 JSON。你只评分、不修改代码、不阻断发布——这是非阻塞的质量信号。"
|
||
"条目分三层：L2 内容丰富（有料耐玩）、L3 留存结构（想再玩的结构前提）、L4 传播钩子（想让别人看/做同款的前提）；"
|
||
"评判要看源码里**真实实现**的玩法、数值成长、解锁阶梯、即时反馈、音效、结算画面与品类元数据，别被空壳或注释骗。"
|
||
)
|
||
|
||
|
||
def _degraded(reason: str) -> dict:
|
||
"""降级结果（非阻塞铁律）：score=None + degraded=True；score 为 None 表示「这次没评出来」，绝不参与达标。
|
||
|
||
groups 同置 None：三分组小计的消费面（品类件验收/回流分析）遇 None 即「本次未评出」，与 score=None 同义。
|
||
"""
|
||
return {"score": None, "max": _MAX, "hits": [], "groups": None, "notes": None, "degraded": True, "reason": reason}
|
||
|
||
|
||
def _coerce_bool(v) -> bool:
|
||
"""把 judge 可能给的多形态命中值归一为 bool（true/1/"是"/"命中" 等 → True；其余 → False）。"""
|
||
if isinstance(v, bool):
|
||
return v
|
||
if isinstance(v, (int, float)):
|
||
return v > 0
|
||
if isinstance(v, str):
|
||
return v.strip().lower() in ("true", "1", "yes", "y", "是", "命中", "hit", "对", "✓", "√")
|
||
return False
|
||
|
||
|
||
def parse_judge_output(text, *, checklist=RICHNESS_CHECKLIST, genre_checklist=None, genre_key=None) -> dict:
|
||
"""纯函数：把 judge 的 LLM 文本输出解析成结构化丰富度评分（含 degraded 兜底）。可单测、零网络、零 agentscope。
|
||
|
||
成功 → {
|
||
score:int(11 条命中总数 0..11), # ★向后兼容字段：仅供 bake_off.richness_dist 读；非「质量总分」。
|
||
max:11, # 设计不设单一总分（诱导 Goodhart），主消费形状 = groups 三分组小计。
|
||
hits:[{name,layer,hit,why}*11], # 逐条：加 layer 层标（L2/L3/L4）
|
||
groups:{L2:{score,max}, L3:..., L4:...}, # ★三分组小计（§3.3 计分制主形状）
|
||
notes:str, degraded:False}
|
||
传 genre_checklist（品类扩展条目，W-GENRE 件④）时 additive 加：
|
||
genre:{key, hits:[…*N], score, max:N, groups:{…}} # ★品类分母独立小计（§4 规范二：不与通用混合平均）
|
||
解析失败（空/非 JSON 对象/缺 checks 数组）→ _degraded(...)（score=None, groups=None, degraded=True）。
|
||
|
||
对齐策略：优先按 name 对齐 judge 的 checks；name 对不上的按位置兜底（容忍模型改名/英文名）——
|
||
checks 数组约定 = 通用条目在前、品类条目紧随其后，位置兜底按段内偏移算。
|
||
layer 层标取自 checklist（不依赖 LLM 回传），故分组小计对模型改名鲁棒。
|
||
本函数**不含任何丰富度的代码判断**——只搬运 judge 的逐条裁决并按层聚合命中数。
|
||
"""
|
||
if not isinstance(text, str) or not text.strip():
|
||
return _degraded("judge 输出为空")
|
||
try:
|
||
# json_repair 容忍 markdown 围栏 / 前后赘语 / 尾逗号等 LLM 常见脏输出。
|
||
import json_repair
|
||
data = json_repair.loads(text)
|
||
except Exception as e: # noqa: BLE001 解析层任何异常都降级，绝不抛
|
||
return _degraded(f"judge 输出解析失败：{type(e).__name__}")
|
||
if not isinstance(data, dict):
|
||
return _degraded("judge 输出不是 JSON 对象")
|
||
checks_raw = data.get("checks")
|
||
if not isinstance(checks_raw, list) or not checks_raw:
|
||
return _degraded("judge 输出缺 checks 数组")
|
||
|
||
by_name = {}
|
||
for c in checks_raw:
|
||
if isinstance(c, dict):
|
||
nm = str(c.get("name", "")).strip()
|
||
if nm:
|
||
by_name[nm] = c
|
||
|
||
def _align(items, offset):
|
||
"""按 name 对齐（对不上按位置兜底，位置 = offset + 段内下标）→ [{name,layer,hit,why}]。"""
|
||
out = []
|
||
for i, (name, layer, _meaning) in enumerate(items):
|
||
c = by_name.get(name)
|
||
pos = offset + i
|
||
if c is None and pos < len(checks_raw) and isinstance(checks_raw[pos], dict):
|
||
c = checks_raw[pos] # name 对不上 → 位置兜底
|
||
hit = _coerce_bool(c.get("hit")) if isinstance(c, dict) else False
|
||
why = (str(c.get("why", "")).strip()[:200]) if isinstance(c, dict) else ""
|
||
out.append({"name": name, "layer": layer, "hit": hit, "why": why}) # layer 取自 checklist（权威）
|
||
return out
|
||
|
||
hits = _align(checklist, 0)
|
||
score = sum(1 for h in hits if h["hit"])
|
||
notes = str(data.get("notes", "")).strip()[:500]
|
||
result = {"score": score, "max": len(checklist), "hits": hits,
|
||
"groups": _group_subtotals(hits, checklist), "notes": notes, "degraded": False}
|
||
|
||
# 品类扩展段（W-GENRE 件④）：checks 数组的通用段之后是品类条目；**分母独立小计**（不并进通用 groups/score）。
|
||
if genre_checklist:
|
||
ghits = _align(genre_checklist, len(checklist))
|
||
result["genre"] = {
|
||
"key": genre_key,
|
||
"hits": ghits,
|
||
"score": sum(1 for h in ghits if h["hit"]),
|
||
"max": len(genre_checklist),
|
||
"groups": _group_subtotals(ghits, genre_checklist),
|
||
}
|
||
return result
|
||
|
||
|
||
def _collect_sources(game_id: str, *, max_bytes: int = _MAX_SRC_BYTES) -> str:
|
||
"""读产物 L3 源码（game-logic/core/render/balance/assets）拼成带文件头的一段文本，总量截断到 max_bytes。
|
||
|
||
只读 L3 游戏本体；读不到的文件跳过；全空 → 返回空串（上层据此降级、跳过 LLM、不白烧钱）。
|
||
"""
|
||
src_dir = cheap_run.game_dir(game_id) / "src"
|
||
parts = []
|
||
total = 0
|
||
for name in _L3_SOURCE_FILES:
|
||
p = src_dir / name
|
||
try:
|
||
if not p.is_file():
|
||
continue
|
||
txt = p.read_text(encoding="utf-8", errors="ignore")
|
||
except OSError:
|
||
continue # 单文件读失败不致命，跳过即可
|
||
chunk = f"\n// ===== {name} =====\n{txt}\n"
|
||
parts.append(chunk)
|
||
total += len(chunk.encode("utf-8"))
|
||
if total >= max_bytes:
|
||
break
|
||
joined = "".join(parts)
|
||
enc = joined.encode("utf-8")
|
||
if len(enc) > max_bytes: # 字节级截断兜底（防单个大文件超限撑爆 token）
|
||
joined = enc[:max_bytes].decode("utf-8", "ignore") + "\n// [源码已截断]"
|
||
return joined
|
||
|
||
|
||
def _build_judge_user(src_text: str, brief: str = "", genre_key: str = None, genre_checklist=None) -> str:
|
||
"""拼 judge 的 user 消息：brief（可选）+ 11 条清单（带层标）[+ 品类扩展清单] + 产物源码 + 严格 JSON 输出契约。
|
||
|
||
纯确定性字符串拼接。**不传品类时输出与既有版本逐字节一致**（金标复验基线稳，规范三锚定纪律）；
|
||
传品类时通用清单后 additive 追加品类段、输出契约条数 = 11+N（checks 通用在前、品类紧随）。
|
||
"""
|
||
checklist_lines = "\n".join(
|
||
f"{i + 1}. [{layer}] {name}：{meaning}" for i, (name, layer, meaning) in enumerate(RICHNESS_CHECKLIST)
|
||
)
|
||
brief_block = f"这款游戏的 brief（玩家想要的）：{brief}\n\n" if brief else ""
|
||
n_genre = len(genre_checklist) if genre_checklist else 0
|
||
if n_genre:
|
||
genre_lines = "\n".join(
|
||
f"{len(RICHNESS_CHECKLIST) + i + 1}. [{layer}] {name}：{meaning}"
|
||
for i, (name, layer, meaning) in enumerate(genre_checklist)
|
||
)
|
||
genre_block = (
|
||
f"\n另有 {n_genre} 条【{genre_key} 品类扩展清单】（分母独立、不与上面 {len(RICHNESS_CHECKLIST)} 条混合，同样逐条独立判命中）：\n"
|
||
f"{genre_lines}\n"
|
||
)
|
||
total_note = f"共 {len(RICHNESS_CHECKLIST) + n_genre} 条（通用 {len(RICHNESS_CHECKLIST)} 条在前、品类 {n_genre} 条紧随其后）"
|
||
name_note = "name 用上面各条的中文名"
|
||
else:
|
||
genre_block = ""
|
||
total_note = f"共 {len(RICHNESS_CHECKLIST)} 条"
|
||
name_note = f"name 用上面 {len(RICHNESS_CHECKLIST)} 条的中文名"
|
||
return (
|
||
f"{brief_block}"
|
||
f"下面是这款便宜档小游戏的 L3 源码。请逐条评判它是否命中这 {len(RICHNESS_CHECKLIST)} 条丰富度清单（每条前的 [L2]/[L3]/[L4] 是分层标注——"
|
||
"L2 内容丰富、L3 留存结构、L4 传播钩子——仅用于分组，不改变你对每条的独立判命中），"
|
||
"据**源码里真实实现了的玩法/数值/音效/解锁/结算画面/品类元数据**判断（别被注释或空壳骗：比如只 import 了 audioMusic "
|
||
"但收益处没真调 playSfx，则「音反馈」不算命中）：\n\n"
|
||
f"{checklist_lines}\n"
|
||
f"{genre_block}\n"
|
||
"=== 源码开始 ===\n"
|
||
f"{src_text}\n"
|
||
"=== 源码结束 ===\n\n"
|
||
"严格只输出以下 JSON（不要任何额外文字、不要 markdown 围栏）：\n"
|
||
f'{{"checks":[{{"name":"即时反馈","hit":true,"why":"一句中文理由"}}, … {total_note}，'
|
||
f'{name_note}、hit 为 true/false], "notes":"整体一句话点评"}}'
|
||
)
|
||
|
||
|
||
def _extract_text(resp) -> str:
|
||
"""从 ChatResponse 抽纯文本：content 是 TextBlock/ToolCall... 序列，只取 text 块（兼容 pydantic 块与 dict 块）。"""
|
||
content = getattr(resp, "content", None)
|
||
if content is None and isinstance(resp, dict):
|
||
content = resp.get("content")
|
||
if isinstance(content, str):
|
||
return content
|
||
parts = []
|
||
for b in (content or []):
|
||
if isinstance(b, dict):
|
||
if b.get("type") == "text" and isinstance(b.get("text"), str):
|
||
parts.append(b["text"])
|
||
elif getattr(b, "type", None) == "text":
|
||
t = getattr(b, "text", None)
|
||
if isinstance(t, str):
|
||
parts.append(t)
|
||
return "".join(parts)
|
||
|
||
|
||
async def verify_richness(game_id: str, *, brief: str = "", model=None, sources=None,
|
||
timeout: float = 90.0, max_src_bytes: int = _MAX_SRC_BYTES,
|
||
genre: str = None) -> dict:
|
||
"""生成完成后跑一次 LLM 丰富度评分（非阻塞·只报告）。**绝不抛、绝不阻断、绝不进 verdict / 不改达标**。
|
||
|
||
Args:
|
||
game_id: 本 run 游戏 id（读 games/amgen-<id>/src/ 下 L3 源码）。
|
||
brief: 本局 brief（喂 judge 做上下文，提升「情感锚/解锁」等判断质量）；缺省空串。
|
||
model: 可选注入的 LLM 客户端（单测用 fake model 注入、零网络）；None → 用 _bootstrap.build_cheap_model
|
||
新建**独立** M3 实例（judge token 不污染生成成本台账 costRmb——richness 是非阻塞 add-on）。
|
||
sources: 可选直接注入的源码文本（单测用，绕过文件 I/O）；None → 从 game_id 收集。
|
||
timeout: judge LLM 调用超时秒数；超时 → degraded（绝不卡死收口）。
|
||
max_src_bytes: 喂 judge 的源码上限。
|
||
genre: 品类 key（W-GENRE 件④）；命中 fixtures/genre-rubrics/<genre>.json（唯一数据源）→
|
||
同一次评分 additive 追加品类扩展条目（分母独立小计落 result["genre"]）；
|
||
None/fixture 缺 → 行为与既有版本完全一致（prompt 逐字节不变）。
|
||
|
||
Returns:
|
||
成功 → parse_judge_output 的结构化评分（含 judgeTokens 观测，best-effort）；
|
||
任何失败（src 空/LLM 异常/超时/解析失败）→ _degraded(...)（score=None, degraded=True, reason）。
|
||
"""
|
||
try:
|
||
src_text = sources if sources is not None else _collect_sources(game_id, max_bytes=max_src_bytes)
|
||
if not src_text or not src_text.strip():
|
||
return _degraded("产物 src 为空/读不到，跳过 LLM 评分")
|
||
|
||
# 惰性 import：agentscope / _bootstrap 较重，且让上面的 parse_judge_output 等纯函数能在无 agentscope 环境单测。
|
||
from agentscope.message import SystemMsg, UserMsg
|
||
m = model
|
||
if m is None:
|
||
import _bootstrap # 代理旁路 + key 注入由 _bootstrap.build_cheap_model 内部处理
|
||
m = _bootstrap.build_cheap_model(max_tokens=_JUDGE_MAX_TOKENS)
|
||
|
||
# 品类扩展（唯一数据源 = fixtures/genre-rubrics/<genre>.json，经 load_genre_checklist；
|
||
# 没有 fixture → 静默按 None——评分尺没有它的条目，别让评分半路造尺）。
|
||
genre_checklist = (load_genre_checklist(genre) or None) if genre else None
|
||
genre_key = genre if genre_checklist else None
|
||
|
||
sys_msg = SystemMsg(name="system", content=_JUDGE_SYSTEM_GENRE if genre_checklist else _JUDGE_SYSTEM)
|
||
user_msg = UserMsg(name="user", content=_build_judge_user(src_text, brief,
|
||
genre_key=genre_key, genre_checklist=genre_checklist))
|
||
# 硬超时包裹真模型调用：judge 卡住绝不能拖死收口（非阻塞铁律）。
|
||
resp = await asyncio.wait_for(m([sys_msg, user_msg]), timeout=timeout)
|
||
result = parse_judge_output(_extract_text(resp), genre_checklist=genre_checklist, genre_key=genre_key)
|
||
|
||
# best-effort 附 judge token（外部 LLM 调用可观测；与生成成本台账隔离——judge 用独立 model 实例）。
|
||
try:
|
||
ti, to = m.usage_sum()
|
||
result["judgeTokens"] = {"in": ti, "out": to, "total": ti + to}
|
||
except Exception: # noqa: BLE001 观测字段失败不影响评分主体
|
||
pass
|
||
return result
|
||
except (asyncio.TimeoutError, TimeoutError):
|
||
return _degraded(f"judge LLM 超时（>{timeout}s）")
|
||
except Exception as e: # noqa: BLE001 非阻塞铁律：任何异常（网络/import/解析）都降级，绝不抛断生成主链
|
||
return _degraded(f"judge 异常：{type(e).__name__}: {e}")
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# 玩法地板判定(W-AXIS 波2 · 质量模型 SoT 裁定三的代码兑现)
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# 与上面的丰富度评分是两物,别混:
|
||
# · 丰富度(verify_richness):评「有多丰富/多耐玩」的【程度】——非阻塞软信号,只观测、永不拒发(L2)。
|
||
# · 玩法地板(judge_gameplay_floor):裁「作为游戏成不成立」的【地板布尔】——阻断放行(L1 的判定半)。
|
||
# 裁定三钉死的分界:判定只裁三类【拒绝】的有无,不判程度高低:
|
||
# · broken —— 真人玩不通 / 到不了终局 / 起不来 / 卡死一屏 / 核心操作无响应;
|
||
# · hollow —— 空壳:核心循环没有决策层(点哪都一样、无输赢取舍、纯自动结算的挂机点击),判「有没有」不判「多好」;
|
||
# · off_brief —— 不切题:画面/玩法与 brief 要的游戏是两回事。
|
||
# 三条铁律(裁定三):
|
||
# ① 出题≠被考(2026-07-10 创始人修正边界):判定只吃运行证据不吃自报;判定模型允许 M3(生成同源)——
|
||
# 地板只裁「有无」不裁品味、同源盲区风险最小,且 07-09 spike 实证 M3 当场抓出 M3 自己写的渲染 bug;
|
||
# 同源残余风险由金标正反例+创始人抽玩定标兜底。修正动因=glm-5.2 唯一通道(闲鱼二手中转)图像支持
|
||
# 按池轮换、07-09 整日全盲,判定层可用性不能押注二手供给。SoT 修订随验收 v2 双评审落档。
|
||
# ② fail-closed:判定失败/评不出/无证据 → 不放行、标 degraded(绝不静默降级成自动通过)。
|
||
# ③ 校准与阻断开关:金标正反例复验 + 创始人抽玩定标;judge.blocking 开关默认 true,可一键整体回退到修订前口径。
|
||
# 判定模型走 new-api:生产档由 generation.yaml judge.model 定(现 MiniMax-M3,一手直连通道);
|
||
# 单局判定成本目标 ≤¥0.3(实测 4-6 帧约 ¥0.10,记账进 judge.costRmb)。
|
||
|
||
# 判定默认档(genconfig 不可达时的编译回落,生产以 YAML 为准):2026-07-10 随创始人决策切 M3。
|
||
# 07-09 探针实证 glm-5.2 与 MiniMax-M3 均能读图;glm-5.2 因二手供给整日全盲弃用。
|
||
_JUDGE_DEFAULT_MODEL = "MiniMax-M3"
|
||
# 三类拒绝的机读键 → 对外类名(off_brief 用下划线,与 LLM 输出的 camelCase offBrief 解耦)。
|
||
_FLOOR_CLASSES = (("broken", "broken"), ("hollow", "hollow"), ("offBrief", "off_brief"))
|
||
|
||
_FLOOR_SYSTEM = (
|
||
"你是轻量小游戏的【玩法地板判定 agent】。你不是打分器——只判「这款游戏作为一个游戏,地板成不成立」,"
|
||
"只裁【有没有】、绝不裁【做得多好】(丰富/耐玩/好玩的程度问题不归你,归另一套软评分)。"
|
||
"你只看运行证据(真玩截图 / 运行日志 / 取证状态时间线),绝不采信任何自报、承诺或源码里写了什么。"
|
||
"逐条判这三类【拒绝】是否成立:\n"
|
||
"· broken(坏死):真人根本玩不通 —— 起不来 / 一直卡在同一屏 / 核心操作点了没反应 / 到不了任何终局或结算。\n"
|
||
"· hollow(空壳):核心循环没有决策层 —— 点哪都一样、没有输赢、没有取舍、判错没有代价,纯自动结算的挂机点击。"
|
||
"只判「有没有决策层」,不判「决策做得多有层次」。\n"
|
||
"· off_brief(不切题):画面与玩法跟 brief 要的游戏是两回事(主题漂移 / 品类不符)。\n"
|
||
"任一类成立 → 整体判 reject;三类都不成立 → 判 accept。"
|
||
"证据不足以断定某一类是否成立时,该类按【成立(有问题)】算——地板判定 fail-closed,宁可错拦、不可放过坏游戏"
|
||
"(后面有人工复核与金标校准兜)。严格只输出 JSON,不要任何额外文字、不要 markdown 围栏。"
|
||
)
|
||
|
||
_FLOOR_OUTPUT_CONTRACT = (
|
||
"严格只输出以下 JSON(problem=true 表示「这一类有问题、应拒绝」,why 给一句中文依据;"
|
||
"imagesSeen=你在本条消息里【真实看到】的截图张数,整数——一张都看不到就如实填 0,"
|
||
"这个字段用于检测图像传输故障,绝不据它奖惩你):\n"
|
||
'{"imagesSeen":0,"broken":{"problem":false,"why":"…"},"hollow":{"problem":false,"why":"…"},'
|
||
'"offBrief":{"problem":false,"why":"…"},"verdict":"accept","notes":"整体一句话"}'
|
||
)
|
||
|
||
|
||
def _degraded_floor(reason: str) -> dict:
|
||
"""fail-closed 降级结果(裁定三②):评不出/无证据/调用失败 → 不放行、标 degraded。
|
||
|
||
accepted=False 是【地板判定的 fail-closed 语义】,与丰富度的 score=None(非阻塞观测)完全不同:
|
||
这里 degraded 直接构成「拒绝」(rejectClasses=['degraded']),阻断放行,由人工复核兜。
|
||
"""
|
||
return {"accepted": False, "verdict": "reject", "rejectClasses": ["degraded"],
|
||
"checks": [], "notes": None, "degraded": True, "reason": reason}
|
||
|
||
|
||
def parse_floor_judgment(text) -> dict:
|
||
"""纯函数:把判定 LLM 的 JSON 输出解析成结构化地板裁决(含 fail-closed 兜底)。可单测、零网络、零 agentscope。
|
||
|
||
成功 → {
|
||
accepted: bool, # = 三类均无问题 且 LLM verdict 未判 reject(两者取交,任一拒即拒)
|
||
verdict: "accept"|"reject",
|
||
rejectClasses: [str], # 命中的拒绝类子集(broken/hollow/off_brief)
|
||
checks: [{class, problem, why}*3],
|
||
notes: str, degraded: False}
|
||
解析失败 / 非对象 / 缺任一类判定字段 → _degraded_floor(...)(accepted=False、fail-closed)。
|
||
|
||
每一类容忍两种形状:{problem,why} 对象,或裸 bool(模型偷懒直接给 true/false);缺字段一律 fail-closed
|
||
(无法确认地板 → 保守拒绝),绝不把「没判出来」静默当成通过。
|
||
"""
|
||
if not isinstance(text, str) or not text.strip():
|
||
return _degraded_floor("判定输出为空")
|
||
try:
|
||
import json_repair # 容忍 markdown 围栏 / 前后赘语 / 尾逗号等 LLM 常见脏输出
|
||
data = json_repair.loads(text)
|
||
except Exception as e: # noqa: BLE001 解析层任何异常都 fail-closed,绝不抛
|
||
return _degraded_floor(f"判定输出解析失败：{type(e).__name__}")
|
||
if not isinstance(data, dict):
|
||
return _degraded_floor("判定输出不是 JSON 对象")
|
||
checks = []
|
||
for raw_key, cname in _FLOOR_CLASSES:
|
||
node = data.get(raw_key)
|
||
if isinstance(node, dict):
|
||
problem = _coerce_bool(node.get("problem"))
|
||
why = str(node.get("why", "")).strip()[:200]
|
||
elif node is not None: # 裸 bool/字符串:模型没按对象给,也接住
|
||
problem = _coerce_bool(node)
|
||
why = ""
|
||
else: # 缺这一类判定 → 无法确认地板 → fail-closed(不静默判过)
|
||
return _degraded_floor(f"判定输出缺 {raw_key} 字段(无法确认地板,fail-closed)")
|
||
checks.append({"class": cname, "problem": problem, "why": why})
|
||
reject_classes = [c["class"] for c in checks if c["problem"]]
|
||
llm_verdict = str(data.get("verdict", "")).strip().lower()
|
||
# accepted 需两者一致取 accept:三类均无问题 且 LLM 自评 verdict 不是 reject(任一给拒绝信号 → 拒,fail-closed 偏严)。
|
||
accepted = (not reject_classes) and (llm_verdict != "reject")
|
||
# imagesSeen:模型自报真实看到的截图张数(图像盲检测的结构化信号;缺失/非数字 → None=未知,不触发盲重掷)。
|
||
# 背景(2026-07-09 基线终审实锤):网关对 glm-5.2 按请求体确定性路由,~15% 载荷永远落在丢图通道——
|
||
# 同载荷重试必盲、任何字节微扰即换通道。靠措辞 grep 判盲太脆,改为契约字段模型自报。
|
||
images_seen = None
|
||
try:
|
||
raw_seen = data.get("imagesSeen")
|
||
if raw_seen is not None and not isinstance(raw_seen, bool):
|
||
images_seen = max(0, int(raw_seen))
|
||
except (TypeError, ValueError):
|
||
images_seen = None
|
||
return {"accepted": accepted, "verdict": "accept" if accepted else "reject",
|
||
"rejectClasses": reject_classes, "checks": checks, "imagesSeen": images_seen,
|
||
"notes": str(data.get("notes", "")).strip()[:300], "degraded": False}
|
||
|
||
|
||
class _JudgeResp:
|
||
"""判定多模态调用的极简响应载体(content=纯文本);与丰富度 _extract_text 的 str 分支兼容。"""
|
||
|
||
def __init__(self, text: str):
|
||
self.content = text if isinstance(text, str) else ""
|
||
|
||
|
||
class _NewapiVisionModel:
|
||
"""判定用的薄多模态客户端:直连 new-api /v1/chat/completions(OpenAI 兼容,支持 image_url 图像块)。
|
||
|
||
为什么不复用 build_cheap_model(agentscope M3):判定层要的是薄而可控的多模态 HTTP 客户端——
|
||
直连网关对图像块形状可控(image_url data-URI 已 probe 坐实)、finish_reason/reasoning 元信息可留痕,
|
||
与生成栈(agentscope 全家桶)解耦;仍走 new-api 统一出口(base+/v1 + NEWAPI_KEY),不违统一网关铁律。
|
||
(判定模型 2026-07-10 起=MiniMax-M3,由 YAML judge.model 定;出题≠被考边界修正见 §判定地板头注。)
|
||
trust_env=False:判定目标在内网 Tailscale(new-api 100.64.0.8),彻底绕开本机 clash 代理(fake-ip 拦本地/内网,
|
||
§7 内网直连必绕代理)。records 记 (in,out,cached) 供 judge 段成本记账(与生成成本台账隔离——判定用独立档)。
|
||
"""
|
||
|
||
def __init__(self, model_name: str, base_url: str, api_key: str, *,
|
||
max_tokens: int = 1500, timeout: float = 120.0):
|
||
self.model_name = model_name
|
||
self.base_url = base_url.rstrip("/")
|
||
self.api_key = api_key
|
||
self.max_tokens = max_tokens
|
||
self.timeout = timeout
|
||
self.records = [] # [(input_tokens, output_tokens, cached_tokens)]
|
||
self.last_meta = {} # 最近一次响应的诊断元信息(finishReason/reasoningChars)——content 空时据此归因
|
||
|
||
async def __call__(self, messages):
|
||
import httpx # noqa: PLC0415 惰性 import(顶层零重依赖;与 cheap_verify 纯函数可无 httpx 环境单测同源)
|
||
body = {"model": self.model_name, "max_tokens": self.max_tokens, "messages": messages}
|
||
async with httpx.AsyncClient(trust_env=False, timeout=self.timeout) as h:
|
||
r = await h.post(
|
||
f"{self.base_url}/chat/completions",
|
||
headers={"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"},
|
||
json=body,
|
||
)
|
||
r.raise_for_status()
|
||
d = r.json()
|
||
u = d.get("usage") or {}
|
||
pi = int(u.get("prompt_tokens") or u.get("input_tokens") or 0)
|
||
po = int(u.get("completion_tokens") or u.get("output_tokens") or 0)
|
||
cached = 0
|
||
det = u.get("prompt_tokens_details") or {}
|
||
if isinstance(det, dict):
|
||
cached = int(det.get("cached_tokens") or 0)
|
||
self.records.append((pi, po, cached))
|
||
ch = (d.get("choices") or [{}])[0]
|
||
msg = ch.get("message") or {}
|
||
text = msg.get("content", "")
|
||
# 判定器自身的真相层:留 finish_reason 与 reasoning 长度。glm-5.2 带思考(reasoning_content),思考长度是
|
||
# 随机变量——吃光 max_tokens 时 content 为空、finish_reason=length(w2b-verify 2026-07-09 实测案),据此可诊断。
|
||
self.last_meta = {"finishReason": ch.get("finish_reason"),
|
||
"reasoningChars": len(msg.get("reasoning_content") or "")}
|
||
return _JudgeResp(text if isinstance(text, str) else "")
|
||
|
||
def usage_sum(self):
|
||
return (sum(r[0] for r in self.records), sum(r[1] for r in self.records))
|
||
|
||
|
||
def _wg1_evidence_dir(game_id: str) -> Path:
|
||
"""判卷证据目录 = _wg1-gen/<id>/evidence/(九门 play 落 first-paint/after-play/midplay/game-log/verdict 处)。"""
|
||
return cheap_run.wg1_game_dir(game_id) / "evidence"
|
||
|
||
|
||
def _natural_key(name: str):
|
||
"""midplay-2 排在 midplay-10 前:按名字里的数字段自然排序(纯字母段保原序)。"""
|
||
import re # noqa: PLC0415
|
||
return [int(t) if t.isdigit() else t for t in re.split(r"(\d+)", name)]
|
||
|
||
|
||
def _collect_frames(evidence_dir, *, max_frames: int = 6) -> list:
|
||
"""收 evidence 目录的真玩截图 → [data:image/png;base64,...],顺序 first-paint → 局中连拍 midplay-* → after-play。
|
||
|
||
超 max_frames 时保头尾(首帧+局末)、中间连拍均匀采样(判定需要「局前→局中→局末」的时间跨度,不能只留一头)。
|
||
读不到目录 / 无截图 → []（上层据此 fail-closed:真玩过的游戏必有 first-paint,缺失即证据异常）。
|
||
"""
|
||
try:
|
||
ev = Path(evidence_dir)
|
||
if not ev.is_dir():
|
||
return []
|
||
ordered = []
|
||
fp = ev / "first-paint.png"
|
||
if fp.is_file():
|
||
ordered.append(fp)
|
||
ordered.extend(sorted(ev.glob("midplay-*.png"), key=lambda p: _natural_key(p.name)))
|
||
ap = ev / "after-play.png"
|
||
if ap.is_file():
|
||
ordered.append(ap)
|
||
if len(ordered) > max_frames:
|
||
head, tail, mid = ordered[0], ordered[-1], ordered[1:-1]
|
||
keep = max_frames - 2
|
||
if keep <= 0:
|
||
ordered = [head, tail]
|
||
else:
|
||
step = len(mid) / keep
|
||
ordered = [head] + [mid[int(i * step)] for i in range(keep)] + [tail]
|
||
out = []
|
||
for p in ordered:
|
||
try:
|
||
out.append("data:image/png;base64," + base64.b64encode(p.read_bytes()).decode())
|
||
except OSError:
|
||
continue
|
||
return out
|
||
except Exception: # noqa: BLE001 证据装配失败 → 返回空(上层 fail-closed),绝不抛
|
||
return []
|
||
|
||
|
||
def _read_state_timeline(evidence_dir) -> str:
|
||
"""从九门 verdict.json 抽一段取证状态时间线(判定的文字证据,补截图看不清的数值/终态信号)。
|
||
|
||
只搬运机械门已测出的客观信号(不加解读):H_progress 的 score 玩前/玩后、E_live 的 played/phase/活性、
|
||
latch 的终局 phase 与是否驻留、D_render 的亮度。读不到/坏 → 空串(判定仍靠截图,不伪造)。
|
||
"""
|
||
try:
|
||
ev = Path(evidence_dir)
|
||
p = ev / "verdict.json"
|
||
if not p.is_file():
|
||
return ""
|
||
v = json.loads(p.read_text(encoding="utf-8"))
|
||
guards = (v or {}).get("guards") or {}
|
||
lines = []
|
||
hg = guards.get("H_progress") or {}
|
||
for c in (hg.get("checks") or []):
|
||
if isinstance(c, dict):
|
||
lines.append(f"进展[{c.get('path')}] {c.get('op')}: 玩前={c.get('before')} 玩后={c.get('after')}"
|
||
f"（机械门判 {'达成' if c.get('pass') else '未达成'}）")
|
||
latch = hg.get("latch") or {}
|
||
if isinstance(latch, dict) and latch:
|
||
lines.append(f"终局 latch: 到达终态={bool(latch.get('pass'))} 此刻 phase={latch.get('phaseNow') or latch.get('after')}"
|
||
f" {(latch.get('reason') or '')[:60]}")
|
||
el = guards.get("E_live") or {}
|
||
if isinstance(el, dict) and el:
|
||
lines.append(f"活性 E_live: 真进过游玩={el.get('played')} 玩后 phase={el.get('phase1')}"
|
||
f" 去重画面态数={el.get('distinctStates')} 状态真变={el.get('liveByState')}")
|
||
dr = guards.get("D_render") or {}
|
||
if isinstance(dr, dict) and dr:
|
||
lines.append(f"渲染 D_render: 亮度={dr.get('bright')} 最大通道={dr.get('maxCh')}")
|
||
return "\n".join(lines)
|
||
except Exception: # noqa: BLE001 取证时间线 best-effort,读不出降级空串
|
||
return ""
|
||
|
||
|
||
def _read_game_log_text(evidence_dir, *, max_entries: int = 12, max_chars: int = 800) -> str:
|
||
"""从 evidence/game-log.json 摘运行时日志(游戏语义 + 插件告警),给判定补运行期信号。读不到/坏 → 空串。"""
|
||
try:
|
||
ev = Path(evidence_dir)
|
||
p = ev / "game-log.json"
|
||
if not p.is_file():
|
||
return ""
|
||
arr = json.loads(p.read_text(encoding="utf-8"))
|
||
if not isinstance(arr, list) or not arr:
|
||
return ""
|
||
|
||
def _fmt(e):
|
||
if not isinstance(e, dict):
|
||
return str(e)[:100]
|
||
return f"[{e.get('scope')}:{e.get('tag')}] {e.get('msg') or ''}"[:100]
|
||
|
||
return "\n".join(_fmt(e) for e in arr[-max_entries:])[:max_chars]
|
||
except Exception: # noqa: BLE001 日志摘要 best-effort
|
||
return ""
|
||
|
||
|
||
def _build_floor_messages(brief: str, state_text: str, log_text: str, data_uris: list) -> list:
|
||
"""拼判定的 OpenAI 多模态消息:brief + 取证时间线 + 日志 + 「首帧/局中/局末」截图序列 + 输出契约。"""
|
||
head = (
|
||
f"这款游戏的 brief（玩家想要的）：{brief or '(未提供)'}\n\n"
|
||
"下面给你这一局【真玩】留下的运行证据。请只据这些运行证据判三类拒绝是否成立。\n\n"
|
||
f"【取证状态时间线（机械门客观测得，未加解读）】\n{state_text or '（无）'}\n\n"
|
||
f"【运行时日志（节选）】\n{log_text or '（无）'}\n\n"
|
||
f"【真玩截图】接下来 {len(data_uris)} 张图按时间顺序给出：第 1 张=首帧（刚进入），"
|
||
"中间=局中连拍（真玩过程），最后 1 张=局末（真玩结束时）。请对照 brief 看这些画面是不是一款玩得通、"
|
||
"有决策层、且切题的游戏：\n"
|
||
)
|
||
parts = [{"type": "text", "text": head}]
|
||
for uri in data_uris:
|
||
parts.append({"type": "image_url", "image_url": {"url": uri}})
|
||
parts.append({"type": "text", "text": "\n" + _FLOOR_OUTPUT_CONTRACT})
|
||
return [{"role": "system", "content": _FLOOR_SYSTEM},
|
||
{"role": "user", "content": parts}]
|
||
|
||
|
||
def _build_judge_model(model_name: str, max_tokens: int, timeout: float):
|
||
"""装判定多模态客户端(直连 new-api /v1)。key 从 env/凭据档解析(内网阶段单一事实源)。"""
|
||
import _bootstrap # noqa: PLC0415 代理旁路时序与 key 注入由 _bootstrap/client 处理
|
||
_bootstrap.ensure_api_key_env()
|
||
from worker import client # noqa: PLC0415
|
||
base = client.resolve_base_url().rstrip("/")
|
||
if not base.endswith("/v1"):
|
||
base = base + "/v1"
|
||
return _NewapiVisionModel(model_name, base, client.get_api_key(),
|
||
max_tokens=max_tokens, timeout=timeout)
|
||
|
||
|
||
def _estimate_judge_cost(model_name: str, ti: int, to: int, cached: int):
|
||
"""据 new-api /api/pricing 权威倍率把判定 token 折成 ¥(与生成成本台账同口径 observability.cost.compute)。
|
||
|
||
best-effort:取价失败(网关不可达 / 无 httpx)→ None（judge 段记 tokens、costRmb 留空,不伪造成本）。
|
||
"""
|
||
try:
|
||
from observability import cost as _cost, newapi_pricing as _np # noqa: PLC0415
|
||
params = _np.fetch_pricing_params()
|
||
if not params:
|
||
return None
|
||
d = _cost.compute(model_name, ti, to, params["pricing"], params["qpu"],
|
||
params["usd_rate"], group_ratio=1.0, cached_tokens=cached)
|
||
return round(d["rmb"], 5)
|
||
except Exception: # noqa: BLE001 成本记账 best-effort,绝不连累判定主体
|
||
return None
|
||
|
||
|
||
def _persist_judge_json(evidence_dir, result: dict, raw_text: str) -> None:
|
||
"""判定真相层落盘(best-effort,绝不抛):evidence/judge.json = 完整地板裁决 + LLM 原始输出头(截断 2000 字)。
|
||
|
||
动机(W-AXIS 波2 验收发现,2026-07-09):判定结果此前只进 run-summary → 后端 trace,本地证据目录零痕迹——
|
||
verdict.json 孤零零 pass=true,盘上看不出这局判定发生过什么;degraded(如「判定输出为空」)更无从诊断。
|
||
判定器自己也要有真相层:每次判定(含 degraded)都在证据目录留完整裁决与原始输出。目录不存在则静默跳过(单测 fake gid)。
|
||
"""
|
||
try:
|
||
ev = Path(evidence_dir) if evidence_dir else None
|
||
if ev is None or not ev.is_dir():
|
||
return
|
||
payload = dict(result)
|
||
payload["rawTextHead"] = (raw_text or "")[:2000]
|
||
payload["ts"] = int(time.time() * 1000)
|
||
(ev / "judge.json").write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
except Exception: # noqa: BLE001 落盘失败绝不连累判定主体
|
||
pass
|
||
|
||
|
||
async def judge_gameplay_floor(game_id: str, *, brief: str = "", model=None, frames=None,
|
||
state_text=None, log_text=None, evidence_dir=None,
|
||
model_name: str = None, max_tokens: int = 1500,
|
||
timeout: float = 120.0, max_frames: int = 6) -> dict:
|
||
"""对一局真玩证据跑独立多模态玩法地板判定(阻断权威;裁定三)。**fail-closed**:任何失败/无证据 → degraded（=拒绝）。
|
||
|
||
与 verify_richness 的非阻塞铁律相反——这里失败不是「静默降级放行」,而是「fail-closed 拒绝」:
|
||
评不出的游戏不许自动过,标 degraded 交人工复核。
|
||
|
||
Args:
|
||
game_id: 本局 id（读 _wg1-gen/<id>/evidence/ 下截图/日志/verdict）。
|
||
brief: 本局 brief（off_brief 判定必需）。
|
||
model: 可注入的判定 LLM 客户端（单测用 fake，零网络）；None → 建 glm-5.2 多模态客户端。
|
||
frames/state_text/log_text: 可直接注入的证据（单测用，绕文件 I/O）；None → 从 evidence_dir 收集。
|
||
evidence_dir: 证据目录；None → _wg1-gen/<id>/evidence/。
|
||
model_name/max_tokens/timeout/max_frames: 判定档旋钮（缺省走 glm-5.2 / 1500 / 120s / 6 帧）。
|
||
|
||
Returns:
|
||
parse_floor_judgment 的结构（附 judgeTokens/costRmb/model/frames 观测），或 _degraded_floor(...)（fail-closed）。
|
||
"""
|
||
mn = model_name or _JUDGE_DEFAULT_MODEL
|
||
ev = None
|
||
try:
|
||
ev = evidence_dir if evidence_dir is not None else _wg1_evidence_dir(game_id)
|
||
imgs = frames if frames is not None else _collect_frames(ev, max_frames=max_frames)
|
||
if not imgs:
|
||
# 无真玩截图:真玩过的游戏必落 first-paint.png,缺失即证据异常 → fail-closed（绝不无证据放行）。
|
||
r = _degraded_floor("真玩截图证据缺失（无 first-paint/midplay/after-play），fail-closed 不放行")
|
||
r["model"] = mn
|
||
_persist_judge_json(ev, r, "")
|
||
return r
|
||
st = state_text if state_text is not None else _read_state_timeline(ev)
|
||
lg = log_text if log_text is not None else _read_game_log_text(ev)
|
||
m = model if model is not None else _build_judge_model(mn, max_tokens, timeout)
|
||
|
||
async def _roll(msgs):
|
||
"""一掷判定 = 一次调用 + 空输出有界重试一次(2026-07-09 w2b-verify 实测案):glm-5.2 带思考,
|
||
思考长度是随机变量——吃光 max_tokens 时 content 为空;同证据复跑即可得答案(主会话复现坐实)。
|
||
重试前放大 max_tokens 对症;绝不无限重试(判定成本有账)。放大取 max(原值, ×2封顶6000):
|
||
base 已调大(YAML 202752)时绝不反向缩小——旧式 min(×2,6000) 在大 base 下会把上限缩回 6000。"""
|
||
t = ""
|
||
used = 0
|
||
for attempt in range(2):
|
||
resp = await asyncio.wait_for(m(msgs), timeout=timeout)
|
||
used = attempt + 1
|
||
t = _extract_text(resp)
|
||
if isinstance(t, str) and t.strip():
|
||
break
|
||
if attempt == 0 and hasattr(m, "max_tokens"):
|
||
cur = int(getattr(m, "max_tokens", 1500))
|
||
m.max_tokens = max(cur, min(cur * 2, 6000))
|
||
return t, used
|
||
|
||
text, attempts_used = await _roll(_build_floor_messages(brief, st, lg, imgs))
|
||
result = parse_floor_judgment(text)
|
||
# 图像盲微扰重掷(2026-07-09 基线终审实锤):网关对 glm-5.2 按请求体【确定性】路由,部分载荷永远落在
|
||
# 丢图通道——同载荷重试必盲(3 例假阴复判 3/3 仍盲),而 brief 尾加一个空格微扰载荷即换路由、当场看见。
|
||
# 故:送了帧而模型自报 imagesSeen=0 → 微扰重掷一次;重掷仍自报 0 → 这是判定仪器故障不是游戏证据缺失,
|
||
# fail-closed degraded 归因「图像通道盲」交人工,不再把好游戏错杀成 hollow/off_brief(基线 3/20 假阴案)。
|
||
# imagesSeen=None(模型没报/老 fake)不触发,向后兼容。
|
||
if imgs and not result.get("degraded") and result.get("imagesSeen") == 0:
|
||
text2, used2 = await _roll(_build_floor_messages(brief + " ", st, lg, imgs))
|
||
attempts_used += used2
|
||
second = parse_floor_judgment(text2)
|
||
if not second.get("degraded") and second.get("imagesSeen") == 0:
|
||
second = _degraded_floor("判定模型两掷均自报看不见截图(网关同载荷确定性丢图)——判定仪器故障 fail-closed,待人工复核")
|
||
second["imageBlindRetried"] = True
|
||
result, text = second, text2
|
||
result["model"] = mn
|
||
result["frames"] = len(imgs)
|
||
if attempts_used > 1:
|
||
result["retries"] = attempts_used - 1
|
||
# 判定器真相层:附最近响应的 finish_reason(空输出归因的第一手证据;fake 模型无此属性则省略)。
|
||
meta = getattr(m, "last_meta", None)
|
||
if isinstance(meta, dict) and meta.get("finishReason") is not None:
|
||
result["finishReason"] = meta.get("finishReason")
|
||
# 判定成本记账（独立档，不污染生成 costRmb）——best-effort。
|
||
try:
|
||
ti, to = m.usage_sum()
|
||
cached = sum(r[2] for r in getattr(m, "records", []) if len(r) > 2)
|
||
result["judgeTokens"] = {"in": ti, "out": to, "total": ti + to}
|
||
result["costRmb"] = _estimate_judge_cost(mn, ti, to, cached)
|
||
except Exception: # noqa: BLE001 成本/观测字段失败不影响判定主体
|
||
pass
|
||
_persist_judge_json(ev, result, text)
|
||
return result
|
||
except (asyncio.TimeoutError, TimeoutError):
|
||
r = _degraded_floor(f"判定 LLM 超时（>{timeout}s），fail-closed 不放行")
|
||
r["model"] = mn
|
||
_persist_judge_json(ev, r, "")
|
||
return r
|
||
except Exception as e: # noqa: BLE001 fail-closed:任何异常（网络/import/解析）都判 degraded=拒绝,绝不放行
|
||
r = _degraded_floor(f"判定异常：{type(e).__name__}: {e}")
|
||
r["model"] = mn
|
||
_persist_judge_json(ev, r, "")
|
||
return r
|
||
|
||
|
||
def _judge_cfg() -> dict:
|
||
"""读判定档配置（genconfig judge.*；worker 不可达时回落编译默认，保证无 agentscope 环境也能取值/单测）。"""
|
||
try:
|
||
from worker import genconfig # noqa: PLC0415
|
||
g = genconfig.get
|
||
except Exception: # noqa: BLE001 无 worker（纯单测环境）→ 用编译默认
|
||
def g(_area, _key, default):
|
||
return default
|
||
return {
|
||
"blocking": bool(g("judge", "blocking", True)),
|
||
"model": g("judge", "model", _JUDGE_DEFAULT_MODEL),
|
||
"max_tokens": int(g("judge", "max_tokens", 1500)),
|
||
"timeout_s": float(g("judge", "timeout_s", 120.0)),
|
||
"max_frames": int(g("judge", "max_frames", 6)),
|
||
"cost_target_rmb": float(g("judge", "cost_target_rmb", 0.3)),
|
||
}
|
||
|
||
|
||
def _writeback_verdict_json(evidence_dir, accepted: bool, judge_summary: dict) -> None:
|
||
"""把最终验收权威增量写回盘上 evidence/verdict.json(best-effort,绝不抛)。
|
||
|
||
字段只加不减:`pass` 物理保留(语义=预筛/机械九门),新增 `accepted`(=预筛∧玩法判定,最终权威)与
|
||
`judge` 摘要段。动机(W-AXIS 波2 验收发现):此前判定只进 run-summary→后端 trace,盘上 verdict.json
|
||
孤零零 pass=true——证据链与真实验收结论脱节(§6.13 亲眼验收看的就是盘上证据)。verdict.json 不存在则跳过。
|
||
"""
|
||
try:
|
||
p = (Path(evidence_dir) / "verdict.json") if evidence_dir else None
|
||
if p is None or not p.is_file():
|
||
return
|
||
v = json.loads(p.read_text(encoding="utf-8"))
|
||
if not isinstance(v, dict):
|
||
return
|
||
v["accepted"] = bool(accepted)
|
||
v["judge"] = judge_summary
|
||
p.write_text(json.dumps(v, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
except Exception: # noqa: BLE001 写回失败绝不连累判定主链
|
||
pass
|
||
|
||
|
||
async def apply_gameplay_judge(summary: dict, *, game_id: str, brief: str = "",
|
||
model=None, blocking=None, evidence_dir=None,
|
||
prefilter=None) -> dict:
|
||
"""把玩法地板判定接到 run-summary，落 ok 语义切换（W-AXIS 波2 · SoT 裁定三代码兑现）。
|
||
|
||
ok 语义（裁定三）:`ok = 预筛 ∧ 判定`。verdict 的 `pass` 字段物理保留、语义收窄为「预筛（机械九门）通过」,
|
||
新增 `accepted`（= 预筛 ∧ 玩法地板判定）与 `judge` 段（裁决/三类理由/degraded/成本）——字段只加不减。
|
||
|
||
· 预筛 = 显式 `prefilter` 入参;缺省(None)回落 summary 现有的 ok。**预筛必须是机械九门口径**(裁定一:
|
||
driven 时九门全量 AND)——Service 路 summary.ok=九门 judge.passed 天然合规可走缺省;CLI(cheap_studio)路
|
||
summary.ok=finished(模型自称收敛),**不是**九门信号,必须显式传 prefilter=finished∧verdict.pass
|
||
(2026-07-09 W3 真跑暴露:三门挂的局按 finished 预筛被判定放行成 accepted=True,违裁定三,已修)。
|
||
· 判定只对【过筛者】跑（省 ¥:预筛没过的游戏已被机械门拒,不再花判定钱;对齐诊断档 §5「预筛先挡明显死」）。
|
||
· blocking=True（默认）→ ok=accepted（判定阻断放行）;blocking=False → ok=预筛（判定观测不阻断,整体回退位）。
|
||
· fail-closed:证据缺/调用失败/解析失败 → judge.degraded=True → floor 不成立 → (blocking) accepted=False。
|
||
|
||
绝不抛（顶层兜底）:additive 写 summary['accepted'] 与 summary['judge']，并按上式更新 summary['ok']。
|
||
blocking/evidence_dir 缺省 None：blocking→读 genconfig；evidence_dir→按 game_id 推 _wg1-gen/<id>/evidence/。
|
||
"""
|
||
try:
|
||
cfg = _judge_cfg()
|
||
blk = cfg["blocking"] if blocking is None else bool(blocking)
|
||
ev = evidence_dir if evidence_dir is not None else _wg1_evidence_dir(game_id)
|
||
prefilter = bool(summary.get("ok")) if prefilter is None else bool(prefilter)
|
||
if not prefilter:
|
||
# 预筛未过:机械门已拒,不花判定钱。accepted=预筛(False);ok 维持 False。
|
||
summary["judge"] = {"ran": False, "blocking": blk,
|
||
"reason": "预筛（机械九门）未过，不跑玩法判定"}
|
||
summary["accepted"] = False
|
||
summary["ok"] = False
|
||
_writeback_verdict_json(ev, False, summary["judge"])
|
||
return summary
|
||
floor = await judge_gameplay_floor(
|
||
game_id, brief=brief, model=model, evidence_dir=ev,
|
||
model_name=cfg["model"], max_tokens=cfg["max_tokens"],
|
||
timeout=cfg["timeout_s"], max_frames=cfg["max_frames"])
|
||
floor_ok = bool(floor.get("accepted")) and not bool(floor.get("degraded"))
|
||
accepted = prefilter and floor_ok
|
||
summary["judge"] = {
|
||
"ran": True, "blocking": blk,
|
||
"verdict": floor.get("verdict"),
|
||
"accepted": bool(floor.get("accepted")),
|
||
"rejectClasses": floor.get("rejectClasses"),
|
||
"checks": floor.get("checks"),
|
||
"degraded": bool(floor.get("degraded")),
|
||
"reason": floor.get("reason"),
|
||
"notes": floor.get("notes"),
|
||
"model": floor.get("model"),
|
||
"frames": floor.get("frames"),
|
||
"costRmb": floor.get("costRmb"),
|
||
"judgeTokens": floor.get("judgeTokens"),
|
||
}
|
||
summary["accepted"] = accepted
|
||
summary["ok"] = accepted if blk else prefilter
|
||
# 阻断路且未过:degraded 走可重试类失败因（result_out 缺省 llm_error 即可重试；不覆盖已有 quota_exhausted）。
|
||
if blk and not accepted and floor.get("degraded"):
|
||
summary.setdefault("failureReason", "llm_error")
|
||
# 盘上证据链与最终验收结论对齐(judge.json 由 judge_gameplay_floor 已落;这里补 verdict.json 增量段)。
|
||
_writeback_verdict_json(ev, accepted, summary["judge"])
|
||
return summary
|
||
except Exception as e: # noqa: BLE001 判定接线级异常:fail-closed（blocking→不放行）,observe 模式保预筛。
|
||
blk = True if blocking is None else bool(blocking)
|
||
try:
|
||
blk = _judge_cfg()["blocking"] if blocking is None else bool(blocking)
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
prefilter = bool(summary.get("ok")) if prefilter is None else bool(prefilter)
|
||
summary["judge"] = {"ran": False, "blocking": blk, "degraded": True,
|
||
"reason": f"判定接线异常（fail-closed）：{type(e).__name__}: {e}"}
|
||
summary["accepted"] = False if blk else prefilter
|
||
summary["ok"] = False if blk else prefilter
|
||
try: # 接线异常路也尽力对齐盘上证据(ev 推导自身可能失败,再兜一层)
|
||
_writeback_verdict_json(
|
||
evidence_dir if evidence_dir is not None else _wg1_evidence_dir(game_id),
|
||
summary["accepted"], summary["judge"])
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
return summary
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 便捷自跑：python cheap-worker/cheap_verify.py <gameId> ["brief"] —— 对已有产物真跑一次评分。
|
||
import json
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
gid = sys.argv[1] if len(sys.argv) > 1 else "cheap-smoke1"
|
||
bf = sys.argv[2] if len(sys.argv) > 2 else ""
|
||
out = asyncio.run(verify_richness(gid, brief=bf))
|
||
print(json.dumps(out, ensure_ascii=False, indent=2))
|