pyRBM/RNN-RBM.ipynb

In [1]:

# RNN-RBM
#
# Architecture: unrolled StackRnn (UNROLL_DEPTH entities, alternating t % depth)
#   visible = [context(CONTEXT_SIZE) | x_t(SENSORY_SIZE)]   hidden = CONTEXT_SIZE
#
# Training data: Moby Dick opening, encoded with a 40-char vocabulary.

import numpy as np_cpu
import matplotlib.pyplot as plt
from rbm.stack_rnn import StackRnn
from rbm.matrix import np, convert
from rbm.entity import EntityParams, TrainingParams
from rbm.status import CheckpointStatus

In [2]:

# ── Config ─────────────────────────────────────────────────────────────────
SENSORY_SIZE   = 40      # numVisibleX * numVisibleY = 1 * 40
CONTEXT_SIZE   = 256     # numContext = numHidden
UNROLL_DEPTH   = 1       # number of time-step entities (1 = shared weights)
LEARNING_RATE  = 0.05    # learningRate
MOMENTUM       = 0.9     # momentum
NUM_EPOCHS     = 500     # numEpochs
MINI_BATCH     = 100     # miniBatchSize
NUM_GIBBS      = 3       # numGibbs
RAO_BLACKWELL  = True    # doRaoBlackwell
L2_LAMBDA      = 0.0     # weightDecay
PRJ_NAME       = "RNN-RBM"
WORK_DIR       = "results"

# Vocabulary (40 chars)
ALLOWED     = set(' .!?ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789')
chars       = sorted(ALLOWED)
idx_to_char = {i: c for i, c in enumerate(chars)}
char_to_idx = {c: i for i, c in enumerate(chars)}

In [3]:

# ── Training sentence ──────────────────────────────────────────────────────
SENTENCE = (
    "Call me Ishmael. Some years ago, never mind how long precisely, "
    "having little money in my pocket and nothing particular to interest "
    "me on shore, I thought I would sail about a little."
)

# Convert to uppercase, keep only in-vocab characters
SENTENCE  = ''.join(c for c in SENTENCE.upper() if c in ALLOWED)
N_SAMPLES = len(SENTENCE)

sensory_np = np_cpu.zeros((N_SAMPLES, SENSORY_SIZE), dtype=np_cpu.float64)
for i, c in enumerate(SENTENCE):
    sensory_np[i, char_to_idx[c]] = 1.0

decoded = SENTENCE
print(f"Sentence : '{decoded}'")
print(f"Length   : {N_SAMPLES} chars")

Sentence : 'CALL ME ISHMAEL. SOME YEARS AGO NEVER MIND HOW LONG PRECISELY HAVING LITTLE MONEY IN MY POCKET AND NOTHING PARTICULAR TO INTEREST ME ON SHORE I THOUGHT I WOULD SAIL ABOUT A LITTLE.'
Length   : 180 chars

In [4]:

# ── Visualise one-hot encoding ─────────────────────────────────────────────
step = max(1, N_SAMPLES // 40)    # show at most 40 tick labels
tick_pos = range(0, N_SAMPLES, step)

plt.figure(figsize=(16, 3))
plt.imshow(sensory_np.T, aspect='auto', cmap='hot', vmin=0, vmax=1)
plt.colorbar(label='activation')
plt.xlabel('Position')
plt.ylabel('Vocab index')
plt.title(f'One-hot encoding  ({N_SAMPLES} chars)')
plt.xticks(tick_pos, [decoded[i] for i in tick_pos], fontsize=8)
plt.tight_layout()
plt.show()

In [5]:

# ── Prepare sequences for StackRnn ─────────────────────────────────────────
# Tile the single sentence N_REPEAT times to form a real batch so GPU/CPU
# matrix ops are (N_REPEAT, 168) instead of (1, 168).
N_REPEAT  = 500
sequences = np_cpu.tile(sensory_np[np_cpu.newaxis, :, :], (N_REPEAT, 1, 1))
print(f"sequences shape : {sequences.shape}  →  {N_REPEAT} × {N_SAMPLES} chars")

sequences shape : (500, 180, 40)  →  500 × 180 chars

In [6]:

# ── Build model ────────────────────────────────────────────────────────────
rnn = StackRnn(PRJ_NAME, WORK_DIR)
for layer in StackRnn.make_unrolled(
    UNROLL_DEPTH,
    sensory_size=SENSORY_SIZE,
    h_size=CONTEXT_SIZE,
    entity_params=EntityParams(
        do_gaussian_visible=False,
        do_gaussian_hidden=False,
        num_gibbs_samples=NUM_GIBBS,
    ),
    training_params=TrainingParams(
        learning_rate=LEARNING_RATE,
        momentum=MOMENTUM,
        num_epochs=NUM_EPOCHS,
        mini_batch_size=MINI_BATCH,
        num_gibbs_samples=NUM_GIBBS,
        do_rao_blackwell=RAO_BLACKWELL,
        l2_lambda=L2_LAMBDA,
    ),
):
    rnn.append(layer)

rnn.state_init(0.01)
rnn.state_load()

e = rnn.from_index(0).entity
print(f"Mode       : {'shared' if rnn.is_shared else 'unrolled'}")
print(f"Layers     : {rnn.num_layers()} (alternating t % {rnn.num_layers()})")
print(f"Visible    : {rnn.h_size()} (context) + {rnn.sensory_size()} (sensory) = {e.shape[0]}")
print(f"Hidden     : {rnn.h_size()}")
print(f"Parameters : {e.shape[0] * e.shape[1] * rnn.num_layers():,}  ({rnn.num_layers()} × {e.shape[0] * e.shape[1]:,})")

Mode       : shared
Layers     : 1 (alternating t % 1)
Visible    : 256 (context) + 40 (sensory) = 296
Hidden     : 256
Parameters : 75,776  (1 × 75,776)

In [ ]:

# ── Train: predict next character from current context ─────────────────────
# Unrolled layers alternate: layer[t % num_layers] owns each time-step position.
#   Step 1 — advance context: h_t = layer[t%D].forward([h_{t-1} | x_t])
#   Step 2 — CD on [h_t | x_{t+1}] using the same layer[t%D]
from rbm.train import cd_binary_binary, _to_gpu
from rbm.matrix import rms_error_accu

h_sz             = rnn.h_size()
num_seq, T, s_sz = sequences.shape
n_layers         = rnn.num_layers()

for idx in range(n_layers):
    rnn.from_index(idx).entity.grad_zero()

d_progress = 100.0 / NUM_EPOCHS
progress   = 0.0
entity0    = rnn.from_index(0).entity
status     = CheckpointStatus(rnn.state_save, update_interval=10)
status.on_change(entity0)

for epoch in range(NUM_EPOCHS):
    h = np.zeros((num_seq, h_sz))
    err_total = 0.0

    for t in range(T - 1):
        entity_t = rnn.from_index(t % n_layers).entity
        x_t   = _to_gpu(sequences[:, t,   :])
        x_tp1 = _to_gpu(sequences[:, t+1, :])

        # Step 1: advance context
        h = entity_t.forward(np.concatenate([h, x_t], axis=1))

        # Step 2: CD on [h_t | x_{t+1}]
        vis = np.concatenate([h, x_tp1], axis=1)
        dwhv, dbv, dbh = cd_binary_binary(entity_t, vis)
        grad = entity_t.grad_compute(dbv, dbh, dwhv)
        entity_t.state_adjust(grad, 1.0 / num_seq)
        err_total += rms_error_accu(vis - entity_t.reconstruct(entity_t.forward(vis)))

    progress += d_progress
    if status.want_report(round(progress)):
        if not status.on_change(entity0, {
            "progress": {"value": round(progress), "unit": "%"},
            "err_rms":  {"value": err_total / (T - 1), "unit": ""},
        }):
            break

rnn.state_save()

-------------------------------------------
Entity-296x256: progress              : 0%
Entity-296x256: err_rms              : 0.00012437576727553383
Entity-296x256: l2_norm              : 195.74862335488334

In [ ]:

# ── Reconstruction: teacher-forced ─────────────────────────────────────────
# Step through each sample, reconstruct, compare to input.
rnn.reset(batch_size=1)
recons = []
for i in range(N_SAMPLES):
    x_t = np.array(sensory_np[i][np_cpu.newaxis, :])
    h   = rnn.step(x_t)
    rec = convert(rnn.reconstruct(h))[0]   # (40,)
    recons.append(rec)

recons_np = np_cpu.array(recons)           # (12, 40)
decoded_recon = ''.join(idx_to_char[int(np_cpu.argmax(recons_np[i]))] for i in range(N_SAMPLES))

print(f"Input         : '{decoded}'")
print(f"Reconstruction: '{decoded_recon}'")

correct = sum(decoded[i] == decoded_recon[i] for i in range(N_SAMPLES))
print(f"Char accuracy : {correct}/{N_SAMPLES} = {100*correct/N_SAMPLES:.0f}%")

In [ ]:

# ── Visualise reconstruction ───────────────────────────────────────────────
step = max(1, N_SAMPLES // 40)
tick_pos = range(0, N_SAMPLES, step)

step = max(1, N_SAMPLES // 40)    # show at most 40 tick labels
tick_pos = range(0, N_SAMPLES, step)

plt.figure(figsize=(16, 3))
plt.imshow(sensory_np.T, aspect='auto', cmap='hot', vmin=0, vmax=1)
plt.colorbar(label='activation')
plt.xlabel('Position')
plt.ylabel('Vocab index')
plt.title(f'Input ({N_SAMPLES} chars)')
plt.xticks(tick_pos, [decoded[i] for i in tick_pos], fontsize=8)
plt.tight_layout()
plt.show()

plt.figure(figsize=(16, 3))
plt.imshow(recons_np.T, aspect='auto', cmap='hot', vmin=0, vmax=1)
plt.colorbar(label='activation')
plt.xlabel('Position')
plt.ylabel('Vocab index')
plt.title(f'Reconstruction (acc {100*correct/N_SAMPLES:.0f}%)')
plt.xticks(tick_pos, [decoded[i] for i in tick_pos], fontsize=8)
plt.tight_layout()
plt.show()

In [ ]:

# ── Next-step prediction ───────────────────────────────────────────────────
# Given h_t (context after seeing x_t), predict x_{t+1} via clamped Gibbs.
def predict_next(rnn, context, n_gibbs=NUM_GIBBS, temperature=1.0):
    entity  = rnn.next_entity()
    h_sz, s_sz = rnn.h_size(), rnn.sensory_size()
    x_init  = (np.random.rand(1, s_sz) > 0.5).astype(float)
    visible = np.concatenate([context, x_init], axis=1)
    for _ in range(n_gibbs):
        h       = entity.forward(visible)
        visible = entity.reconstruct(h)
        visible[:, :h_sz] = context
    probs = convert(visible[:, h_sz:])[0]
    probs = np_cpu.power(np_cpu.clip(probs, 1e-10, 1.0), 1.0 / temperature)
    probs /= probs.sum()
    return probs

rnn.reset(batch_size=1)
h         = np.zeros((1, CONTEXT_SIZE))
predicted = ''

for i in range(N_SAMPLES - 1):
    x_t = np.array(sensory_np[i][np_cpu.newaxis, :])
    h   = rnn.step(x_t)                         # advance to h_t
    probs = predict_next(rnn, h.copy())          # predict x_{t+1} from h_t
    predicted += idx_to_char[int(np_cpu.argmax(probs))]

print(f"Input (t+1)         : '{decoded[1:]}'")
print(f"Predicted from h_t  : '{predicted}'")
correct = sum(decoded[i+1] == predicted[i] for i in range(N_SAMPLES - 1))
print(f"Next-step accuracy  : {correct}/{N_SAMPLES-1} = {100*correct/(N_SAMPLES-1):.0f}%")