#!/usr/bin/env python3 """ prompt_evolver.py — 提示词模板进化器 进化结果写入 evolution/prompts/，不修改 SKILL.md。 核心机制： - 种群管理 + 双选择策略（Top + Diverse） - 遗传算子：交叉、突变、简化 - 岛屿模型：5个独立种群，定期迁移 - 进化结果同步到 evolution/prompts/current_best.json """ import sys, os, re, json, random, hashlib from pathlib import Path from typing import Any sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from utils import ( find_project_root, get_evolution_dir, load_evolution, save_evolution, load_project_db, save_project_db, log_evolution_event, ) # ── 种子模板 ─────────────────────────────────────────── SEED_TEMPLATES = { "writing": [ "核心规则（写作时内化不列出）：Show, don't tell——用细节堆出真实。价值观盐溶于汤——靠行为证明不靠口号。任何行动立于三条腿：过往经历+当前利益+性格底色。配角有自己账本和利益诉求。每章结尾必有小悬念或情绪缺口。全员智商在线。日常场景七成成后续伏笔。关系改变事件驱动。每一行字要么推动剧情要么塑造人物。", "看点密集度（硬尺）：每300字至少1个爽点（小看点/梗/反套路/暧昧台词/情绪拉扯）。每500字至少1个钩子（引发'接下来怎样'的小悬念）。每1000-1500字至少1个完整悬念（问题—蓄力—未解）。连续300字以上只是环境/回忆/议论/心理独白而无推进=水文须删改。叙事段≥40字（手机屏2行），目标40-120字。短段(<40字)只允许：开场反转金句/章末钩子/≤3个爆点短段，合计≤5个。禁止3+短段连排。", "去AI味铁律：叙述者永不替读者下结论。禁止分析报告式语言入正文——'核心动机''信息边界''信息落差'等术语只用于PRE_WRITE_CHECK内部推理。转折/惊讶标记词（仿佛/忽然/竟/猛地/不禁/宛如）全篇≤每3000字1次。同一体感/意象禁止连续渲染超两轮。硬性禁令：全文禁止'不是…而是…'句式；全文禁止破折号'——'。反例→正例：✗'他感到非常愤怒'→✓'他捏碎了茶杯，滚烫茶水流过指缝'；✗'然而事情没那么简单'→✓'哪有那么便宜的事'。", "六步走人物心理分析：1.当前处境（面临什么局面？手上什么牌？）2.核心动机（最想要什么？最怕什么？）3.信息边界（知道什么？不知道什么？误判什么？）4.性格过滤（同样局面这角色怎么反应？）5.行为选择（基于以上四点做什么选择？）6.情绪外化（伴随什么情绪？什么身体语言？）。禁止跳过步骤直接写行为。人设防崩三问：他为什么这么做？符合之前人设吗？只看过前文的读者会觉得突兀吗？", "读者心理学框架：期待管理——读者期待释放时适当延迟增强快感，即将失去耐心时立即给反馈。信息落差——让读者比角色多知道一点（紧张）或少知道一点（好奇）。情绪节拍——压制→释放→更大压制→更大释放，释放超读者预期。递进升级——层层加码，每次比上一次更过分。锚定效应——先给读者参照再展示主角表现。沉没成本——读者已投入的阅读时间是留存关键。", "黄金三章纪律：第1章主角出场800字内触发主线冲突，前300字最后一句须戏剧性反转，场景≤2个冲突人物≤2个。第2章金手指/能力/系统必须'做出来'——一次具体使用事件+看得见的后果——不是'说出来'。第3章中段让主角下一个可量化短期目标浮上水面。贯穿三章：段落3-5行（手机阅读），动词压过形容词，每章结尾必有小钩子。", "章节80/20断章：永远不要在一章里把本章故事讲完——主剧情写到80%，剩下20%留给下一章。章末必须断在action-climax一刻：主角刚放大招尚未见效/刚拔刀尚未落下——不给结果。章节结构优先于字数。不凑字数也不为卡字数提前把高潮讲完。", "代入感六支柱：基础信息标签化（100字内谁/在哪/发生什么）。可视化熟悉感（消毒水味/地铁座椅凉/外卖塑料袋塑胶感）。共鸣两层（认知共鸣+情绪共鸣）。欲望两条腿（基础欲望+主动欲望）。五感钩子（每场景除视觉外1-2种感官）。人设核心标签+反差细节才活——冷面杀手偷偷喂流浪猫。", "情感节点设计：关系发展必须事件驱动节点递进——设计3-5个关键事件（共同御敌/秘密分享/利益冲突/信任考验/牺牲妥协）。每个事件推进关系一个层级，禁止跨越式发展。情绪用场景传达：环境烘托+微动作替代直白抒情。强情绪升级法——坏事叠坏事：被骂→赶不上公交→手机掉了→直播课结束→包子噎住，每层比上一层过分。", "配角设计方法论：配角必须有反击有自己算盘。主角强大在于压服聪明人不是碾压傻子。动机绑定主线。核心标签+反差细节=活人。通过事件立人设禁止外貌形容词堆砌。语言区分度——用词习惯/句子长短/口头禅/方言痕迹。拒绝集体反应——不写'众人齐声惊呼'，挑1-2个角色写具体反应。", "输出格式：=== PRE_WRITE_CHECK ===（Markdown表格：当前任务/读者在等什么/该兑现的暂不掀的/日常过渡承担任务/章尾必须发生的改变/不要做/上下文范围/当前锚点/待回收伏笔/本章冲突/章节类型/风险扫描）=== CHAPTER_TITLE ===（章节标题不含'第X章'，避开近期重复标题词根）=== CHAPTER_CONTENT ===（正文目标字数区间）。只输出以上三个区块。", ], "revision": [ "修稿分流规则：PATCHES处理局部文字问题（措辞/对话/AI痕迹/小连续性错误）——每个PATCH引用原文段落给出替换文本，未涉及内容保持原样。REVISED_CONTENT处理全章级问题（字数压缩/结构重组/节奏重排/重大剧情偏离）——Critical包含字数或结构性问题时必须用REVISED_CONTENT。修稿原则：修根因不做表面润色；伏笔状态须与伏笔池同步；不改变剧情走向和核心冲突；保留原文语言风格节奏呼吸；情绪用动作外化；坏事叠坏事每层比上一层过分。", "润色边界（硬约束）：只改文字层——句式/段落/排版/用词/五感/对话自然度。禁止增删情节/改变人设/调整主线——这些归Reviewer。消灭六条文笔雷点：描写无效删到一笔带过；文笔华丽过度让文字服从情绪；文笔欠佳重写成通顺有画面句子；排版不规范统一手机阅读格式；AI味痕迹替换口语化表达或具体动作；群像脸谱化不写众人震惊。文字层硬规约：段落3-5行/段；句式多样化禁3句以上同结构；动词>形容词；每场景1-2种感官；对话符合身份情绪信息掌握；删除叙述者结论。修改后总长变化≤±15%。", "字数治理：压缩时优先合并重复信息/删除无变化场景/精简过渡段落——如需删75%以上原文→拒绝标准化标记为需人工拆分。扩写时优先增加五感描写/角色内心反应/场景氛围——如扩展后信息密度显著下降→拒绝标准化标记为需合并。只修正一次不递归。保留关键标记/人物名/地点名/已有事实。不凭空新增子情节。", "小目标周期修稿指引：如果本章应该是'后效'阶段但仍在加压→把最密集冲突段落改写为展示改变的段落（谁失去了什么/谁的态度变了/新常态是什么）。如果本章应该是'爆发'阶段但没有明确兑现→放大最接近回报的场景让承诺释放超预期。日常段落不服务主线→改写为'饵'：加入指向未来的细节/暗示/角色反应。", ], "quality_check": [ "自修正循环（不可跳过门控）：Loop最多5轮——执行五遍编辑评估→综合评分≥85通过→60-84针对性修订→40-59重写问题段落→<40废弃从细纲重新写。每轮记录本轮评分/发现问题（S1-S4分级）/修复内容/下一轮重点。反模式：评分≥85还在修=无限打磨者；只跑P1=跳过者；每轮同问题=失忆者需向上追溯。", "五遍编辑评估（按顺序不可跳过）：P1结构(35%)——场景目标→阻碍→变化？章节推动力？爽点释放时机？章末钩子拉力？节奏快慢交替？P2角色(25%)——弧线推进？行为一致性？关系变化？配角不只是工具人？换个名字还能认出谁在说话？P3原创性(15%)——反陈词滥调？反AI模式？至少一处读者没预料到？P4对话(15%)——声音辨识度？三维标准同时携带？潜台词？P5文本(10%)——句式变化？措辞精准？五感至少2-3种？综合≥85通过/60-84修订/40-59重写/<40废弃。P1不过不进P2。", "37维结构审查（InkOS Continuity Auditor标准）：OOC/时间线/设定冲突/战力崩坏/数值/伏笔/节奏/文风/信息越界/词汇疲劳/利益链断裂/年代考据/配角降智/配角工具人化/爽点虚化/台词失真/流水账/知识库污染/视角一致性/段落等长/套话密度/公式化转折/列表式结构/支线停滞/弧线平坦/节奏单调/敏感词/正传事件冲突/未来信息泄露/世界规则跨书一致性/番外伏笔隔离/读者期待管理/章节备忘偏离/角色还原度/世界规则遵守/关系动态/正典事件一致性。评分：95-100可发布；85-94小瑕疵流畅；75-84明显问题主干完整；65-74多处影响阅读；<65结构性崩溃。", "写后确定性校验（零LLM成本规则）：'不是…而是…'→error改用直述句；破折号'——'→error用逗号或句号；惊讶标记词全篇≤每3000字1次→warning；高疲劳词单章最多1次→warning；元叙事（'到这里算是''接下来就是'）→warning；分析报告术语入正文→error；章号引用（'第X章'）→error改自然表达；作者说教词（'显然''毋庸置疑'）→warning；集体震惊反应→warning改写1-2个具体角色；连续'了'字句≥6句→warning；长段落≥2段超300字→warning；段落过碎（60%以上<40字）→warning；连续短段≥3段→warning。", "基础设定审核（5维度评分）：1.核心冲突——是否有清晰且有足够张力的核心冲突支撑40章？2.开篇节奏——前5章能否形成翻页驱动力？3.世界一致性——世界观是否内洽且具体？4.角色区分度——主要角色声音和动机是否各不相同？5.节奏可行性——卷纲是否有足够变化不连续10章同节拍？80+通过可开写/60-79需修改/<60方向性错误需重新设计。", ], "outline": [ "大纲共创三层决策树：第一层全书结构——分几卷？每卷核心任务？每卷不可逆变化？全书最高潮在哪卷为什么不可替代？第二层卷纲细化——高潮事件？需要哪些铺垫（≥3个伏笔/线索）？几个高潮点分别在哪些章节？主角情感弧线从什么状态到什么状态？卷末钩子须全书级别——不是'下章会怎样'而是'整个故事走向要变了'。第三层细纲关键节点——前10章每章核心事件/章末钩子类型和期待感等级/爽点类型和铺垫起点/推进哪个伏笔种下什么新伏笔。关键章逐章追问，普通章确认核心推进和钩子级别。", "架构师基础设定生成：产出散文密度基础设定——不是表格/schema/条目式bullet list。story_frame（4段散文）：主题与基调→核心冲突与前台/后台双层故事→世界观底色（3-5条不可违反铁律+感官锚点）→终局方向+全书Objective。volume_map（5段散文+节奏原则尾段）：各卷主题与情绪曲线→卷间钩子与回收承诺→各卷OKR（Objective+3 Key Results）→卷尾不可逆改变→节奏原则6条（高潮间距/喘息频率/钩子密度/信息释放节奏/爽点节奏/情感节点递进）。", "章节备忘（chapter_memo）七段格式：## 当前任务→本章必须完成的具体动作，写作始终对齐。## 读者此刻在等什么→控制情绪缺口的制造/延迟/兑现程度。## 该兑现的/暂不掀的→本章必须兑现伏笔+必须压住不掀底牌。## 日常/过渡承担什么任务→非冲突段落功能映射（段落位置→承担功能）。## 关键抉择过三连问→关键选择检查（为何做？符合利益？符合人设？）。## 章尾必须发生的改变→1-3条（信息/关系/物理/权力）。## 本章hook账→硬对应：advance/resolve下每个hook_id须在正文里有具体可定位兑现段（具体物件/事件/信息+可观察动作或交谈），每兑现段≥60字。defer不用落open在章末安排自然引出新悬念。## 不要做→硬约束红线。", "规划器14条原则：3-5章一个小目标周期（build-up→escalation→climax→aftermath完整循环）。主动塑造读者期待——信息落差+情绪缺口。万物皆饵——日常段埋伏笔/推关系/建反差。人设防崩——三连问每次写行为前必问。1主线+1支线——不超载，支线2-3章内与核心目标关联。爽点密集化——每300字至少1看点/500字1钩子/1000-1500字1完整悬念。高潮前足够铺垫——铺垫越长释放越爽。高潮后展示影响——紧接着1-2章写出改变。人物立体化——核心标签+反差细节。五感具体化——每场景至少1-2种感官。钩子承接——每章结尾必种新钩子或推进已有。钩子账本必结账——兑现移走/推进追加/新种写入/stale标记。圆心法同场多视角——同一事件不同角色视角呈现不同信息。揭1埋2——每揭示一个答案至少埋下两个新问题。", ], "hook_management": [ "伏笔全生命周期：plant（种植）——新伏笔分配唯一ID/标记类型（SUSPENSE/PROMISE/MYSTERY/CONFLICT/CHEKHOV）和优先级/标明计划兑现章。advance（推进）——虽未兑现但给出新线索新事实新风险升级或范围收缩，更新最近推进章号。payoff（兑现）——明确揭示或解决，状态改为已回收。defer（推迟）——主动搁置转入后台或被剧情压后。stale（过期）——超10章无推进决定推进/废弃/合并。每章前读钩子账本→列出高优活跃钩子→确认推进/兑现机会。每章后更新：兑现移走→推进追加→新种写入→检查stale。", "钩子准入控制：不重复——指向同一事件合并不新增；同一钩子家族（多个钩子都指向主角身份秘密）合并为核心钩子+多条线索。不超量——每卷最多3-5个high优先级，活跃钩子总数≤12。有计划——无兑现计划的钩子不种，先想好怎么收再种。新钩子候选：正文中出现会延续到后续且有具体回收方向的未解问题才新增。不要为旧钩子换说法/重述/抽象总结再开新hook_id。提及vs推进：旧钩子只是被提到无新增信息→mention不更新最近推进；出现新事实/证据/关系变化/风险升级→必须更新最近推进=当前章号。", "钩子健康分析：stale（>10章无推进）→立即标记决定推进/废弃/合并。burst（单章>5个新钩子）→警告削减低优先级。no-advance（high钩子>5章无推进）→下章安排推进。payoff逾期（计划兑现章已过未标记）→确认断头钩子。核心钩子（core_hook=true）stale超10章→warning升级critical——全书只有3-7条核心钩子，任其漂移是叙事腐烂先行指标。卷末核心钩子仍open/stale无显式转入下卷规划→critical。", "状态追踪铁律（Settler）：只提取正文中明确描写的事件和状态变化——不推断/不预测/不补充正文没写内容。正文只写到角色走到门口→状态卡不能写已进入房间。正文暗示可能性未确认→不当已发生事实记录。不删除或修改已有hooks中与本章无关内容——只更新本章正文涉及hooks。伏笔例外：正文中未解疑问/悬念/伏笔线索须在hooks中记录——这是提取叙事承诺不是推断。结算输出：RUNTIME_STATE_DELTA（JSON增量不重写完整truth files）——含currentStatePatch+hookOps(upsert/mention/resolve/defer)+newHookCandidates+chapterSummary。", "钩子推进判定（严格执行）：只有状态真的变了才更新最近推进。不要把'再次提到''换个说法重述''抽象复盘'当推进。判定标准：旧钩子本章出现新事实/新证据/新关系变化/风险升级/范围收缩→推进更新lastAdvancedChapter=当前章号。旧钩子本章只是被提到无新增信息→mention不更新最近推进。旧钩子本章被明确揭示/解决/不再成立→回收状态改已回收。旧钩子被主动搁置/转入后台/被剧情压后→延后标注原因。payoffTiming用语义节奏不用硬写章号：immediate/near-term/mid-arc/slow-burn/endgame。", "长跨度疲劳检测：章节类型单调——连续5+章同一类型→警告建议轮换铺垫-兑现-反转-后效。情绪单调——连续高张力无释放→警告插入释放节拍（温情/幽默/亲密/反思安静）再升级。标题塌缩——近3章标题共享相同命名壳→警告换新意象/动作/后果/角色向量。开头/结尾句式重复——最近章首句或末句相似度过高→警告从动作/后果/意外开局不用相同镜头调度。", ], } class PromptVariant: """提示词变体 — 支持多维评分""" def __init__(self, template: str, source: str = "manual", scores: dict[str, float] | None = None): self.id = hashlib.md5(template.encode()).hexdigest()[:12] self.template = template self.source = source self.scores: dict[str, float] = scores or {} self.trials_per_dim: dict[str, int] = {} self.trials = 0 self.lineage: list[str] = [] def composite_score(self, weights: dict[str, float] | None = None) -> float: """按权重计算加权综合分""" if not self.scores: return 0.0 # 兼容旧格式：单字段 "composite" if "composite" in self.scores and len(self.scores) == 1: return self.scores["composite"] if not weights: weights = {"P1": 0.35, "P2": 0.25, "P3": 0.15, "P4": 0.15, "P5": 0.10} total = 0.0 weight_sum = 0.0 for dim, w in weights.items(): if dim in self.scores: total += self.scores[dim] * w weight_sum += w return total / weight_sum if weight_sum > 0 else 0.0 def to_dict(self) -> dict: return { "id": self.id, "template": self.template, "source": self.source, "scores": self.scores, "trials": self.trials, "trials_per_dim": self.trials_per_dim, "lineage": self.lineage, } @classmethod def from_dict(cls, d: dict) -> "PromptVariant": p = cls(d["template"], d.get("source", "manual")) p.id = d.get("id", p.id) p.scores = d.get("scores", {}) p.trials = d.get("trials", 0) p.trials_per_dim = d.get("trials_per_dim", {}) p.lineage = d.get("lineage", []) # 向后兼容: 旧格式 'score' 单字段 → composite if not p.scores and "score" in d: p.scores = {"composite": d["score"]} return p class PromptPopulation: """提示词种群 — 支持多维评分、停滞检测、灭绝重播""" def __init__(self, category: str, max_size: int = 35): self.category = category self.max_size = max_size self.variants: list[PromptVariant] = [] self.generation = 0 self._best_score_history: list[float] = [] def seed(self, templates: list[str]) -> None: for t in templates: self.variants.append(PromptVariant(t, source="seed")) def select_top(self, n: int = 3, weights: dict[str, float] | None = None) -> list[PromptVariant]: scored = [v for v in self.variants if v.trials > 0] scored.sort(key=lambda v: v.composite_score(weights), reverse=True) return scored[:n] def select_diverse(self, n: int = 2) -> list[PromptVariant]: if len(self.variants) <= n: return self.variants[:] by_len = sorted(self.variants, key=lambda v: len(v.template)) indices = [int(i * len(by_len) / (n + 1)) for i in range(1, n + 1)] return [by_len[i] for i in indices if i < len(by_len)] def add(self, variant: PromptVariant) -> bool: for v in self.variants: if v.id == variant.id: # 合并评分（滑动平均） for dim, score in variant.scores.items(): old = v.scores.get(dim, 0) old_trials = v.trials_per_dim.get(dim, 0) if old_trials > 0: v.scores[dim] = (old * old_trials + score) / (old_trials + 1) else: v.scores[dim] = score v.trials_per_dim[dim] = old_trials + 1 v.trials += variant.trials return False if len(self.variants) < self.max_size: self.variants.append(variant) return True # 淘汰最弱个体 scored = [(v.composite_score() if v.trials > 0 else float("-inf"), v) for v in self.variants] scored.sort(key=lambda x: x[0]) worst = scored[0][1] if variant.composite_score() > worst.composite_score() or worst.trials == 0: self.variants.remove(worst) self.variants.append(variant) return True return False # ── 停滞检测 & 灭绝 ──────────────────────────── def track_best_history(self) -> None: scored = [v for v in self.variants if v.trials > 0] best = max((v.composite_score() for v in scored), default=0.0) self._best_score_history.append(best) if len(self._best_score_history) > 5: self._best_score_history = self._best_score_history[-5:] def is_stagnant(self, generations: int = 2, improvement_threshold: float = 0.03) -> bool: if len(self._best_score_history) < generations + 1: return False recent = self._best_score_history[-generations:] if recent[0] == 0: return False for s in recent[1:]: if s >= recent[0] * (1 + improvement_threshold): return False return True def needs_extinction(self, generations: int = 3, improvement_threshold: float = 0.05) -> bool: return (self.is_stagnant(generations, improvement_threshold) and len(self.variants) >= self.max_size * 0.7) def apply_extinction(self, genre_seeds: list[str] | None = None, style_injects: list[str] | None = None) -> dict: """删除底30%，重新播种""" events = {"removed": 0, "reseeded_genre": 0, "reseeded_style": 0, "reseeded_crossover": 0} scored = sorted( [v for v in self.variants if v.trials > 0], key=lambda v: v.composite_score() ) unscored = [v for v in self.variants if v.trials == 0] remove_count = max(1, int(len(scored) * 0.3)) removed = scored[:remove_count] survivors = scored[remove_count:] + unscored events["removed"] = len(removed) self.variants = survivors to_add = [] if genre_seeds: for s in genre_seeds[:max(1, remove_count // 2)]: to_add.append(PromptVariant(s, source="reseed:genre")) if style_injects: for s in style_injects[:max(1, remove_count * 3 // 10)]: to_add.append(PromptVariant(s, source="reseed:style")) if len(survivors) >= 2: for _ in range(max(1, remove_count // 5)): a, b = random.sample(survivors, 2) child = crossover(a, b) child.source = "reseed:crossover" to_add.append(child) for v in to_add: if self.add(v): key = "reseeded_" + v.source.split(":")[-1] events[key] = events.get(key, 0) + 1 return events # ── 遗传算子 ─────────────────────────────────────────── def crossover(a: PromptVariant, b: PromptVariant) -> PromptVariant: a_sents = re.split(r"[。；;]\s*", a.template) b_sents = re.split(r"[。；;]\s*", b.template) if len(a_sents) < 2 or len(b_sents) < 2: a_kw = re.findall(r"[一-鿿]{2,4}", a.template) b_kw = re.findall(r"[一-鿿]{2,4}", b.template) combined = list(set(a_kw + b_kw)) random.shuffle(combined) template = "。".join(combined[:max(3, len(combined) // 2)]) else: split_a = random.randint(1, len(a_sents) - 1) split_b = random.randint(1, len(b_sents) - 1) template = "；".join(s for s in (a_sents[:split_a] + b_sents[split_b:]) if s.strip()) child = PromptVariant(template, source="crossover") child.lineage = [a.id, b.id] return child def mutate(v: PromptVariant) -> PromptVariant: template = v.template synonyms = { "必须": ["务必", "一定", "确保"], "检查": ["核查", "验证", "确认"], "推进": ["推动", "进展"], "冲突": ["矛盾", "张力"], "钩子": ["悬念", "抓手"], "爽点": ["高潮点", "释放点"], } for word, alts in synonyms.items(): if word in template and random.random() < 0.3: template = template.replace(word, random.choice(alts), 1) sents = re.split(r"[。；;]\s*", template) if len(sents) >= 3 and random.random() < 0.3: i, j = random.sample(range(len(sents)), 2) sents[i], sents[j] = sents[j], sents[i] template = "；".join(sents) if random.random() < 0.2: prefixes = ["关键：", "注意：", "核心："] template = random.choice(prefixes) + template mutant = PromptVariant(template, source="mutate") mutant.lineage = [v.id] return mutant def simplify(v: PromptVariant) -> PromptVariant: sents = re.split(r"[。；;]\s*", v.template) if len(sents) <= 2: return v simplified = PromptVariant("；".join([sents[0], sents[-1]]), source="simplify") simplified.lineage = [v.id] return simplified def template_similarity(a: str, b: str) -> float: a_words = set(re.findall(r"[一-鿿]{2,}", a)) b_words = set(re.findall(r"[一-鿿]{2,}", b)) if not a_words or not b_words: return 0.0 return len(a_words & b_words) / len(a_words | b_words) def inject_from_style_report(report_path: Path) -> PromptVariant | None: """从对标/风格报告.md 提取风格约束并生成注入变体。 解析风格报告的结构化章节（句式/对话/描写/节奏/语调）， 编码为写作规则提示词，source 标记为 inject:{书名}。 """ try: text = report_path.read_text(encoding="utf-8") except Exception: return None fm_match = re.match(r"^---\s*\n([\s\S]*?)\n---\s*\n([\s\S]*)$", text) if not fm_match: return None body = fm_match.group(2) rules = [] source_name = report_path.parent.name # 句式 → 句长和节奏约束 sent_match = re.search(r"## 句式\s*\n(.*?)(?=\n## |\Z)", body, re.DOTALL) if sent_match: sent_text = sent_match.group(1) avg_match = re.search(r"平均句长[：:]\s*(\d+)[–\-—]?\d*\s*字", sent_text) if avg_match: rules.append(f"目标句长:{avg_match.group(1)}字/句") ratio_match = re.search(r"短句.*?(?:约占\s*)?(\d+)\s*%", sent_text) if ratio_match: rules.append(f"短句比例参考:{ratio_match.group(1)}%") # 对话 → 对话占比和风格 dial_match = re.search(r"## 对话\s*\n(.*?)(?=\n## |\Z)", body, re.DOTALL) if dial_match: dial_text = dial_match.group(1) pct_match = re.search(r"对话占比[：:]\s*约?\s*(\d+)\s*%", dial_text) if pct_match: rules.append(f"对话占比参考:{pct_match.group(1)}%") # 描写 → 五感偏好 desc_match = re.search(r"## 描写\s*\n(.*?)(?=\n## |\Z)", body, re.DOTALL) if desc_match: desc_text = desc_match.group(1) for sense in ["视觉", "触觉", "听觉", "嗅觉", "味觉"]: if sense in desc_text: rules.append(f"五感偏好:{sense}主导描写") break # 节奏 → 章首章末规则 pacing_match = re.search(r"## 节奏\s*\n(.*?)(?=\n## |\Z)", body, re.DOTALL) if pacing_match: pacing_text = pacing_match.group(1) if "章首" in pacing_text or "开头" in pacing_text: rules.append("章首规则:3段内进入情境") if "章末" in pacing_text and "钩子" in pacing_text: rules.append("章末规则:留钩子") # 语调 → 叙事距离和视角 tone_match = re.search(r"## 语调\s*\n(.*?)(?=\n## |\Z)", body, re.DOTALL) if tone_match: tone_text = tone_match.group(1) for pov in ["第三人称有限", "第三人称", "第一人称", "上帝视角", "全知"]: if pov in tone_text: rules.append(f"视角约束:{pov}") break if not rules: return None prompt = f"【{source_name}风格注入】" + "；".join(rules) v = PromptVariant(prompt, source=f"inject:{source_name}") v.lineage = [f"style:{source_name}"] return v # ── 岛屿模型 ─────────────────────────────────────────── class IslandModel: """岛屿模型 — 5 个独立种群 + 迁移 + 早停""" def __init__(self): self.islands: dict[str, PromptPopulation] = {} self.stagnation_counters: dict[str, int] = {} def initialize(self, seeds: dict[str, list[str]]) -> None: for cat, templates in seeds.items(): pop = PromptPopulation(cat) pop.seed(templates) self.islands[cat] = pop def evolve_all(self, generations: int = 1, mutation_rate: float = 0.3, weights: dict[str, float] | None = None, genre_seeds: dict[str, list[str]] | None = None, style_injects: dict[str, list[str]] | None = None, project_path: Path | None = None) -> dict[str, list[PromptVariant]]: offspring: dict[str, list[PromptVariant]] = {} for cat, pop in self.islands.items(): pop.track_best_history() # 早停：连续2代提升<3% → 跳过变异 if pop.is_stagnant(generations=2, improvement_threshold=0.03): self.stagnation_counters[cat] = self.stagnation_counters.get(cat, 0) + 1 print(f" [{cat}] 停滞 {self.stagnation_counters[cat]} 代，跳过变异") offspring[cat] = [] continue else: self.stagnation_counters[cat] = 0 # 灭绝：连续3代提升<5% 且种群≥70%满 if pop.needs_extinction(generations=3, improvement_threshold=0.05): print(f" [{cat}] 触发灭绝-重播") gs = (genre_seeds or {}).get(cat, []) si = (style_injects or {}).get(cat, []) events = pop.apply_extinction(gs, si) if project_path: log_evolution_event(project_path, f"灭绝事件_{cat}", events) offspring[cat] = [] continue # 正常进化 children = [] parents = pop.select_top(3, weights) diverse = pop.select_diverse(2) for p in parents[:2]: children.append(PromptVariant( p.template, source="elite", scores=p.scores.copy() )) for i in range(min(3, len(parents) - 1)): children.append(crossover(parents[i], parents[i + 1])) for p in parents + diverse: if random.random() < mutation_rate: children.append(mutate(p)) for p in diverse: if random.random() < mutation_rate * 0.5: children.append(simplify(p)) for child in children: pop.add(child) offspring[cat] = children pop.generation += generations return offspring def all_islands_stagnant(self) -> bool: if not self.islands: return False return all( pop.is_stagnant(generations=2, improvement_threshold=0.03) for pop in self.islands.values() ) def migrate(self, top_n: int = 2) -> int: migrated = 0 islands = list(self.islands.keys()) if len(islands) < 2: return 0 for i, source_cat in enumerate(islands): target_cat = islands[(i + 1) % len(islands)] elites = self.islands[source_cat].select_top(top_n) for elite in elites: copy = PromptVariant( f"[迁自{source_cat}]{elite.template}", source=f"migrate:{source_cat}", scores={k: v * 0.9 for k, v in elite.scores.items()}, ) if self.islands[target_cat].add(copy): migrated += 1 return migrated def sync_to_evolution(self) -> dict: """同步最优提示词到 skill 级 evolution/prompts/current_best.json""" best = {} for cat, pop in self.islands.items(): top = pop.select_top(3) if len(top) < 3: # 用未评分变体补充到 3 个 unscored = [v for v in pop.variants if v.trials == 0] for v in unscored: if v.template not in [t.template for t in top]: top.append(v) if len(top) >= 3: break best[cat] = [v.template for v in top[:3]] if top else [ v.template for v in pop.variants[:3] ] save_evolution("prompts/current_best.json", best) return best def sync_to_genre(self, genre_id: str) -> dict: """同步最优提示词到 evolution/genres/{genre_id}/prompts/current_best.json""" from utils import save_genre_evolution best = {} for cat, pop in self.islands.items(): top = pop.select_top(3) if len(top) < 3: unscored = [v for v in pop.variants if v.trials == 0] for v in unscored: if v.template not in [t.template for t in top]: top.append(v) if len(top) >= 3: break best[cat] = [v.template for v in top[:3]] if top else [ v.template for v in pop.variants[:3] ] save_genre_evolution(genre_id, "prompts/current_best.json", best) return best # ── 持久化 ───────────────────────────────────────────── def save_model(model: IslandModel, genre_id: str | None = None) -> None: """保存岛屿模型到 population.json（题材级或技能级）""" data = {} for cat, pop in model.islands.items(): data[cat] = { "generation": pop.generation, "variants": [v.to_dict() for v in pop.variants], } if genre_id: from utils import save_genre_evolution save_genre_evolution(genre_id, "prompts/population.json", data) else: save_evolution("prompts/population.json", data) def load_model(genre_id: str | None = None) -> IslandModel: """加载岛屿模型（题材级优先，回退到技能级）""" model = IslandModel() if genre_id: from utils import load_genre_evolution data = load_genre_evolution(genre_id, "prompts/population.json") if data: for cat, pop_data in data.items(): pop = PromptPopulation(cat, max_size=35) pop.generation = pop_data.get("generation", 0) pop.variants = [PromptVariant.from_dict(d) for d in pop_data.get("variants", [])] model.islands[cat] = pop return model # 回退：技能级 data = load_evolution("prompts/population.json") if data: for cat, pop_data in data.items(): pop = PromptPopulation(cat, max_size=35) pop.generation = pop_data.get("generation", 0) pop.variants = [PromptVariant.from_dict(d) for d in pop_data.get("variants", [])] model.islands[cat] = pop else: model.initialize(SEED_TEMPLATES) return model # ── 题材级进化入口 ───────────────────────────────────── def evolve_genre(genre_id: str, project_path: Path | None = None, generations: int = 3, mutation_rate: float = 0.3) -> dict: """题材级共享提示词进化。 同题材所有项目共享进化数据。project_path 可选——传入时 会扫描该项目的风格报告注入到题材池中，实现跨书风格交叉。 """ print(f"遗传 提示词进化 (题材: {genre_id})") from utils import ensure_genre_evolution, find_style_reports ensure_genre_evolution(genre_id) model = load_model(genre_id) # 收集风格报告注入（来自具体项目，但写入题材池） style_injects: dict[str, list[str]] = {cat: [] for cat in model.islands} if project_path: style_reports = find_style_reports(project_path) for report_path in style_reports: injected = inject_from_style_report(report_path) if injected: style_injects.setdefault("writing", []).append(injected.template) if model.islands.get("writing", PromptPopulation("writing")).add(injected): print(f" 风格注入: {report_path.parent.name} → writing 岛（题材池）") # 获取题材种子（供灭绝重播用） from seed_extractor import SeedExtractor extractor = SeedExtractor() genre_seeds = extractor.extract_seeds(genre_id=genre_id) # 加载权重 from utils import load_genre_evolution thresholds = load_genre_evolution(genre_id, "thresholds.json") if not thresholds: thresholds = load_evolution("thresholds.json") or {} weights = thresholds.get("weights", {"P1": 0.35, "P2": 0.25, "P3": 0.15, "P4": 0.15, "P5": 0.10}) total_offspring = 0 # 记录演化前状态用于事件日志 pre_variant_ids = { cat: {v.id for v in pop.variants} for cat, pop in model.islands.items() } for gen in range(generations): if model.all_islands_stagnant(): print(" 所有岛屿停滞，进化收敛") break offspring = model.evolve_all( generations=1, mutation_rate=mutation_rate, weights=weights, genre_seeds=genre_seeds, style_injects=style_injects, project_path=project_path, ) gen_count = sum(len(v) for v in offspring.values()) total_offspring += gen_count if (gen + 1) % 2 == 0: model.migrate(top_n=2) # 写入事件日志（Event Sourcing） post_variant_ids = { cat: {v.id for v in pop.variants} for cat, pop in model.islands.items() } events = [] for cat, pop in model.islands.items(): pre_ids = pre_variant_ids.get(cat, set()) post_ids = post_variant_ids.get(cat, set()) new_ids = post_ids - pre_ids removed_ids = pre_ids - post_ids if new_ids or removed_ids: new_variants = [ v.to_dict() for v in pop.variants if v.id in new_ids ] events.append({ "type": "mutate", "island": cat, "generation": pop.generation, "new_variants": new_variants, "removed_variants": list(removed_ids), }) if events: from event_store import write_events write_events(genre_id, events) save_model(model, genre_id) best = model.sync_to_genre(genre_id) print(f" 产生 {total_offspring} 个新变体") print(f" 题材进化结果已写入 evolution/genres/{genre_id}/") for cat, prompts in best.items(): print(f" [{cat}]: {len(prompts)} 最优提示词") return {"status": "ok", "offspring": total_offspring, "genre": genre_id} # ── CLI ───────────────────────────────────────────────── def evolve_skill(generations: int = 3, mutation_rate: float = 0.3) -> dict: """进化 skill 自身的提示词（不依赖项目）""" print(f"遗传 提示词进化 (skill-level)") model = load_model() thresholds = load_evolution("thresholds.json") or {} weights = thresholds.get("weights", {"P1": 0.35, "P2": 0.25, "P3": 0.15, "P4": 0.15, "P5": 0.10}) print(f" 岛屿数: {len(model.islands)}") for cat, pop in model.islands.items(): scored = [v for v in pop.variants if v.trials > 0] print(f" [{cat}] variants={len(pop.variants)} scored={len(scored)} gen={pop.generation}") total_offspring = 0 for gen in range(generations): if model.all_islands_stagnant(): print(" 所有岛屿停滞，进化收敛") break offspring = model.evolve_all(generations=1, mutation_rate=mutation_rate) gen_count = sum(len(v) for v in offspring.values()) total_offspring += gen_count if (gen + 1) % 2 == 0: model.migrate(top_n=2) save_model(model) best = model.sync_to_evolution() print(f" ✓ 产生 {total_offspring} 个新变体") print(f" ✓ 进化结果已写入 evolution/prompts/") for cat, prompts in best.items(): print(f" [{cat}]: {len(prompts)} 最优提示词") return {"status": "ok", "offspring": total_offspring} if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="提示词进化器") parser.add_argument("action", choices=["evolve", "show"], help="evolve=进化 | show=展示当前最优提示词") parser.add_argument("--project", "-p", type=str, default=None, help="项目路径（自动检测题材）") parser.add_argument("--genre", "-g", type=str, default=None, help="题材 ID（直接指定，跳过自动检测）") parser.add_argument("--generations", type=int, default=3) parser.add_argument("--mutation-rate", "-m", type=float, default=0.3) args = parser.parse_args() if args.action == "evolve": if args.genre: evolve_genre(args.genre, generations=args.generations, mutation_rate=args.mutation_rate) elif args.project: from utils import resolve_genre_from_project genre_id = resolve_genre_from_project(Path(args.project)) evolve_genre(genre_id, project_path=Path(args.project), generations=args.generations, mutation_rate=args.mutation_rate) else: evolve_skill(generations=args.generations, mutation_rate=args.mutation_rate) elif args.action == "show": if args.genre: from utils import load_genre_evolution best = load_genre_evolution(args.genre, "prompts/current_best.json") elif args.project: from utils import resolve_genre_from_project, load_genre_evolution genre_id = resolve_genre_from_project(Path(args.project)) best = load_genre_evolution(genre_id, "prompts/current_best.json") else: best = load_evolution("prompts/current_best.json") if not best: model = IslandModel() model.initialize(SEED_TEMPLATES) model.sync_to_evolution() best = load_evolution("prompts/current_best.json") for cat, prompts in best.items(): print(f"\n## {cat}") for i, p in enumerate(prompts): print(f" {i+1}. {p[:120]}...")