Building LSTM from Scratch with NumPy

import numpy as np
np.random.seed(0)
 
def sigmoid(z):
    return 1.0 / (1.0 + np.exp(-z))
 
def dsigmoid(a):
    """সিগময়েডের ডেরিভেটিভ, যেখানে অ্যাক্টিভেটেড মান a = sigmoid(z) দেওয়া আছে।"""
    return a * (1 - a)
 
def dtanh(a):
    """ট্যানএইচ (tanh)-এর ডেরিভেটিভ, যেখানে অ্যাক্টিভেটেড মান a = tanh(z) দেওয়া আছে।"""
    return 1 - a ** 2
 
 
class LSTMFromScratch:
    """একক লেয়ারের LSTM এবং একটি লিনিয়ার আউটপুট লেয়ার, সম্পূর্ণ NumPy-তে তৈরি।"""
 
    def __init__(self, input_size, hidden_size):
        self.input_size = input_size
        self.hidden_size = hidden_size
        concat_size = input_size + hidden_size
        scale = 1.0 / np.sqrt(hidden_size)   # গুরুত্বপূর্ণ: ট্রেনিং শুরুর সময় গেটগুলোকে স্যাচুরেশন থেকে বাঁচায়
 
        # প্রতি গেটের জন্য একটি করে কম্বাইন্ড ওয়েট ম্যাট্রিক্স
        self.Wf = np.random.randn(hidden_size, concat_size) * scale
        self.Wi = np.random.randn(hidden_size, concat_size) * scale
        self.Wc = np.random.randn(hidden_size, concat_size) * scale
        self.Wo = np.random.randn(hidden_size, concat_size) * scale
        self.bf = np.zeros((hidden_size, 1))
        self.bi = np.zeros((hidden_size, 1))
        self.bc = np.zeros((hidden_size, 1))
        self.bo = np.zeros((hidden_size, 1))
 
        # আউটপুট লেয়ার: চূড়ান্ত হিডেন স্টেটকে একটি স্কেলার প্রেডিকশনে ম্যাপিং করে
        self.Wy = np.random.randn(1, hidden_size) * scale
        self.by = np.zeros((1, 1))
 
    def forward(self, xs):
        """
        একটি পুরো সিকোয়েন্সের ওপর ফরোয়ার্ড পাস (forward pass)।
        xs: কলাম ভেক্টরের লিস্ট, যার প্রতিটির শেপ (input_size, 1)।
        """
        h = np.zeros((self.hidden_size, 1))
        C = np.zeros((self.hidden_size, 1))
        caches = []
 
        for x in xs:
            concat = np.vstack([x, h])              # [x_t; h_{t-1}] ভেক্টর তৈরি
            f = sigmoid(self.Wf @ concat + self.bf) # ফরগেট গেট
            i = sigmoid(self.Wi @ concat + self.bi) # ইনপুট গেট
            c_tilde = np.tanh(self.Wc @ concat + self.bc)  # ক্যান্ডিডেট বা ড্রাফট
            C = f * C + i * c_tilde                 # সেল আপডেট  ← লক্ষ্য করুন: এখানে যোগ হচ্ছে
            o = sigmoid(self.Wo @ concat + self.bo) # আউটপুট গেট
            h = o * np.tanh(C)                      # হিডেন স্টেট
            caches.append((concat, f, i, c_tilde, C, o, h))
 
        y_pred = self.Wy @ h + self.by
        return y_pred, h, C, caches

    def backward(self, y_true, y_pred, h, caches, lr=0.03):
        """ব্যাকপ্রোপাগেশন থ্রু টাইম (BPTT), এবং এরপর একটি গ্রেডিয়েন্ট-ডিসেন্ট আপডেট।"""
        # প্রতিটি ওয়েটের জন্য গ্রেডিয়েন্ট অ্যাকুমুলেটর ইনিশিয়ালাইজ করা
        dWf = np.zeros_like(self.Wf); dWi = np.zeros_like(self.Wi)
        dWc = np.zeros_like(self.Wc); dWo = np.zeros_like(self.Wo)
        dbf = np.zeros_like(self.bf); dbi = np.zeros_like(self.bi)
        dbc = np.zeros_like(self.bc); dbo = np.zeros_like(self.bo)
 
        dy = y_pred - y_true         # d(MSE)/d(y_pred)
        dWy = dy @ h.T
        dby = dy
        dh_next = self.Wy.T @ dy
        dC_next = np.zeros((self.hidden_size, 1))
 
        for t in reversed(range(len(caches))):
            concat, f, i, c_tilde, C_t, o, h_t = caches[t]
            C_prev = caches[t-1][4] if t > 0 else np.zeros((self.hidden_size, 1))
 
            # হিডেন স্টেট এবং সেল স্টেট উভয় পথ থেকেই গ্রেডিয়েন্ট যোগ বা সঞ্চিত হচ্ছে
            dC = dC_next + dh_next * o * dtanh(np.tanh(C_t))
            do = dh_next * np.tanh(C_t)   # আউটপুট গেট
            df = dC * C_prev               # ফরগেট গেট
            di = dC * c_tilde              # ইনপুট গেট
            dc_tilde = dC * i              # ক্যান্ডিডেট
 
            dzf = df * dsigmoid(f)
            dzi = di * dsigmoid(i)
            dzc = dc_tilde * dtanh(c_tilde)
            dzo = do * dsigmoid(o)
 
            dWf += dzf @ concat.T; dbf += dzf
            dWi += dzi @ concat.T; dbi += dzi
            dWc += dzc @ concat.T; dbc += dzc
            dWo += dzo @ concat.T; dbo += dzo
 
            dconcat = (self.Wf.T @ dzf + self.Wi.T @ dzi +
                       self.Wc.T @ dzc + self.Wo.T @ dzo)
            dh_next = dconcat[self.input_size:, :]  # h_{t-1}-এর গ্রেডিয়েন্ট
            dC_next = dC * f                         # ← এটাই সেই 'constant error carousel'
 
        for param, grad in [(self.Wf, dWf), (self.Wi, dWi), (self.Wc, dWc), (self.Wo, dWo),
                             (self.bf, dbf), (self.bi, dbi), (self.bc, dbc), (self.bo, dbo),
                             (self.Wy, dWy), (self.by, dby)]:
            param -= lr * grad

def make_batch(seq_len=4):
    """র্যান্ডম সিকোয়েন্স তৈরি করা; এর টার্গেট হলো ভেতরের উপাদানগুলোর যোগফল।"""
    xs_raw = np.random.uniform(-1, 1, seq_len)
    xs = [np.array([[v]]) for v in xs_raw]
    y_true = np.array([[xs_raw.sum()]])
    return xs, y_true
 
 
model = LSTMFromScratch(input_size=1, hidden_size=8)
losses = []
 
for epoch in range(300):
    xs, y_true = make_batch(seq_len=4)
    y_pred, h, C, caches = model.forward(xs)
    loss = float(np.squeeze((y_pred - y_true) ** 2))
    losses.append(loss)
    model.backward(y_true, y_pred, h, caches, lr=0.03)
 
    if epoch % 50 == 0:
        recent_avg = np.mean(losses[max(0, epoch-29):epoch+1])
        print(f"epoch {epoch:4d} | recent avg loss: {recent_avg:.4f}")
 
print(f"final 30-epoch avg loss: {np.mean(losses[-30:]):.4f}")