框架: 拆书敏感降级链(M3→M2.7→deepseek全失败硬停)+提示词缓存重排(固定前置/变化尾置)+缓存命中打印

.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"<think>.*?</think>", "", 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

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

@@ -25,7 +25,7 @@ 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
+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 次。"""
+    """调模型 → JSON 容错提取 → 形状校验（不合格带提示重试 1 次）。
-    content, usage = chat(prompt, model=model, system=system)
+
-    for retry in range(2):
+    敏感降级（创始人 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)
+            content, usage = chat(prompt, model=m, system=system)
-            if all(k in data for k in need_keys):
+        except SensitiveError as e:
-                return data, usage
+            if mi < len(chain) - 1:
-            err = f"缺少必需键 {need_keys}"
+                print(f"[敏感] {m} 拒绝，降级到 {chain[mi + 1]}", file=sys.stderr)
-        except Exception as e:  # json_repair 也救不回来的输出
+                continue
-            err = str(e)[:200]
+            raise SensitiveHardStop(f"降级链全撞敏感（{' → '.join(chain)}）: {e}")
-        if retry == 0:
+        # 拿到内容：JSON 形状校验，不合格带提示重试 1 次（仍用当前模型 m）
-            content, u2 = chat(prompt + f"\n\n【重试提示】上次输出无法解析（{err}），请严格按输出规则只输出一个 JSON 对象。",
+        for retry in range(2):
-                               model=model, system=system)
+            try:
-            # 只合并数值键：上游 usage 带嵌套 dict（如 *_tokens_details），dict+dict 会崩（放量实测）
+                data = extract_json(content)
-            usage = {k: usage.get(k, 0) + u2.get(k, 0) for k in set(usage) | set(u2)
+                if all(k in data for k in need_keys):
-                     if isinstance(usage.get(k, 0), (int, float)) and isinstance(u2.get(k, 0), (int, float))}
+                    if m != model:  # 用了降级模型，标注供审计
-    raise RuntimeError(f"JSON 形状重试仍失败: {err}")
+                        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:,}")