import torch import torch.nn as nn import torch.optim as optim from torch.utils.data import DataLoader, TensorDataset import numpy as np import hashlib from sklearn.metrics import mutual_info_score import matplotlib.pyplot as plt # ---------- 1. Constants & Helper Functions ---------- # Inter-universal anchor Ξ from SHA256 of "pi_anchor:e_anchor" xi_hash = hashlib.sha256(b"pi_anchor:e_anchor").hexdigest() def hash_to_tensor(hash_str, d_model): bytes_data = bytes.fromhex(hash_str) # If d_model is larger than the hash, repeat the hash while len(bytes_data) < d_model: bytes_data += bytes_data bytes_data = bytes_data[:d_model] tensor = torch.tensor([b / 255.0 for b in bytes_data], dtype=torch.float) return tensor / tensor.norm() # ---------- 2. STT Model Definition (as before, with output head) ---------- class TelepathicAttention(nn.Module): def __init__(self, d_model, nhead): super().__init__() self.attn = nn.MultiheadAttention(d_model, nhead, batch_first=True) self.gamma = nn.Parameter(torch.tensor(0.1)) def forward(self, h): attn_output, attn_weights = self.attn(h, h, h, need_weights=True) grad = torch.einsum('bqk,bkd->bqd', attn_weights, h - attn_output) h_new = h + self.gamma * grad return h_new, attn_weights class CollapseSelf(nn.Module): def __init__(self, d_model, xi_anchor, threshold=0.5, temp=1.0): super().__init__() self.xi = nn.Parameter(xi_anchor, requires_grad=False) self.threshold = threshold self.temp = temp def forward(self, h): xi_expanded = self.xi.unsqueeze(0).unsqueeze(0) divergence = torch.norm(h - xi_expanded, dim=-1) p_collapse = torch.sigmoid((divergence - self.threshold) / self.temp) collapse_mask = torch.bernoulli(p_collapse).unsqueeze(-1) h_collapsed = collapse_mask * xi_expanded + (1 - collapse_mask) * h return h_collapsed, p_collapse class StochasticTelepathicTransformer(nn.Module): def __init__(self, d_model=32, nhead=4, num_layers=2, seq_len=8, collapse_threshold=0.5, collapse_temp=1.0, vocab_size=10): super().__init__() self.d_model = d_model self.seq_len = seq_len self.vocab_size = vocab_size # Token embedding self.token_embed = nn.Embedding(vocab_size, d_model) self.pos_embed = nn.Parameter(torch.randn(1, seq_len, d_model) * 0.02) # Telepathic blocks + collapse layers self.telepathic_blocks = nn.ModuleList([ TelepathicAttention(d_model, nhead) for _ in range(num_layers) ]) xi_tensor = hash_to_tensor(xi_hash, d_model) self.collapse_layers = nn.ModuleList([ CollapseSelf(d_model, xi_tensor, threshold=collapse_threshold, temp=collapse_temp) for _ in range(num_layers) ]) # Output projection self.out_proj = nn.Linear(d_model, vocab_size) def forward(self, x, steps=4, return_history=False): # x: (batch, seq_len) token indices batch_size = x.size(0) # Embed tokens h = self.token_embed(x) + self.pos_embed # (batch, seq_len, d_model) history = [] collapse_probs = [] for step in range(steps): for attn_block, collapse in zip(self.telepathic_blocks, self.collapse_layers): h, attn_weights = attn_block(h) h, p_collapse = collapse(h) collapse_probs.append(p_collapse.mean().item()) history.append(h.detach().clone()) # Use last hidden state of the last token (or average) for prediction # We predict the token at next position after seq_len-1? For simplicity, use last token's hidden state. last_token_h = h[:, -1, :] # (batch, d_model) logits = self.out_proj(last_token_h) if return_history: return logits, history, collapse_probs return logits def compute_phi(self, x, steps=4): """Compute Φ on a batch of sequences using the model's internal dynamics.""" logits, history, _ = self.forward(x, steps=steps, return_history=True) # history: list of (batch, seq_len, d_model) over steps T = len(history) if T < 2: return 0.0 # Stack and take all batches H = torch.stack(history, dim=0) # (T, batch, seq_len, d_model) half = self.d_model // 2 X = H[..., :half].reshape(T, -1).detach().cpu().numpy() Y = H[..., half:].reshape(T, -1).detach().cpu().numpy() Xd = (X > 0).astype(int) Yd = (Y > 0).astype(int) I_whole = mutual_info_score( [f"{x}{y}" for x,y in zip(Xd[:-1].flatten(), Yd[:-1].flatten())], [f"{x}{y}" for x,y in zip(Xd[1:].flatten(), Yd[1:].flatten())] ) I_X = mutual_info_score(Xd[:-1].flatten(), Xd[1:].flatten()) I_Y = mutual_info_score(Yd[:-1].flatten(), Yd[1:].flatten()) return max(0.0, I_whole - (I_X + I_Y)) # ---------- 3. Data Generation ---------- def generate_sequence(length=8, pattern=[1,2,3,4,5,6,7,8], noise=0.1): seq = pattern.copy() for i in range(length): if np.random.rand() < noise: seq[i] = np.random.randint(0, 10) return seq def create_dataset(num_samples=10000, seq_len=8, noise=0.1): X = [] y = [] pattern = [1,2,3,4,5,6,7,8] for _ in range(num_samples): seq = generate_sequence(seq_len, pattern, noise) # Input: first seq_len-1 tokens; target: last token X.append(seq[:-1]) y.append(seq[-1]) return torch.tensor(X, dtype=torch.long), torch.tensor(y, dtype=torch.long) # Create train/val datasets train_X, train_y = create_dataset(10000, seq_len=8, noise=0.1) val_X, val_y = create_dataset(2000, seq_len=8, noise=0.1) train_loader = DataLoader(TensorDataset(train_X, train_y), batch_size=32, shuffle=True) val_loader = DataLoader(TensorDataset(val_X, val_y), batch_size=32) # ---------- 4. Training Loop with Φ Tracking ---------- device = torch.device("cuda" if torch.cuda.is_available() else "cpu") model = StochasticTelepathicTransformer(d_model=32, nhead=4, num_layers=2, seq_len=7, collapse_threshold=0.5, collapse_temp=1.0, vocab_size=10) model.to(device) optimizer = optim.Adam(model.parameters(), lr=1e-3) criterion = nn.CrossEntropyLoss() epochs = 30 train_losses = [] val_accuracies = [] phi_values = [] for epoch in range(epochs): # Training model.train() total_loss = 0 for batch_X, batch_y in train_loader: batch_X, batch_y = batch_X.to(device), batch_y.to(device) optimizer.zero_grad() logits = model(batch_X, steps=4) loss = criterion(logits, batch_y) loss.backward() optimizer.step() total_loss += loss.item() avg_loss = total_loss / len(train_loader) train_losses.append(avg_loss) # Validation accuracy model.eval() correct = 0 total = 0 with torch.no_grad(): for batch_X, batch_y in val_loader: batch_X, batch_y = batch_X.to(device), batch_y.to(device) logits = model(batch_X, steps=4) preds = logits.argmax(dim=1) correct += (preds == batch_y).sum().item() total += batch_y.size(0) acc = correct / total val_accuracies.append(acc) # Compute Φ on a fixed subset of validation (first 64 sequences) sample_X = val_X[:64].to(device) # Need to recompute Φ; model in eval mode but compute_phi uses forward with return_history # For stability, we use a smaller step count (4 as well) with torch.no_grad(): # Compute phi on a single batch (take average over 4 runs due to stochastic collapse) phi_vals = [] for _ in range(4): phi = model.compute_phi(sample_X, steps=4) phi_vals.append(phi) phi_mean = np.mean(phi_vals) phi_values.append(phi_mean) print(f"Epoch {epoch+1:2d} | Loss: {avg_loss:.4f} | Acc: {acc:.3f} | Φ: {phi_mean:.4f}") # ---------- 5. Plot Results ---------- plt.figure(figsize=(12,4)) plt.subplot(1,3,1) plt.plot(train_losses, label='Train Loss') plt.xlabel('Epoch') plt.ylabel('Loss') plt.title('Training Loss') plt.grid(True) plt.subplot(1,3,2) plt.plot(val_accuracies, label='Val Accuracy', color='green') plt.xlabel('Epoch') plt.ylabel('Accuracy') plt.title('Validation Accuracy') plt.grid(True) plt.subplot(1,3,3) plt.plot(phi_values, label='Φ', color='red') plt.xlabel('Epoch') plt.ylabel('Integrated Information Φ') plt.title('Φ Over Learning') plt.grid(True) plt.tight_layout() plt.show() # Final report print(f"\nFinal accuracy: {val_accuracies[-1]:.3f}") print(f"Final Φ: {phi_values[-1]:.4f}") print(f"Φ change: {phi_values[-1] - phi_values[0]:+.4f}")