"""
河图洛书 V4.0 - 道之篇章（永续进化版）
永久运行，直到手动停止
"""
import os
import time
import random
import math
from typing import Tuple, Dict, Any, Optional, List
from collections import deque
from dataclasses import dataclass
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torch.utils.data import DataLoader
import numpy as np
try:
import torchvision
import torchvision.transforms as transforms
HAS_TORCHVISION = True
except ImportError:
HAS_TORCHVISION = False
print("警告: torchvision未安装")
# ==================== 配置 ====================
@dataclass
class DaoConfig:
min_agents: int = 3
max_agents: int = 9
mutation_rate: float = 0.05
mutation_strength: float = 0.1
accuracy_threshold: float = 0.99
# ==================== 河图生数成数模块 ====================
class HeTuShengCheng(nn.Module):
"""河图生数与成数 - 先理解后记忆"""
def __init__(self, hidden_dim: int = 128):
super().__init__()
# 生数阶段：理解网络
self.sheng_network = nn.Sequential(
nn.Linear(64 * 7 * 7, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, hidden_dim // 2),
nn.ReLU(),
nn.Linear(hidden_dim // 2, 10)
)
# 成数阶段：记忆网络
self.cheng_network = nn.Sequential(
nn.Linear(64 * 7 * 7, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, 10)
)
self._comprehension = 0.0
self._cheng_active = False
self._sheng_epochs = 0
self.sheng_optimizer = None
self.cheng_optimizer = None
def forward(self, x: torch.Tensor, mode: str = "auto") -> torch.Tensor:
if mode == "sheng" or (mode == "auto" and not self._cheng_active):
return self.sheng_network(x)
else:
out = self.cheng_network(x)
return out * (0.5 + self._comprehension * 0.5)
def train_sheng(self, x: torch.Tensor, y: torch.Tensor) -> float:
if self._cheng_active:
return 0.0
if self.sheng_optimizer is None:
self.sheng_optimizer = optim.Adam(self.sheng_network.parameters(), lr=0.001)
self.sheng_optimizer.zero_grad()
logits = self.sheng_network(x)
loss = F.cross_entropy(logits, y)
loss.backward()
self.sheng_optimizer.step()
pred = logits.argmax(dim=1)
correct = (pred == y).float().mean()
self._comprehension = 0.95 * self._comprehension + 0.05 * correct.item()
return loss.item()
def train_cheng(self, x: torch.Tensor, y: torch.Tensor) -> float:
if not self._cheng_active:
return 0.0
if self.cheng_optimizer is None:
self.cheng_optimizer = optim.Adam(self.cheng_network.parameters(), lr=0.0005)
self.cheng_optimizer.zero_grad()
logits = self.cheng_network(x)
loss = F.cross_entropy(logits, y)
loss.backward()
self.cheng_optimizer.step()
return loss.item()
def update_epoch(self) -> bool:
self._sheng_epochs += 1
if not self._cheng_active and self._sheng_epochs >= 5 and self._comprehension >= 0.7:
self._cheng_active = True
return True
return False
@property
def cheng_active(self) -> bool:
return self._cheng_active
@property
def comprehension(self) -> float:
return self._comprehension
@property
def sheng_epochs(self) -> int:
return self._sheng_epochs
# ==================== 卷积特征提取器 ====================
class FeatureExtractor(nn.Module):
"""CNN特征提取器"""
def __init__(self):
super().__init__()
self.conv1 = nn.Conv2d(1, 32, 3, padding=1)
self.bn1 = nn.BatchNorm2d(32)
self.conv2 = nn.Conv2d(32, 64, 3, padding=1)
self.bn2 = nn.BatchNorm2d(64)
self.pool = nn.MaxPool2d(2)
self.dropout = nn.Dropout(0.25)
def forward(self, x: torch.Tensor) -> torch.Tensor:
x = self.pool(F.relu(self.bn1(self.conv1(x))))
x = self.pool(F.relu(self.bn2(self.conv2(x))))
x = x.view(x.size(0), -1)
return self.dropout(x)
# ==================== 河图洛书 V4.0 智能体 ====================
class HeTuLuoShuAgentV4(nn.Module):
"""河图洛书 V4.0 智能体"""
def __init__(self, agent_id: int, num_classes: int = 10, parent=None):
super().__init__()
self.agent_id = agent_id
self.num_classes = num_classes
self.generation = 0
self.parent_id = -1
self.total_epochs = 0
# 特征提取器
self.feature_extractor = FeatureExtractor()
# 河图生数成数
self.hetu = HeTuShengCheng(128)
if parent:
self.inherit_from(parent)
self._role = "pu"
self.entropy_history = deque(maxlen=200)
self.best_accuracy = 0.0
self.current_accuracy = 0.0
print(f"[Agent {agent_id}] 诞生 (Gen={self.generation})")
def inherit_from(self, parent):
"""继承父智能体"""
self.load_state_dict(parent.state_dict(), strict=False)
self.generation = parent.generation + 1
self.parent_id = parent.agent_id
self.mutate()
def mutate(self):
"""变异"""
with torch.no_grad():
for param in self.parameters():
if random.random() < 0.05:
noise = torch.randn_like(param) * 0.05
param.data = param.data + noise
def forward(self, x: torch.Tensor) -> Tuple[torch.Tensor, Dict]:
features = self.feature_extractor(x)
logits = self.hetu(features)
probs = F.softmax(logits, dim=-1)
entropy = -(probs * torch.log(probs + 1e-8)).sum(dim=-1).mean()
self.entropy_history.append(entropy.item())
return logits, {'entropy': entropy.item(), 'comprehension': self.hetu.comprehension}
def train_one_epoch(self, dataloader: DataLoader) -> Dict:
self.train()
total_loss = 0.0
correct = 0
total = 0
steps = 0
device = next(self.parameters()).device
for x, y in dataloader:
x = x.to(device)
y = y.to(device).long()
features = self.feature_extractor(x)
if not self.hetu.cheng_active:
loss = self.hetu.train_sheng(features, y)
else:
loss = self.hetu.train_cheng(features, y)
with torch.no_grad():
logits = self.hetu(features)
pred = logits.argmax(dim=1)
correct += (pred == y).sum().item()
total_loss += loss
total += y.size(0)
steps += 1
accuracy = correct / max(total, 1)
self.current_accuracy = accuracy
if accuracy > self.best_accuracy:
self.best_accuracy = accuracy
self.total_epochs += 1
entered_cheng = self.hetu.update_epoch()
if entered_cheng:
print(f"[Agent {self.agent_id}] ✨ 理解度 {self.hetu.comprehension:.3f} → 进入成数阶段")
return {
'loss': total_loss / max(steps, 1),
'accuracy': accuracy,
'best_accuracy': self.best_accuracy,
'comprehension': self.hetu.comprehension,
'phase': 'cheng' if self.hetu.cheng_active else 'sheng'
}
@torch.no_grad()
def evaluate(self, dataloader: DataLoader) -> Dict:
self.eval()
correct = 0
total = 0
device = next(self.parameters()).device
for x, y in dataloader:
x = x.to(device)
y = y.to(device).long()
features = self.feature_extractor(x)
logits = self.hetu(features)
pred = logits.argmax(dim=1)
correct += (pred == y).sum().item()
total += y.size(0)
return {'accuracy': correct / max(total, 1)}
def get_role(self) -> str:
return self._role
def set_role(self, role: str):
self._role = role
def should_retire(self, threshold: float = 0.99) -> tuple:
if self.total_epochs < 20:
return False, "训练不足"
if self.best_accuracy >= threshold:
return True, f"达到 {self.best_accuracy:.3f}"
return False, "继续"
def report(self) -> Dict:
return {
'id': self.agent_id,
'gen': self.generation,
'phase': '🌸' if self.hetu.cheng_active else '🌱',
'comp': round(self.hetu.comprehension, 3),
'acc': round(self.current_accuracy, 3),
'best': round(self.best_accuracy, 3),
'role': self._role
}
# ==================== 调度器 ====================
class DaoScheduler:
def __init__(self, config: DaoConfig):
self.config = config
self.agents: Dict[int, HeTuLuoShuAgentV4] = {}
self.current_evolution_id = None
self.total_born = 0
self.total_retired = 0
self.cycle = 0
print("[调度器] 初始化完成")
def init_agents(self, count: int):
for i in range(count):
agent = HeTuLuoShuAgentV4(i)
self.agents[i] = agent
self.total_born += 1
def elect_evolution(self):
if not self.agents:
return
def score(a):
return a.current_accuracy * 0.7 + a.hetu.comprehension * 0.3
best = max(self.agents.values(), key=score)
if self.current_evolution_id != best.agent_id:
if self.current_evolution_id:
old = self.agents.get(self.current_evolution_id)
if old:
old.set_role('pu')
self.current_evolution_id = best.agent_id
best.set_role('evolution')
def check_retire(self):
to_retire = []
for aid, agent in self.agents.items():
should, reason = agent.should_retire(self.config.accuracy_threshold)
if should:
to_retire.append((aid, reason))
for aid, reason in to_retire:
print(f"[调度器] Agent {aid} 退役: {reason}")
del self.agents[aid]
self.total_retired += 1
def get_status(self) -> Dict:
return {
'count': len(self.agents),
'evolution': self.current_evolution_id,
'born': self.total_born,
'retired': self.total_retired,
'cycle': self.cycle,
'agents': [a.report() for a in self.agents.values()]
}
# ==================== 主演化系统（永续运行） ====================
class HeTuLuoShuSystemV4:
def __init__(self, config: DaoConfig = None):
self.config = config or DaoConfig()
self.scheduler = DaoScheduler(self.config)
self.train_loader = None
self.test_loader = None
self.running = True
print("\n" + "=" * 70)
print("河图洛书 V4.0 - 道之篇章（永续进化版）")
print("=" * 70)
print(" 1. 河图生数成数分离 - 先理解后记忆")
print(" 2. 理解度实时更新 - 从🌱到🌸的成长")
print(" 3. 代际传承 - 知识积累")
print(" 4. 永续运行 - 直到手动停止")
print("=" * 70)
print("\n按 Ctrl+C 停止系统\n")
def prepare_data(self):
print("准备数据...")
transform = transforms.Compose([
transforms.ToTensor(),
transforms.Normalize((0.5,), (0.5,))
])
if HAS_TORCHVISION:
train = torchvision.datasets.MNIST(root='./data', train=True, download=True, transform=transform)
test = torchvision.datasets.MNIST(root='./data', train=False, download=True, transform=transform)
print("使用MNIST数据集")
else:
train = torch.utils.data.TensorDataset(torch.randn(2000, 1, 28, 28), torch.randint(0, 10, (2000,)))
test = torch.utils.data.TensorDataset(torch.randn(500, 1, 28, 28), torch.randint(0, 10, (500,)))
print("使用模拟数据")
self.train_loader = DataLoader(train, batch_size=64, shuffle=True)
self.test_loader = DataLoader(test, batch_size=64, shuffle=False)
def train_agent(self, agent: HeTuLuoShuAgentV4, epochs: int = 2):
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
agent.to(device)
for _ in range(epochs):
train_stats = agent.train_one_epoch(self.train_loader)
eval_stats = agent.evaluate(self.test_loader)
phase_mark = "🌱" if train_stats['phase'] == 'sheng' else "🌸"
phase_name = "生数" if train_stats['phase'] == 'sheng' else "成数"
print(f" {phase_mark} Agent {agent.agent_id} | "
f"{phase_name} | "
f"理解度={train_stats['comprehension']:.3f} | "
f"Train={train_stats['accuracy']:.3f} | "
f"Test={eval_stats['accuracy']:.3f}")
def print_status(self):
"""打印状态"""
status = self.scheduler.get_status()
print("\n" + "=" * 70)
print(f"📊 系统状态 (第{status['cycle']}轮)")
print(f" 活跃: {status['count']} | 总诞生: {status['born']} | 总退役: {status['retired']}")
print(f" 主进化态: {status['evolution']}")
print(f"\n 当前智能体:")
for a in status['agents']:
print(f" {a['phase']} Agent {a['id']}: Gen={a['gen']}, "
f"理解={a['comp']}, Acc={a['acc']}, Best={a['best']}, Role={a['role']}")
print("=" * 70)
def run(self):
"""永续运行，直到手动停止"""
self.prepare_data()
self.scheduler.init_agents(3)
print("\n🚀 系统启动，永续进化中...\n")
last_status = time.time()
try:
while self.running:
self.scheduler.cycle += 1
# 选举主进化态
self.scheduler.elect_evolution()
# 训练所有智能体
for agent in list(self.scheduler.agents.values()):
epochs = 3 if agent.get_role() == 'evolution' else 1
self.train_agent(agent, epochs)
# 检查退役和补充
self.scheduler.check_retire()
# 每30秒打印状态
if time.time() - last_status > 30:
self.print_status()
last_status = time.time()
time.sleep(0.1)
except KeyboardInterrupt:
print("\n\n用户手动停止")
self.print_status()
print("\n河图洛书 V4.0 已停止")
def main():
config = DaoConfig(min_agents=3, max_agents=9, accuracy_threshold=0.99)
system = HeTuLuoShuSystemV4(config)
system.run()
if __name__ == '__main__':
main()
|
鄂公网安备42060002000282号 )
