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

[StackRnn] - add unrolled (own-weights) mode; each position gets its own RBM

Browse Source 
Previously a single shared-weight Entity processed every time step.  Now:
- Shared mode  (1 layer via make_layer):  original behaviour unchanged
- Unrolled mode (N layers via make_unrolled): layers[t] owns W_t, b_v_t, b_h_t

New API:
  make_unrolled(T, sensory_size, h_size, ...)  → list[Layer]
  next_entity()   → Entity for the upcoming step() call
  current_entity() → Entity from the most recent step() call
  is_shared        → bool

moby_rnn.ipynb: switch Build model cell to make_unrolled(T=100); update
  predict_next() to use rnn.next_entity() for position-correct Gibbs sampling
README_moby_rnn.md: redraw temporal-unrolling ASCII art showing per-position
  weights W_t; update parameter count and checkpoint file listing

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
jens
2026-05-31 11:52:01 +02:00
co-authored by Claude Sonnet 4.6
parent 0b302a9c12
commit 3698c85ed4
4 changed files with 229 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/tests/test_rnn.py
+51 -64
View File
@@ -1,8 +1,4 @@
"""Test for StackRnn: recurrent RBM with concatenated [h_{t-1} | x_t] visible layer.
Uses a repeating binary pattern as a minimal synthetic sequence so the model
has something learnable to compress and predict.
"""
"""Tests for StackRnn — shared-weights and unrolled (own-weights) modes."""
import numpy as _np_cpu
from rbm.stack_rnn import StackRnn
from rbm.matrix import np, convert
@@ -10,94 +6,85 @@ from rbm.entity import EntityParams, TrainingParams
from rbm.status import Status
SENSORY_SIZE = 8
H_SIZE = 4
T = 16 # sequence length
NUM_SEQ = 10 # sequences in the batch
WORK_DIR = "../../results"
H_SIZE = 4
T = 16 # sequence length
NUM_SEQ = 10 # sequences in the batch
WORK_DIR = "../../results"
_PARAMS = TrainingParams(learning_rate=0.05, momentum=0.5, num_epochs=5,
do_rao_blackwell=True)
def _make_sequences() -> _np_cpu.ndarray:
"""Binary sequences: each row is one sequence of T frames."""
rng = _np_cpu.random.RandomState(0)
rng = _np_cpu.random.RandomState(0)
base = (rng.rand(NUM_SEQ, SENSORY_SIZE) > 0.5).astype(_np_cpu.float64)
seqs = _np_cpu.stack([base] * T, axis=1) # (NUM_SEQ, T, SENSORY_SIZE)
return seqs
return _np_cpu.stack([base] * T, axis=1) # (NUM_SEQ, T, SENSORY_SIZE)
def test_rnn_single_layer():
def test_rnn_shared():
"""Shared-weights mode: one entity reused at every time step."""
seqs = _make_sequences()
rnn = StackRnn("test_rnn", WORK_DIR)
layer = StackRnn.make_layer(
"layer0", SENSORY_SIZE, H_SIZE,
EntityParams(do_gaussian_visible=False, do_gaussian_hidden=False),
TrainingParams(learning_rate=0.05, momentum=0.5, num_epochs=20,
mini_batch_size=0, do_rao_blackwell=True),
)
rnn.append(layer)
rnn = StackRnn("test_rnn_shared", WORK_DIR)
rnn.append(StackRnn.make_layer("layer0", SENSORY_SIZE, H_SIZE,
EntityParams(), _PARAMS))
rnn.state_init(0.01)
# Confirm entity shape
assert rnn.sensory_size() == SENSORY_SIZE, "sensory_size mismatch"
assert rnn.h_size() == H_SIZE, "h_size mismatch"
assert rnn.is_shared
assert rnn.sensory_size() == SENSORY_SIZE
assert rnn.h_size() == H_SIZE
rnn.train(seqs)
# Inference: step through one sequence
rnn.reset(batch_size=1)
seq0 = seqs[0] # (T, SENSORY_SIZE) numpy
for t in range(T):
x_t = np.array(seq0[t][None, :]) # (1, SENSORY_SIZE) on device
h = rnn.step(x_t)
assert h.shape == (1, H_SIZE), f"step output shape wrong at t={t}"
h = rnn.step(np.array(seqs[0, t][None, :]))
assert h.shape == (1, H_SIZE), f"bad shape at t={t}"
# Reconstruction from final hidden state
recon = rnn.reconstruct(h)
assert recon.shape == (1, SENSORY_SIZE), "reconstruct shape wrong"
assert recon.shape == (1, SENSORY_SIZE)
print(f"Original x_T : {convert(np.array(seq0[-1][None, :]))}")
print(f"Reconstructed: {convert(recon)}")
print("test_rnn_single_layer: [passed]")
print(f"Original : {convert(np.array(seqs[0, -1][None, :]))}")
print(f"Recon : {convert(recon)}")
print("test_rnn_shared: [passed]")
def test_rnn_two_layers():
"""Two-layer recurrent stack: layer 1 receives h_0 as its sensory input."""
H_SIZE_0, H_SIZE_1 = 6, 3
def test_rnn_unrolled():
"""Unrolled mode: T entities, one per time step, each with own weights."""
seqs = _make_sequences()
rnn = StackRnn("test_rnn2", WORK_DIR)
rnn.append(StackRnn.make_layer(
"layer0", SENSORY_SIZE, H_SIZE_0,
EntityParams(),
TrainingParams(learning_rate=0.05, num_epochs=10),
))
rnn.append(StackRnn.make_layer(
"layer1", H_SIZE_0, H_SIZE_1,
EntityParams(),
TrainingParams(learning_rate=0.05, num_epochs=10),
))
rnn = StackRnn("test_rnn_unrolled", WORK_DIR)
for layer in StackRnn.make_unrolled(T, SENSORY_SIZE, H_SIZE,
EntityParams(), _PARAMS):
rnn.append(layer)
rnn.state_init(0.01)
assert not rnn.is_shared
assert rnn.num_layers() == T
rnn.train(seqs)
# Inference — _t advances through all T entities
rnn.reset(batch_size=1)
for t in range(T):
x_t = np.array(seqs[0, t][None, :])
h = rnn.step(x_t)
h = rnn.step(np.array(seqs[0, t][None, :]))
assert h.shape == (1, H_SIZE), f"bad shape at t={t}"
assert rnn._t == t + 1
assert h.shape == (1, H_SIZE_1)
print("test_rnn_two_layers: [passed]")
recon = rnn.reconstruct(h)
assert recon.shape == (1, SENSORY_SIZE)
# next_entity wraps around modularly
rnn._t = T + 3
assert rnn.next_entity() is rnn.from_index(3).entity
print("test_rnn_unrolled: [passed]")
def test_rnn_save_load():
seqs = _make_sequences()
rnn = StackRnn("test_rnn_sl", WORK_DIR)
rnn.append(StackRnn.make_layer(
"layer0", SENSORY_SIZE, H_SIZE,
EntityParams(),
TrainingParams(num_epochs=5),
))
rnn = StackRnn("test_rnn_sl", WORK_DIR)
for layer in StackRnn.make_unrolled(T, SENSORY_SIZE, H_SIZE,
EntityParams(), _PARAMS):
rnn.append(layer)
rnn.state_init(0.01)
rnn.train(seqs)
rnn.state_save()
@@ -106,7 +93,7 @@ def test_rnn_save_load():
if __name__ == "__main__":
test_rnn_single_layer()
test_rnn_two_layers()
test_rnn_shared()
test_rnn_unrolled()
test_rnn_save_load()
print("All RNN tests passed.")