自定义Python凝聚聚类算法：如何更新标签列表？

问题根源 #

你的代码里标签更新逻辑出错的核心原因是：距离矩阵的索引随合并操作不断变化，但你直接用矩阵的当前索引去修改原始样本的标签，导致索引对应关系混乱。比如第一次合并后，距离矩阵删除了j行/列，此时矩阵里的i已经不是原始样本的i，再用这个i去更新标签就会错误映射。

解决思路 #

1. 维护聚类与原始样本的映射：用一个列表clusters记录每个聚类包含的原始样本索引，比如初始时每个聚类是单个样本[[0], [1], [2], ...]，合并时把j聚类的样本全部并入i聚类，再删除j聚类。
2. 延迟标签赋值：不要每次合并就修改self.labels_，而是在聚类完成后，遍历每个原始样本，找到它所属的聚类ID，统一生成最终标签。
3. 正确更新距离矩阵：距离矩阵的行/列对应当前聚类的索引，更新时基于当前矩阵的大小操作，避免使用原始样本数量。

修正后的代码 #

import numpy as np
import math

class AgglomerativeClustering:
    def __init__(self, n_clusters=2):
        self.n_clusters = n_clusters
        
    def fit(self, X):
        n_samples = X.shape[0]
        # 初始化：每个聚类对应一个原始样本的索引列表
        self.clusters = [[i] for i in range(n_samples)]
        n_clusters = n_samples
        
        # 初始化距离矩阵（基于原始样本）
        distance_matrix = self._single_linkage(X)
        
        while n_clusters > self.n_clusters:
            # 找到当前距离矩阵中最小的元素对应的聚类对(i,j)
            i, j = np.unravel_index(np.argmin(distance_matrix), distance_matrix.shape)
            # 确保i < j，避免后续删除时索引混乱
            if i > j:
                i, j = j, i
            
            # 合并聚类：将j聚类的样本并入i聚类
            self.clusters[i].extend(self.clusters[j])
            # 删除j聚类
            del self.clusters[j]
            
            # 更新距离矩阵：计算新聚类i与其他所有聚类的距离（单链接取最小）
            for k in range(distance_matrix.shape[0]):
                if k != i and k != j:
                    distance_matrix[i, k] = min(distance_matrix[i, k], distance_matrix[j, k])
                    distance_matrix[k, i] = distance_matrix[i, k]
            
            # 删除j行和j列
            distance_matrix = np.delete(distance_matrix, j, axis=0)
            distance_matrix = np.delete(distance_matrix, j, axis=1)
            
            # 标记当前聚类数量减少
            n_clusters -= 1
        
        # 生成最终标签：遍历每个原始样本，找到所属的聚类ID
        self.labels_ = np.zeros(n_samples, dtype=int)
        for cluster_id, samples in enumerate(self.clusters):
            for sample_idx in samples:
                self.labels_[sample_idx] = cluster_id
        
        return self
    
    def _single_linkage(self, X):
        n_samples = X.shape[0]
        distance_matrix = np.zeros((n_samples, n_samples))
        for i in range(n_samples):
            for j in range(n_samples):
                if i == j:
                    distance_matrix[i, j] = math.inf
                else:
                    distance_matrix[i, j] = np.sqrt(np.sum((X[i] - X[j])**2))
                    distance_matrix[j, i] = distance_matrix[i, j]
        return distance_matrix

关键修改点说明 #

• 聚类映射维护：self.clusters始终记录当前所有聚类的原始样本集合，合并操作直接操作这个列表，避免索引混乱。
• 标签生成时机：最后统一生成标签，确保每个原始样本的标签对应最终的聚类ID，不会出现中间合并导致的索引错误。
• 距离矩阵更新优化：合并时先更新i行/列的距离（取单链接的最小值），再删除j行/列，保证矩阵始终对应当前的聚类集合。

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