Преглед на файлове

feat: 注入通用方法论策略卡并接入上下文链路

lingfengQAQ преди 3 месеца
родител
ревизия
a394136b5d

+ 3 - 0
.claude/scripts/data_modules/config.py

@@ -186,6 +186,9 @@ class DataModulesConfig:
     context_writing_guidance_max_items: int = 6
     context_writing_guidance_low_score_threshold: float = 75.0
     context_writing_guidance_hook_diversify: bool = True
+    context_methodology_enabled: bool = True
+    context_methodology_genre_whitelist: tuple[str, ...] = ("*",)
+    context_methodology_label: str = "digital-serial-v1"
     context_writing_checklist_enabled: bool = True
     context_writing_checklist_min_items: int = 3
     context_writing_checklist_max_items: int = 6

+ 41 - 1
.claude/scripts/data_modules/context_manager.py

@@ -23,7 +23,7 @@ from .context_weights import (
     TEMPLATE_WEIGHTS as CONTEXT_TEMPLATE_WEIGHTS,
     TEMPLATE_WEIGHTS_DYNAMIC_DEFAULT as CONTEXT_TEMPLATE_WEIGHTS_DYNAMIC_DEFAULT,
 )
-from .genre_aliases import normalize_genre_token
+from .genre_aliases import normalize_genre_token, to_profile_key
 from .genre_profile_builder import (
     build_composite_genre_hints,
     extract_genre_section,
@@ -31,6 +31,8 @@ from .genre_profile_builder import (
     parse_genre_tokens,
 )
 from .writing_guidance_builder import (
+    build_methodology_guidance_items,
+    build_methodology_strategy_card,
     build_guidance_items,
     build_writing_checklist,
     is_checklist_item_completed,
@@ -358,12 +360,23 @@ class ContextManager:
         )
 
         guidance = list(guidance_bundle.get("guidance") or [])
+        methodology_strategy: Dict[str, Any] = {}
+
+        if self._is_methodology_enabled_for_genre(genre_profile):
+            methodology_strategy = build_methodology_strategy_card(
+                chapter=chapter,
+                reader_signal=reader_signal,
+                genre_profile=genre_profile,
+                label=str(getattr(self.config, "context_methodology_label", "digital-serial-v1")),
+            )
+            guidance.extend(build_methodology_guidance_items(methodology_strategy))
 
         checklist = self._build_writing_checklist(
             chapter=chapter,
             guidance_items=guidance,
             reader_signal=reader_signal,
             genre_profile=genre_profile,
+            strategy_card=methodology_strategy,
         )
 
         checklist_score = self._compute_writing_checklist_score(
@@ -392,11 +405,13 @@ class ContextManager:
             "guidance_items": guidance[:limit],
             "checklist": checklist,
             "checklist_score": checklist_score,
+            "methodology": methodology_strategy,
             "signals_used": {
                 "has_low_score_ranges": bool(low_ranges),
                 "hook_types": hook_types,
                 "top_patterns": top_patterns,
                 "genre": genre,
+                "methodology_enabled": bool(methodology_strategy.get("enabled")),
             },
         }
 
@@ -543,6 +558,7 @@ class ContextManager:
         guidance_items: List[str],
         reader_signal: Dict[str, Any],
         genre_profile: Dict[str, Any],
+        strategy_card: Dict[str, Any] | None = None,
     ) -> List[Dict[str, Any]]:
         _ = chapter
         if not getattr(self.config, "context_writing_checklist_enabled", True):
@@ -558,11 +574,35 @@ class ContextManager:
             guidance_items=guidance_items,
             reader_signal=reader_signal,
             genre_profile=genre_profile,
+            strategy_card=strategy_card,
             min_items=min_items,
             max_items=max_items,
             default_weight=default_weight,
         )
 
+    def _is_methodology_enabled_for_genre(self, genre_profile: Dict[str, Any]) -> bool:
+        if not bool(getattr(self.config, "context_methodology_enabled", False)):
+            return False
+
+        whitelist_raw = getattr(self.config, "context_methodology_genre_whitelist", ("*",))
+        if isinstance(whitelist_raw, str):
+            whitelist_iter = [whitelist_raw]
+        else:
+            whitelist_iter = list(whitelist_raw or [])
+
+        whitelist = {str(token).strip().lower() for token in whitelist_iter if str(token).strip()}
+        if not whitelist:
+            return True
+        if "*" in whitelist or "all" in whitelist:
+            return True
+
+        genre = str((genre_profile or {}).get("genre") or "").strip()
+        if not genre:
+            return False
+
+        profile_key = to_profile_key(genre)
+        return profile_key in whitelist
+
     def _compact_json_text(self, content: Any, budget: Optional[int]) -> str:
         raw = json.dumps(content, ensure_ascii=False)
         if budget is None or len(raw) <= budget:

+ 63 - 0
.claude/scripts/data_modules/tests/test_context_manager.py

@@ -461,6 +461,69 @@ def test_context_manager_genre_aliases_normalized_for_profile_lookup(temp_projec
     assert "直播文" in (profile.get("genres") or [])
 
 
+def test_context_manager_enables_methodology_for_xianxia(temp_project):
+    state = {
+        "project": {"genre": "修仙"},
+        "protagonist_state": {"name": "韩立"},
+        "chapter_meta": {},
+        "disambiguation_warnings": [],
+        "disambiguation_pending": [],
+    }
+    temp_project.state_file.write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")
+
+    manager = ContextManager(temp_project)
+    manager.config.context_writing_checklist_max_items = 8
+    payload = manager.build_context(21, template="plot", use_snapshot=False, save_snapshot=False)
+
+    guidance = payload["sections"]["writing_guidance"]["content"]
+    strategy = guidance.get("methodology") or {}
+    assert strategy.get("enabled") is True
+    assert strategy.get("pilot") == "xianxia"
+    assert strategy.get("genre_profile_key") == "xianxia"
+    assert guidance.get("signals_used", {}).get("methodology_enabled") is True
+    assert isinstance(strategy.get("observability"), dict)
+
+
+def test_context_manager_enables_methodology_for_non_xianxia_by_default(temp_project):
+    state = {
+        "project": {"genre": "xuanhuan"},
+        "protagonist_state": {"name": "萧炎"},
+        "chapter_meta": {},
+        "disambiguation_warnings": [],
+        "disambiguation_pending": [],
+    }
+    temp_project.state_file.write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")
+
+    manager = ContextManager(temp_project)
+    payload = manager.build_context(21, template="plot", use_snapshot=False, save_snapshot=False)
+
+    guidance = payload["sections"]["writing_guidance"]["content"]
+    strategy = guidance.get("methodology") or {}
+    assert strategy.get("enabled") is True
+    assert strategy.get("genre_profile_key") == "xuanhuan"
+    assert guidance.get("signals_used", {}).get("methodology_enabled") is True
+
+
+def test_context_manager_allows_methodology_whitelist_restriction(temp_project):
+    state = {
+        "project": {"genre": "直播文"},
+        "protagonist_state": {"name": "林默"},
+        "chapter_meta": {},
+        "disambiguation_warnings": [],
+        "disambiguation_pending": [],
+    }
+    temp_project.state_file.write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")
+
+    manager = ContextManager(temp_project)
+    manager.config.context_methodology_genre_whitelist = ("xianxia",)
+    payload = manager.build_context(21, template="plot", use_snapshot=False, save_snapshot=False)
+
+    guidance = payload["sections"]["writing_guidance"]["content"]
+    strategy = guidance.get("methodology") or {}
+    assert strategy == {}
+    assert guidance.get("signals_used", {}).get("methodology_enabled") is False
+
+
 def test_context_manager_compact_text_truncation(temp_project):
     manager = ContextManager(temp_project)
     manager.config.context_compact_text_enabled = True

+ 16 - 0
.claude/scripts/data_modules/tests/test_extract_chapter_context.py

@@ -201,6 +201,19 @@ def test_render_text_contains_writing_guidance_section(tmp_path):
                 "completion_rate": 0.66,
                 "required_completion_rate": 0.75,
             },
+            "methodology": {
+                "enabled": True,
+                "framework": "digital-serial-v1",
+                "pilot": "xianxia",
+                "genre_profile_key": "xianxia",
+                "chapter_stage": "confront",
+                "observability": {
+                    "next_reason_clarity": 78.0,
+                    "anchor_effectiveness": 74.0,
+                    "rhythm_naturalness": 72.0,
+                },
+                "signals": {"risk_flags": ["pattern_overuse_watch"]},
+            },
         },
     }
 
@@ -215,6 +228,9 @@ def test_render_text_contains_writing_guidance_section(tmp_path):
     assert "### 执行评分" in text
     assert "- 评分: 81.5" in text
     assert "- 复合题材: xuanhuan + realistic" in text
+    assert "## 长篇方法论策略" in text
+    assert "- 适用题材: xianxia" in text
+    assert "next_reason=78.0" in text
 
 
 def test_render_text_contains_rag_assist_section_when_hits_exist(tmp_path):

+ 208 - 1
.claude/scripts/data_modules/writing_guidance_builder.py

@@ -26,6 +26,183 @@ GENRE_GUIDANCE_TEXT: dict[str, str] = {
 }
 
 
+GENRE_METHOD_ANCHORS: dict[str, dict[str, str]] = {
+    "xianxia": {
+        "pressure_source": "资源争夺/境界压制",
+        "release_target": "主角主动破局并拿到可见收益",
+    },
+    "urban-power": {
+        "pressure_source": "阶层卡位/权力压制",
+        "release_target": "主角通过资源博弈拿到地位与回报",
+    },
+    "romance": {
+        "pressure_source": "关系误解/情感拉扯",
+        "release_target": "关系位移落地并形成下一步承诺",
+    },
+    "mystery": {
+        "pressure_source": "线索缺失/规则冲突",
+        "release_target": "给出可验证的新线索并保留未知区",
+    },
+    "rules-mystery": {
+        "pressure_source": "规则反噬/代价递增",
+        "release_target": "用代价换突破并留下更高阶规则问题",
+    },
+    "zhihu-short": {
+        "pressure_source": "信息落差/立场对撞",
+        "release_target": "反转兑现并形成高强度尾钩",
+    },
+    "substitute": {
+        "pressure_source": "身份误读/情绪对峙",
+        "release_target": "误解链推进到明确决断",
+    },
+    "esports": {
+        "pressure_source": "战术压制/节奏失衡",
+        "release_target": "关键决策生效并转化为局势优势",
+    },
+    "livestream": {
+        "pressure_source": "舆论波动/数据下滑",
+        "release_target": "当场反制形成可见数据回弹",
+    },
+    "cosmic-horror": {
+        "pressure_source": "认知失真/规则侵蚀",
+        "release_target": "以明确代价换阶段性生存窗口",
+    },
+    "history-travel": {
+        "pressure_source": "历史惯性/礼教阻力",
+        "release_target": "知识优势兑现并引发新的连锁反应",
+    },
+    "game-lit": {
+        "pressure_source": "系统规则限制/资源稀缺",
+        "release_target": "数值突破并暴露更高层级威胁",
+    },
+}
+
+
+def build_methodology_strategy_card(
+    *,
+    chapter: int,
+    reader_signal: Dict[str, Any],
+    genre_profile: Dict[str, Any],
+    label: str = "digital-serial-v1",
+) -> Dict[str, Any]:
+    genre = str(genre_profile.get("genre") or "").strip()
+    profile_key = to_profile_key(genre) or "general"
+
+    hook_usage = reader_signal.get("hook_type_usage") or {}
+    pattern_usage = reader_signal.get("pattern_usage") or {}
+    review_trend = reader_signal.get("review_trend") or {}
+    low_ranges = reader_signal.get("low_score_ranges") or []
+
+    dominant_hook = ""
+    if isinstance(hook_usage, dict) and hook_usage:
+        dominant_hook = max(hook_usage.items(), key=lambda kv: kv[1])[0]
+
+    dominant_pattern = ""
+    if isinstance(pattern_usage, dict) and pattern_usage:
+        dominant_pattern = max(pattern_usage.items(), key=lambda kv: kv[1])[0]
+
+    overall_avg = float(review_trend.get("overall_avg") or 0.0)
+    has_low_range = bool(low_ranges)
+    hook_variety = len(hook_usage) if isinstance(hook_usage, dict) else 0
+    pattern_variety = len(pattern_usage) if isinstance(pattern_usage, dict) else 0
+
+    next_reason_clarity = 70.0 + (4.0 if has_low_range else 8.0)
+    anchor_effectiveness = 68.0 + (6.0 if dominant_hook else 0.0) + (4.0 if overall_avg >= 75 else -4.0)
+    rhythm_naturalness = 65.0 + min(10.0, float(hook_variety + pattern_variety) * 2.0)
+
+    risk_flags: List[str] = []
+    if has_low_range:
+        risk_flags.append("low_score_recency")
+    if dominant_pattern:
+        risk_flags.append("pattern_overuse_watch")
+    if overall_avg > 0 and overall_avg < 75:
+        risk_flags.append("readability_guard")
+
+    stage_mod = chapter % 5
+    if stage_mod in {1, 2}:
+        stage = "build_up"
+    elif stage_mod in {3, 4}:
+        stage = "confront"
+    else:
+        stage = "release"
+
+    anchor_preset = GENRE_METHOD_ANCHORS.get(
+        profile_key,
+        {
+            "pressure_source": "生存目标/资源竞争",
+            "release_target": "主角完成阶段目标并留下新的行动理由",
+        },
+    )
+
+    return {
+        "enabled": True,
+        "framework": label,
+        "pilot": profile_key,
+        "genre_profile_key": profile_key,
+        "chapter_stage": stage,
+        "emotion_anchor": {
+            "pressure_source": anchor_preset["pressure_source"],
+            "release_target": anchor_preset["release_target"],
+            "position_hint": "前段设压,中后段释放,避免固定字位打点",
+        },
+        "long_arc_controls": {
+            "map_transition": "阶段切换承接既有资产与关系账本,避免能力与收益归零",
+            "power_guard": "关键胜利必须给机制理由(信息/资源/代价/策略)",
+            "antagonist_model": "反派需具备目标-手段-代价三要素,避免工具人推进",
+        },
+        "serialization_ops": {
+            "next_reason": "章末或后段给出可复述的下一章动机句",
+            "interaction_note": "保留一个可讨论分歧点,便于连载互动反馈",
+        },
+        "observability": {
+            "next_reason_clarity": round(max(0.0, min(100.0, next_reason_clarity)), 2),
+            "anchor_effectiveness": round(max(0.0, min(100.0, anchor_effectiveness)), 2),
+            "rhythm_naturalness": round(max(0.0, min(100.0, rhythm_naturalness)), 2),
+        },
+        "signals": {
+            "dominant_hook": dominant_hook,
+            "dominant_pattern": dominant_pattern,
+            "risk_flags": risk_flags,
+        },
+    }
+
+
+def build_methodology_guidance_items(strategy_card: Dict[str, Any]) -> List[str]:
+    if not isinstance(strategy_card, dict) or not strategy_card.get("enabled"):
+        return []
+
+    observability = strategy_card.get("observability") or {}
+    signals = strategy_card.get("signals") or {}
+    risk_flags = list(signals.get("risk_flags") or [])
+    stage = str(strategy_card.get("chapter_stage") or "build_up")
+    genre_key = str(strategy_card.get("genre_profile_key") or strategy_card.get("pilot") or "general")
+
+    stage_text = {
+        "build_up": "本章以铺压为主,优先做威胁与代价的可感知铺垫。",
+        "confront": "本章以正面对抗为主,确保破局路径清晰可复盘。",
+        "release": "本章以释放与余波为主,给出实质收益并引出下一问。",
+    }.get(stage, "本章保持压力-破局-余波的完整链路。")
+
+    items = [
+        f"方法论策略(通用/{genre_key}):{stage_text}",
+        "长线控制:换图承接旧资产,避免主角进入新地图后能力与资源归零。",
+        "机制控制:关键胜利必须写出机制理由与代价,不用纯光环碾压。",
+        (
+            "连载互动:保留一个可讨论分歧点,强化下章追更动机。"
+            f"(next_reason={observability.get('next_reason_clarity')})"
+        ),
+    ]
+
+    if "pattern_overuse_watch" in risk_flags:
+        dominant_pattern = str(signals.get("dominant_pattern") or "").strip()
+        if dominant_pattern:
+            items.append(f"风险修正:近期“{dominant_pattern}”偏高频,本章补一个异质副轴避免疲劳。")
+    if "readability_guard" in risk_flags:
+        items.append("风险修正:近期审查均分偏低,本章优先保证段落动作-结果闭环与可读性。")
+
+    return items
+
+
 def build_guidance_items(
     *,
     chapter: int,
@@ -103,6 +280,7 @@ def build_writing_checklist(
     guidance_items: List[str],
     reader_signal: Dict[str, Any],
     genre_profile: Dict[str, Any],
+    strategy_card: Dict[str, Any] | None = None,
     min_items: int,
     max_items: int,
     default_weight: float,
@@ -198,6 +376,32 @@ def build_writing_checklist(
             verify_hint="主冲突与题材核心承诺保持一致。",
         )
 
+    if isinstance(strategy_card, dict) and strategy_card.get("enabled"):
+        _add_item(
+            "methodology_next_reason",
+            "方法论:下章动机需可复述(章末或后段均可)",
+            weight=default_weight,
+            required=False,
+            source="methodology.next_reason",
+            verify_hint="提炼一句“为什么要点下一章”的动机句。",
+        )
+        _add_item(
+            "methodology_power_guard",
+            "方法论:越级与破局给出机制理由与代价",
+            weight=default_weight,
+            required=False,
+            source="methodology.power_guard",
+            verify_hint="至少写清1个机制理由与1个代价。"
+        )
+        _add_item(
+            "methodology_antagonist_pressure",
+            "方法论:反派行动具备目标-手段-代价",
+            weight=default_weight,
+            required=False,
+            source="methodology.antagonist",
+            verify_hint="反派不是工具人推进,需有可解释行动逻辑。",
+        )
+
     for idx, text in enumerate(guidance_items, start=1):
         if len(items) >= max_items:
             break
@@ -267,5 +471,8 @@ def is_checklist_item_completed(item: Dict[str, Any], reader_signal: Dict[str, A
     if source.startswith("fallback"):
         return True
 
-    return False
+    if source.startswith("methodology."):
+        # 方法论条目当前作为软提示,仅做观察与引导,不参与扣分。
+        return True
 
+    return False

+ 22 - 0
.claude/scripts/extract_chapter_context.py

@@ -478,6 +478,7 @@ def _render_text(payload: Dict[str, Any]) -> str:
     guidance_items = writing_guidance.get("guidance_items") or []
     checklist = writing_guidance.get("checklist") or []
     checklist_score = writing_guidance.get("checklist_score") or {}
+    methodology = writing_guidance.get("methodology") or {}
     if guidance_items or checklist:
         lines.append("## 写作执行建议")
         lines.append("")
@@ -526,6 +527,27 @@ def _render_text(payload: Dict[str, Any]) -> str:
 
         lines.append("")
 
+    if isinstance(methodology, dict) and methodology.get("enabled"):
+        lines.append("## 长篇方法论策略")
+        lines.append("")
+        lines.append(f"- 框架: {methodology.get('framework')}")
+        methodology_scope = methodology.get("genre_profile_key") or methodology.get("pilot") or "general"
+        lines.append(f"- 适用题材: {methodology_scope}")
+        lines.append(f"- 章节阶段: {methodology.get('chapter_stage')}")
+        observability = methodology.get("observability") or {}
+        if observability:
+            lines.append(
+                "- 指标: "
+                f"next_reason={observability.get('next_reason_clarity')}, "
+                f"anchor={observability.get('anchor_effectiveness')}, "
+                f"rhythm={observability.get('rhythm_naturalness')}"
+            )
+        signals = methodology.get("signals") or {}
+        risk_flags = list(signals.get("risk_flags") or [])
+        if risk_flags:
+            lines.append(f"- 风险标记: {', '.join(str(flag) for flag in risk_flags)}")
+        lines.append("")
+
     reader_signal = payload.get("reader_signal") or {}
     review_trend = reader_signal.get("review_trend") or {}
     if review_trend: