# hetu_novel_amplifier.py
# 头脑风暴 · 河图洛书放大器 V1.0
# 设计哲学：道的镜像。道驱动一切，万物自己演化。
# 火2：从masterpieces取句子（10-30句），每1000轮重新扫描，自动清理失效文件
# 木3：组合成文章（≤3000字）
# 水1：润色白化文（≤3000字）
# 金4：固化章节（≤3000字）
import os
import sys
import time
import json
import random
import re
import math
import hashlib
import pickle
import shutil
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"
# ==================== 独立文件夹 ====================
RECOVERY_DIR = "recovery_放大器"
for d in ["cache_放大器", "masterpieces", "output_放大器", "logs_放大器", "checkpoints_放大器", RECOVERY_DIR]:
os.makedirs(d, exist_ok=True)
def call_deepseek(prompt: str, max_tokens: int = 4096, temperature: float = 0.75) -> str:
"""调用API，超时120秒，最大输出4096 tokens（约3000字）"""
cache_key = hashlib.md5(prompt.encode()).hexdigest()
cache_file = f"cache_放大器/{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
}
# 超时设置为120秒，确保生成3000字内容
response = requests.post(DEEPSEEK_API_URL, json=data, headers=headers, timeout=120)
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):
print(f" 🔄 正在加载π第 {self.pointer//self.chunk_size + 1} 批小数位...")
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])
return
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, sheng_min, sheng_max):
return int(sheng_min + (sheng_max - sheng_min) * self.get_sheng_ratio())
def get_bian_length(self, bian_min, bian_max):
return int(bian_min + (bian_max - bian_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})
# ==================== 火2：从masterpieces取句子（按需读取，自动清理失效文件） ====================
class Fire2:
def __init__(self, corpus_paths: List[str] = None):
self.corpus_paths = corpus_paths or ["masterpieces"]
self.file_paths = [] # 只存文件名，不读内容
self.last_scan_round = 0
self._scan_files()
print(f" 🔥 火2完成，发现 {len(self.file_paths)} 个文件")
def _scan_files(self):
"""扫描文件夹，只收集文件路径"""
all_files = []
for path in self.corpus_paths:
if not os.path.exists(path):
continue
for fname in os.listdir(path):
if fname.endswith('.txt'):
full_path = os.path.join(path, fname)
if os.path.exists(full_path):
all_files.append(full_path)
self.file_paths = all_files
print(f" 📂 火2重新扫描: 发现 {len(self.file_paths)} 个文件")
def reload_files(self):
"""重新扫描文件夹，更新文件列表"""
old_count = len(self.file_paths)
self._scan_files()
new_count = len(self.file_paths)
if new_count != old_count:
print(f" 🔄 火2文件列表已更新: {old_count} → {new_count}")
def _read_sentences_from_file(self, file_path: str) -> List[str]:
"""从单个文件读取句子，若文件不存在则返回空"""
if not os.path.exists(file_path):
return []
try:
with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
text = f.read()
except:
return []
# 按标点分割
sentences = re.split(r'[。！？\n]+', text)
result = []
for s in sentences:
s = s.strip()
if len(s) > 10 and re.search(r'[\u4e00-\u9fff]', s):
if not re.match(r'^第\d+轮作品', s):
result.append(s)
return result
def _clean_invalid_files(self):
"""从列表中移除已不存在的文件"""
valid = [p for p in self.file_paths if os.path.exists(p)]
if len(valid) != len(self.file_paths):
removed = len(self.file_paths) - len(valid)
self.file_paths = valid
print(f" 🧹 火2清理: 移除了 {removed} 个不存在文件")
def get_sentences(self, dao_novelty: float) -> List[str]:
"""随机选文件 → 读内容 → 提取句子，若文件不存在则从列表中移除"""
# 先清理失效文件
self._clean_invalid_files()
if not self.file_paths:
return []
# 目标句子数：10-30
target_count = 10 + int(dao_novelty * 20)
# 随机选 5-15 个文件
file_count = min(len(self.file_paths), max(5, target_count // 3))
selected_files = random.sample(self.file_paths, file_count)
all_sentences = []
for fpath in selected_files:
# 再次确认文件存在
if not os.path.exists(fpath):
if fpath in self.file_paths:
self.file_paths.remove(fpath)
continue
sentences = self._read_sentences_from_file(fpath)
if sentences:
all_sentences.extend(sentences)
else:
# 文件存在但读不到句子，可能是空文件或格式问题，移除它以避免重复尝试
if fpath in self.file_paths:
self.file_paths.remove(fpath)
print(f" 🗑️ 火2移除空文件: {os.path.basename(fpath)}")
if len(all_sentences) >= target_count:
break
# 如果还不够，继续补
remaining = [f for f in self.file_paths if f not in selected_files]
while len(all_sentences) < target_count and remaining:
fpath = random.choice(remaining)
remaining.remove(fpath)
if not os.path.exists(fpath):
if fpath in self.file_paths:
self.file_paths.remove(fpath)
continue
sentences = self._read_sentences_from_file(fpath)
if sentences:
all_sentences.extend(sentences)
else:
if fpath in self.file_paths:
self.file_paths.remove(fpath)
print(f" 🗑️ 火2移除空文件: {os.path.basename(fpath)}")
# 如果还是不够，用已有句子重复填充
if len(all_sentences) < target_count:
if all_sentences:
all_sentences = all_sentences * (target_count // len(all_sentences) + 1)
all_sentences = all_sentences[:target_count]
else:
return []
# 打乱并截取
random.shuffle(all_sentences)
return all_sentences[:target_count]
def get_state(self) -> dict:
return {"file_count": len(self.file_paths), "last_scan_round": self.last_scan_round}
def restore_state(self, state: dict):
self.last_scan_round = state.get("last_scan_round", 0)
# ==================== 木3：组合句子成文章 ====================
class Mu3:
def generate(self, sentences: List[str], dao_novelty: float, sheng_ratio: float) -> str:
if not sentences:
return ""
temp = 0.6 + sheng_ratio * 0.4
if len(sentences) > 30:
sentences = random.sample(sentences, 30)
target_length = 2000 + int(dao_novelty * 1000)
prompt = f"""请将以下句子组合成一篇连贯的中文文章。
【要求】
- 文章长度：{target_length}字左右
- 可以调整句子顺序，加入连接词和过渡句
- 保持原句的核心意思
- 文章要有起承转合，段落分明
- 不要加说明文字，只输出文章正文
【素材句子】：
{chr(10).join([f'{i+1}. {s}' for i, s in enumerate(sentences)])}
只输出文章正文："""
result = call_deepseek(prompt, max_tokens=4096, temperature=temp)
if result and len(result) > 100:
if len(result) > 3000:
result = result[:3000]
return result
if sentences:
return "".join(random.sample(sentences, min(5, len(sentences)))) + "（组合失败）"
return ""
def get_state(self) -> dict:
return {}
def restore_state(self, state: dict):
pass
# ==================== 水1：润色白化文 ====================
class Shui1:
def __init__(self):
self.dao = None
def set_dao(self, dao):
self.dao = dao
def polish(self, text: str, dao_novelty: float, bian_ratio: float) -> str:
if not text or len(text) < 50:
return text
temp = 0.5 + bian_ratio * 0.3
prompt = f"""请将以下文章润色成流畅的白话文：
【要求】
- 保持原意不变
- 语句通顺自然
- 段落清晰
- 长度控制在3000字以内
- 不要加说明文字
原文：
{text}
润色后的文章："""
result = call_deepseek(prompt, max_tokens=4096, temperature=temp)
if result and len(result) > 100:
if len(result) > 3000:
result = result[:3000]
return result
return text
def get_state(self) -> dict:
return {}
def restore_state(self, state: dict):
pass
# ==================== 金4：固化章节 ====================
class Jin4:
def __init__(self, max_size=100):
self.masterpieces = []
self.max_size = max_size
def solidify(self, candidates: List[str], dao_novelty: float, round_num: int) -> Tuple[List[str], List[float]]:
if not candidates:
return [], []
prompt = f"为以下每个作品评分（0-1分），每行一个分数：\n" + "\n".join([c[:200] for c in candidates])
result = call_deepseek(prompt, max_tokens=100, temperature=0.3)
scores = []
if result:
for line in result.strip().split('\n'):
try:
score = float(re.search(r'(\d+\.?\d*)', line).group(1))
scores.append(min(1.0, max(0.0, score)))
except:
scores.append(0.5)
while len(scores) < len(candidates):
scores.append(0.5)
if not scores:
return [], []
max_score = max(scores)
good_works, good_scores = [], []
for work, score in zip(candidates, scores):
if score == max_score:
good_works.append(work)
good_scores.append(score)
polished = self.expand_to_chapter(work, dao_novelty)
if polished:
self.masterpieces.append(polished)
else:
self.masterpieces.append(work)
if len(self.masterpieces) > self.max_size:
self.masterpieces = self.masterpieces[-self.max_size:]
return good_works, good_scores
def expand_to_chapter(self, work: str, dao_novelty: float) -> str:
if not work or len(work) < 10:
return work
target_length = min(3000, max(2000, len(work) * 3))
prompt = f"""请将以下内容扩写成一篇完整的小说章节：
【要求】
- 保持原意不变
- 情节完整，有起承转合
- 字数控制在{target_length}字左右
- 不要加说明文字，只输出章节正文
原文：
{work}
扩写后的章节："""
result = call_deepseek(prompt, max_tokens=4096, temperature=0.5 + dao_novelty * 0.3)
if result and len(result) > 200:
if len(result) > 3000:
result = result[:3000]
return result.strip()
return work
def get_state(self) -> dict:
return {"masterpieces": self.masterpieces[-100:]}
def restore_state(self, state: dict):
self.masterpieces = state.get("masterpieces", [])
# ==================== 老师 ====================
class Teacher:
def __init__(self, teacher_id: int, student_name: str):
self.id = teacher_id
self.student_name = student_name
self.history = []
def evaluate(self, work: str, dao_novelty: float) -> Tuple[float, str]:
work_slice = work[:300] if len(work) > 300 else work
prompt = f"你是老师{self.id}，评判{self.student_name}。给出分数（0-1分）和评语。格式：分数|评语\n作业：{work_slice}"
result = call_deepseek(prompt, max_tokens=150, temperature=0.4)
score = 0.5
comment = ""
if result and '|' in result:
parts = result.split('|')
try:
score = float(parts[0].strip())
comment = parts[1].strip()[:40]
except:
pass
else:
score = min(1.0, len(work) / 50) * 0.5 + (len(set(work)) / max(1, len(work))) * 0.5
score = score * (0.8 + dao_novelty * 0.3)
score = min(1.0, max(0.0, score))
self.history.append((time.time(), work[:30], score))
if len(self.history) > 100:
self.history = self.history[-100:]
return score, comment
def get_state(self) -> dict:
return {"history": self.history[-50:]}
def restore_state(self, state: dict):
self.history = state.get("history", [])
# ==================== 洛书中心 ====================
class LuoShuCenter:
def __init__(self, dao: DaoEngine, checkpoint_dir: str = "checkpoints_放大器"):
self.dao = dao
self.hetu_center = HeTuCenter()
self.rhythm = RhythmController()
self.checkpoint_dir = checkpoint_dir
os.makedirs(checkpoint_dir, exist_ok=True)
print("\n📚 加载语料（从 masterpieces 取句子）...")
self.fire2 = Fire2(["masterpieces"])
self.mu3 = Mu3()
self.shui1 = Shui1()
self.shui1.set_dao(dao)
self.jin4 = Jin4()
self.teacher6 = Teacher(6, "水1")
self.teacher7 = Teacher(7, "火2")
self.teacher8 = Teacher(8, "木3")
self.teacher9 = Teacher(9, "金4")
self.round = 0
self.log_entries = []
self._load_checkpoint()
def _get_checkpoint_path(self) -> str:
return os.path.join(self.checkpoint_dir, "full_checkpoint.pkl")
def _get_tmp_path(self) -> str:
return self._get_checkpoint_path() + ".tmp"
def _get_backup_path(self, round_num: int) -> str:
return os.path.join(RECOVERY_DIR, f"checkpoint_{round_num}.pkl")
def _restore_pi_pointer(self, checkpoint: dict) -> bool:
pointer_sources = []
dao_state = checkpoint.get("dao_state", {})
if "pointer" in dao_state:
pointer_sources.append(("检查点", dao_state["pointer"]))
if "last_valid_pointer" in checkpoint:
pointer_sources.append(("last_valid_pointer", checkpoint["last_valid_pointer"]))
main_path = self._get_checkpoint_path()
if os.path.exists(main_path):
try:
with open(main_path, 'rb') as f:
main_cp = pickle.load(f)
main_dao = main_cp.get("dao_state", {})
if "pointer" in main_dao:
pointer_sources.append(("主检查点", main_dao["pointer"]))
except:
pass
seen = set()
unique_sources = []
for name, ptr in pointer_sources:
if ptr not in seen:
seen.add(ptr)
unique_sources.append((name, ptr))
for name, ptr in unique_sources:
try:
print(f" 🔄 尝试从 {name} 恢复π指针: {ptr}")
self.dao.pointer = ptr
self.dao.digits = []
self.dao._load_next_chunk()
test_digit = self.dao.get_digit()
self.dao.pointer -= 1
print(f" ✅ π指针恢复成功（来源: {name}）")
return True
except Exception as e:
print(f" ⚠️ 从 {name} 恢复失败: {e}")
continue
print(f" ⚠️ 所有π指针来源均失败，重置为0")
self.dao.pointer = 0
self.dao.digits = []
self.dao._load_next_chunk()
return True
def save_checkpoint(self):
checkpoint = {
"round": self.round,
"dao_state": self.dao.get_state(),
"last_valid_pointer": self.dao.get_pointer(),
"rhythm_state": self.rhythm.get_state(),
"hetu_state": self.hetu_center.get_save_state(),
"fire2_state": self.fire2.get_state(),
"jin4_state": self.jin4.get_state(),
"teacher6_state": self.teacher6.get_state(),
"teacher7_state": self.teacher7.get_state(),
"teacher8_state": self.teacher8.get_state(),
"teacher9_state": self.teacher9.get_state(),
"log_entries": self.log_entries[-100:],
"timestamp": datetime.now().isoformat()
}
tmp_path = self._get_tmp_path()
with open(tmp_path, 'wb') as f:
pickle.dump(checkpoint, f)
main_path = self._get_checkpoint_path()
os.replace(tmp_path, main_path)
if self.round % 100000 == 0 and self.round > 0:
backup_path = self._get_backup_path(self.round)
try:
shutil.copy2(main_path, backup_path)
print(f" 💾 备份检查点已保存: {backup_path}")
except Exception as e:
print(f" ⚠️ 备份保存失败: {e}")
def _load_checkpoint(self):
paths_to_try = [self._get_checkpoint_path(), self._get_tmp_path()]
backup_files = []
if os.path.exists(RECOVERY_DIR):
for f in os.listdir(RECOVERY_DIR):
if f.startswith("checkpoint_") and f.endswith(".pkl"):
try:
round_num = int(f.split("_")[1].split(".")[0])
backup_files.append((round_num, os.path.join(RECOVERY_DIR, f)))
except:
pass
if backup_files:
backup_files.sort(key=lambda x: x[0], reverse=True)
paths_to_try.append(backup_files[0][1])
for path in paths_to_try:
if not os.path.exists(path):
continue
try:
with open(path, 'rb') as f:
checkpoint = pickle.load(f)
print(f" 📂 加载检查点文件成功，正在恢复状态...")
self.round = checkpoint.get("round", 0)
self._restore_pi_pointer(checkpoint)
self.rhythm.restore_state(checkpoint.get("rhythm_state", {}))
self.hetu_center.restore_state(checkpoint.get("hetu_state", {}))
self.fire2.restore_state(checkpoint.get("fire2_state", {}))
self.jin4.restore_state(checkpoint.get("jin4_state", {}))
self.teacher6.restore_state(checkpoint.get("teacher6_state", {}))
self.teacher7.restore_state(checkpoint.get("teacher7_state", {}))
self.teacher8.restore_state(checkpoint.get("teacher8_state", {}))
self.teacher9.restore_state(checkpoint.get("teacher9_state", {}))
self.log_entries = checkpoint.get("log_entries", [])
print(f" 📂 加载检查点成功，从第 {self.round} 轮继续 (来源: {path})")
print(f" 🔄 π指针: {self.dao.pointer}")
return
except Exception as e:
print(f" ⚠️ 加载 {path} 失败: {e}")
continue
print(" 📂 未找到有效检查点，从头开始")
def run_cycle(self):
self.round += 1
dao_novelty = self.dao.get_novelty(6)
self.rhythm.update()
sheng_ratio = self.rhythm.get_sheng_ratio()
bian_ratio = self.rhythm.get_bian_ratio()
# ===== 每1000轮重新扫描 masterpieces 文件夹 =====
if self.round % 1000 == 0:
print(f" 🔄 第{self.round}轮：重新扫描 masterpieces 文件夹...")
self.fire2.reload_files()
print(f"\n{'─'*70}")
print(f"第 {self.round} 轮 | 道新奇度: {dao_novelty:.4f} | 生节:{sheng_ratio:.2f} | 变节:{bian_ratio:.2f}")
# 火2：取句子
sentences = self.fire2.get_sentences(dao_novelty)
print(f" 🔥 火2: 取 {len(sentences)} 个句子")
if sentences:
score7, comment7 = self.teacher7.evaluate(" ".join(sentences[:3]), dao_novelty)
self.hetu_center.update_sheng(1, score7)
self.hetu_center.update_cheng(7, score7)
print(f" 🔥 火2(生1): {len(sentences)}句 | 师7(成7):{score7:.2f} | {comment7}")
else:
score7 = 0.0
print(f" 🔥 火2(生1): 无句子")
# 木3：组合文章
if sentences:
article = self.mu3.generate(sentences, dao_novelty, sheng_ratio)
score8, comment8 = self.teacher8.evaluate(article, dao_novelty)
self.hetu_center.update_sheng(2, score8)
self.hetu_center.update_cheng(8, score8)
print(f" 🌳 木3(生2): 生成 {len(article)} 字")
print(f" 师8(成8):{score8:.2f} | {comment8}")
else:
article = ""
score8 = 0.0
print(f" 🌳 木3(生2): 无文章")
# 水1：润色
if article:
polished = self.shui1.polish(article, dao_novelty, bian_ratio)
if polished:
score6, comment6 = self.teacher6.evaluate(polished, dao_novelty)
self.hetu_center.update_sheng(3, score6)
self.hetu_center.update_cheng(6, score6)
print(f" 💧 水1(生3): 润色 {len(polished)} 字")
print(f" 师6(成6):{score6:.2f} | {comment6}")
else:
score6 = 0.5
print(f" 💧 水1(生3): 无输出")
polished = article
else:
score6 = 0.0
polished = ""
print(f" 💧 水1(生3): 无输入")
# 金4：固化
if polished:
candidates = [polished]
good_works, good_scores = self.jin4.solidify(candidates, dao_novelty, self.round)
if good_works:
best_work = good_works[0]
best_score = good_scores[0]
if self.jin4.masterpieces:
saved_work = self.jin4.masterpieces[-1]
else:
saved_work = best_work
score9, comment9 = self.teacher9.evaluate(saved_work, dao_novelty)
self.hetu_center.update_sheng(4, score9)
self.hetu_center.update_cheng(9, score9)
print(f" 💎 金4(生4): 固化章节 | 师9(成9):{score9:.2f} | {comment9}")
print(f" 章节字数: {len(saved_work)}")
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
with open(f"output_放大器/round_{self.round}_{timestamp}.txt", 'w', encoding='utf-8') as f:
f.write(saved_work)
else:
print(f" 💎 金4(生4): 未固化新作品")
score9 = 0.0
else:
print(f" 💎 金4(生4): 无输入")
score9 = 0.0
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}]")
self.log_entries.append({
"round": self.round, "dao_novelty": dao_novelty,
"sheng_ratio": sheng_ratio, "bian_ratio": bian_ratio,
"sheng": full_state['sheng'], "cheng": full_state['cheng'],
"sentence_count": len(sentences),
"word_count": len(polished) if polished else 0
})
if self.round % 10000 == 0:
self.save_checkpoint()
self.save_log()
def save_log(self):
with open(f"logs_放大器/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 📝 日志已保存，当前轮数: {self.round}")
def run_forever(self):
print("\n" + "="*70)
print("🌀 头脑风暴 · 河图洛书放大器 V1.0 - 道在π中")
print(" 火2: 从 masterpieces 取句子（10-30句），每1000轮重新扫描文件夹")
print(" 木3: 组合成文章（≤3000字）")
print(" 水1: 润色白化文（≤3000字）")
print(" 金4: 固化章节（≤3000字）")
print(" 老师: 调用API评分，自己进化")
print(" 每1万轮保存检查点")
print(" 不加任何人为设定。道驱动一切，万物自己演化")
print("="*70)
print("\n🚀 启动！按 Ctrl+C 停止\n")
try:
while True:
self.run_cycle()
except KeyboardInterrupt:
print(f"\n\n⏸️ 停止。运行了 {self.round} 轮")
print(f" 道消耗: {self.dao.get_pointer()} 位π")
print(f" 金池作品: {len(self.jin4.masterpieces)}")
self.save_checkpoint()
self.save_log()
print("\n 状态已保存，下次运行继续")
print(" 它不完美，但它是道的镜像。")
def main():
print("\n" + "="*70)
print("🌀 头脑风暴 · 河图洛书放大器 V1.0")
print(" 从 masterpieces 取句子，组合成章节")
print(" 每1000轮重新扫描文件夹，自动加入新文件并清理失效文件")
print(" 道驱动一切，不加人为设定")
print("="*70 + "\n")
dao = DaoEngine()
luoshu = LuoShuCenter(dao)
luoshu.run_forever()
if __name__ == "__main__":
main()
|
鄂公网安备42060002000282号 )
