import torch import torch.nn as nn import torch.nn.functional as F import torch.optim as optim from torchvision import datasets, transforms from torch.utils.data import DataLoader import math # ===================================================================== # 1. THE ARCHITECTURE (ODE-CCT Skeleton Layering) # ===================================================================== class CCTMnistNet(nn.Module): def __init__(self): super(CCTMnistNet, self).__init__() # Stationary Structural Components (Fixed Geometry) self.conv1 = nn.Conv2d(1, 16, kernel_size=3, padding=1) self.conv2 = nn.Conv2d(16, 32, kernel_size=3, padding=1) self.pool = nn.MaxPool2d(2, 2) self.fc1 = nn.Linear(32 * 7 * 7, 128) self.fc2 = nn.Linear(128, 10) def forward(self, x): # Maps raw trajectory variables into lower-dimensional manifolds x = self.pool(F.relu(self.conv1(x))) x = self.pool(F.relu(self.conv2(x))) x = x.view(-1, 32 * 7 * 7) x = F.relu(self.fc1(x)) return self.fc2(x) # ===================================================================== # 2. THE CCT THEORY ENGINE (Loops, Constraints, and Entropy Dynamics) # ===================================================================== class CCTTheoryEngine: def __init__(self, base_lr=0.01): self.base_lr = base_lr self.loss_history = [] self.entropy_history = [] self.bifurcation_caches = [] def calculate_shannon_entropy(self, outputs): """Helper to calculate real-time model operational uncertainty H(T).""" probs = F.softmax(outputs, dim=1) # Avoid log(0) probs = torch.clamp(probs, min=1e-8) entropy = -torch.sum(probs * torch.log(probs), dim=1).mean().item() return entropy def monitor_and_adapt(self, current_loss, outputs, step, optimizer): """ Processes batch steps by treating loss/entropy as dynamic ODE systems. Applies theory features to inject micro-adjustments directly to training. """ self.loss_history.append(current_loss) current_entropy = self.calculate_shannon_entropy(outputs) self.entropy_history.append(current_entropy) # Default operational state values adjusted_lr = self.base_lr noise_filter_active = False attractor_pull = 0.0 if len(self.loss_history) < 2: return adjusted_lr, noise_filter_active, attractor_pull # Extracting instantaneous derivatives (dx/dt) d_loss = self.loss_history[-1] - self.loss_history[-2] d_entropy = self.entropy_history[-1] - self.entropy_history[-2] # ------------------------------------------------------------- # FEATURE 1: Stochastic Noise Suppression # ------------------------------------------------------------- # If loss fluctuates violently outside regular law gradients, treat as chaos noise. if abs(d_loss) > 0.5 and current_loss > 1.5: noise_filter_active = True # Instantly damp learning rate to isolate operations from volatile noise spikes adjusted_lr = self.base_lr * 0.1 # ------------------------------------------------------------- # FEATURE 2: Bifurcation Point Caching # ------------------------------------------------------------- # Detect critical fork-in-the-road inflection thresholds where state trajectories flip. elif d_loss < -0.2 and d_entropy < -0.1: print(f"\n[CCT ENGINE] Bifurcation Point Cached at step {step}! Sharp optimization dive.") self.bifurcation_caches.append((step, current_loss)) # Lock the optimal path momentum by providing a high velocity gradient adjusted_lr = self.base_lr * 1.5 # ------------------------------------------------------------- # FEATURE 3: Dynamic Threshold Shifting # ------------------------------------------------------------- # Dynamically scale operational complexity based on system convergence load. # If structural entropy H(T) is low, the model is confident; allow finer gradient steps. elif current_entropy < 0.2: adjusted_lr = self.base_lr * 0.5 # ------------------------------------------------------------- # FEATURE 4: Dynamic Attractor Synthesis # ------------------------------------------------------------- # If the model is stagnating in a flat gradient phase-space plateau, # synthetically inject cross-entropy anchor mass to drag it forward. if len(self.loss_history) > 5 and abs(sum(self.loss_history[-5:]) / 5 - current_loss) < 0.01: attractor_pull = 0.05 # Artificial gradient acceleration magnitude # Update real-time optimizer trajectory constraints for param_group in optimizer.param_groups: param_group['lr'] = adjusted_lr return adjusted_lr, noise_filter_active, attractor_pull # ===================================================================== # 3. COMPREHENSIVE TRAINING & EVALUATION PIPELINE # ===================================================================== def run_cct_mnist_pipeline(): # Setup infrastructure device = torch.device("cuda" if torch.cuda.is_available() else "cpu") print(f"Executing CCT Pipeline on execution environment: {device}") transform = transforms.Compose([ transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,)) ]) train_dataset = datasets.MNIST(root='../data', train=True, download=True, transform=transform) test_dataset = datasets.MNIST(root='../data', train=False, download=True, transform=transform) train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True) test_loader = DataLoader(test_dataset, batch_size=1000, shuffle=False) model = CCTMnistNet().to(device) base_lr = 0.01 optimizer = optim.SGD(model.parameters(), lr=base_lr, momentum=0.9) criterion = nn.CrossEntropyLoss() # Initialize our concept framework overseer cct_engine = CCTTheoryEngine(base_lr=base_lr) epochs = 2 global_step = 0 print("\n--- Beginning CCT Multi-Manifold Training Loop ---") for epoch in range(1, epochs + 1): model.train() running_loss = 0.0 for batch_idx, (data, target) in enumerate(train_loader): data, target = data.to(device), target.to(device) optimizer.zero_grad() outputs = model(data) loss = criterion(outputs, target) # Extract underlying metrics to feed into the theoretical engine loss_val = loss.item() global_step += 1 # Calculate adjustments using our system features current_lr, noise_suppressed, attractor_force = cct_engine.monitor_and_adapt( loss_val, outputs, global_step, optimizer ) # Apply Dynamic Attractor Synthesis modification directly into the target objective if attractor_force > 0: loss = loss + (attractor_force * torch.norm(outputs)) loss.backward() optimizer.step() running_loss += loss_val if batch_idx % 200 == 0: status_flags = [] if noise_suppressed: status_flags.append("Noise_Suppression=ACTIVE") if attractor_force > 0: status_flags.append(f"Attractor_Pull={attractor_force}") status_str = f" [{', '.join(status_flags)}]" if status_flags else "" print(f"Epoch: {epoch} | Batch: {batch_idx}/{len(train_loader)} | " f"Batch Loss: {loss_val:.4f} | Dynamic LR: {current_lr:.5f}{status_str}") print(f"-> Epoch {epoch} Macro Average Loss: {running_loss/len(train_loader):.4f}") # ===================================================================== # FEATURE 5: Real-Time Testset Evaluation & State Validation # ===================================================================== model.eval() test_loss = 0 correct = 0 with torch.no_grad(): for data, target in test_loader: data, target = data.to(device), target.to(device) outputs = model(data) # Sum up evaluation phase spaces test_loss += criterion(outputs, target).item() pred = outputs.argmax(dim=1, keepdim=True) correct += pred.eq(target.view_as(pred)).sum().item() test_loss /= len(test_loader) test_acc = 100. * correct / len(test_loader.dataset) print(f"=== [EVAL SYSTEM] Test Set Evaluation ===") print(f"Average System Loss: {test_loss:.4f} | System Accuracy: {correct}/{len(test_loader.dataset)} ({test_acc:.2f}%)\n") if __name__ == '__main__': run_cct_mnist_pipeline()