407 lines
25 KiB
Python
407 lines
25 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 json
|
||
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}")
|
||
|
||
|
||
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))
|