男频网文自动生成框架 — 架构设计

核心设计哲学

不是"让AI写小说",而是"构建一个会讲故事的系统"。

AI直出的本质问题:单次调用无状态、无全局意图、无戏剧意识。 解法:用四层架构将"写小说"拆解为四种不同性质的工作,每层解决一类核心问题。

INPUT: 基础Idea(题材/主角/核心矛盾/金手指)
         ↓
┌─────────────────────────────────┐
│  Layer 1: 世界观引擎              │  ← 一次性初始化,持久化
├─────────────────────────────────┤
│  Layer 2: 剧情导演(沙盒驱动)    │  ← 每章循环运行
├─────────────────────────────────┤
│  Layer 3: 一致性引擎              │  ← 读写贯穿全程
├─────────────────────────────────┤
│  Layer 4: 文本生成与审计          │  ← 最终输出
└─────────────────────────────────┘
         ↓
OUTPUT: 章节文本(每章2000-3000字)

Layer 1:世界观引擎

解决的问题:AI没有稳定世界观基础,写着写着世界规则开始矛盾。

1.1 核心数据结构

WorldBible(世界圣经,持久化JSON/数据库)
├── universe/
│   ├── power_system.md      # 修炼体系(等级名+战力对比+突破条件)
│   ├── geography.md         # 地图/大陆/秘境/换图路线
│   ├── factions.md          # 势力格局(门派/家族/王朝,强弱关系)
│   ├── economy.md           # 资源/货币/稀缺资源分布
│   └── history.md           # 历史背景(埋伏笔的空间)
├── characters/
│   ├── protagonist.json     # 主角完整档案
│   ├── antagonists/         # 反派档案(含动机/能力/秘密)
│   └── supporting/          # 重要配角档案
└── rules/
    ├── forbidden.json       # 禁忌(绿帽/长期虐主等红线)
    └── genre_laws.json      # 类型法则(男频爽感公式参数)

1.2 初始化流程

Step 1: 解析输入Idea → 提取{题材, 主角设定, 金手指, 核心矛盾}
Step 2: 生成世界观草案(多方案)→ 选优/合并
Step 3: 填充WorldBible所有字段
Step 4: 逻辑自洽性检查(境界体系数值平衡/势力关系无矛盾)
Step 5: 生成全书三大高潮节点(大X规划)
Step 6: 生成卷级大纲(每卷核心爽点、主角状态起终点)

1.3 关键设计:主角能力成长曲线

# 提前规划好的能力曲线,防止后期天花板崩塌
growth_curve = {
    "volume_1": {"start_rank": 1, "end_rank": 5, "key_milestone": "进入大宗门"},
    "volume_2": {"start_rank": 5, "end_rank": 12, "key_milestone": "核心弟子"},
    "volume_3": {"start_rank": 12, "end_rank": 25, "key_milestone": "碾压前辈"},
    # ...
}

Layer 2:剧情导演(沙盒驱动)

解决的问题:AI缺"人味"、情节同质化、爽感节奏失控。

2.1 核心创新:双引擎剧情生成

传统做法:AI直接"写下一章发生什么" → 缺乏意图,人味为零。

本框架:上帝导演 × 角色沙盒 = 有意图的真实感

┌──────────────────┐    设定戏剧框架    ┌──────────────────┐
│  上帝导演 (Director) │ ─────────────→ │  角色沙盒 (Sandbox) │
│  知道全局走向       │                  │  角色按自身逻辑行动  │
│  主动安排冲突节点   │ ←───────────── │  涌现出意外行为      │
└──────────────────┘    反馈涌现结果    └──────────────────┘
         ↓
    场景方案(含人物对话/行动/结果)

2.2 上帝导演(Director Agent)

职责:掌管全局意图,保证爽感节奏。

class Director:
    def plan_next_chapter(self, context: ChapterContext) -> SceneDirective:
        """
        返回本章的戏剧指令,不是具体文字。
        """
        return SceneDirective(
            chapter_no=N,
            dramatic_purpose="本章功能",  # 积累压力/爽点释放/伏笔埋设/换图过渡
            shuang_requirement=ShuangLevel,  # 本章必须有什么级别的爽点
            hook_requirement="章末钩子方向",  # 下一章期待感的方向
            conflict_type="人物冲突/资源争夺/外部威胁/内心抉择",
            forbidden_actions=["不能让主角在本章输太惨", "不能推进X线索"],
            character_states_before={...},  # 各角色进入本章的状态
        )

爽点调度器(嵌入Director内):

# 强制执行3X节奏
class ShuangScheduler:
    chapters_since_small_shuang: int  # 必须 <= 3
    chapters_since_medium_shuang: int  # 必须 <= 5
    chapters_since_big_shuang: int    # 必须 <= 15

    def get_required_shuang_level(self) -> ShuangLevel:
        if self.chapters_since_small_shuang >= 3:
            return ShuangLevel.SMALL  # 强制小爽点
        if self.chapters_since_medium_shuang >= 5:
            return ShuangLevel.MEDIUM  # 强制中高潮
        return ShuangLevel.NONE  # 可以是铺垫章

弧管理器(嵌入Director内):

class ArcManager:
    current_arc: int         # 当前处于第几卷
    pressure_accumulated: float  # 压力积累度(满100触发大高潮)
    foreshadows: List[Foreshadow]  # 待回收的伏笔

    def assess_arc_state(self) -> ArcState:
        # 返回:积累期/临界期/高潮期/回落期

2.3 角色沙盒(Character Sandbox)

职责:让角色按自身逻辑行动,产生真实人味。

每个重要角色有独立Agent,包含:

class CharacterAgent:
    identity: CharacterProfile   # 性格/动机/弱点/秘密
    current_state: CharacterState # 当前知识/情绪/目标/位置

    def decide_action(self, scene: SceneDirective, world_state: WorldState) -> CharacterAction:
        """
        在Director的框架约束内,按自己的性格和动机做决策。
        注意:角色不知道全局剧情,只知道自己知道的。
        """

关键约束:角色Agent的上下文仅包含"该角色应当知道的信息",不包含全局剧情。这保证了角色信息不对称的真实感(反派不知道主角已突破,盟友有自己的小算盘)。

2.4 人物弧光系统(CharacterArcSystem)

核心矛盾:严格固化人物设定 → 一致性好但人物死板;允许随意演变 → 有人味但性格飘移。 解法:把 CharacterProfile 拆成不可变核心 + 可演变弧光维度,演变须通过弧光闸门

2.4.1 双层人物结构

@dataclass
class CharacterCore:
    """永不改变。角色的 DNA。"""
    name: str
    archetype: str           # 底层原型,如"孤傲天才"/"草根逆袭"
    nine_type: int           # 九型人格(1-9,决定压力下的本能反应)
    voice_pattern: str       # 口头禅/说话节奏/惯用句式(文风层面)
    fundamental_desire: str  # 最底层渴望(百万字才可能变一次,慎用)


@dataclass
class ArcTrait:
    """单个可演变的性格维度。"""
    name: str               # 如 "对他人的信任度" / "对权力的执念" / "孤独感"
    value: float            # 当前值,范围 -1.0(极端负极)到 +1.0(极端正极)
    history: list[tuple[int, float]]  # [(chapter_no, value), ...] 变化轨迹
    planned_direction: int  # +1 或 -1:这个维度在全书中大方向是升还是降
    planned_endpoint: float # 本卷结束时这个维度应到达的目标值


@dataclass
class CharacterProfile:
    """运行时实际使用的人物档案,版本化存储。"""
    core: CharacterCore                  # 不可变
    arc_traits: dict[str, ArcTrait]      # 可演变维度
    current_belief: str                  # 当前"谎言/错误信念"(可随arc改变)
    current_surface_goal: str            # 当前表面渴望(按剧情段落更新)
    significant_experiences: list[str]   # 重大经历摘要(影响弧光判断的依据)
    version: int                         # 每次弧光变化 +1,便于追溯

预设弧光维度(主角,初始化时填写)

PROTAGONIST_ARC_DIMENSIONS = [
    # (维度名, 初始值, 计划方向, 卷末目标)
    ("对他人的信任度",  +0.6, -1, +0.2),  # 第一卷被背叛 → 下降
    ("对权力的执念",    0.0,  +1, +0.5),   # 随实力增长执念加深
    ("孤傲程度",        +0.8, -1, +0.3),   # 经历伙伴牺牲后磨去一些棱角
    ("对弱者的同情",    +0.4, +1, +0.8),   # 越来越有担当(需要感为主题)
]
# 类型 + 主题决定具体维度,上方只是玄幻类示例。

2.4.2 弧光演变闸门(ArcEvolutionGate)

演变必须满足所有约束才能写入:

class ArcEvolutionGate:

    # ── 频率约束 ──────────────────────────────────────────────────
    MIN_CHAPTERS_BETWEEN_CHANGES = 20   # 同一维度两次变化至少间隔20章
    MAX_TRAIT_CHANGES_PER_VOLUME = 2    # 每卷最多允许2个维度发生显著变化

    # ── 幅度约束(按触发事件类型决定最大delta)────────────────────
    TRIGGER_MAX_DELTA = {
        "顿悟/体悟":         0.08,   # 感悟型,细水长流
        "重大失败/惨败":      0.10,
        "重要盟友/导师死亡":  0.12,
        "代价沉重的胜利":     0.10,
        "背叛(亲近者)":     0.15,  # 最强烈的单次冲击
        "长期压迫后的爆发":   0.12,
    }
    # 单次变化绝对上限,无论触发类型
    HARD_MAX_DELTA = 0.20

    # ── 方向约束 ──────────────────────────────────────────────────
    # 变化方向必须和 planned_direction 同向(±5%容差允许轻微反复)
    # 防止剧情写着写着人物弧光来回横跳

    @staticmethod
    def validate(
        profile: CharacterProfile,
        trait_name: str,
        proposed_delta: float,
        trigger_type: str,
        chapter_no: int,
    ) -> tuple[bool, str]:
        """
        返回 (approved: bool, reason: str)。
        """
        trait = profile.arc_traits.get(trait_name)
        if not trait:
            return False, f"维度 {trait_name} 不存在于该角色"

        # 1. 频率检查
        last_change_ch = max((ch for ch, _ in trait.history), default=0)
        if chapter_no - last_change_ch < ArcEvolutionGate.MIN_CHAPTERS_BETWEEN_CHANGES:
            return False, f"距上次变化仅{chapter_no - last_change_ch}章,未达最小间隔"

        # 2. 幅度检查
        max_allowed = ArcEvolutionGate.TRIGGER_MAX_DELTA.get(trigger_type, 0.08)
        if abs(proposed_delta) > min(max_allowed, ArcEvolutionGate.HARD_MAX_DELTA):
            return False, f"变化幅度{proposed_delta:.2f}超出{trigger_type}允许的{max_allowed:.2f}"

        # 3. 方向检查(允许轻微反复,但大方向必须正确)
        if proposed_delta * trait.planned_direction < -0.05:
            return False, f"变化方向与规划弧光方向相反(规划:{trait.planned_direction:+d})"

        # 4. 卷内变化次数检查(由 ArcManager 传入当前卷已变化次数)
        return True, "approved"

    @staticmethod
    def apply(profile: CharacterProfile, trait_name: str,
              delta: float, chapter_no: int) -> CharacterProfile:
        """通过闸门后,实际更新弧光,返回新版本 profile。"""
        trait = profile.arc_traits[trait_name]
        new_value = max(-1.0, min(1.0, trait.value + delta))
        trait.history.append((chapter_no, new_value))
        trait.value = new_value
        profile.version += 1
        return profile

2.4.3 弧光里程碑规划(初始化时由 Director 设定)

@dataclass
class ArcMilestone:
    """全书预规划的弧光关键节点,Director 把它当作剧情目标之一。"""
    at_volume_end: int          # 第几卷末实现
    trait_name: str
    target_value: float         # 达到此值
    required_trigger_type: str  # 什么类型的事件触发它
    narrative_expression: str   # 文本层面如何表现("他第一次对陌生人伸出手")

# Director 规划章节时,检查:当前章是否是触发某个里程碑的合适时机?
# 是 → 在 SceneDirective 中加入 arc_evolution_hint,TextGenerator 据此写出相应细节

2.4.4 OOC 判断标准更新

人物行为合理性判断依据从"原始固定设定"改为"当前弧光状态":

def is_character_ooc(action: CharacterAction, profile: CharacterProfile) -> bool:
    """
    OOC 检查:行为是否与「当前版本」的 arc_traits 明显矛盾。
    注意:profile.version 会随弧光演变递增,所以"一致性"是动态的。

    例:第100章主角 trust_level 已从 +0.6 降到 +0.1(经历过背叛)
         此时行为"主动向陌生人托付秘密"→ OOC
         但第1章同样行为 → 正常
    """
    current_trust   = profile.arc_traits["对他人的信任度"].value
    current_arrogance = profile.arc_traits["孤傲程度"].value
    # ... 基于当前 arc_trait 值,而非初始值,进行合理性判断

2.5 场景生成流程

每章执行流程:

1. Director.plan_next_chapter()
   → 产出 SceneDirective(本章意图/爽点要求/禁忌)

2. 角色沙盒运行:
   - 注入 SceneDirective 作为"场景背景"
   - 各角色 Agent 依次行动/反应
   - 涌现出:对话草稿/行动决策/冲突走向

3. Director.review_emergence()
   - 检查涌现结果是否满足爽点要求
   - 若不满足:调整场景参数,重新运行沙盒(最多3次)
   - 若满足:输出场景方案

4. 产出 ScenePlan(送往Layer 4):
   {
     "plot_summary": "本章情节摘要",
     "key_events": [...],
     "dialogue_seeds": [...],
     "shuang_moment": "具体爽点位置和内容",
     "chapter_hook": "章末钩子",
     "state_changes": {...}  # 哪些状态会变化
   }

跨层基础设施:详略规划器(DetailPacingPlanner)

解决的问题:角色沙盒涌现机制倾向持续产出戏剧事件,导致两个问题: 1. 事件密集失真:主角每天都遭遇危机,读者失去真实感和情感投资 2. 高潮后缺缓冲:大戏剧冲突后立即进入下一个高压段,读者情绪疲劳

解法:在 Director 之上增加一层节奏调度器,主动规划"时间跳跃"和"缓冲章",管理全书的密度节律。


DP1:章节类型体系

class ChapterType(Enum):
    DRAMATIC     = "dramatic"    # 高强度戏剧冲突(打脸/战斗/揭秘/大逆转)
    TENSION      = "tension"     # 中强度张力(谋划/铺垫/威胁渐近)
    NORMAL       = "normal"      # 标准推进(普通情节单元)
    BUFFER       = "buffer"      # 缓冲章(高潮后喘息,有内容但无大冲突)
    SLICE        = "slice"       # 生活流章节(日常有意义的片段)
    TRAINING     = "training"    # 成长蓄力(修炼/准备/积累)
    TIMESKIP     = "timeskip"    # 时间跳跃摘要(跨越无戏剧价值的时段)


# 各类型的叙事密度分值(0.0=极静,1.0=极紧张)
CHAPTER_DENSITY = {
    ChapterType.DRAMATIC:  0.90,
    ChapterType.TENSION:   0.65,
    ChapterType.NORMAL:    0.45,
    ChapterType.BUFFER:    0.20,
    ChapterType.SLICE:     0.15,
    ChapterType.TRAINING:  0.30,
    ChapterType.TIMESKIP:  0.00,  # 不计入密度窗口(不是真正的"章")
}


@dataclass
class ChapterPacingPlan:
    """每章在规划阶段由 DetailPacingPlanner 填写,传给 Director 作为约束。"""
    chapter_no: int
    chapter_type: ChapterType
    density_score: float

    # TIMESKIP 专用
    skip_story_time_span: tuple[StoryDate, StoryDate] | None  # 跳过的故事时段
    skip_summary_hint: str | None    # 提示 Director 此段时间内发生了什么(WorldStateDB 需更新)
    skip_key_state_changes: dict     # 跳过期间的重要状态变化

    # BUFFER / SLICE 专用
    buffer_focus: str | None         # 见 DP3 缓冲章内容规范
    allowed_shuang_types: list[str]  # 缓冲章允许的爽点类型(情感/成长,不允许战力大爽)

    # 故事时间跨度(所有类型)
    story_time_start: StoryDate
    story_time_end: StoryDate        # 一章可以覆盖几小时到几天的故事时间

DP2:密度追踪与触发规则

class DetailPacingPlanner:
    """
    在 ArcManager 规划每卷章节序列时调用,
    对 Director 输出的初始章节序列进行密度审计和修正。
    """

    # ── 滚动密度窗口 ───────────────────────────────────────────────
    DENSITY_WINDOW   = 10   # 以最近10章的平均密度作为健康指标
    TARGET_AVG       = 0.50 # 理想10章均值(过高=疲劳,过低=注水)
    HIGH_DENSITY_CAP = 0.70 # 滚动均值不得持续超过此值
    LOW_DENSITY_FLOOR= 0.30 # 滚动均值不得持续低于此值(纯水)

    # ── 高潮后强制缓冲 ─────────────────────────────────────────────
    POST_BIG_CLIMAX_BUFFER  = (2, 4)   # BIG 爽点后,强制插入2-4章缓冲/生活流(RhythmJitter采样)
    POST_MED_CLIMAX_BUFFER  = (1, 2)   # MEDIUM 爽点后,1-2章缓冲
    BUFFER_BEFORE_NEXT_HIGH = 3        # 缓冲结束后至少3章 NORMAL 才能进入下一个 DRAMATIC

    # ── 连续高密度上限 ─────────────────────────────────────────────
    MAX_CONSECUTIVE_DRAMATIC = 5       # 连续5章 DRAMATIC/TENSION → 强制插入 BUFFER 或 SLICE
    MAX_CONSECUTIVE_LOW      = 4       # 连续4章 BUFFER/SLICE/TRAINING → 强制回升至 NORMAL

    # ── 时间跳跃触发条件 ──────────────────────────────────────────
    TIMESKIP_TRIGGER = {
        "story_days_without_value": 30,  # 故事时间超过30天没有戏剧价值 → 考虑跳跃
        "min_skip_story_days": 7,        # 跳跃最短跨度(少于7天不值得跳)
        "max_skip_story_days": 365,      # 单次跳跃最长1年故事时间
        "worldstate_update_required": True,  # 跳跃期间 WorldStateDB 仍需更新状态
    }

    def audit_and_patch(self, chapter_sequence: list[ChapterPacingPlan]) -> list[ChapterPacingPlan]:
        """
        审计 Director 规划的章节序列,自动插入缓冲章/时间跳跃,修正过密或过稀区段。
        返回修正后的序列(章节总数可能增加)。
        """
        patched = []
        consecutive_high = 0
        consecutive_low  = 0

        for i, ch in enumerate(chapter_sequence):
            density = CHAPTER_DENSITY[ch.chapter_type]

            # 连续高密度过载 → 插缓冲
            if density >= CHAPTER_DENSITY[ChapterType.TENSION]:
                consecutive_high += 1
                consecutive_low   = 0
            else:
                consecutive_high  = 0
                consecutive_low  += 1

            if consecutive_high > self.MAX_CONSECUTIVE_DRAMATIC:
                buffer = self._generate_buffer_chapter(ch, reason="high_density_overload")
                patched.append(buffer)
                consecutive_high = 0

            # 连续低密度过稀 → 拉回节奏
            if consecutive_low > self.MAX_CONSECUTIVE_LOW:
                ch.chapter_type  = ChapterType.NORMAL
                ch.density_score = CHAPTER_DENSITY[ChapterType.NORMAL]
                consecutive_low  = 0

            patched.append(ch)

        return patched

    def inject_post_climax_buffer(
        self, after_chapter: int, climax_level: str
    ) -> list[ChapterPacingPlan]:
        """BIG/MEDIUM 爽点后强制插入缓冲序列。"""
        count_range = self.POST_BIG_CLIMAX_BUFFER if climax_level == "BIG" \
                      else self.POST_MED_CLIMAX_BUFFER
        count = RhythmJitter.sample(
            base=(count_range[0] + count_range[1]) // 2,
            sigma=0.5, lo=count_range[0], hi=count_range[1]
        )
        buffers = []
        focuses = self._select_buffer_focuses(count)  # 不同缓冲章有不同焦点
        for k, focus in enumerate(focuses):
            buffers.append(ChapterPacingPlan(
                chapter_no=after_chapter + k + 1,
                chapter_type=ChapterType.BUFFER,
                density_score=CHAPTER_DENSITY[ChapterType.BUFFER],
                buffer_focus=focus,
                allowed_shuang_types=["emotional_shuang", "relationship_warmth"],
            ))
        return buffers

    def plan_timeskip(
        self, from_date: StoryDate, to_date: StoryDate, context: str
    ) -> ChapterPacingPlan:
        """生成一章时间跳跃计划。"""
        return ChapterPacingPlan(
            chapter_type=ChapterType.TIMESKIP,
            skip_story_time_span=(from_date, to_date),
            skip_summary_hint=f"从{from_date}到{to_date},{context}",
            skip_key_state_changes={},  # 由 Director 填写,后写入 WorldStateDB
        )

    def _select_buffer_focuses(self, count: int) -> list[str]:
        """选择多个缓冲章的内容焦点,确保多样性。"""
        pool = ["emotional_recovery", "relationship_deepening",
                "world_breathing", "training_montage", "humor_slice"]
        return random.sample(pool, min(count, len(pool)))

DP3:缓冲章内容规范

缓冲章不是"废章",有各自的叙事功能:

BUFFER_CHAPTER_SPECS = {

    "emotional_recovery": {
        "desc": "高潮后的情绪落地。主角/盟友处理战斗/挫折的心理余震。",
        "forbidden": ["新的重大冲突", "新反派登场", "战力大爽点"],
        "required": ["身体感受细节(疲惫/伤痛/放松)", "至少一句真实的内心独白"],
        "shuang_type": "emotional_shuang",
        "example": "大战后主角独坐废墟,想到的第一件事不是胜利,而是某人说过的一句话",
    },

    "relationship_deepening": {
        "desc": "加深主角与重要盟友/情感线角色的情感联结。",
        "forbidden": ["主要矛盾推进", "新威胁引入"],
        "required": ["实质性对话(推进关系而非填充)", "一个令读者记住的细节或玩笑"],
        "shuang_type": "relationship_warmth",
        "example": "主角和老搭档因为一件小事争论,最后莫名其妙都笑了",
        "arc_value": "缓冲章的情感积累 → 未来高光场景的情感放大器",
    },

    "world_breathing": {
        "desc": "展示世界的日常运转,建立沉浸感。主角不在中心,世界自行呼吸。",
        "forbidden": ["主角解决重大问题"],
        "required": ["有趣的世界细节(非说教)", "普通 NPC 的视角片段"],
        "shuang_type": "world_immersion",
        "example": "集市的喧嚣,一个小摊贩和顾客的争论,道出了这个世界的某条潜规则",
        "arc_value": "当世界有厚度,后续大事件才有重量",
    },

    "training_montage": {
        "desc": "压缩展示主角成长/修炼过程,跨越较长故事时间。",
        "forbidden": ["逐日详写(太细 = 注水)", "停滞感(必须有可见进步)"],
        "required": ["清晰的起止对比(前后能力差异)", "一个标志性的领悟瞬间"],
        "shuang_type": "growth_shuang",
        "story_time_coverage": "7天-3个月故事时间",  # 压缩较长时段
        "example": "三个月,他把那套剑法练到了第七式……直到某天黎明他突然明白了第八式的意思",
    },

    "humor_slice": {
        "desc": "轻松喜剧性章节,调节全书情绪基调,防止过于沉重。",
        "forbidden": ["强行滑稽(硬喜剧)", "与全书气质严重不符"],
        "required": ["至少一个真实可笑的情节", "笑点来自角色性格而非外部安排"],
        "shuang_type": "humor_shuang",
        "frequency": "全书不超过8%章节",  # 太多失去新鲜感
        "example": "主角用绝世战技完成了一件极其日常的蠢事,围观者目瞪口呆",
    },
}

DP4:时间跳跃的世界状态更新协议

跳跃不是"什么都没发生",WorldStateDB 仍需记录跳跃期间的变化:

class TimeskipStateUpdater:
    """
    在时间跳跃章节生成后,向 WorldStateDB 注入跳跃期间的状态变化。
    变化内容由 Director 推断,不需要逐章细写。
    """

    TIMESKIP_STATE_TEMPLATE = """
以下是从 {from_date} 到 {to_date} 这段时间内需要更新的世界状态:
请根据当前 WorldBible 和角色设定,推断合理的变化并填写以下字段:

{{
  "power_level_changes": {{          // 主角/NPC 实力变化
    "protagonist": {{"delta_rank": 2, "reason": "闭关修炼"}},
    ...
  }},
  "relationship_changes": {{         // 角色关系变化
    "ally_trust_delta": +0.1,
    "reason": "共同渡过平静期,信任加深"
  }},
  "world_event_summaries": [         // 世界层面发生的事(势力变化/季节/大事件)
    "北方霜宗派出使者,与青云门会谈破裂"
  ],
  "foreshadow_developments": [       // 暗中推进的伏笔(无需写到正文)
    "反派在此期间完成了某个准备动作"
  ],
  "protagonist_daily_life_note": ""  // 一句话:此段时间主角的主要状态
}}
"""

    def update(self, skip_plan: ChapterPacingPlan, world_state: WorldStateDB):
        """调用 Director LLM 推断跳跃期间变化,写入 WorldStateDB。"""
        prompt = self.TIMESKIP_STATE_TEMPLATE.format(
            from_date=skip_plan.skip_story_time_span[0],
            to_date=skip_plan.skip_story_time_span[1],
        )
        changes = llm_call(prompt, schema=TIMESKIP_CHANGE_SCHEMA)
        world_state.apply_timeskip_changes(changes)

DP5:与 ShuangScheduler 的协调

缓冲章/时间跳跃章不计入爽点间隔计数,但不能无限期推迟:

# 扩充到 R2 爽点节奏规则
SHUANG_SCHEDULER_PACING_PATCH = {

    # 缓冲章的爽点豁免
    "buffer_exempt_from_shuang_counter": True,
    # 即:缓冲章不累积"距上次爽点的章数"计数器
    # 但豁免有上限:
    "max_exempt_chapters_consecutive": 3,
    # 连续超过3章豁免(即使是 BUFFER)→ 计数器恢复累积
    # 防止系统借"缓冲章"名义无限拖延爽点

    # 缓冲章允许的替代爽点(不触发 ShuangScheduler 强制,但鼓励)
    "buffer_shuang_substitutes": {
        "emotional_recovery":     "emotional_shuang",    # 情感释放
        "relationship_deepening": "relationship_warmth", # 关系进展
        "training_montage":       "growth_shuang",       # 成长满足
        "humor_slice":            "humor_shuang",        # 笑点释放
        "world_breathing":        None,                  # 可以无爽点(纯沉浸)
    },

    # 时间跳跃章的处理
    "timeskip_resets_counter": False,  # 时间跳跃不重置计数器(时间流逝=积累压力)
    "timeskip_adds_to_counter": True,  # 反而:每次跳跃后读者期待值升高
    "timeskip_counter_bonus": 1,       # 跳跃等效于计数器+1(加快下次爽点触发)
}

DP6:补充 ContradictionChecker 规则

PACING_CONSISTENCY_CHECKS = [
    {
        "name": "BufferAfterBigClimaxCheck",
        "desc": "BIG 爽点后是否有至少2章缓冲(直接接下一个 DRAMATIC = 情绪疲劳)",
        "severity": "MUST",
    },
    {
        "name": "DensityRollingAvgCheck",
        "desc": "最近10章密度均值是否在 [0.30, 0.70] 区间内",
        "severity": "SHOULD",
        "action": "超出时向 Director 注入节奏调整建议",
    },
    {
        "name": "TimeskipStateConsistencyCheck",
        "desc": "时间跳跃后,角色状态是否与跳跃期间的推断变化一致",
        "severity": "MUST",
    },
    {
        "name": "SlicePurposeCheck",
        "desc": "SLICE/BUFFER 章节是否有实质内容(非空洞注水)",
        "severity": "SHOULD",
        "check": "buffer_focus 字段是否对应了 DP3 规范中的 required 项",
    },
]

跨层基础设施:冰山世界模拟器(IcebergWorldSimulator)

设计意图:主角视角只是世界的 1/8。同一时间轴上,其他势力/人物/阴谋正在独立推进。这些"幕后实体"提前创建在 WorldBible 中、随剧情同步演化,只有只言片语渗透到主角视野。读者感知到世界在自行运转,而非"只有主角时世界才有意义"。

类比:主角在村子里打架,与此同时帝都在政变、北境在入侵、某古老传承在苏醒——读者通过商队口中的流言、城门告示、路过的逃难者隐约感知这一切,直到某卷主角撞进这些事件的核心。


IW1:幕后实体(OffscreenEntity)

class EntityType(Enum):
    FACTION      = "faction"     # 势力(宗门/王朝/商会/地下组织)
    POWERHOUSE   = "powerhouse"  # 个体强者(隐世高手/未来反派/命运交叉点)
    CONSPIRACY   = "conspiracy"  # 阴谋线(某个正在推进的计划)
    NATURAL_FORCE= "natural"     # 自然力(秘境开启/天劫降临/古阵复苏)
    RIVAL_LINE   = "rival"       # 平行主角线(同级别竞争者,未来对手)


@dataclass
class OffscreenEntity:
    entity_id: str
    entity_type: EntityType
    name: str

    # ── 目标与议程 ──────────────────────────────────────────────
    current_agenda: str          # 此刻正在做什么(随卷更新)
    long_term_goal: str          # 终极图谋(WorldBible 初始化时设定,极少改变)
    power_trajectory: list[str]  # 各卷末预计实力标签("初期潜伏→中期成型→末期爆发")

    # ── 状态 ────────────────────────────────────────────────────
    current_state: dict          # {location, power_level, resources, allies, secrets}
    secret_actions: list[str]    # 正在秘密进行的行动(读者永远看不到全貌,只看到痕迹)
    known_to_world: str          # 世界公众对这个实体的认知(可以与实际状态完全不同)

    # ── 与主角的交集规划 ─────────────────────────────────────────
    planned_intersection_arc: int     # 第几卷与主角正面碰撞
    intersection_trigger: str         # 什么事件让两条线相交
    pre_intersection_hint_arcs: list[int]  # 哪几卷应该让主角隐约感知到它的存在

    # ── 暗示配额 ────────────────────────────────────────────────
    hint_budget_per_arc: int = 2      # 每卷最多向主角视野渗透几次(通常1-3)
    hint_min_interval_chapters: int = 15  # 同一实体两次暗示间隔最少章数

IW2:世界事件流(WorldEventStream)

幕后实体的每个行动产生 WorldEvent,存入事件流:

@dataclass
class WorldEvent:
    event_id: str
    story_time: StoryDate
    entity_id: str              # 哪个幕后实体引发

    # ── 双层描述(冰山结构) ──────────────────────────────────────
    actual_description: str     # 实际发生了什么(只存入 WorldStateDB,读者不直接看到)
    rumor_variants: list[str]   # 失真版本:流言/以讹传讹/片面目击(用于暗示)
    # 例:actual="血魔教屠灭青云城分部,夺取封印残片"
    #     rumors=["青云城死了不少人,说是山匪","有人说是宗门内斗","城门三天没开"]

    importance: float           # 0.0-1.0,决定暗示频率和力度
    secrecy: float              # 0.0-1.0,越高越难被主角直接察觉(0=公开新闻,1=绝密)

    # ── MemoryGraph 集成 ───────────────────────────────────────
    foreshadow_node_id: str     # 对应 MemoryGraph 中的 FORESHADOW 节点
    # 当主角最终碰撞这条线时,ContradictionChecker 可回溯验证暗示的一致性


class BackgroundWorldSimulator:

    entities: dict[str, OffscreenEntity]
    event_stream: list[WorldEvent]           # 按故事时间排序

    def advance_arc(self, arc_no: int, world_state: WorldStateDB) -> list[WorldEvent]:
        """
        每卷开始时调用。让幕后实体推进各自的议程,产生新 WorldEvent。
        实体状态更新写入 WorldStateDB(即使主角不知道)。
        """
        new_events = []
        for entity in self.entities.values():
            if arc_no < entity.planned_intersection_arc:
                # 尚未与主角交集:独立演化
                event = self._simulate_entity_arc_progress(entity, arc_no)
                entity.current_state = self._update_entity_state(entity, arc_no)
                world_state.record_background_event(event)
                new_events.append(event)
            # 交集弧:实体进入主线,不再由此模块管理
        return new_events

    def _simulate_entity_arc_progress(
        self, entity: OffscreenEntity, arc_no: int
    ) -> WorldEvent:
        """
        调用 Director LLM 推断实体本卷的行动,生成 WorldEvent。
        Prompt 包含:实体档案 + 当前卷世界背景 + 上一卷实体状态。
        """
        prompt = ENTITY_PROGRESS_PROMPT.format(entity=entity, arc_no=arc_no)
        raw = llm_call(prompt, schema=WORLD_EVENT_SCHEMA)
        return WorldEvent(**raw)

IW3:暗示机制(HintInjector)

控制幕后信息向主角视野渗透的频率、方式和力度:

class HintType(Enum):
    RUMOR       = "rumor"        # NPC 闲聊/商队消息/酒馆八卦
    AFTERMATH   = "aftermath"    # 主角遭遇事件后果(难民/废墟/死尸/气息残留)
    NPC_MENTION = "npc_mention"  # 见多识广的 NPC 一句话提及
    ITEM_REF    = "item_ref"     # 物品来历暗示("这是北境某地的货色……")
    ENVIRON     = "environ"      # 环境征兆(天色/灵气异动/动物异常)
    OFFICIAL    = "official"     # 世界内的"新闻媒介"(通缉令/宗门公告/悬赏榜)


@dataclass
class HintOpportunity:
    world_event_id: str
    hint_type: HintType
    hint_content: str            # 实际渗透给读者的文字(模糊/片面)
    delivery_method: str         # 谁/什么载体传递这个暗示
    fit_score: float             # 与当前场景的自然契合度(0-1)
    obscurity: float             # 暗示的模糊程度(0=明示,1=只有回头看才懂)


class HintInjector:

    MAX_HINTS_PER_CHAPTER   = 2    # 每章最多注入2个暗示(通常0-1个)
    MIN_HINT_INTERVAL       = 15   # 同一事件两次暗示间隔章数
    MIN_OBSCURITY           = 0.4  # 交集前的暗示不能太明显(至少40%模糊度)

    def select_hints(
        self,
        scene: ScenePlan,
        current_chapter: int,
        pending_events: list[WorldEvent],
    ) -> list[HintOpportunity]:
        """
        从待渗透事件中挑选最适合当前场景的 0-2 个暗示。

        选择原则:
          1. 优先将要与主角交集的实体(距离 intersection_arc 越近,暗示越频繁)
          2. 优先与当前场景地点/角色有关联的实体
          3. 暗示多样性:同一章不用同一种 HintType
          4. 保持模糊度(交集前读者不应能拼出完整画面)
        """
        candidates = []
        for event in pending_events:
            entity = self.entities[event.entity_id]

            # 冷却检查
            last_hint_ch = self._last_hint_chapter(event.event_id)
            if current_chapter - last_hint_ch < self.MIN_HINT_INTERVAL:
                continue

            # 交集临近度权重(交集前1-2卷显著增加频率)
            arcs_to_intersection = entity.planned_intersection_arc - self._current_arc()
            urgency = 1.0 / max(arcs_to_intersection, 0.5)

            # 场景契合度
            fit = self._compute_scene_fit(event, scene)

            candidates.append((event, urgency * fit * event.importance))

        candidates.sort(key=lambda x: x[1], reverse=True)
        selected = candidates[:self.MAX_HINTS_PER_CHAPTER]
        return [self._craft_hint(event, scene) for event, _ in selected]

    def _craft_hint(self, event: WorldEvent, scene: ScenePlan) -> HintOpportunity:
        """
        选择最自然的暗示类型,从 event.rumor_variants 中挑选最合适的版本,
        或调用 LLM 生成新的失真版本。
        失真原则:细节错误但情绪正确(读者感受到"有大事"但不知道具体是什么)。
        """
        ...

IW4:与 TextGenerator 的集成

暗示注入到生成 Prompt,但必须"藏"进正常叙事:

HINT_INJECTION_PROMPT_ADDITION = """
【冰山世界暗示(选用 {hint_count} 个,自然融入场景,不可生硬插入)】
{hints}

暗示融入要求:
  - 每个暗示最多占 1-3 句话,不展开,不解释
  - 通过对话的"只言片语"或环境描写传递,主角不必在意
  - 禁止让主角对暗示做过多反应(他/她现在有自己的事)
  - 暗示的细节可以与实际事件有出入(这是流言的本质)

示例融入方式:
  ✗ 生硬:"主角听说北方发生了大事,思考了很久。"
  ✓ 自然:(在酒馆取餐时)旁桌的商人说,"……北边又封城了,说是闹山匪,
            我看哪,哪是什么山匪……" 主角没搭理,端着汤回到位置。
"""

class TextGenerator:
    def generate_chapter(self, directive: SceneDirective) -> str:
        # ...常规检索...

        # 冰山暗示注入
        hints = self.hint_injector.select_hints(
            scene=directive.scene_plan,
            current_chapter=directive.chapter_no,
            pending_events=self.background_sim.get_pending_hint_events(),
        )
        if hints:
            directive.extra_constraints.append(
                HINT_INJECTION_PROMPT_ADDITION.format(
                    hint_count=len(hints),
                    hints="\n".join(f"- {h.hint_content}" for h in hints),
                )
            )

        return self.llm.generate(self.prompt_builder.build(directive))

IW5:MemoryGraph 中的背景事件存储

# 幕后实体档案 → WORLD_FACT 节点(高 importance,永不软删除)
entity_node = MemoryNode(
    node_type=NodeType.WORLD_FACT,
    content=f"[幕后] {entity.name}:{entity.long_term_goal},当前行动:{entity.current_agenda}",
    importance=0.9,
    metadata={"type": "offscreen_entity", "entity_id": entity.entity_id,
               "planned_intersection_arc": entity.planned_intersection_arc},
)

# 每次暗示注入 → EVENT 节点,建立 FORESHADOWS 边
hint_node = MemoryNode(
    node_type=NodeType.EVENT,
    content=f"第{chapter_no}章暗示:{hint.hint_content}(实际指向:{world_event.actual_description})",
    importance=0.5,
    metadata={"type": "iceberg_hint", "entity_id": entity.entity_id,
               "obscurity": hint.obscurity},
)
# FORESHADOWS 边:hint_node → entity_node(暗示指向幕后实体)
# 当主角碰撞这条线时:ContradictionChecker 回溯所有 hint_node,
# 确保过去的暗示与现在正面展开的事实不矛盾(失真版本可以错,核心情绪要一致)

# 主角碰撞时:FORESHADOW 状态改为 resolved,建立 RESOLVES 边

IW6:交集规划与碰撞一致性

幕后实体进入主线时,碰撞质量取决于前置铺垫的密度:

INTERSECTION_QUALITY_RULES = {
    # 最小铺垫要求
    "min_hints_before_intersection": 3,   # 正式碰撞前至少3次暗示(低于此 = 突然冒出来)
    "min_arcs_of_background_existence": 1, # 至少存在1卷再与主角交集(不能当场创建当场碰撞)

    # 碰撞时的叙事处理
    "callback_required": True,  # 碰撞时必须呼应之前至少1次暗示("就是酒馆那个商人说的……")
    "power_credibility": True,  # 实体的实力必须与铺垫中的体量感一致

    # 碰撞后的读者期待管理
    "reader_retrospection_value": True,  # 读者应在碰撞后有"原来如此"的满足感
    # ContradictionChecker 验证:回溯所有该实体的 hint_node,
    # 检查暗示的"情绪方向"是否与碰撞事实一致(不要求细节精确,要求感觉对)
}

Layer 3:一致性引擎

解决的问题:超20万字后时间线模糊、前后矛盾、人物性格飘移。

3.1 两大核心组件

A. 世界状态数据库(WorldStateDB)

WorldStateDB(每章更新)
├── timeline/
│   └── events.jsonl         # 按章节顺序的事件流水账
├── characters/
│   └── {name}_state.json    # 每个角色的当前状态快照
│       ├── location         # 当前位置
│       ├── power_level      # 当前境界
│       ├── known_facts      # 已知信息(防止"角色忘记了某事")
│       ├── goals            # 当前目标
│       ├── relationships    # 与其他角色的关系值
│       └── items            # 持有物品/法宝
├── world/
│   ├── available_secrets.json  # 哪些秘密已揭露/哪些还未揭露
│   └── active_foreshadows.json # 已埋设但未回收的伏笔
└── chapter_registry/
    └── {chapter_N}.json     # 每章的关键事实摘要

B. 上下文注入器(ContextInjector)

⚠️ 下方为早期 naive 草案,已由"跨层基础设施:记忆图谱与上下文检索"(M1-M7 节)完整替代。 实际实现请以 ContextRetriever + ContextBudgetManager 为准。 此处仅保留接口形状,供理解调用链。

class ContextInjector:
    def build_context(self, chapter_no: int, scene_plan: ScenePlan) -> GenerationContext:
        # 委托给 ContextRetriever,不再做 naive 的 last_N_chapters 截取
        query = RetrievalQuery.from_scene_plan(scene_plan, chapter_no)
        retrieved = self.retriever.retrieve(query, budget=TEXT_GEN_BUDGET)
        return GenerationContext.from_retrieved(retrieved)

3.2 写后更新(Post-Write Update)

每章写完后自动执行:

def post_chapter_update(chapter_text: str, scene_plan: ScenePlan):
    # 1. 提取章节事实(LLM辅助)
    facts = extract_facts(chapter_text)
    # 2. 更新角色状态
    for char, changes in facts.state_changes.items():
        state_db.update(char, changes)
    # 3. 标记已回收的伏笔
    for foreshadow in facts.resolved_foreshadows:
        foreshadow_db.mark_resolved(foreshadow)
    # 4. 注册新伏笔
    for new_f in facts.new_foreshadows:
        foreshadow_db.register(new_f)
    # 5. 更新时间线
    timeline.append(ChapterSummary(chapter_no, facts.key_events))

3.3 一致性检查器(ContradictionChecker)

在Layer 4生成文本后、输出前运行:

class ContradictionChecker:
    checks = [
        PowerLevelCheck,        # 主角境界是否和状态DB一致
        LocationCheck,          # 角色位置是否合理
        KnowledgeCheck,         # 角色是否"知道不该知道的事"
        ForeshadowCheck,        # 是否有伏笔被遗忘太久(超15章未提及)
        ShuangDensityCheck,     # 爽点是否符合3章/5章节奏
        ForbiddenCheck,         # 是否触碰禁忌(绿帽/人设崩塌等)
    ]

    def check(self, chapter_text: str, context: GenerationContext) -> CheckResult:
        violations = []
        for check in self.checks:
            result = check.run(chapter_text, context)
            if result.has_violation:
                violations.append(result)
        return CheckResult(violations)

跨层基础设施:记忆图谱与上下文检索

解决的问题:百万字体量下,与主角/情节相关的事实、关系、伏笔无法在每次 LLM 调用时全量传入。需要一套能"按需提取最相关记忆"的系统,同时保留事实间的关联结构。

M1:核心问题拆解

naive 方案的失败原因:
  get_recent_events(last_N=10)  →  遗漏深层历史伏笔
  全量注入 WorldBible           →  token 爆炸,无法缩放
  关键字匹配检索                →  遗漏语义相关但字面不同的内容

需要解决的三个子问题:
  1. 如何表示记忆之间的关联(数据结构)
  2. 如何高效检索"与当前场景最相关"的记忆(检索算法)
  3. 如何在 token 预算内装入最有价值的上下文(预算分配)

M2:MemoryGraph — 记忆图谱数据结构

所有记忆单元作为图节点,关系作为有向边。图同时维护向量索引,支持语义检索。

# ---- 节点类型 ----

@dataclass
class MemoryNode:
    id: str                        # uuid
    node_type: NodeType            # 见下方枚举
    content: str                   # 文本内容(用于 embedding 和 LLM 读取)
    embedding: list[float]         # 语义向量(OpenAI/local embedding)
    chapter_created: int           # 哪章产生
    chapter_last_referenced: int   # 最近哪章被提及(用于 recency 权重)
    importance: float              # 0.0-1.0,由 Director 评分或规则计算
    metadata: dict                 # 类型特定字段,见下方


class NodeType(Enum):
    EVENT          = "event"        # 发生的具体事件(战斗/对话/发现)
    CHARACTER_STATE = "char_state"  # 某章时某角色的状态快照
    RELATIONSHIP   = "relation"     # 两角色间关系(含历史变化)
    WORLD_FACT     = "world_fact"   # 世界规则/已确立的设定事实
    FORESHADOW     = "foreshadow"   # 伏笔线索(open / resolved)
    ARC_MOMENT     = "arc_moment"   # 卷级关键戏剧节点(高潮/转折/揭秘)
    SUMMARY        = "summary"      # 压缩摘要(章摘要/卷摘要/全书摘要)


# ---- 边类型 ----

class EdgeType(Enum):
    CAUSES          = "causes"          # 事件A导致事件B
    INVOLVES        = "involves"        # 事件涉及某角色
    RESOLVES        = "resolves"        # 事件回收了某伏笔
    FORESHADOWS     = "foreshadows"     # 事件埋设了某伏笔
    EVOLVES_FROM    = "evolves_from"    # 角色状态B由状态A演变
    OCCURS_AT       = "occurs_at"       # 事件发生在某地点
    RELATES_TO      = "relates_to"      # 语义相关(自动建立,权重低)
    CONTRADICTS     = "contradicts"     # 潜在矛盾标记(由 ContradictionChecker 建立)
    PART_OF_ARC     = "part_of_arc"     # 属于某卷弧

@dataclass
class MemoryEdge:
    src: str         # node id
    dst: str         # node id
    edge_type: EdgeType
    weight: float    # 关联强度 0.0-1.0
    chapter_created: int


# ---- 图结构(in-memory + 持久化) ----

class MemoryGraph:
    nodes: dict[str, MemoryNode]           # id → node
    edges: list[MemoryEdge]
    adjacency: dict[str, list[MemoryEdge]] # 出边索引
    char_index: dict[str, list[str]]       # character_name → node_ids(快速按角色查)
    type_index: dict[NodeType, list[str]]  # node_type → node_ids
    vector_store: VectorStore              # 向量数据库(ChromaDB / Qdrant / FAISS)

    def add_node(self, node: MemoryNode):
        self.nodes[node.id] = node
        self.vector_store.upsert(node.id, node.embedding, node.content)
        # 更新索引 ...

    def add_edge(self, edge: MemoryEdge):
        self.edges.append(edge)
        self.adjacency.setdefault(edge.src, []).append(edge)

M3:分层记忆结构(解决长期一致性问题)

原始事件节点随章节增长 → token 不可控。解法:多层压缩,检索时按需解压。

Level 0: 原始文本      每章 2000-3000 字,不直接传给 LLM(只存储)
Level 1: 事实节点      post_chapter_update 提取,每章 ~15-30 个节点
Level 2: 章节摘要      SUMMARY 节点,每章 ~150 token,始终保留
Level 3: 卷弧摘要      每卷结束时生成,~400 token,高度压缩
Level 4: 全书摘要      每 50 章更新一次,~800 token,只保留最关键转折

检索时优先级:
  近期(< 15 章)→ 直接用 Level 1 事实节点(精确)
  中期(15-100 章)→ Level 2 章摘 + 语义命中的 Level 1 节点
  远期(> 100 章)→ Level 3/4 摘要 + 极高 importance 的 Level 1 节点

压缩规则:
  超过 50 章的 Level 1 节点,若 importance < 0.4 且 30 章未被引用 → 软删除(只保留摘要)
  伏笔节点(FORESHADOW)和 ARC_MOMENT 节点永不软删除
class HierarchicalMemory:
    def compress_chapter(self, chapter_no: int):
        """章结束时:提取事实节点 + 生成章摘要。"""
        raw_text = self.storage.get_chapter_text(chapter_no)
        facts = self.fact_extractor.extract(raw_text)       # LLM辅助提取
        summary_text = self.summarizer.summarize(raw_text)  # LLM生成150-token摘要

        # 建立事实节点并写入图
        for fact in facts:
            node = MemoryNode(
                node_type=NodeType.EVENT,
                content=fact.description,
                embedding=embed(fact.description),
                chapter_created=chapter_no,
                importance=fact.importance_score,
                ...
            )
            self.graph.add_node(node)
            # 建立边:INVOLVES(涉及角色)、OCCURS_AT(地点)、FORESHADOWS/RESOLVES
            self._build_edges(node, fact)

        # 章摘要节点
        summary_node = MemoryNode(
            node_type=NodeType.SUMMARY,
            content=summary_text,
            chapter_created=chapter_no,
            importance=0.8,
            metadata={"level": 2, "covers_chapters": [chapter_no]},
        )
        self.graph.add_node(summary_node)

    def compress_arc(self, arc_chapters: list[int]):
        """卷结束时:将本卷章摘要合并为卷摘要。"""
        chapter_summaries = [self.graph.get_summary(ch) for ch in arc_chapters]
        arc_summary_text = self.summarizer.summarize_arc(chapter_summaries)
        arc_node = MemoryNode(
            node_type=NodeType.SUMMARY,
            content=arc_summary_text,
            importance=1.0,
            metadata={"level": 3, "covers_chapters": arc_chapters},
        )
        self.graph.add_node(arc_node)

M4:ContextRetriever — 检索算法

每次 LLM 调用前执行,从 MemoryGraph 中检索最相关的记忆集合。

class ContextRetriever:

    def retrieve(self, query: RetrievalQuery, budget: ContextBudget) -> RetrievedContext:
        """
        query 来自 Director 的 SceneDirective,包含:
          - scene_description: 本章场景描述
          - active_characters: 本章出现的角色列表
          - location: 场景地点
          - active_foreshadow_ids: Director 认为本章应提及的伏笔
          - current_chapter_no: 当前章号
        """
        # ── Step 1: 构建检索查询向量 ──────────────────────────────────
        query_text = self._build_query_text(query)
        query_vec = embed(query_text)

        # ── Step 2: 向量检索(语义相关性) ──────────────────────────────
        # top-K 初始候选,K 较大(允许后续过滤)
        semantic_hits = self.graph.vector_store.search(
            query_vec, top_k=30,
            filter={"chapter_created": {"$lte": query.current_chapter_no - 1}}
        )

        # ── Step 3: 图展开(关联节点扩展) ───────────────────────────────
        # 从语义命中节点出发,沿特定边类型展开 1-2 跳
        expanded = set(h.node_id for h in semantic_hits)
        for node_id in list(expanded):
            neighbors = self.graph.get_neighbors(
                node_id,
                edge_types=[EdgeType.CAUSES, EdgeType.INVOLVES,
                             EdgeType.FORESHADOWS, EdgeType.RESOLVES],
                max_hops=2,
            )
            expanded.update(neighbors)

        # ── Step 4: 强制包含项(不受向量检索约束) ──────────────────────
        # 4a. 活跃伏笔(Director 指定必须提及的)
        for fid in query.active_foreshadow_ids:
            expanded.add(fid)
        # 4b. 本章活跃角色的最新状态快照
        for char in query.active_characters:
            latest_state = self.graph.get_latest_char_state(char)
            if latest_state:
                expanded.add(latest_state.id)
        # 4c. 近5章的章摘要(始终包含)
        for ch in range(max(1, query.current_chapter_no - 5), query.current_chapter_no):
            summary_id = self.graph.get_chapter_summary_id(ch)
            if summary_id:
                expanded.add(summary_id)
        # 4d. 当前卷弧摘要(若存在)
        arc_summary = self.graph.get_current_arc_summary()
        if arc_summary:
            expanded.add(arc_summary.id)

        # ── Step 5: 评分与排序 ───────────────────────────────────────
        candidates = [self.graph.nodes[nid] for nid in expanded if nid in self.graph.nodes]
        scored = [(node, self._score(node, query_vec, query.current_chapter_no))
                  for node in candidates]
        scored.sort(key=lambda x: x[1], reverse=True)

        # ── Step 6: token 预算分配 ────────────────────────────────────
        return self._pack_within_budget(scored, budget)

    def _score(self, node: MemoryNode, query_vec, current_chapter: int) -> float:
        """综合评分:语义相关 × 重要性 × 时间衰减 × 角色相关性。"""
        semantic_sim  = cosine_similarity(node.embedding, query_vec)   # 0-1
        importance    = node.importance                                  # 0-1
        recency_decay = 1.0 / (1.0 + 0.02 * (current_chapter - node.chapter_last_referenced))
        # 伏笔节点和 ARC_MOMENT 节点额外加权
        type_bonus    = 1.5 if node.node_type in (NodeType.FORESHADOW, NodeType.ARC_MOMENT) else 1.0
        return semantic_sim * importance * recency_decay * type_bonus

    def _build_query_text(self, query: RetrievalQuery) -> str:
        """将 SceneDirective 压缩为检索用的自然语言查询。"""
        chars = "、".join(query.active_characters)
        return f"{query.scene_description} 涉及角色:{chars} 地点:{query.location}"

M5:ContextBudgetManager — token 预算分配

# 每次 LLM 调用的 token 预算(可按场景类型调整)
CONTEXT_BUDGET_SLOTS = {
    "world_core_rules":    600,   # WorldBible 中与本章相关的核心规则(始终注入)
    "book_summary":        800,   # Level-4 全书摘要(始终注入)
    "arc_summary":         400,   # 当前卷摘要(始终注入)
    "recent_chapters":    1200,   # 近5章章摘(始终注入)
    "char_states":         600,   # 活跃角色当前状态(始终注入)
    "semantic_hits":      1000,   # 向量检索 + 图展开的相关记忆(按分排序填充)
    "active_foreshadows":  400,   # 活跃伏笔列表(始终注入)
    "scene_directive":     400,   # Director 的本章指令(始终注入)
    # 合计上限: ~5400 token(为生成文本留出空间)
}

class ContextBudgetManager:
    def pack(self, scored_nodes: list, slots: dict) -> str:
        """将节点按优先级装入各 slot,超出 slot 上限则截断。"""
        context_parts = {}

        # 强制注入 slot(不受语义检索影响)
        context_parts["world_core_rules"]   = self._get_world_rules()
        context_parts["book_summary"]       = self._get_book_summary()
        context_parts["arc_summary"]        = self._get_arc_summary()
        context_parts["recent_chapters"]    = self._get_recent_summaries(last_n=5)
        context_parts["char_states"]        = self._get_char_states()
        context_parts["active_foreshadows"] = self._get_active_foreshadows()
        context_parts["scene_directive"]    = self._get_scene_directive()

        # 动态填充 slot:语义命中节点按分排序填入
        semantic_tokens_used = 0
        semantic_budget = slots["semantic_hits"]
        semantic_content = []
        for node, score in scored_nodes:
            if node.node_type == NodeType.SUMMARY and node.metadata.get("level", 0) <= 2:
                continue  # 已由 recent_chapters slot 覆盖,跳过
            token_cost = estimate_tokens(node.content)
            if semantic_tokens_used + token_cost > semantic_budget:
                break
            semantic_content.append(node.content)
            semantic_tokens_used += token_cost
        context_parts["semantic_hits"] = "\n---\n".join(semantic_content)

        return self._format_context(context_parts)

M6:每章更新流程(写入 MemoryGraph)

章节文本生成完毕
       ↓
[FactExtractor] LLM辅助,从章节文本提取:
   - 发生了哪些事件(EVENT 节点)
   - 角色状态变化(CHARACTER_STATE 节点更新)
   - 新埋设的伏笔(FORESHADOW 节点,status=open)
   - 回收了哪些伏笔(FORESHADOW 节点,status=resolved)
   - 角色关系变化(RELATIONSHIP 节点更新)
       ↓
[EdgeBuilder] 根据提取结果建立边:
   EVENT → INVOLVES → CHARACTER
   EVENT → FORESHADOWS/RESOLVES → FORESHADOW
   EVENT → OCCURS_AT → LOCATION
   CHARACTER_STATE → EVOLVES_FROM → 前一状态
       ↓
[Summarizer] 生成章节 Level-2 摘要(~150 token)
       ↓
[SoftDeleter] 检查 50 章前的低重要性节点,标记 soft_deleted=True
       ↓
[ArcCompressor] 若当前章是卷结尾,生成 Level-3 卷摘要

M7:与各层的集成接口

# Layer 2(Director)调用检索
class Director:
    def plan_next_chapter(self, chapter_no: int) -> SceneDirective:
        directive = self._generate_directive(chapter_no)  # 基础规划
        # 检索相关记忆,注入到 Director 的规划上下文
        retrieval_query = RetrievalQuery(
            scene_description=directive.dramatic_purpose,
            active_characters=directive.character_states_before.keys(),
            location=directive.location,
            active_foreshadow_ids=self.foreshadow_tracker.get_active_ids(),
            current_chapter_no=chapter_no,
        )
        relevant_memory = self.retriever.retrieve(retrieval_query, budget=DIRECTOR_BUDGET)
        return self._refine_directive(directive, relevant_memory)

# Layer 4(TextGenerator)调用检索(更细粒度)
class TextGenerator:
    def generate_chapter(self, directive: SceneDirective) -> str:
        retrieval_query = RetrievalQuery(
            scene_description=directive.scene_description,
            active_characters=directive.active_characters,
            ...
        )
        context = self.retriever.retrieve(retrieval_query, budget=TEXT_GEN_BUDGET)
        prompt = self.prompt_builder.build(directive, context)
        return self.llm.generate(prompt)

跨层基础设施:叙事时间线管理器(NarrativeTimelineManager)

解决的问题:当前架构默认叙事顺序 = 故事发生顺序(线性)。插叙、倒叙、闪前等手法需要"叙事时间"和"故事时间"解耦,并管理读者在每个叙事节点上的认知状态。


NT1:核心概念拆分

故事时间线(StoryTimeline):事件按因果/时间顺序排列,与"真实发生"的顺序一致。
叙事时间线(NarrativeLine):读者阅读的顺序,章节在此序列上排列。

两者之间的关系决定叙事手法:
  NarrativeLine[ch] = StoryTimeline[ch]     → 平铺直叙(linear)
  NarrativeLine[ch] < StoryTimeline[ch-1]   → 倒叙/插叙(analepsis)
  NarrativeLine[ch] > StoryTimeline[current]→ 闪前/预叙(prolepsis)
  多条 StoryTimeline 交替出现              → 交叉叙事(parallel)
class FrameType(Enum):
    LINEAR      = "linear"       # 正常推进
    FLASHBACK   = "flashback"    # 整章倒叙(时间跳回过去)
    FLASH_EMBED = "flash_embed"  # 章内插叙(几段,非整章)
    FLASHFORWARD= "flashforward" # 闪前(短暂预示未来)
    PARALLEL    = "parallel"     # 交叉叙事(两条时间线交替)
    IN_MEDIAS   = "in_medias"    # 开篇倒叙(从高潮切入再补前情)


@dataclass
class NarrativeFrame:
    """描述单个章节的叙事时间属性。"""
    narrative_chapter: int      # 读者视角的章节编号
    story_time: StoryDate       # 本章内容在故事中发生的时间点
    frame_type: FrameType

    # 非 LINEAR 帧专用字段
    anchor_chapter: int | None  # 本帧"归属"的现在时章节编号(用于嵌套管理)
    temporal_label: str         # 给读者看的时间标记,如"三年前"/"片刻之后"
    return_hook: str            # 倒叙/闪前结束后,如何衔接回当前时间线

    # 戏剧功能(Director 规划时填写)
    dramatic_purpose: str       # 为什么在这里用非线性?信息揭示/情感对比/悬念制造
    reader_irony_target: str    # 若有意制造"读者比角色先知道"的反讽,描述它

NT2:叙事时间线规划器(NarrativePlanner)

Director 规划卷结构时,同时规划非线性叙事节点:

class NarrativePlanner:
    """
    管理全书/全卷的叙事节点序列。
    在 ArcManager.start_volume() 时初始化,按需调整。
    """

    # ── 使用约束(防止滥用导致读者迷失) ──────────────────────────
    NONLINEAR_RULES = {
        "max_consecutive_nonlinear": 2,      # 不超过连续2章非线性帧
        "min_linear_gap_after_nonlinear": 3, # 非线性段落后至少3章线性章回正
        "max_narrative_lag_chapters": 50,    # 叙事时间最多落后故事时间50章
        "temporal_marker_required": True,    # 时间跳跃必须有章首/节首时间标记
        "return_hook_required": True,        # 非线性段结束必须有衔接钩子
        "shuang_requirement_in_nonlinear":   # 非线性章仍须爽点(情感爽点/认知爽点可替代战力爽点)
            "SMALL_OR_EMOTIONAL",
    }

    # ── 适合触发非线性叙事的场景 ───────────────────────────────────
    NONLINEAR_TRIGGERS = {
        "mystery_setup":     "flashback",    # 出现谜题 → 倒叙揭秘
        "character_depth":   "flash_embed",  # 情感高光 → 插叙往事强化代入
        "climax_entry":      "in_medias",    # 卷首开篇倒叙,制造悬念
        "foreshadow_reveal": "flashforward", # 给读者一个未来碎片,吊胃口
        "parallel_storyline":"parallel",     # 反派/盟友线发展到关键点
    }

    def plan_nonlinear_slots(self, arc_plan: VolumePlan) -> list[NarrativeFrame]:
        """
        在卷规划阶段确定哪几章使用非线性叙事,输出帧序列。
        约束:
          - 非线性帧总数 ≤ 卷总章数 × 0.20(最多20%的章节用非线性手法)
          - 必须指定每帧的 dramatic_purpose(无目的的花哨 = 废章)
        """
        ...

NT3:读者认知追踪器(ReaderKnowledgeTracker)

非线性叙事的核心价值在于"读者知道的"和"角色知道的"产生差异。需要精确追踪:

class ReaderKnowledgeTracker:
    """
    追踪读者在每个叙事章节后所掌握的信息,
    与角色的 known_facts 分开维护。
    用于制造戏剧反讽(dramatic irony)和管理悬念张力。
    """

    # 两套独立的事实集合
    story_facts_by_time: dict[StoryDate, list[str]]  # 故事中真实发生的事
    reader_revealed_by_chapter: dict[int, list[str]] # 读者在第N章后知道的事

    def record_reveal(self, narrative_chapter: int, facts: list[str]):
        """章节生成后,记录读者新获得的信息(含倒叙/闪前揭示的历史信息)。"""
        self.reader_revealed_by_chapter.setdefault(narrative_chapter, []).extend(facts)

    def get_reader_knowledge_at(self, narrative_chapter: int) -> set[str]:
        """读者读完第N章后拥有的完整信息集合。"""
        known = set()
        for ch in range(1, narrative_chapter + 1):
            known.update(self.reader_revealed_by_chapter.get(ch, []))
        return known

    def get_dramatic_irony_opportunities(
        self, narrative_chapter: int, active_chars: list[str]
    ) -> list[IronyOpportunity]:
        """
        找出:读者知道 X,但场景中的角色不知道 X。
        这是在文本层面制造悬念/恐惧/期待的素材。

        例:读者在倒叙章看到了师父的秘密 → 当前章师父出场 → 
            读者以"上帝视角"观察主角对师父的信任 → 悬念感极强
        """
        reader_knows = self.get_reader_knowledge_at(narrative_chapter)
        irony_list = []
        for char in active_chars:
            char_knows = self.worldstate_db.get_char_known_facts(char)
            reader_exclusive = reader_knows - char_knows  # 读者知道但角色不知道的
            if reader_exclusive:
                irony_list.append(IronyOpportunity(
                    character=char,
                    reader_knows=reader_exclusive,
                    irony_type="reader_ahead",  # 读者先知
                ))
            char_exclusive = char_knows - reader_knows  # 角色知道但读者不知道的
            if char_exclusive:
                irony_list.append(IronyOpportunity(
                    character=char,
                    char_knows=char_exclusive,
                    irony_type="dramatic_suspense",  # 读者有感知但缺细节
                ))
        return irony_list

NT4:ContextRetriever 的时态感知扩展

非线性帧生成时,检索的是过去/未来状态,不是当前状态:

class ContextRetriever:
    # ...(现有 retrieve() 方法保持不变,处理 LINEAR 帧)

    def retrieve_for_frame(self, frame: NarrativeFrame, budget: ContextBudget) -> RetrievedContext:
        """
        根据帧类型选择不同的检索策略。
        """
        if frame.frame_type == FrameType.LINEAR:
            return self.retrieve(RetrievalQuery.from_frame(frame), budget)

        elif frame.frame_type in (FrameType.FLASHBACK, FrameType.FLASH_EMBED):
            return self._retrieve_historical(frame, budget)

        elif frame.frame_type == FrameType.FLASHFORWARD:
            return self._retrieve_for_flash_forward(frame, budget)

        elif frame.frame_type == FrameType.PARALLEL:
            return self._retrieve_parallel_line(frame, budget)

    def _retrieve_historical(self, frame: NarrativeFrame, budget: ContextBudget) -> RetrievedContext:
        """
        倒叙/插叙专用检索:以 frame.story_time 为时间锚,
        检索该过去时刻的世界状态,而非当前状态。
        """
        # 1. 从 WorldStateDB 拉取 story_time 时刻的角色状态快照
        historical_char_states = self.worldstate_db.get_snapshot_at_story_time(
            frame.story_time, frame.active_characters
        )

        # 2. 向量检索:只搜索 story_time 之前产生的记忆节点
        query_vec = embed(frame.scene_description)
        semantic_hits = self.graph.vector_store.search(
            query_vec, top_k=20,
            filter={"story_time": {"$lte": frame.story_time}}  # 关键:按故事时间过滤
        )

        # 3. 注入读者的现有认知(制造戏剧反讽的素材)
        reader_knowledge = self.reader_tracker.get_reader_knowledge_at(
            frame.narrative_chapter - 1  # 读者在看这章之前知道的
        )
        irony_opportunities = self.reader_tracker.get_dramatic_irony_opportunities(
            frame.narrative_chapter, frame.active_characters
        )

        # 4. 强制包含:连接"过去帧"和"当前时间线"的锚点信息
        # (让TextGenerator知道这段回忆与现在有什么关联,才能写好衔接)
        anchor_context = self._get_anchor_context(frame.anchor_chapter)

        return RetrievedContext(
            character_states=historical_char_states,
            semantic_hits=...,
            reader_irony_opportunities=irony_opportunities,
            anchor_return_context=anchor_context,
            temporal_offset=frame.temporal_label,
        )

    def _retrieve_for_flash_forward(self, frame: NarrativeFrame, budget: ContextBudget) -> RetrievedContext:
        """
        闪前专用:内容是"未来碎片",不做全量世界状态检索。
        只检索:当前已知的伏笔(将被未来呼应)+ 当前角色欲望/目标(闪前是其投影)。
        闪前内容必须模糊——只给意象,不给剧情。
        """
        active_foreshadows = self.foreshadow_tracker.get_all_open()
        protagonist_desires = self.worldstate_db.get_char_state(PROTAGONIST_ID).goals

        return RetrievedContext(
            active_foreshadows=active_foreshadows,
            protagonist_current_desires=protagonist_desires,
            note="闪前内容必须模糊化,禁止明确剧透具体情节",
        )

NT5:WorldStateDB 的历史快照查询

WorldStateDB 当前按叙事章节号存储快照,需扩展为支持按故事时间查询:

class WorldStateDB:
    # 现有:按叙事章节存储
    chapter_snapshots: dict[int, WorldSnapshot]   # narrative_chapter → snapshot

    # 新增:按故事时间索引
    story_time_index: dict[str, int]  # story_time_key → narrative_chapter
    # story_time_key = f"{year}-{month}-{day}" 字符串化

    def get_snapshot_at_story_time(
        self, story_time: StoryDate, char_ids: list[str]
    ) -> dict[str, CharacterState]:
        """
        返回角色在 story_time 时刻的状态。
        实现:找到 story_time 之前最近的叙事章节快照,从中提取角色状态。
        """
        # 找最近的已知快照
        closest_ch = max(
            (ch for ch, snap in self.chapter_snapshots.items()
             if snap.story_time <= story_time),
            default=1,
        )
        snapshot = self.chapter_snapshots[closest_ch]
        return {cid: snapshot.character_states[cid]
                for cid in char_ids if cid in snapshot.character_states}

    def record_story_time_for_chapter(self, narrative_chapter: int, story_time: StoryDate):
        """每章生成时记录其故事时间,建立双向索引。"""
        key = f"{story_time.year}-{story_time.month}-{story_time.day}"
        self.story_time_index[key] = narrative_chapter
        self.chapter_snapshots[narrative_chapter].story_time = story_time

NT6:TextGenerator 的非线性帧提示模板

不同帧类型对应不同生成策略:

NONLINEAR_PROMPT_TEMPLATES = {

    FrameType.FLASHBACK: """
【叙事模式:整章倒叙】
当前章节为回忆/倒叙场景,时间设定:{temporal_label}
衔接锚点(与现在时间线的连接):{anchor_context}

写作要求:
1. 章节开头必须有清晰的时间切换标记("三年前……"/"彼时……"等)
2. 倒叙内容必须与当前故事时间线有明确关联——为什么在这里回忆?
3. 历史状态约束:{historical_char_states}(人物在此时的状态,勿与现在混淆)
4. 章末衔接钩子:{return_hook}(自然过渡回现在时间线)
5. 爽点要求:{shuang_requirement}(倒叙章不能纯粹信息堆砌,必须有情感/认知层面的爽感)
{irony_hint}
""",

    FrameType.FLASH_EMBED: """
【叙事模式:章内插叙】
本章主线时间:当前。其中第{embed_position}段落插入回忆:{temporal_label}
插叙长度约束:不超过全章20%篇幅(插叙是点缀,不是主体)

写作要求:
1. 插叙通过感官触发(气味/声音/物品)自然进入,不用"他想起了……"直接引入
2. 插叙结束后不换场景,直接续接主线("他回过神来……"等过渡)
3. 插叙内容必须改变读者对当前场景的理解(为什么插?插了有什么新意义?)
""",

    FrameType.FLASHFORWARD: """
【叙事模式:闪前预叙】
本段为未来碎片,约{word_limit}字,高度模糊化。

写作要求:
1. 只呈现意象,不呈现具体情节(读者猜不出具体发生了什么,但有强烈预感)
2. 必须使用感官细节(视听触),禁止叙述性语句
3. 呈现可用意象:{oracle_imagery}(可引用命理系统提供的意象)
4. 闪前结束后,用"然而那是……之后的事了"或留白直接断开,不解释
""",

    FrameType.PARALLEL: """
【叙事模式:交叉叙事】
本章交替呈现两条时间线:
  A线:{line_a_description}({line_a_chars},当前故事时间)
  B线:{line_b_description}({line_b_chars},{line_b_time_relation})

写作要求:
1. 每段切换以"***"或空行+时间标记分隔
2. A/B线各自有自己的小钩子,在最后一段制造"两线即将相交"的感觉
3. 两线的情感节奏要有对比(一线紧张/一线平静,或一线爽/一线压抑)
4. 单次切换段落字数:500-800字(太短=碎,太长=读者忘了另一条线)
""",
}

NT7:非线性叙事的爽感补偿规则(新增 R 规则)

非线性帧不能绕过 ShuangScheduler,但允许用不同类型的爽点替代:

# 补充到 R2 爽点节奏规则
NONLINEAR_SHUANG_SUBSTITUTION = {
    # 倒叙/插叙章允许用"认知爽点"替代战力爽点
    "cognitive_shuang": [
        "揭开主角过去的关键谜底(读者获得'原来如此'的智识满足)",
        "倒叙中展现主角早期被压制的屈辱,使现在的打脸更有力量",
        "通过过去视角揭示某个 NPC 的真实面目(改变读者对当前情节的理解)",
    ],
    "emotional_shuang": [
        "回忆与已逝重要人物的最后相处(情感爆发)",
        "展现主角当前强大背后的代价/孤独(深化代入感)",
    ],
    # 约束:认知/情感爽点最多连续2章替代战力爽点
    # 第3章必须有战力层面的爽点兑现(防止读者认为"又是注水回忆")
    "max_substitution_consecutive": 2,
}

NT8:与 ContradictionChecker 的集成

非线性叙事引入新的矛盾风险:时态矛盾(倒叙中用了当前才有的知识/能力)。

# 扩展 ContradictionChecker 的检查项
NONLINEAR_CONSISTENCY_CHECKS = [
    {
        "name": "TemporalStateCheck",
        "desc": "倒叙/插叙场景中,角色状态是否与该历史时刻一致(不能用未来境界/知识)",
        "severity": "MUST",
        "applies_to": [FrameType.FLASHBACK, FrameType.FLASH_EMBED],
        "check": "compare chapter character_states with historical_snapshot at frame.story_time",
    },
    {
        "name": "TemporalMarkerCheck",
        "desc": "非线性帧是否有明确的时间切换标记(防读者迷失)",
        "severity": "MUST",
        "applies_to": [FrameType.FLASHBACK, FrameType.FLASHFORWARD, FrameType.PARALLEL],
    },
    {
        "name": "ReturnHookCheck",
        "desc": "非线性帧结束时是否有衔接回主时间线的钩子/过渡",
        "severity": "MUST",
        "applies_to": [FrameType.FLASHBACK, FrameType.FLASH_EMBED, FrameType.PARALLEL],
    },
    {
        "name": "FlashForwardVaguenessCheck",
        "desc": "闪前内容是否足够模糊(禁止明确剧透具体情节)",
        "severity": "SHOULD",
        "applies_to": [FrameType.FLASHFORWARD],
    },
    {
        "name": "NonlinearDensityCheck",
        "desc": "是否连续超过2章使用非线性叙事(读者迷失风险)",
        "severity": "MUST",
        "check": "count consecutive nonlinear frames in NarrativeLine",
    },
]

Layer 4:文本生成与审计

解决的问题:AI文风平淡、情感密度低、开篇暴露AI感。

4.1 差异化生成策略

根据场景类型选择不同的生成Prompt策略:

场景类型 生成策略 核心要求
高光爽点 慢镜头模式 感官细节×5,情感密度最大,节奏拉长
打脸/碾压 围观者视角 强化围观者反应,打脸对象的心理崩溃
升级突破 内外同步模式 内在感悟 + 外在表现同步描写
铺垫积累 水流模式 节奏快,信息密,伏笔自然融入
对话冲突 话剧模式 潜台词密集,每句话同时做两件事
过渡章节 效率模式 推进为主,不拖,结尾必须有钩子

4.2 人味注入机制

针对"情感密度不足"的专项解决方案:

class HumanFlavorInjector:
    """
    在高光段落生成时,强制注入人味提示。
    """
    flavor_templates = {
        "grief": [
            "不要写'他悲痛欲绝',要写:他做了什么动作?手在哪?眼睛看哪里?",
            "悲伤的具体化:时间感(这一秒有多长)、身体感觉、一个荒诞的小细节",
        ],
        "anger": [
            "愤怒时的克制比爆发更有力量",
            "愤怒的具体化:他控制自己不做什么?他在想什么奇怪的事?",
        ],
        "triumph": [
            "胜利后的第一个念头往往不是'我赢了',而是某个具体的小事",
            "胜利中必须有一个瑕疵或代价,纯粹的爽感反而显假",
        ],
    }

    def inject(self, scene_type: str, base_prompt: str) -> str:
        # 根据场景情感类型,向prompt注入人味要求

4.3 三层审计流程

生成章节文本
      ↓
┌─────────────────────────────┐
│  审计 A:一致性检查(Layer 3)  │  → 发现矛盾 → 标注修复点
└─────────────────────────────┘
      ↓(通过)
┌─────────────────────────────┐
│  审计 B:质量检查               │
│  - 三章测试(钩子/爽点/渴望)    │
│  - 人味评分(情感具体度)        │
│  - 爽点密度(是否触发调度器要求)  │
└─────────────────────────────┘
      ↓(通过 or 修复后)
┌─────────────────────────────┐
│  审计 C:可读性检查             │
│  - 段落长度(超5行自动拆分)     │
│  - 引号开头格式检查             │
│  - AI特征词检测(提示词漏删等)  │
└─────────────────────────────┘
      ↓
输出最终章节

4.4 修复循环

审计不通过 → 生成修复指令 → 重新生成(最多3次)
3次后仍不通过 → 标记为[HUMAN_REVIEW],跳过继续生成后续章节

系统总流程

初始化阶段(一次性)

输入Idea
  ↓
Layer 1: 生成WorldBible
  ↓
Director: 规划全书三大高潮 + 卷级大纲
  ↓
ArcManager: 初始化压力/节奏参数
  ↓
WorldStateDB: 初始化角色状态快照
  ↓
Ready

章节生成循环(每章执行)

┌──────────────────────────────────────────────────┐
│                    章节生成循环                      │
│                                                  │
│  1. ShuangScheduler.check() → 确定本章爽点要求      │
│  2. ArcManager.assess() → 确定本章戏剧功能          │
│  3. Director.plan() → 生成 SceneDirective          │
│  4. CharacterSandbox.run() → 生成角色行动/对话      │
│  5. Director.review() → 验证爽点,必要时重跑沙盒     │
│  6. ContextInjector.build() → 注入一致性上下文       │
│  7. TextGenerator.generate() → 生成章节文本         │
│  8. HumanFlavorInjector.post_process() → 人味增强   │
│  9. ContradictionChecker.check() → 一致性审计       │
│  10. QualityChecker.check() → 质量审计              │
│  11. PostChapterUpdate() → 更新WorldStateDB         │
│  12. 输出章节 / 标记[HUMAN_REVIEW]                  │
│                                                  │
│  → 继续下一章                                       │
└──────────────────────────────────────────────────┘

痛点解决映射

AI写网文痛点 本框架的解法 所在层
超20万字逻辑崩塌 WorldStateDB + ContradictionChecker Layer 3
情感密度不足,缺人味 差异化生成策略 + HumanFlavorInjector Layer 4
爽感节奏失控 ShuangScheduler强制执行3X节奏 Layer 2
同质化严重 CharacterSandbox涌现机制产生意外 Layer 2
缺少全局意图 Director明确每章戏剧目的 Layer 2
角色性格飘移 CharacterAgent有固定性格约束 Layer 2
伏笔遗忘 active_foreshadows + ForeshadowCheck Layer 3
开篇暴露AI感 三章测试强制检查 Layer 4

关键数据模型

# 核心数据结构概览

@dataclass
class SceneDirective:
    chapter_no: int
    dramatic_purpose: str       # "积累压力" | "爽点释放" | "伏笔埋设" | "换图过渡"
    required_shuang: ShuangLevel  # NONE | SMALL | MEDIUM | BIG
    chapter_hook_direction: str   # 章末钩子的方向
    conflict_type: str
    forbidden: List[str]          # 本章绝对不能出现的
    character_entry_states: Dict  # 各角色进入本章的状态

@dataclass  
class CharacterAgent:
    profile: CharacterProfile     # 固定:性格/动机/弱点/价值观
    state: CharacterState         # 动态:当前知识/位置/情绪/目标
    known_world: WorldView        # 该角色视角的世界认知(非全知)

@dataclass
class WorldStateDB:
    timeline: List[ChapterEvent]
    character_states: Dict[str, CharacterState]
    active_foreshadows: List[Foreshadow]
    resolved_foreshadows: List[Foreshadow]
    world_secrets: Dict[str, SecretStatus]  # 未揭露/已揭露

@dataclass
class GenerationContext:
    scene_plan: ScenePlan
    world_rules: str              # 相关世界规则
    character_states: Dict        # 涉及角色的当前状态
    recent_events: List           # 最近10章关键事件
    active_foreshadows: List      # 需要推进或注意的伏笔
    forbidden_contradictions: List # 不能违反的已知事实

MVP优先级(分阶段实现)

Phase 1:能跑通,生成可读内容

  • [ ] Layer 1:WorldBible生成(基础版)
  • [ ] Layer 3:WorldStateDB(基础事实记录)
  • [ ] Layer 4:文本生成(无差异化,基础Prompt)
  • [ ] ShuangScheduler(强制爽点节奏)

Phase 2:解决一致性问题

  • [ ] ContradictionChecker(4项关键检查)
  • [ ] ContextInjector(精准上下文注入)
  • [ ] PostChapterUpdate(自动状态更新)

Phase 3:解决人味问题

  • [ ] CharacterSandbox(角色沙盒涌现)
  • [ ] HumanFlavorInjector(人味注入)
  • [ ] 差异化生成策略(高光/打脸/升级等分类)

Phase 4:完整Director系统

  • [ ] Director Agent(完整戏剧意图规划)
  • [ ] ArcManager(全书弧度管理)
  • [ ] ForeshadowTracker(伏笔追踪)
  • [ ] 质量审计完整流程

量级估算

目标:300万字长篇小说

每章约 2500 字
总章数:300万 ÷ 2500 = 1200 章

每章消耗 Token 估算:
- ContextInjector 注入:~3000 tokens
- Director规划:~1000 tokens
- 沙盒运行:~2000 tokens
- 文本生成:~4000 tokens(输出)
- 审计:~2000 tokens
≈ 12000 tokens/章(输入+输出)

总计:1200 × 12000 = 1440万 tokens
按当前定价估算:约 $50-150(取决于具体模型组合)

最核心的设计决策说明

为什么用角色沙盒而非直接让AI写剧情?

直接让AI写"下一章发生什么" → AI会选择"安全"路线,情节平淡,缺乏人味。

角色沙盒的本质:把问题从"AI写故事"变成"AI扮演角色并做决策"

角色Agent在回答"面对这个情境,我(这个有具体性格和动机的人)会做什么?"时,会产生远比"主角应该做什么推动情节"更真实的答案。意外性和人味来自角色逻辑的自洽,而非剧情需要。

为什么要爽点调度器而非让AI自己把握节奏?

AI倾向安全写作。没有强制约束时,AI会持续铺垫、持续缓和,永远不会主动触发高潮。

ShuangScheduler的本质:将"3章一小高潮"从原则变成代码约束。每章开始前,调度器明确告知Director"这章必须有小爽点",Director则负责在沙盒框架内实现它。

一致性靠向量检索还是精确查询?

两者都需要: - 精确查询:角色当前境界、已揭露的秘密、角色位置 → 用数据库 - 语义检索:相关历史情节、类似的已发生事件 → 用向量数据库

只用向量检索会有精度问题(遗漏关键事实);只用精确查询会有遗漏(找不到隐性矛盾)。


规则体系:写作范式编码

将 note/ 下所有笔记的核心范式,转化为系统各层可执行的规则集合。 规则分为三个优先级:HARD(硬约束,违反=生成中止)、MUST(强约束,违反=标记重写)、SHOULD(软约束,违反=降分)。


R0:全局度量公式(Director的核心指标)

# 每章生成前,Director用此公式评估当前全书状态
shuang_value = (
    protagonist_growth_rate     # 主角成长速度(境界/能力)
    * antagonist_strength       # 当前对手强度(越强越值钱)
    * face_slap_impact          # 打脸力度(围观者震惊程度)
    * reward_richness           # 奖励丰富度(获得了什么)
    / max(setback_duration, 1)  # 挫折持续时长(章数)
)

# 目标:每5章的 shuang_value 均值必须 > 阈值
# 若低于阈值,Director强制提升下章爽点强度

RhythmJitter:节奏抖动工具(反八股文机制)

注记来源参数均为笔记推荐的"标准值",每本书初始化时采样一次,全书共享同一份参数(保证内部一致性)。章节级别不重新采样,否则读者会察觉节奏异常。

目的:让每本书的节奏参数在合理区间内略有不同,避免所有AI生成小说呈现相同的"机械节拍"。

import random

class RhythmJitter:
    """
    两级节奏扰动系统:
      Level 1 — 书级基线(book_baseline):每本书初始化时采样一次,决定这本书的"节奏风格"。
      Level 2 — 卷级参数(volume_params):每卷开始时在书级基线附近再采样,实现书内节奏自然漂移。
      Level 3 — 章级微扰(chapter_jitter):仅对间隔计数器加 ±1 随机扰动,不改变核心参数。

    设计原则:
      - 书级基线 sigma 大(定风格),卷级 sigma 小(在风格内漂移),章级只做微调。
      - 不同叙事阶段(开卷/中段/高潮前)可叠加相位偏置(phase_bias),主动控制节奏起伏。
      - 与人物/世界观相关的参数(伏笔追踪、元素窗口)只在书级采样,不随卷漂移。
    """

    @staticmethod
    def sample(base: int, sigma: float, lo: int, hi: int) -> int:
        """截断正态分布 → 整数。"""
        return max(lo, min(hi, round(random.gauss(base, sigma))))

    @staticmethod
    def init_book_baseline() -> dict:
        """Level 1:书级基线,每本书采样一次,存入 WorldBible.book_baseline。
        决定这本书的整体节奏偏好(紧凑型 vs 舒展型)。
        """
        s = RhythmJitter.sample
        return {
            # 核心节奏基线(卷级参数以此为中心漂移)
            "small_interval_base":      s(3,  0.6, lo=2, hi=5),
            "medium_interval_base":     s(5,  1.0, lo=4, hi=7),
            "big_interval_base":        s(15, 2.5, lo=11, hi=20),
            "suppression_base":         s(5,  1.0, lo=3, hi=7),
            "cooldown_big_base":        s(3,  0.6, lo=2, hi=5),
            "cooldown_medium_base":     s(1,  0.4, lo=1, hi=3),
            # 以下参数只在书级定,不随卷漂移
            "foreshadow_max_age":       s(15, 2.0, lo=11, hi=19),
            "foreshadow_reminder_interval": s(10, 1.5, lo=7, hi=13),
            "pre_climax_buildup_min":   s(5,  1.0, lo=3, hi=8),
            "post_climax_cooldown":     s(2,  0.5, lo=1, hi=4),
            "map_pre_announce":         s(5,  1.0, lo=3, hi=8),
            "element_diversity_window": s(5,  0.8, lo=3, hi=7),
            "max_setback_chapters":     s(5,  0.8, lo=3, hi=7),
            "arc_chapter_flex":         round(random.uniform(0.10, 0.20), 2),
        }

    # 各叙事阶段的相位偏置(乘以基线值,<1 收紧节奏,>1 放松节奏)
    PHASE_BIAS = {
        "opening":       0.85,  # 开卷:节奏偏快,迅速抓住读者
        "rising":        1.00,  # 上升段:标准节奏
        "middle":        1.10,  # 中段:略松,积累矛盾
        "pre_climax":    0.80,  # 高潮前:收紧,制造窒息感
        "climax":        0.70,  # 高潮:最快节奏
        "denouement":    1.20,  # 尾声:最松,让读者喘气
    }

    @staticmethod
    def init_volume_params(baseline: dict, phase: str = "rising") -> dict:
        """Level 2:卷级参数,每卷开始时调用,存入 ArcManager.current_volume_params。
        在书级基线附近小幅漂移,并叠加当前叙事阶段的相位偏置。

        phase: 当前叙事阶段,见 PHASE_BIAS。
        """
        s = RhythmJitter.sample
        bias = RhythmJitter.PHASE_BIAS.get(phase, 1.0)

        def drift(key: str, sigma: float, lo: int, hi: int) -> int:
            """在基线值附近做小 sigma 漂移,再乘以相位偏置。"""
            center = round(baseline[key] * bias)
            return s(center, sigma, lo, hi)

        return {
            "small_interval":           drift("small_interval_base",  0.4, lo=2, hi=5),
            "medium_interval":          drift("medium_interval_base", 0.6, lo=3, hi=8),
            "big_interval":             drift("big_interval_base",    1.5, lo=10, hi=21),
            "max_suppression_chapters": drift("suppression_base",     0.6, lo=2, hi=7),
            "cooldown_after_big":       drift("cooldown_big_base",    0.4, lo=1, hi=5),
            "cooldown_after_medium":    drift("cooldown_medium_base", 0.3, lo=1, hi=3),
        }

    @staticmethod
    def chapter_jitter(value: int) -> int:
        """Level 3:章级微扰,给间隔计数器加 ±1 随机扰动。
        仅在 ShuangScheduler 判断"是否触发爽点"时调用,不修改存储参数。
        避免每章严格在第 N 章触发,产生机械节拍感。
        """
        return max(1, value + random.choice([-1, 0, 0, 1]))  # 0 出现概率 2/4,偏向不变

# 调用链(伪代码):
#   WorldBible.__init__:  self.book_baseline = RhythmJitter.init_book_baseline()
#   ArcManager.start_volume(phase):  self.volume_params = RhythmJitter.init_volume_params(baseline, phase)
#   ShuangScheduler.check(counter, key):  threshold = RhythmJitter.chapter_jitter(volume_params[key])

R1:硬性禁忌规则(HARD — 触发=生成立即中止+报错)

HARD_FORBIDDEN = [
    # 男频第一死穴
    Rule("NTR", "主角伴侣/爱人被他人占有/出轨/玷污"),

    # 人设崩塌(判断依据:当前 arc_traits 状态,非初始固化设定)
    Rule("CHARACTER_OOC", "主角行为与当前弧光状态明显矛盾,且无已批准的 ArcChange 作为依据"),
    # 注:弧光演变本身不是OOC,"未经闸门的突变"才是OOC。
    # 例:trust_level 已降至 -0.3 的主角突然全盘托付陌生人 → OOC
    #     trust_level 从 +0.6 经过背叛事件下降到 +0.1 → 合规弧光演变,非OOC

    # 猪猡主角
    Rule("IQ_DROP", "主角为推进剧情而做出智商明显掉线的决策"),

    # 长期虐主无出路(连续N章触发)
    Rule("ENDLESS_VICTIM", "主角连续{MAX_SETBACK_CHAPTERS}章以上无任何爽点/进展"),

    # 世界规则自毁
    Rule("WORLD_RULE_BREAK", "章节内容违反WorldBible中已定义的基础规则"),
]

MAX_SETBACK_CHAPTERS = rhythm_params["max_setback_chapters"]
# 基准值5;每本书初始化时由 RhythmJitter 采样(范围 3-7)

R2:爽点节奏规则(MUST — ShuangScheduler的核心参数)

SHUANG_RHYTHM = {
    # 从 ArcManager.current_volume_params 读取(每卷开始时由 RhythmJitter.init_volume_params() 采样)
    # 书内不同卷的节奏会有自然漂移,开卷/高潮前偏快,中段偏松。
    # 下方为期望均值 → 实际采样范围(括号内,含相位偏置影响)
    "small_interval":           volume_params["small_interval"],           # ~3  (2-5),高潮相位可压到2
    "medium_interval":          volume_params["medium_interval"],          # ~5  (3-8)
    "big_interval":             volume_params["big_interval"],             # ~15 (10-21)
    "max_suppression_chapters": volume_params["max_suppression_chapters"], # ~5  (2-7)
    "cooldown_after_big":       volume_params["cooldown_after_big"],       # ~3  (1-5)
    "cooldown_after_medium":    volume_params["cooldown_after_medium"],    # ~1  (1-3)
}

# 三级扰动总结:
#   Level 1(书级):确定基线风格,例如"这本书整体节奏偏快"
#   Level 2(卷级):在基线上漂移 + 叙事相位偏置,例如"本卷是高潮前,节奏收紧"
#   Level 3(章级):ShuangScheduler 触发前对阈值 ±1 微扰,消除机械节拍感

# 爽点等级定义
SHUANG_LEVELS = {
    "SMALL": ["小打脸", "小装逼", "获得普通资源", "境界微进步", "小帮助他人"],
    "MEDIUM": ["重要打脸", "境界大突破", "获得珍贵资源/法宝", "击败中等反派"],
    "BIG": ["卷级大高潮", "击败主要反派", "大逆转", "大收获", "换地图前的终章爆发"],
}

R3:情节单元规则(MUST — 每个ScenePlan必须满足)

每个情节单元(场景)必须包含以下四要素,缺一标记重写:

SCENE_REQUIREMENTS = {
    "goal":     "主角在本场景想要什么?(必须明确,不能模糊)",
    "conflict": "什么阻止他得到?(阻力越强,爽点越值钱)",
    "result":   "他得到了吗?(通常部分得到或以意外方式得到)",
    "hook":     "本场景如何推进到下一场景?(章末必须有悬念/期待)",
}

# 打脸情节专项结构(10章完整循环)
FACE_SLAP_STRUCTURE = {
    "hatred_build":  (1, 2),  # 章数:龙套建立仇恨/嘲讽
    "suppression":   (1, 1),  # 章数:主角被压制/受辱
    "counter":       (1, 1),  # 章数:主角开始反踩
    "slap":          (2, 3),  # 章数:打脸过程+围观者反应
    "reward":        (1, 3),  # 章数:【必须】主角获得具体好处
    # reward是整个循环的灵魂:无收获的打脸=废章
}

# 围观者反应强化规则(打脸/高潮场景)
SPECTATOR_AMPLIFICATION = True  # 必须写围观者震惊/改变态度/传播

R4:YY七要素注入规则(SHOULD — Director规划时参考)

每段情节(约10章)至少使用以下2个要素:

YY_SEVEN_ELEMENTS = [
    "奇遇",  # 意外获得传承/机缘/天材地宝
    "升级",  # 境界/能力提升
    "泡妞",  # 与女性角色产生情感进展
    "寻宝",  # 寻找/争夺珍贵资源
    "发财",  # 获得大量财富/资源
    "欺人",  # 主角压制/碾压对手
    "助人",  # 主角帮助盟友/弱者(建立情义线)
]

# 修真/仙侠专属10要素(类型为修真/仙侠时启用)
XIANXIA_ELEMENTS = [
    "救美", "采药", "夺宝", "炼丹", "反抢",
    "踩人", "装逼", "体悟", "副本", "升级反杀"
]

# 组合规则:每个情节段落的元素要保持多样性,不能连续使用同一要素
element_diversity_window = rhythm_params["element_diversity_window"]  # ~5 (3-7)

R5:开篇三章特别规则(MUST — 全书最严格的窗口)

OPENING_RULES = {
    # 第1章规则
    "ch1_conflict_in_first_para": True,   # 第1章第1段必须有矛盾/张力
    "ch1_protagonist_appears": True,       # 第1章主角必须出现
    "ch1_max_supporting_chars": 3,         # 配角不超过3个
    "ch1_end_hook": True,                  # 第1章结尾必须有钩子

    # 前3章规则
    "ch3_establish_desire": True,          # 3章内确立主角核心渴望
    "ch3_establish_obstacle": True,        # 3章内确立主要阻力
    "ch3_has_small_shuang": True,         # 3章内至少1个小爽点释放
    "ch3_protagonist_viewpoint": True,    # 视角必须跟着主角走

    # 金手指规则
    "golden_finger_hint_by_ch2": True,   # 金手指最晚第2章有暗示
    "golden_finger_constraints": True,   # 金手指必须有限制(防止无敌流)

    # 世界观展示规则(AGAINST 直接说教)
    "worldview_through_action": True,    # 世界规则通过角色行动展示,禁止大段说明
    "no_worldview_dump_in_ch1": True,    # 第1章禁止大段世界背景介绍
}

R6:人物塑造规则(MUST — CharacterAgent初始化时建立,运行时动态演变)

人物塑造分两个层面:CharacterCore 永不变(一致性基础);ArcTrait 允许演变(弧光深度)。 行为合理性判断依据"当前 arc_traits 状态",不依据初始设定(见 2.4.4 OOC 更新说明)。

PROTAGONIST_RULES = {
    # ── CharacterCore 层(永不变) ────────────────────────────────
    "core_archetype_fixed": True,        # 底层原型固定(孤傲天才不会变成话痨暖男)
    "nine_type_fixed": True,             # 九型人格固定(压力下的本能反应不变)
    "voice_pattern_fixed": True,         # 说话风格/口头禅固定(文风层面标识)

    # ── ArcTrait 层(受控演变) ───────────────────────────────────
    "arc_traits_required": True,         # 必须预设 2-4 个可演变维度
    "arc_direction_pre_planned": True,   # 每个维度的大方向在全书初始化时规划好
    "arc_milestones_required": True,     # 全书至少3个弧光里程碑(与卷级高潮绑定)

    # ── 演变约束(引用 ArcEvolutionGate 参数) ─────────────────────
    "min_chapters_between_changes": 20,  # 同维度两次演变间隔 ≥ 20 章
    "max_trait_changes_per_volume": 2,   # 每卷最多 2 个维度发生变化
    "max_single_delta": 0.20,            # 单次演变幅度绝对上限(20% of range)
    "trigger_required": True,            # 演变必须由剧情事件触发,不能无由突变

    # ── 行为逻辑约束 ──────────────────────────────────────────────
    "iq_baseline": "CONSISTENT",        # 智商在线,不为推进剧情降智
    "moral_high_ground": True,          # 不主动欺负人,被逼才反击
    "desire_explicit": True,             # 表面渴望清晰(可随剧情段落更新)
    "desire_vs_need_tension": True,     # 渴望 ≠ 需要(两者矛盾是弧光核心张力)
    "backstory_iceberg": True,          # 幕后故事只露1/8,其余埋在行为中

    # ── 演变可见性要求 ─────────────────────────────────────────────
    "arc_change_must_show_not_tell": True,  # 弧光变化通过行为/对话体现,禁止直接说明
    # 错误:"他变得更加信任他人了。"
    # 正确:(过去从不借兵器给人)第N章他把自己的法宝递给了一个受伤的陌生人。
}

ANTAGONIST_RULES = {
    "hate_buildup_required": True,       # 反派必须先铺垫其恶/嚣张,才能被打脸
    "strength_credibility": True,        # 对手强度可信(不能弱到无意义)
    "motivation_coherent": True,         # 反派有自己的逻辑,不只是"为了被打脸"
    "mirror_function": True,             # 反派是主角弧光的镜像(代表主角可能的黑化路径)
    # 反派不需要弧光系统(配角/反派性格脸谱化反而易记)
}

SUPPORTING_CHAR_RULES = {
    "no_function_overlap": True,         # 配角功能不重复
    "single_dominant_trait": True,       # 配角性格鲜明单一(一个标志性特质让读者记住)
    "serves_protagonist": True,          # 配角价值:衬托主角 + 推动剧情
    "no_stealing_spotlight": True,       # 高光场景不能抢主角风头
    # 重要配角(篇幅 > 20 章)可以有1-2个 ArcTrait,但参数更严格(间隔 ≥ 30 章)
}

# ── 弧光演变的频率/幅度速查表 ─────────────────────────────────────
ARC_EVOLUTION_BUDGET = {
    # 全书 100 万字(约 400-500 章)的演变预算
    "total_milestones_planned":    6,    # 全书 6 个里程碑(约每卷1-2个)
    "max_cumulative_delta":        1.2,  # 单个维度全书累计最大变化量(从 -1 到 +1 仅用1.2)
    # 即:读者感受到"他变了",但不会感受到"这人换了个灵魂"
}

R7:伏笔追踪规则(MUST — ForeshadowTracker执行)

FORESHADOW_RULES = {
    # 伏笔生命周期
    "must_resolve": True,              # 有伏必有应,不允许永久悬空
    "max_age_without_mention": rhythm_params["foreshadow_max_age"],        # ~15 (11-19)
    "forced_reminder_interval": rhythm_params["foreshadow_reminder_interval"],  # ~10 (7-13)
    "resolve_window": (10, 30),        # 伏笔在埋设后10-30章内回收(固定,不抖动)

    # 伏笔质量规则
    "natural_embedding": True,         # 伏笔要自然融入情节,不能刻意点出
    "multiple_functions": True,        # 伏笔段落同时承担其他叙事功能
    "max_concurrent_foreshadows": 7,   # 同时存在的活跃伏笔上限(认知负荷硬约束,不抖动)

    # 回收质量规则
    "resolution_surprise": True,       # 伏笔回收要有"情理之中,意料之外"的感觉
    "resolution_rewarding": True,      # 读者发现伏笔被回收时应感到智力满足
}

R8:高潮设计规则(MUST — Director每卷规划时执行)

CLIMAX_DESIGN_RULES = {
    # 全书三大高潮(必须在初始化时规划好)
    "three_act_climaxes": True,        # 前中后各一个大高潮
    "reverse_engineering": True,       # 先定大高潮,再反推铺垫情节

    # 高潮前积累规则(压力建立)
    "pre_climax_buildup_min": rhythm_params["pre_climax_buildup_min"],  # ~5 (3-8)
    "antagonist_hype_required": True,  # 高潮前反复强化对手的强大
    "protagonist_loss_required": True, # 高潮前主角有明显损失/限制
    "near_despair_moment": True,       # 高潮来临前主角几乎无望(然后逆转)

    # 高潮质量规则
    "foreshadow_callback": True,       # 高潮逆转必须呼应之前埋设的伏笔
    "unexpected_but_logical": True,    # 意料之外,情理之中(不能强行逆转)
    "inner_outer_climax_sync": True,   # 最佳效果:内在冲突与外在冲突同时高潮

    # 高潮后规则
    "spectator_reaction": True,        # 高潮后必须写围观者态度改变
    "next_arc_hook": True,             # 卷结尾大高潮后立即埋下下一卷的钩子
    "cooldown_chapters": rhythm_params["post_climax_cooldown"],  # ~2 (1-4)

    # 卷间连接规则
    "arc_start_reset": True,           # 每卷开始主角重置到相对弱小状态(新竞争层级)
    "arc_end_reward": True,            # 每卷结尾主角有明确收获(不能结束得很惨)
}

R9:地图/换图规则(MUST — ArcManager管理)

MAP_TRANSITION_RULES = {
    # 换图时机规则
    "pre_announce_chapters": rhythm_params["map_pre_announce"],  # ~5 (3-8)
    "pre_announce_attractive": True,   # 新地图必须被描述为更有资源/更高竞争层级

    # 换图时的主角状态重置
    "power_reset_relative": True,      # 主角在新地图重新变成"弱小者"(相对竞争层级)
    "keep_absolute_growth": True,      # 但绝对实力确实比上一地图强(成长不回退)

    # 换图触发条件
    "trigger": "上一地图的最大反派被击败 OR 上一地图已无挑战价值",

    # 换图爽点
    "transition_shuang": True,         # 换图本身是爽点:新世界震惊于主角的强大
}

R10:文本生成质量规则(SHOULD — TextGenerator的Prompt约束)

TEXT_QUALITY_RULES = {
    # 段落规则
    "max_paragraph_lines": 5,          # 超过5行自动换段
    "no_quote_opening": True,          # 禁止引号开头段落(网文格式禁忌)

    # 情感描写规则(Human Flavor强化)
    "emotion_concretize": True,        # 禁止抽象情感描写,必须具体化
    # 错误示例:"他悲痛欲绝"
    # 正确示例:"他站在原地,手机滑落也没有去捡。"

    "specific_body_language": True,    # 重要情感时刻必须有身体语言/动作细节
    "internal_monologue_authentic": True, # 内心独白要有"人性的复杂和矛盾"

    # 细节规则
    "detail_must_function": True,      # 每个细节必须有功能(推情节/揭人物/造氛围)
    "no_decorative_description": True, # 禁止纯装饰性描写(不推情节的景物堆砌)
    "indirect_over_direct": True,      # 间接表达>直接说明("他嫉妒了"→写动作)

    # 对话规则
    "each_line_does_two_things": True, # 每句对话同时做两件事
    "subtext_encouraged": True,        # 潜台词>直说(内心想的和说出来的可以不同)
    "character_voice_distinct": True,  # 不同角色说话风格可区分
    "no_infodump_dialogue": True,      # 禁止说明书式对话(互相解释读者已知信息)

    # 节奏规则
    "fast_scene_short_sentences": True, # 动作场景用短句/短段落
    "slow_scene_sensory_detail": True,  # 情感场景用感官细节拉长节奏
    "scene_transition_types": ["顺接", "跳接", "暗切", "对比切"],  # 选择合适的衔接
}

R11:场景类型专项规则(TextGenerator差异化策略细则)

SCENE_TYPE_RULES = {

    "face_slap": {
        # 打脸/碾压场景
        "spectator_perspective": True,      # 必须有围观者视角的震惊/议论
        "antagonist_psychological_collapse": True,  # 必须写对手的心理/神态崩溃
        "reward_moment_slowdown": True,     # 主角获得好处的瞬间放慢节奏
        "protagonist_calm_contrast": True,  # 主角越淡定,爽感越强(与围观者震惊形成对比)
    },

    "power_up": {
        # 升级突破场景
        "inner_outer_sync": True,           # 内在感悟与外在变化同步描写
        "pre_breakthrough_near_failure": True,  # 突破前必须有"几乎失败"的危机感
        "environment_reaction": True,       # 境界突破时周围环境要有反应(天地异象)
        "new_strength_demonstration": True, # 突破后立即用实战展示新强度
    },

    "fight": {
        # 打斗场景(四段式)
        "setup_hatred": True,               # 铺垫:矛盾激化到不可调和
        "spectator_commentary": True,       # 围观者嘴炮/判断(建立对比预期)
        "process_with_strategy": True,      # 过程:主角不只是蛮力,有策略
        "five_senses": True,                # 五感描写(视听触味嗅至少三种)
        "decisive_blow_design": True,       # 最终一击要有记忆点/设计感
    },

    "transition": {
        # 过渡/铺垫章节
        "speed_priority": True,             # 速度优先,不拖沓
        "foreshadow_weave": True,           # 自然融入伏笔(不着痕迹)
        "end_hook_mandatory": True,         # 结尾必须有钩子(即使是过渡章也不能无钩子)
        "relationship_advance": True,       # 可推进配角关系/感情线(充分利用"注水"空间)
    },

    "emotional_peak": {
        # 情感高光场景(非战斗)
        "slowdown_mandatory": True,         # 节奏必须放慢
        "specific_not_abstract": True,      # 禁止抽象情感,要具体化
        "small_imperfection": True,         # 高光时刻加一个小缺陷/代价(纯粹爽感反而假)
        "sensory_richness": True,           # 感官细节×5
    },
}

R12:一致性检查规则(ContradictionChecker参数化)

CONSISTENCY_CHECKS = [
    {
        "name": "PowerLevelCheck",
        "desc": "主角/NPC当前境界与状态DB是否一致",
        "severity": "MUST",
    },
    {
        "name": "KnowledgeCheck",
        "desc": "角色是否使用了'不应该知道'的信息",
        "severity": "MUST",
        "note": "反派不能知道主角已突破,除非有合理的信息来源",
    },
    {
        "name": "ForeshadowAgeCheck",
        "desc": "是否有活跃伏笔超过15章未被提及",
        "severity": "MUST",
        "action": "触发时向Director注入'该伏笔需要提及'的指令",
    },
    {
        "name": "CharacterConsistencyCheck",
        "desc": "角色行为是否与其profile(性格/动机/弱点)一致",
        "severity": "MUST",
    },
    {
        "name": "TimelineCheck",
        "desc": "事件发生时间/地点是否与timeline记录一致",
        "severity": "MUST",
    },
    {
        "name": "ShuangDensityCheck",
        "desc": "当前章爽点密度是否满足调度器要求",
        "severity": "SHOULD",
        "action": "不满足时向TextGenerator注入爽点增强指令",
    },
    {
        "name": "ForbiddenCheck",
        "desc": "是否触碰R1硬性禁忌",
        "severity": "HARD",
        "action": "立即中止生成,返回错误",
    },
    {
        "name": "WorldRuleCheck",
        "desc": "是否违反WorldBible中定义的世界规则",
        "severity": "MUST",
    },
    {
        "name": "OpeningWindowCheck",
        "desc": "前3章是否满足R5开篇特别规则(仅对第1-3章执行)",
        "severity": "MUST",
    },
]

R13:卷级弧度规则(ArcManager参数化,英雄之旅映射)

# 每一卷映射到英雄之旅12阶段的简化版
#
# 【弹性说明】
# chapters 字段为"标准卷(约70章)"下的参考区间。
# ArcManager 实际执行时乘以浮动因子:
#   actual_range = (lo * flex, hi * flex)
#   flex = volume_total_chapters / 70 * (1 ± rhythm_params["arc_chapter_flex"])
# arc_chapter_flex 范围 0.10-0.20(由 RhythmJitter 采样),
# 使得每卷情节节点位置有 ±15% 的自然漂移,不完全死板。
VOLUME_ARC_TEMPLATE = {
    "phase_1_opening": {
        "chapters": (1, 5),
        "stage": "普通世界重置",
        "rule": "主角在新地图/新阶段重新变弱,遭遇新的威胁/机遇",
        "shuang_level": "SMALL",
    },
    "phase_2_call": {
        "chapters": (3, 8),
        "stage": "冒险召唤",
        "rule": "新的核心矛盾/目标出现,读者明白本卷要解决什么",
        "shuang_level": "SMALL",
    },
    "phase_3_tests": {
        "chapters": (8, 40),
        "stage": "测试/盟友/敌人",
        "rule": "主线积累:结交盟友、与中等反派交手、积累资源",
        "shuang_level": "SMALL+MEDIUM交替",
        "note": "占本卷最大篇幅,爽点要保持密度",
    },
    "phase_4_approach": {
        "chapters": (35, 50),
        "stage": "接近洞穴",
        "rule": "主角已经很强,但BOSS仍遥不可及,制造'最后一步'的紧迫感",
        "shuang_level": "MEDIUM",
    },
    "phase_5_ordeal": {
        "chapters": (48, 58),
        "stage": "最黑暗时刻",
        "rule": "主角遭遇最大危机,几乎失败(必须!),读者处于绝望边缘",
        "shuang_level": "NONE/ACCUMULATION",
        "note": "压力密度最大,但这是高潮爆发的前提",
    },
    "phase_6_climax": {
        "chapters": (56, 65),
        "stage": "大高潮爆发",
        "rule": "逆转+击败卷BOSS+获得本卷最大奖励,伏笔全部回收",
        "shuang_level": "BIG",
    },
    "phase_7_return": {
        "chapters": (63, 70),
        "stage": "回归+钩子",
        "rule": "庆祝胜利(让读者享受)→ 发现更大威胁/秘密(钩入下一卷)",
        "shuang_level": "SMALL+NEXT_ARC_HOOK",
    },
}

R14:类型特化规则(WorldBible初始化时按题材加载)

GENRE_SPECIFIC_RULES = {

    "xianxia": {  # 修真/仙侠
        "no_western_elements_before_1M_words": True,  # 100万字前禁止西方元素
        "long_term_goal": "长生/成仙/证道",           # 必须有明确的终极追求
        "short_term_goals": ["宝物收集", "境界提升", "恩怨了结"],
        "manufacture_system": True,                    # 必须有炼丹/炼器/符咒等制造体系
        "power_gap_visible": True,                    # 不同境界的战力差距要可感知
        "forbidden_degenerate_romance": True,         # 过于流氓化破坏仙气
    },

    "xuanhuan": {  # 玄幻(东方奇幻)
        "random_encounter_all_beautiful": True,       # 随机所有女性都美(类型惯例)
        "upgrade_system_numeric": True,               # 等级必须数值化(给读者升级感)
        "map_change_resets_competition": True,        # 换地图重置竞争层级
        "villain_background_escalation": True,        # 每个被打倒的对手有更大的背景
        "basic_formula": "打儿子→打老子→打祖宗→打师傅→灭山头→换地图→重复",
    },

    "dushi": {  # 都市
        "realistic_details_required": True,           # 细节必须真实(消费水平/职场规则)
        "protagonist_justification": True,            # 主角赢必须让读者觉得"理所应当"
        "escalation_credible": True,                  # 发展速度要有合理性(过快=不信)
        "power_backer_credible": True,                # 靠山要有可信分量
        "villain_must_be_hated": True,                # 打脸对象先要充分可恨
    },

    "chuanyue": {  # 穿越/重生
        "advantage_show_within_ch3": True,            # 前3章内必须展示"降维打击"的具体优势
        "modern_knowledge_application": True,         # 现代知识在新世界的应用要合理落地
        "butterfly_effect_awareness": True,           # 穿越改变历史有因果逻辑
        "no_convenient_coincidences": True,           # 巧合必须有根据(不能太方便)
    },

    "wangyou": {  # 网游
        "numeric_power_system": True,                 # 等级/属性数值化
        "hidden_talent_reveal_early": True,           # 隐藏职业/天赋第2章必须暗示
        "game_logic_consistent": True,                # 游戏规则必须自洽
        "system_novelty": True,                       # 金手指/系统要有独特之处(防止同质化)
    },
}

R15:开头漏斗规则(读者留存最关键30000字)

# 读者漏斗模型:前30000字是最大漏斗
# 每一关过滤多少读者取决于质量
OPENING_FUNNEL_GATES = [
    {
        "gate": "标题+简介",
        "rule": "书名3-6字,简单好记,体现核心YY点;简介包含设定+处境+悬念",
        "chapter": 0,
    },
    {
        "gate": "第1章留存",
        "rules": [
            "第1段有矛盾/张力",
            "主角在第1章出现",
            "无大段世界观说明",
            "第1章结尾有钩子",
        ],
        "chapter": 1,
    },
    {
        "gate": "前3章留存",
        "rules": [
            "确立核心渴望",
            "确立主要阻力",
            "至少1个小爽点",
            "世界规则通过行动展示",
        ],
        "chapters": (1, 3),
    },
    {
        "gate": "前6万字留存",
        "rules": [
            "第一个中级高潮出现",
            "主要人物关系建立",
            "伏笔雏形出现",
            "读者对主角未来充满期待",
        ],
        "word_count": 60000,
    },
]

规则优先级总览

HARD(硬约束):R1 禁忌规则
                → 触发=生成中止,不可跳过

MUST(强约束):R2 爽点节奏
                R3 情节单元四要素
                R5 开篇三章
                R6 人物塑造
                R7 伏笔追踪
                R8 高潮设计
                R9 换图规则
                R12 一致性检查
                R13 卷级弧度
                → 触发=标记重写,允许最多3次重试

SHOULD(软约束):R4 YY七要素
                  R10 文本质量
                  R11 场景专项
                  R14 类型特化
                  R15 开头漏斗
                  → 触发=降低质量评分,不强制重写

特色约束系统:命理引擎(MingliEngine)

设计意图:将子平术命理推演嵌入剧情概率层,实现两个目标: 1. 叙事连贯性:主角/关键角色的"命格"与故事中的占卜预言共用同一套数据源,预言在架构层有真实依据而非凭空捏造。 2. 概率权重调制:命格与当前时柱的五行生克关系,作为 Director 规划事件时的软性概率权重,使剧情走向带有"天命"色彩而非完全随机。

所有命理推演通过 DeepSeek API 完成,本地只做数据结构管理和结果解析,不实现命理计算逻辑。


ML1:故事历法 → 干支映射(MingliCalendar)

@dataclass
class StoryDate:
    """故事内时间单位,与现实历法解耦。"""
    year: int    # 故事纪年(如:大盛历 150 年)
    month: int   # 1-12
    day: int     # 1-30(简化)
    hour_index: int  # 0-11(对应十二时辰)


@dataclass
class GanZhi:
    """天干地支对。"""
    gan: str   # 甲乙丙丁戊己庚辛壬癸
    zhi: str   # 子丑寅卯辰巳午未申酉戌亥


@dataclass
class TimeGanZhi:
    """某一时刻的四柱干支。"""
    year_gz:  GanZhi
    month_gz: GanZhi
    day_gz:   GanZhi
    hour_gz:  GanZhi

    def to_prompt_str(self) -> str:
        return (f"年柱{self.year_gz.gan}{self.year_gz.zhi} "
                f"月柱{self.month_gz.gan}{self.month_gz.zhi} "
                f"日柱{self.day_gz.gan}{self.day_gz.zhi} "
                f"时柱{self.hour_gz.gan}{self.hour_gz.zhi}")


class MingliCalendar:
    """
    将故事内历法映射到干支周期。

    映射策略:
      - WorldBible 初始化时指定"纪年起点对应的真实干支年份"(epoch_ganzhi_year)
      - 之后按 60 年甲子周期推算
      - 月/日/时的干支按五虎遁年起月法、五鼠遁日起时法标准推算
      - 推算逻辑通过 DeepSeek 完成,本地只存储结果

    示例:
      大盛历元年 = 甲子年(epoch = "甲子")
      大盛历 150 年 = 甲子 + 149 年 = 癸卯年(150 mod 60 = 30 → 对应天干地支表第30位)
    """
    epoch_ganzhi_year: str  # WorldBible 初始化时设定,如 "甲子"

    def get_time_ganzhi(self, date: StoryDate) -> TimeGanZhi:
        """调用 MingliOracle 计算指定时刻的四柱,缓存结果。"""
        cache_key = f"{date.year}-{date.month}-{date.day}-{date.hour_index}"
        if cache_key in self._cache:
            return self._cache[cache_key]
        result = MingliOracle.compute_time_ganzhi(self.epoch_ganzhi_year, date)
        self._cache[cache_key] = result
        return result

ML2:角色八字(CharacterBaZi)

@dataclass
class CharacterBaZi:
    """
    角色的命盘,WorldBible 初始化时由 MingliOracle 生成。
    """
    character_id: str
    birth_date: StoryDate      # 出生时刻(故事内时间)
    natal_chart: TimeGanZhi    # 出生四柱

    # 命理特征(由 MingliOracle 分析生成)
    day_master: str            # 日主天干(代表自身五行属性)
    day_master_element: str    # 日主五行:金/木/水/火/土
    strength: str              # 身强/身弱/中和
    yong_shen: str             # 用神(最有利的五行)
    ji_shen: str               # 忌神(最不利的五行)

    # 命格标签(影响角色命运走向的宏观判断)
    fate_tags: list[str]       # 如:["贵人多助", "大器晚成", "孤克之命", "将星入命"]

    # 大运排列(每步10年)
    major_periods: list[GanZhi]  # 按时间顺序排列的大运,约8-10步

    def get_current_major_period(self, current_story_year: int) -> GanZhi:
        """根据当前故事年份,返回角色正走的大运。"""
        age = current_story_year - self.birth_date.year
        period_index = min(age // 10, len(self.major_periods) - 1)
        return self.major_periods[period_index]


# WorldBible 中的命盘存储
class MingliRegistry:
    charts: dict[str, CharacterBaZi]  # character_id → BaZi

    def register(self, character_id: str, birth_date: StoryDate):
        """初始化时为每个主要角色生成命盘。"""
        chart = MingliOracle.generate_bazi(character_id, birth_date)
        self.charts[character_id] = chart

    def get(self, character_id: str) -> CharacterBaZi:
        return self.charts[character_id]

ML3:DeepSeek API 接口(MingliOracle)

所有命理计算委托给 DeepSeek,本地只做请求构造和响应解析。

DEEPSEEK_SYSTEM_PROMPT = """
你是一位精通子平术的命理学家。请严格按照子平术方法论进行推算,
输出必须是可直接解析的 JSON,不要包含任何解释性文字在 JSON 外部。
"""

class MingliOracle:
    api_key: str
    model: str = "deepseek-chat"
    base_url: str = "https://api.deepseek.com"

    # ── 接口 1:计算指定时刻的干支四柱 ─────────────────────────────
    @staticmethod
    def compute_time_ganzhi(epoch_year_gz: str, date: StoryDate) -> TimeGanZhi:
        prompt = f"""
已知纪年起点(故事元年)对应干支年:{epoch_year_gz}
请推算以下故事时间的四柱干支:
  纪年:第 {date.year} 年,{date.month} 月,{date.day} 日,第 {date.hour_index} 时辰(0=子时,1=丑时…)

使用五虎遁年起月法推算月柱,五鼠遁日起时法推算时柱。
返回 JSON:
{{
  "year_gz":  {{"gan": "甲", "zhi": "子"}},
  "month_gz": {{"gan": "丙", "zhi": "寅"}},
  "day_gz":   {{"gan": "戊", "zhi": "午"}},
  "hour_gz":  {{"gan": "庚", "zhi": "申"}}
}}
"""
        raw = MingliOracle._call(prompt)
        return TimeGanZhi(**{k: GanZhi(**v) for k, v in raw.items()})

    # ── 接口 2:生成角色命盘 ─────────────────────────────────────
    @staticmethod
    def generate_bazi(character_id: str, birth_date: StoryDate,
                      epoch_year_gz: str) -> CharacterBaZi:
        natal = MingliOracle.compute_time_ganzhi(epoch_year_gz, birth_date)
        prompt = f"""
请对以下八字进行子平术分析:
八字:{natal.to_prompt_str()}
角色身份提示:{character_id}(用于选择命格偏向,勿直接引用)

分析内容及 JSON 格式:
{{
  "day_master": "戊",
  "day_master_element": "土",
  "strength": "身强",        // 身强/身弱/中和
  "yong_shen": "火",         // 用神五行
  "ji_shen": "水",           // 忌神五行
  "fate_tags": ["将星入命", "大器晚成"],
  "major_periods": [         // 大运,从起运开始每10年一步,列8步
    {{"gan": "己", "zhi": "未"}},
    {{"gan": "庚", "zhi": "申"}}
  ]
}}
"""
        raw = MingliOracle._call(prompt)
        return CharacterBaZi(
            character_id=character_id,
            birth_date=birth_date,
            natal_chart=natal,
            **raw,
            major_periods=[GanZhi(**p) for p in raw["major_periods"]],
        )

    # ── 接口 3:核心推演 — 当前命格与时柱的吉凶分析 ────────────────
    @staticmethod
    def divine(bazi: CharacterBaZi, current_time: TimeGanZhi,
               scene_context: str) -> "MingliInfluence":
        current_period = bazi.get_current_major_period(...)
        prompt = f"""
命主八字:{bazi.natal_chart.to_prompt_str()}
日主:{bazi.day_master}({bazi.day_master_element}),身{bazi.strength},用神{bazi.yong_shen},忌神{bazi.ji_shen}
当前大运:{current_period.gan}{current_period.zhi}
当前流年流月:{current_time.year_gz.gan}{current_time.year_gz.zhi}年 {current_time.month_gz.gan}{current_time.month_gz.zhi}月
当前时辰:{current_time.day_gz.gan}{current_time.day_gz.zhi}日{current_time.hour_gz.gan}{current_time.hour_gz.zhi}时
当前剧情情境:{scene_context}

请用子平术分析命主此时的运势,重点关注:
1. 当前大运流年对用神/忌神的影响(得令/失令/冲克)
2. 当前时辰对事件成败的即时影响
3. 与情境相关的吉凶征兆

返回 JSON,格式严格如下:
{{
  "luck_power": 0.65,                    // 0.0(极衰)到 1.0(极旺)
  "success_modifier": +0.15,             // 事件成功概率调整量,范围 -0.30 到 +0.30
  "element_state": {{                     // 当前五行力量状态
    "金": 0.3, "木": 0.1, "水": 0.5, "火": 0.6, "土": 0.4
  }},
  "favorable_themes": ["贵人相助", "以柔克刚"],   // 此时适合的剧情主题
  "warning_themes": ["口舌是非", "破财之兆"],      // 此时应避免的剧情主题
  "oracle_fragments": [                  // 可直接嵌入小说占卜场景的诗句/卦辞风格短语
    "水火既济,先济后困",
    "庚金临子,潜龙勿用"
  ],
  "symbolic_imagery": [                  // 可织入叙事的意象(天象/动物/颜色)
    "乌云压顶", "白虎出林", "北方有水"
  ],
  "divination_summary": "此时命主运势...", // 50字内的白话总结,供 Director 参考
  "confidence": 0.8                      // 推算可信度(时辰不明时降低)
}}
"""
        raw = MingliOracle._call(prompt)
        return MingliInfluence(character_id=bazi.character_id,
                               computation_time=current_time, **raw)

    @staticmethod
    def _call(prompt: str) -> dict:
        """统一调用入口,含重试和 JSON 解析。"""
        from openai import OpenAI
        client = OpenAI(api_key=MingliOracle.api_key,
                        base_url=MingliOracle.base_url)
        for attempt in range(3):
            resp = client.chat.completions.create(
                model=MingliOracle.model,
                messages=[
                    {"role": "system", "content": DEEPSEEK_SYSTEM_PROMPT},
                    {"role": "user",   "content": prompt},
                ],
                response_format={"type": "json_object"},
                temperature=0.2,  # 命理推算要稳定,低 temperature
            )
            try:
                return json.loads(resp.choices[0].message.content)
            except json.JSONDecodeError:
                if attempt == 2:
                    raise

ML4:命理影响结构体(MingliInfluence)

Director 和 TextGenerator 消费此结构体:

@dataclass
class MingliInfluence:
    character_id: str
    computation_time: TimeGanZhi

    # ── Director 使用 ────────────────────────────────────────────
    luck_power: float           # 0.0-1.0,当前运势强度
    success_modifier: float     # 事件成功概率调整量 (-0.30 ~ +0.30)
    favorable_themes: list[str] # 适合本章推进的剧情主题(正向权重)
    warning_themes: list[str]   # 应避免的剧情主题(负向权重)
    divination_summary: str     # 50字总结,供 Director 规划时参考

    # ── TextGenerator / 占卜场景使用 ──────────────────────────────
    oracle_fragments: list[str]  # 可直接用于预言/卦辞/铭文的诗化短语
    symbolic_imagery: list[str]  # 可织入环境描写的意象

    # ── 元数据 ─────────────────────────────────────────────────
    element_state: dict[str, float]  # 五行当前力量分布
    confidence: float                # 推算可信度

ML5:与 Director 的集成

命理影响作为软性概率权重,不强制覆盖 Director 决策:

class Director:
    def plan_next_chapter(self, chapter_no: int) -> SceneDirective:
        # ... 常规规划 ...

        # 命理权重注入(SHOULD 级别,不覆盖 MUST/HARD 规则)
        influences = {}
        for char_id in active_characters:
            bazi = self.mingli_registry.get(char_id)
            current_time_gz = self.mingli_calendar.get_time_ganzhi(current_story_date)
            influence = MingliOracle.divine(bazi, current_time_gz, scene_context)
            influences[char_id] = influence

        # 将命理影响注入 SceneDirective 的概率偏置
        directive.mingli_influences = influences
        directive.protagonist_luck = influences[PROTAGONIST_ID].luck_power

        # 命理主题与 YY 七要素选择权重叠加
        # 例:favorable_themes 包含"贵人相助" → 提升本章出现"助人"要素的概率
        directive.element_weights = self._compute_element_weights(influences)

        return directive

    def _compute_element_weights(self, influences: dict) -> dict:
        """
        将命理 favorable/warning_themes 映射到 YY 七要素的概率权重。
        不修改调度器的爽点节奏(MUST 规则),只在同等条件下倾向选择更"合命"的类型。
        """
        weights = {elem: 1.0 for elem in YY_SEVEN_ELEMENTS}
        protagonist_inf = influences.get(PROTAGONIST_ID)
        if protagonist_inf:
            if "贵人相助" in protagonist_inf.favorable_themes:
                weights["助人"] *= 1.4
            if "破财之兆" in protagonist_inf.warning_themes:
                weights["发财"] *= 0.6   # 此时发财情节略压
        return weights

ML6:占卜预言场景的架构级支持

当 Director 规划含占卜/预言的场景时,直接从 MingliInfluence 提取材料:

DIVINATION_SCENE_TEMPLATE = """
本章含占卜/预言场景。请使用以下命理材料构建预言内容,
使预言在风格上符合奇幻世界的神秘感,同时与角色命格有真实对应。

命理材料(禁止直接透露技术细节,转化为世界内的预言语言):
  角色运势强度:{luck_power:.0%}
  适合主题:{favorable_themes}
  警示主题:{warning_themes}
  可用诗句/卦辞:{oracle_fragments}
  可用意象:{symbolic_imagery}

要求:
  - 预言必须模糊(读者事后回看才恍然大悟,不能太直白)
  - oracle_fragments 至少用1条,改写融入世界观语言
  - symbolic_imagery 至少用1个意象织入场景描写
  - 预言的最终应验(章节中的实际结果)需与 luck_power 大致相符
    (luck_power > 0.6 → 预言偏吉;luck_power < 0.4 → 预言偏凶)
"""

# 在 TextGenerator 生成含占卜场景的章节时,自动注入此模板

ML7:命理引擎调用频率与缓存策略

MINGLI_CALL_STRATEGY = {
    # 高频计算(每章调用)
    "per_chapter_divine": True,          # 每章开始前为主角推算当前吉凶
    "cache_same_day_gz": True,           # 同一故事日期内多章共用同一推算结果

    # 低频计算(初始化/特定节点)
    "generate_bazi_on_init": True,       # WorldBible 初始化时为所有主要角色生成命盘
    "regenerate_on_arc_change": False,   # 命盘不因弧光变化而重新生成(命是定数)

    # 特殊场景(按需调用)
    "on_divination_scene": True,         # 占卜场景额外调用,获取 oracle_fragments
    "on_major_climax": True,             # 卷级高潮前推算,影响高潮走向的命运感

    # 成本控制
    "protagonist_only_by_default": True, # 默认只为主角推算;重要 NPC 按需
    "batch_supporting_chars": False,     # 配角命盘批量初始化,运势不每章推算
}

# 估算 DeepSeek API 调用量(百万字小说):
# - 初始化命盘:主角 + 5 主要 NPC = 6 次
# - 每章运势:~400 章 × 1 次 = 400 次
# - 占卜场景额外调用:~20 次
# 总计约 430 次 DeepSeek 调用
# DeepSeek-chat 约 $0.00014/1K tokens,每次约 1K tokens → 约 $0.06 全书

ML8:命理数据在 MemoryGraph 中的存储

命理推算结果存入 MemoryGraph,供日后检索:

# 命盘存为 WORLD_FACT 节点(永不软删除)
bazi_node = MemoryNode(
    node_type=NodeType.WORLD_FACT,
    content=f"{char_id} 命盘:{bazi.natal_chart.to_prompt_str()},日主{bazi.day_master},用神{bazi.yong_shen}",
    importance=1.0,  # 最高重要性,永不被压缩
    metadata={"type": "bazi", "character_id": char_id},
)

# 每章推算结果存为 EVENT 节点(可软删除,但 oracle_fragments 保留)
divine_node = MemoryNode(
    node_type=NodeType.EVENT,
    content=f"第{chapter_no}章命理推算:{influence.divination_summary}",
    importance=0.3 if not has_divination_scene else 0.8,
    metadata={
        "type": "mingli_divine",
        "oracle_fragments": influence.oracle_fragments,  # 永久保留,占卜场景可回溯引用
        "luck_power": influence.luck_power,
    },
)

# 占卜场景建立 FORESHADOWS 边(命理预言 → 未来事件)
# 使得预言和应验之间有显式图关系,ContradictionChecker 可验证应验合理性