"""
河图洛书智能体 v1.3 (HeTu LuoShu Agent)
永续进化版：循环无端，自动续跑，永久记忆
核心特性：
- 无限训练循环，永不终止（可优雅退出并保存状态）
- 启动时自动检测最新检查点，从中断处继续
- 定期自动保存模型权重 + 记忆状态 + 优化器状态 + 当前epoch计数
- 支持手动中断（Ctrl+C）安全退出并保存
- 完全体现“道循环永无断绝，循环无端”的设计理念
"""
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torch.utils.data import DataLoader
from torchvision import datasets, transforms
import copy
import os
import signal
import sys
from datetime import datetime
# ================== 安全边界 ==================
class SafetyBoundary:
PARAM_BOUNDS = {
'sheng_strength': (0.05, 0.95),
'ke_strength': (0.01, 0.80),
'yang_weights': (0.05, 0.95),
'yin_weights': (0.05, 0.95),
'center_state': (-2.0, 2.0),
'coupling_coef': (0.10, 0.90),
}
COUPLING_SPARSITY_TARGET = 0.3
COUPLING_SPARSITY_WEIGHT = 0.01
STRUCTURE = {
'num_kun_units': 8,
'num_heads': 8,
'num_memory_slots': 8,
}
# ================== 1. 洛书核心 ==================
class LuoShuCore(nn.Module):
def __init__(self):
super().__init__()
self.state = nn.Parameter(torch.randn(9) * 0.1)
ke = torch.zeros(9, 9)
ke[8,7] = 1.0; ke[7,4] = 0.8; ke[4,5] = 0.9
ke[5,6] = 0.7; ke[6,8] = 0.6
self.register_buffer('ke_matrix', ke)
def forward(self, feedback):
influence = torch.matmul(feedback, self.ke_matrix)
new_state = self.state + 0.05 * torch.tanh(influence)
self.state.data = new_state
dispatch = torch.sigmoid(self.state)
return dispatch, self.state
# ================== 2. 阴循环（坤八卦） ==================
class YinXunHuan(nn.Module):
def __init__(self, feat_dim=32):
super().__init__()
self.cores = nn.ModuleList([LuoShuCore() for _ in range(8)])
self.sheng_strength = nn.Parameter(torch.ones(8) * 0.5)
self.ke_strength = nn.Parameter(torch.ones(8) * 0.3)
def apply_safety_guard(self):
with torch.no_grad():
self.sheng_strength.data = torch.clamp(self.sheng_strength.data, 0.05, 0.95)
self.ke_strength.data = torch.clamp(self.ke_strength.data, 0.01, 0.80)
def forward(self, feedbacks, global_dispatch=None):
local_dispatches = []
for i, core in enumerate(self.cores):
fb = feedbacks[i] if feedbacks[i] is not None else torch.zeros(9, device=core.state.device)
dispatch, _ = core(fb)
dispatch = dispatch * self.sheng_strength[i]
local_dispatches.append(dispatch)
if global_dispatch is not None:
for i in range(8):
ke_idx = (i + 2) % 8
ke_factor = 1 - self.ke_strength[i] * local_dispatches[ke_idx].mean()
local_dispatches[i] = local_dispatches[i] * ke_factor
local_dispatches[i] = local_dispatches[i] * (0.5 + global_dispatch.mean())
return local_dispatches
# ================== 3. 阳循环（乾总调度器） ==================
class YangXunHuan(nn.Module):
def __init__(self):
super().__init__()
self.core = LuoShuCore()
self.in_proj = nn.Linear(8, 9)
def forward(self, kun_losses):
feedback = self.in_proj(kun_losses.unsqueeze(0)).squeeze(0)
dispatch, _ = self.core(feedback)
return dispatch
# ================== 4. 外层循环（下界+上界） ==================
class WaiCengXunHuan(nn.Module):
def __init__(self, in_ch=1, out_ch=32, num_classes=10):
super().__init__()
self.group_ch = out_ch // 8
self.conv_sheng = nn.ModuleList([
nn.Conv2d(in_ch, self.group_ch, 3, padding=1) for _ in range(8)
])
self.conv_cheng = nn.ModuleList([
nn.Conv2d(in_ch, self.group_ch, 5, padding=2) for _ in range(8)
])
self.fusion = nn.Conv2d(out_ch * 2, out_ch, 1)
self.act = nn.ReLU()
self.gap = nn.AdaptiveAvgPool2d(1)
self.fc = nn.Linear(out_ch, num_classes)
def forward_features(self, x, local_dispatches):
sheng_parts = []
cheng_parts = []
for i in range(8):
scale = 0.5 + local_dispatches[i].mean()
sheng = self.act(self.conv_sheng[i](x)) * scale
cheng = self.act(self.conv_cheng[i](x)) * scale
sheng_parts.append(sheng)
cheng_parts.append(cheng)
sheng_out = torch.cat(sheng_parts, dim=1)
cheng_out = torch.cat(cheng_parts, dim=1)
combined = torch.cat([sheng_out, cheng_out], dim=1)
out = self.act(self.fusion(combined))
return out
def forward_classifier(self, x):
return self.fc(self.gap(x).flatten(1))
def forward(self, x, local_dispatches):
feat = self.forward_features(x, local_dispatches)
logits = self.forward_classifier(feat)
return logits, feat
# ================== 5. 中层循环（注意力） ==================
class ZhongCengXunHuan(nn.Module):
def __init__(self, dim=32, num_heads=8):
super().__init__()
self.num_heads = num_heads
self.head_dim = dim // num_heads
self.qkv = nn.Linear(dim, dim * 3)
self.proj = nn.Linear(dim, dim)
self.min_dist = nn.Parameter(torch.ones(num_heads) * 2)
def forward(self, x, local_dispatches):
B, C, H, W = x.shape
N = H * W
x_flat = x.flatten(2).permute(0, 2, 1)
qkv = self.qkv(x_flat).reshape(B, N, 3, self.num_heads, self.head_dim)
qkv = qkv.permute(2, 0, 3, 1, 4)
q, k, v = qkv[0], qkv[1], qkv[2]
attn_out = []
for h in range(self.num_heads):
min_d = int(self.min_dist[h].item())
with torch.no_grad():
coords_y, coords_x = torch.meshgrid(torch.arange(H), torch.arange(W), indexing='ij')
coords = torch.stack([coords_x.flatten(), coords_y.flatten()], dim=1).float().to(x.device)
dist = torch.cdist(coords, coords)
mask = (dist < min_d).bool()
attn = torch.matmul(q[:, h], k[:, h].transpose(-2, -1)) / (self.head_dim ** 0.5)
attn = attn.masked_fill(mask.unsqueeze(0), -float('inf'))
attn = F.softmax(attn, dim=-1)
out_h = torch.matmul(attn, v[:, h])
attn_out.append(out_h)
out = torch.stack(attn_out, dim=2).reshape(B, N, C)
out = self.proj(out)
for h in range(self.num_heads):
scale = 0.5 + local_dispatches[h % 8].mean()
out[:, :, h * self.head_dim h + 1) * self.head_dim] *= scale
out = out.permute(0, 2, 1).view(B, C, H, W)
return x + out
# ================== 6. 记忆库（持久化存储，带自动保存） ==================
class HTLSMemory(nn.Module):
def __init__(self, input_dim=32, hidden_dim=64, num_slots=8, persistent=True):
super().__init__()
self.num_slots = num_slots
self.hidden_dim = hidden_dim
self.persistent = persistent
self.rnns = nn.ModuleList([nn.GRUCell(input_dim, hidden_dim) for _ in range(num_slots)])
self.proj = nn.Linear(hidden_dim * num_slots, input_dim)
self.states = [None] * num_slots
def reset_states(self):
self.states = [None] * self.num_slots
def save_states(self, path):
states_to_save = []
for s in self.states:
if s is not None:
states_to_save.append(s.cpu())
else:
states_to_save.append(None)
torch.save(states_to_save, path)
def load_states(self, path, device='cpu'):
loaded = torch.load(path, map_location=device)
for i, s in enumerate(loaded):
if s is not None:
self.states[i] = s.to(device)
else:
self.states[i] = None
def forward(self, x, local_dispatches):
batch_size = x.size(0)
device = x.device
slot_outputs = []
for i in range(self.num_slots):
if (self.states[i] is None or
self.states[i].size(0) != batch_size or
(not self.persistent and i == 0)):
self.states[i] = torch.zeros(batch_size, self.hidden_dim, device=device)
scale = 0.5 + local_dispatches[i].mean()
x_mod = x * scale
new_state = self.rnns[i](x_mod, self.states[i])
if self.persistent:
self.states[i] = new_state.detach()
else:
self.states[i] = None
slot_outputs.append(new_state)
out = torch.cat(slot_outputs, dim=1)
out = self.proj(out)
return out
# ================== 7. 完整智能体（五环 + 穿中心转向 + 持久记忆） ==================
class HeTuLuoShuAgent(nn.Module):
def __init__(self, num_classes=10, memory_persistent=True):
super().__init__()
self.yang = YangXunHuan()
self.yin = YinXunHuan(feat_dim=32)
self.tu = nn.Parameter(torch.ones(8) * 0.5)
self.zhong = ZhongCengXunHuan(dim=32, num_heads=8)
self.wai = WaiCengXunHuan(in_ch=1, out_ch=32, num_classes=num_classes)
self.memory = HTLSMemory(input_dim=32, hidden_dim=64, num_slots=8, persistent=memory_persistent)
self.center_proj = nn.Linear(32, 32)
self.memory_persistent = memory_persistent
def apply_safety_guard(self):
with torch.no_grad():
self.tu.data = torch.clamp(self.tu.data, 0.10, 0.90)
self.yin.apply_safety_guard()
def reset_memory(self):
self.memory.reset_states()
def save_full_state(self, base_path, epoch=None, optimizer=None):
suffix = f"_epoch{epoch}" if epoch is not None else ""
torch.save(self.state_dict(), f"{base_path}_model{suffix}.pth")
self.memory.save_states(f"{base_path}_memory{suffix}.pt")
if optimizer is not None and epoch is not None:
opt_state = {'optimizer': optimizer.state_dict(), 'epoch': epoch}
torch.save(opt_state, f"{base_path}_optimizer{suffix}.pt")
if epoch is not None:
print(f"Saved full state (model + memory) at epoch {epoch}")
def load_full_state(self, base_path, epoch=None, device='cpu', load_optimizer=False):
suffix = f"_epoch{epoch}" if epoch is not None else ""
# 查找最新的检查点（如果epoch为None，尝试找最新的）
if epoch is None:
import glob
files = glob.glob(f"{base_path}_model_epoch*.pth")
if not files:
raise FileNotFoundError(f"No checkpoint found at {base_path}")
# 提取epoch数字并排序
epochs = [int(f.split('_epoch')[-1].split('.')[0]) for f in files]
epoch = max(epochs)
suffix = f"_epoch{epoch}"
print(f"Auto-resuming from latest epoch {epoch}")
state_dict = torch.load(f"{base_path}_model{suffix}.pth", map_location=device)
self.load_state_dict(state_dict)
self.memory.load_states(f"{base_path}_memory{suffix}.pt", device)
if load_optimizer:
opt_path = f"{base_path}_optimizer{suffix}.pt"
if os.path.exists(opt_path):
opt_state = torch.load(opt_path, map_location=device)
return opt_state['optimizer'], opt_state['epoch']
return None, epoch
def forward(self, x):
B = x.shape[0]
device = x.device
with torch.no_grad():
dummy_feedbacks = [None] * 8
dummy_local = self.yin(dummy_feedbacks, None)
dummy_x = self.wai.forward_features(x, dummy_local)
dummy_logits = self.wai.forward_classifier(dummy_x)
dummy_probs = F.softmax(dummy_logits, dim=-1)
entropy = -(dummy_probs * torch.log(dummy_probs + 1e-8)).sum(dim=-1)
yin_losses = entropy[:8] if B >= 8 else torch.cat([entropy, torch.zeros(8 - B, device=device)])
global_dispatch = self.yang(yin_losses)
feedbacks = [None] * 8
local_dispatches = self.yin(feedbacks, global_dispatch)
for i in range(8):
local_dispatches[i] = local_dispatches[i] * self.tu[i]
x = self.wai.forward_features(x, local_dispatches)
x_pool = F.adaptive_avg_pool2d(x, (1, 1)).flatten(1)
center_context = self.center_proj(x_pool)
x = x + center_context.view(B, -1, 1, 1) * 0.2
x = self.zhong(x, local_dispatches)
x_pool2 = F.adaptive_avg_pool2d(x, (1, 1)).flatten(1)
mem_out = self.memory(x_pool2, local_dispatches)
mem_out = mem_out.view(-1, 32, 1, 1)
x = x + mem_out * 0.3
logits = self.wai.forward_classifier(x)
return logits, {
'global_dispatch': global_dispatch,
'local_dispatches': local_dispatches,
'tu': self.tu,
'yin_losses': yin_losses,
'entropy': entropy.mean()
}
# ================== 8. 训练器（永续循环，自动续跑） ==================
class PerpetualTrainer:
def __init__(self, model, device, lr=0.001, checkpoint_dir='./heluo_perpetual'):
self.model = model
self.device = device
self.criterion = nn.CrossEntropyLoss()
self.base_lr = 0.0001
self.max_lr = lr
self.warmup_epochs = 3
self.current_epoch = 0
self.optimizer = optim.AdamW(model.parameters(), lr=self.base_lr, weight_decay=1e-4)
self.scheduler = optim.lr_scheduler.CosineAnnealingLR(self.optimizer, T_max=50)
self.checkpoint_dir = checkpoint_dir
os.makedirs(checkpoint_dir, exist_ok=True)
self.save_interval_epochs = 1 # 每1个epoch自动保存
self.running = True
# 注册信号处理，优雅退出
signal.signal(signal.SIGINT, self.signal_handler)
signal.signal(signal.SIGTERM, self.signal_handler)
def signal_handler(self, sig, frame):
print("\n收到中断信号，正在保存状态并退出...")
self.running = False
self.save_checkpoint(self.current_epoch)
sys.exit(0)
def update_learning_rate(self):
if self.current_epoch < self.warmup_epochs:
lr = self.base_lr + (self.max_lr - self.base_lr) * (self.current_epoch / self.warmup_epochs)
for param_group in self.optimizer.param_groups:
param_group['lr'] = lr
else:
self.scheduler.step()
def save_checkpoint(self, epoch):
self.model.save_full_state(os.path.join(self.checkpoint_dir, 'heluo'), epoch=epoch, optimizer=self.optimizer)
def load_latest_checkpoint(self):
# 尝试恢复最新的检查点
import glob
model_files = glob.glob(os.path.join(self.checkpoint_dir, 'heluo_model_epoch*.pth'))
if not model_files:
print("未找到已有检查点，从头开始训练。")
return 0
# 提取epoch
epochs = [int(f.split('_epoch')[-1].split('.')[0]) for f in model_files]
latest_epoch = max(epochs)
print(f"发现检查点 epoch {latest_epoch}，正在恢复...")
opt_state, loaded_epoch = self.model.load_full_state(
os.path.join(self.checkpoint_dir, 'heluo'), epoch=latest_epoch, device=self.device, load_optimizer=True
)
if opt_state:
self.optimizer.load_state_dict(opt_state)
print(f"优化器状态已恢复，继续从 epoch {loaded_epoch+1} 开始训练")
return loaded_epoch
else:
print("仅恢复模型和记忆，优化器状态未找到，将从头开始学习率调度。")
return latest_epoch
def train_epoch(self, train_loader):
self.model.train()
total_loss = 0
correct = 0
total = 0
for batch_idx, (data, target) in enumerate(train_loader):
if not self.running:
break
data, target = data.to(self.device), target.to(self.device)
self.optimizer.zero_grad()
logits, aux = self.model(data)
loss = self.criterion(logits, target)
dispatch_std = torch.std(torch.stack([d.mean() for d in aux['local_dispatches']]))
diversity_loss = 0.05 * max(0, 1.0 - dispatch_std.item())
coupling_std = torch.std(aux['tu'])
sparsity_loss = SafetyBoundary.COUPLING_SPARSITY_WEIGHT * max(
0, SafetyBoundary.COUPLING_SPARSITY_TARGET - coupling_std.item()
)
total_loss_val = loss + diversity_loss + sparsity_loss
total_loss_val.backward()
torch.nn.utils.clip_grad_norm_(self.model.parameters(), 1.0)
self.optimizer.step()
self.model.apply_safety_guard()
total_loss += loss.item()
pred = logits.argmax(dim=1)
correct += pred.eq(target).sum().item()
total += target.size(0)
if batch_idx % 100 == 0:
acc = 100. * correct / total if total > 0 else 0
current_lr = self.optimizer.param_groups[0]['lr']
print(f"Epoch {self.current_epoch} [{batch_idx}/{len(train_loader)}] "
f"Loss: {loss.item():.4f} | Acc: {acc:.2f}% | LR: {current_lr:.5f}")
avg_loss = total_loss / len(train_loader)
accuracy = 100. * correct / total
return avg_loss, accuracy
def evaluate(self, test_loader):
self.model.eval()
correct = 0
total = 0
test_loss = 0
with torch.no_grad():
for data, target in test_loader:
data, target = data.to(self.device), target.to(self.device)
logits, _ = self.model(data)
loss = self.criterion(logits, target)
test_loss += loss.item()
pred = logits.argmax(dim=1)
correct += pred.eq(target).sum().item()
total += target.size(0)
accuracy = 100. * correct / total
avg_loss = test_loss / len(test_loader)
return avg_loss, accuracy
def perpetual_train(self, train_loader, test_loader, save_interval_epochs=1):
# 尝试恢复到上次中断的进度
start_epoch = self.load_latest_checkpoint()
self.current_epoch = start_epoch
# 如果恢复时优化器状态未恢复，则重置学习率调度状态
if start_epoch > 0:
# 手动调整 scheduler 的步数
for _ in range(start_epoch):
self.scheduler.step()
print(f"开始永续进化，起始 epoch = {self.current_epoch}")
print("道循环永无断绝，训练将永远继续... (按 Ctrl+C 安全退出)\n")
best_acc = 0.0
while self.running:
self.update_learning_rate()
train_loss, train_acc = self.train_epoch(train_loader)
if not self.running:
break
test_loss, test_acc = self.evaluate(test_loader)
print(f"\nEpoch {self.current_epoch} 完成")
print(f" 训练 - Loss: {train_loss:.4f} | Acc: {train_acc:.2f}%")
print(f" 测试 - Loss: {test_loss:.4f} | Acc: {test_acc:.2f}%")
with torch.no_grad():
tu = self.model.tu.cpu().numpy()
print(f" 土层耦合系数: [{tu[0]:.2f}, {tu[1]:.2f}, {tu[2]:.2f}, {tu[3]:.2f}, "
f"{tu[4]:.2f}, {tu[5]:.2f}, {tu[6]:.2f}, {tu[7]:.2f}]")
print(f" 土层标准差: {tu.std():.4f}")
# 保存最佳模型
if test_acc > best_acc:
best_acc = test_acc
torch.save(self.model.state_dict(), os.path.join(self.checkpoint_dir, 'heluo_best_model.pth'))
self.model.memory.save_states(os.path.join(self.checkpoint_dir, 'heluo_best_memory.pt'))
print(f" ★ 新最佳模型！准确率: {best_acc:.2f}% (记忆已同步保存)")
# 定期保存检查点
if self.current_epoch % save_interval_epochs == 0:
self.save_checkpoint(self.current_epoch)
self.current_epoch += 1
print("-" * 70)
# ================== 9. 主程序 ==================
def main():
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
print("=" * 70)
print("河图洛书智能体 v1.3 - 永续进化版")
print("道循环：无始无终，自动续跑，永久记忆")
print("=" * 70)
transform_train = transforms.Compose([
transforms.RandomAffine(degrees=5, translate=(0.05, 0.05)),
transforms.ToTensor(),
transforms.Normalize((0.1307,), (0.3081,))
])
transform_test = transforms.Compose([
transforms.ToTensor(),
transforms.Normalize((0.1307,), (0.3081,))
])
train_dataset = datasets.MNIST('.', train=True, download=True, transform=transform_train)
test_dataset = datasets.MNIST('.', train=False, download=True, transform=transform_test)
train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True, num_workers=0)
test_loader = DataLoader(test_dataset, batch_size=100, shuffle=False, num_workers=0)
model = HeTuLuoShuAgent(num_classes=10, memory_persistent=True).to(device)
trainer = PerpetualTrainer(model, device, lr=0.001, checkpoint_dir='./heluo_perpetual')
total_params = sum(p.numel() for p in model.parameters())
print(f"\n模型参数量: {total_params:,}")
print(f"设备: {device}")
print("记忆模式: 永久持久化（跨epoch保留）")
print("检查点目录: ./heluo_perpetual")
print("训练将无限继续，每1个epoch自动保存，可用 Ctrl+C 安全终止并保存状态。")
trainer.perpetual_train(train_loader, test_loader, save_interval_epochs=1)
if __name__ == '__main__':
main()
|
鄂公网安备42060002000282号 )
