前言

上一篇使用逻辑回归完成了二分类。逻辑回归会从训练数据中学习一组参数，再根据线性得分、Sigmoid 函数和分类阈值作出判断。K 近邻算法采用另一种思路：它面对一个新样本时，不先建立一条显式公式，而是回到训练数据中寻找与它最接近的若干样本，再根据这些“邻居”的类别进行预测。

这种方法很像向附近的人询问意见。不过，算法里的“附近”不是地理位置，而是由特征和距离公式共同定义的特征空间。本篇使用 scikit-learn 内置的葡萄酒数据集，根据 13 个化学成分特征预测 3 个类别。该数据集只用于机器学习教学，本篇重点是理解 KNN 的工作方式，不进行食品质量、医学或商业判断。

K 近邻算法是什么

K 近邻算法（K-Nearest Neighbors，KNN）是一种基于实例的监督学习算法。“监督学习”表示训练样本带有已知标签；“基于实例”表示预测主要依赖保存下来的训练样本，而不是一条预先学好的复杂公式。

它的基本假设可以写成：

特征空间中距离较近的样本，可能具有相似的类别或目标值。

在分类任务中，KNN 通常依次完成下面几步：

1. 保存训练数据；
2. 接收一个新样本；
3. 计算新样本与训练样本之间的距离；
4. 找到距离最近的 K 个样本；
5. 查看这些邻居的类别；
6. 通过投票得到预测类别。

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）。当参数p=2时，它对应欧氏距离。本篇在代码里显式写出p=2，只围绕欧氏距离理解算法，不展开所有距离度量。

需要注意，“距离近”只表示两个样本在当前特征和当前度量方式下相似。若特征选得不合理，或者某个数字并不能表达现实中的接近关系，那么较小的数学距离也不能保证两个对象在现实意义上真的相似。

为什么必须重视特征尺度

葡萄酒数据中的不同特征并不处在同一个数量级。例如，实际读取结果显示，alcohol（酒精含量）的范围是 11.03～14.83，而proline（脯氨酸含量）的范围是 278～1680。

假设两个样本的酒精含量相差 1，脯氨酸含量相差 500。直接计算欧氏距离时，后者平方后的数值会远大于前者，脯氨酸几乎可能单独决定谁更近。此时模型看似使用了 13 个特征，实际距离却可能被少数大尺度特征主导。

因此，KNN 通常需要标准化。本篇使用StandardScaler()，把每个特征转换到以训练均值为中心、标准差相近的尺度。正确顺序是：

先拆分数据 → 只用训练集拟合 StandardScaler → 用同一缩放规则转换验证集和测试集 → 在缩放后的特征空间计算距离

完整流程放进Pipeline：

Pipeline( steps=[ ("scaler", StandardScaler()), ( "classifier", KNeighborsClassifier( n_neighbors=5, ), ), ] )

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 常用两种邻居权重方式。

均匀投票

weights="uniform"

每个邻居拥有相同的投票权。距离非常近和稍远的邻居影响相同，规则简单、容易解释。

距离加权

weights="distance"

距离越近的邻居影响越大。当最近样本比稍远样本更有参考价值时，这种方式可能有效。不过，距离加权并不一定始终更好。本篇会在同一个验证集上比较uniform与distance，再按预先规定的规则选择，而不是凭直觉决定。

使用葡萄酒数据集

数据通过下面的接口加载：

from sklearn.datasets import load_wine dataset = load_wine(as_frame=True) X = dataset.data y = dataset.target

as_frame=True让特征和标签以 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.13～0.66，而proline的范围为 278～1680。对依赖距离的 KNN 来说，这正是标准化不可省略的直接证据。

训练集、验证集和测试集

本篇使用两次train_test_split。第一次把完整数据拆成约 70% 的训练集和 30% 的临时集；第二次再把临时集平均分成验证集和测试集。两次拆分都使用random_state=42固定随机过程，并分别用完整标签与临时标签进行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(strategy="most_frequent")

这个多数类基线始终预测开发数据中样本最多的类别。葡萄酒数据是三分类，因此除准确率外，还计算宏平均精确率、宏平均召回率和宏平均 F1：

average="macro" zero_division=0

宏平均会先分别计算每个类别的指标，再对三个类别取平均，让每个类别拥有相同权重。zero_division=0则明确规定某个类别从未被预测时如何处理除零情况。实际基线在测试集上的准确率为 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 组结果如下：

选择规则也提前固定：先比较验证集宏平均 F1，再比较准确率；若仍相同，优先较大的 K，以获得相对平滑的局部决策；最后优先规则更简单的uniform。因此，本次选中K=15、weights="uniform"。这里选择 K=15 不是因为看过测试集，而是因为它与 K=1 在验证指标上并列后，由预先规定的稳定性规则胜出。

验证集只有 27 个样本，0.9630 也就是只有少量预测差异。这个结果适合演示选择流程，却不足以证明某个参数在其他数据上必然更好。真实项目通常还会使用交叉验证获得更稳定的估计。

训练最终 KNN 模型

参数确定后，把训练集和验证集合并成 151 个样本的开发集，创建一条新的StandardScaler + KNeighborsClassifierPipeline，并在开发集上重新训练：

x_development = pd.concat([x_train, x_validation], axis=0) y_development = pd.concat([y_train, y_validation], axis=0) 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_distance=True, )

本篇从测试集选择原始索引为 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 的“概率”通常来自邻居投票比例或距离加权后的结果。例如K=5且均匀投票时，若 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(f"Python: {platform.python_version()}") print(f"NumPy: {np.__version__}") print(f"Pandas: {pd.__version__}") print(f"Scikit-learn: {sklearn.__version__}") print(f"Interpreter: {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_frame=True) X = dataset.data.copy() y = dataset.target.copy() target_names = np.asarray(dataset.target_names, dtype=str) 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(dtype=float)).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_format=lambda 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_size=0.30, random_state=RANDOM_STATE, stratify=y, ) x_validation, x_test, y_validation, y_test = train_test_split( x_temporary, y_temporary, test_size=0.50, random_state=RANDOM_STATE, stratify=y_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_neighbors=n_neighbors, weights=weights, p=2, ), ), ] ) 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, average="macro", zero_division=0, ) ), "recall_macro": float( recall_score( y_true, y_predicted, average="macro", zero_division=0, ) ), "f1_macro": float( f1_score( y_true, y_predicted, average="macro", zero_division=0, ) ), } 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, average="macro", zero_division=0, ) ), } ) results = pd.DataFrame(records) print("\n验证参数结果：") print( results.to_string( index=False, 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], kind="mergesort", ) 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], axis=0) y_development = pd.concat([y_train, y_validation], axis=0) 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(strategy="most_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, labels=list(labels)) print("\n测试集模型对比：") print( comparison.to_string( index=False, 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_distance=True, ) 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], precision=4, separator=", ") ) print(f"邻居类别: {neighbor_labels.tolist()}") print("邻居明细：") for rank, (distance, position, label) in enumerate( zip(distances[0], indices[0], neighbor_labels), start=1, ): 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(axis=None) and new_sample.le(x_development.max()).all(axis=None) ) 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(index=False, float_format=lambda 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 直观但并不简单：预测成本、无关特征、类别不平衡和维度灾难都会影响它的使用。

下一篇将学习《决策树与随机森林入门》。我们会从“按照条件不断划分数据”的直觉出发，理解树模型如何处理非线性关系，以及随机森林为什么要组合多棵树。