feat(cheap): T4 生产 create 路 genre 最小路由(brief→五类确定性词表·先验序=特异性降序·无命中回退通用) (M3 cutover 放行检查单增补)

- cheap_genre_route:纯词表路由(零 LLM/零 IO/确定性),规则表数据化单源(_GENRE_RULES 顺序即先验序:
  非遗>TRPG>经营>解谜>剧情,特异词前泛词后);映射对齐 GENRE_BY_TEMPLATE 五类模板
- cheap_service_driver:取代硬编码 scaffold_template=None——命中选 per-genre 黄金骨架+genre 透传丰富度评分
  (生产 create 路与 S5 bake_off 复验同轨);无命中 (None,None) 走通用 _template(路由只加不减)
- tests/test_genre_route:五类命中/无命中回退/多类共现先验序/GENRE_BY_TEMPLATE 一致性/确定性/脏输入 14 用例

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
lili 2026-07-03 14:17:54 -07:00
parent 9e1e943503
commit d2b916f112
3 changed files with 189 additions and 3 deletions

View File

@ -0,0 +1,65 @@
"""cheap_genre_route.py — brief → genre 确定性关键词路由(生产 create 路最小品类接线)。
背景(M3 cutover plan放行检查单增补 2026-07-04):live Service 创建路 cheap_service_driver 此前硬编码
scaffold_template=None,不选品类骨架不透传 genreW-GENRE 四件套(品类 skill/骨架/few-shot/评分尺)
生产 create 路不通电, S5 复验(bake_off)是显式传品类的,复验口径与放量后生产口径不同轨本模块补最小
路由:纯词表匹配 LLM确定性( brief 恒同结果),词表映射到既有 GENRE_BY_TEMPLATE 的五类模板;
无命中回退 None(走通用 _template,与旧行为一致路由只加不减,失配面=原行为)
规则表数据化:_GENRE_RULES 是唯一规则源(顺序即先验序),改词表/调先验只动这一张表,不动逻辑
"""
from typing import Optional, Tuple
# 品类路由规则表(唯一规则源;列表顺序 = 先验序)。
# 先验序依据:按【词面特异性】降序——特异词品类在前(非遗/TRPG 的词几乎不出现在别的品类 brief 里,
# 误伤率最低),泛词品类垫后(「剧情/故事」在各品类 brief 里都常见,最易过泛匹配)。多类词共现时取
# 序前者:如「经营一家店铺,顾客的故事各不相同」同时命中经营与剧情 → 判经营(特异性高者定品类,
# 泛词只是修饰)。每条:(genre 键, scaffold 模板名, 关键词元组)——genre/模板对齐 cheap_verify.GENRE_BY_TEMPLATE。
_GENRE_RULES: Tuple[Tuple[str, str, Tuple[str, ...]], ...] = (
# ① 非遗(heritage):非遗工艺词特异性最高,与其他品类零重叠。
("heritage", "_template-feiyi", (
"非遗", "剪纸", "皮影", "刺绣", "年画", "扎染", "泥塑", "糖画", "风筝制作",
"传统工艺", "传统手艺", "民俗", "戏曲", "京剧", "昆曲", "陶艺", "木雕", "漆器",
)),
# ② TRPG:跑团/骰子/地下城词特异性高。
("trpg", "_template-trpg", (
"trpg", "跑团", "骰子", "地下城", "龙与地下城", "dnd", "角色扮演冒险", "冒险者小队",
"桌面角色扮演", "属性检定", "掷骰",
)),
# ③ 经营(sim-business):开店/客人/生意词较特异(「服务」单字面泛,组合词收窄)。
("sim-business", "_template-shop", (
"经营", "开店", "店铺", "商店", "餐厅", "饭店", "奶茶店", "咖啡店", "咖啡馆", "面馆",
"小卖部", "杂货店", "顾客", "客人", "接客", "点单", "上菜", "生意", "收银", "摆摊",
)),
# ④ 解谜(puzzle):谜题/推理词。
("puzzle", "_template-puzzle", (
"解谜", "谜题", "谜语", "推理", "密室", "华容道", "数独", "拼图", "解密", "机关",
"找不同", "找茬",
)),
# ⑤ 剧情(narrative):故事/分支词最泛,垫底(仅当没有更特异品类命中时才路由到剧情)。
("narrative", "_template-story", (
"剧情", "故事", "文字冒险", "互动小说", "视觉小说", "对话选择", "分支", "多结局",
"章节", "叙事",
)),
)
def route_genre(brief: Optional[str]) -> Tuple[Optional[str], Optional[str]]:
"""把用户 brief 确定性路由到 (scaffold_template, genre);无命中 → (None, None) 走通用模板(原行为)。
匹配口径:大小写不敏感的子串包含(brief 是自然语言短句,词表词都是低歧义中文词/固定拼写,子串够用
不引分词依赖); _GENRE_RULES 先验序扫,首个命中即定(多类词共现取先验序前者)/非字符串 brief
无命中回退纯函数 IO LLM brief 恒同结果(生产路由必须可复验)
Returns:
(scaffold_template, genre): ("_template-shop", "sim-business");无命中 (None, None)
"""
if not brief or not isinstance(brief, str):
return None, None
text = brief.lower()
for genre, template, keywords in _GENRE_RULES:
for kw in keywords:
if kw.lower() in text:
return template, genre
return None, None

View File

@ -219,8 +219,15 @@ async def drive_cheap_generation(job: dict, *, base_url: str | None = None, user
brief = job.get("brief") or "" brief = job.get("brief") or ""
t0 = time.time() t0 = time.time()
# create 路默认参(对齐现 _default_run_fn):通用 scaffold(无 per-genre 模板)、通用系统提示、write_whitelist=None。 # create 路参数:brief→genre 确定性关键词路由(T4 最小品类接线,取代硬编码 scaffold_template=None)。
scaffold_template = None # 命中五类(经营/剧情/TRPG/非遗/解谜)→ 选 per-genre 黄金骨架 + genre 透传丰富度评分(生产 create 路
# 与 S5 bake_off 复验口径同轨);无命中 → (None, None) 走通用 _template(与旧行为一致,路由只加不减)。
# write_whitelist=None(create 不收窄,对齐现 _default_run_fn)。
from cheap_genre_route import route_genre # noqa: PLC0415
scaffold_template, genre = route_genre(brief)
if scaffold_template:
print(f"[cheap-driver] game={game_id} 品类路由命中:genre={genre} template={scaffold_template}", flush=True)
scaffold_desc = None scaffold_desc = None
write_whitelist = None write_whitelist = None
# 失控兜底轮数(与三闸 150 对齐;成本上界=max_repairs=6 优雅终止,而非轮数闸;放开防单 POST 多续修被 EXCEED_MAX_ITERS 提前切断)。 # 失控兜底轮数(与三闸 150 对齐;成本上界=max_repairs=6 优雅终止,而非轮数闸;放开防单 POST 多续修被 EXCEED_MAX_ITERS 提前切断)。
@ -318,8 +325,9 @@ async def drive_cheap_generation(job: dict, *, base_url: str | None = None, user
summary = _build_summary(game_id, brief, verdict, svc, turn, t0) summary = _build_summary(game_id, brief, verdict, svc, turn, t0)
# ⑩ reply 外收口:非阻塞丰富度 LLM 评分(additive;不进 verdict、不改 ok;token 不污染生成成本台账)。 # ⑩ reply 外收口:非阻塞丰富度 LLM 评分(additive;不进 verdict、不改 ok;token 不污染生成成本台账)。
# genre 透传(T4):品类路由命中时同一次评分 additive 追加品类扩展条目(与 run_studio 的 genre 参数同义)。
try: try:
summary["richness"] = await cheap_verify.verify_richness(game_id, brief=brief) summary["richness"] = await cheap_verify.verify_richness(game_id, brief=brief, genre=genre)
rv = summary["richness"] rv = summary["richness"]
print(f"[cheap-driver] game={game_id} 丰富度 score={rv.get('score')}/{rv.get('max')} " print(f"[cheap-driver] game={game_id} 丰富度 score={rv.get('score')}/{rv.get('max')} "
f"groups={rv.get('groups')} degraded={rv.get('degraded')}", flush=True) f"groups={rv.get('groups')} degraded={rv.get('degraded')}", flush=True)

View File

@ -0,0 +1,113 @@
"""test_genre_route.py — brief→genre 确定性关键词路由单测(T4 生产 create 路品类接线)。
守的不变量:五类各自命中且映射到 GENRE_BY_TEMPLATE 登记的模板;无命中回退 (None, None)(=旧通用行为);
多类词共现取先验序(特异性高者胜);/脏输入不炸;确定性( brief 恒同结果)
:cheap-worker/.venv/bin/python -m pytest cheap-worker/tests/test_genre_route.py -v
"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # → cheap-worker/
from cheap_genre_route import route_genre # noqa: E402
from cheap_verify import GENRE_BY_TEMPLATE # noqa: E402
# ───────────────────────── 五类命中 ─────────────────────────
def test_sim_business_hits():
for brief in ("经营一家奶茶店", "开店卖包子招待顾客", "餐厅上菜小游戏", "收银台接客人"):
t, g = route_genre(brief)
assert (t, g) == ("_template-shop", "sim-business"), brief
def test_narrative_hits():
for brief in ("一个多结局的文字冒险", "互动小说,对话选择推动剧情", "分支叙事小游戏"):
t, g = route_genre(brief)
assert (t, g) == ("_template-story", "narrative"), brief
def test_trpg_hits():
for brief in ("跑团风格的冒险", "掷骰过检定的 TRPG", "地下城探险,骰子决定命运"):
t, g = route_genre(brief)
assert (t, g) == ("_template-trpg", "trpg"), brief
def test_heritage_hits():
for brief in ("剪纸主题小游戏", "体验非遗皮影戏", "做一款刺绣传统工艺游戏"):
t, g = route_genre(brief)
assert (t, g) == ("_template-feiyi", "heritage"), brief
def test_puzzle_hits():
for brief in ("华容道解谜", "一个密室逃脱谜题", "数独小游戏"):
t, g = route_genre(brief)
assert (t, g) == ("_template-puzzle", "puzzle"), brief
# ───────────────────────── 无命中回退(=旧通用行为)─────────────────────────
def test_no_hit_falls_back_to_none():
for brief in ("一个简单的点击得分小游戏", "接住掉落的水果", "打砖块"):
assert route_genre(brief) == (None, None), brief
def test_empty_and_dirty_input():
for brief in (None, "", 123, [], {}):
assert route_genre(brief) == (None, None)
# ───────────────────────── 多类共现取先验序 ─────────────────────────
def test_cooccurrence_specific_beats_generic():
"""经营词 + 剧情泛词共现 → 判经营(特异性高者定品类,泛词只是修饰;先验序=规则表序)。"""
t, g = route_genre("经营一家小店,每位顾客都有自己的故事")
assert (t, g) == ("_template-shop", "sim-business")
def test_cooccurrence_heritage_beats_business():
"""非遗词 + 经营词共现 → 判非遗(非遗特异性序最前:『经营剪纸铺』的灵魂是剪纸)。"""
t, g = route_genre("经营一家剪纸铺,传承传统手艺")
assert (t, g) == ("_template-feiyi", "heritage")
def test_cooccurrence_trpg_beats_narrative():
t, g = route_genre("跑团故事:骰子决定每章剧情走向")
assert (t, g) == ("_template-trpg", "trpg")
# ───────────────────────── 一致性 ─────────────────────────
def test_templates_registered_in_genre_by_template():
"""路由产出的每个模板都必须在 GENRE_BY_TEMPLATE 登记,且 genre 键一致(不造孤儿模板/键漂移)。"""
from cheap_genre_route import _GENRE_RULES
for genre, template, _ in _GENRE_RULES:
assert GENRE_BY_TEMPLATE.get(template) == genre, \
f"规则表 {template}{genre} 与 GENRE_BY_TEMPLATE({GENRE_BY_TEMPLATE.get(template)})漂移"
def test_deterministic():
"""确定性:同 brief 重复路由恒同结果(生产路由必须可复验)。"""
brief = "经营奶茶店的故事"
assert all(route_genre(brief) == route_genre(brief) for _ in range(5))
def test_case_insensitive():
assert route_genre("一场 DND 冒险")[1] == "trpg"
assert route_genre("一场 dnd 冒险")[1] == "trpg"
if __name__ == "__main__":
_fns = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)]
_failed = 0
for _fn in _fns:
try:
_fn()
print(f" PASS {_fn.__name__}")
except Exception as e: # noqa: BLE001
_failed += 1
print(f" FAIL {_fn.__name__}: {type(e).__name__}: {e}")
print(f"\n{len(_fns) - _failed}/{len(_fns)} passed")
sys.exit(1 if _failed else 0)