大数跨境

导师急了:“KNN都不会,还直接学Transformer?”我:“Codex能跑!”导师:“K近邻、欧氏距离、加权KNN先给我搞懂!”

导师急了:“KNN都不会,还直接学Transformer?”我:“Codex能跑!”导师:“K近邻、欧氏距离、加权KNN先给我搞懂!” 机器学习和人工智能AI
2026-08-21
0

哈喽,大家好~

今天继续基础模型分享,第二篇我们聊聊 K近邻(K-Nearest Neighbors,KNN)

这个模型其实特别适合刚开始学机器学习的时候理解,因为它背后的想法一点都不绕。

你可以先记住一句话:我要判断一个新样本属于哪一类,就先看看离它最近的几个训练样本都是什么类别。

比如你到了一个陌生地方,不知道眼前这个区域到底算“商业区”还是“住宅区”,最直接的办法就是看看周围最近的一圈建筑。如果附近大部分都是住宅,那这个位置大概率也更像住宅区。

KNN做的事情,本质上就是把这种“看邻居”的直觉变成数学计算。

KNN既可以做分类,也可以做回归。这篇我们先以最常见的 KNN分类 为主。

它主要看两件事:一个是“谁离我近”,也就是距离怎么计算;另一个是“邻居怎么看”,也就是最近的 个样本怎么投票。

常见距离有欧氏距离、曼哈顿距离等。分类时最简单的做法,就是看最近的 个邻居里哪个类别最多;也可以做距离加权,让离查询点越近的邻居拥有更大的投票权重。

KNN的优点很直观:实现简单,没有复杂的参数拟合过程,而且对一些非线性边界也能处理得不错。

它的缺点也很明显:需要保留训练样本,朴素实现下预测时要计算查询样本和大量训练样本之间的距离,所以数据一大,预测速度和内存压力都会上来。维度特别高时,距离本身也会越来越难区分,这就是我们经常说的“维度灾难”。

核心公式

先看最常见的欧氏距离:

对于一个待预测样本 ,找到距离它最近的 个训练样本,对应标签记为

最普通的多数投票可以写成:

也就是哪个类别出现得最多,就把查询样本判成哪个类别。

如果想让近的邻居更重要,可以使用距离加权:

其中一个很常见的权重写法是:

这里加一个很小的 ,主要是为了避免两个样本距离刚好为0时出现除零问题。

完整案例

下面我们不用特别规整的线性数据,而是故意造一个更有意思的数据集:一部分是多个簇,一部分是同心圆。

这么做的目的很简单,就是让分类边界明显带有非线性结构,这样更容易看到 变化之后,KNN到底发生了什么。

我们主要看四件事:原始数据长什么样、不同 的决策边界有什么区别、一个测试点到底参考了哪些邻居,以及验证集表现如何随着 变化。

import torch
import numpy as np
import matplotlib.pyplot as plt
from sklearn.datasets import make_blobs, make_circles
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, confusion_matrix
import seaborn as sns


# 多簇数据(3个簇)
X1, y1 = make_blobs(n_samples=3000, centers=[(-4,0),(0,4),(4,0)], cluster_std=1.0, random_state=42)
# 环形数据(两类:内圈与外圈)
X2, y2 = make_circles(n_samples=3000, factor=0.5, noise=0.05, random_state=1)
# 将环形的标签改为 3/4 以避免和前面3个簇标签冲突(多类任务)
y2 = y2 + 3

# 合并
X = np.vstack([X1, X2])
y = np.hstack([y1, y2])

# 划分训练/验证/测试集
X_train, X_tmp, y_train, y_tmp = train_test_split(X, y, test_size=0.4, random_state=0, stratify=y)
X_val, X_test, y_val, y_test = train_test_split(X_tmp, y_tmp, test_size=0.5, random_state=1, stratify=y_tmp)

# 转为torch张量
device = torch.device('cpu')
X_train_t = torch.tensor(X_train, dtype=torch.float32, device=device)
X_val_t = torch.tensor(X_val, dtype=torch.float32, device=device)
X_test_t = torch.tensor(X_test, dtype=torch.float32, device=device)
y_train_t = torch.tensor(y_train, dtype=torch.long, device=device)

这里用了分层抽样 stratify=y,这样训练集、验证集和测试集里的类别比例不会因为随机切分发生太大变化。

另外,这个案例只有两个数值特征,而且两个特征本身处在相近尺度,所以这里没有额外做标准化。真实业务里如果特征量纲差很多,KNN通常要先把尺度处理好,不然数值范围特别大的特征很容易直接主导距离。

实现KNN预测函数

KNN和很多模型不太一样,它没有“先训练出一堆模型参数,再拿模型预测”这个过程。

更准确一点说,它属于一种 惰性学习(lazy learning):训练阶段主要是把样本保存下来,真正的计算压力更多发生在预测阶段。

原来的写法如果一次把很大的查询集合全部传给 torch.cdist,会直接构造一个“查询样本数 × 训练样本数”的距离矩阵。像后面画决策边界时网格点很多,一次性算完会比较吃内存。

所以这里稍微改一下,按批次计算距离。原理没有变,但普通电脑跑起来会稳很多。

def knn_predict(X_train, y_train, X_query, k=5, weighted=False, batch_size=1024):
    """
    X_train: (N_train, d) tensor
    y_train: (N_train,) tensor (整数标签)
    X_query: (N_query, d) tensor
    返回: (N_query,) numpy array 预测标签
    """

    classes = torch.unique(y_train)
    pred_batches = []

    for start in range(0, X_query.shape[0], batch_size):
        X_batch = X_query[start:start + batch_size]

        # 1) 计算当前批次样本到所有训练样本的欧氏距离
        dists = torch.cdist(X_batch, X_train)

        # 2) 找到每个query最近的k个训练点
        knn_dists, knn_idx = torch.topk(dists, k=k, largest=False)
        labels = y_train[knn_idx]

        # 3) 投票
        if not weighted:
            scores = torch.stack([(labels == c).sum(dim=1for c in classes], dim=1)
        else:
            weights = 1.0 / (knn_dists + 1e-8)
            scores = torch.stack([(weights * (labels == c)).sum(dim=1for c in classes], dim=1)

        batch_preds = classes[scores.argmax(dim=1)]
        pred_batches.append(batch_preds.cpu())

    return torch.cat(pred_batches).numpy()

这个版本还有一个小变化:把原来“逐个查询点循环投票”改成了批量统计类别得分。这样代码的含义还是很直观,但跑大一点的数据时会舒服不少。

原始数据散点图

先别急着看模型,第一步还是看看数据本身。

plt.figure(figsize=(8,6))
palette = ['#e41a1c','#377eb8','#4daf4a','#984ea3','#ff7f00']
for cls in np.unique(y):
    mask = (y==cls)
    plt.scatter(X[mask,0], X[mask,1], s=30, color=palette[int(cls)%len(palette)], label=f'class {int(cls)}', alpha=0.9, edgecolor='k', linewidth=0.3)
plt.title('图1:合成数据分布(3簇 + 环形)')
plt.legend()
plt.xlabel('x1'); plt.ylabel('x2')
plt.show()

从图上能看到,一部分数据是明显分开的簇,另一部分数据是内外环结构。

这个例子想表达的不是“线性模型一定做不好”,而是当类别边界明显带有弯曲、局部结构时,KNN这种直接根据附近样本做判断的方法,会非常直观地表现出它的优势。

不同 的决策边界

接下来在整个二维平面上铺一层网格,然后让KNN预测每个网格点属于哪个类别。

这样一来,我们就能直接看到 到底在控制什么。

def plot_decision_boundary(k, ax, weighted=False):
    # 网格
    x_min, x_max = X[:,0].min()-1.0, X[:,0].max()+1.0
    y_min, y_max = X[:,1].min()-1.0, X[:,1].max()+1.0
    xx, yy = np.meshgrid(np.linspace(x_min,x_max,300), np.linspace(y_min,y_max,300))
    grid = np.c_[xx.ravel(), yy.ravel()]
    grid_t = torch.tensor(grid, dtype=torch.float32, device=device)
    preds_grid = knn_predict(X_train_t, y_train_t, grid_t, k=k, weighted=weighted)
    Z = preds_grid.reshape(xx.shape)
    ax.contourf(xx, yy, Z, levels=np.arange(-0.5, len(np.unique(y))+0.51), cmap='Pastel1', alpha=0.7)
    # 训练点
    for cls in np.unique(y_train):
        mask = (y_train==cls)
        ax.scatter(X_train[mask,0], X_train[mask,1], color=palette[int(cls)%len(palette)], s=30, edgecolor='k', linewidth=0.3)
    ax.set_title(f'k={k} {"(weighted)" if weighted else ""}')
    ax.set_xlim(x_min,x_max); ax.set_ylim(y_min,y_max)

fig, axes = plt.subplots(2,2, figsize=(12,10))
ks = [1,3,9,25]
for ax,k in zip(axes.flatten(), ks):
    plot_decision_boundary(k, ax)
plt.suptitle('图2:不同k值的KNN决策边界', fontsize=16)
plt.show()

这里最值得看的其实不是“哪个 一定最好”,而是边界怎么变化。

时,模型几乎跟着单个训练样本走,边界会非常碎,对噪声特别敏感,典型表现就是方差高、容易过拟合。

随着 变成3、9,邻居数量增加,局部偶然噪声的影响会被平均掉,边界慢慢变得平滑。

到了 ,平滑会更加明显。但 并不是越大越好,如果邻居范围大到把不同局部结构都混在一起,模型就可能开始欠拟合。

所以你可以把 简单理解成一个“局部程度”的旋钮:** 小,更相信附近很小的一块区域; 大,更相信更大范围里的整体趋势。**

看一个测试点到底参考了谁

很多同学第一次学KNN,真正理解它往往不是靠公式,而是看这一张图。

我们挑一个测试点,把离它最近的7个训练样本圈出来,再把这些邻居的距离画出来。

# 选取一个测试点
idx_example = 10  # 任选一个测试点的索引(相对X_test)
pt = X_test[idx_example]
pt_t = torch.tensor(pt.reshape(1,-1), dtype=torch.float32, device=device)

k = 7
dists = torch.cdist(pt_t, X_train_t).cpu().numpy().ravel()
knn_idx = np.argsort(dists)[:k]
knn_labels = y_train[knn_idx]
knn_dists = dists[knn_idx]

plt.figure(figsize=(12,5))
# 左:散点,显示训练点和选中点、邻居
plt.subplot(1,2,1)
plt.title('图3左:测试点与其最近的k个邻居(圈出)')
for cls in np.unique(y_train):
    mask = (y_train==cls)
    plt.scatter(X_train[mask,0], X_train[mask,1], color=palette[int(cls)%len(palette)], s=25, alpha=0.8)
plt.scatter(pt[0], pt[1], color='black', s=120, marker='X', label='query', edgecolor='white')
plt.scatter(X_train[knn_idx,0], X_train[knn_idx,1], s=130, facecolors='none', edgecolors='k', linewidths=2, label='k neighbors')
plt.legend()
plt.xlabel('x1'); plt.ylabel('x2')

# 右:邻居距离柱状图
plt.subplot(1,2,2)
plt.title('图3右:这k个邻居到查询点的距离(升序)')
bars = plt.bar(range(k), knn_dists, color=plt.cm.viridis(np.linspace(0,1,k)))
plt.xlabel('neighbor rank')
plt.ylabel('distance')
for i,v in enumerate(knn_dists):
    plt.text(i, v+0.02f'{knn_labels[i]}', ha='center')
plt.tight_layout()
plt.show()

左边能直接看到查询点周围到底有哪些邻居,右边则把这些邻居按照距离从近到远排出来,柱子上方标的是对应类别。

如果最近的一批邻居距离都比较近,而且类别高度一致,一般说明这个局部区域的判断比较稳定。

反过来,如果最近邻的类别本身就混得很厉害,或者查询点离训练样本整体都比较远,那就要谨慎一点。后者意味着这个点可能落在训练数据比较稀疏、模型其实不太“熟”的区域。

验证集上随 变化的准确率

前面的决策边界主要是帮助理解,真正选 不能只靠眼睛看图。

这里用验证集比较一组不同的 ,同时看看普通投票和距离加权投票的表现。

ks = list(range(1,51,2))  # 奇数k
accs = []
accs_weighted = []
for k in ks:
    pred = knn_predict(X_train_t, y_train_t, X_val_t, k=k, weighted=False)
    accs.append(accuracy_score(y_val, pred))
    pred_w = knn_predict(X_train_t, y_train_t, X_val_t, k=k, weighted=True)
    accs_weighted.append(accuracy_score(y_val, pred_w))

best_k = ks[int(np.argmax(accs))]
best_k_weighted = ks[int(np.argmax(accs_weighted))]

plt.figure(figsize=(8,5))
plt.plot(ks, accs, marker='o', color='#d73027', label='unweighted')
plt.plot(ks, accs_weighted, marker='s', color='#1a9641', label='distance-weighted')
plt.axvline(best_k, color='blue', linestyle='--', alpha=0.7, label=f'best unweighted k={best_k}')
plt.axvline(best_k_weighted, color='purple', linestyle=':', alpha=0.7, label=f'best weighted k={best_k_weighted}')
plt.xlabel('k')
plt.ylabel('validation accuracy')
plt.title('图4:随k变化的验证集准确率(比较加权与不加权)')
plt.legend()
plt.show()

很多数据上,你会看到 太小时验证集表现不够稳定,随着 增大先改善,之后又可能因为过度平滑而下降。不过这个“先升后降”不是必须出现的固定形状,具体还是要看数据。

这里还有一个特别容易写错的地方:上面这段代码是用一个独立验证集选 ,不是交叉验证。

如果数据量不大,希望选出来的 更稳,可以进一步使用K折交叉验证;但这篇为了把KNN本身讲清楚,先用训练集、验证集、测试集这套最直观的流程。

最后再看一次测试集

既然前面已经专门留了测试集,就别浪费它。

正确的使用方式是:先根据验证集决定 以及是否使用距离加权,全部选择完成后,再让测试集出场一次,看看最终泛化效果。

测试集不要拿来反复挑参数,否则它就会慢慢变成另一个“验证集”。

best_unweighted_acc = max(accs)
best_weighted_acc = max(accs_weighted)

if best_weighted_acc > best_unweighted_acc:
    final_k = best_k_weighted
    final_weighted = True
else:
    final_k = best_k
    final_weighted = False

pred_test = knn_predict(X_train_t, y_train_t, X_test_t, k=final_k, weighted=final_weighted)
test_acc = accuracy_score(y_test, pred_test)

print(f'final k = {final_k}')
print(f'distance weighted = {final_weighted}')
print(f'test accuracy = {test_acc:.4f}')

cm = confusion_matrix(y_test, pred_test)
plt.figure(figsize=(7,6))
sns.heatmap(cm, annot=True, fmt='d', cmap='YlGnBu')
plt.xlabel('predicted label')
plt.ylabel('true label')
plt.title('图5:测试集混淆矩阵')
plt.show()

准确率告诉我们整体分对了多少,混淆矩阵则能继续看具体是哪几个类别容易互相分错。

这一步对多分类任务很有用。因为有时候总体准确率看起来不错,但某一个小类别几乎全被错分,如果只盯着一个accuracy,很容易把这个问题漏掉。

实际使用时,我更关注这几个地方

特征尺度这个问题真的很容易踩坑。比如一个特征范围是0到1,另一个是0到100000,直接算欧氏距离时,后者很可能把前者的作用完全盖住。所以真实数据里,只要特征量纲差异明显,通常都要考虑标准化或其他合适的尺度处理。

怎么选也别死记“必须取3、5、7”。奇数 在二分类的普通投票里确实能减少平票情况,但它不是理论规定,更不是所有任务都必须这样。比较稳妥的做法还是验证集或者交叉验证。

距离怎么选要看数据。普通连续数值特征里欧氏距离很常见;一些稀疏向量、文本向量场景里,也经常会考虑余弦距离。距离定义一换,“谁是邻居”其实就跟着换了,所以这不是一个无关紧要的小参数。

数据规模和维度上来以后,KNN的代价会越来越明显。中低维数据可以考虑KD-Tree、Ball-Tree等索引结构;更大规模的向量检索场景,经常会用FAISS、Annoy等近似最近邻方法。不过维度特别高时,除了算得慢,更麻烦的是不同样本之间的距离可能越来越接近,邻居本身就没那么好区分了。

还有类别不平衡。假设95%的训练样本都是A类,只有5%是B类,单纯多数投票很容易天然偏向A类。这时候可以尝试距离加权、类别权重、重采样,也别只看accuracy,最好结合召回率、F1、混淆矩阵一起判断。

总结

KNN其实就是一个很朴素的模型:一个新样本来了,先找到离它最近的 个训练样本,再根据这些邻居做判断。

它几乎没有复杂的参数训练过程,但这不代表它“没有参数可调”。 、距离度量、是否加权、特征尺度,都会直接影响最后结果。

刚开始学KNN时,我觉得最值得真正看懂的不是公式,而是两张图:不同 下的决策边界,以及某个查询点周围到底有哪些邻居。

这两张图看懂以后,过拟合、欠拟合、距离加权、特征标准化这些问题,基本就都能顺着理解下来了。

【声明】内容源于网络
0
0
机器学习和人工智能AI
让我们一起期待 AI 带给我们的每一场变革!推送最新行业内最新最前沿人工智能技术!
内容 381
粉丝 0
机器学习和人工智能AI 让我们一起期待 AI 带给我们的每一场变革!推送最新行业内最新最前沿人工智能技术!
总阅读5.0k
粉丝0
内容381