#小说：《浮世寸心》
# hetu_luoshu_v7_memory_v11.py
# 河图洛书 V7.11 - 《浮世寸心》
# 一部永远在生长的言情小说
#
# 核心结构不变：火2 → 木3 → 水1 → 金4 → 老师评分
#
# 学习机制：
# 火2（全能知识收集者）：每1轮学习，每1轮输出知识卡片
# 木3（全局热点捕捉者）：每10轮学习，每10轮输出热点报告
# 水1（创意技巧专家）：每100轮学习，每100轮输出章节提纲
# 金4（写作技巧专家）：每1000轮学习，每1000轮输出小说章节
# 老师7/8/6/9：每50轮学习，优化评分标准
#
# 共享图书馆：火2和木3写入，所有人可读
# 学生个人笔记：每个学生自己的学习笔记
# 老师个人记忆：评分历史 + 优化后的评分标准
#
# 禁用姓氏：顾、冯、赵、万、郑、张、谢、关、余、李、扬、龚、刘、姚、蔚、钱、孙
import os
import sys
import time
import json
import random
import re
import math
import hashlib
import pickle
import requests
from collections import Counter
from typing import List, Dict, Tuple, Optional
from datetime import datetime
# ==================== API配置 ====================
DEEPSEEK_API_KEY = "sk-KEY"
DEEPSEEK_API_URL = "https://api.deepseek.com/v1/chat/completions"
# V11独立文件夹
for d in ["cache_v11", "learning_material", "WZ", "logs_v11", "checkpoints_v11", "learned_skills"]:
os.makedirs(d, exist_ok=True)
def call_deepseek(prompt: str, max_tokens: int = 200, temperature: float = 0.7) -> str:
cache_key = hashlib.md5(prompt.encode()).hexdigest()
cache_file = f"cache_v11/{cache_key}.json"
if os.path.exists(cache_file):
try:
with open(cache_file, 'r', encoding='utf-8') as f:
return json.load(f)["response"]
except:
pass
try:
headers = {"Authorization": f"Bearer {DEEPSEEK_API_KEY}", "Content-Type": "application/json"}
data = {
"model": "deepseek-chat",
"messages": [{"role": "user", "content": prompt}],
"max_tokens": max_tokens,
"temperature": temperature
}
response = requests.post(DEEPSEEK_API_URL, json=data, headers=headers, timeout=30)
if response.status_code == 200:
result = response.json()["choices"][0]["message"]["content"]
with open(cache_file, 'w', encoding='utf-8') as f:
json.dump({"prompt": prompt, "response": result}, f, ensure_ascii=False)
return result
return ""
except Exception as e:
return ""
# ==================== 道：π引擎（完全不变）====================
class DaoEngine:
def __init__(self, chunk_size=10000):
self.chunk_size = chunk_size
self.digits = []
self.pointer = 0
self._load_next_chunk()
def _load_next_chunk(self):
try:
import gmpy2
gmpy2.get_context().precision = (self.pointer + self.chunk_size + 100) * 4
pi = gmpy2.const_pi()
pi_str = format(pi, f'.{self.pointer + self.chunk_size + 50}f')
pi_digits = pi_str.replace('.', '')
segment = pi_digits[self.pointer:self.pointer + self.chunk_size]
self.digits.extend([int(ch) for ch in segment])
except ImportError:
from decimal import Decimal, getcontext
getcontext().prec = self.pointer + self.chunk_size + 50
pi = Decimal(0)
for k in range(self.pointer + self.chunk_size + 20):
pi += (Decimal(1)/(16**k)) * (
Decimal(4)/(8*k+1) - Decimal(2)/(8*k+4) -
Decimal(1)/(8*k+5) - Decimal(1)/(8*k+6)
)
pi_str = str(pi)[2:]
segment = pi_str[self.pointer:self.pointer + self.chunk_size]
self.digits.extend([int(ch) for ch in segment])
def get_novelty(self, length=8) -> float:
while self.pointer + length >= len(self.digits):
self._load_next_chunk()
segment = self.digits[self.pointer:self.pointer+length]
self.pointer += length
value = 0
for i, d in enumerate(segment):
value += d * (0.1 ** (i+1))
return value
def get_digit(self) -> int:
if self.pointer >= len(self.digits):
self._load_next_chunk()
digit = self.digits[self.pointer]
self.pointer += 1
return digit
def get_digits(self, count: int) -> List[int]:
result = []
for _ in range(count):
result.append(self.get_digit())
return result
def get_pointer(self) -> int:
return self.pointer
def get_state(self) -> dict:
return {"pointer": self.pointer}
def restore_state(self, state: dict):
self.pointer = state.get("pointer", 0)
self.digits = []
self._load_next_chunk()
# ==================== 节奏控制器（完全不变）====================
class RhythmController:
def __init__(self):
self.sheng_phase = 0
self.bian_phase = 0
self.sheng_speed = 0.2 * 2 * math.pi / 5
self.bian_speed = 2 * math.pi / 1
def update(self):
self.sheng_phase = (self.sheng_phase + self.sheng_speed) % (2 * math.pi)
self.bian_phase = (self.bian_phase + self.bian_speed) % (2 * math.pi)
def get_sheng_ratio(self):
return 0.55 + 0.25 * math.sin(self.sheng_phase)
def get_bian_ratio(self):
return 0.55 + 0.35 * math.sin(self.bian_phase)
def get_sheng_length(self, base_min=15, base_max=40):
return int(base_min + (base_max - base_min) * self.get_sheng_ratio())
def get_bian_length(self, base_min=8, base_max=25):
return int(base_min + (base_max - base_min) * self.get_bian_ratio())
def get_state(self) -> dict:
return {"sheng_phase": self.sheng_phase, "bian_phase": self.bian_phase}
def restore_state(self, state: dict):
self.sheng_phase = state.get("sheng_phase", 0)
self.bian_phase = state.get("bian_phase", 0)
# ==================== 河图中央（完全不变）====================
class HeTuCenter:
def __init__(self):
self.sheng_info = {"1": 0.0, "2": 0.0, "3": 0.0, "4": 0.0}
self.cheng_info = {"6": 0.0, "7": 0.0, "8": 0.0, "9": 0.0}
self.global_state = {"sheng": 0.0, "cheng": 0.0, "balance": 0.0}
def update_sheng(self, idx: int, value: float):
self.sheng_info[str(idx)] = value
self._update_global_state()
def update_cheng(self, idx: int, value: float):
self.cheng_info[str(idx)] = value
self._update_global_state()
def _update_global_state(self):
self.global_state["sheng"] = sum(self.sheng_info.values()) / 4
self.global_state["cheng"] = sum(self.cheng_info.values()) / 4
self.global_state["balance"] = self.global_state["sheng"] / (self.global_state["cheng"] + 0.01)
def get_full_state(self):
return {"sheng": self.sheng_info.copy(), "cheng": self.cheng_info.copy(), "global": self.global_state.copy()}
def get_save_state(self):
return {"sheng_info": self.sheng_info, "cheng_info": self.cheng_info, "global_state": self.global_state}
def restore_state(self, state: dict):
self.sheng_info = state.get("sheng_info", {"1": 0.0, "2": 0.0, "3": 0.0, "4": 0.0})
self.cheng_info = state.get("cheng_info", {"6": 0.0, "7": 0.0, "8": 0.0, "9": 0.0})
self.global_state = state.get("global_state", {"sheng": 0.0, "cheng": 0.0, "balance": 0.0})
# ==================== 工具函数 ====================
def get_all_txt_files_recursive(root_dir: str) -> List[str]:
"""递归获取所有txt文件（遍历所有子文件夹和孙文件夹）"""
txt_files = []
if not os.path.exists(root_dir):
return txt_files
for dirpath, dirnames, filenames in os.walk(root_dir):
for filename in filenames:
if filename.endswith('.txt'):
txt_files.append(os.path.join(dirpath, filename))
return txt_files
# ==================== 共享图书馆 ====================
class SharedLibrary:
"""共享图书馆：火2和木3写入，所有人可读"""
# 禁用姓氏
FORBIDDEN_SURNAMES = {'顾', '冯', '赵', '万', '郑', '张', '谢', '关', '余', '李', '扬', '龚', '刘', '姚', '蔚', '钱', '孙'}
def __init__(self, max_size: int = 10000):
self.knowledge_cards = [] # 火2的知识卡片
self.hot_reports = [] # 木3的热点报告
self.max_size = max_size
def add_knowledge_card(self, card: str, round_num: int):
"""火2写入知识卡片"""
if card and card not in [k["card"] for k in self.knowledge_cards]:
self.knowledge_cards.append({
"card": card,
"round": round_num,
"timestamp": time.time()
})
if len(self.knowledge_cards) > self.max_size:
self.knowledge_cards = self.knowledge_cards[-self.max_size:]
def add_hot_report(self, report: str, round_num: int):
"""木3写入热点报告"""
if report:
self.hot_reports.append({
"report": report,
"round": round_num,
"timestamp": time.time()
})
if len(self.hot_reports) > self.max_size // 10:
self.hot_reports = self.hot_reports[-self.max_size // 10:]
def get_recent_knowledge(self, n: int = 10) -> List[str]:
"""获取最近n条知识卡片"""
return [k["card"] for k in self.knowledge_cards[-n:]]
def get_recent_hot_reports(self, n: int = 3) -> List[str]:
"""获取最近n份热点报告"""
return [h["report"] for h in self.hot_reports[-n:]]
def get_all_knowledge_text(self) -> str:
"""获取所有知识卡片的文本"""
return "\n".join([k["card"] for k in self.knowledge_cards[-50:]])
def get_all_hot_reports_text(self) -> str:
"""获取所有热点报告的文本"""
return "\n".join([h["report"] for h in self.hot_reports[-10:]])
def _is_forbidden_name(self, name: str) -> bool:
if not name or len(name) < 2:
return False
surname = name[0]
if len(name) >= 2 and name[:2] in ['欧阳', '慕容', '上官', '司马', '东方', '独孤', '南宫']:
surname = name[:2]
return surname in self.FORBIDDEN_SURNAMES
def get_forbidden_surnames_text(self) -> str:
return "、".join(sorted(self.FORBIDDEN_SURNAMES))
def get_state(self) -> dict:
return {
"knowledge_cards": self.knowledge_cards[-1000:],
"hot_reports": self.hot_reports[-100:]
}
def restore_state(self, state: dict):
self.knowledge_cards = state.get("knowledge_cards", [])
self.hot_reports = state.get("hot_reports", [])
# ==================== 学生个人笔记 ====================
class StudentPersonalNote:
"""每个学生自己的个人笔记"""
def __init__(self, student_name: str):
self.student_name = student_name
self.notes = [] # 个人学习笔记
self.last_learn_round = 0
def add_note(self, note: str, round_num: int):
if note:
self.notes.append({
"note": note,
"round": round_num,
"timestamp": time.time()
})
if len(self.notes) > 500:
self.notes = self.notes[-500:]
self.last_learn_round = round_num
def get_recent_notes(self, n: int = 10) -> List[str]:
return [n["note"] for n in self.notes[-n:]]
def get_state(self) -> dict:
return {"notes": self.notes[-500:], "last_learn_round": self.last_learn_round}
def restore_state(self, state: dict):
self.notes = state.get("notes", [])
self.last_learn_round = state.get("last_learn_round", 0)
# ==================== 老师个人记忆 ====================
class TeacherPersonalMemory:
"""每个老师自己的记忆（评分历史 + 评分标准）"""
def __init__(self, teacher_id: int, student_name: str):
self.teacher_id = teacher_id
self.student_name = student_name
self.score_history = [] # 评分历史
self.scoring_standards = [] # 优化后的评分标准
self.student_common_issues = [] # 学生常见问题
self.last_learn_round = 0
def add_score(self, round_num: int, work: str, score: float, scores_detail: dict, comment: str):
self.score_history.append({
"round": round_num,
"work_preview": work[:100],
"score": score,
"scores_detail": scores_detail,
"comment": comment,
"timestamp": time.time()
})
if len(self.score_history) > 500:
self.score_history = self.score_history[-500:]
def get_recent_scores(self, n: int = 20) -> List[dict]:
return self.score_history[-n:]
def add_standard(self, standard: str, round_num: int):
if standard and standard not in [s["standard"] for s in self.scoring_standards[-10:]]:
self.scoring_standards.append({
"standard": standard,
"round": round_num
})
if len(self.scoring_standards) > 50:
self.scoring_standards = self.scoring_standards[-50:]
def add_issue(self, issue: str, round_num: int):
if issue:
self.student_common_issues.append({
"issue": issue,
"round": round_num
})
if len(self.student_common_issues) > 20:
self.student_common_issues = self.student_common_issues[-20:]
def get_scoring_standards_text(self) -> str:
return "\n".join([s["standard"] for s in self.scoring_standards])
def get_common_issues_text(self) -> str:
return "\n".join([i["issue"] for i in self.student_common_issues])
def get_state(self) -> dict:
return {
"score_history": self.score_history[-500:],
"scoring_standards": self.scoring_standards,
"student_common_issues": self.student_common_issues,
"last_learn_round": self.last_learn_round
}
def restore_state(self, state: dict):
self.score_history = state.get("score_history", [])
self.scoring_standards = state.get("scoring_standards", [])
self.student_common_issues = state.get("student_common_issues", [])
self.last_learn_round = state.get("last_learn_round", 0)
# ==================== 火2（全能知识收集者）====================
class Fire2WithMemory:
def __init__(self):
self.shared_library = None
self.personal_note = StudentPersonalNote("火2")
self.last_output_round = 0
self.output_interval = 1 # 每1轮输出
self.read_files = set() # 已读文件记录（存储完整路径）
def set_shared_library(self, library: SharedLibrary):
self.shared_library = library
def _get_all_txt_files(self, root_dir: str) -> List[str]:
"""递归获取所有txt文件（遍历所有子文件夹和孙文件夹）"""
txt_files = []
if not os.path.exists(root_dir):
return txt_files
for dirpath, dirnames, filenames in os.walk(root_dir):
for filename in filenames:
if filename.endswith('.txt'):
txt_files.append(os.path.join(dirpath, filename))
return txt_files
def _learn_from_files(self, round_num: int) -> List[str]:
"""从 learning_material 文件夹（递归所有子文件夹）学习"""
learned = []
learning_dir = "learning_material"
if not os.path.exists(learning_dir):
return learned
# 递归获取所有txt文件
all_files = self._get_all_txt_files(learning_dir)
if not all_files:
return learned
# 找一个未读的文件
unread_files = [f for f in all_files if f not in self.read_files]
if not unread_files:
# 都读过了，重新开始
self.read_files.clear()
unread_files = all_files
target_file = random.choice(unread_files)
self.read_files.add(target_file)
try:
with open(target_file, 'r', encoding='utf-8') as f:
content = f.read(3000)
# 获取相对路径用于显示
rel_path = os.path.relpath(target_file, learning_dir)
prompt = f"""从以下文本中提取写作知识或技巧，每条一句话，最多5条：
{content[:2000]}
输出格式：每行一条知识，不要编号："""
result = call_deepseek(prompt, max_tokens=300, temperature=0.5)
if result:
for line in result.strip().split('\n'):
line = line.strip()
if line and len(line) > 5:
learned.append(line)
self.personal_note.add_note(f"从{rel_path}学到：{line}", round_num)
if self.shared_library:
self.shared_library.add_knowledge_card(line, round_num)
except Exception as e:
pass
return learned
def _generate_knowledge_card(self, round_num: int, dao_novelty: float, learned: List[str]) -> Optional[str]:
"""生成知识卡片"""
if learned:
# 如果学到了新知识，直接作为知识卡片输出
card = learned[0]
else:
# 否则让AI生成一条知识
prompt = f"""生成一条言情小说写作知识或技巧，用于长篇连载言情小说《浮世寸心》。
要求：
1. 内容具体、实用
2. 30字以内
3. 只输出内容，不要解释："""
result = call_deepseek(prompt, max_tokens=50, temperature=0.5 + dao_novelty * 0.3)
if result:
card = result.strip()
self.personal_note.add_note(f"AI生成：{card}", round_num)
if self.shared_library:
self.shared_library.add_knowledge_card(card, round_num)
else:
card = None
return card
def learn_and_output(self, round_num: int, dao_novelty: float) -> Optional[str]:
"""每1轮：学习 + 输出知识卡片"""
# 学习：从 learning_material 文件夹读取文件
learned = self._learn_from_files(round_num)
# 输出：生成知识卡片
if round_num - self.last_output_round >= self.output_interval:
self.last_output_round = round_num
return self._generate_knowledge_card(round_num, dao_novelty, learned)
return None
def get_state(self) -> dict:
return {
"last_output_round": self.last_output_round,
"personal_note": self.personal_note.get_state(),
"read_files": list(self.read_files)
}
def restore_state(self, state: dict):
self.last_output_round = state.get("last_output_round", 0)
self.personal_note.restore_state(state.get("personal_note", {}))
self.read_files = set(state.get("read_files", []))
# ==================== 木3（全局热点捕捉者）====================
class Mu3WithMemory:
def __init__(self):
self.shared_library = None
self.personal_note = StudentPersonalNote("木3")
self.last_output_round = 0
self.output_interval = 10 # 每10轮输出
self.last_hot_capture_round = 0
def set_shared_library(self, library: SharedLibrary):
self.shared_library = library
def _capture_hot_topics(self, round_num: int):
"""捕捉网上热点（用AI模拟当前流行趋势）"""
prompt = f"""请列出当前网上流行的3个热点话题或流行趋势（适合融入言情小说）。
要求：
1. 每个热点一句话
2. 涉及大众情绪、流行话题、生活热点
3. 输出格式：每行一个热点
例如：
- 职场年轻人的精神内耗
- 女性独立意识觉醒
- 慢生活、治愈系成为新追求"""
result = call_deepseek(prompt, max_tokens=200, temperature=0.7)
if result:
for line in result.strip().split('\n'):
line = line.strip()
if line and len(line) > 5:
self.personal_note.add_note(f"热点捕捉：{line}", round_num)
def _generate_hot_report(self, round_num: int, dao_novelty: float) -> Optional[str]:
"""生成热点报告"""
# 从个人笔记中获取最近的热点
recent_notes = self.personal_note.get_recent_notes(10)
hot_topics = [n for n in recent_notes if "热点捕捉" in n]
hot_text = "\n".join(hot_topics[-5:]) if hot_topics else "暂无热点数据"
prompt = f"""根据以下热点，生成一份言情小说创作热点报告（100字左右）：
{hot_text}
要求：
1. 总结当前适合融入言情小说的流行元素
2. 给出创作建议
3. 只输出报告内容："""
result = call_deepseek(prompt, max_tokens=150, temperature=0.6 + dao_novelty * 0.3)
if result:
report = result.strip()
self.personal_note.add_note(f"热点报告：{report[:100]}", round_num)
if self.shared_library:
self.shared_library.add_hot_report(report, round_num)
return report
return None
def learn_and_output(self, round_num: int, dao_novelty: float) -> Optional[str]:
"""每10轮：学习（捕捉热点）+ 输出热点报告"""
# 学习：捕捉网上热点
if round_num - self.last_hot_capture_round >= self.output_interval:
self.last_hot_capture_round = round_num
self._capture_hot_topics(round_num)
# 输出：生成热点报告
if round_num - self.last_output_round >= self.output_interval:
self.last_output_round = round_num
return self._generate_hot_report(round_num, dao_novelty)
return None
def get_state(self) -> dict:
return {
"last_output_round": self.last_output_round,
"last_hot_capture_round": self.last_hot_capture_round,
"personal_note": self.personal_note.get_state()
}
def restore_state(self, state: dict):
self.last_output_round = state.get("last_output_round", 0)
self.last_hot_capture_round = state.get("last_hot_capture_round", 0)
self.personal_note.restore_state(state.get("personal_note", {}))
# ==================== 水1（创意技巧专家）====================
class Shui1WithMemory:
def __init__(self):
self.shared_library = None
self.personal_note = StudentPersonalNote("水1")
self.last_output_round = 0
self.output_interval = 100 # 每100轮输出
self.pond = [] # 章节提纲池
def set_shared_library(self, library: SharedLibrary):
self.shared_library = library
def learn(self, round_num: int, dao_novelty: float):
"""每轮学习：从共享图书馆学习"""
if not self.shared_library:
return
# 获取最近的知识和热点
knowledge = self.shared_library.get_recent_knowledge(5)
hot_reports = self.shared_library.get_recent_hot_reports(2)
if knowledge or hot_reports:
learn_text = "\n".join(knowledge + hot_reports)
prompt = f"""根据以下知识和热点，学习创意技巧：
{learn_text[:500]}
总结1条创意技巧（情节设计、人物设定、情感冲突等），一句话："""
result = call_deepseek(prompt, max_tokens=100, temperature=0.6)
if result:
self.personal_note.add_note(f"创意学习：{result.strip()}", round_num)
def output(self, round_num: int, dao_novelty: float) -> Optional[List[Dict]]:
"""每100轮输出章节提纲"""
if round_num - self.last_output_round < self.output_interval:
return None
if not self.shared_library:
return None
self.last_output_round = round_num
# 获取共享图书馆的知识和热点
knowledge = self.shared_library.get_recent_knowledge(10)
hot_reports = self.shared_library.get_recent_hot_reports(3)
personal_notes = self.personal_note.get_recent_notes(10)
knowledge_text = "\n".join(knowledge) if knowledge else "暂无"
hot_text = "\n".join(hot_reports) if hot_reports else "暂无"
notes_text = "\n".join(personal_notes) if personal_notes else "暂无"
prompt = f"""作为创意技巧专家，根据以下素材生成3个不同的小说章节提纲。
【知识库】
{knowledge_text[:800]}
【热点报告】
{hot_text[:500]}
【个人学习笔记】
{notes_text[:300]}
要求：
1. 每个提纲包括：标题（8-15字）、情节概要（50字）
2. 3个提纲要有不同的创意方向
3. 不要带“第X章”字样
输出格式（每行一个提纲，用|分隔）：
标题|情节概要"""
result = call_deepseek(prompt, max_tokens=800, temperature=0.8 + dao_novelty * 0.3)
self.pond = []
if result:
for line in result.strip().split('\n'):
if '|' in line:
parts = line.split('|')
if len(parts) >= 2:
title = parts[0].strip()
outline = parts[1].strip()
self.pond.append({
"title": title,
"plot_outline": outline,
"score": 0.5,
"round": round_num
})
# 确保有3个提纲
while len(self.pond) < 3:
self.pond.append({
"title": f"创意{len(self.pond)+1}",
"plot_outline": "待补充",
"score": 0.5,
"round": round_num
})
self.pond = self.pond[:3]
self.personal_note.add_note(f"生成了{len(self.pond)}个章节提纲", round_num)
return self.pond.copy()
def get_pond(self) -> List[Dict]:
return self.pond.copy()
def get_state(self) -> dict:
return {
"last_output_round": self.last_output_round,
"pond": self.pond,
"personal_note": self.personal_note.get_state()
}
def restore_state(self, state: dict):
self.last_output_round = state.get("last_output_round", 0)
self.pond = state.get("pond", [])
self.personal_note.restore_state(state.get("personal_note", {}))
# ==================== 金4（写作技巧专家）====================
class Jin4WithMemory:
def __init__(self):
self.shared_library = None
self.personal_note = StudentPersonalNote("金4")
self.last_output_round = 0
self.output_interval = 1000 # 每1000轮输出
self.masterpieces = []
def set_shared_library(self, library: SharedLibrary):
self.shared_library = library
def learn(self, round_num: int, dao_novelty: float):
"""每轮学习：从共享图书馆学习写作技巧"""
if not self.shared_library:
return
knowledge = self.shared_library.get_recent_knowledge(10)
hot_reports = self.shared_library.get_recent_hot_reports(3)
if knowledge or hot_reports:
learn_text = "\n".join(knowledge + hot_reports)
prompt = f"""根据以下知识，学习写作技巧：
{learn_text[:500]}
总结1条写作技巧（文笔、节奏、对话、描写等），一句话："""
result = call_deepseek(prompt, max_tokens=100, temperature=0.6)
if result:
self.personal_note.add_note(f"写作学习：{result.strip()}", round_num)
def output(self, round_num: int, dao_novelty: float, pond: List[Dict]) -> Tuple[Optional[str], Optional[Dict]]:
"""每1000轮输出小说章节"""
if round_num - self.last_output_round < self.output_interval:
return None, None
if not pond:
return None, None
self.last_output_round = round_num
# 获取下一章编号
chapter_num = len(self.masterpieces) + 1
# 选择提纲（选分数最高的）
selected = max(pond, key=lambda x: x.get("score", 0))
print(f" 💎 金4: 准备写第{chapter_num}章《{selected['title']}》")
# 获取前文背景
prev_context = self._get_previous_context()
# 获取学到的写作技巧
writing_tips = self.personal_note.get_recent_notes(10)
tips_text = "\n".join(writing_tips[-5:]) if writing_tips else ""
# 获取禁用姓氏
forbidden_text = ""
if self.shared_library:
forbidden_text = f"禁止使用以下姓氏：{self.shared_library.get_forbidden_surnames_text()}\n请使用其他姓氏（如：林、苏、陈、王、江、宋、周、吴、许、沈、陆、程、姜、白、秦等）\n"
prompt = f"""你是言情小说《浮世寸心》的写手。请写第{chapter_num}章。
{prev_context}
{forbidden_text}
【学到的写作技巧】
{tips_text}
【本章信息】
标题：{selected['title']}
情节概要：{selected['plot_outline']}
要求：
1. 严格承接前文，保持人物姓名一致
2. 人物姓氏不得使用禁止列表中的姓氏
3. 章节正文第一行必须是“第{chapter_num}章 {selected['title']}”，不要加#号
4. 第一行之后空一行，再接正文
5. 语言优美流畅，符合言情小说风格
6. 情感真挚，有感染力
7. 情节完整，有起承转合
8. 结尾留有悬念
9. 字数在2500字左右
只输出章节正文："""
temp = 0.7 + dao_novelty * 0.3
result = call_deepseek(prompt, max_tokens=3800, temperature=min(1.0, temp))
if not result or len(result) < 1200:
return None, None
# 保存章节
safe_title = re.sub(r'[\\/*?:"<>|]', '', selected["title"][:20])
wz_file = f"WZ/第{chapter_num:03d}章_{safe_title}.txt"
with open(wz_file, 'w', encoding='utf-8') as f:
f.write(result)
self.masterpieces.append({
"chapter_num": chapter_num,
"title": selected["title"],
"content_preview": result[:200]
})
# 提取章节摘要存入个人笔记
self.personal_note.add_note(f"第{chapter_num}章《{selected['title']}》已写，{len(result)}字", round_num)
print(f" 已保存: {wz_file}，{len(result)}字")
return result, selected
def _get_previous_context(self) -> str:
"""获取前文背景"""
if not self.masterpieces:
return "这是小说的第一章，还没有前文。"
context = "【前文回顾】\n"
for ch in self.masterpieces[-3:]:
context += f"第{ch['chapter_num']}章《{ch['title']}》：{ch['content_preview'][:150]}...\n\n"
context += "【当前状态】\n"
context += f"上一章是第{self.masterpieces[-1]['chapter_num']}章《{self.masterpieces[-1]['title']}》\n"
return context
def get_state(self) -> dict:
return {
"last_output_round": self.last_output_round,
"masterpieces": self.masterpieces[-100:],
"personal_note": self.personal_note.get_state()
}
def restore_state(self, state: dict):
self.last_output_round = state.get("last_output_round", 0)
self.masterpieces = state.get("masterpieces", [])
self.personal_note.restore_state(state.get("personal_note", {}))
# ==================== 老师（四维度评分 + 学习）====================
class TeacherWithMemory:
def __init__(self, teacher_id: int, student_name: str):
self.id = teacher_id
self.student_name = student_name
self.personal_memory = TeacherPersonalMemory(teacher_id, student_name)
self.shared_library = None
self.learn_interval = 50 # 每50轮学习
self.last_learn_round = 0
def set_shared_library(self, library: SharedLibrary):
self.shared_library = library
def evaluate(self, work: str, dao_novelty: float, round_num: int) -> Tuple[float, str, dict]:
"""四维度评分，并记录历史"""
# 每50轮学习一次
if round_num - self.last_learn_round >= self.learn_interval:
self.last_learn_round = round_num
self._learn(round_num)
# 获取评分标准
standards = self.personal_memory.get_scoring_standards_text()
issues = self.personal_memory.get_common_issues_text()
standards_text = f"【评分标准参考】\n{standards}\n" if standards else ""
issues_text = f"【学生常见问题】\n{issues}\n" if issues else ""
prompt = f"""你是老师{self.id}，评判{self.student_name}的言情小说创作。请从四个维度评分（0-1分）：
{standards_text}
{issues_text}
作品：{work[:800]}
输出格式（四个分数和评语，用|分隔）：
创意|文笔|结构|深度|评语
评分标准：
- 创意：情节新颖度、情感设计独特度
- 文笔：语言优美度、描写生动度
- 结构：节奏把控、起承转合
- 深度：情感厚度、主题内涵
只输出数字和评语："""
result = call_deepseek(prompt, max_tokens=150, temperature=0.4)
creativity = 0.5
writing = 0.5
structure = 0.5
depth = 0.5
comment = ""
if result and '|' in result:
parts = result.split('|')
if len(parts) >= 5:
try:
creativity = float(parts[0].strip())
writing = float(parts[1].strip())
structure = float(parts[2].strip())
depth = float(parts[3].strip())
comment = parts[4].strip()[:100]
except:
pass
creativity = min(1.0, max(0.0, creativity))
writing = min(1.0, max(0.0, writing))
structure = min(1.0, max(0.0, structure))
depth = min(1.0, max(0.0, depth))
score = (creativity + writing + structure + depth) / 4
score = score * (0.8 + dao_novelty * 0.3)
score = min(1.0, max(0.0, score))
scores_detail = {
"creativity": creativity,
"writing": writing,
"structure": structure,
"depth": depth
}
# 记录评分历史
self.personal_memory.add_score(round_num, work, score, scores_detail, comment)
return score, comment, scores_detail
def _learn(self, round_num: int):
"""老师每50轮学习：分析评分历史，优化评分标准"""
recent_scores = self.personal_memory.get_recent_scores(20)
if len(recent_scores) < 10:
return
# 分析评分趋势
high_scores = [s for s in recent_scores if s["score"] > 0.8]
low_scores = [s for s in recent_scores if s["score"] < 0.5]
learn_text = ""
if high_scores:
prompt = f"""分析以下高分作品的共同特点，总结评分标准：
{high_scores[:5]}
输出1条评分标准："""
result = call_deepseek(prompt, max_tokens=100, temperature=0.5)
if result:
self.personal_memory.add_standard(result.strip(), round_num)
learn_text += f"评分标准优化：{result.strip()}\n"
if low_scores:
prompt = f"""分析以下低分作品的常见问题，总结学生需要改进的地方：
{low_scores[:5]}
输出1条常见问题："""
result = call_deepseek(prompt, max_tokens=100, temperature=0.5)
if result:
self.personal_memory.add_issue(result.strip(), round_num)
learn_text += f"学生问题发现：{result.strip()}\n"
if learn_text:
print(f" 📖 老师{self.id}学习了：{learn_text[:80]}...")
def get_state(self) -> dict:
return {
"personal_memory": self.personal_memory.get_state(),
"last_learn_round": self.last_learn_round
}
def restore_state(self, state: dict):
self.personal_memory.restore_state(state.get("personal_memory", {}))
self.last_learn_round = state.get("last_learn_round", 0)
# ==================== 洛书中心 ====================
class LuoShuCenterMemoryV11:
def __init__(self, dao: DaoEngine, checkpoint_dir: str = "checkpoints_v11"):
self.dao = dao
self.hetu_center = HeTuCenter()
self.rhythm = RhythmController()
self.checkpoint_dir = checkpoint_dir
os.makedirs(checkpoint_dir, exist_ok=True)
print("\n" + "="*70)
print("📚 河图洛书 V7.11 - 《浮世寸心》")
print(" 一部永远在生长的言情小说")
print("")
print(" 🔥 火2（全能知识收集者）: 每1轮学习+输出")
print(" 🌳 木3（全局热点捕捉者）: 每10轮学习+输出")
print(" 🌊 水1（创意技巧专家）: 每100轮输出")
print(" 💎 金4（写作技巧专家）: 每1000轮输出")
print(" 👨🏫 老师7/8/6/9: 每50轮学习优化评分标准")
print(" 📚 共享图书馆: 火2+木3写入，所有人可读")
print(" 📝 学生个人笔记: 每人独立")
print(" 🚫 禁用姓氏: 顾、冯、赵、万、郑、张、谢、关、余、李、扬、龚、刘、姚、蔚、钱、孙")
print(" ☯ 道驱动一切，按Ctrl+C停止")
print("="*70)
# 共享图书馆
self.shared_library = SharedLibrary()
# 4个学生
self.fire2 = Fire2WithMemory()
self.mu3 = Mu3WithMemory()
self.shui1 = Shui1WithMemory()
self.jin4 = Jin4WithMemory()
# 注入共享图书馆
self.fire2.set_shared_library(self.shared_library)
self.mu3.set_shared_library(self.shared_library)
self.shui1.set_shared_library(self.shared_library)
self.jin4.set_shared_library(self.shared_library)
# 4个老师
self.teacher7 = TeacherWithMemory(7, "火2")
self.teacher8 = TeacherWithMemory(8, "木3")
self.teacher6 = TeacherWithMemory(6, "水1")
self.teacher9 = TeacherWithMemory(9, "金4")
# 注入共享图书馆给老师
self.teacher7.set_shared_library(self.shared_library)
self.teacher8.set_shared_library(self.shared_library)
self.teacher6.set_shared_library(self.shared_library)
self.teacher9.set_shared_library(self.shared_library)
self.round = 0
self.log_entries = []
self._load_checkpoint()
def _get_checkpoint_path(self) -> str:
return os.path.join(self.checkpoint_dir, "full_checkpoint_v11.pkl")
def save_checkpoint(self):
checkpoint = {
"round": self.round,
"dao_state": self.dao.get_state(),
"rhythm_state": self.rhythm.get_state(),
"hetu_state": self.hetu_center.get_save_state(),
"shared_library_state": self.shared_library.get_state(),
"fire2_state": self.fire2.get_state(),
"mu3_state": self.mu3.get_state(),
"shui1_state": self.shui1.get_state(),
"jin4_state": self.jin4.get_state(),
"teacher7_state": self.teacher7.get_state(),
"teacher8_state": self.teacher8.get_state(),
"teacher6_state": self.teacher6.get_state(),
"teacher9_state": self.teacher9.get_state(),
"log_entries": self.log_entries[-100:],
"timestamp": datetime.now().isoformat()
}
tmp_path = self._get_checkpoint_path() + ".tmp"
with open(tmp_path, 'wb') as f:
pickle.dump(checkpoint, f)
os.replace(tmp_path, self._get_checkpoint_path())
def _load_checkpoint(self):
if not os.path.exists(self._get_checkpoint_path()):
print(" 📂 未找到检查点，从头开始")
print(" 📖 开始创作《浮世寸心》...")
return
try:
with open(self._get_checkpoint_path(), 'rb') as f:
checkpoint = pickle.load(f)
self.round = checkpoint.get("round", 0)
self.dao.restore_state(checkpoint.get("dao_state", {}))
self.rhythm.restore_state(checkpoint.get("rhythm_state", {}))
self.hetu_center.restore_state(checkpoint.get("hetu_state", {}))
if "shared_library_state" in checkpoint:
self.shared_library.restore_state(checkpoint["shared_library_state"])
self.fire2.restore_state(checkpoint.get("fire2_state", {}))
self.mu3.restore_state(checkpoint.get("mu3_state", {}))
self.shui1.restore_state(checkpoint.get("shui1_state", {}))
self.jin4.restore_state(checkpoint.get("jin4_state", {}))
self.teacher7.restore_state(checkpoint.get("teacher7_state", {}))
self.teacher8.restore_state(checkpoint.get("teacher8_state", {}))
self.teacher6.restore_state(checkpoint.get("teacher6_state", {}))
self.teacher9.restore_state(checkpoint.get("teacher9_state", {}))
self.log_entries = checkpoint.get("log_entries", [])
print(f" 📂 加载检查点成功，从第 {self.round} 轮继续")
print(f" 《浮世寸心》已写章节: {len(self.jin4.masterpieces)} 章")
except Exception as e:
print(f" ⚠️ 加载检查点失败: {e}，从头开始")
def run_cycle(self):
self.round += 1
dao_novelty = self.dao.get_novelty(6)
self.rhythm.update()
print(f"\n{'─'*70}")
print(f"🧠 第 {self.round} 轮 | 道新奇度: {dao_novelty:.4f}")
# ===== 火2：每1轮学习+输出 =====
knowledge_card = self.fire2.learn_and_output(self.round, dao_novelty)
if knowledge_card:
score7, comment7, scores7 = self.teacher7.evaluate(knowledge_card, dao_novelty, self.round)
self.hetu_center.update_sheng(1, score7)
self.hetu_center.update_cheng(7, score7)
print(f" 🔥 火2: 输出知识卡片「{knowledge_card[:40]}...」| 师7:{score7:.3f}")
else:
self.hetu_center.update_sheng(1, 0.5)
self.hetu_center.update_cheng(7, 0.5)
# ===== 木3：每10轮学习+输出 =====
hot_report = self.mu3.learn_and_output(self.round, dao_novelty)
if hot_report:
score8, comment8, scores8 = self.teacher8.evaluate(hot_report, dao_novelty, self.round)
self.hetu_center.update_sheng(2, score8)
self.hetu_center.update_cheng(8, score8)
print(f" 🌳 木3: 输出热点报告「{hot_report[:40]}...」| 师8:{score8:.3f}")
else:
self.hetu_center.update_sheng(2, 0.5)
self.hetu_center.update_cheng(8, 0.5)
# ===== 水1：每轮学习，每100轮输出 =====
self.shui1.learn(self.round, dao_novelty)
pond = self.shui1.output(self.round, dao_novelty)
if pond:
# 老师6评价水1的提纲
outline_text = "\n".join([f"{p['title']}|{p['plot_outline']}" for p in pond])
score6, comment6, scores6 = self.teacher6.evaluate(outline_text, dao_novelty, self.round)
self.hetu_center.update_sheng(3, score6)
self.hetu_center.update_cheng(6, score6)
print(f" 🌊 水1: 输出了{len(pond)}个章节提纲 | 师6:{score6:.3f}")
else:
self.hetu_center.update_sheng(3, 0.5)
self.hetu_center.update_cheng(6, 0.5)
# ===== 金4：每轮学习，每1000轮输出 =====
self.jin4.learn(self.round, dao_novelty)
chapter, selected = self.jin4.output(self.round, dao_novelty, self.shui1.get_pond())
if chapter:
score9, comment9, scores9 = self.teacher9.evaluate(chapter, dao_novelty, self.round)
self.hetu_center.update_sheng(4, score9)
self.hetu_center.update_cheng(9, score9)
print(f" 💎 金4: 写了{len(chapter)}字章节 | 师9:{score9:.3f}")
else:
self.hetu_center.update_sheng(4, 0.5)
self.hetu_center.update_cheng(9, 0.5)
# 汇总
full_state = self.hetu_center.get_full_state()
sheng_str = f"{full_state['sheng']['1']:.2f}/{full_state['sheng']['2']:.2f}/{full_state['sheng']['3']:.2f}/{full_state['sheng']['4']:.2f}"
cheng_str = f"{full_state['cheng']['6']:.2f}/{full_state['cheng']['7']:.2f}/{full_state['cheng']['8']:.2f}/{full_state['cheng']['9']:.2f}"
print(f" 📊 汇总 | 生:[{sheng_str}] | 成:[{cheng_str}]")
print(f" 📖 《浮世寸心》已写{len(self.jin4.masterpieces)}章 | 共享图书馆:知识{len(self.shared_library.knowledge_cards)}条,热点{len(self.shared_library.hot_reports)}份")
self.log_entries.append({
"round": self.round,
"dao_novelty": dao_novelty,
"chapters_written": len(self.jin4.masterpieces)
})
if self.round % 10 == 0:
self.save_checkpoint()
if self.round % 100 == 0:
self.save_log()
def save_log(self):
with open(f"logs_v11/run_log_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json", 'w', encoding='utf-8') as f:
json.dump(self.log_entries[-500:], f, ensure_ascii=False, indent=2)
print(f"\n 📝 日志已保存，已写{len(self.jin4.masterpieces)}章")
def run_forever(self):
print("\n🚀 启动！按 Ctrl+C 停止\n")
print("📖 《浮世寸心》—— 一部永远在生长的言情小说\n")
try:
while True:
self.run_cycle()
time.sleep(0.3)
except KeyboardInterrupt:
print(f"\n\n⏸️ 停止。运行了 {self.round} 轮")
print(f" 《浮世寸心》已写章节: {len(self.jin4.masterpieces)} 章")
print(f" 共享图书馆: 知识{len(self.shared_library.knowledge_cards)}条, 热点{len(self.shared_library.hot_reports)}份")
self.save_checkpoint()
self.save_log()
print("\n 状态已保存，下次运行继续")
print(" 《浮世寸心》—— 未完待续...")
def main():
print("\n" + "="*70)
print("🐉 河图洛书 V7.11 - 《浮世寸心》")
print(" 一部永远在生长的言情小说")
print(" 火2(全能收集)→木3(热点捕捉)→水1(创意)→金4(写作)")
print(" 老师7/8/6/9 每50轮学习优化")
print(" 共享图书馆 + 学生个人笔记 + 老师个人记忆")
print("="*70 + "\n")
dao = DaoEngine()
luoshu = LuoShuCenterMemoryV11(dao)
luoshu.run_forever()
if __name__ == "__main__":
main()
|
鄂公网安备42060002000282号 )
