Update network.py

network.py

@@ -1,19 +1,19 @@
-# cognitive_net/network.py
 import torch
 import torch.nn as nn
 import torch.optim as optim
 import math
-from typing import Dict, List, Optional
+import numpy as np
+from typing import Dict, Optional
 from .node import CognitiveNode
 
 class DynamicCognitiveNet(nn.Module):
-    """Self-organizing neural architecture with structural plasticity"""
+    """Jaringan dengan manajemen koneksi yang robust"""
     def __init__(self, input_size: int, output_size: int):
         super().__init__()
         self.input_size = input_size
         self.output_size = output_size
         
-        # Neural population
+        # Node input/output
         self.input_nodes = nn.ModuleList([
             CognitiveNode(i, 1) for i in range(input_size)
         ])
@@ -21,31 +21,31 @@ class DynamicCognitiveNet(nn.Module):
             CognitiveNode(input_size + i, 1) for i in range(output_size)
         ])
         
-        # Structural configuration
-        self.connections: Dict[str, nn.Parameter] = nn.ParameterDict()
+        # Manajemen koneksi
+        self.connections = nn.ParameterDict()
         self._init_base_connections()
         
-        # Meta-learning components
+        # Konteks pembelajaran
         self.emotional_state = nn.Parameter(torch.tensor(0.0))
         self.optimizer = optim.AdamW(self.parameters(), lr=0.001)
         self.loss_fn = nn.MSELoss()
 
     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):
+        """Inisialisasi koneksi input-output"""
+        for in_node in self.input_nodes:
+            for out_node in 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:
-        # Input processing
+        # Pemrosesan input
         activations = {}
         for i, node in enumerate(self.input_nodes):
             activations[node.id] = node(x[i].unsqueeze(0))
         
-        # Output integration
+        # Integrasi output
         outputs = []
         for out_node in self.output_nodes:
             integrated = []
@@ -58,58 +58,63 @@ class DynamicCognitiveNet(nn.Module):
                 combined = sum(integrated) / math.sqrt(len(integrated))
                 outputs.append(out_node(combined))
         
-        return torch.cat(outputs)
+        return torch.stack(outputs).squeeze()
 
     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_weight = weight * 0.95
+        """Update struktur dengan validasi koneksi"""
+        # Adaptasi kekuatan koneksi
+        for conn_id in list(self.connections.keys()):
+            new_weight = self.connections[conn_id] + 0.1 * global_reward
             self.connections[conn_id].data = new_weight.clamp(-1, 1)
         
-        # Structural neurogenesis
+        # Pembuatan koneksi baru
         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
-                )
+            new_conn = self._find_underutilized_connection()
+            if new_conn and new_conn not in self.connections:
+                self.connections[new_conn] = nn.Parameter(torch.randn(1) * 0.1)
 
-    def _generate_connection(self) -> str:
-        """Create new input-output connection based on activity"""
+    def _find_underutilized_connection(self) -> Optional[str]:
+        """Mencari koneksi input-output yang underutilized"""
         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 input_act or not output_act:
-            return ""
+            return None
             
-        src = min(input_act, key=input_act.get)  # type: ignore
-        tgt = min(output_act, key=output_act.get)  # type: ignore
+        src = min(input_act, key=lambda k: input_act[k])
+        tgt = min(output_act, key=lambda k: output_act[k])
         return f"{src}->{tgt}"
 
     def train_step(self, x: torch.Tensor, y: torch.Tensor) -> float:
+        """Training step dengan error handling"""
         self.optimizer.zero_grad()
-        pred = self(x)
-        loss = self.loss_fn(pred, y)
         
-        # Structural regularization
+        try:
+            pred = self(x)
+            loss = self.loss_fn(pred, y)
+        except RuntimeError as e:
+            print(f"Error selama forward pass: {e}")
+            return float('nan')
+        
+        # Regularisasi struktural
         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()
+        try:
+            total_loss.backward()
+            self.optimizer.step()
+        except RuntimeError as e:
+            print(f"Error selama backpropagation: {e}")
+            return float('nan')
         
-        # Emotional state adaptation
+        # Update state emosional
         self.emotional_state.data = torch.sigmoid(
             self.emotional_state + (0.5 - loss.item()) * 0.1
         )
         
-        # Structural reorganization
+        # Update struktur
         self.structural_update(0.5 - loss.item())
         
         return total_loss.item()