如何在SciKit K Means等聚类方法中约束指定数据点所属聚类？

原生Scikit-learn提供的KMeans类没有直接开放指定样本固定归属簇的接口，但可以通过修改迭代过程中的样本分配逻辑实现需求，完全保留原有加权计算聚类重心的能力，且被约束的样本会全程参与聚类中心的计算，不会被默认分配到距离最近的簇。

核心实现原理 #

标准KMeans迭代分为两个重复执行的步骤直到收敛：

• 分配步骤：计算所有样本到当前各聚类中心的距离，将样本分配到距离最近的簇
• 更新步骤：根据每个簇内的所有样本（以及对应的样本权重），计算新的簇中心（也就是需要的加权重心）

需要的约束只需要修改「分配步骤」的逻辑即可，更新步骤完全可以沿用原生逻辑，不需要改动cluster_centers_的计算规则：

• 提前维护一份需要固定归属的样本索引与对应目标簇的映射表
• 每次执行分配步骤时，无约束样本正常按距离分配到最近簇；被约束的样本直接跳过距离计算，强制分配到预先指定的簇
• 后续更新簇中心时，被约束的样本会和同簇其他样本一起参与加权计算，自然会拉动簇中心向约束样本偏移，满足约束影响中心位置的要求

可直接运行的代码实现 #

基于原生KMeans继承重写，兼容原类的所有参数，支持传入样本权重计算加权重心：

import numpy as np
from sklearn.cluster import KMeans
from sklearn.utils.validation import check_array

class ConstrainedKMeans(KMeans):
    def __init__(self, n_clusters=8, fixed_assignments=None, sample_weight=None, **kwargs):
        """
        参数说明
        ----------
        n_clusters : 聚类簇数，和原生KMeans一致
        fixed_assignments : dict, 固定归属映射，格式为 {样本行索引: 指定的簇编号（从0开始）}
        sample_weight : array-like of shape (n_samples,), 每个样本的权重，用于计算加权重心
        **kwargs : 其他原生KMeans支持的参数，比如random_state、n_init等
        """
        super().__init__(n_clusters=n_clusters, **kwargs)
        self.fixed_assignments = fixed_assignments if fixed_assignments is not None else {}
        self.sample_weight = sample_weight

    def _labels_inertia(self, X, sample_weight=None):
        # 复用原生的欧氏距离计算逻辑
        distances = self._euclidean_distances(X, self.cluster_centers_, squared=True)
        # 无约束样本默认分配到距离最近的簇
        labels = distances.argmin(axis=1)
        # 强制覆盖约束样本的分配结果，跳过距离判断
        for sample_idx, target_cluster in self.fixed_assignments.items():
            labels[sample_idx] = target_cluster
        # 按权重计算簇内平方和（惯性）
        sw = sample_weight if sample_weight is not None else self.sample_weight
        if sw is None:
            sw = np.ones(X.shape[0])
        inertia = np.sum((distances[np.arange(X.shape[0]), labels]) * sw)
        return labels, inertia

    def fit(self, X, y=None, sample_weight=None):
        X = check_array(X, accept_sparse="csr")
        sw = sample_weight if sample_weight is not None else self.sample_weight
        return super().fit(X, y, sample_weight=sw)

调用示例 #

# 测试数据：100个2维样本
X = np.random.rand(100, 2)
# 约束：第0个样本强制分到0号簇，第1个样本强制分到2号簇
fixed_map = {0:0, 1:2}
# 初始化模型，3个簇，传入约束和自定义权重
ckm = ConstrainedKMeans(
    n_clusters=3,
    fixed_assignments=fixed_map,
    sample_weight=np.random.rand(100), # 替换成实际使用的样本权重
    random_state=42
)
ckm.fit(X)
# 查看聚类中心，和原生KMeans一样取cluster_centers_即可
print(ckm.cluster_centers_)
# 验证约束样本的标签
print(ckm.labels_[0], ckm.labels_[1]) # 固定输出0 2，不会随距离计算结果变化

注意事项 #

• 配置fixed_assignments时需要保证每个簇至少有一个样本（不管是固定样本还是无约束样本），避免出现空簇导致迭代报错。
• 不需要使用SemiSupervisedKMeans实现该需求：半监督KMeans的约束是must-link/cannot-link的成对约束，不是强制指定单个样本的永久归属，逻辑和需求不匹配。
• 如果不想重写类，也可以用原生KMeans手动迭代：设置max_iter=1，循环调用fit，每次拿到预测标签后手动修改约束样本的标签，再把修改后的标签作为下一轮的初始标签传入，直到聚类中心收敛，效果和上述封装类完全一致。

内容的提问来源于stack exchange，提问作者Tomatoflee