236 lines
16 KiB
Python
236 lines
16 KiB
Python
#!/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 re
|
||
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 "（第一章，为空）"
|
||
# 绝对字数上限比抽象比例对 LLM 更可执行（首轮实测 M3 按比例会写到 9.8%–19%）
|
||
wc = len(text.replace("\n", "").replace(" ", ""))
|
||
cap = max(60, int(wc * 0.05))
|
||
return f"""【muse 创作实验台·拆书 2b·脚手架 pass】
|
||
{IDENTITY}
|
||
|
||
【功能指令(parse-book 逐章内环)】
|
||
对参考书《{title}》第 {ch} 章《{ch_title}》（正文附后，约 {wc} 字）：
|
||
1) 逆推本章细纲：章目标/关键事件/出场角色/伏笔动作(埋·推·收)/章末钩子。**细纲全文不得超过 {cap} 字**（章正文的 5%；超标会被校验脚本机械退回）。细纲是结构骨架，不是缩写复述——用短语与分号，不写完整句子。
|
||
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 张。
|
||
- 归型走判据；五型之外一律不出卡。**先按判据定 type，再只用该 type 字段表里的中文 key 填 fields——混用他型字段=机械拒卡**（如 trope 的「公式步骤」不得出现在 craft 卡里）。
|
||
- **五型都是候选，不要把一切归成 craft**：整场武力对抗→combat；整场情绪戏→emotion；拍卖/谈判/审讯等场景公式→scene_pattern；跨章复用的情节公式→trope；craft 只留给「单点装置」（删去它场景仍成立）。type 值必须与 fields 所用字段表同型。
|
||
- **脱敏红线**：只写抽象结构与手法归纳，严禁抄录原文；≥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]
|
||
# 判重名录排除本章自己的卡：否则重跑时 M3 对着自己首轮的卡判重不出卡，
|
||
# ingest 幂等软删旧卡后插 0 张——卡被自己删了（重跑自噬,实测病）
|
||
cards = conn.execute(
|
||
"""SELECT draft_payload->>'型', draft_payload->>'名称' FROM muse_knowledge_draft
|
||
WHERE tenant_id=%s AND source_type='parse_book' AND deleted=FALSE
|
||
AND NOT (source_id=%s AND (draft_payload->>'章序')::int=%s)""",
|
||
(TENANT, work_id, ch)).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:
|
||
p1 = scaffold_prompt(title, ch, ch_title, text, prev)
|
||
data, usage = m3_json(p1, 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)
|
||
# 比例超标是 M3 高频病：带压缩指令重试 1 次（首轮实测 10/15 章超标）
|
||
if not ok and "细纲比例" in out:
|
||
cap = max(60, int(len(re.sub(r"\s", "", text)) * 0.05))
|
||
data, usage = m3_json(
|
||
p1 + f"\n\n【重试】上次细纲 {len(data['outline'])} 字超标被退回。"
|
||
f"压缩到 {cap} 字以内：只留章目标/关键事件/伏笔动作/钩子，删掉一切修饰与过程描述。",
|
||
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)
|
||
# source 机械回填：book/chapter 是调用侧确定数据，不依赖 LLM 自觉
|
||
#（实测 M3 偶发 source=null 整批被拒）；anchor 是内容性字段，缺时用 brief 兜底
|
||
for c in data.get("cards") or []:
|
||
src = c.get("source") or {}
|
||
c["source"] = {"book": src.get("book") or title,
|
||
"chapter": src.get("chapter") or f"第{ch}章 {ch_title}",
|
||
"anchor": src.get("anchor") or c.get("brief") or ""}
|
||
_, 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)
|