基于NumPy从零实现神经网络：小批量梯度下降预测输出异常

问题根源分析 #

你的代码存在几个核心逻辑错误，直接导致了预测输出维度与预期不符的问题：

1. shuffle_batch函数返回逻辑错误
   你使用return而非yield，导致函数生成第一个batch后就直接终止，无法遍历所有训练批次。同时fit方法里还错误地将batch数据赋值给了输出变量，完全违背训练逻辑。

2. Dense层反向传播权重梯度计算错误
   对self.input.T的冗余reshape操作破坏了梯度维度，导致权重更新时维度不匹配，最终扰乱了整个网络的输出结构。正确的权重梯度应该和权重本身维度一致。

3. fit方法训练循环逻辑混乱

   • 仅处理了单个batch，没有遍历所有训练数据
   • 损失计算与平均方式错误，无法反映真实训练误差
   • 变量赋值逻辑颠倒，将训练数据直接当作输出处理

4. predict方法冗余循环
   逐个处理输入数据不仅效率低下，还容易引发维度匹配问题，其实可以直接批量处理输入。

修正后的完整代码 #

import numpy as np

# 定义激活函数与损失函数
def relu(x):
    return np.maximum(x, 0)

def relu_prime(x):
    return np.where(x > 0, 1, 0)

def mse(y_true, y_pred):
    return np.mean(np.power(y_true - y_pred, 2))

def mse_prime(y_true, y_pred):
    return 2 * (y_pred - y_true) / y_true.size

class Dense:
    def __init__(self, feat_size, out_size):
        # He初始化，提升训练稳定性
        self.weights = (np.random.normal(0, 1, feat_size*out_size) * np.sqrt(2/feat_size)).reshape(feat_size, out_size)
        self.bias = np.random.rand(1, out_size) - 0.5

    def forward(self, input_data):
        self.input = input_data
        self.output = np.dot(self.input, self.weights) + self.bias
        return self.output

    def backward(self, output_der, lr):
        # 计算输入梯度与权重梯度
        input_der = np.dot(output_der, self.weights.T)
        # 修正权重梯度计算：保证维度与权重一致
        weight_der = np.dot(self.input.T, output_der)
        # 偏置梯度取平均，避免batch大小影响更新幅度
        bias_der = np.mean(output_der, axis=0, keepdims=True)
        
        # 更新网络参数
        self.weights -= lr * weight_der
        self.bias -= lr * bias_der
        return input_der

class ActLayer:
    def __init__(self, act, act_prime):
        self.act = act
        self.act_prime = act_prime

    def forward(self, input_data):
        self.input = input_data
        self.output = self.act(self.input)
        return self.output

    def backward(self, output_der, lr):
        return self.act_prime(self.input) * output_der

def shuffle_batch(X, y, batch_size):
    rnd_idx = np.random.permutation(len(X))
    n_batches = len(X) // batch_size
    # 使用生成器返回所有batch
    for batch_idx in np.array_split(rnd_idx, n_batches):
        yield X[batch_idx], y[batch_idx]
    # 处理剩余不足一个batch的样本
    if len(X) % batch_size != 0:
        yield X[rnd_idx[n_batches*batch_size:]], y[rnd_idx[n_batches*batch_size:]]

class Network:
    def __init__(self, loss, loss_prime):
        self.layers = []
        self.loss = loss
        self.loss_prime = loss_prime

    def add(self, layer):
        self.layers.append(layer)

    def predict(self, input_data):
        # 直接批量处理输入，避免维度问题
        layer_output = input_data
        for layer in self.layers:
            layer_output = layer.forward(layer_output)
        return layer_output

    def fit(self, X_train, y_train, epochs, lr, batch_size):
        n_samples = len(X_train)
        for epoch in range(epochs):
            total_err = 0.0
            # 遍历所有训练batch
            for X_batch, y_batch in shuffle_batch(X_train, y_train, batch_size):
                # 前向传播
                layer_output = X_batch
                for layer in self.layers:
                    layer_output = layer.forward(layer_output)
                # 累加当前batch的总损失
                batch_err = self.loss(y_batch, layer_output)
                total_err += batch_err * len(X_batch)
                
                # 反向传播更新参数
                gradient = self.loss_prime(y_batch, layer_output)
                for layer in reversed(self.layers):
                    gradient = layer.backward(gradient, lr)
            
            # 计算并打印平均损失
            avg_err = total_err / n_samples
            print(f'epoch {epoch+1}/{epochs} error={avg_err:.6f}')

# 训练与测试代码
x_train = np.array([[[0,0]], [[0,1]], [[1,0]], [[1,1]]])
y_train = np.array([[[0]], [[1]], [[1]], [[0]]])
x_train = x_train.reshape(-1, 2)
y_train = y_train.reshape(-1, 1)

# 初始化网络
net = Network(mse, mse_prime)
net.add(Dense(2, 3))
net.add(ActLayer(relu, relu_prime))
net.add(Dense(3, 1))

# 开始训练
net.fit(x_train, y_train, epochs=1000, lr=0.1, batch_size=2)

# 测试模型
out = net.predict(x_train)
print("预测结果：")
print(out)

关键修正点说明 #

1. shuffle_batch改为生成器：用yield替代return，确保遍历所有训练批次，包括最后剩余的样本。
2. 修正权重梯度计算：移除错误的reshape操作，直接通过矩阵点积得到与权重维度匹配的梯度，同时对偏置梯度取平均，保证更新幅度的稳定性。
3. 重构训练循环：正确遍历每个batch，累加总损失后计算平均误差，真实反映训练状态。
4. 简化预测逻辑：直接批量处理输入数据，既提升效率又避免维度匹配问题。

预期输出 #

运行修正后的代码，每个输入样本会对应1个预测值，类似如下结果：

预测结果：
[[0.0087]
 [0.9912]
 [0.9895]
 [0.0103]]

内容的提问来源于stack exchange，提问作者CheeseBurger