pyRBM/context33.ipynb

In [1]:

# context33.prj → pyRBM
#
# Architecture: shared-weights StackRnn (1 entity, reused every time step)
#   visible = [context(128) | x_t(40)] = 168   hidden = 128
#
# All hyperparameters taken verbatim from context33.prj.
# Training data: 20-character sentence, encoded with the moby 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 from context33.prj ──────────────────────────────────────────────
SENSORY_SIZE   = 40      # numVisibleX * numVisibleY = 1 * 40
CONTEXT_SIZE   = 128     # numContext = numHidden
LEARNING_RATE  = 0.05    # learningRate
MOMENTUM       = 0.5     # momentum
NUM_EPOCHS     = 1000    # numEpochs
MINI_BATCH     = 100     # miniBatchSize
NUM_GIBBS      = 3       # numGibbs
RAO_BLACKWELL  = True    # doRaoBlackwell
L2_LAMBDA      = 0.0     # weightDecay
PRJ_NAME       = "context33"
WORK_DIR       = "results"

# Vocabulary (40 chars, matching numVisibleY=40 in context33.prj)
# Same encoding as moby_rnn.ipynb
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   = "MOBY DICK IS A WHALE"   # exactly 20 chars, all in vocab
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}'  ({N_SAMPLES} chars)")
print(f"Vocab    : {repr(''.join(chars))}")

Sentence : 'MOBY DICK IS A WHALE'  (20 chars)
Vocab    : ' !.0123456789?ABCDEFGHIJKLMNOPQRSTUVWXYZ'

In [4]:

# ── Visualise one-hot encoding ─────────────────────────────────────────────
plt.figure(figsize=(14, 3))
plt.imshow(sensory_np.T, aspect='auto', cmap='Blues', vmin=0, vmax=1)
plt.colorbar(label='activation')
plt.xlabel('Position')
plt.ylabel('Vocab index')
plt.title(f'One-hot encoding: "{decoded}"')
plt.xticks(range(N_SAMPLES), list(decoded))
plt.tight_layout()
plt.show()

In [5]:

# ── Prepare sequences for StackRnn ─────────────────────────────────────────
sequences = sensory_np[np_cpu.newaxis, :, :]   # (1, 20, 40)
print(f"sequences shape : {sequences.shape}  →  1 sequence of {N_SAMPLES} chars")

sequences shape : (1, 20, 40)  →  1 sequence of 20 chars

In [6]:

# ── Build model ────────────────────────────────────────────────────────────
rnn = StackRnn(PRJ_NAME, WORK_DIR)
rnn.append(StackRnn.make_layer(
    "layer0",
    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.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"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]:,}")

Mode       : shared
Visible    : 128 (context) + 40 (sensory) = 168
Hidden     : 128
Parameters : 21,504

In [7]:

# ── Train: predict next character from current context ─────────────────────
# Pairs used: (h_t, x_{t+1})
#   Step 1 — advance context: h_t = forward([h_{t-1} | x_t])  (no weight update)
#   Step 2 — CD update on visible = [h_t | x_{t+1}]           (predict next)
from rbm.train import cd_binary_binary, _to_gpu
from rbm.matrix import rms_error_accu

entity  = rnn.from_index(0).entity
h_sz    = rnn.h_size()
num_seq, T, s_sz = sequences.shape

rnn.state_init(0.01)   # fresh weights for the new training objective
entity.grad_zero()

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

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

    for t in range(T - 1):
        x_t   = _to_gpu(sequences[:, t,   :])   # current char x_t
        x_tp1 = _to_gpu(sequences[:, t+1, :])   # next char   x_{t+1}

        # Step 1: advance context to h_t (no training)
        h = entity.forward(np.concatenate([h, x_t], axis=1))

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

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

rnn.state_save()

-------------------------------------------
Entity-168x128: progress              : 0%
Entity-168x128: err_rms              : 0.006513810748197312
Entity-168x128: l2_norm              : 0.16668707179363768

-------------------------------------------
Entity-168x128: progress              : 10%
Entity-168x128: err_rms              : 0.005327613080843449
Entity-168x128: l2_norm              : 2.1248952398421848

-------------------------------------------
Entity-168x128: progress              : 20%
Entity-168x128: err_rms              : 0.0022508816936137495
Entity-168x128: l2_norm              : 7.285520227496324

-------------------------------------------
Entity-168x128: progress              : 30%
Entity-168x128: err_rms              : 0.00023891875964096135
Entity-168x128: l2_norm              : 11.942612143538451

-------------------------------------------
Entity-168x128: progress              : 40%
Entity-168x128: err_rms              : 0.0003654588775567404
Entity-168x128: l2_norm              : 14.038897529481288

-------------------------------------------
Entity-168x128: progress              : 50%
Entity-168x128: err_rms              : 0.0005439847855184065
Entity-168x128: l2_norm              : 15.413432769275325

-------------------------------------------
Entity-168x128: progress              : 60%
Entity-168x128: err_rms              : 4.005501369432108e-05
Entity-168x128: l2_norm              : 16.458922493692164

-------------------------------------------
Entity-168x128: progress              : 70%
Entity-168x128: err_rms              : 1.826141900561072e-05
Entity-168x128: l2_norm              : 17.16704910391379

-------------------------------------------
Entity-168x128: progress              : 80%
Entity-168x128: err_rms              : 1.072283482720063e-05
Entity-168x128: l2_norm              : 17.9025294873916

-------------------------------------------
Entity-168x128: progress              : 90%
Entity-168x128: err_rms              : 1.808959411651762e-05
Entity-168x128: l2_norm              : 18.510211685821016

-------------------------------------------
Entity-168x128: progress              : 100%
Entity-168x128: err_rms              : 1.8806019913429117e-05
Entity-168x128: l2_norm              : 19.03956209476975

In [8]:

# ── 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}%")

Input        : 'MOBY DICK IS A WHALE'
Reconstruction: 'KOBY DICK IS A WHALE'
Char accuracy : 19/20 = 95%

In [9]:

# ── Visualise reconstruction ───────────────────────────────────────────────
fig, axes = plt.subplots(1, 2, figsize=(14, 3))

axes[0].imshow(sensory_np.T, aspect='auto', cmap='Blues', vmin=0, vmax=1)
axes[0].set_title(f'Input: "{decoded}"')
axes[0].set_xticks(range(N_SAMPLES))
axes[0].set_xticklabels(list(decoded))
axes[0].set_ylabel('Vocab index')

axes[1].imshow(recons_np.T, aspect='auto', cmap='Blues', vmin=0, vmax=1)
axes[1].set_title(f'Reconstruction: "{decoded_recon}"')
axes[1].set_xticks(range(N_SAMPLES))
axes[1].set_xticklabels(list(decoded_recon))

plt.tight_layout()
plt.show()

In [10]:

# ── 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}%")

Input (t+1)         : 'OBY DICK IS A WHALE'
Predicted from h_t  : 'OIY DICK IS A WHALE'
Next-step accuracy  : 18/19 = 95%