jens/pyRBM
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactored

Browse Source
This commit is contained in:
jens
2025-12-19 15:20:04 +01:00
parent a293fc31a0
commit a47922cb1c
8 changed files with 92 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files
layer_test.ipynb
+2 -2
View File
@@ -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}\")"
]
},
src/rbm/entity.py
+21 -3
View File
@@ -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)
src/rbm/layer.py
+5 -7
View File
@@ -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)
src/rbm/params.py
-74
View File
@@ -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]")
src/rbm/stack_deep.py
+4 -4
View File
@@ -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):
src/rbm/stack_factory.py
+8 -11
View File
@@ -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)
src/rbm/train.py
+42 -14
View File
@@ -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
src/tests/test_xor.py
+10 -9
View File
@@ -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__":