Improved and refactored FCN modules as well as Parameters, Optimizers and a Training Loop
def logsumexp(inp):
a = inp.max(dim=1).values
return a + ((inp-a[:,None]).exp().sum(-1)).log()
def log_softmax_improved(inp):
return inp - logsumexp(inp).unsqueeze(-1)
def fit(epochs, model, loss_func, train, valid):
for epoch in range(epochs):
for batch in range(math.ceil(len(train)//bs)):
start = batch*bs
end = batch*bs + bs
train_batch = train[start:end]
valid_batch = valid[start:end]
loss = loss_func(model(train_batch), valid_batch)
loss_func.backward(loss, )
model.backward()
with torch.no_grad():
for l in model.layers:
if hasattr(l, 'w'):
l.w -= l.w.g * lr
l.b -= l.b.g * lr
l.w.g = 0
l.b.g = 0
print(f'Epoch {epoch+1}, Accuracy: {accuracy(model(xt), yt)}')
l_data = torch.zeros([4,6])
s_data = torch.randn([4,2])
param_l = Parameter(l_data)
param_s = Parameter(s_data)
param_l
param_s
class SequentialModel():
"Model for executing forward and backward passes on a given list of `layers`"
def __init__(self, *args):
self.layers = list(args)
self.training = True
def __repr__(self):
"Prints out all modules of model"
res = ["(Layer" + str(i+1) + "): " + str(m) for i,m in enumerate(self.layers)]
return "\n".join(res)
def __call__(self, x):
"Execute forward pass on `x` throuh `self.layers`"
for l in self.layers: x = l(x)
return x
def backward(self):
"Execute backward pass on `x` throuh `self.layers`"
for l in reversed(self.layers): l.backward()
def parameters(self):
"Get iterator over all parameters in layers of `self.layers`"
for l in self.layers:
for p in l.parameters(): yield p
class CrossEntropy(Module):
def forward(self, inp, targ):
return cross_entropy(inp, targ)
def bwd(self, loss, inp, targ):
inp_s = softmax(inp)
inp_s[range(targ.shape[0]), targ.long()] -= 1
inp.g = inp_s / targ.shape[0]
class Linear(Module):
def __init__(self, in_d, out_d, relu_after, req_grad=True):
super().__init__()
self.w = Parameter(get_weight(in_d, out_d, relu_after), req_grad)
self.b = Parameter(torch.zeros(out_d), req_grad)
def forward(self, xb): return xb @ self.w.d + self.b.d
def bwd(self, out, inp):
inp.g = out.g @ self.w.d.t()
self.w.update(inp.t() @ out.g)
self.b.update(out.g.sum(0))
def __repr__(self): return f'Linear({self.w.d.shape[0]}, {self.w.d.shape[1]})'
class ReLU(Module):
def forward(self, x): return x.clamp_min_(0.)-0.5
def bwd(self, out, inp):
inp.g = (inp>0).float() * out.g
def __repr__(self): return f'ReLU()'
model = SequentialModel(Linear(n_in,50, True), ReLU(), Linear(50,n_out, False))
model
for p in model.parameters(): print(p)
optim = Optimizer(model.parameters(), 0.5)
def fit(epochs, model, optim, loss_func, train, valid):
"Fit function 2: Added easier optimization steps"
for epoch in range(epochs):
for batch in range(math.ceil(len(train)//bs)):
start = batch*bs
end = batch*bs + bs
train_batch = train[start:end]
valid_batch = valid[start:end]
loss = loss_func(model(train_batch), valid_batch)
loss_func.backward()
model.backward()
optim.step()
optim.zero_grad()
print(f'Epoch {epoch+1}, Accuracy: {accuracy(model(xt), yt)}')
fit(3, model, optim, CrossEntropy(), xt, yt)
class DataLoader():
"Container class to iterate over a dataset given a batcht size"
def __init__(self, ds, bs):
"Provide a dataset, `ds`, and batchsize, `bs`"
self.ds, self.bs = ds,bs
def __iter__(self):
for i in range(0, len(self.ds), self.bs): yield self.ds[i:i+self.bs]
ds = Dataset(xt, yt)
dl = DataLoader(ds, bs)
def fit(epochs, model, optim, loss_func, data_loader):
"Training Loop 3: Refactored out for easy databunch usage"
for epoch in range(epochs):
for xb, yb in data_loader:
loss = loss_func(model(xb), yb)
loss_func.backward()
model.backward()
optim.step()
optim.zero_grad()
print(f'Epoch {epoch+1}, Accuracy: {accuracy(model(xt), yt)}')
m, o, lf = get_model(0.5)
fit(3, m, o, lf, dl)
small_ds = Dataset(*ds[:10])
random = Batcher(small_ds, 4, True)
not_random = Batcher(small_ds, 4, False)
[r for r in random]
[nr for nr in not_random]
train, valid = get_datasets()
m, o, lf = get_model(0.1)
fit(3, m, o, lf, train, valid)