From dc998d30a21f3a63ed4e7ca0030d76f35e9c1545 Mon Sep 17 00:00:00 2001 From: Jens Ahrensfeld Date: Thu, 1 Jan 2026 12:05:39 +0100 Subject: [PATCH] - initroduced types: BB-RBM GB-RBM, GG-RBM as property of entity - training simplification and for clarity: added specialized cd_funcs for each entity type --- src/rbm/entity.py | 30 +++++++++++++- src/rbm/layout.py | 2 +- src/rbm/matrix.py | 2 +- src/rbm/model.py | 2 +- src/rbm/train.py | 89 +++++++++++++++++++++++++++++++++++++++-- src/tests/test_norbs.py | 2 +- 6 files changed, 119 insertions(+), 8 deletions(-) diff --git a/src/rbm/entity.py b/src/rbm/entity.py index 77e5fb2..8031121 100644 --- a/src/rbm/entity.py +++ b/src/rbm/entity.py @@ -1,6 +1,6 @@ from .state import RbmState from .matrix import prob, Mat - +from enum import Enum class EntityParams: def __init__(self, do_gaussian_visible: bool = False, do_gaussian_hidden: bool = False, num_gibbs_samples: int = 1): # Entity parameters @@ -21,11 +21,29 @@ class EntityParams: return obj class Entity: + class Type(Enum): + BB_RBM = "BB-RBM" + BG_RBM = "BG-RBM" + GB_RBM = "GB-RBM" + GG_RBM = "GG-RBM" + def __init__(self, shape: tuple[int, int], params: EntityParams): self.shape = shape self.params = params self.state = RbmState.from_layer_params(shape) self.grad = RbmState.from_layer_params(shape) + self.type = None + if params.do_gaussian_visible: + if params.do_gaussian_hidden: + self.type = Entity.Type.GG_RBM + else: + self.type = Entity.Type.GB_RBM + else: + if params.do_gaussian_hidden: + self.type = Entity.Type.BG_RBM + else: + self.type = Entity.Type.BB_RBM + def __call__(self, x: Mat): return self.forward(x) @@ -72,6 +90,7 @@ class Entity: return prob(state) + def _h_to_pv(self, h: Mat) -> Mat: state = self.state.h_to_v(h) if self.params.do_gaussian_visible: @@ -79,5 +98,14 @@ class Entity: return prob(state) + def h_given_v(self, v: Mat) -> Mat: + state = self.state.v_to_h(v) + return state + + + def v_given_h(self, h: Mat) -> Mat: + state = self.state.h_to_v(h) + return state + if __name__ == "__main__": print("Test: [passed]") diff --git a/src/rbm/layout.py b/src/rbm/layout.py index 761ae0a..a29db4b 100644 --- a/src/rbm/layout.py +++ b/src/rbm/layout.py @@ -24,6 +24,6 @@ class Horizontal: def train(self, batch: Mat, params: TrainingParams): _batch = np.copy(batch) for unit in self.units: - train(unit, _batch, params, Status(), cd_jens) + train(unit, _batch, params, Status()) _batch = unit.forward(_batch, num_gibbs=params.num_gibbs_samples) diff --git a/src/rbm/matrix.py b/src/rbm/matrix.py index 19bfc63..915822f 100644 --- a/src/rbm/matrix.py +++ b/src/rbm/matrix.py @@ -27,7 +27,7 @@ def gaussian(shape: tuple, mu: float = 0.0, std: float = 1.0) -> Mat: def sample(src: Mat) -> Mat: return (src > uniform(src.shape)).astype(float) -def gaussian_sample(src: Mat, mu=0, std=1.0) -> Mat: +def sample_gaussian(src: Mat, mu=0, std=1.0) -> Mat: return gaussian(src.shape, mu, std) def prob(src: Mat) -> Mat: diff --git a/src/rbm/model.py b/src/rbm/model.py index 4989765..b4a854f 100644 --- a/src/rbm/model.py +++ b/src/rbm/model.py @@ -26,7 +26,7 @@ class Model(ABC): params = [params]*len(entities) _batch = np.copy(batch) for entity, param in zip(entities, params): - train(entity, _batch, param, Status(), cd_jens) + train(entity, _batch, param, Status()) _batch = entity.forward(_batch, num_gibbs=param.num_gibbs_samples) @abstractmethod diff --git a/src/rbm/train.py b/src/rbm/train.py index a6e5d54..c28702b 100644 --- a/src/rbm/train.py +++ b/src/rbm/train.py @@ -1,5 +1,5 @@ from collections.abc import Callable -from .matrix import gaussian, sample, prob, rms_error_accu, Mat, np +from .matrix import sample, sample_gaussian, prob, rms_error_accu, Mat, np from .entity import Entity from .status import Status @@ -50,7 +50,7 @@ def cd_jens(entity: Entity, v_states: Mat, params: TrainingParams): h_probs = h_states if entity.params.do_gaussian_hidden: - h_states += gaussian(h_states.shape) + h_states += sample_gaussian(h_states) else: h_probs = entity.forward(v_states) if params.do_rao_blackwell: @@ -83,6 +83,81 @@ def cd_jens(entity: Entity, v_states: Mat, params: TrainingParams): return dw, dbv, dbh +def cd_binary_binary(entity: Entity, data_pos: Mat, params: TrainingParams): + # Positive phase + h_probs_pos = entity.h_given_v(data_pos) + + # Update weights (positive phase) + dw = np.dot(np.transpose(data_pos), h_probs_pos) + dbh = np.sum(h_probs_pos, 0) + dbv = np.sum(data_pos, 0) + + # Sample hidden states + if params.do_rao_blackwell: + h_states_pos = h_probs_pos + else: + h_states_pos = sample(h_probs_pos) + + # Negative phase + data_neg = entity.v_given_h(h_states_pos) + h_probs_neg = entity.h_given_v(data_neg) + + # Gibbs sampling with training params + # ToDo + + # Update weights (negative phase) + dw -= np.dot(np.transpose(data_neg), h_probs_neg) + dbh -= np.sum(h_probs_neg, 0) + dbv -= np.sum(data_neg, 0) + + return dw, dbv, dbh + +def cd_gaussian_binary(entity: Entity, data_pos: Mat, params: TrainingParams): + # Positive phase + h_probs_pos = entity.h_given_v(data_pos) + + # Update weights (positive phase) + dw = np.dot(np.transpose(data_pos), h_probs_pos) + dbh = np.sum(h_probs_pos, 0) + dbv = np.sum(data_pos, 0) + + # Sample hidden states + h_states_pos = sample(h_probs_pos) + + # Negative phase + data_neg = entity.v_given_h(h_states_pos) + h_probs_neg = entity.h_given_v(data_neg) + + # Update weights (negative phase) + dw -= np.dot(np.transpose(data_neg), h_probs_neg) + dbh -= np.sum(h_probs_neg, 0) + dbv -= np.sum(data_neg, 0) + + return dw, dbv, dbh + +def cd_gaussian_gaussian(entity: Entity, data_pos: Mat, params: TrainingParams): + # Positive phase + h_probs_pos = entity.h_given_v(data_pos) + + # Sample hidden states + h_states_pos = h_probs_pos + sample_gaussian(h_probs_pos) + + # Update weights (positive phase) + dw = np.dot(np.transpose(data_pos), h_states_pos) + dbh = np.sum(h_states_pos, 0) + dbv = np.sum(data_pos, 0) + + # Negative phase + data_neg = entity.v_given_h(h_states_pos) + h_probs_neg = entity.h_given_v(data_neg) + + # Update weights (negative phase) + dw -= np.dot(np.transpose(data_neg), h_probs_neg) + dbh -= np.sum(h_probs_neg, 0) + dbv -= np.sum(data_neg, 0) + + return dw, dbv, dbh + def to_mini_batch(batch: Mat, mini_batch_size: int): mini_batches = [] remain = batch.shape[0] @@ -95,12 +170,20 @@ def to_mini_batch(batch: Mat, mini_batch_size: int): return mini_batches -def train(entity: Entity, batch: Mat, params: TrainingParams, status: Status, cd_func: Callable = cd_jens): +def train(entity: Entity, batch: Mat, params: TrainingParams, status: Status): mini_batch_size = min(params.mini_batch_size, batch.shape[0]) if params.mini_batch_size > 0 else batch.shape[0] d_progress = 100.0 / (batch.shape[0]*params.num_epochs) progress = 0 keep_running = True status.on_change() + cd_func = None + if entity.type == Entity.Type.GB_RBM: + cd_func = cd_gaussian_binary + if entity.type == Entity.Type.BB_RBM: + cd_func = cd_binary_binary + if entity.type == Entity.Type.GG_RBM: + cd_func = cd_gaussian_gaussian + for mini_batch in to_mini_batch(batch, mini_batch_size): if not keep_running: break diff --git a/src/tests/test_norbs.py b/src/tests/test_norbs.py index d0c5d93..855f647 100644 --- a/src/tests/test_norbs.py +++ b/src/tests/test_norbs.py @@ -40,7 +40,7 @@ if __name__ == "__main__": test_batch = read_armadillo(os.path.join(prj_root, f"{prj_name}.test.dat")) # Train - model.train(train_batch, TrainingParams(learning_rate=0.00001, momentum=0.9, do_rao_blackwell=True, num_epochs=100, num_gibbs_samples=3)) + model.train(train_batch, TrainingParams(learning_rate=0.000001, momentum=0.9, do_rao_blackwell=False, num_epochs=1000, num_gibbs_samples=3)) # save state model.save()