Phase 03 - Lesson 10

Construye tu propio Mini Framework

This lesson includes a graded coding exercise that runs in your browser, unlocked with lifetime access.

Has construido neuronas, capas, redes, backprop, activaciones, funciones de perdida, optimizadores, regularizacion, inicializacion y programaciones de tasa de aprendizaje. Todo como piezas separadas. Ahora conectalas en un framework. No PyTorch. No TensorFlow. El tuyo.

Tipo: Build Lenguajes: Python Prerrequisitos: Toda la Fase 03 (Lecciones 01-09) Tiempo: ~120 minutos

Objetivos de Aprendizaje

• Construir un framework completo de deep learning (~500 lineas) con Module, Linear, ReLU, Sigmoid, Dropout, BatchNorm, Sequential, funciones de perdida, optimizadores y DataLoader
• Explicar la abstraccion Module (forward, backward, parameters) y por que es necesario alternar entre los modos de entrenamiento/evaluacion
• Conectar todos los componentes en un bucle de entrenamiento funcional que entrena una red de 4 capas en clasificacion de circulo
• Mapear cada componente de tu framework a su equivalente en PyTorch (nn.Module, nn.Sequential, optim.Adam, DataLoader)

El Problema

Tienes diez lecciones de bloques de construccion dispersos en archivos separados. Una clase Value aqui, un bucle de entrenamiento alla, inicializacion de pesos en otro archivo, programaciones de tasa de aprendizaje en otro mas. Para entrenar una red, copias y pegas de cinco lecciones diferentes y las conectas a mano.

Eso es lo que resuelven los frameworks. PyTorch te da nn.Module, nn.Sequential, optim.Adam, DataLoader y un patron de bucle de entrenamiento que une todo eso. TensorFlow te da keras.Layer, keras.Sequential, keras.optimizers.Adam. Esto no es magia. Son patrones organizativos que hacen posible definir, entrenar y evaluar redes sin reinventar la plomeria cada vez.

Vas a construir lo mismo en ~500 lineas de Python. Sin numpy. Sin dependencias externas. Un framework que puede definir cualquier red feedforward, entrenarla con SGD o Adam, dividir los datos en lotes, aplicar dropout y batch normalization, usar cualquier activacion y programar la tasa de aprendizaje.

Cuando termines, vas a entender exactamente que sucede cuando escribes model = nn.Sequential(...) en PyTorch. Vas a entender por que existen model.train() y model.eval(). Vas a entender por que optimizer.zero_grad() es una llamada separada. Vas a entender todo eso, porque lo construiste todo.

El Concepto

La Abstraccion Module

Cada capa en PyTorch hereda de nn.Module. Un Module tiene tres responsabilidades:

1. forward() -- calcula la salida dada la entrada
2. parameters() -- retorna todos los pesos entrenables
3. backward() -- calcula gradientes (manejado por autograd en PyTorch, explicito en el nuestro)

Una capa Linear es un Module. Una activacion ReLU es un Module. Una capa de dropout es un Module. Una capa de batch normalization es un Module. Todas tienen la misma interfaz.

Contenedor Sequential

nn.Sequential encadena Modules. Pasada forward: alimenta los datos por el Module 1, luego el Module 2, luego el Module 3. Pasada backward: invierte la cadena. El contenedor mismo es un Module -- tiene forward(), parameters() y backward(). Este es el patron composite: una secuencia de Modules es, ella misma, un Module.

Modo de Entrenamiento vs Evaluacion

El Dropout pone neuronas en cero de forma aleatoria durante el entrenamiento, pero deja pasar todo durante la evaluacion. La batch normalization usa estadisticas del lote durante el entrenamiento, pero promedios moviles durante la evaluacion. Los metodos train() y eval() alternan este comportamiento. Cada Module tiene una bandera training.

Optimizador

El optimizador actualiza los parametros usando sus gradientes. SGD: param -= lr * grad. Adam: mantiene estimaciones de momento y varianza y, luego, actualiza. El optimizador no conoce la arquitectura de la red -- solo ve una lista plana de parametros y sus gradientes.

DataLoader

Dividir en lotes importa por dos razones. Primero, no puedes meter el dataset entero en memoria en problemas grandes. Segundo, el descenso de gradiente por mini-lotes proporciona un ruido que ayuda a escapar de minimos locales. El DataLoader divide los datos en lotes y, opcionalmente, los mezcla entre epocas.

Arquitectura del Framework

graph TD
    subgraph "Modules"
        Linear["Linear<br/>W*x + b"]
        ReLU["ReLU<br/>max(0, x)"]
        Sigmoid["Sigmoid<br/>1/(1+e^-x)"]
        Dropout["Dropout<br/>mascara cero aleatoria"]
        BatchNorm["BatchNorm<br/>normaliza activaciones"]
    end

    subgraph "Containers"
        Sequential["Sequential<br/>encadena modules"]
    end

    subgraph "Loss Functions"
        MSE["MSELoss<br/>(pred - target)^2"]
        BCE["BCELoss<br/>entropia cruzada binaria"]
    end

    subgraph "Optimizers"
        SGD["SGD<br/>param -= lr * grad"]
        Adam["Adam<br/>momentos adaptativos"]
    end

    subgraph "Data"
        DataLoader["DataLoader<br/>lotes + mezcla"]
    end

    Sequential --> |"contiene"| Linear
    Sequential --> |"contiene"| ReLU
    Sequential --> |"forward/backward"| MSE
    SGD --> |"actualiza"| Sequential
    DataLoader --> |"alimenta"| Sequential

Bucle de Entrenamiento

sequenceDiagram
    participant DL as DataLoader
    participant M as Model
    participant L as Loss
    participant O as Optimizer

    loop Cada Epoca
        DL->>M: lote de entradas
        M->>M: pasada forward (capa por capa)
        M->>L: predicciones
        L->>L: calcula la perdida
        L->>M: pasada backward (gradientes)
        M->>O: parametros + gradientes
        O->>M: parametros actualizados
        O->>O: pone gradientes en cero
    end

Jerarquia de Module

classDiagram
    class Module {
        +forward(x)
        +backward(grad)
        +parameters()
        +train()
        +eval()
    }

    class Linear {
        -weights
        -biases
        +forward(x)
        +backward(grad)
    }

    class ReLU {
        +forward(x)
        +backward(grad)
    }

    class Sequential {
        -modules[]
        +forward(x)
        +backward(grad)
        +parameters()
    }

    Module <|-- Linear
    Module <|-- ReLU
    Module <|-- Sequential
    Sequential *-- Module

Construyelo

Paso 1: Clase Base Module

La interfaz abstracta que cada capa implementa.

class Module:
    def __init__(self):
        self.training = True

    def forward(self, x):
        raise NotImplementedError

    def backward(self, grad):
        raise NotImplementedError

    def parameters(self):
        return []

    def train(self):
        self.training = True

    def eval(self):
        self.training = False

Paso 2: Capa Linear

El bloque de construccion fundamental. Almacena pesos y sesgos, calcula Wx + b en el forward y gradientes de peso/entrada en el backward.

import math
import random


class Linear(Module):
    def __init__(self, fan_in, fan_out):
        super().__init__()
        std = math.sqrt(2.0 / fan_in)
        self.weights = [[random.gauss(0, std) for _ in range(fan_in)] for _ in range(fan_out)]
        self.biases = [0.0] * fan_out
        self.weight_grads = [[0.0] * fan_in for _ in range(fan_out)]
        self.bias_grads = [0.0] * fan_out
        self.fan_in = fan_in
        self.fan_out = fan_out
        self.input = None

    def forward(self, x):
        self.input = x
        output = []
        for i in range(self.fan_out):
            val = self.biases[i]
            for j in range(self.fan_in):
                val += self.weights[i][j] * x[j]
            output.append(val)
        return output

    def backward(self, grad):
        input_grad = [0.0] * self.fan_in
        for i in range(self.fan_out):
            self.bias_grads[i] += grad[i]
            for j in range(self.fan_in):
                self.weight_grads[i][j] += grad[i] * self.input[j]
                input_grad[j] += grad[i] * self.weights[i][j]
        return input_grad

    def parameters(self):
        params = []
        for i in range(self.fan_out):
            for j in range(self.fan_in):
                params.append((self.weights, i, j, self.weight_grads))
            params.append((self.biases, i, None, self.bias_grads))
        return params

Paso 3: Modules de Activacion

ReLU, Sigmoid y Tanh como Modules. Cada uno guarda en cache lo que necesita para la pasada backward.

class ReLU(Module):
    def __init__(self):
        super().__init__()
        self.mask = None

    def forward(self, x):
        self.mask = [1.0 if v > 0 else 0.0 for v in x]
        return [max(0.0, v) for v in x]

    def backward(self, grad):
        return [g * m for g, m in zip(grad, self.mask)]


class Sigmoid(Module):
    def __init__(self):
        super().__init__()
        self.output = None

    def forward(self, x):
        self.output = []
        for v in x:
            v = max(-500, min(500, v))
            self.output.append(1.0 / (1.0 + math.exp(-v)))
        return self.output

    def backward(self, grad):
        return [g * o * (1 - o) for g, o in zip(grad, self.output)]


class Tanh(Module):
    def __init__(self):
        super().__init__()
        self.output = None

    def forward(self, x):
        self.output = [math.tanh(v) for v in x]
        return self.output

    def backward(self, grad):
        return [g * (1 - o * o) for g, o in zip(grad, self.output)]

Paso 4: Module Dropout

Pone elementos en cero de forma aleatoria durante el entrenamiento. Escala los elementos restantes por 1/(1-p) para que los valores esperados se mantengan iguales. No hace nada durante la evaluacion.

class Dropout(Module):
    def __init__(self, p=0.5):
        super().__init__()
        self.p = p
        self.mask = None

    def forward(self, x):
        if not self.training:
            return x
        self.mask = [0.0 if random.random() < self.p else 1.0 / (1 - self.p) for _ in x]
        return [v * m for v, m in zip(x, self.mask)]

    def backward(self, grad):
        if self.mask is None:
            return grad
        return [g * m for g, m in zip(grad, self.mask)]

Paso 5: Module BatchNorm

Normaliza las activaciones a media cero y varianza unitaria por feature a lo largo del lote. Mantiene estadisticas moviles para el modo de evaluacion.

class BatchNorm(Module):
    def __init__(self, size, momentum=0.1, eps=1e-5):
        super().__init__()
        self.size = size
        self.gamma = [1.0] * size
        self.beta = [0.0] * size
        self.gamma_grads = [0.0] * size
        self.beta_grads = [0.0] * size
        self.running_mean = [0.0] * size
        self.running_var = [1.0] * size
        self.momentum = momentum
        self.eps = eps
        self.x_norm = None
        self.std_inv = None
        self.batch_input = None

    def forward_batch(self, batch):
        batch_size = len(batch)
        output_batch = []

        if self.training:
            mean = [0.0] * self.size
            for sample in batch:
                for j in range(self.size):
                    mean[j] += sample[j]
            mean = [m / batch_size for m in mean]

            var = [0.0] * self.size
            for sample in batch:
                for j in range(self.size):
                    var[j] += (sample[j] - mean[j]) ** 2
            var = [v / batch_size for v in var]

            self.std_inv = [1.0 / math.sqrt(v + self.eps) for v in var]

            self.x_norm = []
            self.batch_input = batch
            for sample in batch:
                normed = [(sample[j] - mean[j]) * self.std_inv[j] for j in range(self.size)]
                self.x_norm.append(normed)
                output = [self.gamma[j] * normed[j] + self.beta[j] for j in range(self.size)]
                output_batch.append(output)

            for j in range(self.size):
                self.running_mean[j] = (1 - self.momentum) * self.running_mean[j] + self.momentum * mean[j]
                self.running_var[j] = (1 - self.momentum) * self.running_var[j] + self.momentum * var[j]
        else:
            std_inv = [1.0 / math.sqrt(v + self.eps) for v in self.running_var]
            for sample in batch:
                normed = [(sample[j] - self.running_mean[j]) * std_inv[j] for j in range(self.size)]
                output = [self.gamma[j] * normed[j] + self.beta[j] for j in range(self.size)]
                output_batch.append(output)

        return output_batch

    def forward(self, x):
        result = self.forward_batch([x])
        return result[0]

    def backward(self, grad):
        if self.x_norm is None:
            return grad
        for j in range(self.size):
            self.gamma_grads[j] += self.x_norm[0][j] * grad[j]
            self.beta_grads[j] += grad[j]
        return [grad[j] * self.gamma[j] * self.std_inv[j] for j in range(self.size)]

    def parameters(self):
        params = []
        for j in range(self.size):
            params.append((self.gamma, j, None, self.gamma_grads))
            params.append((self.beta, j, None, self.beta_grads))
        return params

Paso 6: Contenedor Sequential

Encadena modules. El forward va de izquierda a derecha, el backward va de derecha a izquierda.

class Sequential(Module):
    def __init__(self, *modules):
        super().__init__()
        self.modules = list(modules)

    def forward(self, x):
        for module in self.modules:
            x = module.forward(x)
        return x

    def backward(self, grad):
        for module in reversed(self.modules):
            grad = module.backward(grad)
        return grad

    def parameters(self):
        params = []
        for module in self.modules:
            params.extend(module.parameters())
        return params

    def train(self):
        self.training = True
        for module in self.modules:
            module.train()

    def eval(self):
        self.training = False
        for module in self.modules:
            module.eval()

Paso 7: Funciones de Perdida

MSE y Entropia Cruzada Binaria. Cada una retorna el valor de la perdida y proporciona un backward() que retorna el gradiente.

class MSELoss:
    def __call__(self, predicted, target):
        self.predicted = predicted
        self.target = target
        n = len(predicted)
        self.loss = sum((p - t) ** 2 for p, t in zip(predicted, target)) / n
        return self.loss

    def backward(self):
        n = len(self.predicted)
        return [2 * (p - t) / n for p, t in zip(self.predicted, self.target)]


class BCELoss:
    def __call__(self, predicted, target):
        self.predicted = predicted
        self.target = target
        eps = 1e-7
        n = len(predicted)
        self.loss = 0
        for p, t in zip(predicted, target):
            p = max(eps, min(1 - eps, p))
            self.loss += -(t * math.log(p) + (1 - t) * math.log(1 - p))
        self.loss /= n
        return self.loss

    def backward(self):
        eps = 1e-7
        n = len(self.predicted)
        grads = []
        for p, t in zip(self.predicted, self.target):
            p = max(eps, min(1 - eps, p))
            grads.append((-t / p + (1 - t) / (1 - p)) / n)
        return grads

Paso 8: Optimizadores SGD y Adam

Ambos reciben una lista de parametros y actualizan los pesos usando los gradientes.

class SGD:
    def __init__(self, parameters, lr=0.01):
        self.params = parameters
        self.lr = lr

    def step(self):
        for container, i, j, grad_container in self.params:
            if j is not None:
                container[i][j] -= self.lr * grad_container[i][j]
            else:
                container[i] -= self.lr * grad_container[i]

    def zero_grad(self):
        for container, i, j, grad_container in self.params:
            if j is not None:
                grad_container[i][j] = 0.0
            else:
                grad_container[i] = 0.0


class Adam:
    def __init__(self, parameters, lr=0.001, beta1=0.9, beta2=0.999, eps=1e-8):
        self.params = parameters
        self.lr = lr
        self.beta1 = beta1
        self.beta2 = beta2
        self.eps = eps
        self.t = 0
        self.m = [0.0] * len(parameters)
        self.v = [0.0] * len(parameters)

    def step(self):
        self.t += 1
        for idx, (container, i, j, grad_container) in enumerate(self.params):
            if j is not None:
                g = grad_container[i][j]
            else:
                g = grad_container[i]

            self.m[idx] = self.beta1 * self.m[idx] + (1 - self.beta1) * g
            self.v[idx] = self.beta2 * self.v[idx] + (1 - self.beta2) * g * g

            m_hat = self.m[idx] / (1 - self.beta1 ** self.t)
            v_hat = self.v[idx] / (1 - self.beta2 ** self.t)

            update = self.lr * m_hat / (math.sqrt(v_hat) + self.eps)

            if j is not None:
                container[i][j] -= update
            else:
                container[i] -= update

    def zero_grad(self):
        for container, i, j, grad_container in self.params:
            if j is not None:
                grad_container[i][j] = 0.0
            else:
                grad_container[i] = 0.0

Paso 9: DataLoader

Divide los datos en lotes, mezclandolos opcionalmente en cada epoca.

class DataLoader:
    def __init__(self, data, batch_size=32, shuffle=True):
        self.data = data
        self.batch_size = batch_size
        self.shuffle = shuffle

    def __iter__(self):
        indices = list(range(len(self.data)))
        if self.shuffle:
            random.shuffle(indices)
        for start in range(0, len(indices), self.batch_size):
            batch_indices = indices[start:start + self.batch_size]
            batch = [self.data[i] for i in batch_indices]
            inputs = [item[0] for item in batch]
            targets = [item[1] for item in batch]
            yield inputs, targets

    def __len__(self):
        return (len(self.data) + self.batch_size - 1) // self.batch_size

Paso 10: Entrena una Red de 4 Capas en la Clasificacion de Circulo

Conectalo todo. Define un modelo, elige una perdida, elige un optimizador, ejecuta el bucle de entrenamiento.

def make_circle_data(n=500, seed=42):
    random.seed(seed)
    data = []
    for _ in range(n):
        x = random.uniform(-2, 2)
        y = random.uniform(-2, 2)
        label = 1.0 if x * x + y * y < 1.5 else 0.0
        data.append(([x, y], [label]))
    return data


def train():
    random.seed(42)

    model = Sequential(
        Linear(2, 16),
        ReLU(),
        Linear(16, 16),
        ReLU(),
        Linear(16, 8),
        ReLU(),
        Linear(8, 1),
        Sigmoid(),
    )

    criterion = BCELoss()
    optimizer = Adam(model.parameters(), lr=0.01)

    data = make_circle_data(500)
    split = int(len(data) * 0.8)
    train_data = data[:split]
    test_data = data[split:]

    loader = DataLoader(train_data, batch_size=16, shuffle=True)

    model.train()

    for epoch in range(100):
        total_loss = 0
        total_correct = 0
        total_samples = 0

        for batch_inputs, batch_targets in loader:
            batch_loss = 0
            for x, t in zip(batch_inputs, batch_targets):
                pred = model.forward(x)
                loss = criterion(pred, t)
                batch_loss += loss

                optimizer.zero_grad()
                grad = criterion.backward()
                model.backward(grad)
                optimizer.step()

                predicted_class = 1.0 if pred[0] >= 0.5 else 0.0
                if predicted_class == t[0]:
                    total_correct += 1
                total_samples += 1

            total_loss += batch_loss

        avg_loss = total_loss / total_samples
        accuracy = total_correct / total_samples * 100

        if epoch % 10 == 0 or epoch == 99:
            print(f"Epoch {epoch:3d} | Loss: {avg_loss:.6f} | Train Accuracy: {accuracy:.1f}%")

    model.eval()
    correct = 0
    for x, t in test_data:
        pred = model.forward(x)
        predicted_class = 1.0 if pred[0] >= 0.5 else 0.0
        if predicted_class == t[0]:
            correct += 1
    test_accuracy = correct / len(test_data) * 100
    print(f"\nTest Accuracy: {test_accuracy:.1f}% ({correct}/{len(test_data)})")

    return model, test_accuracy

Usalo

Aqui esta el equivalente en PyTorch de lo que acabas de construir:

import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset

model = nn.Sequential(
    nn.Linear(2, 16),
    nn.ReLU(),
    nn.Linear(16, 16),
    nn.ReLU(),
    nn.Linear(16, 8),
    nn.ReLU(),
    nn.Linear(8, 1),
    nn.Sigmoid(),
)

criterion = nn.BCELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)

for epoch in range(100):
    model.train()
    for inputs, targets in dataloader:
        optimizer.zero_grad()
        predictions = model(inputs)
        loss = criterion(predictions, targets)
        loss.backward()
        optimizer.step()

    model.eval()
    with torch.no_grad():
        test_predictions = model(test_inputs)

La estructura es identica. Sequential, Linear, ReLU, Sigmoid, BCELoss, Adam, zero_grad, backward, step, train, eval. Cada concepto mapea uno a uno. La diferencia es que PyTorch maneja el autograd automaticamente (sin necesidad de implementar backward() en cada module), corre en GPU y ha sido optimizado durante anos. Pero los huesos son los mismos.

Ahora, cuando veas codigo PyTorch, sabes exactamente que esta sucediendo en cada linea. Ese entendimiento es el objetivo de todo.

Entregalo

Esta leccion produce:

• outputs/prompt-framework-architect.md -- un prompt para disenar arquitecturas de redes neuronales usando abstracciones de framework

Ejercicios

1. Agrega una clase SoftmaxCrossEntropyLoss para clasificacion multiclase. Aplica softmax a las predicciones, calcula la perdida de entropia cruzada y maneja la pasada backward combinada. Pruebala en un dataset espiral de 3 clases.

2. Implementa la programacion de la tasa de aprendizaje en el optimizador: agrega un metodo set_lr() y conecta la programacion coseno de la Leccion 09. Entrena el clasificador de circulo con warmup + coseno y comparalo con LR constante.

3. Agrega metodos save() y load() al Sequential que serialicen todos los pesos en un archivo JSON y los carguen de vuelta. Verifica que un modelo cargado produzca las mismas predicciones que el original.

4. Implementa el weight decay (regularizacion L2) en el optimizador Adam. Agrega un parametro weight_decay que encoge los pesos hacia cero en cada paso. Compara el entrenamiento con decay=0 vs decay=0.01.

5. Reemplaza el bucle de entrenamiento por muestra con una acumulacion de gradiente por mini-lotes adecuada: acumula los gradientes a lo largo de todas las muestras de un lote, luego divide por el tamano del lote y da un solo paso del optimizador. Mide si esto cambia la velocidad de convergencia.

Terminos Clave

Lectura Adicional

• Paszke et al., "PyTorch: An Imperative Style, High-Performance Deep Learning Library" (2019) -- el articulo que describe las decisiones de diseno de PyTorch
• Chollet, "Deep Learning with Python, Second Edition" (2021) -- el Capitulo 3 cubre las entranas de Keras con la misma abstraccion module/capa
• Johnson, "Tiny-DNN" (https://github.com/tiny-dnn/tiny-dnn) -- un framework de deep learning header-only en C++ para entender las entranas de un framework