再进化一下，4个学生与否个老师都有自已的单独记忆，总的运行状况也有自已的记忆，总之，在关闭后重启无须从头开始。修改一下。
```python
# hetu_luoshu_v7_mirror.py
# 河图洛书 V7.0 - 镜像核心版（完整API增强版）
# 支持递归遍历 learning_material 文件夹
# 道引擎使用 gmpy2.const_pi()，极速启动
# 生与变不同频：生慢（5轮周期），变快（1轮周期）
# 每轮输出完整师生结果，全程透明
# 支持完整状态持久化：4个学生、4个老师、河图中央、洛书中心、道指针全部保存
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
# ==================== DeepSeek API 配置 ====================
DEEPSEEK_API_KEY = "sk-81fdd74045564e0d8863eea589beade6"
DEEPSEEK_API_URL = "https://api.deepseek.com/v1/chat/completions"
# 创建必要目录
for d in ["cache", "learning_material", "masterpieces", "logs", "checkpoints"]:
os.makedirs(d, exist_ok=True)
def call_deepseek(prompt: str, max_tokens: int = 200, temperature: float = 0.7) -> str:
"""调用DeepSeek API，带缓存"""
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
}
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
else:
print(f" ⚠️ API失败: {response.status_code}")
return ""
except Exception as e:
print(f" ⚠️ API异常: {e}")
return ""
# ==================== 道：π引擎（可持久化） ====================
class DaoEngine:
def __init__(self, chunk_size=10000):
self.chunk_size = chunk_size
self.digits = []
self.pointer = 0
self._load_next_chunk()
print(f" ☯ 道引擎就绪，已加载{len(self.digits)}位π数字")
def _load_next_chunk(self):
"""用gmpy2计算π，极快"""
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_pointer(self) -> int:
return self.pointer
def get_state(self) -> dict:
"""保存道引擎状态"""
return {
"pointer": self.pointer,
"digits_len": len(self.digits)
}
def restore_state(self, state: dict):
"""恢复道引擎状态（需要重新加载π）"""
self.pointer = state.get("pointer", 0)
# 重新加载到指定位置
self.digits = []
self._load_next_chunk()
# 确保指针位置正确
while self.pointer >= len(self.digits):
self._load_next_chunk()
print(f" 道引擎恢复到指针: {self.pointer}")
# ==================== 节奏控制器（可持久化） ====================
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_state(self):
return self.global_state
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(root_dir: str) -> List[str]:
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
# ==================== 火2：河图之“生”（可持久化） ====================
class Fire2:
def __init__(self, corpus_paths: List[str]):
self.high_freq = []
self.low_freq = []
self._load_corpus(corpus_paths)
print(f" 🔥 火2完成。高频词:{len(self.high_freq)} 低频词:{len(self.low_freq)}")
def _load_corpus(self, paths):
counter = Counter()
all_files = []
for path in paths:
if os.path.isfile(path) and path.endswith('.txt'):
all_files.append(path)
elif os.path.isdir(path):
all_files.extend(get_all_txt_files(path))
if not all_files:
return
for i, file_path in enumerate(all_files):
if i % 500 == 0 and i > 0:
print(f" 进度: {i}/{len(all_files)}")
try:
with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
text = f.read()
for j in range(len(text)):
for l in range(1, 5):
word = text[j:j+l]
if re.match(r'[\u4e00-\u9fff]{1,4}$', word):
counter[word] += 1
except:
pass
for word, freq in counter.items():
if freq > 10:
self.high_freq.append(word)
elif freq >= 1:
self.low_freq.append(word)
def get_morphemes(self, dao_novelty: float, total: int = 20) -> List[str]:
candidates = (self.high_freq + self.low_freq)[:100]
if not candidates:
return ["道", "德", "仁", "义", "礼", "智", "信", "天", "地", "人"]
novelty_ratio = dao_novelty * 0.3
high_count = int(total * (0.6 - novelty_ratio))
result = []
if self.high_freq:
result.extend(random.sample(self.high_freq, min(high_count, len(self.high_freq))))
if len(result) < total:
prompt = f"从以下词语中选出{total - len(result)}个最有文化内涵的词，只输出词语用逗号分隔：\n{', '.join(candidates[:50])}"
api_result = call_deepseek(prompt, max_tokens=200, temperature=0.5)
if api_result:
new_words = [w.strip() for w in api_result.split(',') if w.strip()][:total - len(result)]
result.extend(new_words)
while len(result) < total:
length = random.choice([2, 3, 4])
base = random.choice(result)[:2] if result else "道"
result.append((base * (length // len(base) + 1))[:length])
random.shuffle(result)
return result[:total]
def get_state(self) -> dict:
return {"high_freq": self.high_freq[:1000], "low_freq": self.low_freq[:1000]}
def restore_state(self, state: dict):
self.high_freq = state.get("high_freq", [])
self.low_freq = state.get("low_freq", [])
# ==================== 木3 ====================
class Mu3:
def generate(self, morphemes: List[str], dao_novelty: float, sheng_ratio: float, sheng_length: int) -> str:
temp = 0.6 + sheng_ratio * 0.4
prompt = f"用以下词语造一个{sheng_length}字左右、有哲理的中文句子：{', '.join(morphemes[:12])}\n只输出句子："
result = call_deepseek(prompt, max_tokens=sheng_length + 20, temperature=temp)
if result and len(result) > 5:
if len(result) > sheng_length:
result = result[:sheng_length]
return result.strip()
return "。".join(random.sample(morphemes, min(3, len(morphemes)))) + "。"
def get_state(self) -> dict:
return {}
def restore_state(self, state: dict):
pass
# ==================== 水1 ====================
class Shui1:
def mutate(self, sentence: str, dao_novelty: float, bian_ratio: float, bian_length: int) -> List[str]:
styles = ["古风", "白话", "哲理", "诗意", "简练"]
style = styles[int(dao_novelty * len(styles)) % len(styles)]
if bian_ratio < 0.33:
intensity = "轻度"
elif bian_ratio < 0.66:
intensity = "中度"
else:
intensity = "强度"
temp = 0.6 + bian_ratio * 0.5
prompt = f"将「{sentence}」改写成{style}风格，{intensity}变体，生成3个不同的变体，每个变体{bian_length}字左右，每行一个："
result = call_deepseek(prompt, max_tokens=bian_length * 3 + 50, temperature=temp)
if result:
variants = [v.strip() for v in result.strip().split('\n') if v.strip()]
trimmed = []
for v in variants:
if len(v) > bian_length:
v = v[:bian_length]
trimmed.append(v)
return trimmed[:3]
return [sentence[:bian_length], sentence[:bian_length//2] + "……", "诚然，" + sentence[:bian_length-2]]
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) -> Tuple[List[str], List[float]]:
if not candidates:
return [], []
prompt = f"为以下每个句子评分（0-1分），评估其文学价值、哲理深度和通顺度。每行一个分数：\n" + "\n".join(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)
good_works, good_scores = [], []
for work, score in zip(candidates, scores):
if score > 0.7:
good_works.append(work)
good_scores.append(score)
self.masterpieces.append(work)
if len(self.masterpieces) > self.max_size:
self.masterpieces = self.masterpieces[-self.max_size:]
return good_works, good_scores
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.threshold = 0.6
self.history = []
def evaluate(self, work: str, dao_novelty: float) -> Tuple[bool, float, str]:
prompt = f"你是老师{self.id}，负责评判{self.student_name}。给出分数（0-1分）和评语。格式：分数|评语\n作业：{work[:150]}"
result = call_deepseek(prompt, max_tokens=100, 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) / 30) * 0.5 + (len(set(work)) / max(1, len(work))) * 0.5
score = score * (0.8 + dao_novelty * 0.3)
passed = score >= self.threshold
self.history.append((time.time(), work[:30], score))
if len(self.history) > 100:
self.history = self.history[-100:]
return passed, min(1.0, max(0.0, score)), comment
def get_state(self) -> dict:
return {"threshold": self.threshold, "history": self.history[-50:]}
def restore_state(self, state: dict):
self.threshold = state.get("threshold", 0.6)
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📚 加载语料...")
self.fire2 = Fire2(["learning_material"])
self.mu3 = Mu3()
self.shui1 = Shui1()
self.jin4 = Jin4()
self.teacher7 = Teacher(7, "火2-语素提取")
self.teacher8 = Teacher(8, "木3-句子生成")
self.teacher6 = Teacher(6, "水1-变体生成")
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 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(),
"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())
print(f" 💾 检查点已保存 (轮数: {self.round})")
def _load_checkpoint(self):
"""加载检查点"""
checkpoint_path = self._get_checkpoint_path()
if not os.path.exists(checkpoint_path):
print(" 📂 未找到检查点，从头开始")
return
try:
with open(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", {}))
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} 轮继续")
except Exception as e:
print(f" ⚠️ 加载检查点失败: {e}，从头开始")
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()
sheng_length = self.rhythm.get_sheng_length(15, 40)
bian_length = self.rhythm.get_bian_length(8, 25)
print(f"\n{'─'*70}")
print(f"第 {self.round} 轮 | 道新奇度: {dao_novelty:.4f} | 生节:{sheng_ratio:.2f}/{sheng_length} | 变节:{bian_ratio:.2f}/{bian_length}")
# 火2
morphemes = self.fire2.get_morphemes(dao_novelty, total=15)
passed7, score7, comment7 = self.teacher7.evaluate(" ".join(morphemes[:5]), dao_novelty)
self.hetu_center.update_sheng(1, score7)
self.hetu_center.update_cheng(7, score7)
print(f" 🔥 火2(生1): {len(morphemes)}语素 | 师7(成7):{score7:.2f} | {comment7}")
# 木3
sentence = self.mu3.generate(morphemes[:10], dao_novelty, sheng_ratio, sheng_length)
passed8, score8, comment8 = self.teacher8.evaluate(sentence, dao_novelty)
self.hetu_center.update_sheng(2, score8)
self.hetu_center.update_cheng(8, score8)
print(f" 🌳 木3(生2): {sentence[:70]}...")
print(f" 师8(成8):{score8:.2f} | {comment8}")
# 水1
variants = self.shui1.mutate(sentence, dao_novelty, bian_ratio, bian_length)
if variants:
best_variant = variants[0]
passed6, score6, comment6 = self.teacher6.evaluate(best_variant, dao_novelty)
self.hetu_center.update_sheng(3, score6)
self.hetu_center.update_cheng(6, score6)
print(f" 💧 水1(生3): {len(variants)}个变体")
for i, v in enumerate(variants[:2]):
print(f" 变体{i+1}: {v[:60]}...")
print(f" 师6(成6):{score6:.2f} | {comment6}")
else:
best_variant = sentence
score6 = 0.5
print(f" 💧 水1(生3): 无变体")
# 金4
candidates = [sentence] + (variants if variants else [])
good_works, good_scores = self.jin4.solidify(candidates, dao_novelty)
score9 = 0
if good_works:
best_work = good_works[0]
best_score = good_scores[0]
passed9, score9, comment9 = self.teacher9.evaluate(best_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" 作品: {best_work[:80]}...")
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
with open(f"masterpieces/round_{self.round}_{timestamp}.txt", 'w', encoding='utf-8') as f:
f.write(f"第{self.round}轮作品\n道新奇度:{dao_novelty:.4f}\n\n{best_work}")
else:
print(f" 💎 金4(生4): 未固化新作品")
# 汇总
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']
})
# 每10轮保存检查点
if self.round % 10 == 0:
self.save_checkpoint()
if self.round % 100 == 0:
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("☯ 河图洛书镜像智能体 V7.0")
print(" 道在π中 | 河图中央汇聚生1-4和成6-9 | 洛书调度")
print(" 生1=火2 | 生2=木3 | 生3=水1 | 生4=金4")
print(" 成6=师6 | 成7=师7 | 成8=师8 | 成9=师9")
print(" 生与变不同频：生慢（5轮周期），变快（1轮周期）")
print(" 每轮输出完整师生结果，全程透明")
print(" 每10轮自动保存检查点，关闭后重启可继续")
print("="*70)
print("\n🚀 启动！按 Ctrl+C 停止\n")
try:
while True:
self.run_cycle()
time.sleep(0.3)
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 状态已保存，下次运行将从第", self.round, "轮继续")
print(" 它不完美，但它是道的镜像。")
# ==================== 主程序 ====================
def main():
print("\n" + "="*70)
print("🐉 河图洛书 V7.0 - 镜像核心版")
print(" 设计哲学：我们不做合道的智能体，我们做道的镜像")
print(" 生与变不同频：阴阳不同步，万物有节律")
print(" 完整状态持久化：4生4师+河图中央+道指针全部保存")
print("="*70 + "\n")
dao = DaoEngine()
luoshu = LuoShuCenter(dao)
luoshu.run_forever()
if __name__ == "__main__":
main()
```
## 持久化说明
| 组件 | 保存内容 | 恢复方式 |
| :--- | :--- | :--- |
| **道引擎(π)** | 指针位置 | 重新计算π到指针位置 |
| **节奏控制器** | 生相位、变相位 | 恢复相位值 |
| **河图中央** | 生1-4值、成6-9值 | 恢复全部状态 |
| **火2** | 高频词表、低频词表(前1000) | 恢复词表 |
| **木3** | 无状态 | - |
| **水1** | 无状态 | - |
| **金4** | 作品库(前100) | 恢复作品库 |
| **老师6/7/8/9** | 阈值、历史记录 | 恢复全部 |
| **洛书中心** | 轮数、日志 | 恢复进度 |
## 运行效果
- **首次运行**：从头开始训练
- **Ctrl+C停止**：自动保存检查点
- **再次运行**：自动加载检查点，从上次轮数继续
快乐的老头，可以运行了。 |
鄂公网安备42060002000282号 )