第12章:无监督学习概论(Introduction to Unsupervised Learning)
一、导读
12.1 本章学习目标
第12章是全书的最后一章,介绍无监督学习的基本概念和方法。与监督学习不同,无监督学习的数据没有标签,目标是从数据中发现隐藏的结构和模式。第2版新增了这一章,补充了无监督学习这一重要领域。
本章的核心学习目标包括:
- 理解无监督学习的基本概念:与监督学习的区别,主要任务类型
- 掌握聚类分析:K-means、层次聚类、DBSCAN等经典方法
- 掌握降维方法:PCA(主成分分析)的原理和应用
- 理解概率模型估计:极大似然估计在无监督学习中的应用
- 了解奇异值分解(SVD):矩阵分解的核心方法
- 实现经典的无监督学习算法
12.2 为什么本章如此重要
无监督学习是机器学习的重要组成部分。现实世界中,大量的数据是没有标签的——互联网上的文本、图像、视频,传感器采集的信号,生物基因数据等。如何从这些无标签数据中提取有用的信息,是机器学习面临的重要挑战。
无监督学习的方法也是许多复杂模型的组成部分。预训练(pre-training)是深度学习的核心范式,而许多预训练方法(如自编码器、生成对抗网络)本质上都是无监督学习。
12.3 核心问题引导
在阅读本章时,请带着以下问题思考:
无监督学习与监督学习的本质区别是什么?
聚类的核心目标是什么?不同的聚类算法有什么不同的"好聚类"定义?
PCA的直觉是什么?为什么保留最大方差的方向是好的?
SVD和PCA有什么关系?SVD在数据压缩中如何应用?
如何评估无监督学习的效果?没有标签怎么知道结果好不好?
无监督学习在深度学习时代有什么新的应用?
二、核心概念详解
12.1 无监督学习概述
12.1.1 定义
无监督学习是从无标签数据中学习模型的方法。给定训练数据 {(x_1), (x_2), ..., (x_N)}(没有对应的标签 y),目标是发现数据中的潜在结构。
12.1.2 主要任务
- 聚类(Clustering):将数据分成若干组,组内相似、组间不同
- 降维(Dimensionality Reduction):将高维数据映射到低维空间
- 密度估计(Density Estimation):估计数据的概率分布
- 异常检测(Anomaly Detection):发现与大多数数据不同的异常点
import numpy as np
def unsupervised_learning_overview():
"""无监督学习概述"""
print("无监督学习的主要任务:")
print()
print("1. 聚类: 将数据分成若干组")
print(" K-means, 层次聚类, DBSCAN, GMM")
print()
print("2. 降维: 减少特征数量")
print(" PCA, t-SNE, UMAP, 自编码器")
print()
print("3. 密度估计: 学习数据的分布")
print(" 高斯混合模型, 核密度估计")
print()
print("4. 异常检测: 发现异常数据点")
print(" 孤立森林, One-Class SVM, LOF")
print()
print("核心挑战: 没有标签,如何评估学习效果?")
unsupervised_learning_overview()12.2 聚类分析
12.2.1 K-means聚类
K-means是最经典的聚类算法。目标是将N个样本分成K个簇,最小化簇内平方和:
J = Σ_k Σ_{x∈C_k} ||x - μ_k||^2其中 μ_k 是簇 C_k 的中心。
算法:
随机初始化K个中心
将每个样本分配到最近的中心
重新计算每个簇的中心
重复2-3直到收敛
import numpy as np
class KMeans:
def __init__(self, k=3, max_iter=100):
self.k = k
self.max_iter = max_iter
def fit(self, X):
N, D = X.shape
# 初始化中心(K-means++)
self.centroids = np.zeros((self.k, D))
idx = np.random.randint(N)
self.centroids[0] = X[idx]
for i in range(1, self.k):
dists = np.min([np.sum((X - c) ** 2, axis=1) for c in self.centroids[:i]], axis=0)
probs = dists / dists.sum()
idx = np.random.choice(N, p=probs)
self.centroids[i] = X[idx]
for iteration in range(self.max_iter):
# 分配
dists = np.array([np.sum((X - c) ** 2, axis=1) for c in self.centroids])
labels = np.argmin(dists, axis=0)
# 更新中心
new_centroids = np.zeros_like(self.centroids)
for j in range(self.k):
mask = labels == j
if np.sum(mask) > 0:
new_centroids[j] = X[mask].mean(axis=0)
else:
new_centroids[j] = self.centroids[j]
# 检查收敛
if np.allclose(new_centroids, self.centroids):
print(f"K-means收敛于第 {iteration + 1} 次迭代")
break
self.centroids = new_centroids
self.labels_ = labels
return self
def predict(self, X):
dists = np.array([np.sum((X - c) ** 2, axis=1) for c in self.centroids])
return np.argmin(dists, axis=0)
# 测试
np.random.seed(42)
X1 = np.random.randn(100, 2) + np.array([3, 3])
X2 = np.random.randn(100, 2) + np.array([-2, -2])
X3 = np.random.randn(100, 2) + np.array([3, -2])
X = np.vstack([X1, X2, X3])
kmeans = KMeans(k=3)
kmeans.fit(X)
print(f"聚类中心:\n{kmeans.centroids}")
print(f"各簇样本数: {[np.sum(kmeans.labels_ == i) for i in range(3)]}")12.2.2 层次聚类
层次聚类不需要指定簇数,而是构建一棵聚类树(dendrogram)。
凝聚层次聚类(自底向上):
每个样本初始化为一个簇
合并最相似的两个簇
重复直到所有样本合并为一个簇
簇间距离的定义:
- 单连接(Single Linkage):两簇中最近样本的距离
- 全连接(Complete Linkage):两簇中最远样本的距离
- 平均连接(Average Linkage):两簇中所有样本对的平均距离
import numpy as np
def agglomerative_clustering(X):
"""凝聚层次聚类"""
N = len(X)
# 计算距离矩阵
dist_matrix = np.zeros((N, N))
for i in range(N):
for j in range(i + 1, N):
d = np.linalg.norm(X[i] - X[j])
dist_matrix[i, j] = d
dist_matrix[j, i] = d
# 初始化:每个样本一个簇
clusters = {i: [i] for i in range(N)}
merge_history = []
active = set(range(N))
while len(active) > 1:
# 找到最近的两个簇
min_dist = float('inf')
merge_i, merge_j = -1, -1
active_list = list(active)
for idx_a in range(len(active_list)):
for idx_b in range(idx_a + 1, len(active_list)):
i, j = active_list[idx_a], active_list[idx_b]
# 平均连接
dists = []
for a in clusters[i]:
for b in clusters[j]:
dists.append(dist_matrix[a, b])
avg_dist = np.mean(dists)
if avg_dist < min_dist:
min_dist = avg_dist
merge_i, merge_j = i, j
# 合并
new_cluster = clusters[merge_i] + clusters[merge_j]
new_id = max(clusters.keys()) + 1
clusters[new_id] = new_cluster
del clusters[merge_i]
del clusters[merge_j]
active.remove(merge_i)
active.remove(merge_j)
active.add(new_id)
merge_history.append((merge_i, merge_j, min_dist))
return merge_history
# 小规模示例
np.random.seed(42)
X_small = np.random.randn(8, 2)
history = agglomerative_clustering(X_small)
print("层次聚类合并历史:")
for i, (a, b, dist) in enumerate(history):
print(f" 步骤{i+1}: 合并簇{a}和簇{b}, 距离={dist:.4f}")12.2.3 DBSCAN
DBSCAN(Density-Based Spatial Clustering of Applications with Noise)基于密度的聚类方法。
核心思想:
- 核心点:半径 ε 内至少有 min_samples 个点
- 边界点:在核心点的 ε 半径内,但自身不是核心点
- 噪声点:既不是核心点也不是边界点
import numpy as np
class DBSCAN:
def __init__(self, eps=0.5, min_samples=5):
self.eps = eps
self.min_samples = min_samples
def fit(self, X):
N = len(X)
labels = np.full(N, -1) # -1表示噪声
cluster_id = 0
for i in range(N):
if labels[i] != -1:
continue
# 找到ε邻域内的点
neighbors = self._region_query(X, i)
if len(neighbors) < self.min_samples:
labels[i] = -1 # 噪声
else:
# 扩展簇
self._expand_cluster(X, labels, i, neighbors, cluster_id)
cluster_id += 1
self.labels_ = labels
return self
def _region_query(self, X, i):
dists = np.sqrt(np.sum((X - X[i]) ** 2, axis=1))
return np.where(dists <= self.eps)[0]
def _expand_cluster(self, X, labels, i, neighbors, cluster_id):
labels[i] = cluster_id
queue = list(neighbors)
while queue:
j = queue.pop(0)
if labels[j] == -1:
labels[j] = cluster_id # 噪声变边界点
if labels[j] != -1 and labels[j] != cluster_id:
continue
labels[j] = cluster_id
j_neighbors = self._region_query(X, j)
if len(j_neighbors) >= self.min_samples:
for nb in j_neighbors:
if labels[nb] == -1 or labels[nb] == -1:
queue.append(nb)
# 测试
np.random.seed(42)
# 生成环形数据
t = np.linspace(0, 2 * np.pi, 100)
X_ring1 = np.column_stack([np.cos(t), np.sin(t)]) + np.random.randn(100, 2) * 0.1
X_ring2 = np.column_stack([3 * np.cos(t), 3 * np.sin(t)]) + np.random.randn(100, 2) * 0.1
X_dbscan = np.vstack([X_ring1, X_ring2])
db = DBSCAN(eps=0.5, min_samples=5)
db.fit(X_dbscan)
n_clusters = len(set(db.labels_)) - (1 if -1 in db.labels_ else 0)
n_noise = list(db.labels_).count(-1)
print(f"DBSCAN发现 {n_clusters} 个簇, {n_noise} 个噪声点")12.3 主成分分析(PCA)
12.3.1 PCA的直觉
PCA的目标是找到数据中方差最大的方向(主成分),然后将数据投影到这些方向上实现降维。
直觉:方差最大的方向包含了数据中最多的信息。
12.3.2 PCA的数学推导
给定数据中心化后的协方差矩阵:
C = (1/N) X^T XPCA找到 C 的特征值和特征向量。特征值从大到小排列,对应的特征向量就是主成分方向。
降维:保留前 k 个主成分,将数据投影到 k 维空间。
import numpy as np
class PCA:
def __init__(self, n_components=2):
self.n_components = n_components
def fit(self, X):
# 中心化
self.mean = X.mean(axis=0)
X_centered = X - self.mean
# 协方差矩阵
cov_matrix = np.cov(X_centered.T)
# 特征分解
eigenvalues, eigenvectors = np.linalg.eigh(cov_matrix)
# 按特征值从大到小排序
idx = np.argsort(eigenvalues)[::-1]
self.eigenvalues = eigenvalues[idx]
self.eigenvectors = eigenvectors[:, idx]
# 保留前k个主成分
self.components = self.eigenvectors[:, :self.n_components]
return self
def transform(self, X):
X_centered = X - self.mean
return X_centered @ self.components
def inverse_transform(self, X_transformed):
return X_transformed @ self.components.T + self.mean
def explained_variance_ratio(self):
return self.eigenvalues[:self.n_components] / self.eigenvalues.sum()
# 测试
np.random.seed(42)
# 生成相关数据
X_pca = np.random.randn(200, 4)
X_pca[:, 1] = X_pca[:, 0] * 0.8 + np.random.randn(200) * 0.3
X_pca[:, 2] = X_pca[:, 0] * 0.5 + np.random.randn(200) * 0.5
X_pca[:, 3] = np.random.randn(200) # 独立噪声
pca = PCA(n_components=2)
pca.fit(X_pca)
print("PCA结果:")
print(f"各主成分的方差解释比: {pca.explained_variance_ratio()}")
print(f"前2个主成分解释的总方差: {pca.explained_variance_ratio().sum():.4f}")
X_reduced = pca.transform(X_pca)
print(f"降维后维度: {X_reduced.shape}")12.4 奇异值分解(SVD)
12.4.1 SVD的定义
任意矩阵 M 可以分解为:
M = U Σ V^T其中 U 和 V 是正交矩阵,Σ 是对角矩阵(奇异值从大到小排列)。
12.4.2 SVD与PCA的关系
对数据中心化后的矩阵 X 进行SVD:
- V 的列就是PCA的主成分方向
- Σ 的平方除以 N 就是PCA的特征值
import numpy as np
def svd_demo():
"""SVD演示"""
np.random.seed(42)
# 创建一个矩阵
M = np.random.randn(100, 5)
# SVD
U, S, Vt = np.linalg.svd(M, full_matrices=False)
print(f"矩阵 M: {M.shape}")
print(f"U: {U.shape}, S: {S.shape}, Vt: {Vt.shape}")
print(f"奇异值: {S[:3]}...")
# 低秩近似
k = 2
M_approx = U[:, :k] @ np.diag(S[:k]) @ Vt[:k, :]
reconstruction_error = np.linalg.norm(M - M_approx) / np.linalg.norm(M)
print(f"\n用前{k}个奇异值近似的相对误差: {reconstruction_error:.4f}")
# 数据压缩
original_size = M.size
compressed_size = k * (M.shape[0] + M.shape[1] + 1)
print(f"压缩比: {original_size / compressed_size:.1f}x")
svd_demo()12.5 无监督学习的评估
12.5.1 聚类评估指标
当有真实标签时:
- 调整兰德指数(ARI):衡量聚类结果与真实标签的一致性
- 归一化互信息(NMI):基于信息论的评估指标
当没有真实标签时:
- 轮廓系数(Silhouette Score):衡量簇内紧密度和簇间分离度
- Davies-Bouldin指数:衡量簇间的相似度
import numpy as np
def silhouette_score_simple(X, labels):
"""简化的轮廓系数"""
N = len(X)
unique_labels = np.unique(labels)
silhouettes = np.zeros(N)
for i in range(N):
# 同簇平均距离 a(i)
same_cluster = X[labels == labels[i]]
if len(same_cluster) > 1:
a_i = np.mean(np.sqrt(np.sum((same_cluster - X[i]) ** 2, axis=1)))
else:
a_i = 0
# 最近异簇平均距离 b(i)
b_i = float('inf')
for label in unique_labels:
if label == labels[i]:
continue
other_cluster = X[labels == label]
if len(other_cluster) > 0:
mean_dist = np.mean(np.sqrt(np.sum((other_cluster - X[i]) ** 2, axis=1)))
b_i = min(b_i, mean_dist)
silhouettes[i] = (b_i - a_i) / max(a_i, b_i)
return np.mean(silhouettes)
# 测试
np.random.seed(42)
X1 = np.random.randn(50, 2) + np.array([3, 3])
X2 = np.random.randn(50, 2) + np.array([-2, -2])
X_eval = np.vstack([X1, X2])
labels_eval = np.array([0] * 50 + [1] * 50)
score = silhouette_score_simple(X_eval, labels_eval)
print(f"轮廓系数: {score:.4f}")
print("轮廓系数范围[-1, 1],越大表示聚类越好")三、实战案例
3.1 聚类算法对比
import numpy as np
def compare_clustering():
"""对比不同聚类算法"""
np.random.seed(42)
# 生成不同类型的数据
# 1. 球形簇
X1 = np.vstack([
np.random.randn(80, 2) + [2, 2],
np.random.randn(80, 2) + [-2, -2],
np.random.randn(80, 2) + [2, -2],
])
# 2. 环形数据
t = np.linspace(0, 2 * np.pi, 120)
X2_inner = np.column_stack([np.cos(t), np.sin(t)]) + np.random.randn(120, 2) * 0.1
X2_outer = np.column_stack([3 * np.cos(t), 3 * np.sin(t)]) + np.random.randn(120, 2) * 0.1
X2 = np.vstack([X2_inner, X2_outer])
print("聚类算法适用场景:")
print()
print(f"{'算法':>15} {'球形簇':>10} {'环形数据':>10} {'噪声鲁棒':>10}")
print("-" * 48)
print(f"{'K-means':>15} {'适合':>10} {'不适合':>10} {'一般':>10}")
print(f"{'层次聚类':>15} {'适合':>10} {'不适合':>10} {'一般':>10}")
print(f"{'DBSCAN':>15} {'适合':>10} {'适合':>10} {'好':>10}")
print(f"{'GMM':>15} {'适合':>10} {'一般':>10} {'一般':>10}")
compare_clustering()3.2 PCA降维可视化
import numpy as np
def pca_dimensionality_demo():
"""PCA降维演示"""
np.random.seed(42)
# 生成高维数据(10维)
n_samples = 200
X_high = np.random.randn(n_samples, 10)
# 添加一些结构
X_high[:, 1] = X_high[:, 0] * 0.8 + np.random.randn(n_samples) * 0.2
X_high[:, 2] = X_high[:, 0] * 0.5 + np.random.randn(n_samples) * 0.3
X_high[:, 3] = X_high[:, 1] * 0.6 + np.random.randn(n_samples) * 0.3
# PCA降维
pca = PCA(n_components=2)
pca.fit(X_high)
print("PCA降维分析:")
print(f"原始维度: {X_high.shape[1]}")
print(f"降维后维度: 2")
print(f"各主成分方差解释比:")
for i, ratio in enumerate(pca.explained_variance_ratio()):
print(f" PC{i+1}: {ratio:.4f}")
print(f"前2个主成分解释的总方差: {pca.explained_variance_ratio().sum():.4f}")
X_2d = pca.transform(X_high)
print(f"\n降维后数据范围:")
print(f" PC1: [{X_2d[:, 0].min():.2f}, {X_2d[:, 0].max():.2f}]")
print(f" PC2: [{X_2d[:, 1].min():.2f}, {X_2d[:, 1].max():.2f}]")
pca_dimensionality_demo()四、本章小结
第12章介绍了无监督学习的基本方法:
无监督学习:从无标签数据中学习,主要任务包括聚类、降维、密度估计、异常检测。
聚类:K-means(球形簇)、层次聚类(树形结构)、DBSCAN(密度聚类,可处理任意形状)。
PCA:找到方差最大的方向进行降维,通过特征分解协方差矩阵实现。
SVD:矩阵分解的核心方法,与PCA密切相关,可用于数据压缩和降维。
评估:轮廓系数等指标可以在没有标签的情况下评估聚类质量。
无监督学习的核心启示是:数据中隐藏着丰富的结构,即使没有标签,我们也能从中发现有意义的模式。
五、思考与练习
思考题
无监督学习与监督学习的本质区别是什么?在什么场景下应该使用无监督学习?
K-means为什么倾向于产生球形簇?如何让它处理非球形数据?
PCA的"方差最大"和"重构误差最小"是等价的吗?为什么?
SVD和特征分解有什么关系?在什么情况下SVD比特征分解更适用?
如何确定聚类的最优簇数K?有哪些方法?
编程练习
从零实现K-means算法(包括K-means++初始化),在不同数据上测试。
实现PCA,在高维数据上进行降维,分析各主成分的方差解释比。
实现DBSCAN,在环形数据上对比K-means和DBSCAN的聚类效果。