尧图网页设计 
前言上一篇使用逻辑回归完成了二分类。逻辑回归会从训练数据中学习一组参数再根据线性得分、Sigmoid 函数和分类阈值作出判断。K 近邻算法采用另一种思路它面对一个新样本时不先建立一条显式公式而是回到训练数据中寻找与它最接近的若干样本再根据这些“邻居”的类别进行预测。这种方法很像向附近的人询问意见。不过算法里的“附近”不是地理位置而是由特征和距离公式共同定义的特征空间。本篇使用 scikit-learn 内置的葡萄酒数据集根据 13 个化学成分特征预测 3 个类别。该数据集只用于机器学习教学本篇重点是理解 KNN 的工作方式不进行食品质量、医学或商业判断。K 近邻算法是什么K 近邻算法K-Nearest NeighborsKNN是一种基于实例的监督学习算法。“监督学习”表示训练样本带有已知标签“基于实例”表示预测主要依赖保存下来的训练样本而不是一条预先学好的复杂公式。它的基本假设可以写成特征空间中距离较近的样本可能具有相似的类别或目标值。在分类任务中KNN 通常依次完成下面几步保存训练数据接收一个新样本计算新样本与训练样本之间的距离找到距离最近的 K 个样本查看这些邻居的类别通过投票得到预测类别。KNN 常被称为惰性学习Lazy Learning算法。这里的“惰性”并不是说算法什么都不做而是说训练阶段主要保存数据没有像线性回归或逻辑回归那样学习一组显式系数。较多计算被推迟到了预测阶段。从邻居投票理解分类先想象一个只有两个特征的简单平面。蓝色圆点代表类别 A橙色方块代表类别 B一个尚未分类的新样本落在两类样本之间。当K 3时算法只查看最近的 3 个邻居。如果其中 2 个属于 A、1 个属于 B那么多数票是 A新样本就被预测为 A。K 3 A 类2 票 B 类1 票 预测A 类如果改成K 5搜索范围会扩大。新增的两个邻居可能都属于 B于是投票变成 A 类 2 票、B 类 3 票预测结果也可能随之变化。这个例子说明K 不只是代码中的一个数字它直接决定模型观察多大范围并影响模型对局部结构的敏感程度。距离怎样衡量样本相似性本篇重点使用欧氏距离Euclidean Distance。它就是平面中两点直线距离向多特征数据的推广。对于两个二维样本A (x₁, y₁) B (x₂, y₂) d(A, B) √((x₁-x₂)² (y₁-y₂)²)计算过程可以拆成四步先分别计算每个特征的差再对差值平方把平方结果相加最后开平方。数据有 13 个特征时只是把 13 个特征的平方差都加入求和核心思路没有改变。scikit-learn 的KNeighborsClassifier默认使用闵可夫斯基距离Minkowski Distance。当参数p2时它对应欧氏距离。本篇在代码里显式写出p2只围绕欧氏距离理解算法不展开所有距离度量。需要注意“距离近”只表示两个样本在当前特征和当前度量方式下相似。若特征选得不合理或者某个数字并不能表达现实中的接近关系那么较小的数学距离也不能保证两个对象在现实意义上真的相似。为什么必须重视特征尺度葡萄酒数据中的不同特征并不处在同一个数量级。例如实际读取结果显示alcohol酒精含量的范围是 11.0314.83而proline脯氨酸含量的范围是 2781680。假设两个样本的酒精含量相差 1脯氨酸含量相差 500。直接计算欧氏距离时后者平方后的数值会远大于前者脯氨酸几乎可能单独决定谁更近。此时模型看似使用了 13 个特征实际距离却可能被少数大尺度特征主导。因此KNN 通常需要标准化。本篇使用StandardScaler()把每个特征转换到以训练均值为中心、标准差相近的尺度。正确顺序是先拆分数据 → 只用训练集拟合 StandardScaler → 用同一缩放规则转换验证集和测试集 → 在缩放后的特征空间计算距离完整流程放进PipelinePipeline( steps[ (scaler, StandardScaler()), ( classifier, KNeighborsClassifier( n_neighbors5, ), ), ] )Pipeline 会让标准化器只从传给fit的数据学习参数保证训练、验证和测试使用同一套缩放规则也方便每次调整 K 时创建一条完整的新流程。这样可以避免先用完整数据计算均值和标准差造成的预处理泄漏。K 值表示什么n_neighbors就是 KNN 中的 K它决定预测时考虑多少个最近邻。K 较小K 1时预测完全由最近的单个样本决定。模型非常关注局部结构训练集表现可能很好但一个噪声点或异常点就可能改变预测。它的决策边界往往比较曲折容易过拟合。K 较大K 较大时模型会汇总更大范围内的邻居预测通常更平滑也不容易被单个噪声点影响。但范围过大可能忽略局部结构出现欠拟合在类别不平衡的数据中多数类别也更容易主导投票。不存在对所有数据都最好的固定 K。K 是需要使用验证集或交叉验证选择的超参数不能在测试集上尝试多个 K 后再挑最高分。为什么常见示例使用奇数 K在二分类任务中偶数 K 更容易出现票数相同。例如K 4时可能得到 A 类 2 票、B 类 2 票。使用奇数 K 可以减少这种平票所以教学示例经常使用 1、3、5、7 等候选值。但这只是一种方便的经验并不是严格规则。多分类任务仍可能形成票数相同不同训练样本也可能与新样本距离相同。scikit-learn 会按照其确定的内部规则给出结果我们不应把“奇数 K”理解成消除一切平票的保证。均匀投票与距离加权KNN 常用两种邻居权重方式。均匀投票weightsuniform每个邻居拥有相同的投票权。距离非常近和稍远的邻居影响相同规则简单、容易解释。距离加权weightsdistance距离越近的邻居影响越大。当最近样本比稍远样本更有参考价值时这种方式可能有效。不过距离加权并不一定始终更好。本篇会在同一个验证集上比较uniform与distance再按预先规定的规则选择而不是凭直觉决定。使用葡萄酒数据集数据通过下面的接口加载from sklearn.datasets import load_wine dataset load_wine(as_frameTrue) X dataset.data y dataset.targetas_frameTrue让特征和标签以 Pandas 对象返回。dataset.data是特征表dataset.target是类别标签dataset.feature_names保存特征名称dataset.target_names保存类别名称。本次实际读取到 178 个样本、13 个数值特征和 3 个类别类别 0、1、2 分别有 59、71、48 个样本。数据随 scikit-learn 安装在本地不需要联网也不需要手动下载文件。这些类别只服务于算法演示不能把模型输出解释为现实中的品质鉴定或商业建议。检查数据开始建模前至少应查看形状、类别分布、缺失值和描述性统计print(X.shape) print(y.shape) print(X.isna().sum()) print(y.value_counts()) print(X.describe())完整代码还检查了 X 与 y 的样本数是否一致、全部特征是否为数值、是否存在无穷值、是否至少有两个类别以及是否存在重复特征行。实际结果是特征形状(178, 13)标签形状(178,)缺失值 0重复特征行 0数值有限性检查通过。查看各特征的最小值、最大值和标准差可以明显看到尺度差异。例如nonflavanoid_phenols的范围约为 0.130.66而proline的范围为 2781680。对依赖距离的 KNN 来说这正是标准化不可省略的直接证据。训练集、验证集和测试集本篇使用两次train_test_split。第一次把完整数据拆成约 70% 的训练集和 30% 的临时集第二次再把临时集平均分成验证集和测试集。两次拆分都使用random_state42固定随机过程并分别用完整标签与临时标签进行stratify分层抽样。实际得到训练集 124 个样本验证集 27 个样本测试集 27 个样本。类别分布分别为训练集{0: 41, 1: 50, 2: 33} 验证集{0: 9, 1: 10, 2: 8} 测试集{0: 9, 1: 11, 2: 7}训练集用于拟合标准化器并保存邻居样本验证集用于比较不同 K 和权重方式测试集只在最终方案确定后评估一次。这个职责划分比“全部数据上跑出一个高分”更重要因为测试集一旦参与选参数就不再是独立的最终检查。建立分类基线为了判断 KNN 是否真的利用了特征我们使用DummyClassifier(strategymost_frequent)这个多数类基线始终预测开发数据中样本最多的类别。葡萄酒数据是三分类因此除准确率外还计算宏平均精确率、宏平均召回率和宏平均 F1averagemacro zero_division0宏平均会先分别计算每个类别的指标再对三个类别取平均让每个类别拥有相同权重。zero_division0则明确规定某个类别从未被预测时如何处理除零情况。实际基线在测试集上的准确率为 0.4074宏平均精确率为 0.1358宏平均召回率为 0.3333宏平均 F1 为 0.1930。使用验证集选择 K 和权重方式候选参数在查看测试结果之前写好candidate_neighbors [1, 3, 5, 7, 9, 11, 15] candidate_weights [uniform, distance]每个组合都创建一条全新的 Pipeline只在训练集上fit然后在验证集计算准确率和宏平均 F1。实际 14 组结果如下K权重方式验证集准确率验证集宏平均 F11uniform0.96300.96281distance0.96300.96283uniform0.92590.92593distance0.92590.92595uniform0.92590.92595distance0.92590.92597uniform0.92590.92597distance0.92590.92599uniform0.92590.92599distance0.92590.925911uniform0.92590.925911distance0.92590.925915uniform0.96300.962815distance0.96300.9628选择规则也提前固定先比较验证集宏平均 F1再比较准确率若仍相同优先较大的 K以获得相对平滑的局部决策最后优先规则更简单的uniform。因此本次选中K15、weightsuniform。这里选择 K15 不是因为看过测试集而是因为它与 K1 在验证指标上并列后由预先规定的稳定性规则胜出。验证集只有 27 个样本0.9630 也就是只有少量预测差异。这个结果适合演示选择流程却不足以证明某个参数在其他数据上必然更好。真实项目通常还会使用交叉验证获得更稳定的估计。训练最终 KNN 模型参数确定后把训练集和验证集合并成 151 个样本的开发集创建一条新的StandardScaler KNeighborsClassifierPipeline并在开发集上重新训练x_development pd.concat([x_train, x_validation], axis0) y_development pd.concat([y_train, y_validation], axis0) final_model build_pipeline(best_k, best_weights) final_model.fit(x_development, y_development)合并的原因是验证集已经完成选参数职责最终模型可以利用这部分有标签数据。测试集仍然不能加入训练否则最终评估会泄漏答案。在测试集上评估最终模型最终 KNN 在 27 个测试样本上的实际指标为准确率 1.0000、宏平均精确率 1.0000、宏平均召回率 1.0000、宏平均 F1 1.0000。显式使用类别顺序[0, 1, 2]得到混淆矩阵[[ 9 0 0] [ 0 11 0] [ 0 0 7]]相比多数类基线KNN 显然利用了特征和邻居信息。但这里必须克制解读测试集仅有 27 个样本并且这是一个结构较清晰的教学数据集。一次固定拆分得到满分不等于模型对所有葡萄酒或真实生产数据都不会出错也不能代替更多拆分、交叉验证和真实业务验证。查看新样本的最近邻Pipeline 会先标准化样本再交给分类器。若想查看最近邻需要取得两个步骤scaler final_model.named_steps[scaler] classifier final_model.named_steps[classifier] transformed_sample scaler.transform(sample) distances, indices classifier.kneighbors( transformed_sample, return_distanceTrue, )本篇从测试集选择原始索引为 7 的样本仅用于解释不把它加入训练。它的真实类别和预测类别都是0 (class_0)。最近的 15 个开发集邻居距离依次为[1.7932, 1.8397, 2.0447, 2.2061, 2.2169, 2.3021, 2.3545, 2.3548, 2.3811, 2.3932, 2.4513, 2.4605, 2.4668, 2.4737, 2.5071]这 15 个邻居的类别都是 0所以均匀投票得到类别 0。indices指向最终模型训练时看到的开发集位置因此代码使用与合并顺序一致的y_development.iloc[indices[0]]取邻居标签。如果误用原始数据的标签位置解释就可能与模型实际邻居错位。查看邻居有助于理解单个预测为什么出现但它不能替代整体测试集评估。一个样本的邻居非常一致也不代表所有测试样本都会如此。对一个手工新样本进行预测代码用开发集各特征中位数构造基础样本再把alcohol、color_intensity和proline调整到开发集的 75% 分位数。列名、列顺序与训练数据完全一致全部数值也经过范围检查没有使用测试标签。实际预测结果是类别编号 0对应名称class_0。分类器实际的classes_顺序为 0、1、2对应概率为类别 0 (class_0)1.0000 类别 1 (class_1)0.0000 类别 2 (class_2)0.0000这只是基于开发集统计量构造的数据点和模型邻居投票结果不是现实葡萄酒鉴定。代码没有手工假定概率列顺序而是读取classifier.classes_再将每一列概率与正确类别对应。KNN 的 predict_proba 表示什么KNN 的“概率”通常来自邻居投票比例或距离加权后的结果。例如K5且均匀投票时若 3 个邻居属于类别 0、1 个属于类别 1、1 个属于类别 2那么输出可能近似为 0.6、0.2、0.2。距离加权时每个邻居的贡献不同概率也会按权重汇总。这些数值表达模型在当前邻居规则下的支持程度不一定是经过严格概率校准的现实发生概率。因此模型输出 0.8 不能自动解释为现实中“有 80% 的真实概率”。KNN 分类与 KNN 回归KNN 不只可以分类。KNeighborsClassifier通过邻居投票预测类别KNeighborsRegressor则通过邻居目标值的平均数或距离加权平均数预测连续结果。两者都依赖距离、K 值和特征尺度只是最后聚合邻居答案的方式不同。本篇集中讨论分类不展开 KNN 回归代码。KNN 为什么训练快、预测慢KNN 训练阶段主要保存训练样本、拟合标准化参数并根据实现建立必要的检索结构因此通常很快。预测阶段却要把每个新样本与大量训练样本进行距离比较找出最近的 K 个邻居再投票或加权。随着训练数据增多预测耗时和保存训练数据所需内存都可能增加。scikit-learn 会根据数据情况使用不同的最近邻搜索算法但本篇不深入 KD Tree、Ball Tree 等内部实现。只需记住KNN 把许多工作从训练阶段推迟到了预测阶段。什么是维度灾难维度灾难Curse of Dimensionality描述的是特征数量增加后高维空间出现的一系列困难空间迅速变得稀疏样本之间距离的差异可能变得不明显“最近邻”不像二维平面中那样直观而且需要更多数据才能覆盖特征空间。无关特征也会进入距离计算可能把真正有用的相似关系冲淡。因此不是特征越多越好。面对高维数据可以考虑删除无关特征、进行特征选择或降维、收集更多数据或者改用其他模型。KNN 的性能和预测速度都可能随着维度增加而下降。KNN 常见问题1. 忘记标准化数值范围较大的特征会主导距离。应使用StandardScaler KNeighborsClassifierPipeline并让标准化器只在训练数据上拟合。2. K 设置过小模型可能过度依赖单个样本对噪声和异常点敏感训练表现很好但泛化较差。3. K 设置过大较大的邻居范围可能淹没局部结构少数类别更难被识别模型可能欠拟合。4. 使用测试集选择 K这样会让测试集参与模型开发使最终指标偏乐观。应使用验证集或交叉验证选择参数测试集只用于最终评估。5. 特征中包含无关信息KNN 会把无关特征也用于计算距离额外维度可能降低效果。需要结合任务进行特征选择而不是把所有字段原样塞给模型。6. 类别不平衡多数类别可能在邻居投票中占优势。可以比较距离权重、选择宏平均指标、合理重采样、收集更多少数类样本或考虑其他模型。本篇主代码不加入复杂重采样。7. 预测速度慢训练样本很多时每次预测都可能需要大量距离计算。对延迟严格的服务应实际测试预测时间和内存成本。8. 新样本超出训练范围模型仍会返回所谓最近邻但这些邻居可能实际上都很远。此时预测可靠性会下降仅有类别输出并不能说明样本在训练分布内。9. 字符串特征不能直接计算欧氏距离类别字段通常需要编码。但把类别编码成 0、1、2 后这些数字之间的距离是否有意义还要结合字段含义判断不能机械套用。10. 缺失值大多数 KNN 实现不能直接处理包含 NaN 的特征。可以在 Pipeline 中先合理填补并确保填补器也只从训练集学习。本篇内置数据没有缺失值因此不展开完整缺失值流水线。KNN 的优点KNN 原理直观几乎不需要复杂训练可以表达非线性分类边界自然支持多分类也可扩展到回归。它还能查看邻居来辅助…59 tokens truncated…感高维数据中还会受到维度灾难影响。类别不平衡时可能偏向多数类predict_proba也通常不代表经过校准的真实概率。完整实践代码下面的代码保存为code/article_09_knn_classifier.py。它不访问网络、不读取外部数据、不导入绘图库也不会生成图片。KNN 比较适合数据规模较小或中等、特征已合理缩放、样本相似性能够通过距离表达、维度不太高并且预测延迟要求不极端的任务。它也适合用作一个直观的分类基线帮助判断局部邻域是否包含有效信息。如果样本数量非常大、特征维度很高、大量字段与任务无关、不同特征无法通过简单距离合理比较或者部署系统对延迟和模型体积要求严格KNN 往往不是首选。缺失严重、类别极度不平衡的数据也需要先解决相应问题。《K 近邻算法入门从距离理解分类》完整示例。 数据来自 scikit-learn 内置葡萄酒数据集不访问网络也不生成图片。 from __future__ import annotations import platform import sys from typing import Any, Sequence try: import numpy as np import pandas as pd import sklearn from sklearn.base import ClassifierMixin from sklearn.datasets import load_wine from sklearn.dummy import DummyClassifier from sklearn.metrics import ( accuracy_score, confusion_matrix, f1_score, precision_score, recall_score, ) from sklearn.model_selection import train_test_split from sklearn.neighbors import KNeighborsClassifier from sklearn.pipeline import Pipeline from sklearn.preprocessing import StandardScaler except ImportError as exc: print(f依赖导入失败{exc}) print(请确认当前 Conda 环境已安装 NumPy、Pandas 和 scikit-learn。) raise SystemExit(1) from exc RANDOM_STATE 42 CANDIDATE_NEIGHBORS [1, 3, 5, 7, 9, 11, 15] CANDIDATE_WEIGHTS [uniform, distance] def check_environment() - None: 输出解释器与直接依赖版本并检查 Python 主版本。 print(环境信息) print(fPython: {platform.python_version()}) print(fNumPy: {np.__version__}) print(fPandas: {pd.__version__}) print(fScikit-learn: {sklearn.__version__}) print(fInterpreter: {sys.executable}) if sys.version_info.major 3: raise RuntimeError(本示例需要 Python 3。) def class_distribution(y: pd.Series) - dict[int, int]: 按类别编号排序并返回样本数量。 counts y.value_counts().sort_index() return {int(label): int(count) for label, count in counts.items()} def load_and_validate_dataset( ) - tuple[pd.DataFrame, pd.Series, np.ndarray[Any, np.dtype[np.str_]]]: 加载内置数据并执行与 KNN 距离计算有关的基本检查。 dataset load_wine(as_frameTrue) X dataset.data.copy() y dataset.target.copy() target_names np.asarray(dataset.target_names, dtypestr) if len(X) ! len(y): raise ValueError(特征与标签的样本数量不一致。) if X.empty: raise ValueError(特征表为空。) if not all(pd.api.types.is_numeric_dtype(dtype) for dtype in X.dtypes): raise TypeError(KNN 示例要求所有输入特征均为数值类型。) missing_count int(X.isna().sum().sum() y.isna().sum()) if missing_count ! 0: raise ValueError(f数据中存在 {missing_count} 个缺失值。) if not np.isfinite(X.to_numpy(dtypefloat)).all(): raise ValueError(特征中存在无穷值。) if y.nunique() 2: raise ValueError(分类任务至少需要两个类别。) labels sorted(int(label) for label in y.unique()) if labels ! list(range(len(target_names))): raise ValueError(类别编号与类别名称无法正确对应。) print(\n数据检查) print(f完整特征形状: {X.shape}) print(f标签形状: {y.shape}) print(f特征数量: {X.shape[1]}) print(f类别数量: {y.nunique()}) print(f类别名称: {target_names.tolist()}) print(f完整类别分布: {class_distribution(y)}) print(f缺失值数量: {missing_count}) print(f重复特征行数量: {int(X.duplicated().sum())}) print(数值有限性检查: 通过) scale_table pd.DataFrame( { minimum: X.min(), maximum: X.max(), standard_deviation: X.std(), } ) print(\n特征尺度范围) print(scale_table.to_string(float_formatlambda value: f{value:.4f})) return X, y, target_names def split_dataset( X: pd.DataFrame, y: pd.Series, ) - tuple[ pd.DataFrame, pd.DataFrame, pd.DataFrame, pd.Series, pd.Series, pd.Series, ]: 按约 70%/15%/15% 分层拆分训练、验证和测试集。 x_train, x_temporary, y_train, y_temporary train_test_split( X, y, test_size0.30, random_stateRANDOM_STATE, stratifyy, ) x_validation, x_test, y_validation, y_test train_test_split( x_temporary, y_temporary, test_size0.50, random_stateRANDOM_STATE, stratifyy_temporary, ) print(\n数据集拆分) print(f训练集样本数: {len(x_train)}) print(f验证集样本数: {len(x_validation)}) print(f测试集样本数: {len(x_test)}) print(f训练集类别分布: {class_distribution(y_train)}) print(f验证集类别分布: {class_distribution(y_validation)}) print(f测试集类别分布: {class_distribution(y_test)}) return x_train, x_validation, x_test, y_train, y_validation, y_test def build_pipeline(n_neighbors: int, weights: str) - Pipeline: 为一个参数组合创建全新的标准化与 KNN 流水线。 if n_neighbors 1: raise ValueError(K 必须是正整数。) if weights not in CANDIDATE_WEIGHTS: raise ValueError(f不支持的权重方式{weights}) return Pipeline( steps[ (scaler, StandardScaler()), ( classifier, KNeighborsClassifier( n_neighborsn_neighbors, weightsweights, p2, ), ), ] ) def evaluate_classifier( y_true: pd.Series, y_predicted: np.ndarray[Any, Any], ) - dict[str, float]: 计算三分类任务的准确率和宏平均指标。 return { accuracy: float(accuracy_score(y_true, y_predicted)), precision_macro: float( precision_score( y_true, y_predicted, averagemacro, zero_division0, ) ), recall_macro: float( recall_score( y_true, y_predicted, averagemacro, zero_division0, ) ), f1_macro: float( f1_score( y_true, y_predicted, averagemacro, zero_division0, ) ), } def search_hyperparameters( x_train: pd.DataFrame, y_train: pd.Series, x_validation: pd.DataFrame, y_validation: pd.Series, ) - pd.DataFrame: 只利用训练集训练候选模型并只利用验证集比较参数。 if max(CANDIDATE_NEIGHBORS) len(x_train): raise ValueError(候选 K 不能大于训练集样本数量。) records: list[dict[str, Any]] [] for n_neighbors in CANDIDATE_NEIGHBORS: for weights in CANDIDATE_WEIGHTS: candidate_model build_pipeline(n_neighbors, weights) candidate_model.fit(x_train, y_train) validation_prediction candidate_model.predict(x_validation) records.append( { n_neighbors: n_neighbors, weights: weights, validation_accuracy: float( accuracy_score(y_validation, validation_prediction) ), validation_f1_macro: float( f1_score( y_validation, validation_prediction, averagemacro, zero_division0, ) ), } ) results pd.DataFrame(records) print(\n验证参数结果) print( results.to_string( indexFalse, formatters{ validation_accuracy: lambda value: f{value:.4f}, validation_f1_macro: lambda value: f{value:.4f}, }, ) ) return results def select_best_configuration(results: pd.DataFrame) - tuple[int, str]: 按预先规定的宏平均 F1、准确率、较大 K、uniform 顺序选参数。 required_columns { n_neighbors, weights, validation_accuracy, validation_f1_macro, } if results.empty or not required_columns.issubset(results.columns): raise ValueError(验证结果为空或缺少必要列。) ranked results.assign( uniform_priority(results[weights] uniform).astype(int) ).sort_values( by[ validation_f1_macro, validation_accuracy, n_neighbors, uniform_priority, ], ascending[False, False, False, False], kindmergesort, ) best ranked.iloc[0] best_k int(best[n_neighbors]) best_weights str(best[weights]) print(\n参数选择规则先比较验证集宏平均 F1再比较准确率) print(若仍相同优先较大的 K最后优先结构更简单的 uniform。) print(f最佳K: {best_k}) print(f最佳权重方式: {best_weights}) return best_k, best_weights def train_final_model( x_train: pd.DataFrame, y_train: pd.Series, x_validation: pd.DataFrame, y_validation: pd.Series, best_k: int, best_weights: str, ) - tuple[Pipeline, pd.DataFrame, pd.Series]: 合并训练集与验证集创建并训练一条新的最终流水线。 x_development pd.concat([x_train, x_validation], axis0) y_development pd.concat([y_train, y_validation], axis0) if best_k len(x_development): raise ValueError(最终 K 不能大于开发集样本数量。) final_model build_pipeline(best_k, best_weights) final_model.fit(x_development, y_development) return final_model, x_development, y_development def evaluate_final_models( final_model: Pipeline, x_development: pd.DataFrame, y_development: pd.Series, x_test: pd.DataFrame, y_test: pd.Series, labels: Sequence[int], ) - tuple[pd.DataFrame, np.ndarray[Any, Any]]: 参数确定后仅在测试集上评估多数类基线和最终 KNN。 baseline: ClassifierMixin DummyClassifier(strategymost_frequent) baseline.fit(x_development, y_development) baseline_prediction baseline.predict(x_test) knn_prediction final_model.predict(x_test) rows [] for model_name, prediction in ( (多数类基线, baseline_prediction), (最终KNN, knn_prediction), ): metrics evaluate_classifier(y_test, prediction) rows.append({model: model_name, **metrics}) comparison pd.DataFrame(rows) matrix confusion_matrix(y_test, knn_prediction, labelslist(labels)) print(\n测试集模型对比) print( comparison.to_string( indexFalse, formatters{ accuracy: lambda value: f{value:.4f}, precision_macro: lambda value: f{value:.4f}, recall_macro: lambda value: f{value:.4f}, f1_macro: lambda value: f{value:.4f}, }, ) ) print(\n最终KNN混淆矩阵) print(matrix) return comparison, matrix def inspect_nearest_neighbors( final_model: Pipeline, x_development: pd.DataFrame, y_development: pd.Series, x_test: pd.DataFrame, y_test: pd.Series, target_names: np.ndarray[Any, np.dtype[np.str_]], ) - None: 查看一条测试样本在最终开发集中的最近邻。 sample x_test.iloc[[0]] true_label int(y_test.iloc[0]) predicted_label int(final_model.predict(sample)[0]) scaler: StandardScaler final_model.named_steps[scaler] classifier: KNeighborsClassifier final_model.named_steps[classifier] transformed_sample scaler.transform(sample) distances, indices classifier.kneighbors( transformed_sample, return_distanceTrue, ) neighbor_labels y_development.iloc[indices[0]].astype(int).to_numpy() print(\n邻居查看) print(f测试样本原始索引: {sample.index[0]}) print( f测试样本真实类别: {true_label} f({target_names[true_label]}) ) print( f测试样本预测类别: {predicted_label} f({target_names[predicted_label]}) ) print( 邻居距离: np.array2string(distances[0], precision4, separator, ) ) print(f邻居类别: {neighbor_labels.tolist()}) print(邻居明细) for rank, (distance, position, label) in enumerate( zip(distances[0], indices[0], neighbor_labels), start1, ): print( f 第{rank}近开发集位置{int(position)}, f原始索引{x_development.index[int(position)]}, f距离{float(distance):.4f}, f类别{int(label)} ({target_names[int(label)]}) ) def build_new_sample(x_development: pd.DataFrame) - pd.DataFrame: 使用开发集统计量构造处于训练数据大致范围内的新样本。 new_sample x_development.median().to_frame().T for feature in (alcohol, color_intensity, proline): new_sample.loc[:, feature] x_development[feature].quantile(0.75) new_sample new_sample.loc[:, x_development.columns] within_range ( new_sample.ge(x_development.min()).all(axisNone) and new_sample.le(x_development.max()).all(axisNone) ) if not within_range: raise ValueError(手工新样本超出了开发数据的特征范围。) return new_sample def predict_new_sample( final_model: Pipeline, new_sample: pd.DataFrame, target_names: np.ndarray[Any, np.dtype[np.str_]], ) - None: 预测手工样本并按分类器实际类别顺序输出概率。 predicted_label int(final_model.predict(new_sample)[0]) probabilities final_model.predict_proba(new_sample)[0] classifier: KNeighborsClassifier final_model.named_steps[classifier] probability_classes classifier.classes_ if len(probabilities) ! len(probability_classes): raise RuntimeError(概率列数量与分类器类别数量不一致。) print(\n手工新样本由开发集统计量构造) print(new_sample.to_string(indexFalse, float_formatlambda value: f{value:.4f})) print(f新样本预测类别编号: {predicted_label}) print(f新样本预测类别名称: {target_names[predicted_label]}) print(新样本各类别概率) for label, probability in zip(probability_classes, probabilities): label_number int(label) print( f 类别 {label_number} ({target_names[label_number]}): f{float(probability):.4f} ) print(f概率之和: {float(np.sum(probabilities)):.4f}) def main() - int: 运行完整的 KNN 分类实验。 try: np.random.seed(RANDOM_STATE) check_environment() X, y, target_names load_and_validate_dataset() ( x_train, x_validation, x_test, y_train, y_validation, y_test, ) split_dataset(X, y) validation_results search_hyperparameters( x_train, y_train, x_validation, y_validation, ) best_k, best_weights select_best_configuration(validation_results) final_model, x_development, y_development train_final_model( x_train, y_train, x_validation, y_validation, best_k, best_weights, ) labels sorted(int(label) for label in y.unique()) evaluate_final_models( final_model, x_development, y_development, x_test, y_test, labels, ) inspect_nearest_neighbors( final_model, x_development, y_development, x_test, y_test, target_names, ) new_sample build_new_sample(x_development) predict_new_sample(final_model, new_sample, target_names) print(\n谨慎结论以上结果只反映固定拆分下的教学数据实验) print(不能直接当作现实葡萄酒鉴定、食品质量判断或商业建议。) print(\n程序正常结束。) return 0 except (ValueError, TypeError, RuntimeError) as exc: print(f\n程序执行失败{type(exc).__name__}: {exc}) return 1 except Exception as exc: # 捕获未预期错误并给出明确类型 print(f\n出现未预期错误{type(exc).__name__}: {exc}) return 1 if __name__ __main__: raise SystemExit(main())运行结果与谨慎结论本次实际运行环境为 Python 3.11.4、NumPy 1.24.3、Pandas 1.5.3、scikit-learn 1.3.0解释器是当前 Condabase环境。程序正常结束14 个候选组合全部完成训练参数选择只查看验证集测试集只在最终参数确定后评估一次。由于数据拆分和代码路径固定相同库版本下应能复现本文结果。不同 scikit-learn、NumPy 或底层数值库版本可能造成输出格式或末位浮点数的轻微差异。更重要的是固定拆分的教学实验不能替代真实场景中的数据审查、重复验证和业务评估。本文总结KNN 不显式学习一组回归系数而是保存训练实例在预测时通过距离寻找邻居并投票。理解它需要抓住四个重点距离怎样定义相似性、特征尺度为什么会改变距离、K 如何控制局部敏感程度以及验证集为什么必须与测试集分工。本篇还完成了标准化 Pipeline、均匀投票与距离加权比较、多分类宏平均指标、最终混淆矩阵、最近邻查询和手工样本概率输出。KNN 直观但并不简单预测成本、无关特征、类别不平衡和维度灾难都会影响它的使用。下一篇将学习《决策树与随机森林入门》。我们会从“按照条件不断划分数据”的直觉出发理解树模型如何处理非线性关系以及随机森林为什么要组合多棵树。 09 K 近邻算法入门：从距离理解分类
前言上一篇使用逻辑回归完成了二分类。逻辑回归会从训练数据中学习一组参数再根据线性得分、Sigmoid 函数和分类阈值作出判断。K 近邻算法采用另一种思路它面对一个新样本时不先建立一条显式公式而是回到训练数据中寻找与它最接近的若干样本再根据这些“邻居”的类别进行预测。这种方法很像向附近的人询问意见。不过算法里的“附近”不是地理位置而是由特征和距离公式共同定义的特征空间。本篇使用 scikit-learn 内置的葡萄酒数据集根据 13 个化学成分特征预测 3 个类别。该数据集只用于机器学习教学本篇重点是理解 KNN 的工作方式不进行食品质量、医学或商业判断。K 近邻算法是什么K 近邻算法K-Nearest NeighborsKNN是一种基于实例的监督学习算法。“监督学习”表示训练样本带有已知标签“基于实例”表示预测主要依赖保存下来的训练样本而不是一条预先学好的复杂公式。它的基本假设可以写成特征空间中距离较近的样本可能具有相似的类别或目标值。在分类任务中KNN 通常依次完成下面几步保存训练数据接收一个新样本计算新样本与训练样本之间的距离找到距离最近的 K 个样本查看这些邻居的类别通过投票得到预测类别。KNN 常被称为惰性学习Lazy Learning算法。这里的“惰性”并不是说算法什么都不做而是说训练阶段主要保存数据没有像线性回归或逻辑回归那样学习一组显式系数。较多计算被推迟到了预测阶段。从邻居投票理解分类先想象一个只有两个特征的简单平面。蓝色圆点代表类别 A橙色方块代表类别 B一个尚未分类的新样本落在两类样本之间。当K 3时算法只查看最近的 3 个邻居。如果其中 2 个属于 A、1 个属于 B那么多数票是 A新样本就被预测为 A。K 3 A 类2 票 B 类1 票 预测A 类如果改成K 5搜索范围会扩大。新增的两个邻居可能都属于 B于是投票变成 A 类 2 票、B 类 3 票预测结果也可能随之变化。这个例子说明K 不只是代码中的一个数字它直接决定模型观察多大范围并影响模型对局部结构的敏感程度。距离怎样衡量样本相似性本篇重点使用欧氏距离Euclidean Distance。它就是平面中两点直线距离向多特征数据的推广。对于两个二维样本A (x₁, y₁) B (x₂, y₂) d(A, B) √((x₁-x₂)² (y₁-y₂)²)计算过程可以拆成四步先分别计算每个特征的差再对差值平方把平方结果相加最后开平方。数据有 13 个特征时只是把 13 个特征的平方差都加入求和核心思路没有改变。scikit-learn 的KNeighborsClassifier默认使用闵可夫斯基距离Minkowski Distance。当参数p2时它对应欧氏距离。本篇在代码里显式写出p2只围绕欧氏距离理解算法不展开所有距离度量。需要注意“距离近”只表示两个样本在当前特征和当前度量方式下相似。若特征选得不合理或者某个数字并不能表达现实中的接近关系那么较小的数学距离也不能保证两个对象在现实意义上真的相似。为什么必须重视特征尺度葡萄酒数据中的不同特征并不处在同一个数量级。例如实际读取结果显示alcohol酒精含量的范围是 11.0314.83而proline脯氨酸含量的范围是 2781680。假设两个样本的酒精含量相差 1脯氨酸含量相差 500。直接计算欧氏距离时后者平方后的数值会远大于前者脯氨酸几乎可能单独决定谁更近。此时模型看似使用了 13 个特征实际距离却可能被少数大尺度特征主导。因此KNN 通常需要标准化。本篇使用StandardScaler()把每个特征转换到以训练均值为中心、标准差相近的尺度。正确顺序是先拆分数据 → 只用训练集拟合 StandardScaler → 用同一缩放规则转换验证集和测试集 → 在缩放后的特征空间计算距离完整流程放进PipelinePipeline( steps[ (scaler, StandardScaler()), ( classifier, KNeighborsClassifier( n_neighbors5, ), ), ] )Pipeline 会让标准化器只从传给fit的数据学习参数保证训练、验证和测试使用同一套缩放规则也方便每次调整 K 时创建一条完整的新流程。这样可以避免先用完整数据计算均值和标准差造成的预处理泄漏。K 值表示什么n_neighbors就是 KNN 中的 K它决定预测时考虑多少个最近邻。K 较小K 1时预测完全由最近的单个样本决定。模型非常关注局部结构训练集表现可能很好但一个噪声点或异常点就可能改变预测。它的决策边界往往比较曲折容易过拟合。K 较大K 较大时模型会汇总更大范围内的邻居预测通常更平滑也不容易被单个噪声点影响。但范围过大可能忽略局部结构出现欠拟合在类别不平衡的数据中多数类别也更容易主导投票。不存在对所有数据都最好的固定 K。K 是需要使用验证集或交叉验证选择的超参数不能在测试集上尝试多个 K 后再挑最高分。为什么常见示例使用奇数 K在二分类任务中偶数 K 更容易出现票数相同。例如K 4时可能得到 A 类 2 票、B 类 2 票。使用奇数 K 可以减少这种平票所以教学示例经常使用 1、3、5、7 等候选值。但这只是一种方便的经验并不是严格规则。多分类任务仍可能形成票数相同不同训练样本也可能与新样本距离相同。scikit-learn 会按照其确定的内部规则给出结果我们不应把“奇数 K”理解成消除一切平票的保证。均匀投票与距离加权KNN 常用两种邻居权重方式。均匀投票weightsuniform每个邻居拥有相同的投票权。距离非常近和稍远的邻居影响相同规则简单、容易解释。距离加权weightsdistance距离越近的邻居影响越大。当最近样本比稍远样本更有参考价值时这种方式可能有效。不过距离加权并不一定始终更好。本篇会在同一个验证集上比较uniform与distance再按预先规定的规则选择而不是凭直觉决定。使用葡萄酒数据集数据通过下面的接口加载from sklearn.datasets import load_wine dataset load_wine(as_frameTrue) X dataset.data y dataset.targetas_frameTrue让特征和标签以 Pandas 对象返回。dataset.data是特征表dataset.target是类别标签dataset.feature_names保存特征名称dataset.target_names保存类别名称。本次实际读取到 178 个样本、13 个数值特征和 3 个类别类别 0、1、2 分别有 59、71、48 个样本。数据随 scikit-learn 安装在本地不需要联网也不需要手动下载文件。这些类别只服务于算法演示不能把模型输出解释为现实中的品质鉴定或商业建议。检查数据开始建模前至少应查看形状、类别分布、缺失值和描述性统计print(X.shape) print(y.shape) print(X.isna().sum()) print(y.value_counts()) print(X.describe())完整代码还检查了 X 与 y 的样本数是否一致、全部特征是否为数值、是否存在无穷值、是否至少有两个类别以及是否存在重复特征行。实际结果是特征形状(178, 13)标签形状(178,)缺失值 0重复特征行 0数值有限性检查通过。查看各特征的最小值、最大值和标准差可以明显看到尺度差异。例如nonflavanoid_phenols的范围约为 0.130.66而proline的范围为 2781680。对依赖距离的 KNN 来说这正是标准化不可省略的直接证据。训练集、验证集和测试集本篇使用两次train_test_split。第一次把完整数据拆成约 70% 的训练集和 30% 的临时集第二次再把临时集平均分成验证集和测试集。两次拆分都使用random_state42固定随机过程并分别用完整标签与临时标签进行stratify分层抽样。实际得到训练集 124 个样本验证集 27 个样本测试集 27 个样本。类别分布分别为训练集{0: 41, 1: 50, 2: 33} 验证集{0: 9, 1: 10, 2: 8} 测试集{0: 9, 1: 11, 2: 7}训练集用于拟合标准化器并保存邻居样本验证集用于比较不同 K 和权重方式测试集只在最终方案确定后评估一次。这个职责划分比“全部数据上跑出一个高分”更重要因为测试集一旦参与选参数就不再是独立的最终检查。建立分类基线为了判断 KNN 是否真的利用了特征我们使用DummyClassifier(strategymost_frequent)这个多数类基线始终预测开发数据中样本最多的类别。葡萄酒数据是三分类因此除准确率外还计算宏平均精确率、宏平均召回率和宏平均 F1averagemacro zero_division0宏平均会先分别计算每个类别的指标再对三个类别取平均让每个类别拥有相同权重。zero_division0则明确规定某个类别从未被预测时如何处理除零情况。实际基线在测试集上的准确率为 0.4074宏平均精确率为 0.1358宏平均召回率为 0.3333宏平均 F1 为 0.1930。使用验证集选择 K 和权重方式候选参数在查看测试结果之前写好candidate_neighbors [1, 3, 5, 7, 9, 11, 15] candidate_weights [uniform, distance]每个组合都创建一条全新的 Pipeline只在训练集上fit然后在验证集计算准确率和宏平均 F1。实际 14 组结果如下K权重方式验证集准确率验证集宏平均 F11uniform0.96300.96281distance0.96300.96283uniform0.92590.92593distance0.92590.92595uniform0.92590.92595distance0.92590.92597uniform0.92590.92597distance0.92590.92599uniform0.92590.92599distance0.92590.925911uniform0.92590.925911distance0.92590.925915uniform0.96300.962815distance0.96300.9628选择规则也提前固定先比较验证集宏平均 F1再比较准确率若仍相同优先较大的 K以获得相对平滑的局部决策最后优先规则更简单的uniform。因此本次选中K15、weightsuniform。这里选择 K15 不是因为看过测试集而是因为它与 K1 在验证指标上并列后由预先规定的稳定性规则胜出。验证集只有 27 个样本0.9630 也就是只有少量预测差异。这个结果适合演示选择流程却不足以证明某个参数在其他数据上必然更好。真实项目通常还会使用交叉验证获得更稳定的估计。训练最终 KNN 模型参数确定后把训练集和验证集合并成 151 个样本的开发集创建一条新的StandardScaler KNeighborsClassifierPipeline并在开发集上重新训练x_development pd.concat([x_train, x_validation], axis0) y_development pd.concat([y_train, y_validation], axis0) final_model build_pipeline(best_k, best_weights) final_model.fit(x_development, y_development)合并的原因是验证集已经完成选参数职责最终模型可以利用这部分有标签数据。测试集仍然不能加入训练否则最终评估会泄漏答案。在测试集上评估最终模型最终 KNN 在 27 个测试样本上的实际指标为准确率 1.0000、宏平均精确率 1.0000、宏平均召回率 1.0000、宏平均 F1 1.0000。显式使用类别顺序[0, 1, 2]得到混淆矩阵[[ 9 0 0] [ 0 11 0] [ 0 0 7]]相比多数类基线KNN 显然利用了特征和邻居信息。但这里必须克制解读测试集仅有 27 个样本并且这是一个结构较清晰的教学数据集。一次固定拆分得到满分不等于模型对所有葡萄酒或真实生产数据都不会出错也不能代替更多拆分、交叉验证和真实业务验证。查看新样本的最近邻Pipeline 会先标准化样本再交给分类器。若想查看最近邻需要取得两个步骤scaler final_model.named_steps[scaler] classifier final_model.named_steps[classifier] transformed_sample scaler.transform(sample) distances, indices classifier.kneighbors( transformed_sample, return_distanceTrue, )本篇从测试集选择原始索引为 7 的样本仅用于解释不把它加入训练。它的真实类别和预测类别都是0 (class_0)。最近的 15 个开发集邻居距离依次为[1.7932, 1.8397, 2.0447, 2.2061, 2.2169, 2.3021, 2.3545, 2.3548, 2.3811, 2.3932, 2.4513, 2.4605, 2.4668, 2.4737, 2.5071]这 15 个邻居的类别都是 0所以均匀投票得到类别 0。indices指向最终模型训练时看到的开发集位置因此代码使用与合并顺序一致的y_development.iloc[indices[0]]取邻居标签。如果误用原始数据的标签位置解释就可能与模型实际邻居错位。查看邻居有助于理解单个预测为什么出现但它不能替代整体测试集评估。一个样本的邻居非常一致也不代表所有测试样本都会如此。对一个手工新样本进行预测代码用开发集各特征中位数构造基础样本再把alcohol、color_intensity和proline调整到开发集的 75% 分位数。列名、列顺序与训练数据完全一致全部数值也经过范围检查没有使用测试标签。实际预测结果是类别编号 0对应名称class_0。分类器实际的classes_顺序为 0、1、2对应概率为类别 0 (class_0)1.0000 类别 1 (class_1)0.0000 类别 2 (class_2)0.0000这只是基于开发集统计量构造的数据点和模型邻居投票结果不是现实葡萄酒鉴定。代码没有手工假定概率列顺序而是读取classifier.classes_再将每一列概率与正确类别对应。KNN 的 predict_proba 表示什么KNN 的“概率”通常来自邻居投票比例或距离加权后的结果。例如K5且均匀投票时若 3 个邻居属于类别 0、1 个属于类别 1、1 个属于类别 2那么输出可能近似为 0.6、0.2、0.2。距离加权时每个邻居的贡献不同概率也会按权重汇总。这些数值表达模型在当前邻居规则下的支持程度不一定是经过严格概率校准的现实发生概率。因此模型输出 0.8 不能自动解释为现实中“有 80% 的真实概率”。KNN 分类与 KNN 回归KNN 不只可以分类。KNeighborsClassifier通过邻居投票预测类别KNeighborsRegressor则通过邻居目标值的平均数或距离加权平均数预测连续结果。两者都依赖距离、K 值和特征尺度只是最后聚合邻居答案的方式不同。本篇集中讨论分类不展开 KNN 回归代码。KNN 为什么训练快、预测慢KNN 训练阶段主要保存训练样本、拟合标准化参数并根据实现建立必要的检索结构因此通常很快。预测阶段却要把每个新样本与大量训练样本进行距离比较找出最近的 K 个邻居再投票或加权。随着训练数据增多预测耗时和保存训练数据所需内存都可能增加。scikit-learn 会根据数据情况使用不同的最近邻搜索算法但本篇不深入 KD Tree、Ball Tree 等内部实现。只需记住KNN 把许多工作从训练阶段推迟到了预测阶段。什么是维度灾难维度灾难Curse of Dimensionality描述的是特征数量增加后高维空间出现的一系列困难空间迅速变得稀疏样本之间距离的差异可能变得不明显“最近邻”不像二维平面中那样直观而且需要更多数据才能覆盖特征空间。无关特征也会进入距离计算可能把真正有用的相似关系冲淡。因此不是特征越多越好。面对高维数据可以考虑删除无关特征、进行特征选择或降维、收集更多数据或者改用其他模型。KNN 的性能和预测速度都可能随着维度增加而下降。KNN 常见问题1. 忘记标准化数值范围较大的特征会主导距离。应使用StandardScaler KNeighborsClassifierPipeline并让标准化器只在训练数据上拟合。2. K 设置过小模型可能过度依赖单个样本对噪声和异常点敏感训练表现很好但泛化较差。3. K 设置过大较大的邻居范围可能淹没局部结构少数类别更难被识别模型可能欠拟合。4. 使用测试集选择 K这样会让测试集参与模型开发使最终指标偏乐观。应使用验证集或交叉验证选择参数测试集只用于最终评估。5. 特征中包含无关信息KNN 会把无关特征也用于计算距离额外维度可能降低效果。需要结合任务进行特征选择而不是把所有字段原样塞给模型。6. 类别不平衡多数类别可能在邻居投票中占优势。可以比较距离权重、选择宏平均指标、合理重采样、收集更多少数类样本或考虑其他模型。本篇主代码不加入复杂重采样。7. 预测速度慢训练样本很多时每次预测都可能需要大量距离计算。对延迟严格的服务应实际测试预测时间和内存成本。8. 新样本超出训练范围模型仍会返回所谓最近邻但这些邻居可能实际上都很远。此时预测可靠性会下降仅有类别输出并不能说明样本在训练分布内。9. 字符串特征不能直接计算欧氏距离类别字段通常需要编码。但把类别编码成 0、1、2 后这些数字之间的距离是否有意义还要结合字段含义判断不能机械套用。10. 缺失值大多数 KNN 实现不能直接处理包含 NaN 的特征。可以在 Pipeline 中先合理填补并确保填补器也只从训练集学习。本篇内置数据没有缺失值因此不展开完整缺失值流水线。KNN 的优点KNN 原理直观几乎不需要复杂训练可以表达非线性分类边界自然支持多分类也可扩展到回归。它还能查看邻居来辅助…59 tokens truncated…感高维数据中还会受到维度灾难影响。类别不平衡时可能偏向多数类predict_proba也通常不代表经过校准的真实概率。完整实践代码下面的代码保存为code/article_09_knn_classifier.py。它不访问网络、不读取外部数据、不导入绘图库也不会生成图片。KNN 比较适合数据规模较小或中等、特征已合理缩放、样本相似性能够通过距离表达、维度不太高并且预测延迟要求不极端的任务。它也适合用作一个直观的分类基线帮助判断局部邻域是否包含有效信息。如果样本数量非常大、特征维度很高、大量字段与任务无关、不同特征无法通过简单距离合理比较或者部署系统对延迟和模型体积要求严格KNN 往往不是首选。缺失严重、类别极度不平衡的数据也需要先解决相应问题。《K 近邻算法入门从距离理解分类》完整示例。 数据来自 scikit-learn 内置葡萄酒数据集不访问网络也不生成图片。 from __future__ import annotations import platform import sys from typing import Any, Sequence try: import numpy as np import pandas as pd import sklearn from sklearn.base import ClassifierMixin from sklearn.datasets import load_wine from sklearn.dummy import DummyClassifier from sklearn.metrics import ( accuracy_score, confusion_matrix, f1_score, precision_score, recall_score, ) from sklearn.model_selection import train_test_split from sklearn.neighbors import KNeighborsClassifier from sklearn.pipeline import Pipeline from sklearn.preprocessing import StandardScaler except ImportError as exc: print(f依赖导入失败{exc}) print(请确认当前 Conda 环境已安装 NumPy、Pandas 和 scikit-learn。) raise SystemExit(1) from exc RANDOM_STATE 42 CANDIDATE_NEIGHBORS [1, 3, 5, 7, 9, 11, 15] CANDIDATE_WEIGHTS [uniform, distance] def check_environment() - None: 输出解释器与直接依赖版本并检查 Python 主版本。 print(环境信息) print(fPython: {platform.python_version()}) print(fNumPy: {np.__version__}) print(fPandas: {pd.__version__}) print(fScikit-learn: {sklearn.__version__}) print(fInterpreter: {sys.executable}) if sys.version_info.major 3: raise RuntimeError(本示例需要 Python 3。) def class_distribution(y: pd.Series) - dict[int, int]: 按类别编号排序并返回样本数量。 counts y.value_counts().sort_index() return {int(label): int(count) for label, count in counts.items()} def load_and_validate_dataset( ) - tuple[pd.DataFrame, pd.Series, np.ndarray[Any, np.dtype[np.str_]]]: 加载内置数据并执行与 KNN 距离计算有关的基本检查。 dataset load_wine(as_frameTrue) X dataset.data.copy() y dataset.target.copy() target_names np.asarray(dataset.target_names, dtypestr) if len(X) ! len(y): raise ValueError(特征与标签的样本数量不一致。) if X.empty: raise ValueError(特征表为空。) if not all(pd.api.types.is_numeric_dtype(dtype) for dtype in X.dtypes): raise TypeError(KNN 示例要求所有输入特征均为数值类型。) missing_count int(X.isna().sum().sum() y.isna().sum()) if missing_count ! 0: raise ValueError(f数据中存在 {missing_count} 个缺失值。) if not np.isfinite(X.to_numpy(dtypefloat)).all(): raise ValueError(特征中存在无穷值。) if y.nunique() 2: raise ValueError(分类任务至少需要两个类别。) labels sorted(int(label) for label in y.unique()) if labels ! list(range(len(target_names))): raise ValueError(类别编号与类别名称无法正确对应。) print(\n数据检查) print(f完整特征形状: {X.shape}) print(f标签形状: {y.shape}) print(f特征数量: {X.shape[1]}) print(f类别数量: {y.nunique()}) print(f类别名称: {target_names.tolist()}) print(f完整类别分布: {class_distribution(y)}) print(f缺失值数量: {missing_count}) print(f重复特征行数量: {int(X.duplicated().sum())}) print(数值有限性检查: 通过) scale_table pd.DataFrame( { minimum: X.min(), maximum: X.max(), standard_deviation: X.std(), } ) print(\n特征尺度范围) print(scale_table.to_string(float_formatlambda value: f{value:.4f})) return X, y, target_names def split_dataset( X: pd.DataFrame, y: pd.Series, ) - tuple[ pd.DataFrame, pd.DataFrame, pd.DataFrame, pd.Series, pd.Series, pd.Series, ]: 按约 70%/15%/15% 分层拆分训练、验证和测试集。 x_train, x_temporary, y_train, y_temporary train_test_split( X, y, test_size0.30, random_stateRANDOM_STATE, stratifyy, ) x_validation, x_test, y_validation, y_test train_test_split( x_temporary, y_temporary, test_size0.50, random_stateRANDOM_STATE, stratifyy_temporary, ) print(\n数据集拆分) print(f训练集样本数: {len(x_train)}) print(f验证集样本数: {len(x_validation)}) print(f测试集样本数: {len(x_test)}) print(f训练集类别分布: {class_distribution(y_train)}) print(f验证集类别分布: {class_distribution(y_validation)}) print(f测试集类别分布: {class_distribution(y_test)}) return x_train, x_validation, x_test, y_train, y_validation, y_test def build_pipeline(n_neighbors: int, weights: str) - Pipeline: 为一个参数组合创建全新的标准化与 KNN 流水线。 if n_neighbors 1: raise ValueError(K 必须是正整数。) if weights not in CANDIDATE_WEIGHTS: raise ValueError(f不支持的权重方式{weights}) return Pipeline( steps[ (scaler, StandardScaler()), ( classifier, KNeighborsClassifier( n_neighborsn_neighbors, weightsweights, p2, ), ), ] ) def evaluate_classifier( y_true: pd.Series, y_predicted: np.ndarray[Any, Any], ) - dict[str, float]: 计算三分类任务的准确率和宏平均指标。 return { accuracy: float(accuracy_score(y_true, y_predicted)), precision_macro: float( precision_score( y_true, y_predicted, averagemacro, zero_division0, ) ), recall_macro: float( recall_score( y_true, y_predicted, averagemacro, zero_division0, ) ), f1_macro: float( f1_score( y_true, y_predicted, averagemacro, zero_division0, ) ), } def search_hyperparameters( x_train: pd.DataFrame, y_train: pd.Series, x_validation: pd.DataFrame, y_validation: pd.Series, ) - pd.DataFrame: 只利用训练集训练候选模型并只利用验证集比较参数。 if max(CANDIDATE_NEIGHBORS) len(x_train): raise ValueError(候选 K 不能大于训练集样本数量。) records: list[dict[str, Any]] [] for n_neighbors in CANDIDATE_NEIGHBORS: for weights in CANDIDATE_WEIGHTS: candidate_model build_pipeline(n_neighbors, weights) candidate_model.fit(x_train, y_train) validation_prediction candidate_model.predict(x_validation) records.append( { n_neighbors: n_neighbors, weights: weights, validation_accuracy: float( accuracy_score(y_validation, validation_prediction) ), validation_f1_macro: float( f1_score( y_validation, validation_prediction, averagemacro, zero_division0, ) ), } ) results pd.DataFrame(records) print(\n验证参数结果) print( results.to_string( indexFalse, formatters{ validation_accuracy: lambda value: f{value:.4f}, validation_f1_macro: lambda value: f{value:.4f}, }, ) ) return results def select_best_configuration(results: pd.DataFrame) - tuple[int, str]: 按预先规定的宏平均 F1、准确率、较大 K、uniform 顺序选参数。 required_columns { n_neighbors, weights, validation_accuracy, validation_f1_macro, } if results.empty or not required_columns.issubset(results.columns): raise ValueError(验证结果为空或缺少必要列。) ranked results.assign( uniform_priority(results[weights] uniform).astype(int) ).sort_values( by[ validation_f1_macro, validation_accuracy, n_neighbors, uniform_priority, ], ascending[False, False, False, False], kindmergesort, ) best ranked.iloc[0] best_k int(best[n_neighbors]) best_weights str(best[weights]) print(\n参数选择规则先比较验证集宏平均 F1再比较准确率) print(若仍相同优先较大的 K最后优先结构更简单的 uniform。) print(f最佳K: {best_k}) print(f最佳权重方式: {best_weights}) return best_k, best_weights def train_final_model( x_train: pd.DataFrame, y_train: pd.Series, x_validation: pd.DataFrame, y_validation: pd.Series, best_k: int, best_weights: str, ) - tuple[Pipeline, pd.DataFrame, pd.Series]: 合并训练集与验证集创建并训练一条新的最终流水线。 x_development pd.concat([x_train, x_validation], axis0) y_development pd.concat([y_train, y_validation], axis0) if best_k len(x_development): raise ValueError(最终 K 不能大于开发集样本数量。) final_model build_pipeline(best_k, best_weights) final_model.fit(x_development, y_development) return final_model, x_development, y_development def evaluate_final_models( final_model: Pipeline, x_development: pd.DataFrame, y_development: pd.Series, x_test: pd.DataFrame, y_test: pd.Series, labels: Sequence[int], ) - tuple[pd.DataFrame, np.ndarray[Any, Any]]: 参数确定后仅在测试集上评估多数类基线和最终 KNN。 baseline: ClassifierMixin DummyClassifier(strategymost_frequent) baseline.fit(x_development, y_development) baseline_prediction baseline.predict(x_test) knn_prediction final_model.predict(x_test) rows [] for model_name, prediction in ( (多数类基线, baseline_prediction), (最终KNN, knn_prediction), ): metrics evaluate_classifier(y_test, prediction) rows.append({model: model_name, **metrics}) comparison pd.DataFrame(rows) matrix confusion_matrix(y_test, knn_prediction, labelslist(labels)) print(\n测试集模型对比) print( comparison.to_string( indexFalse, formatters{ accuracy: lambda value: f{value:.4f}, precision_macro: lambda value: f{value:.4f}, recall_macro: lambda value: f{value:.4f}, f1_macro: lambda value: f{value:.4f}, }, ) ) print(\n最终KNN混淆矩阵) print(matrix) return comparison, matrix def inspect_nearest_neighbors( final_model: Pipeline, x_development: pd.DataFrame, y_development: pd.Series, x_test: pd.DataFrame, y_test: pd.Series, target_names: np.ndarray[Any, np.dtype[np.str_]], ) - None: 查看一条测试样本在最终开发集中的最近邻。 sample x_test.iloc[[0]] true_label int(y_test.iloc[0]) predicted_label int(final_model.predict(sample)[0]) scaler: StandardScaler final_model.named_steps[scaler] classifier: KNeighborsClassifier final_model.named_steps[classifier] transformed_sample scaler.transform(sample) distances, indices classifier.kneighbors( transformed_sample, return_distanceTrue, ) neighbor_labels y_development.iloc[indices[0]].astype(int).to_numpy() print(\n邻居查看) print(f测试样本原始索引: {sample.index[0]}) print( f测试样本真实类别: {true_label} f({target_names[true_label]}) ) print( f测试样本预测类别: {predicted_label} f({target_names[predicted_label]}) ) print( 邻居距离: np.array2string(distances[0], precision4, separator, ) ) print(f邻居类别: {neighbor_labels.tolist()}) print(邻居明细) for rank, (distance, position, label) in enumerate( zip(distances[0], indices[0], neighbor_labels), start1, ): print( f 第{rank}近开发集位置{int(position)}, f原始索引{x_development.index[int(position)]}, f距离{float(distance):.4f}, f类别{int(label)} ({target_names[int(label)]}) ) def build_new_sample(x_development: pd.DataFrame) - pd.DataFrame: 使用开发集统计量构造处于训练数据大致范围内的新样本。 new_sample x_development.median().to_frame().T for feature in (alcohol, color_intensity, proline): new_sample.loc[:, feature] x_development[feature].quantile(0.75) new_sample new_sample.loc[:, x_development.columns] within_range ( new_sample.ge(x_development.min()).all(axisNone) and new_sample.le(x_development.max()).all(axisNone) ) if not within_range: raise ValueError(手工新样本超出了开发数据的特征范围。) return new_sample def predict_new_sample( final_model: Pipeline, new_sample: pd.DataFrame, target_names: np.ndarray[Any, np.dtype[np.str_]], ) - None: 预测手工样本并按分类器实际类别顺序输出概率。 predicted_label int(final_model.predict(new_sample)[0]) probabilities final_model.predict_proba(new_sample)[0] classifier: KNeighborsClassifier final_model.named_steps[classifier] probability_classes classifier.classes_ if len(probabilities) ! len(probability_classes): raise RuntimeError(概率列数量与分类器类别数量不一致。) print(\n手工新样本由开发集统计量构造) print(new_sample.to_string(indexFalse, float_formatlambda value: f{value:.4f})) print(f新样本预测类别编号: {predicted_label}) print(f新样本预测类别名称: {target_names[predicted_label]}) print(新样本各类别概率) for label, probability in zip(probability_classes, probabilities): label_number int(label) print( f 类别 {label_number} ({target_names[label_number]}): f{float(probability):.4f} ) print(f概率之和: {float(np.sum(probabilities)):.4f}) def main() - int: 运行完整的 KNN 分类实验。 try: np.random.seed(RANDOM_STATE) check_environment() X, y, target_names load_and_validate_dataset() ( x_train, x_validation, x_test, y_train, y_validation, y_test, ) split_dataset(X, y) validation_results search_hyperparameters( x_train, y_train, x_validation, y_validation, ) best_k, best_weights select_best_configuration(validation_results) final_model, x_development, y_development train_final_model( x_train, y_train, x_validation, y_validation, best_k, best_weights, ) labels sorted(int(label) for label in y.unique()) evaluate_final_models( final_model, x_development, y_development, x_test, y_test, labels, ) inspect_nearest_neighbors( final_model, x_development, y_development, x_test, y_test, target_names, ) new_sample build_new_sample(x_development) predict_new_sample(final_model, new_sample, target_names) print(\n谨慎结论以上结果只反映固定拆分下的教学数据实验) print(不能直接当作现实葡萄酒鉴定、食品质量判断或商业建议。) print(\n程序正常结束。) return 0 except (ValueError, TypeError, RuntimeError) as exc: print(f\n程序执行失败{type(exc).__name__}: {exc}) return 1 except Exception as exc: # 捕获未预期错误并给出明确类型 print(f\n出现未预期错误{type(exc).__name__}: {exc}) return 1 if __name__ __main__: raise SystemExit(main())运行结果与谨慎结论本次实际运行环境为 Python 3.11.4、NumPy 1.24.3、Pandas 1.5.3、scikit-learn 1.3.0解释器是当前 Condabase环境。程序正常结束14 个候选组合全部完成训练参数选择只查看验证集测试集只在最终参数确定后评估一次。由于数据拆分和代码路径固定相同库版本下应能复现本文结果。不同 scikit-learn、NumPy 或底层数值库版本可能造成输出格式或末位浮点数的轻微差异。更重要的是固定拆分的教学实验不能替代真实场景中的数据审查、重复验证和业务评估。本文总结KNN 不显式学习一组回归系数而是保存训练实例在预测时通过距离寻找邻居并投票。理解它需要抓住四个重点距离怎样定义相似性、特征尺度为什么会改变距离、K 如何控制局部敏感程度以及验证集为什么必须与测试集分工。本篇还完成了标准化 Pipeline、均匀投票与距离加权比较、多分类宏平均指标、最终混淆矩阵、最近邻查询和手工样本概率输出。KNN 直观但并不简单预测成本、无关特征、类别不平衡和维度灾难都会影响它的使用。下一篇将学习《决策树与随机森林入门》。我们会从“按照条件不断划分数据”的直觉出发理解树模型如何处理非线性关系以及随机森林为什么要组合多棵树。