| import gradio as gr |
| import numpy as np |
| import matplotlib.pyplot as plt |
|
|
| def relu(x): |
| return np.maximum(0, x) |
|
|
| def relu_derivative(x): |
| return (x > 0).astype(float) |
|
|
| def tanh(x): |
| return np.tanh(x) |
|
|
| def tanh_derivative(x): |
| return 1 - np.tanh(x)**2 |
|
|
| class EveOptimizer: |
| def __init__(self, params, learning_rate=0.001, beta1=0.9, beta2=0.999, epsilon=1e-8): |
| self.params = params |
| self.lr = learning_rate |
| self.beta1 = beta1 |
| self.beta2 = beta2 |
| self.epsilon = epsilon |
| self.t = 0 |
| self.m = [np.zeros_like(p) for p in params] |
| self.v = [np.zeros_like(p) for p in params] |
| self.fractal_memory = [np.zeros_like(p) for p in params] |
|
|
| def step(self, grads): |
| self.t += 1 |
| for i, (param, grad) in enumerate(zip(self.params, grads)): |
| self.m[i] = self.beta1 * self.m[i] + (1 - self.beta1) * grad |
| self.v[i] = self.beta2 * self.v[i] + (1 - self.beta2) * (grad ** 2) |
|
|
| m_hat = self.m[i] / (1 - self.beta1 ** self.t) |
| v_hat = self.v[i] / (1 - self.beta2 ** self.t) |
|
|
| fractal_factor = self.fractal_adjustment(param, grad) |
| self.fractal_memory[i] = 0.9 * self.fractal_memory[i] + 0.1 * fractal_factor |
|
|
| param -= self.lr * m_hat / (np.sqrt(v_hat) + self.epsilon) * self.fractal_memory[i] |
|
|
| def fractal_adjustment(self, param, grad): |
| c = np.mean(grad) + 1j * np.std(param) |
| z = 0 |
| for _ in range(10): |
| z = z**2 + c |
| if abs(z) > 2: |
| break |
| return 1 / (1 + abs(z)) |
|
|
| class BatchNormalization: |
| def __init__(self, input_shape): |
| self.gamma = np.ones(input_shape) |
| self.beta = np.zeros(input_shape) |
| self.epsilon = 1e-5 |
| self.moving_mean = np.zeros(input_shape) |
| self.moving_var = np.ones(input_shape) |
|
|
| def forward(self, x, training=True): |
| if training: |
| mean = np.mean(x, axis=0) |
| var = np.var(x, axis=0) |
| self.moving_mean = 0.99 * self.moving_mean + 0.01 * mean |
| self.moving_var = 0.99 * self.moving_var + 0.01 * var |
| else: |
| mean = self.moving_mean |
| var = self.moving_var |
|
|
| x_norm = (x - mean) / np.sqrt(var + self.epsilon) |
| out = self.gamma * x_norm + self.beta |
| if training: |
| self.cache = (x, x_norm, mean, var) |
| return out |
|
|
| def backward(self, dout): |
| x, x_norm, mean, var = self.cache |
| m = x.shape[0] |
|
|
| dx_norm = dout * self.gamma |
| dvar = np.sum(dx_norm * (x - mean) * -0.5 * (var + self.epsilon)**(-1.5), axis=0) |
| dmean = np.sum(dx_norm * -1 / np.sqrt(var + self.epsilon), axis=0) + dvar * np.mean(-2 * (x - mean), axis=0) |
|
|
| dx = dx_norm / np.sqrt(var + self.epsilon) + dvar * 2 * (x - mean) / m + dmean / m |
| dgamma = np.sum(dout * x_norm, axis=0) |
| dbeta = np.sum(dout, axis=0) |
|
|
| return dx, dgamma, dbeta |
|
|
| class Reward: |
| def __init__(self): |
| self.lowest_avg_batch_loss = float('inf') |
| self.lowest_max_batch_loss = float('inf') |
| self.best_weights = None |
|
|
| def update(self, avg_batch_loss, max_batch_loss, network): |
| improved = False |
| if avg_batch_loss < self.lowest_avg_batch_loss: |
| self.lowest_avg_batch_loss = avg_batch_loss |
| improved = True |
| if max_batch_loss < self.lowest_max_batch_loss: |
| self.lowest_max_batch_loss = max_batch_loss |
| improved = True |
| if improved: |
| self.best_weights = self.get_network_weights(network) |
|
|
| def get_network_weights(self, network): |
| weights = [] |
| for layer in network.layers: |
| layer_weights = [] |
| for agent in layer.agents: |
| agent_weights = { |
| 'weights': agent.weights.copy(), |
| 'bias': agent.bias.copy(), |
| 'bn_gamma': agent.bn.gamma.copy(), |
| 'bn_beta': agent.bn.beta.copy() |
| } |
| layer_weights.append(agent_weights) |
| weights.append(layer_weights) |
| return weights |
|
|
| def apply_best_weights(self, network): |
| if self.best_weights is not None: |
| for layer, layer_weights in zip(network.layers, self.best_weights): |
| for agent, agent_weights in zip(layer.agents, layer_weights): |
| agent.weights = agent_weights['weights'].copy() |
| agent.bias = agent_weights['bias'].copy() |
| agent.bn.gamma = agent_weights['bn_gamma'].copy() |
| agent.bn.beta = agent_weights['bn_beta'].copy() |
|
|
| class Agent: |
| def __init__(self, id, input_size, output_size, fractal_method): |
| self.id = id |
| self.weights = np.random.randn(input_size, output_size) * np.sqrt(2. / input_size) |
| self.bias = np.zeros((1, output_size)) |
| self.fractal_method = fractal_method |
| self.bn = BatchNormalization((output_size,)) |
|
|
| self.optimizer = EveOptimizer([self.weights, self.bias, self.bn.gamma, self.bn.beta]) |
|
|
| def forward(self, x, training=True): |
| self.last_input = x |
| z = np.dot(x, self.weights) + self.bias |
| z_bn = self.bn.forward(z, training) |
| self.last_output = relu(z_bn) |
| return self.last_output |
|
|
| def backward(self, error, l2_lambda=1e-5): |
| delta = error * relu_derivative(self.last_output) |
| delta, dgamma, dbeta = self.bn.backward(delta) |
|
|
| dw = np.dot(self.last_input.T, delta) + l2_lambda * self.weights |
| db = np.sum(delta, axis=0, keepdims=True) |
|
|
| self.optimizer.step([dw, db, dgamma, dbeta]) |
|
|
| return np.dot(delta, self.weights.T) |
|
|
| def apply_fractal(self, x): |
| return self.fractal_method(x) |
|
|
| class Swarm: |
| def __init__(self, num_agents, input_size, output_size, fractal_method): |
| self.agents = [Agent(i, input_size, output_size, fractal_method) for i in range(num_agents)] |
|
|
| def forward(self, x, training=True): |
| results = [agent.forward(x, training) for agent in self.agents] |
| return np.mean(results, axis=0) |
|
|
| def backward(self, error, l2_lambda): |
| errors = [agent.backward(error, l2_lambda) for agent in self.agents] |
| return np.mean(errors, axis=0) |
|
|
| def apply_fractal(self, x): |
| results = [agent.apply_fractal(x) for agent in self.agents] |
| return np.mean(results, axis=0) |
|
|
| class SwarmNeuralNetwork: |
| def __init__(self, layer_sizes, fractal_methods): |
| self.layers = [] |
| for i in range(len(layer_sizes) - 2): |
| self.layers.append(Swarm(num_agents=3, |
| input_size=layer_sizes[i], |
| output_size=layer_sizes[i+1], |
| fractal_method=fractal_methods[i])) |
| self.output_layer = Swarm(num_agents=1, |
| input_size=layer_sizes[-2], |
| output_size=layer_sizes[-1], |
| fractal_method=fractal_methods[-1]) |
| self.reward = Reward() |
|
|
| def forward(self, x, training=True): |
| self.layer_outputs = [x] |
| for layer in self.layers: |
| x = layer.forward(x, training) |
| self.layer_outputs.append(x) |
| self.final_output = tanh(self.output_layer.forward(x, training)) |
| return self.final_output |
|
|
| def backward(self, error, l2_lambda=1e-5): |
| error = error * tanh_derivative(self.final_output) |
| error = self.output_layer.backward(error, l2_lambda) |
| for i in reversed(range(len(self.layers))): |
| error = self.layers[i].backward(error, l2_lambda) |
|
|
| def train(self, X, y, epochs, batch_size=32, l2_lambda=1e-5, patience=50): |
| best_mse = float('inf') |
| patience_counter = 0 |
|
|
| for epoch in range(epochs): |
| indices = np.arange(len(X)) |
| np.random.shuffle(indices) |
|
|
| self.reward.apply_best_weights(self) |
|
|
| epoch_losses = [] |
| for start_idx in range(0, len(X) - batch_size + 1, batch_size): |
| batch_indices = indices[start_idx:start_idx+batch_size] |
| X_batch = X[batch_indices] |
| y_batch = y[batch_indices] |
|
|
| output = self.forward(X_batch) |
| error = y_batch - output |
|
|
| error = np.clip(error, -1, 1) |
|
|
| self.backward(error, l2_lambda) |
|
|
| epoch_losses.append(np.mean(np.square(error))) |
|
|
| avg_batch_loss = np.mean(epoch_losses) |
| max_batch_loss = np.max(epoch_losses) |
| self.reward.update(avg_batch_loss, max_batch_loss, self) |
|
|
| mse = np.mean(np.square(y - self.forward(X, training=False))) |
|
|
| if epoch % 100 == 0: |
| print(f"Epoch {epoch}, MSE: {mse:.6f}, Avg Batch Loss: {avg_batch_loss:.6f}, Min Batch Loss: {np.min(epoch_losses):.6f}, Max Batch Loss: {max_batch_loss:.6f}") |
|
|
| if mse < best_mse: |
| best_mse = mse |
| patience_counter = 0 |
| else: |
| patience_counter += 1 |
|
|
| if patience_counter >= patience: |
| print(f"Early stopping at epoch {epoch}") |
| break |
|
|
| return best_mse |
|
|
| def apply_fractals(self, x): |
| fractal_outputs = [] |
| for i, layer in enumerate(self.layers): |
| x = self.layer_outputs[i+1] |
| fractal_output = layer.apply_fractal(x) |
| fractal_outputs.append(fractal_output) |
| return fractal_outputs |
|
|
| def sierpinski_fractal(input_data): |
| t = np.linspace(0, 2 * np.pi, input_data.shape[0]) |
| x = np.mean(input_data) * np.cos(t) |
| y = np.mean(input_data) * np.sin(t) |
| return x, y |
|
|
| def mandelbrot_fractal(input_data, max_iter=10): |
| output = np.zeros(input_data.shape[0]) |
| for i in range(input_data.shape[0]): |
| c = input_data[i, 0] + 0.1j * np.std(input_data) |
| z = 0 |
| for n in range(max_iter): |
| if abs(z) > 2: |
| output[i] = n |
| break |
| z = z*z + c |
| else: |
| output[i] = max_iter |
| return output |
|
|
| def julia_fractal(input_data, max_iter=10): |
| output = np.zeros(input_data.shape[0]) |
| c = -0.8 + 0.156j |
| for i in range(input_data.shape[0]): |
| z = input_data[i, 0] + 0.1j * np.std(input_data) |
| for n in range(max_iter): |
| if abs(z) > 2: |
| output[i] = n |
| break |
| z = z*z + c |
| else: |
| output[i] = max_iter |
| return output |
|
|
| def run_snn(epochs, batch_size, l2_lambda, patience): |
| np.random.seed(42) |
|
|
| X = np.linspace(0, 10, 1000).reshape(-1, 1) |
| y = np.sin(X).reshape(-1, 1) |
|
|
| X = (X - X.min()) / (X.max() - X.min()) |
| y = (y - y.min()) / (y.max() - y.min()) |
|
|
| snn = SwarmNeuralNetwork(layer_sizes=[1, 32, 16, 8, 1], |
| fractal_methods=[sierpinski_fractal, mandelbrot_fractal, julia_fractal, julia_fractal]) |
|
|
| snn.train(X, y, epochs=epochs, batch_size=batch_size, l2_lambda=l2_lambda, patience=patience) |
|
|
| y_pred = snn.forward(X, training=False) |
| fractal_outputs = snn.apply_fractals(X) |
|
|
| fig, axs = plt.subplots(2, 2, figsize=(15, 10)) |
| |
| axs[0, 0].plot(X, y, label='True') |
| axs[0, 0].plot(X, y_pred, label='Predicted') |
| axs[0, 0].legend() |
| axs[0, 0].set_title('True vs Predicted') |
| |
| x, y = fractal_outputs[0] |
| axs[0, 1].plot(x, y) |
| axs[0, 1].set_title('Sierpinski Fractal Output') |
| |
| axs[1, 0].plot(X, fractal_outputs[1]) |
| axs[1, 0].set_title('Mandelbrot Fractal Output') |
| |
| axs[1, 1].plot(X, fractal_outputs[2]) |
| axs[1, 1].set_title('Julia Fractal Output') |
|
|
| plt.tight_layout() |
| return fig |
|
|
| with gr.Blocks() as demo: |
| epochs = gr.Slider(1, 10000, value=5000, label="Epochs") |
| batch_size = gr.Slider(1, 100, value=32, label="Batch Size") |
| l2_lambda = gr.Slider(0.0001, 0.1, value=0.00001, label="L2 Lambda") |
| patience = gr.Slider(1, 1000, value=50, label="Patience") |
|
|
| plot = gr.Plot() |
|
|
| def update_plot(epochs, batch_size, l2_lambda, patience): |
| return run_snn(epochs, batch_size, l2_lambda, patience) |
|
|
| btn = gr.Button("Run SNN") |
| btn.click(update_plot, inputs=[epochs, batch_size, l2_lambda, patience], outputs=plot) |
|
|
| demo.launch() |
|
|
