diff --git a/.claude/skills/llm/scripts/llm.py b/.claude/skills/llm/scripts/llm.py
index 5e74923..5bad570 100644
--- a/.claude/skills/llm/scripts/llm.py
+++ b/.claude/skills/llm/scripts/llm.py
@@ -21,6 +21,13 @@ TOKEN = "sk-DyVqO3lDmEvQZ3PqGpbNaaaHZHhbh0xaHRIiynhYSmVlLHl2"
DEFAULT_MODEL = "MiniMax-M3"
+class SensitiveError(Exception):
+ """上游内容安全拦截（响应体含 sensitive，如 new_sensitive 1026）。
+
+ 同模型退避重试必再触发（放量实测每敏感章空烧 3 次），故不在此退避，
+ 立即抛给上层走模型降级链（创始人 2026-07-14：M3→MiniMax-M2.7→deepseek-v4-flash）。"""
+
+
def chat(prompt, model=DEFAULT_MODEL, max_tokens=32000, temperature=0.2,
retries=2, timeout=900, system=None, top_p=None):
"""单轮对话，返回 (content, usage)。网络错/5xx/429 指数退避重试。
@@ -48,6 +55,10 @@ def chat(prompt, model=DEFAULT_MODEL, max_tokens=32000, temperature=0.2,
r = s.post(f"{BASE}/v1/chat/completions",
headers={"Authorization": f"Bearer {TOKEN}"},
json=payload, timeout=timeout)
+ # 内容安全拦截：同模型退避重试必再敏感，立即抛 SensitiveError 交上层
+ # 降级换模型，不在此浪费退避（否则一敏感章空烧 3 次，实测占放量请求 23%）
+ if r.status_code >= 500 and "sensitive" in r.text.lower():
+ raise SensitiveError(f"内容安全拦截 HTTP {r.status_code}: {r.text[:150]}")
# 429/5xx 属于可重试的服务端瞬时问题
if r.status_code in (429,) or r.status_code >= 500:
last_err = f"HTTP {r.status_code}: {r.text[:200]}"
@@ -57,8 +68,10 @@ def chat(prompt, model=DEFAULT_MODEL, max_tokens=32000, temperature=0.2,
content = data["choices"][0]["message"]["content"] or ""
content = re.sub(r".*?", "", content, flags=re.S).strip()
usage = data.get("usage", {})
+ # 缓存命中数（OpenAI 式 prompt_tokens_details.cached_tokens）——验证前缀缓存是否生效、省了多少
+ cached = (usage.get("prompt_tokens_details") or {}).get("cached_tokens", 0)
print(f"[llm] {model} in={usage.get('prompt_tokens', '?')} "
- f"out={usage.get('completion_tokens', '?')} "
+ f"cached={cached} out={usage.get('completion_tokens', '?')} "
f"耗时{time.time() - t0:.0f}s finish={data['choices'][0].get('finish_reason')}",
file=sys.stderr)
return content, usage
diff --git a/.claude/skills/parse-book/scripts/parse_llm.py b/.claude/skills/parse-book/scripts/parse_llm.py
index 95cab58..19b52a2 100644
--- a/.claude/skills/parse-book/scripts/parse_llm.py
+++ b/.claude/skills/parse-book/scripts/parse_llm.py
@@ -25,7 +25,7 @@ import psycopg
# 统一走 llm skill 入口（trust_env/重试/剥离/JSON 容错都在那边）
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[2] / "llm" / "scripts"))
-from llm import chat, extract_json # noqa: E402
+from llm import chat, extract_json, SensitiveError # noqa: E402
DSN = ("postgresql://root:f6710e2d0294eb1c10e26a805a64bc54@100.64.0.8:5433/muse-example"
"?keepalives=1&keepalives_idle=15&keepalives_interval=5&keepalives_count=3")
@@ -87,24 +87,26 @@ def render_contracts(contracts):
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))
+ cap = max(60, int(wc * 0.05)) # 绝对字数上限（纯比例约束 M3 会写超，首轮实测 9.8%–19%）
+ # 缓存友好（read-context「稳定度递减吃前缀缓存」）：固定规则段全部前置吃前缀缓存，
+ # 章号/字数/上限/前文实体/正文（每章必变）一律尾置；cap 值随材料尾置、规则段只说「见文末上限」以保持固定
return f"""【功能指令(parse-book 章级 pass：细纲+实体+范式候选线索)】
-对参考书《{title}》第 {ch} 章《{ch_title}》（正文附后，约 {wc} 字）做三件事：
-1) 逆推本章细纲：章目标/关键事件/出场角色/伏笔动作(埋·推·收)/章末钩子。**细纲全文不得超过 {cap} 字**（章正文的 5%；超标会被校验脚本机械退回）。细纲是结构骨架，不是缩写复述——用短语与分号，不写完整句子。
+对参考书的当前章（书名/章号/正文见文末材料区）做三件事：
+1) 逆推本章细纲：章目标/关键事件/出场角色/伏笔动作(埋·推·收)/章末钩子。**细纲全文不得超过文末给出的字数上限**（约本章正文的 5%；超标会被校验脚本机械退回）。细纲是结构骨架，不是缩写复述——用短语与分号，不写完整句子。
2) 抽实体增量（脚手架级索引）：{ENTITY_CRITERIA}
-判重：下方是前文已收录实体，不重报（除非本章给出新身份，则在一句话摘要里并入）。
+判重：文末给出前文已收录实体，不重报（除非本章给出新身份，则在一句话摘要里并入）。
3) 报范式候选线索（**只报线索不出卡**，出卡由窗级聚类另做）：本章表现突出、疑似可跨书复用的写法。五型候选：craft=单点叙事装置(删去它场景仍成立)；combat=整场武力对抗的打法；emotion=整场情绪戏的推进；scene_pattern=拍卖/谈判/审讯等场景公式；trope=跨章复用的情节公式。
每条线索给四项：type（五型之一，拿不准填"?"）、name（2–8 字短名，作者口头会说的话，禁书内专名禁修饰堆叠）、clue（一句话：这个写法怎么运作，用{PLACEHOLDERS}等抽象指代）、evidence（一句话：本章哪里这样写、为何突出，可用专名）。
0–5 条/章，平庸章 0 条正常；门槛比出卡低——拿不准的报上来，窗级聚类会过滤。
-【前文实体索引】
-{ents}
-
【输出规则（只输出一个 JSON 对象，禁止任何其他文字）】
{{"outline": "细纲文本", "entities": [{{"type": "六型之一", "name": "名称", "brief": "一句话摘要"}}], "hints": [{{"type": "五型之一或?", "name": "短名", "clue": "运作机制一句话", "evidence": "本章证据一句话"}}]}}
+━━━ 本章材料（每章不同，非规则）━━━
+【当前章】《{title}》第 {ch} 章《{ch_title}》｜正文约 {wc} 字｜细纲上限 {cap} 字
+【前文实体索引】
+{ents}
【本章正文】
{text}"""
@@ -112,11 +114,11 @@ def scaffold_prompt(title, ch, ch_title, text, prev_entities):
def window_cards_prompt(title, a, b, stage_outline, ch_outlines, ch_hints, contracts, batch_note):
"""窗级聚类出卡 prompt（B4-S3 核心资产；金标准七条纪律全落于此）。
放量教训：百条线索一次全型聚类=认知超载，M3 退化为逐条转写（37 章窗出 74 张卡）——
- 调用侧按型分批（batch_note 说明本批范围），每批线索几十条以内聚类才真实发生。"""
+ 调用侧按型分批（batch_note 说明本批范围），每批线索几十条以内聚类才真实发生。
+ 缓存友好（read-context「稳定度递减吃前缀缓存」）：合同+七条纪律+输出格式（上千字、
+ 跨书跨窗完全一致）全部前置吃前缀缓存；书名/章范围/本批说明/三层材料（每窗每批必变）尾置。"""
return f"""【功能指令(parse-book 窗级聚类出卡+脱敏红线)】
-《{title}》第 {a}–{b} 章已完成章级分析。下面给你三层材料：阶段大纲、逐章细纲、候选线索。
-{batch_note}
-任务：把候选线索**聚类归并**成「公共范式卡」——同一手法在多章的多次出现归并为一张母卡，带全部实例章号；线索不足处可从细纲的伏笔账（埋·推·收）补证跨章装置。
+把一个窗口内已完成章级分析的候选线索**聚类归并**成「公共范式卡」——同一手法在多章的多次出现归并为一张母卡，带全部实例章号；线索不足处可从细纲的伏笔账（埋·推·收）补证跨章装置。书名、章范围、本批范围与三层材料（阶段大纲/逐章细纲/候选线索）均见文末材料区。
{render_contracts(contracts)}
@@ -132,6 +134,12 @@ def window_cards_prompt(title, a, b, stage_outline, ch_outlines, ch_hints, contr
- fields 只允许使用该卡 type 在上方合同表里列出的中文 key，没证据的 key 省略不编造。**先按判据定 type，再对照该型的表格填字段**——把别的型才有的 key（哪怕内容再有价值）塞进本型=机械拒卡；上一批实测拒卡主因就是给非 craft 卡塞了 craft 才有的字段。
- 脱敏红线：只写抽象结构与手法归纳，严禁抄录原文（≥15 连续字与原文重合=机械拒卡）；书内专名只允许出现在实例定位(anchor)里。
+【输出规则（只输出一个 JSON 对象，禁止任何其他文字；没有值得立的卡时 cards 给空数组）】
+{{"cards": [{{"type": "craft|combat|emotion|scene_pattern|trope", "name": "短名", "brief": "一句话通用摘要", "fields": {{"中文合同key": "值"}}, "instances": [{{"ch": 章号数字, "anchor": "一句话情节定位(可用专名)"}}]}}]}}
+
+━━━ 本窗材料（每窗每批不同，非规则）━━━
+《{title}》第 {a}–{b} 章。{batch_note}
+
【阶段大纲】
{stage_outline}
@@ -139,30 +147,55 @@ def window_cards_prompt(title, a, b, stage_outline, ch_outlines, ch_hints, contr
{ch_outlines}
【逐章候选线索】
-{ch_hints}
+{ch_hints}"""
-【输出规则（只输出一个 JSON 对象，禁止任何其他文字；没有值得立的卡时 cards 给空数组）】
-{{"cards": [{{"type": "craft|combat|emotion|scene_pattern|trope", "name": "短名", "brief": "一句话通用摘要", "fields": {{"中文合同key": "值"}}, "instances": [{{"ch": 章号数字, "anchor": "一句话情节定位(可用专名)"}}]}}]}}"""
+
+# 敏感降级链（创始人 2026-07-14）：主模型撞内容安全 → 依次换这两个 → 都失败=硬停
+FALLBACK_MODELS = ["MiniMax-M2.7", "deepseek-v4-flash"]
+
+
+class SensitiveHardStop(Exception):
+ """主模型 + 两个降级备选全部撞敏感——按创始人指令硬停整个解析并汇报（不跳过、不硬扛）。"""
def m3_json(prompt, model, need_keys, system=IDENTITY):
- """调 M3（身份段走 system 角色）→ 容错提取 JSON → 形状校验；不合格带错误提示重试 1 次。"""
- content, usage = chat(prompt, model=model, system=system)
- for retry in range(2):
+ """调模型 → JSON 容错提取 → 形状校验（不合格带提示重试 1 次）。
+
+ 敏感降级（创始人 2026-07-14）：撞内容安全拦截（SensitiveError）按 [主模型]→
+ MiniMax-M2.7→deepseek-v4-flash 换模型（最多 2 次降级）；全链敏感抛 SensitiveHardStop
+ 交上层硬停。JSON 形状问题非敏感，不换模型（原样抛 RuntimeError，上层按章跳过）。"""
+ chain = [model] + FALLBACK_MODELS # 主模型 + 两个降级备选（最多两次降级）
+ for mi, m in enumerate(chain):
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, system=system)
- # 只合并数值键：上游 usage 带嵌套 dict（如 *_tokens_details），dict+dict 会崩（放量实测）
- usage = {k: usage.get(k, 0) + u2.get(k, 0) for k in set(usage) | set(u2)
- if isinstance(usage.get(k, 0), (int, float)) and isinstance(u2.get(k, 0), (int, float))}
- raise RuntimeError(f"JSON 形状重试仍失败: {err}")
+ content, usage = chat(prompt, model=m, system=system)
+ except SensitiveError as e:
+ if mi < len(chain) - 1:
+ print(f"[敏感] {m} 拒绝，降级到 {chain[mi + 1]}", file=sys.stderr)
+ continue
+ raise SensitiveHardStop(f"降级链全撞敏感（{' → '.join(chain)}）: {e}")
+ # 拿到内容：JSON 形状校验，不合格带提示重试 1 次（仍用当前模型 m）
+ for retry in range(2):
+ try:
+ data = extract_json(content)
+ if all(k in data for k in need_keys):
+ if m != model: # 用了降级模型，标注供审计
+ print(f"[敏感降级成功] 本条由 {m} 产出", file=sys.stderr)
+ return data, usage
+ err = f"缺少必需键 {need_keys}"
+ except Exception as e: # json_repair 也救不回来的输出
+ err = str(e)[:200]
+ if retry == 0:
+ try:
+ content, u2 = chat(prompt + f"\n\n【重试提示】上次输出无法解析（{err}），请严格按输出规则只输出一个 JSON 对象。",
+ model=m, system=system)
+ # 只合并数值键：上游 usage 带嵌套 dict（如 *_tokens_details），dict+dict 会崩（放量实测）
+ usage = {k: usage.get(k, 0) + u2.get(k, 0) for k in set(usage) | set(u2)
+ if isinstance(usage.get(k, 0), (int, float)) and isinstance(u2.get(k, 0), (int, float))}
+ except SensitiveError:
+ break # 重试提示触发敏感（罕见）：当作该模型失败，落到下方 RuntimeError
+ # 当前模型 JSON 形状最终失败（非敏感）：不换模型，原样抛（上层按章跳过）
+ raise RuntimeError(f"JSON 形状重试仍失败（{m}）: {err}")
+ raise RuntimeError("m3_json 降级链异常") # 理论不可达（chain 非空）
def ingest(kind, work_id, key, payload, keyflag="--chapter-order"):
@@ -234,6 +267,19 @@ def chapters(work_id, from_, to, model):
total_out += usage.get("completion_tokens", 0)
ok, out = ingest("scaffold", work_id, ch, data)
click.echo(f" {out}")
+ except SensitiveHardStop as e:
+ # 降级链（主+2 备）全撞敏感——按创始人指令硬停该书解析并汇报，不跳过、不硬扛
+ with psycopg.connect(DSN) as conn:
+ conn.execute(
+ """UPDATE example_parse_task SET scaffold_status='failed', error_message=%s
+ WHERE tenant_id=%s AND work_id=%s AND chapter_id=(
+ SELECT id FROM muse_content_chapter
+ WHERE tenant_id=%s AND work_id=%s AND order_no=%s)""",
+ (str(e)[:500], TENANT, work_id, TENANT, work_id, ch))
+ conn.commit()
+ click.echo(f" ⛔ #{ch} 敏感降级链全失败：{e}")
+ click.echo(f"《{title}》解析按指令硬停于 #{ch}（此前完成章已落库；该章需人工定夺）")
+ return
except RuntimeError as e:
click.echo(f" #{ch} 章级 M3 失败: {e}")
click.echo(f"《{title}》{from_}–{to} 章级完成；token in={total_in:,} out={total_out:,}")