Create network.py

network.py

@@ -0,0 +1,134 @@
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+from torch import optim
+import math
+from typing import Optional
+
+from .node import CognitiveNode
+
+class DynamicCognitiveNet(nn.Module):
+    """Self-organizing cognitive network with structure learning"""
+    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)
+        })
+        self.output_nodes = nn.ModuleList([
+            CognitiveNode(input_size + i, input_size) 
+            for i in range(output_size)
+        ])
+        
+        # Structure learning parameters
+        self.connection_strength = nn.ParameterDict()
+        self.recent_activations = {}
+        self.init_connections()
+        
+        # Emotional context
+        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(
+                    torch.randn(1) * 0.1
+                )
+                
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        # Process inputs
+        activations = {}
+        for i in range(self.input_size):
+            node = self.nodes[f'input_{i}']
+            activations[node.id] = node(x[i].unsqueeze(0))
+            
+        # Propagate through network
+        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)
+                outputs.append(out_act)
+                
+        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
+            else:
+                new_strength = weight * 0.9
+            self.connection_strength[conn_id] = nn.Parameter(
+                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(
+                    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: torch.mean(torch.stack(list(node.recent_activations.values())))
+            for node_id, node in self.nodes.items()
+            if hasattr(node, 'recent_activations') and node.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
+            
+        source = sorted_nodes[0][0]
+        target = sorted_nodes[1][0]
+        
+        return f"{source}->{target}"
+    
+    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())
+        total_loss = loss + 0.01 * reg_loss
+        
+        total_loss.backward()
+        self.optimizer.step()
+        
+        # Update emotional context
+        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())
+        
+        return total_loss.item()