手写机器学习算法：从梯度下降到决策树的源码实现与调试指南

简介基于Python的机器学习算法设计源码包面向有Python基础、希望从代码层面理解并复现机器学习模型的开发者。包内共35个文件以33个Python源文件为核心覆盖MNIST手写数字识别、猫狗图像分类、动漫人脸生成、FCN语义分割、RNN、DCGAN、DiscoGAN、DQN、Faster R-CNN、空间变换网络等典型实现同时提供readme.txt环境配置与使用说明、.gitignore版本管理配置。算法方向涵盖图像分类、目标检测、生成对抗网络和强化学习项目内还包含模型基类、常用工具函数、路径与数学处理模块、多个测试和演示脚本便于从零训练、评估和对比不同模型的性能。压缩包约132KB体量小巧、部署方便已有368人下载学习。适合需要快速获得可运行算法示例、减少重复编码并深入理解算法细节的机器学习开发者和相关专业学生既可直接运行也可作为二次开发基础帮助构建和优化机器学习模型。1. 手写机器学习算法源码调包之外的那块拼图一个在业务里跑了三年模型的工程师大概率不会把线性回归的权重更新公式记错但让他不借助 sklearn 在半小时内写出带早停的 Logistic 回归很多人会卡在“梯度怎么组织成矩阵运算”这一步。这个标题对应的东西正是把“会用”翻译成“能写”的那道工序用 Python 从零实现常见机器学习算法并把代码组织成可读、可扩展的源码工程。它适合两类人——一类是刚学完理论、想确认自己真懂了的初学者另一类是生产环境里被黑盒模型坑过、想掌握调试能力的从业者。前者的收获是公式落地后者的收获是当损失不下降时知道该去哪一行代码里找原因而不是对着训练曲线猜谜。这里不讨论调包技巧而是把算法拆到“能手写实现”的粒度并给出工程化的代码组织方式。2. 机器学习源码工程的基础模块数据集与评估先行2.1 为什么先写数据生成器和评估指标而不是先写模型几乎所有人手写算法时犯的第一个错误是直接去实现模型主体然后随手用 sklearn 的train_test_split切数据、用accuracy_score看结果。这在验证阶段没有大问题但一旦进入调试环节问题就暴露了你无法确认模型学到的规律是数据里真实存在的还是生成数据时无意引入的伪规律。更关键的是评估指标与损失函数的关系不清晰导致你对“模型表现”的判断失真。我一般会先写两个模块一个是可控的数据生成器一个是统一的评估器。前者让你手里有“已知答案”的测试题后者让你在算法还没完全调通时能区分“模型没收敛”和“评估方式有问题”这两种截然不同的故障。这两个模块大约 100 行代码却能把后续所有算法的调试成本降低一半以上。2.2 可复现的数据生成器分类与回归的最小实现数据生成器要满足三个条件可复现、可调难度、可注入噪声。下面给出一个同时支持二分类和回归的生成器骨架它比 sklearn 的make_classification更透明因为你清楚每一条样本的真实生成过程。# data_generator.py import numpy as np def make_linear_regression(n_samples200, n_features3, noise0.1, coefNone, seed42): 生成线性回归数据y X coef 高斯噪声 rng np.random.default_rng(seed) X rng.normal(0, 1, size(n_samples, n_features)) if coef is None: coef rng.uniform(-2, 2, sizen_features) y X coef rng.normal(0, noise, sizen_samples) return X, y, coef def make_binary_classification(n_samples200, n_features4, sep1.0, seed42): 生成二分类数据两类中心点距离由 sep 控制 rng np.random.default_rng(seed) X_pos rng.normal(sep / 2, 1.0, size(n_samples // 2, n_features)) X_neg rng.normal(-sep / 2, 1.0, size(n_samples // 2, n_features)) X np.vstack([X_pos, X_neg]) y np.hstack([np.ones(n_samples // 2), np.zeros(n_samples // 2)]) # 打乱顺序避免训练时前一半全是正样本 idx rng.permutation(n_samples) return X[idx], y[idx]这里的seed参数直接透传给default_rng保证每次运行生成的数据完全一致。noise控制回归任务的信噪比sep控制分类任务的线性可分程度。调试算法初期先把noise调小或把sep调大让模型必须收敛之后再逐步加大难度检验算法的鲁棒性。2.3 指标模块不能只写准确率手写算法时输出层和评估指标常发生隐性错配——比如用 Sigmoid 输出的概率直接比较大小当作预测类别却用均方误差当评估指标。下面这个评估模块覆盖了分类和回归最常见的指标并统一返回字典方便后续在训练循环里直接打印。# metrics.py import numpy as np def accuracy(y_true, y_pred): return np.mean(y_true y_pred) def precision_recall_f1(y_true, y_pred): tp np.sum((y_true 1) (y_pred 1)) fp np.sum((y_true 0) (y_pred 1)) fn np.sum((y_true 1) (y_pred 0)) precision tp / (tp fp 1e-9) recall tp / (tp fn 1e-9) f1 2 * precision * recall / (precision recall 1e-9) return {precision: precision, recall: recall, f1: f1} def mean_squared_error(y_true, y_pred): return np.mean((y_true - y_pred) ** 2) def evaluate(y_true, y_pred, taskclassification): if task classification: return {accuracy: accuracy(y_true, y_pred), **precision_recall_f1(y_true, y_pred)} return {mse: mean_squared_error(y_true, y_pred)}加1e-9是为了防止除零但要注意如果预测结果里完全没有正样本precision 和 recall 本身就没有意义此时应该去检查分类阈值是否合理而不是信任这个数字。这个模块的价值在于所有算法共用同一套评估逻辑避免了每个模型文件里各写一份np.mean(y_pred y_test)导致的风格漂移。3. 从零实现三类核心算法线性模型、决策树与 KMeans 的源码设计3.1 算法源码的公共抽象fit 与 predict 的接口约束手写多个算法后会发现如果没有统一的基类每个模型的调用方式会逐渐走样——有的返回概率有的返回类别有的把训练历史存在不同属性名下。把这套项目当工程做的话第一步不是实现某个具体算法而是定一个极简的接口约束。# base.py from abc import ABC, abstractmethod class BaseEstimator(ABC): def __init__(self): self.is_fitted False abstractmethod def fit(self, X, y): 训练模型 abstractmethod def predict(self, X): 预测 abstractmethod def score(self, X, y): 评估模型并返回指标接口约束的价值体现在调试期你写一个通用的交叉验证脚本传入任何继承BaseEstimator的模型都能直接跑不需要为每个算法单独写一份验证代码。这里不引入 sklearn 的BaseEstimator是为了零依赖也让读者看清楚接口的本质——无非是约定好fit和predict的存在。3.2 线性回归与梯度下降手写反向传播的第一个里程碑线性回归是理解梯度下降的最佳载体。实现时有一个关键决策用解析解正规方程还是梯度下降。两者都是正确方案但既然标题强调“算法设计源码”用梯度下降能覆盖更多可复用的知识点比如学习率、迭代次数、损失记录。下面给出用矩阵运算实现的版本避免显式 for 循环带来的性能陷阱。# linear_regression.py import numpy as np class LinearRegressionGD(BaseEstimator): def __init__(self, lr0.01, n_iters1000, tol1e-6): self.lr lr self.n_iters n_iters self.tol tol self.coef_ None self.loss_history_ [] def fit(self, X, y): # 添加偏置项把 bias 统一成 weight 的一部分 X_aug np.hstack([np.ones((X.shape[0], 1)), X]) n_samples, n_features X_aug.shape self.coef_ np.zeros(n_features) for i in range(self.n_iters): y_pred X_aug self.coef_ grad (2 / n_samples) * X_aug.T (y_pred - y) self.coef_ - self.lr * grad loss np.mean((y_pred - y) ** 2) self.loss_history_.append(loss) if len(self.loss_history_) 1 and abs(self.loss_history_[-2] - loss) self.tol: break self.is_fitted True return self def predict(self, X): X_aug np.hstack([np.ones((X.shape[0], 1)), X]) return X_aug self.coef_ def score(self, X, y): return {mse: mean_squared_error(y, self.predict(X))}核心逻辑是两行grad (2 / n_samples) * X_aug.T (y_pred - y)和self.coef_ - self.lr * grad。前者的含义是误差向量与特征矩阵的转置相乘得到每个维度的梯度方向后者是参数沿负梯度方向更新。这里不用 L2 正则是为了让第一次接触源码的人不被额外项干扰。tol控制早停当相邻两轮损失变化小于阈值时停止避免无效迭代。3.3 决策树信息增益与递归切分的纯 Python 实现决策树是理解“非参数模型”的最佳样本也是源码里最能体现“算法设计”的部分——树本身的数据结构、递归切分逻辑、特征选择标准三者缺一不可。实现时用 CART 回归树的框架分裂标准选择均方误差下降量这样分类和回归任务共用一套代码。# decision_tree.py import numpy as np class Node: def __init__(self, feature_idxNone, thresholdNone, leftNone, rightNone, valueNone): self.feature_idx feature_idx # 切分特征下标 self.threshold threshold # 切分阈值 self.left left self.right right self.value value # 叶节点的预测值 class DecisionTreeRegressor(BaseEstimator): def __init__(self, max_depth5, min_samples_split2): self.max_depth max_depth self.min_samples_split min_samples_split self.root_ None def fit(self, X, y): self.root_ self._build_tree(X, y, depth0) self.is_fitted True return self def _build_tree(self, X, y, depth): n_samples, n_features X.shape # 停止条件达到最大深度、样本太少、目标值唯一 if depth self.max_depth or n_samples self.min_samples_split or len(np.unique(y)) 1: return Node(valuenp.mean(y)) best_feat, best_thr, best_loss None, None, float(inf) cur_loss np.var(y) * n_samples for f in range(n_features): thresholds np.unique(X[:, f]) # 对连续特征取每个唯一值作候选阈值 for t in thresholds: left_mask X[:, f] t if np.sum(left_mask) 0 or np.sum(~left_mask) 0: continue loss (np.var(y[left_mask]) * np.sum(left_mask) np.var(y[~left_mask]) * np.sum(~left_mask)) if loss best_loss: best_loss loss best_feat f best_thr t if best_feat is None: return Node(valuenp.mean(y)) left_mask X[:, best_feat] best_thr left self._build_tree(X[left_mask], y[left_mask], depth 1) right self._build_tree(X[~left_mask], y[~left_mask], depth 1) return Node(feature_idxbest_feat, thresholdbest_thr, leftleft, rightright) def predict(self, X): return np.array([self._traverse(x, self.root_) for x in X]) def _traverse(self, x, node): if node.value is not None: return node.value if x[node.feature_idx] node.threshold: return self._traverse(x, node.left) return self._traverse(x, node.right)这段代码的关键设计有两个。第一叶节点用value存预测值内部节点用feature_idx和threshold存切分规则从结构上区分了两类节点。第二特征搜索遍历的是np.unique(X[:, f])而不是所有样本值这避免了对同一取值的重复计算。性能上纯 Python 版本的决策树在大数据集上比 sklearn 慢很多但作为源码学习它的递归结构更直观。max_depth是最需要调的参数——太浅欠拟合太深过拟合且树变得极不稳定。3.4 KMeans从随机初始化到收敛判断KMeans 和前面两个算法的最大区别是没有标签参与训练因此它的fit接口虽然形式上一样内部逻辑完全不同——不需要计算梯度只需要交替执行“分配中心”和“更新中心”两步。实现的难点在于初始化策略和空簇处理。# kmeans.py import numpy as np class KMeans(BaseEstimator): def __init__(self, n_clusters3, max_iters100, initkmeans, seed42): self.n_clusters n_clusters self.max_iters max_iters self.init init self.seed seed self.centroids_ None self.labels_ None def fit(self, X, yNone): rng np.random.default_rng(self.seed) n_samples X.shape[0] if self.init kmeans: centroids [X[rng.integers(n_samples)]] for _ in range(1, self.n_clusters): dist np.min([np.linalg.norm(X - c, axis1) ** 2 for c in centroids], axis0) probs dist / np.sum(dist) centroids.append(X[rng.choice(n_samples, pprobs)]) else: idx rng.choice(n_samples, self.n_clusters, replaceFalse) centroids X[idx] self.centroids_ np.array(centroids) for _ in range(self.max_iters): # 分配步骤每个样本归入最近的中心 labels self._assign(X) # 更新步骤每个簇的中心取均值 new_centroids [] for k in range(self.n_clusters): cluster_points X[labels k] if len(cluster_points) 0: # 空簇重新随机初始化一个点避免中心消失 new_centroids.append(X[rng.integers(n_samples)]) else: new_centroids.append(np.mean(cluster_points, axis0)) new_centroids np.array(new_centroids) if np.allclose(self.centroids_, new_centroids, atol1e-6): break self.centroids_ new_centroids self.labels_ self._assign(X) self.is_fitted True return self def _assign(self, X): # 返回每个样本距离最近的中心下标 dists np.linalg.norm(X[:, None, :] - self.centroids_[None, :, :], axis2) return np.argmin(dists, axis1)_assign方法里的广播运算值得单独说明X[:, None, :]把 X 扩展成形状(n_samples, 1, n_features)centroids_[None, :, :]扩展成(1, n_clusters, n_features)两者相减后取范数一步得到完整的距离矩阵。这是手写算法时最常用的加速技巧。空簇处理容易被忽略——如果某个中心在分配后没有任何样本它会在下一步变成 NaN 并污染所有距离计算这里直接重新赋值一个随机样本虽然简单但非常有效。4. 算法源码的正确性验证对拍、梯度检查与可视化调试4.1 对拍用 sklearn 作为参考实现但不要迷信结果手写算法的第一道验证工序是对拍differential testing即用相同的数据分别跑自己写的模型和 sklearn 的参考实现逐项比较预测结果。对拍的严谨之处在于不是比较谁的结果更好而是比较差距是否在合理范围内。手写线性回归和 sklearn 的LinearRegression默认对不上因为 sklearn 默认用解析解而手写的是梯度下降解两者在存在特征共线性时会有细微权重差异。正确的对拍方式是给 sklearn 也加上solversgd或者放宽比较的精度。# 对拍脚本的目录组织方式 ml_from_scratch/ ├── base.py # 抽象基类 ├── data_generator.py # 数据生成器 ├── metrics.py # 评估指标 ├── linear_regression.py ├── decision_tree.py ├── kmeans.py └── benchmark.py # 对拍与基准测试脚本对拍脚本的核心逻辑是用一模一样的数据调用两个实现然后比较predict结果的差异程度。对分类器比较准确率对回归器比较预测值的相关系数和最大绝对误差对聚类模型比较调整兰德指数而非逐点标签因为 KMeans 的标签编号是随意的。4.2 梯度检查用数值差分验证反向传播写没写错深度学习框架里有梯度检查gradient checking手写算法同样需要。梯度下降类算法的绝大多数 bug 出现在损失函数对参数的偏导计算上——矩阵维度不对、转置漏掉、符号写反这些问题从最终预测结果很难直接发现因为模型可能仍然收敛只是收敛速度很慢。数值差分法的数学基础是导数的定义对某个参数theta_i加一个极小量epsilon损失变化量除以epsilon就是梯度的近似值。# gradient_check.py import numpy as np def numerical_gradient(loss_fn, theta, epsilon1e-5): loss_fn: 参数为 theta 的函数返回标量损失 grad np.zeros_like(theta) for i in range(len(theta)): theta_plus theta.copy() theta_minus theta.copy() theta_plus[i] epsilon theta_minus[i] - epsilon grad[i] (loss_fn(theta_plus) - loss_fn(theta_minus)) / (2 * epsilon) return grad # 使用示例假设 theta 是线性回归的系数向量 # analytic_grad (2 / n_samples) * X_aug.T (X_aug theta - y) # numeric_grad numerical_gradient(lambda t: np.mean((X_aug t - y) ** 2), theta) # print(np.max(np.abs(analytic_grad - numeric_grad))) # 小于 1e-6 即通过梯度检查的最佳实践是抽检而非全检——高维参数全部数值差分会很慢通常随机选 3 到 5 个维度做检查即可。如果数值梯度和解析梯度的差异在1e-6量级说明反向传播写对了如果差异在1e-2量级说明有符号错误或漏项。注意epsilon不能选得太小否则浮点精度误差会淹没真实的梯度信号。这套方法在实现逻辑回归、Softmax 分类器或简单神经网络时同样有效。4.3 训练过程可视化损失曲线比终态指标透露更多评估一个手写算法的调试手段里最容易被低估的是损失曲线的形态。训练完成后打印最终评估指标只能回答“模型好不好”而观察损失曲线能回答“为什么不收敛”。损失曲线基本能归纳为三种病一是曲线持续下降但斜率缓慢说明学习率太小模型收敛极慢典型表现是训练几百轮损失仍在下降二是曲线先降后震荡并持续上升说明学习率太大参数在最优解附近反复横跳甚至发散三是曲线在前几轮下降后平台期不再变化但模型预测结果依然很差这种不是学习率问题而是特征没有标准化梯度方向被数值范围大的特征主导。# 可视化训练历史的辅助函数 import matplotlib.pyplot as plt def plot_loss_history(model, titleTraining Loss): plt.figure(figsize(8, 4)) plt.plot(model.loss_history_, lw2) plt.xlabel(Iteration) plt.ylabel(MSE) plt.title(title) plt.grid(True, alpha0.3) plt.tight_layout() plt.savefig(loss_curve.png, dpi150)把plot_loss_history保存为独立工具函数而不是在每个算法文件里重复写绘图代码这样可以保证整个项目里的损失图风格统一也方便在调试时快速切换不同模型对比曲线。注意KMeans 没有损失曲线它的等价物是簇内平方和inertia可以在每次迭代后记录一次观察是否单调下降——如果出现上升说明实现里有 bug。5. 源码的可读性与验证脚本手写算法工程化的最后一步当算法正确性验证通过后源码设计的重心转向可维护性。手写算法的代码容易陷入两种极端一是把所有逻辑塞进一个巨型函数外人看不懂二是过度抽象每个方法只有两三行阅读代码需要不停跳转。这个项目的源码要想对后来者包括三个月后的自己友好需要在中间位置找到平衡点。一个值得推荐的实践是为每个算法配一个独立的自测脚本脚本里包含数据生成、模型训练、对拍比较、结果断言四个步骤。这样任何人拿到源码仓库后无需阅读文档就能确认算法实现没有回归。以决策树为例自测脚本长这样# test_decision_tree.py from data_generator import make_linear_regression from decision_tree import DecisionTreeRegressor from metrics import mean_squared_error, evaluate import numpy as np X, y, _ make_linear_regression(n_samples300, n_features4, noise0.3, seed0) X_train, X_test, y_train, y_test X[:240], X[240:], y[:240], y[240:] dt DecisionTreeRegressor(max_depth5, min_samples_split4) dt.fit(X_train, y_train) y_pred dt.predict(X_test) mse mean_squared_error(y_test, y_pred) print(fTest MSE: {mse:.4f}) # 对拍决策树对噪声的拟合能力较弱MSE 应接近噪声方差 0.09 assert mse 0.15, fMSE {mse} too high, decision tree may be overfitting脚本末尾的assert是防线不是摆设。加了断言之后任何对实现的无意修改——比如把np.var写成np.std、把左子树和右子树的掩码条件写反——都会立刻让脚本报错不需要人肉读代码找差异。这套自测脚本的组织方式同样适用于分类模型分类任务可以断言准确率下限聚类任务可以断言不同seed下结果的稳定性。另一个容易被忽略的细节是随机种子管理。手写算法的源码仓库里所有涉及随机性的模块——数据生成、KMeans 初始化、梯度下降的样本洗牌——都应该支持传入seed参数。这不是为了让结果“好看”而是为了在调试时做到完全可复现同样的代码、同样的输入、同样的种子必须得到同样的输出。如果随机性来源过多且不可控你永远无法判断一个改动是修复了 bug 还是撞上了好运。项目中所有算法的基类也应用相同的设计原则fit方法返回self方便链式调用内部不打印训练过程把日志交给调用方处理。最后提一个验证方法层面的技巧算法源码写完不要急着上真实业务数据先用合成数据把每个算法的行为模式摸清楚。线性回归在无噪声数据上的训练误差应该能降到接近零决策树在无噪声数据上调大max_depth可以完全拟合训练集KMeans 在生成的高斯混合数据上恢复出的簇中心应该和原始中心相差不到 0.2。每验证一项就在代码注释里记一笔当时的参数和观察到的现象。这些注释将来就是整份源码最好的文档。本文还有配套的精品资源点击获取