[StackRnn] - add recurrent RBM stack with character-level LM notebook
Implements StackRnn: a cascaded/recurrent RBM where the visible layer at each time step is the concatenation of the previous hidden state (context) and the current sensory input — V[t] = [context | x_t]. Weights are shared across time steps (RTRBM-style concatenation variant). - stack_rnn.py: StackRnn with step(), reconstruct(), reset(), train(), make_layer() factory; supports greedy layer-wise training over sequences of shape (num_seq, T, sensory_size) - test_rnn.py: single-layer, two-layer, and save/load tests - moby_rnn.ipynb: character-level language model on Moby Dick; one-hot encoding, clamped-Gibbs next-char prediction, free text generation, hidden-state trace and character-distribution visualisations Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,112 @@
|
||||
"""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.
|
||||
"""
|
||||
import numpy as _np_cpu
|
||||
from rbm.stack_rnn import StackRnn
|
||||
from rbm.matrix import np, convert
|
||||
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"
|
||||
|
||||
|
||||
def _make_sequences() -> _np_cpu.ndarray:
|
||||
"""Binary sequences: each row is one sequence of T frames."""
|
||||
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
|
||||
|
||||
|
||||
def test_rnn_single_layer():
|
||||
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.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"
|
||||
|
||||
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}"
|
||||
|
||||
# Reconstruction from final hidden state
|
||||
recon = rnn.reconstruct(h)
|
||||
assert recon.shape == (1, SENSORY_SIZE), "reconstruct shape wrong"
|
||||
|
||||
print(f"Original x_T : {convert(np.array(seq0[-1][None, :]))}")
|
||||
print(f"Reconstructed: {convert(recon)}")
|
||||
print("test_rnn_single_layer: [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
|
||||
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.state_init(0.01)
|
||||
|
||||
rnn.train(seqs)
|
||||
|
||||
rnn.reset(batch_size=1)
|
||||
for t in range(T):
|
||||
x_t = np.array(seqs[0, t][None, :])
|
||||
h = rnn.step(x_t)
|
||||
|
||||
assert h.shape == (1, H_SIZE_1)
|
||||
print("test_rnn_two_layers: [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.state_init(0.01)
|
||||
rnn.train(seqs)
|
||||
rnn.state_save()
|
||||
rnn.state_load()
|
||||
print("test_rnn_save_load: [passed]")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_rnn_single_layer()
|
||||
test_rnn_two_layers()
|
||||
test_rnn_save_load()
|
||||
print("All RNN tests passed.")
|
||||
Reference in New Issue
Block a user