标准回答
K-means 通过交替优化「簇分配」与「质心更新」来最小化簇内平方和(inertia)。关键点有三:一是用广播向量化计算距离矩阵把复杂度做实在 O(NKD);二是更新时对每个簇取均值,并显式处理空簇避免 NaN;三是用 k-means++ 让初始质心彼此分散,减少陷入坏局部最优的概率。下面给出可直接运行的实现:
python
import numpy as np
def kmeans_pp_init(X, k, rng):
"""k-means++:按距离平方加权采样初始质心,使其彼此分散。"""
n = X.shape[0]
centroids = [X[rng.integers(n)]] # 随机选第一个质心
for _ in range(1, k):
# 每个点到已有质心的最近平方距离
d2 = np.min(((X[:, None, :] - np.array(centroids)[None, :, :]) ** 2).sum(-1), axis=1)
probs = d2 / d2.sum() # 距离越远被选中概率越大
centroids.append(X[rng.choice(n, p=probs)])
return np.array(centroids)
def kmeans(X, k, max_iter=100, tol=1e-4, seed=0):
rng = np.random.default_rng(seed)
centroids = kmeans_pp_init(X, k, rng)
for _ in range(max_iter):
# 1) 分配:N×K 平方欧氏距离矩阵,取最近质心
dist2 = ((X[:, None, :] - centroids[None, :, :]) ** 2).sum(-1)
labels = dist2.argmin(axis=1)
# 2) 更新:逐簇取均值;空簇用离当前质心最远的点替补
new_centroids = np.empty_like(centroids)
for j in range(k):
members = X[labels == j]
if len(members) == 0:
far = ((X - centroids[j]) ** 2).sum(-1).argmax()
new_centroids[j] = X[far]
else:
new_centroids[j] = members.mean(axis=0)
# 3) 收敛判定:质心移动小于阈值即停止
if np.linalg.norm(new_centroids - centroids) < tol:
centroids = new_centroids
break
centroids = new_centroids
inertia = ((X - centroids[labels]) ** 2).sum() # 簇内平方和
return labels, centroids, inertia
if __name__ == '__main__':
rng = np.random.default_rng(42)
X = np.vstack([rng.normal(m, 0.5, (100, 2)) for m in ([0, 0], [5, 5], [0, 5])])
labels, centroids, inertia = kmeans(X, k=3)
print('inertia=', round(float(inertia), 2))
print('centroids=\n', np.round(centroids, 2))常见误区
⚠️ 常见踩坑
比较距离时多此一举地开方(np.sqrt)拖慢且无意义;更致命的是忘记处理空簇,导致 members.mean() 返回 NaN 让质心全部污染。此外随机初始化对结果影响大,应多次重启取最小 inertia,而非只跑一次。
延伸学习
与本题相关的知识库文章、术语、工具与行业资讯。