feat: integrate reasoning table into story_system_engine build pipeline

webnovel-writer/scripts/data_modules/story_system_engine.py

@@ -43,12 +43,21 @@ class StorySystemEngine:
             genre=route["genre_filter"],
             top_k=2,
         )
-        source_trace = route["source_trace"] + self._build_source_trace(base_context, dynamic_context)
-        anti_patterns = merge_anti_patterns(
+
+        # Reasoning layer
+        primary_genre = str(route.get("meta", {}).get("primary_genre", "") or genre or "").strip()
+        reasoning = self._load_reasoning(primary_genre)
+        ranked = self._apply_reasoning(reasoning, base_context, dynamic_context)
+
+        source_trace = route["source_trace"] + self._build_source_trace_with_reasoning(ranked, reasoning)
+
+        raw_anti = merge_anti_patterns(
             route["route_anti_patterns"],
             self._extract_anti_patterns(base_context),
             self._extract_anti_patterns(dynamic_context),
         )
+        anti_patterns = self._rank_anti_patterns(reasoning, raw_anti)
+
         return {
             "meta": {"query": query, "chapter": chapter, "explicit_genre": genre or ""},
             "master_setting": {
@@ -63,7 +72,7 @@ class StorySystemEngine:
                     "core_tone": route["core_tone"],
                     "pacing_strategy": route["pacing_strategy"],
                 },
-                "base_context": base_context,
+                "base_context": [r for r in ranked if r.get("_priority_rank", 999) < 999],
                 "source_trace": source_trace,
                 "override_policy": {
                     "locked": ["route.primary_genre", "master_constraints.core_tone"],
@@ -82,8 +91,18 @@ class StorySystemEngine:
                     "override_allowed": {
                         "chapter_focus": self._suggest_chapter_focus(query, dynamic_context),
                     },
-                    "dynamic_context": dynamic_context,
+                    "dynamic_context": ranked,
                     "source_trace": source_trace,
+                    "reasoning": (
+                        {
+                            "genre": reasoning.get("题材", ""),
+                            "inject_target": reasoning.get("contract注入层", ""),
+                            "style_priority": reasoning.get("风格优先级", ""),
+                            "pacing_strategy": reasoning.get("节奏默认策略", ""),
+                        }
+                        if reasoning
+                        else {}
+                    ),
                 }
                 if chapter is not None
                 else None
@@ -233,6 +252,110 @@ class StorySystemEngine:
             for text in self._split_multi_value(row.get("毒点"))
         ]
 
+    # ------------------------------------------------------------------
+    # Reasoning / 裁决 layer
+    # ------------------------------------------------------------------
+
+    def _load_reasoning(self, genre: str) -> Dict[str, Any]:
+        """Load matching row from 裁决规则.csv for *genre*."""
+        rows = self._load_csv_rows("裁决规则")
+        genre_norm = self._normalize_text(genre)
+        if not genre_norm:
+            return {}
+        for row in rows:
+            if self._normalize_text(row.get("题材")) == genre_norm:
+                return row
+            aliases = (
+                self._split_multi_value(row.get("关键词"))
+                + self._split_multi_value(row.get("意图与同义词"))
+            )
+            if any(genre_norm == self._normalize_text(a) for a in aliases):
+                return row
+        return {}
+
+    def _apply_reasoning(
+        self,
+        reasoning: Dict[str, Any],
+        base_context: List[Dict[str, Any]],
+        dynamic_context: List[Dict[str, Any]],
+    ) -> List[Dict[str, Any]]:
+        """Rank *base_context* + *dynamic_context* rows using 冲突裁决 priority."""
+        combined = [dict(r) for r in base_context] + [dict(r) for r in dynamic_context]
+        if not reasoning:
+            return combined
+
+        priority_order = [
+            s.strip()
+            for s in str(reasoning.get("冲突裁决") or "").split(">")
+            if s.strip()
+        ]
+        priority_map = {name: idx for idx, name in enumerate(priority_order)}
+
+        genre_label = reasoning.get("题材", "")
+        for row in combined:
+            table = str(row.get("_table") or "")
+            row["_priority_rank"] = priority_map.get(table, 999)
+            row["_reasoning_rule"] = genre_label
+
+        combined.sort(key=lambda r: r["_priority_rank"])
+        return combined
+
+    def _rank_anti_patterns(
+        self,
+        reasoning: Dict[str, Any],
+        anti_patterns: List[Dict[str, Any]],
+    ) -> List[Dict[str, Any]]:
+        """Sort *anti_patterns* by 毒点权重 and append reasoning 反模式."""
+        if not reasoning:
+            return anti_patterns
+
+        weight_order = [
+            s.strip()
+            for s in str(reasoning.get("毒点权重") or "").split(">")
+            if s.strip()
+        ]
+
+        def _sort_key(item: Dict[str, Any]) -> int:
+            text = str(item.get("text") or "")
+            for idx, keyword in enumerate(weight_order):
+                if keyword in text:
+                    return idx
+            return len(weight_order)
+
+        sorted_anti = sorted(anti_patterns, key=_sort_key)
+
+        # Append 反模式 entries from reasoning row
+        existing_texts = {str(a.get("text") or "") for a in sorted_anti}
+        for text in self._split_multi_value(reasoning.get("反模式")):
+            if text and text not in existing_texts:
+                sorted_anti.append(
+                    {"text": text, "source_table": "裁决规则", "source_id": reasoning.get("编号", "")}
+                )
+                existing_texts.add(text)
+
+        return sorted_anti
+
+    def _build_source_trace_with_reasoning(
+        self,
+        ranked: List[Dict[str, Any]],
+        reasoning: Dict[str, Any],
+    ) -> List[Dict[str, Any]]:
+        """Build source trace entries enriched with reasoning metadata."""
+        inject_target = reasoning.get("contract注入层", "") if reasoning else ""
+        trace: List[Dict[str, Any]] = []
+        for row in ranked:
+            trace.append(
+                {
+                    "table": row.get("_table", ""),
+                    "id": row.get("编号", ""),
+                    "summary": row.get("核心摘要", ""),
+                    "reasoning_rule": row.get("_reasoning_rule", ""),
+                    "priority_rank": row.get("_priority_rank", 999),
+                    "inject_target": inject_target,
+                }
+            )
+        return trace
+
     def _empty_route(self, query: str, genre: Optional[str]) -> Dict[str, Any]:
         fallback_genre = str(genre or "未命中题材").strip()
         route_source = "explicit_genre_fallback" if genre else "empty_csv_fallback"

webnovel-writer/scripts/data_modules/tests/test_reasoning_engine.py

@@ -0,0 +1,160 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import csv
+
+from data_modules.story_system_engine import StorySystemEngine
+
+
+def _write_csv(path, headers, rows):
+    with open(path, "w", encoding="utf-8-sig", newline="") as f:
+        writer = csv.DictWriter(f, fieldnames=headers)
+        writer.writeheader()
+        writer.writerows(rows)
+
+
+def _setup_csvs(csv_dir):
+    """Create fixture CSVs for 题材与调性推理, 裁决规则, 桥段套路, 爽点与节奏."""
+    csv_dir.mkdir(exist_ok=True)
+
+    _write_csv(
+        csv_dir / "题材与调性推理.csv",
+        [
+            "编号", "适用技能", "分类", "层级", "关键词", "意图与同义词", "适用题材",
+            "大模型指令", "核心摘要", "详细展开", "题材/流派", "题材别名", "核心调性",
+            "节奏策略", "毒点", "推荐基础检索表", "推荐动态检索表", "默认查询词",
+        ],
+        [
+            {
+                "编号": "GR-001",
+                "适用技能": "write|plan",
+                "分类": "题材路由",
+                "层级": "知识补充",
+                "关键词": "玄幻|退婚",
+                "意图与同义词": "玄幻怎么写",
+                "适用题材": "玄幻",
+                "大模型指令": "先给压抑，再给爆发兑现。",
+                "核心摘要": "玄幻退婚流需要耻辱起手和强兑现。",
+                "详细展开": "",
+                "题材/流派": "玄幻",
+                "题材别名": "退婚流",
+                "核心调性": "压抑蓄势后爆裂反击",
+                "节奏策略": "前压后爆",
+                "毒点": "打脸节奏不能缺最后一拍补刀",
+                "推荐基础检索表": "命名规则|人设与关系",
+                "推荐动态检索表": "桥段套路|爽点与节奏",
+                "默认查询词": "退婚|打脸",
+            },
+        ],
+    )
+
+    _write_csv(
+        csv_dir / "裁决规则.csv",
+        [
+            "编号", "适用技能", "分类", "层级", "关键词", "意图与同义词", "适用题材",
+            "大模型指令", "核心摘要", "详细展开", "题材", "风格优先级", "爽点优先级",
+            "节奏默认策略", "毒点权重", "冲突裁决", "contract注入层", "反模式",
+        ],
+        [
+            {
+                "编号": "RS-001",
+                "适用技能": "write|plan",
+                "分类": "裁决",
+                "层级": "推理层",
+                "关键词": "玄幻",
+                "意图与同义词": "玄幻怎么写",
+                "适用题材": "玄幻",
+                "大模型指令": "按冲突裁决排序命中条目",
+                "核心摘要": "玄幻裁决规则",
+                "详细展开": "",
+                "题材": "玄幻",
+                "风格优先级": "热血冲突 > 冷硬算计",
+                "爽点优先级": "实力碾压 > 逆境翻盘",
+                "节奏默认策略": "快推慢收",
+                "毒点权重": "圣母病 > 情绪标签化 > 逻辑断裂",
+                "冲突裁决": "爽点与节奏 > 桥段套路 > 写作技法",
+                "contract注入层": "CHAPTER_BRIEF.writing_guidance",
+                "反模式": "情绪标签化|角色行为无逻辑",
+            },
+        ],
+    )
+
+    _write_csv(
+        csv_dir / "桥段套路.csv",
+        ["编号", "适用技能", "分类", "层级", "关键词", "适用题材", "核心摘要", "桥段名称", "毒点"],
+        [
+            {
+                "编号": "TR-001",
+                "适用技能": "write",
+                "分类": "桥段",
+                "层级": "知识补充",
+                "关键词": "退婚|打脸",
+                "适用题材": "玄幻",
+                "核心摘要": "退婚现场要给足羞辱和反击空间",
+                "桥段名称": "退婚反击",
+                "毒点": "主角还没反打就被配角替他出手",
+            },
+        ],
+    )
+
+    _write_csv(
+        csv_dir / "爽点与节奏.csv",
+        ["编号", "适用技能", "分类", "层级", "关键词", "适用题材", "核心摘要", "毒点", "节奏类型"],
+        [
+            {
+                "编号": "PA-001",
+                "适用技能": "write",
+                "分类": "节奏",
+                "层级": "知识补充",
+                "关键词": "打脸|兑现",
+                "适用题材": "玄幻",
+                "核心摘要": "兑现必须补刀",
+                "毒点": "打脸收尾太软，没有读者情绪补刀",
+                "节奏类型": "爆发期",
+            },
+        ],
+    )
+
+
+def test_build_with_reasoning_includes_reasoning_rule_in_source_trace(tmp_path):
+    csv_dir = tmp_path / "csv"
+    _setup_csvs(csv_dir)
+
+    engine = StorySystemEngine(csv_dir=csv_dir)
+    contract = engine.build(query="玄幻", genre=None, chapter=1)
+
+    source_trace = contract["master_setting"]["source_trace"]
+    reasoning_entries = [e for e in source_trace if e.get("reasoning_rule") == "玄幻"]
+    assert len(reasoning_entries) > 0, f"Expected reasoning_rule='玄幻' in source_trace, got {source_trace}"
+    assert reasoning_entries[0]["inject_target"] == "CHAPTER_BRIEF.writing_guidance"
+
+
+def test_reasoning_anti_patterns_sorted_by_weight(tmp_path):
+    csv_dir = tmp_path / "csv"
+    _setup_csvs(csv_dir)
+
+    engine = StorySystemEngine(csv_dir=csv_dir)
+    contract = engine.build(query="玄幻", genre=None, chapter=None)
+
+    anti_patterns = contract["anti_patterns"]
+    assert len(anti_patterns) > 0, "Expected non-empty anti_patterns"
+
+    # Check that reasoning 反模式 entries are present
+    texts = [a["text"] for a in anti_patterns]
+    assert "情绪标签化" in texts or "角色行为无逻辑" in texts, (
+        f"Expected reasoning 反模式 in anti_patterns, got {texts}"
+    )
+
+
+def test_reasoning_not_found_falls_back_gracefully(tmp_path):
+    csv_dir = tmp_path / "csv"
+    _setup_csvs(csv_dir)
+
+    engine = StorySystemEngine(csv_dir=csv_dir)
+    # 末日 has no matching row in 裁决规则.csv fixture
+    contract = engine.build(query="末日求生", genre="末日", chapter=None)
+
+    assert contract["master_setting"] is not None
+    assert contract["anti_patterns"] is not None
+    # Should still produce a valid contract without errors
+    assert contract["meta"]["explicit_genre"] == "末日"