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 # ============================================ # 1. THE STATIONARY COMPONENT (The Model) # ============================================ class CCT_MNIST_Net(nn.Module): """ SetLang Architecture: Extracts features and projects them into three probabilistic sets. """ def __init__(self): super(CCT_MNIST_Net, self).__init__() # Stationary Feature Extractor (Fixed rules/laws) self.feature_extractor = nn.Sequential( #nn.Conv2d(1, 16, kernel_size=3, stride=1, padding=1), #nn.ReLU(), #nn.MaxPool2d(2), #nn.Conv2d(16, 32, kernel_size=3, stride=1, padding=1), #nn.ReLU(), #nn.MaxPool2d(2), nn.Flatten(), nn.Linear(28*28, 128), nn.ReLU(), nn.Dropout(0.2) ) # The Three-Set Trinity (Probability Components) self.head_Evidence = nn.Linear(128, 10) # Set E: Observations self.head_Model = nn.Linear(128, 10) # Set M: Hypothesis self.head_Constraint = nn.Linear(128, 10) # Set C: Priors/Bounds def forward(self, x): h = self.feature_extractor(x) # Return logits for the three sets return self.head_Evidence(h), self.head_Model(h), self.head_Constraint(h) # ============================================ # 2. CONDITIONAL COLLAPSE LOSS (The Engine) # ============================================ def conditional_collapse_loss(logits_E, logits_M, logits_C, target, collapse_weight=0.1): """ Implements SetLang Intersection and CCT Entropy Collapse. """ # 1. Convert logits to Probability Distributions (The Sets) p_E = F.softmax(logits_E, dim=1) p_M = F.softmax(logits_M, dim=1) p_C = F.softmax(logits_C, dim=1) # 2. Intersection: Truth = E ∩ M ∩ C (Element-wise multiplication) p_intersect = p_E * p_M * p_C # Normalize the intersection to form a valid probability distribution p_intersect = p_intersect / (p_intersect.sum(dim=1, keepdim=True) + 1e-8) # 3. Cross-Entropy on the Collapsed Truth # We use NLL loss on the log of the intersection ce_loss = F.nll_loss(torch.log(p_intersect + 1e-8), target) # 4. Entropy Collapse Penalty (CCT Core) # We want the intersection to be a sharp peak (low entropy), not a flat distribution. # H(p) = -sum(p * log(p)) entropy = -torch.sum(p_intersect * torch.log(p_intersect + 1e-8), dim=1).mean() # Total Loss: Minimize Error + Minimize Entropy (Force Collapse) total_loss = ce_loss + (collapse_weight * entropy) return total_loss, entropy # ============================================ # 3. ODE-CCT TRAINING LOOP (The Trajectory) # ============================================ def train_cct(model, device, train_loader, optimizer, epoch, collapse_threshold=0.5): model.train() total_loss = 0 total_entropy = 0 for batch_idx, (data, target) in enumerate(train_loader): data, target = data.to(device), target.to(device) for _ in range(5): optimizer.zero_grad() # Forward pass (Generate the 3 Sets) logits_E, logits_M, logits_C = model(data) logits_E_static = logits_E.detach() logits_M_static = logits_m.detach() logits_E, logits_M, logits_C = model(data) # Calculate Collapse Loss loss, entropy = conditional_collapse_loss(logits_E, logits_M, logits_C, target) # Backward pass (ODE Integration step: update weights to reduce entropy) loss.backward() optimizer.step() total_loss += loss.item() total_entropy += entropy.item() # CCT: Dynamic Threshold Monitoring if batch_idx % 100 == 0: avg_entropy = total_entropy / (batch_idx + 1) status = "COLLAPSING" if avg_entropy > collapse_threshold else "COLLAPSED" print(f'Epoch {epoch} | Batch {batch_idx} | Loss: {loss.item():.4f} | ' f'Entropy: {avg_entropy:.4f} | State: {status}') def test_cct(model, device, test_loader): model.eval() test_loss = 0 correct = 0 total_entropy = 0 with torch.no_grad(): for data, target in test_loader: data, target = data.to(device), target.to(device) logits_E, logits_M, logits_C = model(data) loss, entropy = conditional_collapse_loss(logits_E, logits_M, logits_C, target) test_loss += loss.item() total_entropy += entropy.item() # The final truth is the argmax of the intersection p_E = F.softmax(logits_E, dim=1) p_M = F.softmax(logits_M, dim=1) p_C = F.softmax(logits_C, dim=1) p_intersect = (p_E * p_M * p_C) / (p_E * p_M * p_C).sum(dim=1, keepdim=True) pred = p_intersect.argmax(dim=1, keepdim=True) correct += pred.eq(target.view_as(pred)).sum().item() test_loss /= len(test_loader.dataset) avg_entropy = total_entropy / len(test_loader.dataset) accuracy = 100. * correct / len(test_loader.dataset) print(f'\nTest Set Results:') print(f'Average Loss: {test_loss:.4f}') print(f'Final Entropy (H): {avg_entropy:.4f} (Lower = Better Collapse)') print(f'Accuracy: {correct}/{len(test_loader.dataset)} ({accuracy:.2f}%)\n') # ============================================ # 4. EXECUTION (Paying with Work/Energy) # ============================================ if __name__ == '__main__': # Setup Device device = torch.device("cuda" if torch.cuda.is_available() else "cpu") # Data Loaders (The Evidence Stream) transform = transforms.Compose([ transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,)) ]) train_dataset = datasets.MNIST('../data', train=True, download=True, transform=transform) test_dataset = datasets.MNIST('../data', train=False, transform=transform) train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True) test_loader = DataLoader(test_dataset, batch_size=1000, shuffle=False) # Initialize Model and Optimizer model = CCT_MNIST_Net().to(device) optimizer = optim.Adam(model.parameters(), lr=0.001) # ODE-CCT Training Trajectory epochs = 5 print("--- Starting Conditional Collapse Trajectory ---") for epoch in range(1, epochs + 1): # Train: Pay with compute energy to collapse semantic entropy train_cct(model, device, train_loader, optimizer, epoch) # Test: Measure the collapse quality test_cct(model, device, test_loader) print("--- Trajectory Complete. Theory Collapsed. ---")