#!/usr/bin/env python3 """ seed_extractor.py — InkOS 题材感知种子提取器 从 InkOS 题材配置文件（YAML frontmatter + Markdown body）提取 题材特定的初始提示词种子，映射到 novel-write 的 5 岛结构。 回退链：指定题材 → other.md → 硬编码通用种子（SEED_TEMPLATES） """ import sys, os, re, json, yaml from pathlib import Path from typing import Any sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from utils import ( save_genre_evolution, find_project_genre, map_genre_to_inkos_id, resolve_genre_from_project, log_evolution_event, ) # ── 通用回退种子（当 InkOS 不可用时） ───────────────── FALLBACK_SEEDS: dict[str, list[str]] = { "writing": [ "核心规则（写作时内化不列出）：Show, don't tell——用细节堆出真实。价值观盐溶于汤——靠行为证明不靠口号。任何行动立于三条腿：过往经历+当前利益+性格底色。配角有自己账本和利益诉求。每章结尾必有小悬念或情绪缺口。全员智商在线。日常场景七成成后续伏笔。关系改变事件驱动。每一行字要么推动剧情要么塑造人物。", "看点密集度（硬尺）：每300字至少1个爽点。每500字至少1个钩子。每1000-1500字至少1个完整悬念。连续300字以上只是环境/回忆/议论/心理独白而无推进=水文须删改。叙事段≥40字，目标40-120字。短段(<40字)只允许：开场反转金句/章末钩子/≤3个爆点短段，合计≤5个。", "去AI味铁律：叙述者永不替读者下结论。禁止分析报告式语言入正文。转折/惊讶标记词（仿佛/忽然/竟/猛地/不禁/宛如）全篇≤每3000字1次。硬性禁令：全文禁止'不是…而是…'句式；全文禁止破折号'——'。", ], "revision": [ "修稿分流规则：PATCHES处理局部文字问题——每个PATCH引用原文段落给出替换文本。REVISED_CONTENT处理全章级问题——Critical包含字数或结构性问题时必须用REVISED_CONTENT。修稿原则：修根因不做表面润色；伏笔状态须与伏笔池同步；不改变剧情走向和核心冲突。", "润色边界（硬约束）：只改文字层——句式/段落/排版/用词/五感/对话自然度。禁止增删情节/改变人设/调整主线。文字层硬规约：段落3-5行/段；句式多样化；动词>形容词；每场景1-2种感官。修改后总长变化≤±15%。", ], "quality_check": [ "自修正循环（不可跳过门控）：Loop最多5轮——综合评分≥85通过→60-84针对性修订→40-59重写问题段落→<40废弃从细纲重新写。每轮记录本轮评分/发现问题/修复内容/下一轮重点。", "五遍编辑评估（按顺序不可跳过）：P1结构(35%)→P2角色(25%)→P3原创性(15%)→P4对话(15%)→P5文本(10%)。综合≥85通过/60-84修订/40-59重写/<40废弃。P1不过不进P2。", ], "outline": [ "大纲共创三层决策树：第一层全书结构——分几卷？每卷核心任务？不可逆变化？全书最高潮在哪卷？第二层卷纲细化——高潮事件？需要哪些铺垫（≥3个伏笔）？主角情感弧线？卷末钩子须全书级别。第三层细纲关键节点——前10章每章核心事件/章末钩子类型/爽点类型/伏笔推进计划。", "章节备忘七段格式：当前任务→读者此刻在等什么→该兑现的/暂不掀的→日常/过渡承担任务→关键抉择过三连问→章尾必须发生的改变→本章hook账→不要做。", ], "hook_management": [ "伏笔全生命周期：plant（种植）→advance（推进）→payoff（兑现）→defer（推迟）→stale（过期）。每章前读钩子账本→列出高优活跃钩子→确认推进/兑现机会。每章后更新：兑现移走→推进追加→新种写入→检查stale。", "钩子准入控制：不重复——指向同一事件合并不新增。不超量——每卷最多3-5个high优先级，活跃钩子总数≤12。有计划——无兑现计划的钩子不种。新钩子候选：正文中出现会延续到后续且有具体回收方向的未解问题才新增。", ], } class SeedExtractor: """从 InkOS 题材配置提取 5 岛种子提示词""" def __init__(self, inkos_genres_dir: str | None = None): if inkos_genres_dir: self.genres_dir = Path(inkos_genres_dir) else: # 内置题材配置，无外部依赖 self.genres_dir = Path(__file__).resolve().parent.parent / "genres" self._profile_cache: dict[str, dict | None] = {} # ── 题材配置加载 ───────────────────────────────── def _load_profile(self, genre_id: str) -> dict | None: """加载单个题材配置文件，解析 YAML frontmatter + body sections""" if genre_id in self._profile_cache: return self._profile_cache[genre_id] fpath = self.genres_dir / f"{genre_id}.md" if not fpath.exists(): self._profile_cache[genre_id] = None return None text = fpath.read_text(encoding="utf-8") fm_match = re.match(r"^---\s*\n([\s\S]*?)\n---\s*\n([\s\S]*)$", text) if not fm_match: self._profile_cache[genre_id] = None return None try: frontmatter = yaml.safe_load(fm_match.group(1)) except yaml.YAMLError: frontmatter = {} body = fm_match.group(2) sections = {} for sec_match in re.finditer(r"##\s+(.+?)\s*\n([\s\S]*?)(?=\n## |\Z)", body): title = sec_match.group(1).strip() content = sec_match.group(2).strip() sections[title] = content profile = { "frontmatter": frontmatter or {}, "sections": sections, "language": frontmatter.get("language", "zh") if frontmatter else "zh", } self._profile_cache[genre_id] = profile return profile def _load_with_fallback(self, genre_id: str) -> dict: """加载题材配置，失败时回退到 other → None""" profile = self._load_profile(genre_id) if profile: return profile if genre_id != "other": return self._load_profile("other") or {"frontmatter": {}, "sections": {}, "language": "zh"} return {"frontmatter": {}, "sections": {}, "language": "zh"} # ── 字段→岛屿 映射 ─────────────────────────────── def _build_writing_seeds(self, fm: dict, sections: dict, lang: str) -> list[str]: """从题材配置构建 writing 岛种子""" seeds = [] # 疲劳词 → 词汇疲劳规避规则 fatigue_words = fm.get("fatigueWords", []) if fatigue_words: word_list = "、".join(fatigue_words) seeds.append( f"本文疲劳词监控（题材特定）：以下词汇单章最多使用1次——{word_list}。" f"超限即标记为S2警告。这些是{lang == 'en' and 'genre fatigue words' or '题材疲劳词'}，" f"过度使用会让读者产生'又是这套'的审美疲劳。" ) # 节奏规则 pacing = fm.get("pacingRule", "") if pacing: seeds.append(f"节奏控制规则（题材特定）：{pacing}。偏离此节奏需在章节备忘中说明原因。") # 爽点类型 satisfaction = fm.get("satisfactionTypes", []) if satisfaction: types_str = "、".join(satisfaction) seeds.append( f"本文期待感类型优先排序（题材特定）：{types_str}。" f"每章至少兑现其中一种类型。爽点设计优先从该清单选取，清单外类型需在细纲中标注'非标爽点'。" ) # 叙事指导 section narrative_guidance = ( sections.get("叙事指导") or sections.get("Narrative Guidance") or sections.get("Pacing Guidance") or "" ) if narrative_guidance: # 提取关键句子（以句号分隔，取有实质内容的句子） key_lines = [ s.strip() for s in re.split(r"[。.\n]", narrative_guidance) if len(s.strip()) > 10 and not s.strip().startswith("-") ][:5] if key_lines: guidance_text = "。".join(key_lines) seeds.append(f"题材叙事指导（注入写作规则）：{guidance_text}。") return seeds def _build_revision_seeds(self, fm: dict, sections: dict, lang: str) -> list[str]: """从题材配置构建 revision 岛种子""" seeds = [] # 语言铁律 / Language Rules language_rules = sections.get("语言铁律") or sections.get("Language Rules") or "" if language_rules: # 提取具体规则行 rule_lines = [ line.strip().lstrip("- ").lstrip("* ") for line in language_rules.split("\n") if line.strip() and (line.strip().startswith("-") or line.strip().startswith("*") or "→" in line or "✗" in line or "✓" in line) ] if rule_lines: rules_text = "；".join(rule_lines[:8]) seeds.insert(0, f"题材润色特化规则：{rules_text}。这些规则优先于通用润色规则执行。") # 年代考据规则 if fm.get("eraResearch"): seeds.append( "年代考据规则（题材特定）：涉及法律/政策/商业规则/物价/机构名/行业术语必须符合设定年代。" "金融操作和公司运营需有基本可信度。地名和机构名不可凭空编造。" ) return seeds def _build_quality_check_seeds(self, fm: dict, sections: dict, lang: str) -> list[str]: """从题材配置构建 quality_check 岛种子""" seeds = [] # 题材禁忌 → S1/S2 检查项 prohibitions = sections.get("题材禁忌") or sections.get("Genre Prohibitions") or "" if prohibitions: items = [] for line in prohibitions.split("\n"): stripped = line.strip().lstrip("- ").lstrip("* ") if len(stripped) > 10: items.append(stripped) if items: check_items = [] for i, item in enumerate(items[:10]): severity = "S1" if any( kw in item for kw in ["禁止", "不可", "杜绝", "never", "must not", "prohibited"] ) else "S2" check_items.append(f" {i+1}. [{severity}] {item}") seeds.append( f"题材特定质量审查清单：\n" + "\n".join(check_items) + f"\n每章完成后逐项检查，S1项不过直接退回修改。" ) # 审计维度 audit_dims = fm.get("auditDimensions", []) if audit_dims: dims_str = ",".join(str(d) for d in audit_dims) seeds.append( f"37维审查启用维度子集（题材特定）：[{dims_str}]。" f"未启用的维度本作不适用，跳过检查以降低误报。" ) # 数值系统检查 if fm.get("numericalSystem"): seeds.append( "数值一致性检查（题材特定）：统计面板/战力数值/资源计数必须与上下文一致。" "同质资源重复获取须写明衰减。增量逐笔列出，不可跳跃式增长。" ) # 战力体系检查 if fm.get("powerScaling"): seeds.append( "战力体系检查（题材特定）：境界/品阶/等级突破必须有积累过程。" "不可出现无铺垫跳跃式突破。跨大境界需特殊条件（天劫/机缘/丹药）。" "能力上限/附加代价/触发条件/成长路径——四维约束缺一不可。" ) return seeds def _build_outline_seeds(self, fm: dict, sections: dict, lang: str) -> list[str]: """从题材配置构建 outline 岛种子""" seeds = [] # 章节类型 chapter_types = fm.get("chapterTypes", []) if chapter_types: types_str = "、".join(chapter_types) seeds.append( f"章节类型轮换策略（题材特定）：可用类型——{types_str}。" f"连续3章以上同类型触发疲劳警告。每卷类型分布应覆盖所有列出类型。" f"各类型比例为：战斗/推进类40%，布局/过渡类35%，回收/高潮类25%。" ) # 世界观规则 section world_rules = ( sections.get("修炼规则") or sections.get("World Rules") or sections.get("年代与现实约束") or sections.get("System Design Rules") or "" ) if world_rules: key_rules = [] for line in world_rules.split("\n"): stripped = line.strip().lstrip("- ").lstrip("* ") if len(stripped) > 15: key_rules.append(stripped) if key_rules: rules_text = "；".join(key_rules[:6]) seeds.append(f"世界观一致性规则（题材特定，大纲阶段检查）：{rules_text}。") return seeds def _build_hook_management_seeds(self, fm: dict, sections: dict, lang: str) -> list[str]: """从题材配置构建 hook_management 岛种子""" seeds = [] # 题材特定的悬念期待类型 satisfaction = fm.get("satisfactionTypes", []) if satisfaction: types_str = "、".join(satisfaction[:4]) seeds.append( f"题材悬念期待管理：本类型读者的核心期待——{types_str}。" f"伏笔设计优先围绕这几类期待展开。每卷至少种下2个指向不同类型期待的伏笔。" ) # 节奏规则 → 伏笔兑现节奏 pacing = fm.get("pacingRule", "") if pacing: seeds.append( f"伏笔兑现节奏约束：{pacing}。伏笔的advance和payoff需与此节奏同步——" f"快节奏段兑现密度可提高，慢节奏段以种植和推进为主。" ) # 从世界观/修炼规则提取伏笔管理特殊规则 world_rules = ( sections.get("修炼规则") or sections.get("World Rules") or sections.get("年代与现实约束") or "" ) if world_rules: key_lines = [ s.strip().lstrip("- ").lstrip("* ") for s in world_rules.split("\n") if len(s.strip()) > 15 and any( kw in s for kw in ["规则", "约束", "限制", "禁止", "条件", "代价", "rule", "limit", "cannot", "must"] ) ][:3] if key_lines: rules_text = "；".join(key_lines) seeds.append( f"题材伏笔管理特殊规则：{rules_text}。" f"违反这些规则的伏笔在plant阶段即应被拒绝或标记为'需特殊机制'。" ) return seeds # ── 主入口 ────────────────────────────────────── def extract_seeds(self, genre_id: str | None = None, project_path: Path | None = None) -> dict[str, list[str]]: """ 提取题材特定的 5 岛种子提示词。 返回: {island_name: [prompt_strings, ...]} """ # 确定题材 ID if not genre_id and project_path: genre_name = find_project_genre(project_path) genre_id = map_genre_to_inkos_id(genre_name) elif not genre_id: genre_id = "other" profile = self._load_with_fallback(genre_id) fm = profile.get("frontmatter", {}) sections = profile.get("sections", {}) lang = profile.get("language", "zh") # 如果 profile 是空的（InkOS 不可用），用回退种子 if not fm and not sections: print(f" ⚠ InkOS 题材配置不可用，使用通用回退种子") return FALLBACK_SEEDS seeds = { "writing": self._build_writing_seeds(fm, sections, lang), "revision": self._build_revision_seeds(fm, sections, lang), "quality_check": self._build_quality_check_seeds(fm, sections, lang), "outline": self._build_outline_seeds(fm, sections, lang), "hook_management": self._build_hook_management_seeds(fm, sections, lang), } # 对种子过少的岛屿，从 general pool 补充（避免重复） general = self._load_profile("other") for island, current in seeds.items(): if len(current) < 2: # 先从 general profile 补 if general and general.get("frontmatter"): gf = general["frontmatter"] gs = general["sections"] gl = general.get("language", "zh") fallback_builder = getattr(self, f"_build_{island}_seeds", None) if fallback_builder: fb_seeds = fallback_builder(gf, gs, gl) for s in fb_seeds: if s not in current: current.append(s) if len(current) >= 2: break # 还不够 → 从硬编码回退种子补 if len(current) < 2 and island in FALLBACK_SEEDS: for s in FALLBACK_SEEDS[island]: if s not in current: current.append(s) if len(current) >= 2: break return seeds def initialize_genre(self, genre_id: str) -> dict[str, list[str]]: """ 初始化题材级共享进化目录 evolution/genres/{genre_id}/。 创建 population.json（含题材种子）和相关文件。 返回提取的种子字典。 """ seeds = self.extract_seeds(genre_id=genre_id) import hashlib population = {} for island_name, templates in seeds.items(): variants = [] for template in templates: variant_id = hashlib.md5(template.encode()).hexdigest()[:12] variants.append({ "id": variant_id, "template": template, "source": f"seed:{genre_id}", "scores": {}, "trials": 0, "trials_per_dim": {}, "lineage": [], }) population[island_name] = { "generation": 0, "variants": variants, } save_genre_evolution(genre_id, "prompts/population.json", population) save_genre_evolution(genre_id, "prompts/current_best.json", {k: [v["template"] for v in pop["variants"][:3]] for k, pop in population.items()}) default_thresholds = load_evolution("thresholds.json") if default_thresholds: save_genre_evolution(genre_id, "thresholds.json", default_thresholds) else: save_genre_evolution(genre_id, "thresholds.json", { "pass": 85, "revise_low": 60, "rewrite_low": 40, "max_revision_rounds": 3, "weights": {"P1": 0.35, "P2": 0.25, "P3": 0.15, "P4": 0.15, "P5": 0.10}, }) save_genre_evolution(genre_id, "patterns/pattern_library.json", { "patterns": [], "meta": {"total_mined": 0} }) save_genre_evolution(genre_id, "session_state.json", { "recent_chapters": [], "total_feedback_count": 0 }) print(f" ✓ 题材进化目录已初始化: evolution/genres/{genre_id}/") for island, tmpls in seeds.items(): print(f" [{island}]: {len(tmpls)} 条种子") return seeds # ── CLI ───────────────────────────────────────────────── if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="InkOS 题材感知种子提取器") parser.add_argument("--project", "-p", type=str, default=None, help="项目路径（自动检测题材）") parser.add_argument("--genre", "-g", type=str, default=None, help="强制指定题材 ID") parser.add_argument("--show", action="store_true", help="仅显示提取的种子，不初始化") args = parser.parse_args() extractor = SeedExtractor() if args.project: project = Path(args.project) genre_id = args.genre or resolve_genre_from_project(project) else: genre_id = args.genre or "other" if args.show: seeds = extractor.extract_seeds(genre_id=genre_id) for island, tmpls in seeds.items(): print(f"\n## {island} ({len(tmpls)} 条)") for i, t in enumerate(tmpls): print(f" {i+1}. {t[:150]}...") else: extractor.initialize_genre(genre_id)