简介apcluster.zip 提供了一份基于 Python 的 AP 聚类算法实现AP 聚类Affinity Propagation是一种无需预先指定簇数的无中心聚类方法通过样本间的消息传递职责与可用性自动确定聚类中心特别适合数据探索、图像分割、文本挖掘等未知簇数的场景。这份资源适合正在学习聚类算法或需要在 Python 中快速应用 AP 算法的开发者也适合作为算法课程的辅助材料。压缩包体积仅 1KB虽只包含 1 个 py 文件但代码结构简练覆盖了距离矩阵计算、消息迭代更新、收敛判断和簇分配等核心逻辑目前已有 781 人学习下载过。通过这份代码可以深入理解 AP 算法的内部机制包括 R 与 A 的更新公式、阻尼系数的作用以及如何根据最终矩阵输出聚类结果同时也可作为基础模板帮助开发者在实际项目中快速集成或二次开发 AP 聚类功能无论是处理二维样本还是高维特征都能为后续聚类分析提供便捷起点。1. 为什么 AP 聚类在 Python 里值得自己手动实现KMeans 这类划分式聚类有个让人难受的前提你得先告诉它分成几类。AP 聚类Affinity Propagation想解决的是把这个 K 也交给数据自己决定。它把每个样本都看作潜在的聚类中心通过点与点之间互相交换“你适合当中心”和“我信任你”两类消息迭代收敛后自动给出聚类中心和每个点的归属。我第一次看它的论文时觉得公式绕后来发现手写一个 50 行的 Python 实现反而比读公式更容易建立直觉。下面按我常用的顺序展开先讲 R 和 A 两条更新公式再给一份从零写的 AP 聚类 Python 代码最后落到 sklearn 的落地调用与调参技巧。适合刚开始用 Python 做聚类分析、又不想停在 fit 这一步的人。2. AP 聚类的消息传递原理Responsibility 与 Availability 的迭代逻辑2.1 输入是相似度矩阵不是原始特征AP 聚类把样本之间的相似度作为唯一输入。假设有 n 个样本就有一张 n×n 的相似度矩阵 S。S[i][k] 越大表示样本 i 越愿意选择 k 作为自己的代表S[i][i] 称为偏好值 preference可以统一给一个值也可以按样本单独设置。这个设计带来的直接后果是算法不要求每个样本都写在同一个特征空间里。比如你只有两两之间的距离、重叠度或业务评分关系也可以直接当作 S 来跑。对 Python 实现来说最省事的相似度是负欧氏距离。import numpy as np from scipy.spatial.distance import cdist def make_similarity(X): dist_mat cdist(X, X, metriceuclidean) S -dist_mat np.fill_diagonal(S, 0.0) # 对角线先留空后面会被 preference 覆盖 return S这段代码的作用是把特征矩阵 X 换成负距离矩阵距离为 0 的两个样本相似度为 0距离越远相似度越小。np.fill_diagonal(S, 0.0)是把自身距离置为 0避免对角线大值干扰后续计算。参数metric可以改成cityblock或cosine只要满足距离越近相似度越高即可。2.2 两条核心消息Responsibility 和 AvailabilityAP 聚类迭代维护两个矩阵R 和 A。R[i][k] 是责任值样本 i 对“k 应该成为聚类中心”的支持度。它等于 S[i][k] 减去除 k 之外所有竞争者的得分最大值。竞争者得分 A[i][k] S[i][k]。这里加 A 是有意的如果某个 k 已经获得了别人的支持A 为正它才更值得被当作备选。公式为R[i][k] S[i][k] - max_{k ! k}(A[i][k] S[i][k])A[i][k] 是可用性样本 i 是否承认 k 是中心由除 i 自己之外其他样本对 k 的正向责任积累决定。非对角线公式为A[i][k] min(0, R[k][k] sum_{i ! i, k} max(0, R[i][k]))对角线单独算A[k][k] sum_{i ! k} max(0, R[i][k])。对角线可用性越大k 越可能成为中心。def update_r(S, A): n S.shape[0] R np.empty_like(S) for k in range(n): cand A S cand[:, k] -np.inf # 排除 k 自己 R[:, k] S[:, k] - cand.max(axis1) return R def update_a(R): n R.shape[0] pos np.maximum(R, 0) pos[np.arange(n), np.arange(n)] 0 col_sum pos.sum(axis0) A np.empty_like(R) for k in range(n): A[:, k] np.minimum(0, R[k, k] col_sum[k] - pos[:, k]) A[k, k] col_sum[k] return A这两段是手写 AP 聚类最核心的部分。update_r对每一列 k 计算所有 i 的责任先构造候选矩阵 cand在候选里把第 k 列去掉再用列最大值做对比对象本质上就是“你的竞争力比当前最好的对手差多少”。update_a先取 R 的正数部分把对角线清零然后按列求和得到 col_sumA[i][k] 就是 R[k][k] 加上除去 i 自己之后的正向责任总和并截断到 0 以下。对角线 A[k][k] 直接取 col_sum[k]因为它代表的是其他所有样本对 k 的支持累积。注意这段代码每轮会创建多个临时矩阵n 在 500 以内用起来没有压力。2.3 preference 和 damping 是一对互相拉扯的旋钮preference 是 S 对角线填充的值。它不代表任何样本关系而是算法给每个点的“当中心先验值”。preference 调得越大更多样本会被激活成中心聚类数变多调得越小中心越少簇也越粗。常见做法是先取相似度矩阵非对角线元素的中位数跑一版结果再根据业务目标按倍率缩放。damping 是阻尼系数每次迭代把新值与旧值加权合并R_new lambda * R_old (1-lambda) * R_calc。lambda 越大更新越平滑越不会出现两个中心来回横跳的情况但代价是需要的迭代次数变多。参数常见初始值调大后的表现建议preferenceS 非对角线中位数聚类中心变多、簇变小先用中位数再乘 0.5 或 2 试几轮damping0.5迭代步长变小收敛变慢出现中心反复变动时调到 0.9 或 0.95max_iter200迭代上限放宽n 大时先用 100 跑通再加大convergence_iter15中心连续不变多少轮后停止想更快收敛可以降到 5但可靠性略降2.4 什么是“收敛”只看中心是否改变AP 不会等到 R 和 A 完全不变才停止那样太慢。实用的停止条件有两个一是连续 convergence_iter 轮计算出的候选中心集合完全一致二是迭代次数达到 max_iter。候选中心集合由 R[i][i] A[i][i] 0 的样本组成。这个判据很直观一个点自己的责任与可用性之和大于 0说明它既认为自己适合当中心又拿到了其他人的支持。在实现中我会把每次得到的中心索引排序后比较避免因顺序不同误判。3. 用 Python 从零实现 AP 聚类最小可运行代码与参数说明3.1 造一份演示数据先把相似度矩阵算出来网上搜“免费 python 源码大全”时经常会看到把 AP 聚类封装成类的代码但大多数只调 sklearn。与其猜封装不如先用一份能看见每一步的代码。我先生成三团高斯分布的数据每团 30 个点分布中心分别为 (0,0)、(5,5)、(8,1)。这样既能看到聚类中心收敛到真实中心附近又不会因为数据太理想而看不出问题。import numpy as np rng np.random.default_rng(42) true_centers [[0, 0], [5, 5], [8, 1]] X np.vstack([ rng.normal(locc, scale0.8, size(30, 2)) for c in true_centers ]) print(X.shape) # (90, 2)这段代码用default_rng固定随机种子保证你在不同机器上跑出的结果一致。normal的loc是簇中心scale是标准差调大 scale 会让三团数据重叠可以观察 AP 在簇边界模糊时的表现。3.2 完整的 AP 迭代函数把上一章的update_r和update_a组合进一个函数加入 damping、收敛判断和标签分配。这是最小可运行版本。import numpy as np from scipy.spatial.distance import cdist def update_r(S, A): n S.shape[0] R np.empty_like(S) for k in range(n): cand A S cand[:, k] -np.inf R[:, k] S[:, k] - cand.max(axis1) return R def update_a(R): n R.shape[0] pos np.maximum(R, 0) pos[np.arange(n), np.arange(n)] 0 col_sum pos.sum(axis0) A np.empty_like(R) for k in range(n): A[:, k] np.minimum(0, R[k, k] col_sum[k] - pos[:, k]) A[k, k] col_sum[k] return A def ap_cluster(S, preferenceNone, damping0.5, max_iter200, convergence_iter15): n S.shape[0] if preference is None: preference np.median(S[S ! 0]) S S.copy() np.fill_diagonal(S, preference) R np.zeros((n, n)) A np.zeros((n, n)) last_centers None same_count 0 for it in range(max_iter): R damping * R (1 - damping) * update_r(S, A) A damping * A (1 - damping) * update_a(R) centers np.flatnonzero(np.diag(R) np.diag(A) 0) if len(centers) 0: centers np.array([np.argmax(np.diag(S))]) if last_centers is not None and np.array_equal( np.sort(centers), np.sort(last_centers)): same_count 1 if same_count convergence_iter: break else: same_count 0 last_centers centers labels np.argmax(S[:, centers], axis1) return centers, labels, it 1这段代码的流程是先把 preference 填进 S 的对角线循环里先更新 R用新 R 更新 A然后通过 R 和 A 的对角线之和找出候选中心。如果连续convergence_iter轮中心集合相同就提前结束。最后一行把每个样本归到相似度最大的中心上。参数preference不传时用非对角线中位数damping控制更新平滑度max_iter是硬上限convergence_iter是温和停止阈值。3.3 在演示数据上跑一遍看迭代次数和中心S -cdist(X, X) centers, labels, n_iter ap_cluster(S) print(聚类中心索引:, centers) print(迭代轮数:, n_iter)centers是 X 的行索引也就是真实样本的 ID。比如输出[10 55 81]说明第 10、55、81 号样本是三个簇的代表。labels的长度与 X 的样本数一致每个值是中心列表的下标不是全局样本 ID。如果迭代轮数等于 200说明算法到max_iter还没收敛优先把damping调到 0.9 再跑。数据规模建议先跑的参数说明n 300默认 damping0.5, max_iter200一般 20 轮内能收敛300 n 1000damping0.7, max_iter300消息矩阵更大需要一点阻尼n 1000先用测试子集跑通全量跑之前先看中心是否合理3.4 手写版本和 sklearn 版本差在哪手写版最大的价值是把迭代逻辑暴露在眼前方便你改消息公式或加业务约束。但手写版与 sklearn 的AffinityPropagation有几个边界差异sklearn 对update_r和update_a做了更多数值保护比如处理全零行它支持设置affinityprecomputed来绕过特征距离计算它的标签分配策略更保守会先对相似度做归一化再比较。手写版适合学习以及在小样本上做实验生产环境直接使用 sklearn 调用不要把不可控的循环应用到大数据上。4. 用 scikit-learn 快速落地 AP 聚类环境配置与调参实战4.1 在 Linux 或本地把 Python 环境备好Linux 系统安装 python 的教程很多这里只说我常用的最快路径创建虚拟环境安装三个包不污染系统 Python。如果你平时用 VS Code只要让解释器指向.venv路径就能直接在编辑环境里跑起来。python -m venv .venv source .venv/bin/activate pip install numpy scipy scikit-learn matplotlib参数说明python -m venv创建隔离环境source .venv/bin/activate激活它Linux 上如果缺 pip先通过系统包管理装好 python3-pip。安装后用python -c import sklearn; print(sklearn.__version__)确认版本0.22 以上对preference的处理更符合常识。4.2 用 AffinityPropagation 写一个最小调用使用 sklearn 的AffinityPropagation默认使用负欧氏距离作为相似度不需要手动构造矩阵。注意它的构造参数和第 3 章的ap_cluster一一对应但标签分配、收敛判断都已经做过较多调优。from sklearn.cluster import AffinityPropagation import numpy as np model AffinityPropagation( damping0.5, max_iter200, convergence_iter15, preferenceNone, affinityeuclidean ) model.fit(X) print(model.cluster_centers_indices_) print(model.labels_) print(model.n_iter_)代码逻辑很简单model.fit(X)完成整个消息迭代cluster_centers_indices_是中心对应的原始样本行号labels_是每个样本的簇标签n_iter_是实际迭代次数如果它等于max_iter说明没有在迭代上限前收敛。affinityeuclidean时sklearn 内部用负欧氏距离计算相似度当你想用自定义相似度改成affinityprecomputed同时fit传入的必须是方阵。4.3 三个必调参数的经验区间AP 聚类对参数的敏感度比 KMeans 高下面这张表是我在不同数据集上最容易踩到的边界。参数默认值经验区间效果preference中位数中位数的 0.5 ~ 2 倍用 0.5 倍会减少聚类数2 倍会增加聚类数damping0.50.5 ~ 0.95越大迭代越平滑但需要更多轮convergence_iter1510 ~ 30调大能减少误收敛但会增加运行时间一个常见调参流程先跑默认参数看聚类数如果明显多个簇挤到一起把preference降到中位数的 0.5 倍如果两个中心反复争夺同一样本把damping调大到 0.9。对业务输出影响最直接的是 preference因为它直接决定聚类中心的密度。4.4 数据规模大了以后怎么绕开 O(N^2T) 的时间成本AP 聚类的时间复杂度是 O(N^2·T)T 是迭代次数。N1000 时还算轻松N5000 就要考虑内存和核时。最常见的做法是先降维再聚类不要惊讶于中心质量下降。另一种方式是把数据用 MiniBatchKMeans 粗分成 50 个桶然后在每个桶内单独跑 AP再把所有桶里的聚类中心汇总成第二次 AP 的样本。如果业务上允许也可以把相似度矩阵稀疏化只有 0~1 的非负相似度才能直接用稀疏矩阵因为稀疏矩阵里的 0 表示“没有相似度”而不是负距离。对负欧氏距离做稀疏化时非邻居必须填一个足够小的负数否则 0 会被当成完美相似度聚类结果会完全乱掉。不少开源代码在这个地方栽过跟头所以单独提一下。5. 聚类中心出来之后先别急着收工可视化与有效性验证5.1 把中心索引映射回原始样本AP 聚类中心是真实的样本点不像 KMeans 那样得到一个均值向量。这在业务解释上很有用但也带来一个问题它必须从现有样本里选人数据里如果没有一个够格的“典型用户”中心会落在边界上。拿到中心的第一个动作应该是打印中心样本的原始特征确认它和簇内其他样本的关系。exemplar_indices model.cluster_centers_indices_ for i, idx in enumerate(exemplar_indices): print(f簇 {i}: 中心样本 #{idx}, 特征值 {X[idx]})如果中心样本的特征明显是中位数而不是边缘值说明这个簇选得合理如果中心样本和簇内 90% 的样本距离都远调小 preference 再试。5.2 用轮廓系数做横向对比轮廓系数对 AP 聚类并非完美因为 AP 适合非凸簇而轮廓系数偏向凸型分布但横向对比同一数据集下不同 preference 的参数仍然有参考价值。代码很简单from sklearn.metrics import silhouette_score score silhouette_score(X, model.labels_) print(fsilhouette_score: {score:.3f})注意这个分数是在所有样本点两两距离的基础上算的n5000 时也会很耗时。更快的做法是用sample_size1000参数采样计算。如果分数低于 0.2优先怀疑 preference 设得太大聚类数过多导致每个簇太碎。5.3 中心数量异常时先查这三个地方AP 聚类调参失败最常表现在两个极端中心太多或者完全没有中心。先看model.n_iter_是否等于max_iter如果是把 damping 提到 0.9 再跑再看 preference 是否设成中位数的几千倍那会导致每个样本都成为中心如果 preference 被设成很大的负数中心会塌缩到 1 个。调试时可以用二分法固定聚类数把 preference 当作未知数每次按中位数的 0.5 倍和 2 倍做一次扫描画一条聚类数随 preference 变化的曲线比猜参数快得多。最后检查每个簇里离中心最远的样本AP 并不保证每个点都被温柔对待某些样本在迭代结束后对所有中心都持负面态度却被强制归到了最近中心。如果你的业务允许把这种样本当作离群点可以在分配标签前加一个相似度阈值低于阈值的样本独立成“未分配”簇这一步比调 iteration 更能挽救业务解释。本文还有配套的精品资源点击获取