框架: P2放量驱动clean_batch+P3拆书M3直调parse_llm固化(提示词自workflow移植,ingest守卫复用)

.claude/skills/clean/scripts/clean_batch.py

@@ -0,0 +1,81 @@
+#!/usr/bin/env python3
+"""clean skill：放量驱动——每书串行跑 prep→detect→apply 全链，可多进程分书并行。
+
+断点续跑设计：
+- prep 仅在该书 manifest 缺失时执行（防覆盖已切窗）；
+- detect 天然跳过已有 deletions-*.json 的窗；
+- apply 前查 example_clean_log 是否已有该批次（幂等防重删），且要求窗产物齐备。
+单书失败不阻断后续书；重跑本脚本自动补缺。
+"""
+import json
+import pathlib
+import subprocess
+import sys
+
+import click
+import psycopg
+
+DSN = ("postgresql://root:f6710e2d0294eb1c10e26a805a64bc54@100.64.0.8:5433/muse-example"
+       "?keepalives=1&keepalives_idle=15&keepalives_interval=5&keepalives_count=3")
+TENANT = 1
+OUT = pathlib.Path("/tmp/muse-clean")
+HERE = pathlib.Path(__file__).resolve().parent
+PY = str(pathlib.Path(sys.executable))
+
+
+def run(args):
+    """子进程跑 skill 脚本，回显输出；返回是否成功。"""
+    r = subprocess.run([PY, *args], capture_output=True, text=True)
+    for ln in (r.stdout + r.stderr).strip().splitlines():
+        click.echo(f"    {ln}")
+    return r.returncode == 0
+
+
+@click.command()
+@click.option("--work-id", "work_ids", type=int, multiple=True, required=True)
+@click.option("--batch", required=True, help="放量批次号（幂等判断依据，重跑请保持一致）")
+@click.option("--report-dir", default="docs", show_default=True)
+def main(work_ids, batch, report_dir):
+    with psycopg.connect(DSN) as conn:
+        titles = dict(conn.execute(
+            "SELECT id, title FROM muse_content_work WHERE tenant_id=%s AND deleted=FALSE AND id=ANY(%s)",
+            (TENANT, list(work_ids))).fetchall())
+    for wid in work_ids:
+        title = titles.get(wid, f"id{wid}")
+        click.echo(f"═══《{title}》(work={wid}) ═══")
+        d = OUT / str(wid)
+        # 1) 切窗（manifest 已存在则跳过——防覆盖已探测批次）
+        if not (d / "manifest.json").exists():
+            if not run([str(HERE / "clean_prep.py"), "--work-id", str(wid)]):
+                click.echo(f"  [失败] prep，跳过本书")
+                continue
+        manifest = json.loads((d / "manifest.json").read_text())
+        # 2) 探测（detect 内部跳过已有产物窗；单窗失败中断，本书留待重跑补齐）
+        run([str(HERE / "clean_detect.py"), "--work-id", str(wid)])
+        missing = [m["win"] for m in manifest["windows"]
+                   if not (d / f"deletions-{m['win']:03d}.json").exists()]
+        if missing:
+            click.echo(f"  [缺窗] {missing}，本书暂不 apply（重跑本命令自动补）")
+            continue
+        # 3) 幂等防重删：该批次已落审计则跳过 apply
+        with psycopg.connect(DSN) as conn:
+            done = conn.execute(
+                "SELECT 1 FROM example_clean_log WHERE work_id=%s AND batch=%s LIMIT 1",
+                (wid, batch)).fetchone()
+        if done:
+            click.echo(f"  [跳过] 批次 {batch} 已落库")
+            continue
+        # 4) 合并全窗建议 → 真删落库 + 质检报告
+        alls = []
+        for m in manifest["windows"]:
+            alls += json.loads((d / f"deletions-{m['win']:03d}.json").read_text())["deletions"]
+        (d / "deletions-all.json").write_text(
+            json.dumps({"deletions": alls}, ensure_ascii=False, indent=1))
+        report = pathlib.Path(report_dir) / f"清洗-{wid}-{title}.md"
+        ok = run([str(HERE / "clean_apply.py"), "--work-id", str(wid), "--batch", batch,
+                  "--file", str(d / "deletions-all.json"), "--report-md", str(report)])
+        click.echo(f"  {'✓ 完成' if ok else '[失败] apply'}《{title}》建议 {len(alls)} 条，报告 {report}")
+
+
+if __name__ == "__main__":
+    main()

.claude/skills/parse-book/SKILL.md

@@ -31,6 +31,15 @@ disable-model-invocation: true
 - **顺序性只来自增量判重**(新实体要对着已积累实体判重合并),细纲逆推本身章间独立——先顺序跑保正确,并行化留作后续优化;
 - 进度每 10 章报一行(章号/新实体数/累计分型统计)。
 
+**M3 直调形态（创始人 2026-07-13 拍板,现行）**：循环体不再派 opus/haiku 子代理,改为 `scripts/parse_llm.py` 直调 New-API `MiniMax-M3`（经 llm skill）——脚本自己取数（正文/前文实体/已积累卡名录内联进 prompt）→ M3 两 pass（脚手架→范式）→ `parse_ingest` 机械校验入库（比例硬顶/归型/字段越合同/15连字泄漏）。断点续跑按 example_parse_task 状态机逐章跳过已 done 的 pass。
+
+```bash
+# 逐章两 pass 拆一本书的指定章区（断点续跑,重跑自动补失败章）
+.venv/bin/python .claude/skills/parse-book/scripts/parse_llm.py --work-id 4 --from 1 --to 3
+# 进度
+.venv/bin/python .claude/skills/parse-book/scripts/parse_ingest.py progress
+```
+
 ## 步骤(自底向上,与创作期规划的自顶向下互为镜像)
 
 1. 静态分章:import skill(规则,LLM 不参与);

.claude/skills/parse-book/scripts/parse_llm.py

@@ -0,0 +1,210 @@
+#!/usr/bin/env python3
+"""parse-book skill：M3 直调拆书执行器——逐章两 pass（脚手架→范式），入库走 parse_ingest 守卫。
+
+创始人拍板（2026-07-13）：拆书内容生产 LLM=New-API MiniMax-M3（经 llm skill），
+不再派 opus/haiku 子代理。本脚本把 workflow 版的提示词资产（身份段/实体判据/五型合同）
+固化为直调版：脚本自己取数（正文/前文实体/已积累卡名录内联进 prompt）、调 M3、
+容错解析 JSON、经 parse_ingest 机械校验入库（比例硬顶/归型/字段越合同/15连字泄漏全在那边）。
+
+断点续跑：按 example_parse_task 状态机逐章跳过已 done 的 pass；失败记录不阻断后续章。
+"""
+import json
+import pathlib
+import subprocess
+import sys
+
+import click
+import psycopg
+
+# 统一走 llm skill 入口（trust_env/重试/<think>剥离/JSON 容错都在那边）
+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 = 1
+HERE = pathlib.Path(__file__).resolve().parent
+TMP = pathlib.Path("/tmp/muse-parse")
+
+# ── 提示词资产（自 workflow 试拆版移植，与 A3 库内合同同源） ──────────────
+
+IDENTITY = """你是知识抽取员(extractor)，分析槽位的默认绑定件。产出全部是草稿。
+元数据纪律：schema 有什么字段你就抽什么，schema 没有的不抽——字段合同就是抽取 checklist，不自造结构；归型走各型「判据」；归不进任何型的候选=枚举缺口，如实报不硬塞；每字段要有正文证据，置信度低标「?」。
+通则：以正文为准，不脑补正文没写的；基础字段规范填；采纳正文≠确认知识。"""
+
+ENTITY_CRITERIA = ("实体型判据(脚手架级,只要 型/名称/一句话摘要)：character=具名可指认的行动主体；"
+                   "location=有名字的地点/星球/设施；faction=组织/国家/军团/公司；"
+                   "power_system=力量体系/科技体系/修炼阶梯(体系本身,非招式)；"
+                   "item=有名字且有跨章戏份的装备/机甲/物品；event=已发生的重大事件(战役/事故/仪式)。"
+                   "立卡门槛：有跨章戏份潜力；一次性龙套与单场景道具不收。")
+
+CONTRACTS = {
+    "combat": {"中文名": "打斗桥段", "判据": "以武力/超自然力分胜负的对抗场景范式 ‖ 非武力博弈(商战/权谋/斗嘴)→scene_pattern;情绪弧主导→emotion", "字段": [
+        {"key": "对抗类型", "说明": "斗法/斗宝/阵战/围杀/车轮战/追杀"}, {"key": "开局态势", "说明": "强弱差及其理由——没有落差就没有戏"}, {"key": "节拍结构", "说明": "回合推进——试探/僵持/变数/分胜负各占多少"}, {"key": "反转机制", "说明": "靠什么翻盘——法宝/援军/情报/环境/代价兑换"}, {"key": "代价结构", "说明": "赢者付出什么——白赢是范式的失败"}, {"key": "败者处理", "说明": "死/逃/收服/结仇"}]},
+    "craft": {"中文名": "叙事技法", "判据": "单点装置——删去它场景仍成立、读者体验变平 ‖ 承载整场戏→桥段三型;跨章公式→trope", "字段": [
+        {"key": "装置类型", "说明": "伏笔/契诃夫之枪/信息差/重复意象/倒计时/身份错认"}, {"key": "埋设手法", "说明": "如何放得自然——挂在什么载体上不显眼又可回指"}, {"key": "履约方式", "说明": "何时、以什么形式兑现;兑现时如何唤起读者记忆"}, {"key": "间隔纪律", "说明": "埋与收的典型距离;太近廉价、太远失效"}, {"key": "失败模式", "说明": "这个装置最常见的用砸方式"}]},
+    "emotion": {"中文名": "情感桥段", "判据": "以情绪弧为主体的场景(告白/离别/爆发) ‖ 武力分胜负→combat;关系实体本身→character_relation", "字段": [
+        {"key": "情绪类型", "说明": "愤怒/悲怆/温情/羞耻/释然/敬畏"}, {"key": "铺垫结构", "说明": "情绪蓄水的台阶——哪几步把水位抬起来"}, {"key": "爆发点设计", "说明": "临界事件与引爆台词/动作的形态"}, {"key": "收束方式", "说明": "爆发后如何落地——转场/留白/反差"}, {"key": "常见失误", "说明": "这类情绪戏最常写砸的地方"}]},
+    "scene_pattern": {"中文名": "通用桥段", "判据": "承载一场戏的通用场景公式(非打斗非情感主导):拍卖/比试/谈判/夜袭/审讯/宴会 ‖ 武力对抗→combat;情绪弧→emotion", "字段": [
+        {"key": "场景类型", "说明": "拍卖/比试/谈判/夜袭/审讯/宴会/交易"}, {"key": "参与结构", "说明": "几方角色与各自诉求——冲突的来源"}, {"key": "推进节拍", "说明": "这场戏的标准起承转合"}, {"key": "变数设计", "说明": "打破常规走向的手段"}, {"key": "出口设计", "说明": "各方如何离场,留下什么后续"}]},
+    "trope": {"中文名": "套路", "判据": "跨章/跨作品复用的情节公式(扮猪吃虎/废柴逆袭/打脸循环),有可枚举的步骤链 ‖ 单场景装置→craft;单场戏→桥段三型", "字段": [
+        {"key": "公式步骤", "说明": "可枚举的步骤链——每步做什么"}, {"key": "适用条件", "说明": "什么设定/关系下这个套路才成立"}, {"key": "变体谱系", "说明": "常见变形与升级版"}, {"key": "失效风险", "说明": "读者疲劳点;用几次会腻"}, {"key": "组合搭配", "说明": "常与哪些套路/桥段连用"}]},
+}
+
+
+def scaffold_prompt(title, ch, ch_title, text, prev_entities):
+    ents = json.dumps(prev_entities, ensure_ascii=False) if prev_entities else "（第一章，为空）"
+    return f"""【muse 创作实验台·拆书 2b·脚手架 pass】
+{IDENTITY}
+
+【功能指令(parse-book 逐章内环)】
+对参考书《{title}》第 {ch} 章《{ch_title}》（正文附后）：
+1) 逆推本章细纲：章目标/关键事件/出场角色/伏笔动作(埋·推·收)/章末钩子。字数=章正文的 3–5%（3000字章→100–150字），硬顶 8%，超标会被校验脚本退回。细纲是结构骨架，不是缩写复述。
+2) 抽实体增量（脚手架级索引）：{ENTITY_CRITERIA}
+判重：下方是前文已收录实体，不重报（除非本章给出新身份，则在一句话摘要里并入）。
+
+【前文实体索引】
+{ents}
+
+【输出规则（只输出一个 JSON 对象，禁止任何其他文字）】
+{{"outline": "细纲文本", "entities": [{{"type": "六型之一", "name": "名称", "brief": "一句话摘要"}}]}}
+
+【本章正文】
+{text}"""
+
+
+def patterns_prompt(title, ch, ch_title, text, outline, existing_cards):
+    cards = json.dumps(existing_cards, ensure_ascii=False) if existing_cards else "（尚无）"
+    return f"""【muse 创作实验台·拆书 2b·范式 pass】
+{IDENTITY}
+
+【功能指令(parse-book 范式拆取+脱敏红线)】
+对参考书《{title}》第 {ch} 章《{ch_title}》（正文附后），在脚手架细纲之上拆「范式五型」卡：
+{json.dumps(CONTRACTS, ensure_ascii=False)}
+纪律：
+- 宁缺毋滥：一章 0–3 张为常态，只收本章表现突出、可跨书复用的写法；平庸章可出 0 张。
+- 归型走判据；五型之外一律不出卡。
+- **脱敏红线**：只写抽象结构与手法归纳，严禁抄录原文；≥15 连续字与原文重合=校验脚本机械拒卡。卡名与字段值用「主角/强敌/导师」等抽象指代，**不得出现书内专名**（人名/地名/机甲名/组织名）——专名只允许出现在 source.anchor 里。
+- fields 按该型字段合同的中文 key 填；没证据的 key 省略，不编造。
+- 判重：下方已积累卡名录，不与之重复立同义卡。
+
+【本章细纲(脚手架 pass 产出)】
+{outline}
+
+【已积累范式卡名录】
+{cards}
+
+【输出规则（只输出一个 JSON 对象，禁止任何其他文字；没有值得立的卡时 cards 给空数组）】
+{{"cards": [{{"type": "craft|combat|emotion|scene_pattern|trope", "name": "抽象手法名(无书内专名)", "brief": "一句话摘要", "fields": {{"中文合同key": "值"}}, "source": {{"book": "{title}", "chapter": "第{ch}章 {ch_title}", "anchor": "一句话情节定位(抽象指代)"}}}}]}}
+
+【本章正文】
+{text}"""
+
+
+def m3_json(prompt, model, need_keys):
+    """调 M3 → 容错提取 JSON → 形状校验；不合格带错误提示重试 1 次。"""
+    content, usage = chat(prompt, model=model)
+    for retry in range(2):
+        try:
+            data = extract_json(content)
+            if all(k in data for k in need_keys):
+                return data, usage
+            err = f"缺少必需键 {need_keys}"
+        except Exception as e:  # json_repair 也救不回来的输出
+            err = str(e)[:200]
+        if retry == 0:
+            content, u2 = chat(prompt + f"\n\n【重试提示】上次输出无法解析（{err}），请严格按输出规则只输出一个 JSON 对象。",
+                               model=model)
+            usage = {k: usage.get(k, 0) + u2.get(k, 0) for k in set(usage) | set(u2)}
+    raise RuntimeError(f"JSON 形状重试仍失败: {err}")
+
+
+def ingest(kind, work_id, ch, payload):
+    """写临时文件 → parse_ingest 机械校验入库；返回 (是否成功, 输出文本)。"""
+    TMP.mkdir(parents=True, exist_ok=True)
+    f = TMP / f"{work_id}-{ch}-{kind}.json"
+    f.write_text(json.dumps(payload, ensure_ascii=False, indent=1))
+    r = subprocess.run([sys.executable, str(HERE / "parse_ingest.py"), kind,
+                        "--work-id", str(work_id), "--chapter-order", str(ch), "--file", str(f)],
+                       capture_output=True, text=True)
+    return r.returncode == 0, (r.stdout + r.stderr).strip()
+
+
+@click.command()
+@click.option("--work-id", type=int, required=True)
+@click.option("--from", "from_", type=int, required=True)
+@click.option("--to", type=int, required=True)
+@click.option("--model", default="MiniMax-M3", show_default=True)
+def main(work_id, from_, to, model):
+    # 任务行就绪（幂等）
+    subprocess.run([sys.executable, str(HERE / "parse_ingest.py"), "init-tasks",
+                    "--work-id", str(work_id), "--from", str(from_), "--to", str(to)],
+                   capture_output=True, text=True)
+    with psycopg.connect(DSN) as conn:
+        title = conn.execute("SELECT title FROM muse_content_work WHERE id=%s", (work_id,)).fetchone()[0]
+    total_in = total_out = 0
+    for ch in range(from_, to + 1):
+        with psycopg.connect(DSN) as conn:
+            row = conn.execute(
+                """SELECT c.id, c.title, b.content_text, t.scaffold_status, t.pattern_status
+                   FROM muse_content_chapter c
+                   JOIN muse_content_block b ON b.chapter_id=c.id AND b.deleted=FALSE
+                   JOIN example_parse_task t ON t.chapter_id=c.id AND t.tenant_id=c.tenant_id
+                   WHERE c.tenant_id=%s AND c.work_id=%s AND c.order_no=%s AND c.deleted=FALSE""",
+                (TENANT, work_id, ch)).fetchone()
+            if not row:
+                click.echo(f"#{ch} 章或任务行不存在，跳过")
+                continue
+            ch_id, ch_title, text, s_st, p_st = row
+            # 前文实体索引（紧凑：型/名称/摘要），判重用
+            prev = [e for (ents,) in conn.execute(
+                """SELECT s.entities FROM example_parse_scaffold s
+                   JOIN muse_content_chapter c ON c.id=s.chapter_id
+                   WHERE s.tenant_id=%s AND s.work_id=%s AND c.order_no<%s AND s.deleted=FALSE""",
+                (TENANT, work_id, ch)).fetchall() for e in ents]
+            cards = conn.execute(
+                """SELECT draft_payload->>'型', draft_payload->>'名称' FROM muse_knowledge_draft
+                   WHERE tenant_id=%s AND source_type='parse_book' AND deleted=FALSE""",
+                (TENANT,)).fetchall()
+            outline = None
+            if s_st == "done":
+                r = conn.execute(
+                    "SELECT outline_text FROM example_parse_scaffold WHERE tenant_id=%s AND chapter_id=%s AND deleted=FALSE",
+                    (TENANT, ch_id)).fetchone()
+                outline = r[0] if r else None
+
+        # pass1 脚手架（断点续跑：done 跳过）
+        if s_st != "done" or outline is None:
+            try:
+                data, usage = m3_json(scaffold_prompt(title, ch, ch_title, text, prev), model,
+                                      ("outline", "entities"))
+                total_in += usage.get("prompt_tokens", 0)
+                total_out += usage.get("completion_tokens", 0)
+                ok, out = ingest("scaffold", work_id, ch, data)
+                click.echo(f"  {out}")
+                if not ok:
+                    continue  # 比例超标等已记 task failed，重跑本命令补
+                outline = data["outline"]
+            except RuntimeError as e:
+                click.echo(f"  #{ch} 脚手架 M3 失败: {e}")
+                continue
+        # pass2 范式（done 跳过；卡被守卫拒属正常，不算失败）
+        if p_st != "done":
+            try:
+                data, usage = m3_json(patterns_prompt(title, ch, ch_title, text, outline, cards), model,
+                                      ("cards",))
+                total_in += usage.get("prompt_tokens", 0)
+                total_out += usage.get("completion_tokens", 0)
+                _, out = ingest("patterns", work_id, ch, data)
+                click.echo(f"  {out}")
+            except RuntimeError as e:
+                click.echo(f"  #{ch} 范式 M3 失败: {e}")
+    click.echo(f"《{title}》{from_}–{to} 章完成；token in={total_in:,} out={total_out:,}")
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except (psycopg.Error, RuntimeError) as e:
+        click.echo(f"[错误] {type(e).__name__}: {e}", err=True)
+        sys.exit(1)