47 KiB
47 KiB
In [47]:
# 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 CheckpointStatusIn [48]:
# ── Config from context33.prj ──────────────────────────────────────────────
SENSORY_SIZE = 40 # numVisibleX * numVisibleY = 1 * 40
CONTEXT_SIZE = 256 # numContext = numHidden
LEARNING_RATE = 0.05 # learningRate
MOMENTUM = 0.7 # momentum
NUM_EPOCHS = 100 # 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 [49]:
# ── 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 [50]:
# ── 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 [51]:
# ── 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 [52]:
# ── 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 : 256 (context) + 40 (sensory) = 296 Hidden : 256 Parameters : 75,776
In [ ]:
# ── 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
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-296x256: progress : 1% Entity-296x256: err_rms : 4.153017505591589e-06 Entity-296x256: l2_norm : 149.9841498059769
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}%")