223 lines
13 KiB
Python
223 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""review-cards skill：公共知识卡三角色审核（拆书流程常设步骤，MiniMax-M3 执行）。
|
||
|
||
创始人拍板（2026-07-13）：审核是流程内可复用资产——放量时每批卡由 M3 扮演
|
||
番茄作家/起点作家/主编三角色审核；fable/opus 只在起量前校准本 skill 质量
|
||
（金标准见 golden/），起量后做一次总审核。
|
||
|
||
流程：三角色各一次 M3 调用（全批卡）→ 三组评分汇总 → 判定（pass/revise/reject）
|
||
→ 写回 draft_payload.审核 节（可追溯）。--calibrate 与 golden 金标准对照输出一致性。
|
||
"""
|
||
import json
|
||
import pathlib
|
||
import statistics
|
||
import sys
|
||
|
||
import click
|
||
import psycopg
|
||
|
||
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[2] / "llm" / "scripts"))
|
||
from llm import chat, extract_json # noqa: E402
|
||
|
||
DSN = ("postgresql://root:f6710e2d0294eb1c10e26a805a64bc54@100.64.0.8:5433/muse-example"
|
||
"?keepalives=1&keepalives_idle=15&keepalives_interval=5&keepalives_count=3")
|
||
TENANT, ACTOR = 1, "1"
|
||
GOLDEN = pathlib.Path(__file__).resolve().parents[1] / "golden"
|
||
|
||
# 三角色身份（与金标准评审同源）
|
||
ROLES = {
|
||
"番茄作家": "番茄小说网顶级畅销作家（多本10亿+在读爆款）。独特视角：爽点节奏、读者留存、商业性——这卡能不能帮 AI 写出让人上头的东西。",
|
||
"起点作家": "起点中文网白金作家（以结构工艺著称）。独特视角：长篇结构工艺——伏笔埋收距离、信息差维护成本、装置与主线耦合度。",
|
||
"主编": "头部网文平台资深主编（二十年审稿）。独特视角：可复制性与指导价值——这卡进知识库是资产还是噪声；最懂垃圾写作课的套路话术。",
|
||
}
|
||
|
||
# 统一判定闸（自 opus 金标准三评审蒸馏的共性红线）
|
||
CRITERIA = """审查判定要点（每张卡都过一遍）：
|
||
1. 换书测试：抹掉书名/专名/具体数字，换一个完全不同题材（都市悬疑/古装宫斗），这卡还成立、还能指导落笔吗？不成立=一次性妙笔而非范式；
|
||
2. 命名黑话检测：卡名是"作者会说的话"还是"XX式YY化ZZ"的 AI 修饰语堆叠？摘要是通用手法陈述还是原书剧情复述？
|
||
3. 标签判定：装置类型是否乱贴（信息差=必须存在角色A知道而B/读者不知道的落差；契诃夫之枪=预先可见放置+后续引爆；身份错认=必须有人把A当成B）；
|
||
4. 伪精确检测："隔两章以上"这类数字是从原文数出来的还是编的？编的=危险的 false precision；
|
||
例外：带「实测：#a埋→#b收，隔n章」标注的数字是校验脚本按实例章号机械计算的权威值，可信；
|
||
5. 同功重复：与本批其他卡是否同一手法换皮（先知降压/面瘫外化这类 family）；
|
||
6. 干货位检查：失败模式字段是否比手法本身更值钱（是→说明手法总结空洞）；
|
||
7. 专名泄漏：字段里出现原书人名/地名/数字/梗=破坏跨书复用；
|
||
豁免区：「实例」定位与「出处」里的专名是设计允许的（溯源用），不算泄漏——只审名称/摘要/字段；
|
||
8. 场景走位判定：间隔纪律/复用节奏写明「埋收同场景闭环」的，不是跨章装置而是场景内走位，
|
||
可参考性上限 2 分（反例：某"借道具看直播"卡把单场走位包装成范式——金标准 2.67 分）。"""
|
||
|
||
# 机械降档（已知弱点双保险，fable 审查回测 19 卡零误伤）：
|
||
# pass 卡的间隔/节奏字段若声明同场景闭环 → 场景走位而非跨章装置，强制降 revise
|
||
SCENE_WALK_PAT = ("同一场景", "同场景内", "一两个段落", "同一场对话", "同一场戏")
|
||
|
||
SYSTEM_TMPL = """你是{role_desc}
|
||
|
||
背景：muse 是 AI 长篇创作系统，从经典网文拆出「公共范式卡」，未来 AI 写新书时检索这些卡作写作参考。你负责质量审核：内容是否成立、AI 创作时是否可用、是否可参考（跨书复用）。"""
|
||
|
||
REVIEW_PROMPT = """{criteria}
|
||
|
||
下面是本批 {n} 张卡（带序号）。逐卡评分（1-5 整数）并给一句毒舌判词。
|
||
输出规则（只输出一个 JSON 对象；idx=卡片序号原样带回——放量后 LLM 复述卡名易漂移，序号是唯一可靠匹配键）：
|
||
{{"reviews": [{{"idx": 序号, "cheng": 成立性, "yong": 可用性, "can": 可参考性, "verdict": "一句话判词"}}], "overall": "本批系统性问题一句话"}}
|
||
|
||
【卡片】
|
||
{cards}"""
|
||
|
||
|
||
def load_cards(conn, work_id=None):
|
||
sql = """SELECT id, draft_payload FROM muse_knowledge_draft
|
||
WHERE tenant_id=%s AND source_type='parse_book' AND deleted=FALSE"""
|
||
args = [TENANT]
|
||
if work_id:
|
||
sql += " AND source_id=%s"
|
||
args.append(work_id)
|
||
return conn.execute(sql + " ORDER BY id", args).fetchall()
|
||
|
||
|
||
@click.group()
|
||
def cli():
|
||
"""公共知识卡三角色审核（M3 常设步骤）"""
|
||
|
||
|
||
@cli.command()
|
||
@click.option("--work-id", type=int, help="只审该书的卡；不给则全部")
|
||
@click.option("--batch", required=True, help="审核批次号")
|
||
@click.option("--model", default="MiniMax-M3", show_default=True)
|
||
@click.option("--dry-run", is_flag=True, help="只打印不写库")
|
||
def review(work_id, batch, model, dry_run):
|
||
"""三角色审核并写回 draft_payload.审核 节。判定：均分≥3.5 pass / ≥2.5 revise / 其余 reject。"""
|
||
with psycopg.connect(DSN) as conn:
|
||
rows = load_cards(conn, work_id)
|
||
if not rows:
|
||
click.echo("无待审卡")
|
||
return
|
||
# 分页 ≤30 卡/批：放量后单调用塞全库必爆；同批互见保 family 检出
|
||
PAGE = 30
|
||
all_reviews = {role: {} for role in ROLES}
|
||
for p0 in range(0, len(rows), PAGE):
|
||
page = rows[p0:p0 + PAGE]
|
||
cards_json = json.dumps([{"idx": p0 + i + 1, **r[1]} for i, r in enumerate(page)],
|
||
ensure_ascii=False, indent=1)
|
||
for role, desc in ROLES.items():
|
||
content, usage = chat(
|
||
REVIEW_PROMPT.format(criteria=CRITERIA, n=len(page), cards=cards_json),
|
||
model=model, system=SYSTEM_TMPL.format(role_desc=desc))
|
||
data = extract_json(content)
|
||
all_reviews[role].update(
|
||
{int(r["idx"]): r for r in data.get("reviews", []) if r.get("idx")})
|
||
click.echo(f"[{role}] 批{p0 // PAGE + 1} 评 {len(data.get('reviews', []))} 卡 "
|
||
f"(in={usage.get('prompt_tokens')}) 总评: {data.get('overall', '')[:80]}")
|
||
# 汇总判定并写库（按序号匹配）
|
||
stats = {"pass": 0, "revise": 0, "reject": 0}
|
||
for i, (cid, payload) in enumerate(rows):
|
||
name = payload.get("名称")
|
||
scores, per_role = [], {}
|
||
for role in ROLES:
|
||
r = all_reviews[role].get(i + 1)
|
||
if r:
|
||
s = [r.get("cheng", 0), r.get("yong", 0), r.get("can", 0)]
|
||
scores += s
|
||
per_role[role] = {"分": s, "判词": r.get("verdict", "")}
|
||
if not scores:
|
||
continue
|
||
avg = round(statistics.mean(scores), 2)
|
||
verdict = "pass" if avg >= 3.5 else ("revise" if avg >= 2.5 else "reject")
|
||
# 机械双保险：pass 但间隔/节奏字段声明同场景闭环 → 强制降 revise
|
||
fields_text = json.dumps(payload.get("字段") or {}, ensure_ascii=False)
|
||
if verdict == "pass" and any(p in fields_text for p in SCENE_WALK_PAT):
|
||
verdict = "revise"
|
||
stats[verdict] += 1
|
||
click.echo(f" {verdict:6s} {avg} 《{name}》")
|
||
if dry_run:
|
||
continue
|
||
payload["审核"] = {"批次": batch, "模型": model, "均分": avg,
|
||
"判定": verdict, "角色": per_role}
|
||
conn.execute("UPDATE muse_knowledge_draft SET draft_payload=%s, updater=%s WHERE id=%s",
|
||
(json.dumps(payload, ensure_ascii=False), ACTOR, cid))
|
||
if not dry_run:
|
||
conn.commit()
|
||
click.echo(f"审核完成: pass {stats['pass']} / revise {stats['revise']} / reject {stats['reject']}"
|
||
+ ("（dry-run 未写库）" if dry_run else ""))
|
||
|
||
|
||
@cli.command()
|
||
@click.option("--work-id", type=int, help="限定书")
|
||
@click.option("--out", required=True, type=click.Path(), help="输出 markdown 路径（仓库 docs/ 下，给创始人质检）")
|
||
def export(work_id, out):
|
||
"""全部活卡导出为人读样张（创始人确认门的入口物料——他看文件，不读数据库）。
|
||
每卡：审核判定/三角色判词/字段全文/实例章号/审计标记；按书分节、判定排序。"""
|
||
order = {"pass": 0, "revise": 1, "reject": 2, None: 3}
|
||
with psycopg.connect(DSN) as conn:
|
||
rows = conn.execute(
|
||
"""SELECT w.title, d.id, d.draft_payload FROM muse_knowledge_draft d
|
||
JOIN muse_content_work w ON w.id=d.source_id
|
||
WHERE d.tenant_id=%s AND d.source_type='parse_book' AND d.deleted=FALSE"""
|
||
+ (" AND d.source_id=%s" if work_id else "") + " ORDER BY w.title, d.id",
|
||
[TENANT] + ([work_id] if work_id else [])).fetchall()
|
||
books = {}
|
||
for title, did, p in rows:
|
||
books.setdefault(title, []).append((did, p))
|
||
stats = {"pass": 0, "revise": 0, "reject": 0, None: 0}
|
||
for cards in books.values():
|
||
for _, p in cards:
|
||
stats[(p.get("审核") or {}).get("判定")] += 1
|
||
lines = [f"# 公共范式卡样张（{len(rows)} 张）\n",
|
||
f"> 审核判定：**pass {stats['pass']} / revise {stats['revise']} / reject {stats['reject']}**"
|
||
+ (f"（未审 {stats[None]}）" if stats[None] else "") + "。",
|
||
"> pass=建议入公共库；revise=有料但需改；reject=不够格（多为单场景记录冒充范式）。",
|
||
"> 「实例」里的人名/专名是溯源信息（设计允许），卡身字段应无专名。\n"]
|
||
for title, cards in books.items():
|
||
cards.sort(key=lambda x: (order[(x[1].get("审核") or {}).get("判定")], -(x[1].get("审核") or {}).get("均分", 0)))
|
||
lines.append(f"\n## 《{title}》（{len(cards)} 张）\n")
|
||
for did, p in cards:
|
||
audit = p.get("审核") or {}
|
||
lines.append(f"### [{audit.get('判定', '未审')} {audit.get('均分', '')}] "
|
||
f"{p.get('名称')}（{p.get('型')}·库号{did}）\n")
|
||
lines.append(f"**摘要**：{p.get('一句话摘要')}\n")
|
||
for k, v in (p.get("字段") or {}).items():
|
||
lines.append(f"- **{k}**：{v}")
|
||
if p.get("实例"):
|
||
lines.append("- **实例**：" + "；".join(
|
||
f"第{i.get('章')}章「{i.get('定位')}」" for i in p["实例"]))
|
||
marks = [m for m in ("跨窗待证", "判重", "裁剪字段", "改型") if p.get(m)]
|
||
if marks:
|
||
lines.append("- **审计标记**：" + "；".join(
|
||
f"{m}={json.dumps(p[m], ensure_ascii=False)}" for m in marks))
|
||
for role, r in (audit.get("角色") or {}).items():
|
||
lines.append(f"- *{role}*（{'/'.join(map(str, r.get('分', [])))}）：{r.get('判词')}")
|
||
lines.append("")
|
||
pathlib.Path(out).write_text("\n".join(lines))
|
||
click.echo(f"样张已导出：{out}（{len(rows)} 卡）")
|
||
|
||
|
||
@cli.command()
|
||
@click.option("--work-id", type=int, help="限定书")
|
||
@click.option("--batch", required=True, help="要对照的审核批次号（先跑 review）")
|
||
def calibrate(work_id, batch):
|
||
"""与 golden/scores.json 金标准对照：输出逐卡偏差与整体一致性（起量前校准用）。"""
|
||
golden = json.loads((GOLDEN / "scores.json").read_text()) # {卡名: 金标准均分}
|
||
with psycopg.connect(DSN) as conn:
|
||
rows = load_cards(conn, work_id)
|
||
diffs, lines = [], []
|
||
for _, payload in rows:
|
||
name = payload.get("名称")
|
||
audit = payload.get("审核") or {}
|
||
if audit.get("批次") != batch or name not in golden:
|
||
continue
|
||
g, m = golden[name], audit["均分"]
|
||
diffs.append(m - g)
|
||
lines.append((abs(m - g), f" Δ{m - g:+.2f} M3={m} 金={g} 《{name}》"))
|
||
for _, ln in sorted(lines, reverse=True):
|
||
click.echo(ln)
|
||
if diffs:
|
||
click.echo(f"对照 {len(diffs)} 卡：平均偏差 {statistics.mean(diffs):+.2f}，"
|
||
f"平均绝对偏差 {statistics.mean(map(abs, diffs)):.2f}，最大 {max(map(abs, diffs)):.2f}")
|
||
click.echo("校准合格线（拍板前建议）：平均绝对偏差 ≤0.5 且无 ≥1.5 的单卡背离")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
cli()
|
||
except (psycopg.Error, RuntimeError) as e:
|
||
click.echo(f"[错误] {type(e).__name__}: {e}", err=True)
|
||
sys.exit(1)
|