diff --git a/layer_test.ipynb b/layer_test.ipynb index 6c54d97..f6b2caf 100644 --- a/layer_test.ipynb +++ b/layer_test.ipynb @@ -213,8 +213,8 @@ "# Test with test data\n", "test_batch = Mat([[0,0,0], [0,1,0], [1,0,0], [1,1,0]], dtype=np.float64)\n", "for pattern in test_batch:\n", - "\th = layer.entity.gibbs_v_to_h(pattern)\n", - "\tv = layer.entity.gibbs_h_to_v(h)\n", + "\th = layer.entity.forward(pattern)\n", + "\tv = layer.entity.reconstruct(h)\n", "\tprint(f\"P{pattern} : {v}\")" ] }, diff --git a/src/rbm/entity.py b/src/rbm/entity.py index fb10c3f..a57feb7 100644 --- a/src/rbm/entity.py +++ b/src/rbm/entity.py @@ -1,7 +1,25 @@ -from .params import EntityParams from .state import RbmState from .matrix import prob, Mat +class EntityParams: + def __init__(self): + # Entity parameters + self.num_gibbs_samples = 1 + self.do_gaussian_visible = False + self.do_gaussian_hidden = False + + @classmethod + def from_dict(cls, params: dict): + obj = EntityParams() + if "numGibbs" in params: + obj.num_gibbs_samples = params["numGibbs"] + if "doGaussianVisible" in params: + obj.do_gaussian_visible = params["doGaussianVisible"] + if "doGaussianHidden" in params: + obj.do_gaussian_hidden = params["doGaussianHidden"] + + return obj + class Entity: def __init__(self, shape: tuple[int, int], params: EntityParams): self.shape = shape @@ -22,7 +40,7 @@ class Entity: return prob(state) - def gibbs_v_to_h(self, v: Mat) -> Mat: + def forward(self, v: Mat) -> Mat: h = self.v_to_ph(v) for i in range(self.params.num_gibbs_samples-1): h = self.h_to_pv(h) @@ -30,7 +48,7 @@ class Entity: return h - def gibbs_h_to_v(self, h: Mat) -> Mat: + def reconstruct(self, h: Mat) -> Mat: v = self.h_to_pv(h) for i in range(self.params.num_gibbs_samples-1): v = self.v_to_ph(v) diff --git a/src/rbm/layer.py b/src/rbm/layer.py index 7f65f97..327b26a 100644 --- a/src/rbm/layer.py +++ b/src/rbm/layer.py @@ -1,15 +1,13 @@ -from .params import EntityParams from .state import RbmState -from .status import Status -from .train import train -from .entity import Entity -from .matrix import Mat, np +from .train import TrainingParams +from .entity import Entity, EntityParams class Layer: - def __init__(self, name: str, shape: tuple[int, int, int, int], params: EntityParams): + def __init__(self, name: str, shape: tuple[int, int, int, int], entity_params: EntityParams, training_params: TrainingParams): self.name = name self.shape = shape - self.entity = Entity((shape[0]*shape[1]+shape[2], shape[3]), params) + self.entity = Entity((shape[0]*shape[1]+shape[2], shape[3]), entity_params) + self.training_params = training_params def init(self, std: float): self.entity.state.init(mu=0, std=std) diff --git a/src/rbm/params.py b/src/rbm/params.py deleted file mode 100644 index 7525947..0000000 --- a/src/rbm/params.py +++ /dev/null @@ -1,74 +0,0 @@ -import json - -class Params: - def save(self, filename: str): - with open(filename, "w") as fp: - json.dump(self.__dict__, fp, indent=4) - - def load(self, filename: str): - with open(filename, "r") as fp: - self.__dict__ = json.load(fp) - -class EntityParams(Params): - def __init__(self): - # Training parameters - self.learning_rate = 0.1 - self.momentum = 0.5 - self.weight_decay = 0 - self.num_epochs = 1000 - self.mini_batch_size = 0 - self.do_rao_blackwell = False - self.do_gibbs_sample_visible = False - self.do_gibbs_sample_hidden = False - self.do_batch_sample = False - - # Entity parameters - self.num_gibbs_samples = 1 - self.do_gaussian_visible = False - self.do_gaussian_hidden = False - - - @classmethod - def from_dict(cls, params: dict, version: str = '0'): - obj = EntityParams() - - if "0" in version or "1" in version: - # Training parameters - obj.learning_rate = params["learningRate"] - obj.momentum = params["momentum"] - obj.weight_decay = params["weightDecay"] - obj.num_epochs = params["numEpochs"] - obj.mini_batch_size = params["miniBatchSize"] - obj.do_rao_blackwell = params["doRaoBlackwell"] - obj.do_gibbs_sample_visible = params["gibbsDoSampleVisible"] - obj.do_gibbs_sample_hidden = params["gibbsDoSampleHidden"] - obj.do_batch_sample = params["doSampleBatch"] - - # Entity parameters - obj.num_gibbs_samples = params["numGibbs"] - - if "1" in version: - # Entity parameters - obj.do_gaussian_visible = params["doGaussianVisible"] - obj.do_gaussian_hidden = params["doGaussianHidden"] - - return obj - -class LayerParams(Params): - def __init__(self, num_visible, num_hidden): - self.num_visible = num_visible - self.num_hidden = num_hidden - - -if __name__ == "__main__": - test_filename = "../../test_params.json" - p1 = EntityParams() - p1.num_epochs = 31101970 - p1.save(test_filename) - - p2 = EntityParams() - p2.load(test_filename) - - assert p2.__dict__ == p1.__dict__ - - print("Test: [passed]") diff --git a/src/rbm/stack_deep.py b/src/rbm/stack_deep.py index 8ac84e0..2cd7476 100644 --- a/src/rbm/stack_deep.py +++ b/src/rbm/stack_deep.py @@ -19,21 +19,21 @@ class StackDeep(Stack): def train(self, batch: Mat, status=Status()): _batch = np.copy(batch) for index, layer in enumerate(self.layers): - print(f"Train layer {index} for {layer.entity.params.num_epochs} epochs") + print(f"Train layer {index} for {layer.training_params.num_epochs} epochs") _batch = self.batch_from(batch, index) - train(layer.entity, _batch, status=status) + train(layer.entity, _batch, layer.training_params, status=status) def pass_up(self, visible: Mat, from_layer_id: int = 0): h = np.copy(visible) for layer in self.layers[from_layer_id:]: - h = layer.entity.gibbs_v_to_h(h) + h = layer.entity.forward(h) return h def pass_down(self, hidden: Mat, from_layer_id: int = 0): v = np.copy(hidden) for layer in list(reversed(self.layers))[from_layer_id:]: - v = layer.entity.gibbs_h_to_v(v) + v = layer.entity.reconstruct(v) return v def pass_down_up(self, visible: Mat, from_layer_id: int = 0): diff --git a/src/rbm/stack_factory.py b/src/rbm/stack_factory.py index ebda9d9..4b7a81b 100644 --- a/src/rbm/stack_factory.py +++ b/src/rbm/stack_factory.py @@ -1,14 +1,14 @@ import json -from collections.abc import Callable from .stack import StackType from .layer import Layer -from .params import EntityParams +from .entity import EntityParams +from .train import TrainingParams from .stack_deep import StackDeep from .stack_rnn import StackRnn class StackFactory: @classmethod - def from_dict(cls, project: dict, work_dir: str = ".", layer_constructor: Callable = None) -> StackDeep|StackRnn: + def from_dict(cls, project: dict, work_dir: str = ".") -> StackDeep|StackRnn: name = project["stack"]["name"] layers = project["stack"]["layers"] @@ -38,15 +38,12 @@ class StackFactory: num_context = 0 # Determine version by existence of keys - layer_params = layer["rbm"]["params"] - params_version = '0' - if "doGaussianHidden" in layer_params and "doGaussianVisible" in layer_params: - params_version = '1' - - params = EntityParams.from_dict(layer_params, params_version) + params = layer["rbm"]["params"] + entity_params = EntityParams.from_dict(params) + training_params = TrainingParams.from_dict(params) # Create layer - layer_obj = Layer(f"{layer_name}-{layer_id}", (num_visible_x, num_visible_y, num_context, num_hidden), params) + layer_obj = Layer(f"{layer_name}-{layer_id}", (num_visible_x, num_visible_y, num_context, num_hidden), entity_params, training_params) # Add layer to stack obj.append(layer_obj) @@ -54,7 +51,7 @@ class StackFactory: return obj @classmethod - def from_file(cls, filename: str, work_dir: str = ".", layer_constructor: Callable = None) -> StackDeep|StackRnn: + def from_file(cls, filename: str, work_dir: str = ".") -> StackDeep|StackRnn: obj = None with open(filename, "r") as fp: prj = json.load(fp) diff --git a/src/rbm/train.py b/src/rbm/train.py index ba52865..c2dac48 100644 --- a/src/rbm/train.py +++ b/src/rbm/train.py @@ -3,7 +3,35 @@ from .matrix import gaussian, sample, prob, rms_error_accu, Mat, np from .entity import Entity from .status import Status -def cd_jens(entity: Entity, v_states: Mat): +class TrainingParams: + def __init__(self): + # Training parameters + self.learning_rate = 0.1 + self.momentum = 0.5 + self.weight_decay = 0 + self.num_epochs = 1000 + self.mini_batch_size = 0 + self.do_rao_blackwell = False + self.do_gibbs_sample_visible = False + self.do_gibbs_sample_hidden = False + self.do_batch_sample = False + + @classmethod + def from_dict(cls, params: dict): + obj = TrainingParams() + obj.learning_rate = params["learningRate"] + obj.momentum = params["momentum"] + obj.weight_decay = params["weightDecay"] + obj.num_epochs = params["numEpochs"] + obj.mini_batch_size = params["miniBatchSize"] + obj.do_rao_blackwell = params["doRaoBlackwell"] + obj.do_gibbs_sample_visible = params["gibbsDoSampleVisible"] + obj.do_gibbs_sample_hidden = params["gibbsDoSampleHidden"] + obj.do_batch_sample = params["doSampleBatch"] + + return obj + +def cd_jens(entity: Entity, v_states: Mat, params: TrainingParams): v_probs = prob(v_states) h_states = entity.v_to_ph(v_states) h_probs = h_states @@ -12,7 +40,7 @@ def cd_jens(entity: Entity, v_states: Mat): h_states += gaussian(h_states.shape) else: h_probs = entity.v_to_ph(v_states) - if entity.params.do_rao_blackwell: + if params.do_rao_blackwell: h_states = h_probs else: h_states = sample(h_probs) @@ -24,13 +52,13 @@ def cd_jens(entity: Entity, v_states: Mat): # Gibbs sampling with training params for i in range(entity.params.num_gibbs_samples): - if entity.params.do_gibbs_sample_hidden: + if params.do_gibbs_sample_hidden: v_probs = entity.h_to_pv(sample(h_probs)) else: v_probs = entity.h_to_pv(h_probs) # Create hidden representation given v - if entity.params.do_gibbs_sample_visible: + if params.do_gibbs_sample_visible: h_probs = entity.v_to_ph(sample(v_probs)) else: h_probs = entity.v_to_ph(v_probs) @@ -42,14 +70,14 @@ def cd_jens(entity: Entity, v_states: Mat): return dw, dbv, dbh -def train(entity: Entity, batch: Mat, status: Status, cd_func: Callable = cd_jens): +def train(entity: Entity, batch: Mat, params: TrainingParams, status: Status, cd_func: Callable = cd_jens): training_remain = batch.shape[0] - batch_size = min(entity.params.mini_batch_size, training_remain) + batch_size = min(params.mini_batch_size, training_remain) if batch_size == 0: batch_size = training_remain status.on_change() - d_progress = 100.0 / (training_remain*entity.params.num_epochs) + d_progress = 100.0 / (training_remain*params.num_epochs) progress = 0 batch_row_index = 0 keep_running = True @@ -64,18 +92,18 @@ def train(entity: Entity, batch: Mat, status: Status, cd_func: Callable = cd_jen inc_whv = np.zeros(entity.state.w_hv.shape) v_states = mini_batch - if entity.params.do_batch_sample: + if params.do_batch_sample: v_states = sample(mini_batch) - for epochs in range(entity.params.num_epochs): + for epochs in range(params.num_epochs): # Contrastive divergence learning: calculate gradients - dwhv, dbv, dbh = cd_func(entity, v_states) + dwhv, dbv, dbh = cd_func(entity, v_states, params) # Adjust weight and biases - kl = entity.params.learning_rate/batch_size - inc_bv = entity.params.momentum*inc_bv + kl*dbv - inc_bh = entity.params.momentum*inc_bh + kl*dbh - inc_whv = entity.params.momentum*inc_whv + kl*dwhv - entity.params.weight_decay*entity.state.w_hv + kl = params.learning_rate/batch_size + inc_bv = params.momentum*inc_bv + kl*dbv + inc_bh = params.momentum*inc_bh + kl*dbh + inc_whv = params.momentum*inc_whv + kl*dwhv - params.weight_decay*entity.state.w_hv entity.state.b_v += inc_bv entity.state.b_h += inc_bh diff --git a/src/tests/test_xor.py b/src/tests/test_xor.py index 248aa68..2861dd3 100644 --- a/src/tests/test_xor.py +++ b/src/tests/test_xor.py @@ -1,20 +1,21 @@ import os.path -from rbm.params import EntityParams from rbm.layer import Layer from rbm.status import Status -from rbm.train import train +from rbm.train import train, TrainingParams from rbm.matrix import Mat, np +from rbm.entity import EntityParams work_dir = "../../results" def xor(): # Create params - params = EntityParams() - params.do_rao_blackwell = True - params.num_gibbs_samples = 3 + entity_params = EntityParams() + training_params = TrainingParams() + entity_params.do_rao_blackwell = True + entity_params.num_gibbs_samples = 3 # Create layer - layer = Layer("Layer_0", (3, 1, 0, 16), params) + layer = Layer("Layer_0", (3, 1, 0, 16), entity_params, training_params) # Init weights layer.init(0.01) @@ -26,7 +27,7 @@ def xor(): training_batch = Mat([[0,1,1], [0,0,0], [1,1,0], [1,0,1]], dtype=np.float64) # Train layer - train(layer.entity, training_batch, Status()) + train(layer.entity, training_batch, training_params, Status()) # Save weights layer.save(os.path.join(work_dir, "xor_layer0_state.npz")) @@ -34,8 +35,8 @@ def xor(): # Test with test data test_batch = Mat([[0,0,0], [0,1,0], [1,0,0], [1,1,0]], dtype=np.float64) for pattern in test_batch: - h = layer.entity.gibbs_v_to_h(pattern) - v = layer.entity.gibbs_h_to_v(h) + h = layer.entity.forward(pattern) + v = layer.entity.reconstruct(h) print(f"P{pattern} : {v}") if __name__ == "__main__":