steffen/self-replicating-neural-networks
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

robustness cleaned

Browse Source
This commit is contained in:
steffen-illium
2021-06-04 14:03:03 +02:00
parent c9efe0a31b
commit 9abde030af
1 changed files with 74 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

126
journal_robustness.py
View File

@@ -4,7 +4,6 @@ import pandas as pd
import torch
import random
import copy
import numpy as np
from pathlib import Path
from tqdm import tqdm
@@ -21,6 +20,7 @@ from matplotlib import pyplot as plt
def prng():
return random.random()
def generate_perfekt_synthetic_fixpoint_weights():
return torch.tensor([[1.0], [0.0], [0.0], [0.0], [0.0], [0.0], [0.0], [0.0],
[1.0], [0.0], [0.0], [0.0],
@@ -28,15 +28,32 @@ def generate_perfekt_synthetic_fixpoint_weights():
], dtype=torch.float32)
PALETTE = 10 * (
"#377eb8",
"#4daf4a",
"#984ea3",
"#e41a1c",
"#ff7f00",
"#a65628",
"#f781bf",
"#888888",
"#a6cee3",
"#b2df8a",
"#cab2d6",
"#fb9a99",
"#fdbf6f",
)
class RobustnessComparisonExperiment:
@staticmethod
def apply_noise(network, noise: int):
""" Changing the weights of a network to values + noise """
# Changing the weights of a network to values + noise
for layer_id, layer_name in enumerate(network.state_dict()):
for line_id, line_values in enumerate(network.state_dict()[layer_name]):
for weight_id, weight_value in enumerate(network.state_dict()[layer_name][line_id]):
#network.state_dict()[layer_name][line_id][weight_id] = weight_value + noise
# network.state_dict()[layer_name][line_id][weight_id] = weight_value + noise
if prng() < 0.5:
network.state_dict()[layer_name][line_id][weight_id] = weight_value + noise
else:
@@ -55,7 +72,7 @@ class RobustnessComparisonExperiment:
self.epochs = epochs
self.ST_steps = st_steps
self.loss_history = []
self.synthetic = synthetic
self.is_synthetic = synthetic
self.fixpoint_counters = {
"identity_func": 0,
"divergent": 0,
@@ -71,14 +88,14 @@ class RobustnessComparisonExperiment:
self.id_functions = []
self.nets = self.populate_environment()
self.count_fixpoints()
self.time_to_vergence, self.time_as_fixpoint = self.test_robustness()
self.time_to_vergence, self.time_as_fixpoint = self.test_robustness(
seeds=population_size if self.is_synthetic else 1)
self.save()
def populate_environment(self):
loop_population_size = tqdm(range(self.population_size))
nets = []
if self.synthetic:
if self.is_synthetic:
''' Either use perfect / hand-constructed fixpoint ... '''
net_name = f"net_{str(0)}_synthetic"
net = Net(self.net_input_size, self.net_hidden_size, self.net_out_size, net_name)
@@ -86,6 +103,7 @@ class RobustnessComparisonExperiment:
nets.append(net)
else:
loop_population_size = tqdm(range(self.population_size))
for i in loop_population_size:
loop_population_size.set_description("Populating experiment %s" % i)
@@ -99,58 +117,61 @@ class RobustnessComparisonExperiment:
def test_robustness(self, print_it=True, noise_levels=10, seeds=10):
assert (len(self.id_functions) == 1 and seeds > 1) or (len(self.id_functions) > 1 and seeds == 1)
is_synthetic = True if len(self.id_functions) > 1 and seeds == 1 else False
avg_time_to_vergence = [[0 for _ in range(noise_levels)] for _ in
range(seeds if is_synthetic else len(self.id_functions))]
avg_time_as_fixpoint = [[0 for _ in range(noise_levels)] for _ in
range(seeds if is_synthetic else len(self.id_functions))]
time_to_vergence = [[0 for _ in range(noise_levels)] for _ in
range(seeds if self.is_synthetic else len(self.id_functions))]
time_as_fixpoint = [[0 for _ in range(noise_levels)] for _ in
range(seeds if self.is_synthetic else len(self.id_functions))]
row_headers = []
data_pos = 0
# This checks wether to use synthetic setting with multiple seeds
# or multi network settings with a singlee seed
df = pd.DataFrame(columns=['setting', 'noise_level', 'application_step', 'absolute_loss', 'time_to_vergence'])
for i, fixpoint in enumerate(self.id_functions): #1 / n
row_headers.append(fixpoint.name)
for seed in range(seeds): #n / 1
for noise_level in range(noise_levels):
self_application_steps = 0
clone = Net(fixpoint.input_size, fixpoint.hidden_size, fixpoint.out_size,
f"{fixpoint.name}_clone_noise10e-{noise_level}")
clone.load_state_dict(copy.deepcopy(fixpoint.state_dict()))
rand_noise = prng() * pow(10, -noise_level) #n / 1
clone = self.apply_noise(clone, rand_noise)
df = pd.DataFrame(columns=['setting', 'noise_level', 'steps', 'absolute_loss', 'time_to_vergence', 'time_as_fixpoint'])
with tqdm(total=max(len(self.id_functions), seeds)) as pbar:
for i, fixpoint in enumerate(self.id_functions): # 1 / n
row_headers.append(fixpoint.name)
for seed in range(seeds): # n / 1
for noise_level in range(noise_levels):
steps = 0
clone = Net(fixpoint.input_size, fixpoint.hidden_size, fixpoint.out_size,
f"{fixpoint.name}_clone_noise10e-{noise_level}")
clone.load_state_dict(copy.deepcopy(fixpoint.state_dict()))
rand_noise = prng() * pow(10, -noise_level) # n / 1
clone = self.apply_noise(clone, rand_noise)
while not is_zero_fixpoint(clone) and not is_divergent(clone):
# -> before
clone_weight_pre_application = clone.input_weight_matrix()
target_data_pre_application = clone.create_target_weights(clone_weight_pre_application)
while not is_zero_fixpoint(clone) and not is_divergent(clone):
# -> before
clone_weight_pre_application = clone.input_weight_matrix()
target_data_pre_application = clone.create_target_weights(clone_weight_pre_application)
clone.self_application(1, self.log_step_size)
avg_time_to_vergence[i][noise_level] += 1
# -> after
clone_weight_post_application = clone.input_weight_matrix()
target_data_post_application = clone.create_target_weights(clone_weight_post_application)
clone.self_application(1, self.log_step_size)
time_to_vergence[i][noise_level] += 1
# -> after
clone_weight_post_application = clone.input_weight_matrix()
target_data_post_application = clone.create_target_weights(clone_weight_post_application)
absolute_loss = F.l1_loss(target_data_pre_application, target_data_post_application).item()
absolute_loss = F.l1_loss(target_data_pre_application, target_data_post_application).item()
setting = i if is_synthetic else seed
setting = seed if self.is_synthetic else i
if is_identity_function(clone):
avg_time_as_fixpoint[i][noise_level] += 1
# When this raises a Type Error, we found a second order fixpoint!
self_application_steps += 1
else:
self_application_steps = pd.NA # Not a Number!
if is_identity_function(clone):
time_as_fixpoint[i][noise_level] += 1
# When this raises a Type Error, we found a second order fixpoint!
steps += 1
df.loc[df.shape[0]] = [setting, noise_level, self_application_steps,
absolute_loss, avg_time_to_vergence[i][noise_level]]
df.loc[df.shape[0]] = [setting, noise_level, steps, absolute_loss,
time_to_vergence[i][noise_level], time_as_fixpoint[i][noise_level]]
pbar.update(1)
# calculate the average:
# df = df.replace([np.inf, -np.inf], np.nan)
# df = df.dropna()
bf = sns.boxplot(data=df, y='self_application_steps', x='noise_level', )
# Get the measuremts at the highest time_time_to_vergence
df_sorted = df.sort_values('steps', ascending=False).drop_duplicates(['setting', 'noise_level'])
df_melted = df_sorted.reset_index().melt(id_vars=['setting', 'noise_level', 'steps'],
value_vars=['time_to_vergence', 'time_as_fixpoint'],
var_name="Measurement",
value_name="Steps")
# Plotting
sns.set(style='whitegrid')
bf = sns.boxplot(data=df_melted, y='Steps', x='noise_level', hue='Measurement', palette=PALETTE)
bf.set_title('Robustness as self application steps per noise level')
plt.tight_layout()
@@ -158,6 +179,7 @@ class RobustnessComparisonExperiment:
# bx = sns.catplot(data=df[df['absolute_loss'] < 1], y='absolute_loss', x='application_step', kind='box',
# col='noise_level', col_wrap=3, showfliers=False)
directory = Path('output') / 'robustness'
directory.mkdir(parents=True, exist_ok=True)
filename = f"absolute_loss_perapplication_boxplot_grid.png"
filepath = directory / filename
@@ -167,12 +189,12 @@ class RobustnessComparisonExperiment:
col_headers = [str(f"10e-{d}") for d in range(noise_levels)]
print(f"\nAppplications steps until divergence / zero: ")
print(tabulate(avg_time_to_vergence, showindex=row_headers, headers=col_headers, tablefmt='orgtbl'))
print(tabulate(time_to_vergence, showindex=row_headers, headers=col_headers, tablefmt='orgtbl'))
print(f"\nTime as fixpoint: ")
print(tabulate(avg_time_as_fixpoint, showindex=row_headers, headers=col_headers, tablefmt='orgtbl'))
print(tabulate(time_as_fixpoint, showindex=row_headers, headers=col_headers, tablefmt='orgtbl'))
return avg_time_as_fixpoint, avg_time_to_vergence
return time_as_fixpoint, time_to_vergence
def count_fixpoints(self):
exp_details = f"ST steps: {self.ST_steps}"
@@ -198,7 +220,7 @@ if __name__ == "__main__":
ST_steps = 1000
ST_epochs = 5
ST_log_step_size = 10
ST_population_size = 5
ST_population_size = 100
ST_net_hidden_size = 2
ST_net_learning_rate = 0.04
ST_name_hash = random.getrandbits(32)