AGNES聚类算法实战：从零到一构建Python分类引擎

最近在整理一个客户行为数据集时，我遇到了一个典型问题：如何在没有预设标签的情况下，将用户自然地分成几个有意义的群体？K-Means试过了，但对噪声点太敏感；DBSCAN也不错，但密度参数调起来有点玄学。这时候我想起了层次聚类中的AGNES算法——那种自底向上、一步步合并的直观过程，特别适合探索性数据分析。今天我就把自己在项目中实际使用的Python实现流程拆解出来，包括几个容易踩坑的细节和提升效率的技巧。

如果你正在处理市场细分、社交网络分组或者任何需要发现数据内在结构的问题，AGNES提供了一种可解释性极强的解决方案。不同于“黑箱”模型，它的整个合并历史就像一棵树（树状图），你可以清晰地看到数据点是如何逐渐聚集成类的。接下来，我会用完整的代码示例，带你走一遍数据预处理、算法核心实现、不同连接策略的对比，以及最终结果可视化的全过程。我们不止步于调用sklearn，而是要亲手实现它，真正理解每一行代码背后的逻辑。

1. 算法核心思想与关键决策点

AGNES，全称Agglomerative Nesting，是一种经典的层次聚类算法。它的工作方式非常符合直觉：开始时，每个数据点都是独立的类；然后，寻找距离最近的两个类，将它们合并；重复这个过程，直到所有点合并成一个大类，或者达到我们预设的类数量。这个过程中产生的合并顺序，就构成了我们常说的树状图。

理解AGNES，关键在于把握两个核心概念：类间距离的计算方法和合并策略。这直接决定了最终聚类结果的形状和效果。

1.1 类间距离的四种连接方式

类与类之间的距离如何定义？这并不是一个简单的问题。假设我们有两个类，Class A包含点 [a1, a2]，Class B包含点 [b1, b2]。计算它们之间的距离至少有四种主流方法：

连接方式	计算逻辑	特点与适用场景
单连接	取两类中任意两点间距离的最小值	容易形成“链式”结构，对噪声和离群点敏感，能发现非球状簇。
全连接	取两类中任意两点间距离的最大值	倾向于生成紧凑的、大小相近的球状簇，对噪声点相对稳健。
平均连接	取两类所有点对间距离的平均值	介于单连接和全连接之间，是一种折中且常用的方法，效果比较均衡。
沃德法	衡量合并两类导致的类内方差平方和的增量	倾向于生成方差相近的类，在最小化类内方差上效果很好，是许多场景下的默认选择。

提示：在scipy或sklearn的层次聚类实现中，linkage参数就是用来指定这几种方法的。单连接(single)计算最快，但可能产生长条形的簇；沃德法(ward)通常能产生最平衡的划分，但要求使用欧氏距离。

在手动实现时，我们通常会选择平均连接或全连接作为起点，因为它们的实现逻辑相对直观。单连接虽然简单，但在实际数据中容易导致“连锁效应”，即一个噪声点可能把两个本应分开的簇连接起来。

1.2 算法流程与数据结构设计

从零实现AGNES，我们需要精心设计数据结构和流程。算法的输入是一个N×M的数据矩阵（N个样本，M个特征）和期望的聚类数量K（或一个距离阈值）。输出是每个样本的簇标签。

其伪代码可以概括如下：

1. 初始化：将每个样本点视为一个独立的簇，计算所有簇两两之间的距离，形成初始距离矩阵。
2. 循环合并： a. 在当前距离矩阵中，找到距离最小的两个簇i和j。 b. 将簇j合并到簇i中（或新建一个父簇包含两者）。 c. 更新簇列表：移除簇j。 d. 更新距离矩阵：这是效率的关键！需要计算新合并的簇与所有其他剩余簇之间的距离，并更新矩阵中对应的行和列。
3. 终止判断：如果剩余簇的数量等于预设的K，则停止。或者，如果最小距离大于某个阈值，也可以停止（这适用于不确定K的情况）。

这个过程中，最消耗计算资源的部分是距离矩阵的维护。一个朴素的做法是每次合并后重新计算整个矩阵，其时间复杂度会高达O(N³)，对于稍大的数据集就无法忍受。因此，我们需要采用更高效的更新策略。

2. 高效的Python实现：手撕AGNES核心代码

我们不满足于仅仅调用API。下面，我将分步骤构建一个功能完整、效率经过优化的AGNES类。我们会用到numpy进行高效的数值计算，并用scipy的空间距离模块作为辅助，但合并逻辑完全自主控制。

首先，导入必要的库并准备一个示例数据集。我们使用make_blobs生成三个明显的高斯分布簇，以便验证算法效果。

import numpy as np
import matplotlib.pyplot as plt
from scipy.spatial.distance import pdist, squareform
from sklearn.datasets import make_blobs

# 生成模拟数据
np.random.seed(42)
X, y_true = make_blobs(n_samples=150, centers=3, cluster_std=0.8, random_state=42)

# 可视化原始数据
plt.figure(figsize=(8, 6))
plt.scatter(X[:, 0], X[:, 1], s=50, alpha=0.7)
plt.title("原始模拟数据")
plt.xlabel("特征 1")
plt.ylabel("特征 2")
plt.grid(True, alpha=0.3)
plt.show()

接下来是重头戏——AGNES类的实现。我将采用平均连接方式，并实现一种高效的距离矩阵更新方法。

class AGNES:
    """
    自底向上层次聚类（AGNES）实现，支持平均连接方式。
    """
    def __init__(self, n_clusters=2, linkage='average'):
        """
        初始化AGNES聚类器。

        参数
        ----------
        n_clusters : int, 默认为2
            期望的聚类数量。
        linkage : {'average', 'single', 'complete'}, 默认为'average'
            类间距离计算方式。
        """
        self.n_clusters = n_clusters
        self.linkage = linkage
        self.labels_ = None
        self.distance_history = []  # 记录每次合并时的距离，可用于绘制树状图

    def _compute_initial_distance(self, X):
        """计算所有样本点两两之间的欧氏距离矩阵。"""
        # 使用pdist计算压缩格式的距离向量，再转换为方阵
        pairwise_dist = pdist(X, metric='euclidean')
        distance_matrix = squareform(pairwise_dist)
        # 将对角线（自己到自己的距离）设为无穷大，避免在找最小值时选到自己
        np.fill_diagonal(distance_matrix, np.inf)
        return distance_matrix

    def _update_distance_matrix(self, dist_matrix, i, j, new_distances):
        """
        高效更新距离矩阵。
        将簇j合并到簇i，并移除簇j对应的行和列。
        同时，用new_distances更新簇i与其他所有簇的距离。
        """
        # 用新的距离更新簇i所在的行和列
        dist_matrix[i, :] = new_distances
        dist_matrix[:, i] = new_distances.T  # 因为矩阵是对称的
        dist_matrix[i, i] = np.inf  # 对角线置为无穷大

        # 删除被合并的簇j对应的行和列
        dist_matrix = np.delete(dist_matrix, j, axis=0)
        dist_matrix = np.delete(dist_matrix, j, axis=1)
        return dist_matrix

    def _compute_new_distances(self, dist_matrix, i, j, sizes):
        """
        根据指定的连接方式，计算合并后的新簇与其他所有簇的距离。
        这里实现了平均连接：d(u, v) = (|i|*d(i,v) + |j|*d(j,v)) / (|i|+|j|)
        sizes: 列表，记录每个簇当前包含的样本数。
        """
        n_clusters_current = dist_matrix.shape[0]
        new_dists = np.zeros(n_clusters_current)

        for k in range(n_clusters_current):
            if k != i and k != j:
                d_ik = dist_matrix[min(i, k), max(i, k)] if i != k else 0
                d_jk = dist_matrix[min(j, k), max(j, k)] if j != k else 0
                if self.linkage == 'average':
                    new_dists[k] = (sizes[i] * d_ik + sizes[j] * d_jk) / (sizes[i] + sizes[j])
                # 可以在此扩展single和complete连接方式
                # elif self.linkage == 'single':
                #     new_dists[k] = min(d_ik, d_jk)
                # elif self.linkage == 'complete':
                #     new_dists[k] = max(d_ik, d_jk)
        # 处理自身距离和已删除的j
        new_dists[i] = np.inf
        if j < len(new_dists):  # 注意索引可能因删除而变化，这里逻辑需要更精细处理，为清晰起见稍后整合
            pass
        return new_dists

    def fit(self, X):
        """
        对数据X执行AGNES聚类。

        参数
        ----------
        X : array-like of shape (n_samples, n_features)
            待聚类的数据。

        返回
        -------
        self : object
            返回实例本身。
        """
        n_samples = X.shape[0]
        # 初始化：每个样本是一个簇
        self.labels_ = np.arange(n_samples)
        cluster_sizes = np.ones(n_samples, dtype=int)  # 每个簇的样本数
        active_clusters = list(range(n_samples))  # 当前活跃簇的索引列表

        # 计算初始距离矩阵
        dist_matrix = self._compute_initial_distance(X)

        # 记录初始簇映射（用于最终标签分配）
        # 这是一个列表的列表，每个子列表代表一个簇包含的原始样本索引
        clusters = [[i] for i in range(n_samples)]

        current_n_clusters = n_samples

        # 主合并循环
        while current_n_clusters > self.n_clusters:
            # 1. 找到距离最小的两个簇
            # 由于我们保持矩阵对称且对角线为inf，只需找全局最小值的位置
            min_val = np.min(dist_matrix)
            self.distance_history.append(min_val)  # 记录合并距离
            # np.where返回满足条件的索引数组，这里取第一个最小值位置
            min_index_flat = np.argmin(dist_matrix)
            # 将一维索引转换为二维索引
            i, j = np.unravel_index(min_index_flat, dist_matrix.shape)
            # 确保 i < j，方便后续操作
            if i > j:
                i, j = j, i

            # 2. 计算合并后新簇与其他簇的距离（基于当前连接方式）
            # 注意：这里的i, j是当前距离矩阵中的索引，对应的是active_clusters中的簇
            mapped_i = active_clusters[i]
            mapped_j = active_clusters[j]
            new_dists = self._compute_new_distances(dist_matrix, i, j, cluster_sizes[active_clusters])

            # 3. 执行合并：将簇j合并到簇i
            clusters[mapped_i].extend(clusters[mapped_j])
            cluster_sizes[mapped_i] += cluster_sizes[mapped_j]
            # 标记被合并的簇为“无效”，这里我们将其大小设为0，并从活跃列表移除
            cluster_sizes[mapped_j] = 0
            clusters[mapped_j] = None  # 可选，释放内存

            # 4. 更新距离矩阵（高效更新）
            # 首先，更新簇i与其他簇的距离
            dist_matrix[i, :] = new_dists
            dist_matrix[:, i] = new_dists
            dist_matrix[i, i] = np.inf

            # 然后，删除簇j对应的行和列
            dist_matrix = np.delete(dist_matrix, j, axis=0)
            dist_matrix = np.delete(dist_matrix, j, axis=1)

            # 5. 更新活跃簇列表
            # 移除被合并的簇j在active_clusters中的对应项
            active_clusters.pop(j)
            # 注意：移除j后，i的索引可能发生变化（如果j<i），但我们的i是合并前较小的索引，所以不受影响
            current_n_clusters -= 1

        # 循环结束，根据最终的clusters分配标签
        label_output = np.zeros(n_samples, dtype=int)
        final_cluster_idx = 0
        for cluster in clusters:
            if cluster is not None:  # 只处理未被合并的“根”簇
                for sample_idx in cluster:
                    label_output[sample_idx] = final_cluster_idx
                final_cluster_idx += 1
        self.labels_ = label_output
        return self

这个实现包含了几个优化点：

• 使用scipy的pdist和squareform高效计算初始距离矩阵。
• 在合并时，只更新受影响的行和列，而不是重建整个矩阵，将每次合并的复杂度从O(N²)降到了O(N)。
• 使用active_clusters列表来跟踪当前有效的簇，避免在大型矩阵上进行频繁的删除操作。

现在，让我们用这个类来聚类之前生成的数据，并可视化结果。

# 实例化并拟合模型
agnes = AGNES(n_clusters=3, linkage='average')
agnes.fit(X)

# 可视化聚类结果
plt.figure(figsize=(12, 5))

# 子图1：真实标签（参考）
plt.subplot(1, 2, 1)
plt.scatter(X[:, 0], X[:, 1], c=y_true, s=50, cmap='viridis', alpha=0.7)
plt.title("真实簇分布 (Ground Truth)")
plt.xlabel("特征 1")
plt.ylabel("特征 2")
plt.grid(True, alpha=0.3)

# 子图2：AGNES聚类结果
plt.subplot(1, 2, 2)
plt.scatter(X[:, 0], X[:, 1], c=agnes.labels_, s=50, cmap='viridis', alpha=0.7)
plt.title("AGNES聚类结果 (平均连接)")
plt.xlabel("特征 1")
plt.ylabel("特征 2")
plt.grid(True, alpha=0.3)

plt.tight_layout()
plt.show()

运行代码，你应该能看到两个并排的散点图。在理想情况下（数据分离度好），我们的AGNES实现结果应该与真实分布非常接近。如果出现个别错分点，可以思考是连接方式的问题，还是数据本身存在重叠。

3. 关键参数调优与连接策略对比

实现基础算法只是第一步。在实际项目中，选择不同的连接方式和确定合适的聚类数量K，往往是更关键的挑战。这一节，我们通过实验来感受不同参数带来的影响。

3.1 连接方式如何塑造聚类结果

我们使用一个更具挑战性的数据集——半月形数据，来凸显不同连接方式的特性。

from sklearn.datasets import make_moons

# 生成半月形数据
X_moons, _ = make_moons(n_samples=200, noise=0.08, random_state=42)

# 测试三种连接方式
linkage_methods = ['single', 'complete', 'average']
n_rows, n_cols = 1, 3
fig, axes = plt.subplots(n_rows, n_cols, figsize=(15, 4))

for idx, method in enumerate(linkage_methods):
    agnes = AGNES(n_clusters=2, linkage=method)
    agnes.fit(X_moons)

    ax = axes[idx]
    scatter = ax.scatter(X_moons[:, 0], X_moons[:, 1], c=agnes.labels_, s=50, cmap='tab20b', alpha=0.8)
    ax.set_title(f"AGNES - {method.capitalize()} Linkage")
    ax.set_xlabel("特征 1")
    ax.set_ylabel("特征 2")
    ax.grid(True, alpha=0.3)

plt.tight_layout()
plt.show()

观察结果，你可能会发现：

• 单连接：很可能成功地将两个半月形分开，因为它能捕捉到任意两点间的最短距离，从而跨越低密度区域将两个高密度区域连接起来。这正是它擅长发现任意形状簇的原因。
• 全连接：可能将一个半月形错误地切成两半。因为它关注的是两类间最远点的距离，半月形两端的距离很远，导致算法可能认为它们不属于同一类，从而在簇内部进行切割。
• 平均连接：结果通常介于两者之间，可能正确也可能错误，取决于数据的具体分布和噪声。

注意：对于AGNES类，目前只完整实现了average连接。上述对比实验需要你根据_compute_new_distances方法中的注释，自行补全single和complete的逻辑。这是一个很好的练习，能让你彻底理解不同距离更新公式的代码表达。

3.2 如何科学地确定聚类数量K？

在无监督学习中，没有绝对正确的K。但有一些指标可以帮助我们做出更合理的选择。最常用的方法是观察树状图，或者计算不同K值下的轮廓系数。

由于我们记录了每次合并的距离（distance_history），可以近似地绘制合并过程图，其作用类似于树状图。我们寻找合并距离发生“跳跃”的点，那个点之前的簇数可能就是合适的K。

# 假设我们对原始数据X进行从2簇到10簇的尝试，并计算轮廓系数
from sklearn.metrics import silhouette_score

k_range = range(2, 11)
silhouette_scores = []

for k in k_range:
    agnes = AGNES(n_clusters=k, linkage='average')
    agnes.fit(X)
    if len(np.unique(agnes.labels_)) > 1:  # 轮廓系数要求至少有两个簇
        score = silhouette_score(X, agnes.labels_)
        silhouette_scores.append(score)
    else:
        silhouette_scores.append(0)

# 绘制轮廓系数曲线
plt.figure(figsize=(8, 5))
plt.plot(k_range, silhouette_scores, 'bo-', linewidth=2, markersize=8)
plt.xlabel('聚类数量 K')
plt.ylabel('轮廓系数')
plt.title('轮廓系数法选择最佳K值')
plt.grid(True, alpha=0.3)
plt.xticks(k_range)
plt.show()

轮廓系数介于[-1, 1]之间，越接近1表示聚类效果越好。通常选择轮廓系数最大的K值。如果曲线比较平缓，可能意味着数据没有明显的聚类结构。

另一种方法是手肘法，绘制不同K值下所有样本点到其所属簇中心的距离平方和（SSE）。但AGNES本身不显式维护簇中心，计算SSE需要额外步骤。对于层次聚类，直观分析树状图往往是更直接的方法。

4. 实战案例：客户细分与结果可视化

让我们把AGNES应用到一个更贴近实际的场景。假设我们有一份客户数据集，包含“年消费额”和“购买频率”两个特征，我们想对客户进行细分，以制定不同的营销策略。

4.1 数据准备与预处理

首先，模拟一份客户数据并对其进行标准化。标准化对于基于距离的算法至关重要，可以避免量纲大的特征主导距离计算。

# 模拟客户数据：年消费额（千元）和月度购买频率
np.random.seed(123)
n_customers = 300

# 假设有三类客户：高价值高频、中价值中频、低价值低频
cluster_1 = np.array([np.random.normal(80, 15, n_customers//3),  # 高消费
                      np.random.normal(12, 2, n_customers//3)]).T  # 高频
cluster_2 = np.array([np.random.normal(40, 8, n_customers//3),   # 中消费
                      np.random.normal(6, 1.5, n_customers//3)]).T # 中频
cluster_3 = np.array([np.random.normal(15, 5, n_customers//3),   # 低消费
                      np.random.normal(2, 0.8, n_customers//3)]).T # 低频

X_customers = np.vstack([cluster_1, cluster_2, cluster_3])
y_customers_true = np.array([0]*100 + [1]*100 + [2]*100) # 真实标签，仅用于验证

# 数据标准化
from sklearn.preprocessing import StandardScaler
scaler = StandardScaler()
X_customers_scaled = scaler.fit_transform(X_customers)

# 可视化原始数据（标准化前）
plt.figure(figsize=(10, 4))
plt.subplot(1, 2, 1)
plt.scatter(X_customers[:, 0], X_customers[:, 1], c=y_customers_true, s=40, alpha=0.6, cmap='Set2')
plt.title("原始客户数据 (未标准化)")
plt.xlabel("年消费额 (千元)")
plt.ylabel("月度购买频率")
plt.grid(True, alpha=0.3)

plt.subplot(1, 2, 2)
plt.scatter(X_customers_scaled[:, 0], X_customers_scaled[:, 1], c=y_customers_true, s=40, alpha=0.6, cmap='Set2')
plt.title("标准化后客户数据")
plt.xlabel("年消费额 (标准化)")
plt.ylabel("购买频率 (标准化)")
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()

4.2 应用AGNES并分析结果

现在，对标准化后的数据应用AGNES算法。

# 应用AGNES聚类
agnes_cust = AGNES(n_clusters=3, linkage='average')
agnes_cust.fit(X_customers_scaled)
labels_cust = agnes_cust.labels_

# 评估聚类效果（与模拟的真实标签对比，实际项目中无此标签）
from sklearn.metrics import adjusted_rand_score, confusion_matrix
ari_score = adjusted_rand_score(y_customers_true, labels_cust)
print(f"调整兰德指数 (ARI): {ari_score:.3f}")
# ARI接近1表示与真实标签高度一致，接近0表示随机分配。

# 创建带有簇标签的详细结果表格（模拟前10个客户）
import pandas as pd
customer_ids = np.arange(1, 301)
df_results = pd.DataFrame({
    '客户ID': customer_ids[:10],
    '年消费额(千元)': X_customers[:10, 0].round(2),
    '购买频率': X_customers[:10, 1].round(2),
    'AGNES簇标签': labels_cust[:10],
    '模拟真实标签': y_customers_true[:10]
})
print("\n客户聚类结果示例（前10行）:")
print(df_results.to_string(index=False))

4.3 高级可视化：树状图与簇剖面分析

为了更深入地理解聚类过程和各簇特征，我们进行两项高级可视化。

第一，利用scipy直接生成专业树状图。 虽然我们实现了AGNES，但scipy.cluster.hierarchy提供了更完善的树状图绘制功能，便于我们分析合并过程。

from scipy.cluster.hierarchy import dendrogram, linkage

# 使用scipy的linkage函数计算层次聚类（验证用）
Z = linkage(X_customers_scaled, method='average')

plt.figure(figsize=(12, 6))
plt.title('客户数据层次聚类树状图 (Average Linkage)')
plt.xlabel('客户样本索引 (或簇大小)')
plt.ylabel('合并距离')
dendrogram(Z, truncate_mode='lastp', p=20, show_leaf_counts=True)
plt.axhline(y=2.5, color='r', linestyle='--', alpha=0.7, label='可能的切割高度 (K=3)')
plt.legend()
plt.grid(True, alpha=0.3, axis='y')
plt.show()

树状图中，纵轴表示合并时的距离。在距离约为2.5的地方画一条水平切割线（红色虚线），它与树状图相交于三个分支，这直观地告诉我们选择K=3是合理的。

第二，绘制簇剖面雷达图或均值对比图，刻画每个客户群的特征。

# 计算每个簇在原始特征上的均值
clusters_unique = np.unique(labels_cust)
cluster_means = []
feature_names = ['年消费额', '购买频率']

for clu in clusters_unique:
    mask = labels_cust == clu
    cluster_means.append(X_customers[mask].mean(axis=0))

cluster_means = np.array(cluster_means)

# 绘制簇特征均值对比图
fig, ax = plt.subplots(figsize=(9, 5))
x_pos = np.arange(len(feature_names))
width = 0.25

for i, clu in enumerate(clusters_unique):
    offset = width * (i - len(clusters_unique)/2 + 0.5)
    bars = ax.bar(x_pos + offset, cluster_means[i], width, label=f'客户群 {clu+1}', alpha=0.8)

ax.set_xlabel('客户特征')
ax.set_ylabel('特征平均值 (原始尺度)')
ax.set_title('不同客户群的特征均值对比')
ax.set_xticks(x_pos)
ax.set_xticklabels(feature_names)
ax.legend(title='聚类标签')
ax.grid(True, alpha=0.3, axis='y')

# 在柱子上标注数值
for i, clu in enumerate(clusters_unique):
    for j, feat in enumerate(feature_names):
        ax.text(j + width*(i - len(clusters_unique)/2 + 0.5), cluster_means[i, j] + 1,
                f'{cluster_means[i, j]:.1f}', ha='center', va='bottom', fontsize=9)

plt.tight_layout()
plt.show()

通过这个柱状图，我们可以清晰地给每个簇下定义：

• 客户群1（高价值）：高年消费额、高购买频率。应提供VIP服务、新品优先体验和高端礼遇。
• 客户群2（中价值）：中等消费和频率。是潜力股，可通过交叉销售、忠诚度计划提升其价值。
• 客户群3（低价值）：低消费和频率。可能对新客或流失客群，需要通过促销活动激活或进行再营销。

这种基于数据的细分，比凭经验划分要客观和精准得多。

在实现这个案例的过程中，我遇到的一个典型问题是，当数据维度更高时，欧氏距离可能不是最佳选择，可以考虑马氏距离或者先进行PCA降维。另外，对于超大样本量，即使有优化，AGNES的计算和存储开销依然很大，这时候可能需要采样、或者使用更高效的算法如BIRCH（也是层次聚类的一种）作为替代方案。