2026/9/15 4:08:24
FEATURED · 精选文章 
简介这套基于Python的机器学习算法设计源码主要面向机器学习和深度学习开发者用来降低算法实现与落地成本。资源包共包含三十五个文件体积仅一百三十二KB非常轻量。其中三十三个为Python源代码文件覆盖数据预处理、特征工程、模型构建和评估等环节既包含CNN、RNN、自编码器、生成对抗网络等经典深度模型也包含DDPG、DQN等强化学习实现同时提供工具函数与测试脚本方便对照运行和二次开发。另外压缩包中还含有一个readme说明文档介绍环境配置与基本使用方法以及一个.gitignore文件帮助维护项目整洁。目前已有三百六十八人学习浏览既适合作为算法学习参考也可作为实际项目开发的起点让开发者在现有实现基础上更专注于算法优化和业务创新。1. 把机器学习算法写成 Python 源码本质是在设计什么把sklearn里一行model.fit(X, y)展开成上百次迭代、梯度计算与收敛判断这就是「基于 Python 实现的机器学习算法设计源码」在回答的问题用 Python 与 numpy 从零实现数据结构、损失函数、优化过程与预测逻辑而不是对库做二次封装。这类源码解决两类真实需求把模型移植到不能安装 sklearn 的嵌入式环境以及为面试手撕算法、期末备考做系统性准备。适合有一年以上 Python 经验、想掌控模型细节的开发者也适合正在系统学习机器学习理论的学生。正文按算法仓库的常见结构展开先定数据接口与评价契约再实现线性回归、逻辑回归与 K-Means最后用对照实验验收源码。2. 算法设计的第一层数据结构、数据切分与评价函数任何机器学习算法在写梯度之前都要先面对三个公共问题输入长什么样、数据怎么切、结果怎么评。这一层如果不统一后面每个算法都要重复处理形状校验和评价逻辑代码量翻倍不说还容易在对比实验时因为口径不一致得出错误结论。2.1 用 numpy 定义统一的数据契约我一般约定 X 是形状为(n_samples, n_features)的二维 float64 数组y 是(n_samples,)的一维数组分类时取值是从 0 开始的整数回归时是浮点数。这个约定与 sklearn 一致后续实现、对照、测试都省事。把校验逻辑收敛到一个极简的 Dataset 类里import numpy as np from typing import Tuple class Dataset: 极简数据集封装统一 X/y 的类型与形状检查。 def __init__(self, X: np.ndarray, y: np.ndarray): X np.asarray(X, dtypenp.float64) y np.asarray(y) if X.ndim ! 2: raise ValueError(fX 必须是二维数组当前 shape{X.shape}) if y.shape[0] ! X.shape[0]: raise ValueError(X 与 y 的样本数不一致) self.X X self.y y self.n_samples, self.n_features X.shape def __len__(self) - int: return self.n_samples def __getitem__(self, idx) - Tuple[np.ndarray, np.ndarray]: return self.X[idx], self.y[idx]np.asarray负责把 list 静默转成数组dtypenp.float64是数值算法的默认精度低于它会在梯度累加时出现精度漂移。两个形状校验要放在最前面因为后续矩阵乘法、广播全都依赖这两个维度。__getitem__同时支持整数索引和切片训练循环里打乱顺序时执行dataset[rng.permutation(n)]就能拿到重排后的批量不需要额外维护索引数组。2.2 标准化与切分先堵住信息泄露的坑两个数据操作是所有算法共用的应该写成独立函数而不是塞进某个算法内部。第一个是按列标准化第二个是随机切分。它们的正确性直接影响评估结果的可信度def standardize(X: np.ndarray, mean: np.ndarray None, std: np.ndarray None): 按列标准化。测试集必须复用训练集的 mean/std。 mean X.mean(axis0) if mean is None else mean std X.std(axis0) if std is None else std std[std 1e-12] 1.0 # 零方差列直接置 1避免除零 return (X - mean) / std, mean, std def train_test_split(X: np.ndarray, y: np.ndarray, test_size: float 0.2, seed: int 42): 按比例随机切分返回 X_train, X_test, y_train, y_test。 rng np.random.default_rng(seed) idx rng.permutation(X.shape[0]) n_test int(X.shape[0] * test_size) test_idx, train_idx idx[:n_test], idx[n_test:] return X[train_idx], X[test_idx], y[train_idx], y[test_idx]这里最容易犯的错有三个我都写进过自己的踩坑笔记里操作常见错误后果标准化测试集用自身的 mean/std信息泄露评估指标虚高切分切分前不打乱数据类别分布偏移训练集缺类调参用同一份测试集反复调参测试集被「记住」泛化估计失效train_test_split用的np.random.default_rng(seed)是 NumPy 1.17 之后的 Generator 接口比老的np.random.permutation更可复现。注意返回顺序是先 train 后 test我见过不止一个项目把这两个返回值接反导致模型在测试集上「训练」。提示标准化时把测试集接到训练集的 mean/std 上是评估流程里最容易被忽略的信息泄露入口。做特征工程时同样如此先 fit 再 transform。2.3 损失函数与评价指标分类器和回归器共用一套接口损失函数是算法设计里的核心契约它必须是可微的并且作为优化目标要跟最终评价指标趋势一致。回归器默认用均方误差分类器默认用交叉熵评价指标分别用 R² 和 accuracy。最小实现如下def mse_loss(y_true: np.ndarray, y_pred: np.ndarray) - float: return np.mean((y_true - y_pred) ** 2) def cross_entropy_loss(y_true: np.ndarray, y_prob: np.ndarray, eps: float 1e-12) - float: 二分类交叉熵y_prob 是预测为正类的概率。 y_prob np.clip(y_prob, eps, 1 - eps) return -np.mean(y_true * np.log(y_prob) (1 - y_true) * np.log(1 - y_prob)) def accuracy(y_true: np.ndarray, y_pred: np.ndarray) - float: return np.mean(y_true y_pred)交叉熵里的np.clip是为了防止log(0)产生 naneps 取 1e-12 左右太小会数值不稳太大会让损失计算失真。交叉熵与 accuracy 并不是严格单调关系但交叉熵在错误分类时梯度幅度更大、收敛更快这就是分类器普遍用交叉熵做损失、用 accuracy 做评价的原因。这一层接口定下来之后第 3 章所有算法都可以直接复用这就是算法设计里「契约先行」的实际收益。3. 从零实现三类机器学习算法线性回归、逻辑回归与 K-Means这一章是源码设计的核心动作把纸面上的公式翻译成能跑的 Python。三个算法分别代表三类问题连续值回归、二分类、无监督聚类。它们的共同点是都依赖上一章的数据契约代码结构也刻意保持对齐方便对照差异。3.1 线性回归闭式解与梯度下降两条路线对比线性回归是最适合起步的算法因为它同时存在解析解和迭代解两种实现方式。解析解适合特征维度低几千维以内的场景迭代解能平滑过渡到逻辑回归和神经网络。两种路线都实现一下对比着看理解更深class LinearRegression: 支持闭式解与梯度下降两种 fit 路线的线性回归。 def __init__(self, method: str gd, lr: float 0.01, epochs: int 1000): self.method method self.lr lr self.epochs epochs self.w, self.b None, None def fit(self, X: np.ndarray, y: np.ndarray): n, d X.shape if self.method closed: X_aug np.hstack([np.ones((n, 1)), X]) # 偏置并入权重矩阵 w_aug np.linalg.pinv(X_aug) y # 伪逆数值更稳 self.b, self.w w_aug[0], w_aug[1:] else: self.w, self.b np.zeros(d), 0.0 for _ in range(self.epochs): pred X self.w self.b grad_w (2 / n) * (X.T (pred - y)) grad_b (2 / n) * np.sum(pred - y) self.w - self.lr * grad_w self.b - self.lr * grad_b return self def predict(self, X: np.ndarray) - np.ndarray: return X self.w self.b闭式解用np.linalg.pinv(X_aug)直接对增广矩阵做伪逆比先算X^T X再求逆的条件数更好当特征存在共线性时后者会得到数值上爆炸的解伪逆仍然返回最小范数解。梯度下降版本里X.T (pred - y)把「所有样本的梯度求和再平均」压缩成一次矩阵乘法比 for 循环快两个数量级这是 numpy 向量化编程的核心习惯。学习率lr是迭代路线最敏感的参数太大梯度发散、太小收敛缓慢具体调法在第 4 章优化器部分展开。3.2 逻辑回归把分类器的线性输出变成概率逻辑回归在源码层面只比线性回归多两步加一个 sigmoid 激活损失从 MSE 换成交叉熵。但它的梯度形式出奇地简洁推导结果恰好是X.T (prob - y) / n实现如下def sigmoid(z: np.ndarray) - np.ndarray: z np.clip(z, -500, 500) # 防止 exp 溢出为 inf return 1.0 / (1.0 np.exp(-z)) class LogisticRegression: 二分类逻辑回归支持 predict_proba 输出原始概率。 def __init__(self, lr: float 0.1, epochs: int 500): self.lr, self.epochs lr, epochs self.w, self.b None, None def fit(self, X: np.ndarray, y: np.ndarray): n, d X.shape self.w, self.b np.zeros(d), 0.0 for _ in range(self.epochs): prob sigmoid(X self.w self.b) grad_w (1 / n) * (X.T (prob - y)) grad_b (1 / n) * np.sum(prob - y) self.w - self.lr * grad_w self.b - self.lr * grad_b return self def predict_proba(self, X: np.ndarray) - np.ndarray: return sigmoid(X self.w self.b) def predict(self, X: np.ndarray, threshold: float 0.5) - np.ndarray: return (self.predict_proba(X) threshold).astype(int)sigmoid里的 clip 是必须的np.exp(-z)在 z 小于 -745 左右会溢出成 infclip 到 [-500, 500] 既保证数值安全又不影响正常区间的梯度精度。threshold是分类器设计里容易被忽略的决策点默认 0.5 只在正负样本均衡时合理样本不平衡时应拿验证集遍历阈值选 F1 最大或业务召回率达标的点。逻辑回归与线性回归源码高度相似不是巧合它俩同属广义线性模型区别只在 link function这是设计源码时值得刻意保留的结构。3.3 K-Means无监督聚类里的迭代与收敛判断K-Means 没有梯度它的迭代是「指派-更新」两板斧先给每个样本指派最近的中心点再重新计算每个簇的均值。看点是距离计算的向量化与收敛判断的写法class KMeans: 带 k-means 初始化的聚类实现。 def __init__(self, n_clusters: int 3, max_iter: int 300, tol: float 1e-4, seed: int 0): self.k, self.max_iter, self.tol, self.seed n_clusters, max_iter, tol, seed self.centroids, self.labels_ None, None def fit(self, X: np.ndarray): rng np.random.default_rng(self.seed) n X.shape[0] centroids [X[rng.integers(n)]] for _ in range(1, self.k): # k-means 初始化 dist np.min([np.sum((X - c) ** 2, axis1) for c in centroids], axis0) centroids.append(X[rng.choice(n, pdist / dist.sum())]) self.centroids np.array(centroids) for _ in range(self.max_iter): labels self._assign(X) new_centroids np.array([X[labels j].mean(axis0) for j in range(self.k)]) shift np.sum((new_centroids - self.centroids) ** 2) self.centroids, self.labels_ new_centroids, labels if shift self.tol: break return self def _assign(self, X: np.ndarray) - np.ndarray: # 展开公式 a^2 - 2ab b^2 把距离计算转成矩阵乘法 dists (np.sum(X ** 2, axis1)[:, None] - 2 * (X self.centroids.T) np.sum(self.centroids ** 2, axis1)[None, :]) return np.argmin(dists, axis1)_assign里的展开公式避免了三层 for 循环把 O(n·k·d) 的逐点距离换成矩阵乘法在 10 万样本、100 维特征时能快一到两个数量级。k-means初始化让后续中心按距离平方加权采样显著降低落入局部最优的概率。如果某个簇在迭代中没有分到样本mean会得到 nan工业实现通常重新初始化空簇中心演示版本可以加一行判断跳过。收敛判断用中心点位移平方和shift而不是硬跑满max_itertol取 1e-4 左右即可。三个算法放一起看会发现一个共同的设计模式fit负责更新状态predict/predict_proba只读不写。把这个约定保持到后面的每一个模型里训练与推理的边界就清晰了。算法模型状态收敛条件主要瓶颈线性回归闭式解w、b无迭代伪逆计算约 O(n·d²)线性回归 / 逻辑回归梯度w、b跑满 epochs学习率敏感K-Meanscentroids、labels_中心位移 tol距离矩阵内存占用4. 梯度反向传播与优化器把训练循环做成可插拔的 Python 模块当模型从单层走向多层梯度不再能手推闭式解这时需要反向传播和通用优化器。这一章把「损失-梯度-参数更新」三件事拆开让优化器可以插到任何模型上。4.1 梯度检查写反向传播之前先自证的测试方法反向传播是机器学习源码里 bug 最高发的区域一个符号写反、一个维度转置错模型表现为「能跑但 loss 不降」。在写任何反向传播之前先做数值梯度检查用差分近似验证解析梯度def gradient_check(model, X: np.ndarray, y: np.ndarray, eps: float 1e-6) - None: 数值梯度 vs 解析梯度相对误差超过 1e-6 说明反向传播有 bug。 grads model.compute_gradients(X, y) for name, g in grads.items(): param model.params[name] numeric np.zeros_like(param) for idx in np.ndindex(param.shape): orig param[idx] param[idx] orig eps loss_plus model.compute_loss(X, y) param[idx] orig - eps loss_minus model.compute_loss(X, y) param[idx] orig numeric[idx] (loss_plus - loss_minus) / (2 * eps) err np.abs(g - numeric) / (np.abs(g) np.abs(numeric) 1e-12) assert err.max() 1e-6, f{name} 梯度检查失败: {err.max():.2e}数值梯度用中心差分(f(xeps) - f(x-eps)) / 2eps比单侧差分精度高一个数量级。相对误差分母加1e-12是为了防止梯度恰好为零时除零。这个测试要求模型暴露compute_loss与compute_gradients两个方法等于强制每个模型实现可测试的接口设计上反而更干净。梯度检查只适用于基于梯度的模型决策树的剪枝算法、SVM 的 SMO 这类结构化算法没有连续梯度调试时主要靠算法流程图和边界用例推演那是另一套方法论。4.2 优化器统一接口SGD、Momentum、Adam 的结构对比优化器只做一件事根据梯度更新参数。把这件事抽象成统一接口模型代码里就不用再手写w - lr * gradclass SGD: 带动量的随机梯度下降。 def __init__(self, lr: float 0.01, momentum: float 0.9): self.lr, self.momentum lr, momentum self.velocity {} def step(self, params: dict, grads: dict) - None: for name, p in params.items(): v self.velocity.setdefault(name, np.zeros_like(p)) self.velocity[name] self.momentum * v - self.lr * grads[name] params[name] self.velocity[name] class Adam: 自适应矩估计优化器对学习率不那么敏感。 def __init__(self, lr: float 1e-3, beta1: float 0.9, beta2: float 0.999, eps: float 1e-8): self.lr, self.beta1, self.beta2, self.eps lr, beta1, beta2, eps self.m, self.v, self.t {}, {}, 0 def step(self, params: dict, grads: dict) - None: self.t 1 for name, p in params.items(): m self.m.setdefault(name, np.zeros_like(p)) v self.v.setdefault(name, np.zeros_like(p)) self.m[name] self.beta1 * m (1 - self.beta1) * grads[name] self.v[name] self.beta2 * v (1 - self.beta2) * grads[name] ** 2 m_hat self.m[name] / (1 - self.beta1 ** self.t) v_hat self.v[name] / (1 - self.beta2 ** self.t) params[name] - self.lr * m_hat / (np.sqrt(v_hat) self.eps)两个优化器共享同一个step(params, grads)签名模型里只要持有self.params字典和self.compute_gradients就能在不动模型代码的前提下替换优化器。Adam 里的偏差校正m / (1 - beta^t)只在训练初期有作用前 20 步内把一阶矩从零初始化的偏差拉回来。选型时可以参考这张表优化器适用场景默认 lr踩坑点SGD数据量小、特征已标准化0.01~0.1对 lr 极其敏感SGD Momentum层数较多的模型0.01动量从 0.9 起步Adam大多数情况的首选0.001泛化有时略差于调好的 SGD4.3 训练循环骨架与学习率衰减有了优化器接口训练循环可以统一成一个函数这也是整个算法源码仓库的主入口。常见做法是用一个train函数接收模型、优化器、数据与轮数内部完成「前向-算损失-反向-更新」四步def train(model, optimizer, dataset, epochs: int 100, batch_size: int 32, lr_decay: float 0.99): rng np.random.default_rng(0) for epoch in range(epochs): idx rng.permutation(len(dataset)) epoch_loss 0.0 for i in range(0, len(dataset), batch_size): batch_idx idx[i:i batch_size] X_b, y_b dataset[batch_idx] loss, grads model.compute_loss_and_grads(X_b, y_b) optimizer.step(model.params, grads) epoch_loss loss * len(batch_idx) optimizer.lr * lr_decay # 每轮衰减学习率 print(fepoch {epoch:3d} loss {epoch_loss / len(dataset):.6f})batch_size决定每次更新的梯度噪声32 是收敛速度与稳定性的常见折中。lr_decay每轮把学习率乘一个小于 1 的系数让模型在训练后期用更小的步长精调权重避免在最优解附近震荡。如果发现 loss 曲线下降后反弹优先检查lr_decay是否设置、batch_size是否过小而不是急着换模型结构。注意optimizer.lr * lr_decay这一步同步修改了优化器内部状态。因为调度逻辑放在 train 里而不是优化器里换用任何实现了step(params, grads)的优化器训练循环都不用改。5. 对照实验与可视化验证机器学习算法源码的正确性写完了不代表写对了。机器学习源码的验证有三板斧跟成熟库对照、看可视化曲线、做单元测试。5.1 与 sklearn 对照同一份数据、同一个指标最直接的验收方式是用公开数据集同时跑自己的实现和 sklearn 的标准实现指标差距收敛到可接受范围即视为正确。对线性回归和逻辑回归权重向量与 sklearn 的对齐程度可以到小数点后 24 位取决于收敛条件K-Means 因为初始化随机看的是聚类结果的轮廓系数而不是完全一致的标签。from sklearn.datasets import load_breast_cancer from sklearn.model_selection import train_test_split as sk_split from sklearn.linear_model import LogisticRegression as SkLR data load_breast_cancer() X, y data.data, data.target X (X - X.mean(axis0)) / (X.std(axis0) 1e-12) X_tr, X_te, y_tr, y_te sk_split(X, y, test_size0.2, random_state42) my_lr LogisticRegression(lr0.05, epochs1000).fit(X_tr, y_tr) sk_lr SkLR(max_iter1000).fit(X_tr, y_tr) print(my acc: %.4f % accuracy(y_te, my_lr.predict(X_te))) print(sk acc: %.4f % accuracy(y_te, sk_lr.predict(X_te))) print(my w: , np.round(my_lr.w, 4)) print(sk w: , np.round(sk_lr.coef_[0], 4))这份脚本同时验证了分类器效果与权重方向两个层面。如果 accuracy 接近但权重符号相反说明数据没标准化或者标签编码顺序不一致如果权重数值差异大优先怀疑学习率衰减设置与收敛轮数。对照实验的目的不是追求和 sklearn 完全一致而是把「数值正确性」从「结果好坏」里剥离出来。5.2 三张必看的图损失曲线、决策边界与收敛轨迹数值对了还要看过程。我最常画的三张图第一张是训练损失曲线正常形状是单调下降后趋平出现锯齿说明 batch_size 过小或学习率偏大第二张是二分类的决策边界叠加在数据散点图上能直观看到分类器是不是被离群点带偏第三张是 K-Means 的簇心移动轨迹用透明度画出每轮迭代的中心位置能快速判断初始化是否落入局部最优。import matplotlib.pyplot as plt def plot_loss_curve(losses: list, title: str training loss): plt.figure(figsize(6, 4)) plt.plot(losses, lw2) plt.xlabel(epoch) plt.ylabel(loss) plt.title(title) plt.yscale(log) # 对数坐标放大小幅波动 plt.grid(alpha0.3) plt.show()损失曲线用plt.yscale(log)画对数坐标是因为 MSE 和交叉熵在训练后期下降幅度往往跨数量级线性坐标下后半段看起来像一条直线看不出收敛细节。可视化不是最终验收手段但它是定位 bug 阶段最快的手段先把「过程看起来对」做到再谈上线指标。5.3 把「源码正确」固化成可复现的实验记录最后给所有跟着这套流程写源码的人一个习惯每个算法的实验都用一个固定脚本入口复现记录三个字段——数据版本、超参数、验收指标。我自己的目录组织是每个算法一个目录内部放model.py核心实现、train.py训练循环、test_model.py梯度检查与对照实验三个文件的边界严格对应前面章节的划分。跑完一次实验把git rev-parse --short HEAD的输出、超参 json 和验证集指标追加进results.md下次改动后重跑一遍diff 出来的数字就是这次提交的验收结论。本文还有配套的精品资源点击获取
从零实现机器学习算法：Python源码设计与工程实践
简介这套基于Python的机器学习算法设计源码主要面向机器学习和深度学习开发者用来降低算法实现与落地成本。资源包共包含三十五个文件体积仅一百三十二KB非常轻量。其中三十三个为Python源代码文件覆盖数据预处理、特征工程、模型构建和评估等环节既包含CNN、RNN、自编码器、生成对抗网络等经典深度模型也包含DDPG、DQN等强化学习实现同时提供工具函数与测试脚本方便对照运行和二次开发。另外压缩包中还含有一个readme说明文档介绍环境配置与基本使用方法以及一个.gitignore文件帮助维护项目整洁。目前已有三百六十八人学习浏览既适合作为算法学习参考也可作为实际项目开发的起点让开发者在现有实现基础上更专注于算法优化和业务创新。1. 把机器学习算法写成 Python 源码本质是在设计什么把sklearn里一行model.fit(X, y)展开成上百次迭代、梯度计算与收敛判断这就是「基于 Python 实现的机器学习算法设计源码」在回答的问题用 Python 与 numpy 从零实现数据结构、损失函数、优化过程与预测逻辑而不是对库做二次封装。这类源码解决两类真实需求把模型移植到不能安装 sklearn 的嵌入式环境以及为面试手撕算法、期末备考做系统性准备。适合有一年以上 Python 经验、想掌控模型细节的开发者也适合正在系统学习机器学习理论的学生。正文按算法仓库的常见结构展开先定数据接口与评价契约再实现线性回归、逻辑回归与 K-Means最后用对照实验验收源码。2. 算法设计的第一层数据结构、数据切分与评价函数任何机器学习算法在写梯度之前都要先面对三个公共问题输入长什么样、数据怎么切、结果怎么评。这一层如果不统一后面每个算法都要重复处理形状校验和评价逻辑代码量翻倍不说还容易在对比实验时因为口径不一致得出错误结论。2.1 用 numpy 定义统一的数据契约我一般约定 X 是形状为(n_samples, n_features)的二维 float64 数组y 是(n_samples,)的一维数组分类时取值是从 0 开始的整数回归时是浮点数。这个约定与 sklearn 一致后续实现、对照、测试都省事。把校验逻辑收敛到一个极简的 Dataset 类里import numpy as np from typing import Tuple class Dataset: 极简数据集封装统一 X/y 的类型与形状检查。 def __init__(self, X: np.ndarray, y: np.ndarray): X np.asarray(X, dtypenp.float64) y np.asarray(y) if X.ndim ! 2: raise ValueError(fX 必须是二维数组当前 shape{X.shape}) if y.shape[0] ! X.shape[0]: raise ValueError(X 与 y 的样本数不一致) self.X X self.y y self.n_samples, self.n_features X.shape def __len__(self) - int: return self.n_samples def __getitem__(self, idx) - Tuple[np.ndarray, np.ndarray]: return self.X[idx], self.y[idx]np.asarray负责把 list 静默转成数组dtypenp.float64是数值算法的默认精度低于它会在梯度累加时出现精度漂移。两个形状校验要放在最前面因为后续矩阵乘法、广播全都依赖这两个维度。__getitem__同时支持整数索引和切片训练循环里打乱顺序时执行dataset[rng.permutation(n)]就能拿到重排后的批量不需要额外维护索引数组。2.2 标准化与切分先堵住信息泄露的坑两个数据操作是所有算法共用的应该写成独立函数而不是塞进某个算法内部。第一个是按列标准化第二个是随机切分。它们的正确性直接影响评估结果的可信度def standardize(X: np.ndarray, mean: np.ndarray None, std: np.ndarray None): 按列标准化。测试集必须复用训练集的 mean/std。 mean X.mean(axis0) if mean is None else mean std X.std(axis0) if std is None else std std[std 1e-12] 1.0 # 零方差列直接置 1避免除零 return (X - mean) / std, mean, std def train_test_split(X: np.ndarray, y: np.ndarray, test_size: float 0.2, seed: int 42): 按比例随机切分返回 X_train, X_test, y_train, y_test。 rng np.random.default_rng(seed) idx rng.permutation(X.shape[0]) n_test int(X.shape[0] * test_size) test_idx, train_idx idx[:n_test], idx[n_test:] return X[train_idx], X[test_idx], y[train_idx], y[test_idx]这里最容易犯的错有三个我都写进过自己的踩坑笔记里操作常见错误后果标准化测试集用自身的 mean/std信息泄露评估指标虚高切分切分前不打乱数据类别分布偏移训练集缺类调参用同一份测试集反复调参测试集被「记住」泛化估计失效train_test_split用的np.random.default_rng(seed)是 NumPy 1.17 之后的 Generator 接口比老的np.random.permutation更可复现。注意返回顺序是先 train 后 test我见过不止一个项目把这两个返回值接反导致模型在测试集上「训练」。提示标准化时把测试集接到训练集的 mean/std 上是评估流程里最容易被忽略的信息泄露入口。做特征工程时同样如此先 fit 再 transform。2.3 损失函数与评价指标分类器和回归器共用一套接口损失函数是算法设计里的核心契约它必须是可微的并且作为优化目标要跟最终评价指标趋势一致。回归器默认用均方误差分类器默认用交叉熵评价指标分别用 R² 和 accuracy。最小实现如下def mse_loss(y_true: np.ndarray, y_pred: np.ndarray) - float: return np.mean((y_true - y_pred) ** 2) def cross_entropy_loss(y_true: np.ndarray, y_prob: np.ndarray, eps: float 1e-12) - float: 二分类交叉熵y_prob 是预测为正类的概率。 y_prob np.clip(y_prob, eps, 1 - eps) return -np.mean(y_true * np.log(y_prob) (1 - y_true) * np.log(1 - y_prob)) def accuracy(y_true: np.ndarray, y_pred: np.ndarray) - float: return np.mean(y_true y_pred)交叉熵里的np.clip是为了防止log(0)产生 naneps 取 1e-12 左右太小会数值不稳太大会让损失计算失真。交叉熵与 accuracy 并不是严格单调关系但交叉熵在错误分类时梯度幅度更大、收敛更快这就是分类器普遍用交叉熵做损失、用 accuracy 做评价的原因。这一层接口定下来之后第 3 章所有算法都可以直接复用这就是算法设计里「契约先行」的实际收益。3. 从零实现三类机器学习算法线性回归、逻辑回归与 K-Means这一章是源码设计的核心动作把纸面上的公式翻译成能跑的 Python。三个算法分别代表三类问题连续值回归、二分类、无监督聚类。它们的共同点是都依赖上一章的数据契约代码结构也刻意保持对齐方便对照差异。3.1 线性回归闭式解与梯度下降两条路线对比线性回归是最适合起步的算法因为它同时存在解析解和迭代解两种实现方式。解析解适合特征维度低几千维以内的场景迭代解能平滑过渡到逻辑回归和神经网络。两种路线都实现一下对比着看理解更深class LinearRegression: 支持闭式解与梯度下降两种 fit 路线的线性回归。 def __init__(self, method: str gd, lr: float 0.01, epochs: int 1000): self.method method self.lr lr self.epochs epochs self.w, self.b None, None def fit(self, X: np.ndarray, y: np.ndarray): n, d X.shape if self.method closed: X_aug np.hstack([np.ones((n, 1)), X]) # 偏置并入权重矩阵 w_aug np.linalg.pinv(X_aug) y # 伪逆数值更稳 self.b, self.w w_aug[0], w_aug[1:] else: self.w, self.b np.zeros(d), 0.0 for _ in range(self.epochs): pred X self.w self.b grad_w (2 / n) * (X.T (pred - y)) grad_b (2 / n) * np.sum(pred - y) self.w - self.lr * grad_w self.b - self.lr * grad_b return self def predict(self, X: np.ndarray) - np.ndarray: return X self.w self.b闭式解用np.linalg.pinv(X_aug)直接对增广矩阵做伪逆比先算X^T X再求逆的条件数更好当特征存在共线性时后者会得到数值上爆炸的解伪逆仍然返回最小范数解。梯度下降版本里X.T (pred - y)把「所有样本的梯度求和再平均」压缩成一次矩阵乘法比 for 循环快两个数量级这是 numpy 向量化编程的核心习惯。学习率lr是迭代路线最敏感的参数太大梯度发散、太小收敛缓慢具体调法在第 4 章优化器部分展开。3.2 逻辑回归把分类器的线性输出变成概率逻辑回归在源码层面只比线性回归多两步加一个 sigmoid 激活损失从 MSE 换成交叉熵。但它的梯度形式出奇地简洁推导结果恰好是X.T (prob - y) / n实现如下def sigmoid(z: np.ndarray) - np.ndarray: z np.clip(z, -500, 500) # 防止 exp 溢出为 inf return 1.0 / (1.0 np.exp(-z)) class LogisticRegression: 二分类逻辑回归支持 predict_proba 输出原始概率。 def __init__(self, lr: float 0.1, epochs: int 500): self.lr, self.epochs lr, epochs self.w, self.b None, None def fit(self, X: np.ndarray, y: np.ndarray): n, d X.shape self.w, self.b np.zeros(d), 0.0 for _ in range(self.epochs): prob sigmoid(X self.w self.b) grad_w (1 / n) * (X.T (prob - y)) grad_b (1 / n) * np.sum(prob - y) self.w - self.lr * grad_w self.b - self.lr * grad_b return self def predict_proba(self, X: np.ndarray) - np.ndarray: return sigmoid(X self.w self.b) def predict(self, X: np.ndarray, threshold: float 0.5) - np.ndarray: return (self.predict_proba(X) threshold).astype(int)sigmoid里的 clip 是必须的np.exp(-z)在 z 小于 -745 左右会溢出成 infclip 到 [-500, 500] 既保证数值安全又不影响正常区间的梯度精度。threshold是分类器设计里容易被忽略的决策点默认 0.5 只在正负样本均衡时合理样本不平衡时应拿验证集遍历阈值选 F1 最大或业务召回率达标的点。逻辑回归与线性回归源码高度相似不是巧合它俩同属广义线性模型区别只在 link function这是设计源码时值得刻意保留的结构。3.3 K-Means无监督聚类里的迭代与收敛判断K-Means 没有梯度它的迭代是「指派-更新」两板斧先给每个样本指派最近的中心点再重新计算每个簇的均值。看点是距离计算的向量化与收敛判断的写法class KMeans: 带 k-means 初始化的聚类实现。 def __init__(self, n_clusters: int 3, max_iter: int 300, tol: float 1e-4, seed: int 0): self.k, self.max_iter, self.tol, self.seed n_clusters, max_iter, tol, seed self.centroids, self.labels_ None, None def fit(self, X: np.ndarray): rng np.random.default_rng(self.seed) n X.shape[0] centroids [X[rng.integers(n)]] for _ in range(1, self.k): # k-means 初始化 dist np.min([np.sum((X - c) ** 2, axis1) for c in centroids], axis0) centroids.append(X[rng.choice(n, pdist / dist.sum())]) self.centroids np.array(centroids) for _ in range(self.max_iter): labels self._assign(X) new_centroids np.array([X[labels j].mean(axis0) for j in range(self.k)]) shift np.sum((new_centroids - self.centroids) ** 2) self.centroids, self.labels_ new_centroids, labels if shift self.tol: break return self def _assign(self, X: np.ndarray) - np.ndarray: # 展开公式 a^2 - 2ab b^2 把距离计算转成矩阵乘法 dists (np.sum(X ** 2, axis1)[:, None] - 2 * (X self.centroids.T) np.sum(self.centroids ** 2, axis1)[None, :]) return np.argmin(dists, axis1)_assign里的展开公式避免了三层 for 循环把 O(n·k·d) 的逐点距离换成矩阵乘法在 10 万样本、100 维特征时能快一到两个数量级。k-means初始化让后续中心按距离平方加权采样显著降低落入局部最优的概率。如果某个簇在迭代中没有分到样本mean会得到 nan工业实现通常重新初始化空簇中心演示版本可以加一行判断跳过。收敛判断用中心点位移平方和shift而不是硬跑满max_itertol取 1e-4 左右即可。三个算法放一起看会发现一个共同的设计模式fit负责更新状态predict/predict_proba只读不写。把这个约定保持到后面的每一个模型里训练与推理的边界就清晰了。算法模型状态收敛条件主要瓶颈线性回归闭式解w、b无迭代伪逆计算约 O(n·d²)线性回归 / 逻辑回归梯度w、b跑满 epochs学习率敏感K-Meanscentroids、labels_中心位移 tol距离矩阵内存占用4. 梯度反向传播与优化器把训练循环做成可插拔的 Python 模块当模型从单层走向多层梯度不再能手推闭式解这时需要反向传播和通用优化器。这一章把「损失-梯度-参数更新」三件事拆开让优化器可以插到任何模型上。4.1 梯度检查写反向传播之前先自证的测试方法反向传播是机器学习源码里 bug 最高发的区域一个符号写反、一个维度转置错模型表现为「能跑但 loss 不降」。在写任何反向传播之前先做数值梯度检查用差分近似验证解析梯度def gradient_check(model, X: np.ndarray, y: np.ndarray, eps: float 1e-6) - None: 数值梯度 vs 解析梯度相对误差超过 1e-6 说明反向传播有 bug。 grads model.compute_gradients(X, y) for name, g in grads.items(): param model.params[name] numeric np.zeros_like(param) for idx in np.ndindex(param.shape): orig param[idx] param[idx] orig eps loss_plus model.compute_loss(X, y) param[idx] orig - eps loss_minus model.compute_loss(X, y) param[idx] orig numeric[idx] (loss_plus - loss_minus) / (2 * eps) err np.abs(g - numeric) / (np.abs(g) np.abs(numeric) 1e-12) assert err.max() 1e-6, f{name} 梯度检查失败: {err.max():.2e}数值梯度用中心差分(f(xeps) - f(x-eps)) / 2eps比单侧差分精度高一个数量级。相对误差分母加1e-12是为了防止梯度恰好为零时除零。这个测试要求模型暴露compute_loss与compute_gradients两个方法等于强制每个模型实现可测试的接口设计上反而更干净。梯度检查只适用于基于梯度的模型决策树的剪枝算法、SVM 的 SMO 这类结构化算法没有连续梯度调试时主要靠算法流程图和边界用例推演那是另一套方法论。4.2 优化器统一接口SGD、Momentum、Adam 的结构对比优化器只做一件事根据梯度更新参数。把这件事抽象成统一接口模型代码里就不用再手写w - lr * gradclass SGD: 带动量的随机梯度下降。 def __init__(self, lr: float 0.01, momentum: float 0.9): self.lr, self.momentum lr, momentum self.velocity {} def step(self, params: dict, grads: dict) - None: for name, p in params.items(): v self.velocity.setdefault(name, np.zeros_like(p)) self.velocity[name] self.momentum * v - self.lr * grads[name] params[name] self.velocity[name] class Adam: 自适应矩估计优化器对学习率不那么敏感。 def __init__(self, lr: float 1e-3, beta1: float 0.9, beta2: float 0.999, eps: float 1e-8): self.lr, self.beta1, self.beta2, self.eps lr, beta1, beta2, eps self.m, self.v, self.t {}, {}, 0 def step(self, params: dict, grads: dict) - None: self.t 1 for name, p in params.items(): m self.m.setdefault(name, np.zeros_like(p)) v self.v.setdefault(name, np.zeros_like(p)) self.m[name] self.beta1 * m (1 - self.beta1) * grads[name] self.v[name] self.beta2 * v (1 - self.beta2) * grads[name] ** 2 m_hat self.m[name] / (1 - self.beta1 ** self.t) v_hat self.v[name] / (1 - self.beta2 ** self.t) params[name] - self.lr * m_hat / (np.sqrt(v_hat) self.eps)两个优化器共享同一个step(params, grads)签名模型里只要持有self.params字典和self.compute_gradients就能在不动模型代码的前提下替换优化器。Adam 里的偏差校正m / (1 - beta^t)只在训练初期有作用前 20 步内把一阶矩从零初始化的偏差拉回来。选型时可以参考这张表优化器适用场景默认 lr踩坑点SGD数据量小、特征已标准化0.01~0.1对 lr 极其敏感SGD Momentum层数较多的模型0.01动量从 0.9 起步Adam大多数情况的首选0.001泛化有时略差于调好的 SGD4.3 训练循环骨架与学习率衰减有了优化器接口训练循环可以统一成一个函数这也是整个算法源码仓库的主入口。常见做法是用一个train函数接收模型、优化器、数据与轮数内部完成「前向-算损失-反向-更新」四步def train(model, optimizer, dataset, epochs: int 100, batch_size: int 32, lr_decay: float 0.99): rng np.random.default_rng(0) for epoch in range(epochs): idx rng.permutation(len(dataset)) epoch_loss 0.0 for i in range(0, len(dataset), batch_size): batch_idx idx[i:i batch_size] X_b, y_b dataset[batch_idx] loss, grads model.compute_loss_and_grads(X_b, y_b) optimizer.step(model.params, grads) epoch_loss loss * len(batch_idx) optimizer.lr * lr_decay # 每轮衰减学习率 print(fepoch {epoch:3d} loss {epoch_loss / len(dataset):.6f})batch_size决定每次更新的梯度噪声32 是收敛速度与稳定性的常见折中。lr_decay每轮把学习率乘一个小于 1 的系数让模型在训练后期用更小的步长精调权重避免在最优解附近震荡。如果发现 loss 曲线下降后反弹优先检查lr_decay是否设置、batch_size是否过小而不是急着换模型结构。注意optimizer.lr * lr_decay这一步同步修改了优化器内部状态。因为调度逻辑放在 train 里而不是优化器里换用任何实现了step(params, grads)的优化器训练循环都不用改。5. 对照实验与可视化验证机器学习算法源码的正确性写完了不代表写对了。机器学习源码的验证有三板斧跟成熟库对照、看可视化曲线、做单元测试。5.1 与 sklearn 对照同一份数据、同一个指标最直接的验收方式是用公开数据集同时跑自己的实现和 sklearn 的标准实现指标差距收敛到可接受范围即视为正确。对线性回归和逻辑回归权重向量与 sklearn 的对齐程度可以到小数点后 24 位取决于收敛条件K-Means 因为初始化随机看的是聚类结果的轮廓系数而不是完全一致的标签。from sklearn.datasets import load_breast_cancer from sklearn.model_selection import train_test_split as sk_split from sklearn.linear_model import LogisticRegression as SkLR data load_breast_cancer() X, y data.data, data.target X (X - X.mean(axis0)) / (X.std(axis0) 1e-12) X_tr, X_te, y_tr, y_te sk_split(X, y, test_size0.2, random_state42) my_lr LogisticRegression(lr0.05, epochs1000).fit(X_tr, y_tr) sk_lr SkLR(max_iter1000).fit(X_tr, y_tr) print(my acc: %.4f % accuracy(y_te, my_lr.predict(X_te))) print(sk acc: %.4f % accuracy(y_te, sk_lr.predict(X_te))) print(my w: , np.round(my_lr.w, 4)) print(sk w: , np.round(sk_lr.coef_[0], 4))这份脚本同时验证了分类器效果与权重方向两个层面。如果 accuracy 接近但权重符号相反说明数据没标准化或者标签编码顺序不一致如果权重数值差异大优先怀疑学习率衰减设置与收敛轮数。对照实验的目的不是追求和 sklearn 完全一致而是把「数值正确性」从「结果好坏」里剥离出来。5.2 三张必看的图损失曲线、决策边界与收敛轨迹数值对了还要看过程。我最常画的三张图第一张是训练损失曲线正常形状是单调下降后趋平出现锯齿说明 batch_size 过小或学习率偏大第二张是二分类的决策边界叠加在数据散点图上能直观看到分类器是不是被离群点带偏第三张是 K-Means 的簇心移动轨迹用透明度画出每轮迭代的中心位置能快速判断初始化是否落入局部最优。import matplotlib.pyplot as plt def plot_loss_curve(losses: list, title: str training loss): plt.figure(figsize(6, 4)) plt.plot(losses, lw2) plt.xlabel(epoch) plt.ylabel(loss) plt.title(title) plt.yscale(log) # 对数坐标放大小幅波动 plt.grid(alpha0.3) plt.show()损失曲线用plt.yscale(log)画对数坐标是因为 MSE 和交叉熵在训练后期下降幅度往往跨数量级线性坐标下后半段看起来像一条直线看不出收敛细节。可视化不是最终验收手段但它是定位 bug 阶段最快的手段先把「过程看起来对」做到再谈上线指标。5.3 把「源码正确」固化成可复现的实验记录最后给所有跟着这套流程写源码的人一个习惯每个算法的实验都用一个固定脚本入口复现记录三个字段——数据版本、超参数、验收指标。我自己的目录组织是每个算法一个目录内部放model.py核心实现、train.py训练循环、test_model.py梯度检查与对照实验三个文件的边界严格对应前面章节的划分。跑完一次实验把git rev-parse --short HEAD的输出、超参 json 和验证集指标追加进results.md下次改动后重跑一遍diff 出来的数字就是这次提交的验收结论。本文还有配套的精品资源点击获取 
