Update network.py

network.py

@@ -1,130 +1,115 @@
+# cognitive_net/network.py
 import torch
 import torch.nn as nn
-import torch.nn.functional as F
-from torch import optim
+import torch.optim as optim
 import math
-from typing import Optional
-
+from typing import Dict, List, Optional
 from .node import CognitiveNode
 
 class DynamicCognitiveNet(nn.Module):
-    """Self-organizing cognitive network with structure learning"""
+    """Self-organizing neural architecture with structural plasticity"""
     def __init__(self, input_size: int, output_size: int):
         super().__init__()
         self.input_size = input_size
         self.output_size = output_size
         
-        # Initialize core nodes
-        self.nodes = nn.ModuleDict({
-            f'input_{i}': CognitiveNode(i, 1) for i in range(input_size)
-        })
+        # Neural population
+        self.input_nodes = nn.ModuleList([
+            CognitiveNode(i, 1) for i in range(input_size)
+        ])
         self.output_nodes = nn.ModuleList([
             CognitiveNode(input_size + i, 1) for i in range(output_size)
         ])
         
-        # Structure learning parameters
-        self.connection_strength = nn.ParameterDict()
-        self.init_connections()
+        # Structural configuration
+        self.connections: Dict[str, nn.Parameter] = nn.ParameterDict()
+        self._init_base_connections()
         
-        # Emotional context
+        # Meta-learning components
         self.emotional_state = nn.Parameter(torch.tensor(0.0))
-        self.learning_rate = 0.01
-        
-        # Adaptive learning
         self.optimizer = optim.AdamW(self.parameters(), lr=0.001)
         self.loss_fn = nn.MSELoss()
-        
-    def init_connections(self):
-        """Initialize sparse random connections"""
-        for i in range(self.input_size):
-            for out_node in self.output_nodes:
-                conn_id = f'input_{i}->{out_node.id}'
-                self.connection_strength[conn_id] = nn.Parameter(
+
+    def _init_base_connections(self):
+        """Initialize sparse input-output connectivity"""
+        for i, in_node in enumerate(self.input_nodes):
+            for j, out_node in enumerate(self.output_nodes):
+                conn_id = f"{in_node.id}->{out_node.id}"
+                self.connections[conn_id] = nn.Parameter(
                     torch.randn(1) * 0.1
                 )
-                
+
     def forward(self, x: torch.Tensor) -> torch.Tensor:
-        # Process inputs
+        # Input processing
         activations = {}
-        for i in range(self.input_size):
-            node = self.nodes[f'input_{i}']
+        for i, node in enumerate(self.input_nodes):
             activations[node.id] = node(x[i].unsqueeze(0))
-            
-        # Propagate through network
+        
+        # Output integration
         outputs = []
         for out_node in self.output_nodes:
-            input_acts = []
-            for i in range(self.input_size):
-                conn_id = f'input_{i}->{out_node.id}'
-                weight = self.connection_strength.get(conn_id, torch.tensor(0.0))
-                input_acts.append(activations[i] * torch.sigmoid(weight))
-                
-            if input_acts:
-                combined = sum(input_acts) / math.sqrt(len(input_acts))
-                out_act = out_node(combined.unsqueeze(0))
-                outputs.append(out_act)
-                
+            integrated = []
+            for in_node in self.input_nodes:
+                conn_id = f"{in_node.id}->{out_node.id}"
+                weight = torch.sigmoid(self.connections[conn_id])
+                integrated.append(activations[in_node.id] * weight)
+            
+            if integrated:
+                combined = sum(integrated) / math.sqrt(len(integrated))
+                outputs.append(out_node(combined))
+        
         return torch.cat(outputs)
-    
-    def structural_update(self, reward: float):
-        """Adapt network structure based on performance"""
-        # Strengthen productive connections
-        for conn_id, weight in self.connection_strength.items():
-            if reward > 0:
-                new_strength = weight + self.learning_rate * reward
+
+    def structural_update(self, global_reward: float):
+        """Evolutionary architecture modification"""
+        # Connection strength adaptation
+        for conn_id, weight in self.connections.items():
+            if global_reward > 0:
+                new_weight = weight + 0.1 * global_reward
             else:
-                new_strength = weight * 0.9
-            self.connection_strength[conn_id].data = torch.clamp(new_strength, -1, 1)
-            
-        # Add new connections if performance is poor
-        if reward < -0.5 and torch.rand(1).item() < 0.3:
-            new_conn = self._create_new_connection()
-            if new_conn:
-                self.connection_strength[new_conn] = nn.Parameter(
+                new_weight = weight * 0.95
+            self.connections[conn_id].data = new_weight.clamp(-1, 1)
+        
+        # Structural neurogenesis
+        if global_reward < -0.5:
+            new_conn = self._generate_connection()
+            if new_conn not in self.connections:
+                self.connections[new_conn] = nn.Parameter(
                     torch.randn(1) * 0.1
                 )
-                
-    def _create_new_connection(self) -> Optional[str]:
-        """Create new random connection between underutilized nodes"""
-        # Find least active nodes
-        node_activations = {
-            node_id: sum(node.recent_activations.values()) / len(node.recent_activations)
-            for node_id, node in self.nodes.items()
-            if node.recent_activations
-        }
+
+    def _generate_connection(self) -> str:
+        """Create new input-output connection based on activity"""
+        input_act = {n.id: np.mean(n.recent_activations) 
+                    for n in self.input_nodes if n.recent_activations}
+        output_act = {n.id: np.mean(n.recent_activations)
+                     for n in self.output_nodes if n.recent_activations}
         
-        if not node_activations:
-            return None
-            
-        # Select random underutilized node pair
-        sorted_nodes = sorted(node_activations.items(), key=lambda x: x[1])
-        if len(sorted_nodes) < 2:
-            return None
+        if not input_act or not output_act:
+            return ""
             
-        source = sorted_nodes[0][0]
-        target = sorted_nodes[1][0]
-        
-        return f"{source}->{target}"
-    
+        src = min(input_act, key=input_act.get)  # type: ignore
+        tgt = min(output_act, key=output_act.get)  # type: ignore
+        return f"{src}->{tgt}"
+
     def train_step(self, x: torch.Tensor, y: torch.Tensor) -> float:
-        """Execute a single training step"""
         self.optimizer.zero_grad()
         pred = self(x)
         loss = self.loss_fn(pred, y)
         
-        # Add structural regularization
-        reg_loss = sum(torch.abs(w).mean() for w in self.connection_strength.values())
+        # Structural regularization
+        reg_loss = sum(p.abs().mean() for p in self.connections.values())
         total_loss = loss + 0.01 * reg_loss
         
         total_loss.backward()
         self.optimizer.step()
         
-        # Update emotional context
+        # Emotional state adaptation
         self.emotional_state.data = torch.sigmoid(
             self.emotional_state + (0.5 - loss.item()) * 0.1
         )
         
-        # Structural updates
-        self.structural_update(reward=0.5 - loss.item())
+        # Structural reorganization
+        self.structural_update(0.5 - loss.item())
         
         return total_loss.item()