题目重点:Lloyd分配/均值更新,处理空簇和停止条件,从双循环到广播或矩阵乘法计算距离。
题意与假设
这道练习题保留了面经中值得训练的核心约束,并将需要澄清的条件单独列出。现场拿到变体时,先和面试官确认这些条件。
- 有限实数点X[N,D],欧氏平方距离,K不超过不同点数量;题目未规定初始化/空簇/tie。
- 示例复用Apple plain Python实现:第一批K个不同点初始化,tie选小簇编号、空簇保留旧中心;输入点是tuple。
- 达到max_iter后重新分配labels到最终中心,因此labels与返回centers一致,但中心未必已是这些最终labels的均值;只有收敛时才同时满足两条件。
解题路线
- 中心C[K,D];assign每点选择最小平方距离,输出labels[N]。按labels累加sum[K,D]与count[K],非空簇更新为均值。均值是在固定分配下平方误差最小解。
- 固定中心的最近分配与固定分配的均值更新都不增加同一个SSE目标,但非凸仍可能局部最优;不能由目标下降推出全局最优。不同初始化可多次restart,选择最小SSE但增加成本。
- 广播X[:,None,:]-C[None,:,:]形成[N,K,D],沿D求平方和得到[N,K];中心题目记为[D,K]时先转置,不把两个矩阵直接相减。
- 更省临时内存的公式distance²=||X||²[:,None]+||C||²[None,:]-2XCᵀ,只构造[N,K];舍入使极小距离负值可clip0,但很近且大幅值点有消减误差,要考虑中心化/更高精度/分块直接距离。
- N/K很大按点batch生成距离,马上取argmin并聚合sum/count;避免整个[N,K,D]峰值占用。若无法保存全体labels,停止条件可用中心变化或目标变化,并明确定义容差。
- 空簇不能除0;保留、重采样或拿最大误差点各自有不同结果。特征尺度和离群值会改变欧氏簇,先解释标准化与数据清洗,不把所有业务标签当球形簇。
手动走一遍
点(0,0),(2,0),(10,0),(12,0),K2,若初始两个中心0/2,第一轮会把大多数点给后者,随后边界移动至两簇0/2与10/12,中心1/11。脚本33,435组小数据/迭代上限与独立实现对照通过。
Python 参考实现
下面的代码按上文假设实现,适合本地运行并追加自己的测试用例。
代码区可横向滚动;使用键盘时,先聚焦代码区再按方向键查看。
from __future__ import annotations
from math import isfinite
def kmeans(points: list[tuple[float, ...]], k: int,
max_iter: int = 100) -> tuple[list[tuple[float, ...]], list[int]]:
"""Deterministic Lloyd training contract; first k distinct input points seed centers.
Ties choose lower center index. Empty clusters keep their previous center.
"""
if not points or k < 1 or max_iter < 1:
raise ValueError("nonempty points, positive k and max_iter required")
dimension = len(points[0])
if dimension == 0 or any(len(p) != dimension or not all(isfinite(x) for x in p) for p in points):
raise ValueError("points must have same positive dimension and finite coordinates")
seeds = list(dict.fromkeys(points))[:k]
if len(seeds) != k:
raise ValueError("need at least k distinct points")
centers = [tuple(map(float, p)) for p in seeds]
labels = [-1] * len(points)
for _ in range(max_iter):
assigned = [min(range(k), key=lambda c: (
sum((x - y) ** 2 for x, y in zip(p, centers[c])), c)) for p in points]
if assigned == labels:
break
labels = assigned
sums = [[0.0] * dimension for _ in range(k)]
counts = [0] * k
for point, label in zip(points, labels):
counts[label] += 1
for axis, value in enumerate(point):
sums[label][axis] += value
centers = [tuple(value / counts[c] for value in sums[c]) if counts[c]
else centers[c] for c in range(k)]
# Return labels consistent with the final centers after the last allowed update.
labels = [min(range(k), key=lambda c: (
sum((x - y) ** 2 for x, y in zip(p, centers[c])), c)) for p in points]
return centers, labels
复杂度
I轮朴素O(INKD)时间,存输入/labels/center约O(ND+N+KD)。广播额外O(NKD),矩阵式O(NK),分块大小b可降到O(bK)距离内存。多restart再乘次数,不把向量化说成降低渐近时间。
易错点与追问
容易失分的边界
- 题目[D,K]中心矩阵直接当[K,D]。
- 空簇除零。
- 迭代上限到了声称必已收敛。
- 广播峰值内存忽略。
- 把局部最优当全局。
继续深挖
- K-means++与多restart怎样比较成本?
- mini-batch更新与Lloyd整轮更新差别?
- 很近的大向量距离怎么算稳?
- 稀疏高维特征如何处理?
