深度学习学习教程，从入门到精通，表示学习 — 知识点与代码案例详解(15）
·
表示学习 — 知识点与代码案例详解
一、贪心逐层无监督预训练（Greedy Layer-wise Unsupervised Pretraining）
1.1 核心知识点
核心思想： 每一层网络单独进行无监督学习（如自编码器），将每一层学到的表示作为下一层的输入，逐层贪心地训练，最后再用有监督学习微调整个网络。
为什么需要预训练：
- 在深层网络中直接训练容易陷入差的局部极小值
- 无监督预训练可以将参数初始化到一个较好的区域（正则化效果）
- 利用大量无标签数据学习数据的内在结构
训练流程：
- 第1层：用原始输入训练一个自编码器，得到第1层编码器的权重
- 第2层：用第1层编码器的输出作为输入，训练第二个自编码器
- 重复上述步骤直到所有层都预训练完成
- 用有标签数据对整个网络进行有监督微调（Fine-tuning）
1.2 代码案例：贪心逐层预训练的自编码器
import torch # 导入PyTorch深度学习框架
import torch.nn as nn # 导入神经网络模块
import torch.optim as optim # 导入优化器模块
from torch.utils.data import DataLoader, TensorDataset # 导入数据加载工具
import numpy as np # 导入NumPy数值计算库
# ============================================================
# 设置随机种子，确保实验可复现
# ============================================================
torch.manual_seed(42) # 设置PyTorch的CPU随机种子为42
np.random.seed(42) # 设置NumPy的随机种子为42
# ============================================================
# 生成模拟数据（无标签数据 + 少量有标签数据）
# ============================================================
input_dim = 784 # 输入维度（模拟MNIST的28x28像素）
n_unlabeled = 5000 # 无标签样本数量（用于无监督预训练）
n_labeled = 500 # 有标签样本数量（用于有监督微调）
n_classes = 10 # 分类类别数（0-9共10个数字）
# 生成无标签数据（模拟实际场景中大量的无标签数据）
X_unlabeled = torch.randn(n_unlabeled, input_dim) # 随机生成5000x784的张量
# 生成有标签数据和标签
X_labeled = torch.randn(n_labeled, input_dim) # 随机生成500x784的张量
y_labeled = torch.randint(0, n_classes, (n_labeled,)) # 随机生成500个类别标签（0-9）
# ============================================================
# 定义单层自编码器（Autoencoder）——用于贪心逐层预训练
# ============================================================
class SingleLayerAutoencoder(nn.Module):
"""
单层自编码器
结构：输入层 -> 编码层 -> 解码层 -> 输出层（与输入层维度相同）
目标：学习输入数据的压缩表示
"""
def __init__(self, input_size, hidden_size):
"""
构造函数
参数:
input_size: 输入维度（上一层的输出维度或原始输入维度）
hidden_size: 隐藏层维度（这一层要学习的特征维度）
"""
super(SingleLayerAutoencoder, self).__init__() # 调用父类nn.Module的构造函数
# 编码器部分：将输入映射到隐藏表示
self.encoder = nn.Sequential(
nn.Linear(input_size, hidden_size), # 全连接层：输入维度 -> 隐藏维度
nn.Sigmoid() # Sigmoid激活函数，将输出压缩到(0,1)区间
)
# 解码器部分：将隐藏表示重建回输入
self.decoder = nn.Sequential(
nn.Linear(hidden_size, input_size), # 全连接层：隐藏维度 -> 输入维度
nn.Sigmoid() # Sigmoid激活函数，输出范围(0,1)
)
def forward(self, x):
"""
前向传播
参数:
x: 输入张量，形状为(batch_size, input_size)
返回:
decoded: 重建的输出，形状为(batch_size, input_size)
encoded: 编码后的隐藏表示，形状为(batch_size, hidden_size)
"""
encoded = self.encoder(x) # 编码：将输入映射到隐藏空间
decoded = self.decoder(encoded) # 解码：将隐藏表示重建回输入空间
return decoded, encoded # 返回重建结果和隐藏表示
# ============================================================
# 定义完整的贪心逐层预训练类
# ============================================================
class GreedyLayerwisePretraining:
"""
贪心逐层无监督预训练
核心思想：逐层训练自编码器，每层的输入是上一层编码器的输出
"""
def __init__(self, layer_sizes, pretrain_epochs=50, pretrain_lr=0.001):
"""
构造函数
参数:
layer_sizes: 列表，每一层隐藏层的神经元数量
例如 [784, 512, 256, 128] 表示：
- 第1层自编码器：784 -> 512 -> 784
- 第2层自编码器：512 -> 256 -> 512
- 第3层自编码器：256 -> 128 -> 256
pretrain_epochs: 每层预训练的训练轮数
pretrain_lr: 预训练的学习率
"""
self.layer_sizes = layer_sizes # 保存网络各层尺寸
self.pretrain_epochs = pretrain_epochs # 保存预训练轮数
self.pretrain_lr = pretrain_lr # 保存预训练学习率
self.encoders = [] # 初始化编码器列表，用于保存每一层训练好的编码器
def pretrain(self, X_unlabeled, batch_size=64):
"""
执行贪心逐层预训练
参数:
X_unlabeled: 无标签数据，形状为(n_samples, input_dim)
batch_size: 小批量大小
"""
current_input = X_unlabeled # 当前层的输入初始化为原始数据
# 逐层训练自编码器
for i in range(len(self.layer_sizes) - 1): # 遍历每一对相邻层
input_size = self.layer_sizes[i] # 当前层的输入维度
hidden_size = self.layer_sizes[i + 1] # 当前层的隐藏维度（输出维度）
print(f"\n{'='*60}") # 打印分隔线
print(f"预训练第 {i+1} 层: {input_size} -> {hidden_size}") # 打印当前预训练层信息
# 创建当前层的自编码器
autoencoder = SingleLayerAutoencoder(input_size, hidden_size) # 实例化自编码器
# 定义优化器
optimizer = optim.Adam( # 使用Adam优化器
autoencoder.parameters(), # 优化自编码器的所有参数
lr=self.pretrain_lr # 设置学习率
)
# 定义损失函数（均方误差损失，用于重建任务）
criterion = nn.MSELoss() # MSE损失：衡量重建输出与原始输入的差异
# 创建数据加载器（小批量训练）
dataset = TensorDataset(current_input) # 将数据包装成TensorDataset
dataloader = DataLoader( # 创建数据加载器
dataset, # 数据集
batch_size=batch_size, # 每个批次的样本数
shuffle=True # 每个epoch打乱数据顺序
)
# 训练当前层的自编码器
autoencoder.train() # 设置为训练模式（启用Dropout等）
for epoch in range(self.pretrain_epochs): # 遍历每个训练轮次
total_loss = 0 # 初始化当前epoch的累计损失
for (batch_x,) in dataloader: # 遍历每个小批量数据
# 前向传播
reconstructed, _ = autoencoder(batch_x) # 通过自编码器获取重建输出
# 计算损失
loss = criterion(reconstructed, batch_x) # 计算重建损失（输入与输出的MSE）
# 反向传播
optimizer.zero_grad() # 清除之前的梯度
loss.backward() # 计算当前梯度
optimizer.step() # 更新参数
total_loss += loss.item() # 累加损失值
# 每10个epoch打印一次训练进度
if (epoch + 1) % 10 == 0: # 判断是否为第10的倍数轮次
avg_loss = total_loss / len(dataloader) # 计算平均损失
print(f" Epoch [{epoch+1}/{self.pretrain_epochs}], "
f"重建损失: {avg_loss:.6f}") # 打印训练进度
# 保存训练好的编码器
self.encoders.append(autoencoder.encoder) # 将编码器加入列表（解码器不再需要）
# 用当前编码器将输入转换为下一层的输入
autoencoder.eval() # 设置为评估模式
with torch.no_grad(): # 关闭梯度计算（节省内存和计算）
_, current_input = autoencoder(current_input) # 获取编码后的隐藏表示
print(f" 第 {i+1} 层编码器输出形状: {current_input.shape}") # 打印输出形状
def encode(self, x):
"""
使用所有预训练好的编码器进行前向编码
参数:
x: 输入数据
返回:
最终层的隐藏表示
"""
current = x # 初始化当前输入
for encoder in self.encoders: # 遍历所有编码器
encoder.eval() # 设置为评估模式
with torch.no_grad(): # 关闭梯度计算
current = encoder(current) # 逐层编码
return current # 返回最终表示
# ============================================================
# 定义完整的微调分类网络
# ============================================================
class FineTunedNetwork(nn.Module):
"""
基于预训练权重初始化的分类网络
使用预训练的编码器权重初始化各层，然后添加分类头进行有监督微调
"""
def __init__(self, pretrained_encoders, n_classes):
"""
构造函数
参数:
pretrained_encoders: 预训练好的编码器列表
n_classes: 分类类别数
"""
super(FineTunedNetwork, self).__init__()
# 构建特征提取层（使用预训练的编码器）
layers = [] # 初始化层列表
for encoder in pretrained_encoders: # 遍历每个预训练的编码器
# 提取编码器中的线性层和激活函数
for layer in encoder: # 遍历编码器中的每一层
layers.append(layer) # 将层添加到列表中（保留预训练权重）
self.feature_extractor = nn.Sequential(*layers) # 创建特征提取器的顺序容器
# 添加分类头（输出层）
# 获取最后一个编码器的输出维度
last_encoder_output = None
for layer in pretrained_encoders[-1]: # 遍历最后一个编码器的层
if isinstance(layer, nn.Linear): # 找到线性层
last_encoder_output = layer.out_features # 获取输出维度
self.classifier = nn.Linear( # 创建分类输出层
last_encoder_output, # 输入维度 = 最后一层编码器的输出维度
n_classes # 输出维度 = 类别数量
)
def forward(self, x):
"""
前向传播
参数:
x: 输入数据
返回:
logits: 分类的原始输出（未经softmax）
"""
features = self.feature_extractor(x) # 提取特征表示
logits = self.classifier(features) # 分类预测
return logits # 返回预测结果
# ============================================================
# 执行贪心逐层预训练
# ============================================================
layer_sizes = [784, 512, 256, 128] # 定义网络各层尺寸
pretrainer = GreedyLayerwisePretraining( # 创建贪心逐层预训练对象
layer_sizes=layer_sizes, # 传入层尺寸
pretrain_epochs=50, # 设置预训练轮数
pretrain_lr=0.001 # 设置学习率
)
# 执行预训练（使用无标签数据）
pretrainer.pretrain(X_unlabeled) # 开始逐层预训练
# ============================================================
# 使用预训练权重初始化分类网络并进行有监督微调
# ============================================================
model = FineTunedNetwork(pretrainer.encoders, n_classes) # 创建微调网络
# 定义有监督训练的优化器和损失函数
finetune_optimizer = optim.Adam( # 创建Adam优化器
model.parameters(), # 优化模型所有参数
lr=0.0001 # 微调使用较小的学习率（保护预训练权重）
)
criterion_ce = nn.CrossEntropyLoss() # 交叉熵损失（用于分类任务）
# 创建有标签数据的数据加载器
labeled_dataset = TensorDataset(X_labeled, y_labeled) # 包装有标签数据
labeled_loader = DataLoader( # 创建数据加载器
labeled_dataset, # 数据集
batch_size=64, # 批次大小
shuffle=True # 打乱顺序
)
# 微调阶段训练
print(f"\n{'='*60}")
print("开始有监督微调...")
finetune_epochs = 100 # 微调训练轮数
model.train() # 设置模型为训练模式
for epoch in range(finetune_epochs): # 遍历每个训练轮次
total_loss = 0 # 初始化累计损失
correct = 0 # 初始化正确预测数
total = 0 # 初始化总样本数
for batch_x, batch_y in labeled_loader: # 遍历每个小批量
# 前向传播
logits = model(batch_x) # 获取分类输出
loss = criterion_ce(logits, batch_y) # 计算交叉熵损失
# 反向传播
finetune_optimizer.zero_grad() # 清除旧梯度
loss.backward() # 计算新梯度
finetune_optimizer.step() # 更新参数
# 统计
total_loss += loss.item() # 累加损失
_, predicted = torch.max(logits, 1) # 获取预测类别（取最大值的索引）
total += batch_y.size(0) # 累加样本数
correct += (predicted == batch_y).sum().item() # 累加正确预测数
# 每20个epoch打印训练进度
if (epoch + 1) % 20 == 0: # 判断是否打印
accuracy = 100 * correct / total # 计算准确率
print(f"微调 Epoch [{epoch+1}/{finetune_epochs}], "
f"损失: {total_loss/len(labeled_loader):.4f}, "
f"准确率: {accuracy:.2f}%") # 打印进度
print("\n贪心逐层预训练 + 微调完成！")
二、迁移学习和领域自适应（Transfer Learning and Domain Adaptation）
2.1 核心知识点
迁移学习： 将一个领域（源域）上学到的知识迁移到另一个相关领域（目标域），以解决目标域数据不足的问题。
关键假设：
- 源域和目标域的输入空间相同（或有重叠）
- 源域和目标域共享某些底层特征表示
- 源域有大量标注数据，目标域标注数据少或没有
领域自适应（Domain Adaptation）： 迁移学习的子问题，重点在于减小源域和目标域之间的分布差异。
主要方法：
| 方法 | 说明 |
|---|---|
| 特征提取器共享 | 用源域训练好的特征提取器，固定或微调后用于目标域 |
| 对抗性领域自适应 | 通过对抗训练让特征提取器产生领域不变的特征 |
| 最大均值差异（MMD） | 最小化两个领域特征分布在再生核希尔伯特空间中的距离 |
2.2 代码案例一：基于预训练模型的迁移学习（PyTorch）
import torch # 导入PyTorch深度学习框架
import torch.nn as nn # 导入神经网络模块
import torch.optim as optim # 导入优化器模块
import torchvision.models as models # 导入预训练模型库
from torch.utils.data import DataLoader, TensorDataset # 导入数据工具
import numpy as np # 导入NumPy
torch.manual_seed(42) # 设置随机种子
# ============================================================
# 模拟源域数据和目标域数据
# ============================================================
# 假设源域：ImageNet上的图像分类（1000类）
# 目标域：医学图像分类（5类，数据量少）
source_input_dim = 2048 # 源域特征维度（模拟ResNet最后一层输出）
target_input_dim = 2048 # 目标域特征维度（与源域相同）
n_source = 5000 # 源域样本数量（充足）
n_target = 200 # 目标域样本数量（稀少）
n_source_classes = 1000 # 源域类别数
n_target_classes = 5 # 目标域类别数
# 模拟源域数据
X_source = torch.randn(n_source, source_input_dim) # 源域特征
y_source = torch.randint(0, n_source_classes, (n_source,)) # 源域标签
# 模拟目标域数据
X_target = torch.randn(n_target, target_input_dim) # 目标域特征
y_target = torch.randint(0, n_target_classes, (n_target,)) # 目标域标签
# ============================================================
# 方法1：特征提取法（Feature Extraction）
# 思路：冻结预训练模型的特征提取器，只训练新的分类头
# ============================================================
class TransferLearningFeatureExtraction(nn.Module):
"""
迁移学习-特征提取方法
核心：冻结预训练层的参数，只训练新增的分类层
"""
def __init__(self, feature_dim, n_target_classes, freeze_features=True):
"""
构造函数
参数:
feature_dim: 预训练特征的维度
n_target_classes: 目标域类别数
freeze_features: 是否冻结特征提取层
"""
super(TransferLearningFeatureExtraction, self).__init__()
# 模拟预训练的特征提取器（实际中用ResNet、VGG等）
self.feature_extractor = nn.Sequential(
nn.Linear(feature_dim, 512), # 全连接层：特征维度 -> 512
nn.ReLU(), # ReLU激活函数
nn.Dropout(0.3), # Dropout正则化，丢弃概率30%
nn.Linear(512, 256), # 全连接层：512 -> 256
nn.ReLU(), # ReLU激活函数
)
# 如果冻结特征提取器的参数
if freeze_features: # 判断是否冻结
for param in self.feature_extractor.parameters(): # 遍历特征提取器的所有参数
param.requires_grad = False # 设置为不需要计算梯度（冻结参数）
# 新增分类头（针对目标域任务）
self.classifier = nn.Sequential(
nn.Linear(256, 128), # 全连接层：256 -> 128
nn.ReLU(), # ReLU激活函数
nn.Linear(128, n_target_classes), # 输出层：128 -> 目标域类别数
)
def forward(self, x):
"""
前向传播
参数:
x: 输入特征
返回:
分类输出logits
"""
features = self.feature_extractor(x) # 用特征提取器提取特征
logits = self.classifier(features) # 用分类头进行分类
return logits # 返回分类结果
# ============================================================
# 方法2：微调法（Fine-tuning）
# 思路：使用较小的学习率更新预训练层，用正常学习率训练分类头
# ============================================================
class TransferLearningFineTuning(nn.Module):
"""
迁移学习-微调方法
核心：预训练层和新层使用不同的学习率
"""
def __init__(self, feature_dim, n_target_classes):
"""
构造函数
参数:
feature_dim: 预训练特征维度
n_target_classes: 目标域类别数
"""
super(TransferLearningFineTuning, self).__init__()
# 预训练的特征提取器（参数会用小学习率更新）
self.feature_extractor = nn.Sequential(
nn.Linear(feature_dim, 512), # 全连接层
nn.ReLU(), # 激活函数
nn.Linear(512, 256), # 全连接层
nn.ReLU(), # 激活函数
)
# 新增分类头
self.classifier = nn.Sequential(
nn.Linear(256, 128), # 全连接层
nn.ReLU(), # 激活函数
nn.Linear(128, n_target_classes), # 输出层
)
def forward(self, x):
"""前向传播"""
features = self.feature_extractor(x) # 提取特征
logits = self.classifier(features) # 分类
return logits # 返回结果
# ============================================================
# 训练函数
# ============================================================
def train_model(model, X_train, y_train, epochs=50, lr=0.001, model_type='feature_extract'):
"""
通用训练函数
参数:
model: 要训练的模型
X_train: 训练特征
y_train: 训练标签
epochs: 训练轮数
lr: 学习率
model_type: 模型类型（'feature_extract'或'finetune'）
"""
criterion = nn.CrossEntropyLoss() # 交叉熵损失函数
if model_type == 'finetune': # 如果是微调模式
# 对不同层使用不同的学习率
optimizer = optim.Adam([ # 创建参数组
{'params': model.feature_extractor.parameters(), # 特征提取器参数
'lr': lr * 0.1}, # 使用较小的学习率（源域知识的保护）
{'params': model.classifier.parameters(), # 分类头参数
'lr': lr} # 使用正常学习率
])
else: # 如果是特征提取模式
# 只优化分类头（因为特征提取器被冻结了）
optimizer = optim.Adam(
filter(lambda p: p.requires_grad, model.parameters()), # 只选择需要梯度的参数
lr=lr # 正常学习率
)
dataset = TensorDataset(X_train, y_train) # 创建数据集
loader = DataLoader(dataset, batch_size=32, shuffle=True) # 创建数据加载器
model.train() # 设置训练模式
for epoch in range(epochs): # 遍历每个epoch
total_loss = 0 # 初始化损失
correct = 0 # 正确数
total = 0 # 总数
for batch_x, batch_y in loader: # 遍历每个批次
logits = model(batch_x) # 前向传播
loss = criterion(logits, batch_y) # 计算损失
optimizer.zero_grad() # 清除梯度
loss.backward() # 反向传播
optimizer.step() # 更新参数
total_loss += loss.item() # 累加损失
_, predicted = torch.max(logits, 1) # 获取预测
total += batch_y.size(0) # 累加总数
correct += (predicted == batch_y).sum().item() # 累加正确数
if (epoch + 1) % 10 == 0: # 每10轮打印
acc = 100 * correct / total # 计算准确率
print(f" Epoch [{epoch+1}/{epochs}], "
f"Loss: {total_loss/len(loader):.4f}, "
f"Acc: {acc:.2f}%") # 打印进度
# ============================================================
# 运行特征提取方法
# ============================================================
print("="*60)
print("方法1：特征提取法（冻结特征层，只训练分类头）")
model_fe = TransferLearningFeatureExtraction( # 创建特征提取模型
feature_dim=target_input_dim, # 特征维度
n_target_classes=n_target_classes, # 目标域类别数
freeze_features=True # 冻结特征提取器
)
train_model(model_fe, X_target, y_target, # 训练模型
epochs=50, lr=0.001, model_type='feature_extract')
# ============================================================
# 运行微调方法
# ============================================================
print("\n" + "="*60)
print("方法2：微调法（特征层小学习率，分类头正常学习率）")
model_ft = TransferLearningFineTuning( # 创建微调模型
feature_dim=target_input_dim, # 特征维度
n_target_classes=n_target_classes # 目标域类别数
)
train_model(model_ft, X_target, y_target, # 训练模型
epochs=50, lr=0.001, model_type='finetune')
2.3 代码案例二：最大均值差异（MMD）领域自适应
import torch # 导入PyTorch
import torch.nn as nn # 导入神经网络模块
import torch.optim as optim # 导入优化器
import numpy as np # 导入NumPy
torch.manual_seed(42) # 设置随机种子
# ============================================================
# 定义MMD（最大均值差异）损失函数
# ============================================================
class MMDLoss(nn.Module):
"""
最大均值差异（Maximum Mean Discrepancy）损失
原理：MMD衡量两个分布在再生核希尔伯特空间（RKHS）中的距离。
通过最小化MMD，可以让源域和目标域的特征分布尽量接近。
核函数：使用高斯核（RBF核）
K(x, y) = exp(-||x - y||^2 / (2 * sigma^2))
"""
def __init__(self, kernel_bandwidth=1.0):
"""
构造函数
参数:
kernel_bandwidth: 高斯核的带宽参数sigma
"""
super(MMDLoss, self).__init__()
self.kernel_bandwidth = kernel_bandwidth # 保存核带宽参数
def gaussian_kernel(self, x, y):
"""
计算两个样本集之间的高斯核矩阵
参数:
x: 第一组样本，形状为(n_samples_x, n_features)
y: 第二组样本，形状为(n_samples_y, n_features)
返回:
核矩阵，形状为(n_samples_x, n_samples_y)
"""
# 计算||x_i - y_j||^2（欧几里得距离的平方）
# 展开：||x_i - y_j||^2 = ||x_i||^2 + ||y_j||^2 - 2 * x_i · y_j
x_size = x.size(0) # 获取x的样本数
y_size = y.size(0) # 获取y的样本数
dim = x.size(1) # 获取特征维度
# 利用广播机制计算两两之间的距离平方
x_expanded = x.unsqueeze(1).expand(x_size, y_size, dim) # 扩展x: (n_x, 1, d) -> (n_x, n_y, d)
y_expanded = y.unsqueeze(0).expand(x_size, y_size, dim) # 扩展y: (1, n_y, d) -> (n_x, n_y, d)
distance_squared = torch.sum( # 计算距离的平方
(x_expanded - y_expanded) ** 2, # 差的平方
dim=2 # 沿特征维度求和
) # 输出形状: (n_x, n_y)
# 应用高斯核公式
bandwidth = 2.0 * self.kernel_bandwidth ** 2 # 计算2*sigma^2
kernel_matrix = torch.exp(-distance_squared / bandwidth) # 高斯核计算
return kernel_matrix # 返回核矩阵
def forward(self, source_features, target_features):
"""
计算MMD损失
参数:
source_features: 源域特征，形状为(n_source, n_features)
target_features: 目标域特征，形状为(n_target, n_features)
返回:
MMD损失值（标量）
"""
# 计算三种核矩阵
K_ss = self.gaussian_kernel( # 源域-源域核矩阵
source_features, source_features
) # 形状: (n_source, n_source)
K_tt = self.gaussian_kernel( # 目标域-目标域核矩阵
target_features, target_features
) # 形状: (n_target, n_target)
K_st = self.gaussian_kernel( # 源域-目标域核矩阵
source_features, target_features
) # 形状: (n_source, n_target)
# MMD^2 = E[K(x_s, x_s')] + E[K(x_t, x_t')] - 2*E[K(x_s, x_t)]
mmd = K_ss.mean() + K_tt.mean() - 2 * K_st.mean() # 计算MMD的平方
return mmd # 返回MMD损失
# ============================================================
# 定义领域自适应网络
# ============================================================
class DomainAdaptationNetwork(nn.Module):
"""
基于MMD的领域自适应网络
架构：
输入 -> 特征提取器（共享） -> 分类器
-> MMD损失（源域和目标域特征分布对齐）
目标函数 = 分类损失 + lambda * MMD损失
"""
def __init__(self, input_dim, feature_dim, n_classes):
"""
构造函数
参数:
input_dim: 输入特征维度
feature_dim: 共享特征空间的维度
n_classes: 分类类别数
"""
super(DomainAdaptationNetwork, self).__init__()
# 共享特征提取器（源域和目标域共用同一个编码器）
self.feature_extractor = nn.Sequential(
nn.Linear(input_dim, 512), # 全连接层1
nn.ReLU(), # ReLU激活
nn.BatchNorm1d(512), # 批归一化，加速训练和稳定梯度
nn.Dropout(0.3), # Dropout正则化
nn.Linear(512, 256), # 全连接层2
nn.ReLU(), # ReLU激活
nn.BatchNorm1d(256), # 批归一化
nn.Linear(256, feature_dim), # 输出层，映射到共享特征空间
nn.ReLU(), # ReLU激活
)
# 分类器（在共享特征上进行分类）
self.classifier = nn.Sequential(
nn.Linear(feature_dim, 64), # 全连接层
nn.ReLU(), # 激活函数
nn.Linear(64, n_classes), # 输出层
)
def forward(self, x):
"""
前向传播
参数:
x: 输入数据
返回:
logits: 分类输出
features: 提取的特征表示
"""
features = self.feature_extractor(x) # 提取共享特征
logits = self.classifier(features) # 分类
return logits, features # 返回分类结果和特征
# ============================================================
# 训练领域自适应网络
# ============================================================
def train_domain_adaptation():
"""训练领域自适应模型"""
# 参数设置
input_dim = 256 # 输入维度
feature_dim = 64 # 共享特征维度
n_classes = 5 # 类别数
n_source = 1000 # 源域样本数
n_target = 100 # 目标域样本数（很少）
epochs = 100 # 训练轮数
mmd_weight = 0.5 # MMD损失的权重系数lambda
# 生成模拟数据
# 源域数据
X_source = torch.randn(n_source, input_dim) # 源域输入特征
y_source = torch.randint(0, n_classes, (n_source,)) # 源域标签
# 目标域数据（分布略有不同，模拟域偏移）
X_target = torch.randn(n_target, input_dim) * 1.2 + 0.3 # 不同均值和方差的分布
y_target = torch.randint(0, n_classes, (n_target,)) # 目标域标签
# 创建模型
model = DomainAdaptationNetwork(input_dim, feature_dim, n_classes) # 实例化模型
# 优化器和损失函数
optimizer = optim.Adam(model.parameters(), lr=0.001) # Adam优化器
classification_loss = nn.CrossEntropyLoss() # 分类损失函数
mmd_loss_fn = MMDLoss(kernel_bandwidth=1.0) # MMD损失函数
# 数据加载
source_dataset = TensorDataset(X_source, y_source) # 源域数据集
source_loader = DataLoader(source_dataset, batch_size=64, shuffle=True) # 源域加载器
target_dataset = TensorDataset(X_target, y_target) # 目标域数据集
target_loader = DataLoader(target_dataset, batch_size=32, shuffle=True) # 目标域加载器
# 开始训练
print("="*60)
print("开始领域自适应训练（MMD方法）")
model.train() # 设置训练模式
for epoch in range(epochs): # 遍历每个epoch
total_cls_loss = 0 # 分类损失累计
total_mmd_loss = 0 # MMD损失累计
# 同时遍历源域和目标域数据
for (batch_x_s, batch_y_s), (batch_x_t, _) in zip(source_loader, target_loader):
# 前向传播 - 源域
logits_s, features_s = model(batch_x_s) # 源域的分类输出和特征
# 前向传播 - 目标域
logits_t, features_t = model(batch_x_t) # 目标域的分类输出和特征
# 计算分类损失（只在源域上有标签）
cls_loss = classification_loss(logits_s, batch_y_s) # 源域分类损失
# 计算MMD损失（对齐源域和目标域的特征分布）
domain_loss = mmd_loss_fn(features_s, features_t) # MMD损失
# 总损失 = 分类损失 + lambda * MMD损失
total_loss = cls_loss + mmd_weight * domain_loss # 加权组合损失
# 反向传播
optimizer.zero_grad() # 清除旧梯度
total_loss.backward() # 计算梯度
optimizer.step() # 更新参数
total_cls_loss += cls_loss.item() # 累加分类损失
total_mmd_loss += domain_loss.item() # 累加MMD损失
# 打印训练进度
if (epoch + 1) % 20 == 0: # 每20轮打印
print(f"Epoch [{epoch+1}/{epochs}], "
f"分类损失: {total_cls_loss/len(source_loader):.4f}, "
f"MMD损失: {total_mmd_loss/len(source_loader):.4f}")
# 评估目标域准确率
model.eval() # 设置评估模式
with torch.no_grad(): # 关闭梯度计算
logits_t, _ = model(X_target) # 目标域预测
_, predicted = torch.max(logits_t, 1) # 获取预测类别
accuracy = (predicted == y_target).float().mean() # 计算准确率
print(f"\n目标域分类准确率: {accuracy.item() * 100:.2f}%")
# 运行训练
train_domain_adaptation() # 执行训练
三、半监督学习与因果关系（Semi-supervised Learning and Causality）
3.1 核心知识点
半监督学习： 同时利用有标签数据和无标签数据进行学习。核心假设是：
- 流形假设： 相邻的数据点倾向于有相同的标签
- 聚类假设： 数据倾向于形成离散的簇，同一簇中的数据点属于同一类别
- 低密度分离假设： 分类边界倾向于穿过数据的低密度区域
与因果关系的联系：
- 如果我们能识别出数据的潜在生成因素（即因果变量），半监督学习会更有效
- 无标签数据有助于发现数据的内在结构（潜在原因）
- 好的表示应该捕捉到因果因素，而非仅仅是统计相关性
3.2 代码案例：基于图的半监督标签传播
import torch # 导入PyTorch
import torch.nn as nn # 导入神经网络模块
import torch.optim as optim # 导入优化器
import numpy as np # 导入NumPy
from sklearn.neighbors import kneighbors_graph # 导入K近邻图构建工具
torch.manual_seed(42) # 设置随机种子
np.random.seed(42) # 设置NumPy随机种子
# ============================================================
# 生成半监督学习的模拟数据
# ============================================================
def generate_semi_supervised_data(n_total=500, n_labeled=50, n_features=20, n_classes=3):
"""
生成半监督学习数据集
参数:
n_total: 总样本数
n_labeled: 有标签样本数（远小于总数）
n_features: 特征维度
n_classes: 类别数
返回:
features: 特征矩阵
labels: 完整标签（用于评估）
labels_partial: 部分标签（-1表示无标签）
"""
# 为每个类别生成不同分布的数据点
features_list = [] # 特征列表
labels_list = [] # 标签列表
samples_per_class = n_total // n_classes # 每个类别的样本数
for i in range(n_classes): # 遍历每个类别
# 每个类别有不同的均值（模拟不同的簇）
center = np.random.randn(n_features) * 3 # 随机生成类别中心
cluster_data = np.random.randn( # 生成簇内数据
samples_per_class, n_features # 样本数 x 特征维度
) + center # 加上类别中心偏移
features_list.append(cluster_data) # 添加到特征列表
labels_list.extend([i] * samples_per_class) # 添加对应的标签
features = np.vstack(features_list) # 垂直拼接所有类别数据
labels = np.array(labels_list) # 转换标签为numpy数组
# 打乱数据顺序
shuffle_idx = np.random.permutation(len(features)) # 生成随机排列索引
features = features[shuffle_idx] # 按索引打乱特征
labels = labels[shuffle_idx] # 按索引打乱标签
# 创建部分标签（大部分为-1表示无标签）
labels_partial = np.full(n_total, -1) # 初始化所有标签为-1（无标签）
labeled_indices = np.random.choice( # 随机选择有标签的样本索引
n_total, n_labeled, replace=False # 不重复抽样
)
labels_partial[labeled_indices] = labels[labeled_indices] # 填入已知标签
return ( # 返回数据
torch.FloatTensor(features), # 特征张量
torch.LongTensor(labels), # 完整标签
torch.LongTensor(labels_partial), # 部分标签
labeled_indices # 有标签样本的索引
)
# ============================================================
# 定义基于自训练（Self-Training）的半监督学习
# ============================================================
class SemiSupervisedSelfTraining:
"""
自训练半监督学习（Self-Training / Pseudo-Labeling）
步骤：
1. 用有标签数据训练初始模型
2. 用模型对无标签数据进行预测，生成伪标签
3. 选择高置信度的伪标签加入训练集
4. 用扩充后的训练集重新训练模型
5. 重复步骤2-4直到收敛
"""
def __init__(self, input_dim, n_classes, hidden_dim=128):
"""
构造函数
参数:
input_dim: 输入特征维度
n_classes: 类别数
hidden_dim: 隐藏层维度
"""
self.n_classes = n_classes # 保存类别数
# 定义分类网络
self.model = nn.Sequential(
nn.Linear(input_dim, hidden_dim), # 全连接层1
nn.ReLU(), # 激活函数
nn.BatchNorm1d(hidden_dim), # 批归一化
nn.Dropout(0.3), # Dropout
nn.Linear(hidden_dim, hidden_dim // 2), # 全连接层2
nn.ReLU(), # 激活函数
nn.Linear(hidden_dim // 2, n_classes), # 输出层
)
def train_step(self, X_train, y_train, epochs=30, lr=0.001, batch_size=32):
"""
单次训练步骤（在给定的有标签数据上训练模型）
参数:
X_train: 训练特征
y_train: 训练标签
epochs: 训练轮数
lr: 学习率
batch_size: 批次大小
"""
optimizer = optim.Adam(self.model.parameters(), lr=lr) # Adam优化器
criterion = nn.CrossEntropyLoss() # 交叉熵损失
dataset = TensorDataset(X_train, y_train) # 创建数据集
loader = DataLoader(dataset, batch_size=batch_size, shuffle=True) # 数据加载器
self.model.train() # 训练模式
for epoch in range(epochs): # 遍历epoch
for batch_x, batch_y in loader: # 遍历批次
logits = self.model(batch_x) # 前向传播
loss = criterion(logits, batch_y) # 计算损失
optimizer.zero_grad() # 清除梯度
loss.backward() # 反向传播
optimizer.step() # 更新参数
def predict_with_confidence(self, X):
"""
预测并返回置信度
参数:
X: 输入特征
返回:
predictions: 预测类别
confidences: 预测置信度（最大softmax概率）
"""
self.model.eval() # 评估模式
with torch.no_grad(): # 关闭梯度
logits = self.model(X) # 获取分类输出
probabilities = torch.softmax(logits, dim=1) # 计算softmax概率
confidences, predictions = torch.max(probabilities, dim=1) # 获取最大概率和预测类别
return predictions, confidences # 返回结果
def self_train(self, X_all, labels_partial, confidence_threshold=0.9,
max_iterations=5, initial_epochs=50, retrain_epochs=20):
"""
自训练主循环
参数:
X_all: 所有数据（有标签+无标签）
labels_partial: 部分标签（-1表示无标签）
confidence_threshold: 置信度阈值（只有高于此值的伪标签才被接受）
max_iterations: 最大迭代次数
initial_epochs: 初始训练轮数
retrain_epochs: 每次重新训练的轮数
"""
labels = labels_partial.clone() # 克隆标签（会修改）
# 第一步：用初始有标签数据训练模型
labeled_mask = labels >= 0 # 创建有标签样本的布尔掩码
X_labeled = X_all[labeled_mask] # 提取有标签数据
y_labeled = labels[labeled_mask] # 提取有标签标签
print(f"初始有标签样本数: {labeled_mask.sum().item()}")
self.train_step(X_labeled, y_labeled, epochs=initial_epochs) # 初始训练
# 评估初始模型
predictions, _ = self.predict_with_confidence(X_all) # 对所有数据预测
accuracy = (predictions == labels_partial).float()[labels_partial >= 0].mean() # 初始准确率
print(f"初始有标签数据准确率: {accuracy*100:.2f}%")
# 自训练迭代
for iteration in range(max_iterations): # 遍历每次迭代
# 对无标签数据进行预测
predictions, confidences = self.predict_with_confidence(X_all) # 获取预测和置信度
# 选择高置信度的预测作为伪标签
unlabeled_mask = labels < 0 # 找到仍然无标签的样本
high_confidence_mask = ( # 高置信度掩码
unlabeled_mask & (confidences > confidence_threshold) # 无标签 AND 高置信度
)
n_new_labeled = high_confidence_mask.sum().item() # 新增伪标签数量
if n_new_labeled == 0: # 如果没有新增标签
print(f"迭代 {iteration+1}: 没有新的高置信度预测，停止迭代")
break # 退出循环
# 将高置信度预测作为伪标签加入训练集
labels[high_confidence_mask] = predictions[high_confidence_mask] # 更新标签
# 用扩充后的数据重新训练
new_labeled_mask = labels >= 0 # 新的有标签掩码
X_train = X_all[new_labeled_mask] # 扩充后的训练数据
y_train = labels[new_labeled_mask] # 扩充后的标签
print(f"迭代 {iteration+1}: 新增 {n_new_labeled} 个伪标签, "
f"总训练样本: {new_labeled_mask.sum().item()}")
self.train_step(X_train, y_train, epochs=retrain_epochs) # 重新训练模型
return labels # 返回最终标签（包含伪标签）
# ============================================================
# 定义半监督变分自编码器（VAE）—— 学习潜在因果表示
# ============================================================
class SemiSupervisedVAE(nn.Module):
"""
半监督变分自编码器 (M1+M2 模型)
核心思想：
- VAE学习数据的潜在生成因素（因果变量）
- 半监督设置：部分数据有标签，部分没有
- 潜在变量z捕捉数据的连续变化因素
- 标签y捕捉离散的类别信息
损失函数 = 重建损失 + KL散度 + 分类损失
"""
def __init__(self, input_dim, latent_dim, n_classes):
"""
构造函数
参数:
input_dim: 输入维度
latent_dim: 潜在空间维度（连续因果因素的数量）
n_classes: 类别数
"""
super(SemiSupervisedVAE, self).__init__()
self.latent_dim = latent_dim # 保存潜在维度
# 编码器 q(z|x)：将输入映射到潜在分布
self.encoder = nn.Sequential(
nn.Linear(input_dim, 256), # 全连接层
nn.ReLU(), # 激活函数
nn.Linear(256, 128), # 全连接层
nn.ReLU(), # 激活函数
)
self.fc_mu = nn.Linear(128, latent_dim) # 均值网络 μ = q(z|x)
self.fc_logvar = nn.Linear(128, latent_dim) # 对数方差网络 log σ² = q(z|x)
# 解码器 p(x|z, y)：从潜在变量和标签重建输入
self.decoder = nn.Sequential(
nn.Linear(latent_dim + n_classes, 128), # 输入=潜在变量拼接标签
nn.ReLU(), # 激活函数
nn.Linear(128, 256), # 全连接层
nn.ReLU(), # 激活函数
nn.Linear(256, input_dim), # 输出层
nn.Sigmoid(), # Sigmoid激活（输出范围0-1）
)
# 分类器 q(y|x)：从输入预测类别
self.classifier = nn.Sequential(
nn.Linear(input_dim, 128), # 全连接层
nn.ReLU(), # 激活函数
nn.Linear(128, n_classes), # 输出层（类别logits）
)
def encode(self, x):
"""
编码：获取潜在分布的参数
参数:
x: 输入数据
返回:
mu: 潜在分布的均值
logvar: 潜在分布的对数方差
"""
h = self.encoder(x) # 通过编码器获取隐藏表示
mu = self.fc_mu(h) # 计算均值
logvar = self.fc_logvar(h) # 计算对数方差
return mu, logvar # 返回分布参数
def reparameterize(self, mu, logvar):
"""
重参数化技巧：从 q(z|x) 中采样
z = μ + σ * ε, 其中 ε ~ N(0, I)
为什么要重参数化？直接从N(μ,σ²)采样无法反向传播梯度
通过重参数化，将随机性转移到ε（与参数无关的噪声）
参数:
mu: 均值
logvar: 对数方差
返回:
z: 采样的潜在变量
"""
std = torch.exp(0.5 * logvar) # 计算标准差 σ = exp(0.5 * log σ²)
eps = torch.randn_like(std) # 从标准正态分布采样 ε ~ N(0, I)
z = mu + eps * std # 重参数化：z = μ + σ * ε
return z # 返回采样结果
def decode(self, z, y_onehot):
"""
解码：从潜在变量和标签重建输入
参数:
z: 潜在变量
y_onehot: 类别的one-hot编码
返回:
重建的输入
"""
z_y = torch.cat([z, y_onehot], dim=1) # 拼接潜在变量和标签
return self.decoder(z_y) # 通过解码器重建
def forward(self, x, y=None):
"""
前向传播
参数:
x: 输入数据
y: 标签（如果有的话），无标签数据为None
返回:
根据是否有标签返回不同的结果
"""
# 编码
mu, logvar = self.encode(x) # 获取潜在分布参数
z = self.reparameterize(mu, logvar) # 重参数化采样
if y is not None: # 如果有标签
# 将标签转换为one-hot编码
y_onehot = torch.zeros(x.size(0), self.classifier[-1].out_features) # 创建零矩阵
y_onehot.scatter_(1, y.unsqueeze(1), 1) # 填充one-hot编码
x_recon = self.decode(z, y_onehot) # 重建输入
else: # 如果无标签
# 对所有可能的类别进行预测
y_logits = self.classifier(x) # 获取类别logits
y_prob = torch.softmax(y_logits, dim=1) # 计算类别概率
y_pred = y_prob.argmax(dim=1) # 获取预测类别
y_onehot = torch.zeros_like(y_prob) # 创建零矩阵
y_onehot.scatter_(1, y_pred.unsqueeze(1), 1) # 填充one-hot
x_recon = self.decode(z, y_onehot) # 用预测标签重建
return x_recon, mu, logvar # 返回重建结果和分布参数
# ============================================================
# 半监督VAE的损失函数
# ============================================================
def semi_supervised_vae_loss(x_recon, x, mu, logvar, y_logits=None, y_true=None, beta=1.0):
"""
半监督VAE的损失函数
有标签数据的损失：
L = 重建损失 + β * KL散度 + α * 分类损失
无标签数据的损失：
L = 重建损失 + β * KL散度
参数:
x_recon: 重建的输入
x: 原始输入
mu: 潜在分布均值
logvar: 潜在分布对数方差
y_logits: 分类logits（仅无标签数据使用）
y_true: 真实标签（仅有标签数据使用）
beta: KL散度的权重系数（β-VAE思想）
"""
# 重建损失（二元交叉熵或MSE）
recon_loss = nn.functional.mse_loss( # 使用MSE衡量重建质量
x_recon, x, reduction='sum' # 求和（而非平均）
)
# KL散度损失
# KL(q(z|x) || p(z)) = -0.5 * Σ(1 + log σ² - μ² - σ²)
# 其中 p(z) = N(0, I)
kl_loss = -0.5 * torch.sum( # 计算KL散度
1 + logvar - mu.pow(2) - logvar.exp() # 公式展开
)
total_loss = recon_loss + beta * kl_loss # 基础损失 = 重建 + β*KL
# 如果有标签数据，添加分类损失
if y_true is not None and y_logits is not None: # 检查是否有分类信息
cls_loss = nn.functional.cross_entropy( # 交叉熵分类损失
y_logits, y_true, reduction='sum'
)
total_loss += 0.1 * cls_loss # 加入分类损失（权重较小）
return total_loss, recon_loss, kl_loss # 返回总损失及各分量
# ============================================================
# 执行实验
# ============================================================
def run_semi_supervised_experiment():
"""运行半监督学习实验"""
# 生成数据
features, labels, labels_partial, labeled_idx = generate_semi_supervised_data(
n_total=500, # 总样本数
n_labeled=50, # 有标签数（仅10%）
n_features=20, # 特征维度
n_classes=3 # 类别数
)
print("="*60)
print("实验1：自训练半监督学习")
print("="*60)
# 自训练方法
self_trainer = SemiSupervisedSelfTraining( # 创建自训练对象
input_dim=20, # 输入维度
n_classes=3 # 类别数
)
pseudo_labels = self_trainer.self_train( # 执行自训练
features, # 所有数据
labels_partial, # 部分标签
confidence_threshold=0.85, # 置信度阈值
max_iterations=5 # 最大迭代数
)
# 评估最终结果
final_accuracy = (pseudo_labels == labels).float().mean() # 计算标签准确率
print(f"最终标签恢复准确率: {final_accuracy*100:.2f}%")
print("\n" + "="*60)
print("实验2：半监督VAE学习潜在因果表示")
print("="*60)
# 半监督VAE
input_dim = 20 # 输入维度
latent_dim = 5 # 潜在空间维度（因果因素数量）
n_classes = 3 # 类别数
vae = SemiSupervisedVAE(input_dim, latent_dim, n_classes) # 创建VAE
optimizer = optim.Adam(vae.parameters(), lr=0.001) # Adam优化器
# 分离有标签和无标签数据
labeled_mask = labels_partial >= 0 # 有标签掩码
unlabeled_mask = labels_partial < 0 # 无标签掩码
X_labeled = features[labeled_mask] # 有标签数据
y_labeled = labels_partial[labeled_mask] # 有标签标签
X_unlabeled = features[unlabeled_mask] # 无标签数据
epochs = 100 # 训练轮数
vae.train() # 训练模式
for epoch in range(epochs): # 遍历epoch
# 训练有标签数据
x_recon_l, mu_l, logvar_l = vae(X_labeled, y_labeled) # 有标签数据前向传播
y_logits_l = vae.classifier(X_labeled) # 获取分类输出
loss_l, recon_l, kl_l = semi_supervised_vae_loss( # 计算有标签损失
x_recon_l, X_labeled, mu_l, logvar_l,
y_logits=y_logits_l, y_true=y_labeled
)
# 训练无标签数据
x_recon_u, mu_u, logvar_u = vae(X_unlabeled, y=None) # 无标签数据前向传播
loss_u, recon_u, kl_u = semi_supervised_vae_loss( # 计算无标签损失
x_recon_u, X_unlabeled, mu_u, logvar_u
)
# 总损失
total_loss = loss_l + loss_u # 有标签+无标签损失
optimizer.zero_grad() # 清除梯度
total_loss.backward() # 反向传播
optimizer.step() # 更新参数
if (epoch + 1) % 20 == 0: # 每20轮打印
print(f"Epoch [{epoch+1}/{epochs}], "
f"有标签损失: {loss_l.item():.2f}, "
f"无标签损失: {loss_u.item():.2f}")
# 评估分类准确率
vae.eval() # 评估模式
with torch.no_grad(): # 关闭梯度
y_pred_logits = vae.classifier(features) # 对所有数据预测
y_pred = y_pred_logits.argmax(dim=1) # 获取预测类别
accuracy = (y_pred == labels).float().mean() # 计算准确率
print(f"\n半监督VAE分类准确率: {accuracy*100:.2f}%")
# 对比：仅用有标签数据训练的准确率
print("\n对比实验：仅用有标签数据训练")
simple_model = nn.Sequential( # 简单分类器
nn.Linear(20, 64), # 全连接层
nn.ReLU(), # 激活
nn.Linear(64, 3), # 输出层
)
simple_optimizer = optim.Adam(simple_model.parameters(), lr=0.001)
simple_criterion = nn.CrossEntropyLoss()
for _ in range(100): # 训练100轮
logits = simple_model(X_labeled) # 前向传播
loss = simple_criterion(logits, y_labeled) # 计算损失
simple_optimizer.zero_grad() # 清除梯度
loss.backward() # 反向传播
simple_optimizer.step() # 更新
with torch.no_grad(): # 评估
pred = simple_model(features).argmax(dim=1) # 预测
supervised_acc = (pred == labels).float().mean() # 计算准确率
print(f"仅监督学习准确率: {supervised_acc*100:.2f}%")
# 运行实验
run_semi_supervised_experiment() # 执行半监督学习实验
四、分布式表示（Distributed Representation）
4.1 核心知识点
分布式表示 vs 局部表示（Local Representation）：
| 特性 | 局部表示 | 分布式表示 |
|---|---|---|
| 表示方式 | 每个概念由一个单独的神经元表示（one-hot） | 每个概念由多个神经元的激活模式共同表示 |
| 容量 | n个神经元最多表示n个概念 | n个神经元可以表示2^n个概念 |
| 泛化能力 | 差（概念之间独立） | 好（共享特征实现泛化） |
| 例子 | one-hot编码 | 词嵌入、深度学习的隐藏层特征 |
核心优势：
- 指数级表达能力： k个二值特征可以表示2^k个不同概念
- 共享统计强度： 相似的概念共享部分特征，使学习更高效
- 组合泛化： 可以表示从未见过的概念组合
4.2 代码案例：词嵌入（Word Embedding）中的分布式表示
import torch # 导入PyTorch
import torch.nn as nn # 导入神经网络模块
import torch.optim as optim # 导入优化器
import numpy as np # 导入NumPy
torch.manual_seed(42) # 设置随机种子
# ============================================================
# 构建小型语料库和词汇表
# ============================================================
corpus_sentences = [ # 定义语料库（示例句子列表）
"the king loves the queen", # 句子1
"the queen loves the king", # 句子2
"the prince is a young king", # 句子3
"the princess is a young queen", # 句子4
"the king is a man", # 句子5
"the queen is a woman", # 句子6
"a prince is a young man", # 句子7
"a princess is a young woman", # 句子8
]
# 构建词汇表
word_to_idx = {} # 词 -> 索引的映射字典
idx_to_word = {} # 索引 -> 词的映射字典
word_freq = {} # 词频统计字典
for sentence in corpus_sentences: # 遍历每个句子
words = sentence.lower().split() # 将句子转为小写并分词
for word in words: # 遍历每个词
if word not in word_to_idx: # 如果词不在词汇表中
idx = len(word_to_idx) # 获取新索引
word_to_idx[word] = idx # 建立词->索引映射
idx_to_word[idx] = word # 建立索引->词映射
word_freq[word] = 0 # 初始化词频
word_freq[word] += 1 # 增加词频
vocab_size = len(word_to_idx) # 词汇表大小
print(f"词汇表大小: {vocab_size}") # 打印词汇表大小
print(f"词汇表: {word_to_idx}") # 打印词汇表
# ============================================================
# 生成Skip-gram训练数据（词共现对）
# ============================================================
def generate_skipgram_pairs(sentences, word_to_idx, window_size=2):
"""
生成Skip-gram训练对
Skip-gram模型：给定中心词，预测上下文词
参数:
sentences: 句子列表
word_to_idx: 词到索引的映射
window_size: 上下文窗口大小（中心词前后各取多少个词）
返回:
pairs: (中心词索引, 上下文词索引) 的列表
"""
pairs = [] # 初始化训练对列表
for sentence in sentences: # 遍历每个句子
words = sentence.lower().split() # 分词
indices = [word_to_idx[w] for w in words] # 将词转换为索引
for i, center_word in enumerate(indices): # 遍历每个词作为中心词
# 在窗口范围内取上下文词
for j in range( # 遍历窗口范围
max(0, i - window_size), # 左边界（不超出句子开头）
min(len(indices), i + window_size + 1) # 右边界（不超出句子结尾）
):
if i != j: # 排除中心词本身
context_word = indices[j] # 获取上下文词索引
pairs.append((center_word, context_word)) # 添加训练对
return pairs # 返回训练对列表
# 生成训练数据
training_pairs = generate_skipgram_pairs( # 调用函数生成训练对
corpus_sentences, word_to_idx, window_size=2
)
print(f"训练对数量: {len(training_pairs)}") # 打印训练对数量
# ============================================================
# 定义Skip-gram模型
# ============================================================
class SkipGramModel(nn.Module):
"""
Skip-gram词嵌入模型
架构：
中心词 -> 嵌入层(查找表) -> 投影层 -> 输出层(softmax) -> 上下文词预测
核心思想：用中心词的嵌入来预测其上下文词
训练完成后，嵌入层的权重就是每个词的分布式表示
"""
def __init__(self, vocab_size, embedding_dim):
"""
构造函数
参数:
vocab_size: 词汇表大小
embedding_dim: 嵌入维度（分布式表示的维度）
"""
super(SkipGramModel, self).__init__()
# 输入嵌入矩阵 W_in: vocab_size × embedding_dim
# 每一行对应一个词的嵌入向量（分布式表示）
self.input_embeddings = nn.Embedding( # 创建嵌入层
vocab_size, # 嵌入表大小=词汇表大小
embedding_dim # 每个词的嵌入维度
)
# 输出嵌入矩阵 W_out: embedding_dim × vocab_size
# 用于计算与上下文词的相似度
self.output_embeddings = nn.Embedding( # 创建输出嵌入层
vocab_size, # 嵌入表大小=词汇表大小
embedding_dim # 嵌入维度（与输入相同）
)
# 初始化嵌入权重
nn.init.xavier_uniform_(self.input_embeddings.weight) # Xavier初始化输入嵌入
nn.init.xavier_uniform_(self.output_embeddings.weight) # Xavier初始化输出嵌入
def forward(self, center_words, context_words, negative_words):
"""
前向传播（使用负采样优化）
负采样思想：
- 正样本：(中心词, 真实上下文词)
- 负样本：(中心词, 随机采样的非上下文词)
- 目标：最大化正样本的相似度，最小化负样本的相似度
参数:
center_words: 中心词索引，形状(batch_size,)
context_words: 正样本上下文词索引，形状(batch_size,)
negative_words: 负样本词索引，形状(batch_size, n_neg)
返回:
loss: 损失值
"""
# 获取中心词的嵌入向量
center_embeds = self.input_embeddings(center_words) # (batch_size, embed_dim)
# 获取正样本上下文词的嵌入向量
context_embeds = self.output_embeddings(context_words) # (batch_size, embed_dim)
# 获取负样本词的嵌入向量
negative_embeds = self.output_embeddings(negative_words) # (batch_size, n_neg, embed_dim)
# 计算正样本得分：中心词嵌入与上下文词嵌入的点积
pos_score = torch.sum( # 点积求和
center_embeds * context_embeds, # 逐元素相乘
dim=1 # 沿嵌入维度求和
) # (batch_size,)
pos_score = torch.clamp(pos_score, max=10) # 防止数值溢出
pos_loss = -torch.log(torch.sigmoid(pos_score) + 1e-10) # 正样本损失：-log σ(s)
# 计算负样本得分
neg_score = torch.bmm( # 批量矩阵乘法
negative_embeds, # (batch_size, n_neg, embed_dim)
center_embeds.unsqueeze(2) # (batch_size, embed_dim, 1)
).squeeze(2) # (batch_size, n_neg)
neg_score = torch.clamp(neg_score, max=10) # 防止数值溢出
neg_loss = -torch.log(1 - torch.sigmoid(neg_score) + 1e-10) # 负样本损失：-log(1-σ(s))
# 总损失 = 正样本损失 + 负样本损失之和
loss = pos_loss.mean() + neg_loss.sum(dim=1).mean() # 计算平均损失
return loss # 返回损失值
def get_embedding(self, word_idx):
"""
获取指定词的分布式表示（嵌入向量）
参数:
word_idx: 词的索引
返回:
该词的嵌入向量
"""
with torch.no_grad(): # 关闭梯度计算
return self.input_embeddings( # 从嵌入层获取向量
torch.LongTensor([word_idx]) # 传入索引
).squeeze(0) # 去除多余维度
# ============================================================
# 负采样函数
# ============================================================
def get_negative_samples(batch_size, n_neg, vocab_size, positive_indices):
"""
生成负采样样本
使用基于词频的采样分布（频率高的词更容易被采为负样本）
参数:
batch_size: 批次大小
n_neg: 每个正样本对应的负样本数
vocab_size: 词汇表大小
positive_indices: 正样本索引（需要排除）
返回:
负样本索引张量，形状(batch_size, n_neg)
"""
# 简单均匀采样（实际中常用频率的3/4次方作为采样分布）
negative_samples = torch.randint( # 随机生成整数
0, vocab_size, # 范围：[0, vocab_size)
(batch_size, n_neg) # 形状
)
return negative_samples # 返回负样本
# ============================================================
# 训练Skip-gram模型
# ============================================================
embedding_dim = 20 # 嵌入维度
n_neg = 5 # 负样本数量
epochs = 200 # 训练轮数
batch_size = 32 # 批次大小
learning_rate = 0.01 # 学习率
model = SkipGramModel(vocab_size, embedding_dim) # 创建Skip-gram模型
optimizer = optim.Adam(model.parameters(), lr=learning_rate) # Adam优化器
# 将训练对转换为张量
centers = torch.LongTensor([p[0] for p in training_pairs]) # 中心词索引张量
contexts = torch.LongTensor([p[1] for p in training_pairs]) # 上下文词索引张量
print("\n" + "="*60)
print("开始训练Skip-gram模型...")
model.train() # 设置训练模式
for epoch in range(epochs): # 遍历每个epoch
total_loss = 0 # 初始化累计损失
# 随机打乱训练数据
perm = torch.randperm(len(centers)) # 生成随机排列
centers_shuffled = centers[perm] # 打乱中心词
contexts_shuffled = contexts[perm] # 打乱上下文词
# 分批训练
for i in range(0, len(centers), batch_size): # 按批次遍历
batch_centers = centers_shuffled[i:i+batch_size] # 取出当前批次的中心词
batch_contexts = contexts_shuffled[i:i+batch_size] # 取出当前批次的上下文词
# 生成负样本
negative_words = get_negative_samples( # 调用负采样函数
len(batch_centers), # 当前批次大小
n_neg, # 每个正样本的负样本数
vocab_size, # 词汇表大小
batch_contexts # 正样本索引（需排除）
)
# 前向传播
loss = model( # 通过模型计算损失
batch_centers, # 中心词
batch_contexts, # 上下文词
negative_words # 负样本
)
# 反向传播
optimizer.zero_grad() # 清除旧梯度
loss.backward() # 计算新梯度
optimizer.step() # 更新参数
total_loss += loss.item() # 累加损失
if (epoch + 1) % 50 == 0: # 每50轮打印
print(f"Epoch [{epoch+1}/{epochs}], "
f"损失: {total_loss / (len(centers) // batch_size):.4f}")
# ============================================================
# 分析学到的分布式表示
# ============================================================
print("\n" + "="*60)
print("分布式表示分析")
print("="*60)
def cosine_similarity(vec_a, vec_b):
"""
计算两个向量的余弦相似度
参数:
vec_a: 向量a
vec_b: 向量b
返回:
余弦相似度值（范围-1到1）
"""
dot_product = torch.dot(vec_a, vec_b) # 计算点积
norm_a = torch.norm(vec_a) # 计算a的L2范数
norm_b = torch.norm(vec_b) # 计算b的L2范数
return dot_product / (norm_a * norm_b + 1e-8) # 余弦相似度公式
# 获取所有词的嵌入
print("\n各词的分布式表示（前5维）:")
print("-" * 40)
for word, idx in word_to_idx.items(): # 遍历词汇表
embedding = model.get_embedding(idx) # 获取嵌入向量
print(f" {word:12s}: {embedding[:5].numpy().round(3)}") # 打印前5维
# 计算词之间的相似度
print("\n词相似度矩阵（余弦相似度）:")
print("-" * 40)
# 选择部分词进行相似度分析
selected_words = ["king", "queen", "prince", "princess", "man", "woman"] # 选定词
selected_indices = [word_to_idx[w] for w in selected_words if w in word_to_idx] # 获取索引
# 打印相似度表头
print(f"{'':12s}", end="") # 空白对齐
for word in selected_words: # 遍历选定词
if word in word_to_idx: # 如果在词汇表中
print(f"{word:>12s}", end="") # 打印列名
print()
for w1 in selected_words: # 遍历行
if w1 not in word_to_idx: # 跳过不在词汇表中的词
continue
print(f"{w1:12s}", end="") # 打印行名
emb1 = model.get_embedding(word_to_idx[w1]) # 获取词1的嵌入
for w2 in selected_words: # 遍历列
if w2 not in word_to_idx: # 跳过不在词汇表中的词
continue
emb2 = model.get_embedding(word_to_idx[w2]) # 获取词2的嵌入
sim = cosine_similarity(emb1, emb2) # 计算余弦相似度
print(f"{sim.item():>12.4f}", end="") # 打印相似度值
print() # 换行
# 向量运算示例：king - man + woman ≈ queen
print("\n向量运算示例:")
print("-" * 40)
if all(w in word_to_idx for w in ["king", "man", "woman", "queen"]): # 检查词是否在词汇表
king_vec = model.get_embedding(word_to_idx["king"]) # king的向量
man_vec = model.get_embedding(word_to_idx["man"]) # man的向量
woman_vec = model.get_embedding(word_to_idx["woman"]) # woman的向量
queen_vec = model.get_embedding(word_to_idx["queen"]) # queen的向量
# 计算 king - man + woman
result_vec = king_vec - man_vec + woman_vec # 类比向量运算
# 找到最相似的词
best_word = None # 最佳匹配词
best_sim = -1 # 最佳相似度
for word, idx in word_to_idx.items(): # 遍历词汇表
if word in ["king", "man", "woman"]: # 排除参与运算的词
continue
emb = model.get_embedding(idx) # 获取当前词的嵌入
sim = cosine_similarity(result_vec, emb) # 计算与结果向量的相似度
if sim > best_sim: # 如果更相似
best_sim = sim # 更新最佳相似度
best_word = word # 更新最佳匹配词
print(f" king - man + woman = ?") # 打印运算表达式
print(f" 最匹配的词: {best_word} (相似度: {best_sim:.4f})") # 打印结果
# 展示分布式表示的维度含义
print("\n" + "="*60)
print("分布式表示 vs 局部表示（One-Hot）的对比")
print("="*60)
print("\n局部表示（One-Hot编码）:")
for word in ["king", "queen", "man"]: # 遍历示例词
idx = word_to_idx.get(word) # 获取索引
if idx is not None: # 如果存在
onehot = np.zeros(vocab_size) # 创建零向量
onehot[idx] = 1 # 对应位置设为1
print(f" {word:12s}: {onehot.astype(int)}") # 打印one-hot编码
print("\n分布式表示（Embedding）:")
for word in ["king", "queen", "man"]: # 遍历示例词
idx = word_to_idx.get(word) # 获取索引
if idx is not None: # 如果存在
emb = model.get_embedding(idx) # 获取嵌入向量
print(f" {word:12s}: {emb.detach().numpy().round(3)[:8]}...") # 打印前8维
print(f"\n词汇表大小: {vocab_size}")
print(f"One-Hot维度: {vocab_size}（稀疏，只有1个非零元素）")
print(f"分布式表示维度: {embedding_dim}（密集，所有元素都参与编码）")
print(f"表达能力: One-Hot最多编码{vocab_size}个概念, "
f"分布式表示可编码2^{embedding_dim}={2**embedding_dim}个概念")
五、得益于深度的指数增益（Exponential Gains from Depth）
5.1 核心知识点
核心定理： 深度网络可以用指数级少的神经元来表示某些函数，而浅层网络需要指数级多的神经元才能表示相同的函数。
具体来说：
- 存在一类函数，深度为k的网络用O(n)个神经元就能表示
- 而深度为k-1的网络需要O(2^n)个神经元
- 深度使得网络能够以分层组合的方式构建复杂的函数
直观理解： 每增加一层，网络可以在上一层的基础上进行进一步的组合和抽象，就像"折叠"空间一样，逐层构建越来越复杂的决策边界。
5.2 代码案例：深度 vs 宽度的表达能力对比
import torch # 导入PyTorch
import torch.nn as nn # 导入神经网络模块
import torch.optim as optim # 导入优化器
import numpy as np # 导入NumPy
import time # 导入时间模块（用于计时）
torch.manual_seed(42) # 设置随机种子
# ============================================================
# 实验1：深层网络 vs 浅层网络学习组合函数的能力
# ============================================================
print("="*60)
print("实验1：深度网络 vs 浅层网络表达能力对比")
print("="*60)
# 创建一个需要多层组合才能学习的目标函数
# 目标：f(x) = sin(x1) * cos(x2) + x3^2 - |x4|
# 这个函数需要多层非线性变换才能高效表示
def target_function(X):
"""
目标函数：一个具有多层次结构的复杂函数
参数:
X: 输入矩阵，形状(n_samples, 4)
返回:
函数值，形状(n_samples, 1)
"""
x1, x2, x3, x4 = X[:, 0:1], X[:, 1:2], X[:, 2:3], X[:, 3:4] # 分离各维度
return torch.sin(x1) * torch.cos(x2) + x3**2 - torch.abs(x4) # 组合函数
# 生成训练数据
n_train = 2000 # 训练样本数
n_test = 500 # 测试样本数
input_dim = 4 # 输入维度
X_train = torch.randn(n_train, input_dim) * 2 # 训练数据（范围-2到2）
X_test = torch.randn(n_test, input_dim) * 2 # 测试数据
y_train = target_function(X_train) # 训练标签
y_test = target_function(X_test) # 测试标签
# ============================================================
# 定义不同深度的网络
# ============================================================
def create_deep_network(input_dim, hidden_dim, output_dim, n_layers):
"""
创建指定深度的全连接网络
参数:
input_dim: 输入维度
hidden_dim: 隐藏层维度
output_dim: 输出维度
n_layers: 隐藏层数量（深度）
返回:
网络模型
"""
layers = [] # 初始化层列表
# 第一层：输入 -> 隐藏
layers.append(nn.Linear(input_dim, hidden_dim)) # 第一个全连接层
layers.append(nn.ReLU()) # ReLU激活函数
layers.append(nn.BatchNorm1d(hidden_dim)) # 批归一化
# 中间层：隐藏 -> 隐藏
for _ in range(n_layers - 1): # 添加n_layers-1个中间层
layers.append(nn.Linear(hidden_dim, hidden_dim)) # 全连接层
layers.append(nn.ReLU()) # ReLU激活函数
layers.append(nn.BatchNorm1d(hidden_dim)) # 批归一化
# 输出层：隐藏 -> 输出
layers.append(nn.Linear(hidden_dim, output_dim)) # 输出全连接层
return nn.Sequential(*layers) # 返回顺序容器
# ============================================================
# 创建不同深度和宽度的网络
# ============================================================
# 配置列表：(名称, 深度, 隐藏维度, 总参数量说明)
configs = [
("浅而宽 (2层, 512单元)", 2, 512), # 浅层宽网络
("中等 (4层, 256单元)", 4, 256), # 中等深度网络
("深而窄 (8层, 128单元)", 8, 128), # 深层窄网络
("更深 (12层, 64单元)", 12, 64), # 更深网络
]
models = {} # 存储所有模型
results = {} # 存储训练结果
for name, depth, hidden_dim in configs: # 遍历每种配置
model = create_deep_network( # 创建网络
input_dim=input_dim, # 输入维度
hidden_dim=hidden_dim, # 隐藏层维度
output_dim=1, # 输出维度
n_layers=depth # 网络深度
)
# 计算参数量
n_params = sum( # 计算总参数数量
p.numel() for p in model.parameters() # 遍历所有参数并计算元素数
)
print(f"\n{name}") # 打印配置名
print(f" 参数量: {n_params:,}") # 打印参数量
models[name] = model # 存储模型
# ============================================================
# 训练函数
# ============================================================
def train_and_evaluate(model, X_train, y_train, X_test, y_test,
epochs=200, lr=0.001, model_name=""):
"""
训练并评估模型
参数:
model: 神经网络模型
X_train: 训练输入
y_train: 训练标签
X_test: 测试输入
y_test: 测试标签
epochs: 训练轮数
lr: 学习率
model_name: 模型名称（用于打印）
返回:
训练损失历史和测试损失历史
"""
optimizer = optim.Adam(model.parameters(), lr=lr) # Adam优化器
scheduler = optim.lr_scheduler.StepLR( # 学习率调度器
optimizer, step_size=50, gamma=0.5 # 每50轮学习率减半
)
criterion = nn.MSELoss() # 均方误差损失
train_losses = [] # 训练损失历史
test_losses = [] # 测试损失历史
dataset = TensorDataset(X_train, y_train) # 创建训练数据集
loader = DataLoader(dataset, batch_size=64, shuffle=True) # 数据加载器
start_time = time.time() # 记录开始时间
model.train() # 训练模式
for epoch in range(epochs): # 遍历每个epoch
epoch_loss = 0 # 初始化epoch损失
for batch_x, batch_y in loader: # 遍历每个批次
pred = model(batch_x) # 前向传播
loss = criterion(pred, batch_y) # 计算损失
optimizer.zero_grad() # 清除梯度
loss.backward() # 反向传播
# 梯度裁剪（防止深层网络的梯度爆炸）
torch.nn.utils.clip_grad_norm_( # 裁剪梯度范数
model.parameters(), # 模型参数
max_norm=1.0 # 最大范数
)
optimizer.step() # 更新参数
epoch_loss += loss.item() # 累加损失
scheduler.step() # 更新学习率
# 记录训练和测试损失
model.eval() # 评估模式
with torch.no_grad(): # 关闭梯度
train_pred = model(X_train) # 训练集预测
test_pred = model(X_test) # 测试集预测
train_loss = criterion(train_pred, y_train).item() # 训练MSE
test_loss = criterion(test_pred, y_test).item() # 测试MSE
train_losses.append(train_loss) # 记录训练损失
test_losses.append(test_loss) # 记录测试损失
model.train() # 恢复训练模式
if (epoch + 1) % 50 == 0: # 每50轮打印
print(f" Epoch [{epoch+1}/{epochs}], "
f"训练MSE: {train_loss:.6f}, "
f"测试MSE: {test_loss:.6f}")
elapsed = time.time() - start_time # 计算训练耗时
print(f" 训练耗时: {elapsed:.2f}秒") # 打印耗时
return train_losses, test_losses # 返回损失历史
# ============================================================
# 训练所有模型并对比
# ============================================================
from torch.utils.data import DataLoader, TensorDataset # 导入数据工具
print("\n" + "="*60)
print("开始训练和对比...")
for name, model in models.items(): # 遍历所有模型
print(f"\n训练: {name}") # 打印模型名
train_losses, test_losses = train_and_evaluate( # 训练并评估
model, X_train, y_train, X_test, y_test, # 数据
epochs=200, lr=0.001, model_name=name # 参数
)
results[name] = { # 存储结果
'train_loss': train_losses[-1], # 最终训练损失
'test_loss': test_losses[-1], # 最终测试损失
}
# 打印最终对比结果
print("\n" + "="*60)
print("最终结果对比:")
print("="*60)
print(f"{'模型名称':<30s} {'训练MSE':>12s} {'测试MSE':>12s}")
print("-" * 56)
for name in results: # 遍历结果
r = results[name] # 获取当前模型的结果
print(f"{name:<30s} {r['train_loss']:>12.6f} {r['test_loss']:>12.6f}") # 打印
# ============================================================
# 实验2：理论证明 —— 深度的指数效率
# ============================================================
print("\n" + "="*60)
print("实验2：深度的指数效率（奇偶函数实验）")
print("="*60)
def parity_function(X):
"""
奇偶函数（Parity Function）：输入为二进制串，输出为1的个数的奇偶性
这是一个经典的需要深度的函数：
- 深度为O(log n)的网络可以用O(n)个神经元计算
- 浅层网络需要O(2^n)个神经元
参数:
X: 二进制输入矩阵，形状(n_samples, n_bits)
返回:
奇偶函数值，形状(n_samples, 1)
"""
return torch.sum(X, dim=1, keepdim=True) % 2 # 求和后取模2
# 生成奇偶函数数据
n_bits = 8 # 二进制位数
n_samples = 5000 # 样本数
X_parity = torch.randint(0, 2, (n_samples, n_bits)).float() # 随机二进制输入
y_parity = parity_function(X_parity) # 计算奇偶标签
# 分割训练集和测试集
split = int(0.8 * n_samples) # 80%训练
X_parity_train, X_parity_test = X_parity[:split], X_parity[split:] # 分割输入
y_parity_train, y_parity_test = y_parity[:split], y_parity[split:] # 分割标签
# 定义浅层和深层网络
class ShallowParityNet(nn.Module):
"""
浅层网络：理论上需要O(2^n)个隐藏单元才能表示奇偶函数
"""
def __init__(self, n_bits, hidden_size):
"""
构造函数
参数:
n_bits: 输入位数
hidden_size: 隐藏层大小
"""
super(ShallowParityNet, self).__init__()
self.net = nn.Sequential(
nn.Linear(n_bits, hidden_size), # 输入层 -> 隐藏层
nn.ReLU(), # ReLU激活
nn.Linear(hidden_size, 1), # 隐藏层 -> 输出
nn.Sigmoid(), # Sigmoid（输出0-1之间的概率）
)
def forward(self, x):
"""前向传播"""
return self.net(x) # 顺序执行
class DeepParityNet(nn.Module):
"""
深层网络：理论上只需O(n)个神经元（O(log n)层）
"""
def __init__(self, n_bits, hidden_size, n_layers):
"""
构造函数
参数:
n_bits: 输入位数
hidden_size: 隐藏层大小
n_layers: 隐藏层数量
"""
super(DeepParityNet, self).__init__()
layers = [nn.Linear(n_bits, hidden_size), nn.ReLU()] # 第一层
for _ in range(n_layers - 1): # 添加中间层
layers.extend([nn.Linear(hidden_size, hidden_size), nn.ReLU()])
layers.extend([nn.Linear(hidden_size, 1), nn.Sigmoid()]) # 输出层
self.net = nn.Sequential(*layers) # 创建顺序容器
def forward(self, x):
"""前向传播"""
return self.net(x) # 顺序执行
# 训练对比函数
def train_parity_model(model, X_train, y_train, X_test, y_test, epochs=100, lr=0.001):
"""
训练奇偶函数模型
参数:
model: 神经网络模型
X_train, y_train: 训练数据和标签
X_test, y_test: 测试数据和标签
epochs: 训练轮数
lr: 学习率
返回:
最终测试准确率
"""
optimizer = optim.Adam(model.parameters(), lr=lr) # Adam优化器
criterion = nn.BCELoss() # 二元交叉熵损失
dataset = TensorDataset(X_train, y_train) # 创建数据集
loader = DataLoader(dataset, batch_size=128, shuffle=True) # 数据加载器
model.train() # 训练模式
for epoch in range(epochs): # 遍历epoch
for batch_x, batch_y in loader: # 遍历批次
pred = model(batch_x) # 前向传播
loss = criterion(pred, batch_y) # 计算损失
optimizer.zero_grad() # 清除梯度
loss.backward() # 反向传播
optimizer.step() # 更新参数
# 评估
model.eval() # 评估模式
with torch.no_grad(): # 关闭梯度
pred_test = model(X_test) # 测试集预测
pred_labels = (pred_test > 0.5).float() # 二值化预测
accuracy = (pred_labels == y_test).float().mean() # 计算准确率
return accuracy.item() # 返回准确率
# 对比实验
print("\n浅层网络实验（不同隐藏单元数）:")
for hidden_size in [16, 64, 256, 1024]: # 不同宽度
model = ShallowParityNet(n_bits, hidden_size) # 创建浅层网络
n_params = sum(p.numel() for p in model.parameters()) # 计算参数量
acc = train_parity_model(model, X_parity_train, y_parity_train, # 训练
X_parity_test, y_parity_test)
print(f" 隐藏单元: {hidden_size:5d}, 参数量: {n_params:8d}, 准确率: {acc*100:.2f}%")
print("\n深层网络实验（不同深度）:")
for n_layers in [1, 2, 3, 4]: # 不同深度
hidden_size = 32 # 固定隐藏层宽度
model = DeepParityNet(n_bits, hidden_size, n_layers) # 创建深层网络
n_params = sum(p.numel() for p in model.parameters()) # 计算参数量
acc = train_parity_model(model, X_parity_train, y_parity_train, # 训练
X_parity_test, y_parity_test)
print(f" 深度: {n_layers}, 隐藏单元: {hidden_size}, "
f"参数量: {n_params:8d}, 准确率: {acc*100:.2f}%")
六、提供发现潜在原因的线索（Providing Clues for Discovering Underlying Causes）
6.1 核心知识点
核心思想： 好的表示学习应该能够发现数据背后的潜在生成因素（latent generative factors）。这些因素是产生观测数据的真正原因。
因果层次：
观测数据 x ← [混杂/变换] ← 潜在因素 z₁, z₂, ..., zₖ
↑
我们想要恢复的
关键观点：
- 数据的变化（variation）通常由少量潜在因素控制
- 表示学习的目标是解纠缠（disentangle）这些因素
- 每个潜在维度应该对应一个有意义的、独立的变化因素
解纠缠表示的评价标准：
- 每个潜在维度只捕获一个因素
- 不同因素之间的变化是独立的
- 表示是可解释的
6.2 代码案例：β-VAE 学习解纠缠的潜在表示
import torch # 导入PyTorch
import torch.nn as nn # 导入神经网络模块
import torch.optim as optim # 导入优化器
import numpy as np # 导入NumPy
torch.manual_seed(42) # 设置随机种子
np.random.seed(42) # 设置NumPy随机种子
# ============================================================
# 生成具有已知潜在因素的数据集
# 模拟场景：2D图形数据，有3个潜在因素控制变化
# z1: 水平位置 (0到1)
# z2: 垂直位置 (0到1)
# z3: 大小 (0.5到1.5)
# ============================================================
def generate_factor_data(n_samples=5000, image_size=16):
"""
生成具有已知潜在因素的数据集
每个样本是一个小图像，由3个潜在因素生成：
- z1: 水平位置
- z2: 垂直位置
- z3: 圆的大小
参数:
n_samples: 样本数量
image_size: 图像尺寸（正方形）
返回:
images: 生成的图像数据，形状(n_samples, image_size*image_size)
factors: 潜在因素，形状(n_samples, 3)
"""
images = np.zeros((n_samples, image_size, image_size)) # 初始化图像数组
factors = np.zeros((n_samples, 3)) # 初始化因素数组
for i in range(n_samples): # 遍历每个样本
# 随机生成潜在因素
z1 = np.random.uniform(0.2, 0.8) # 水平位置（范围0.2-0.8，避免边界）
z2 = np.random.uniform(0.2, 0.8) # 垂直位置
z3 = np.random.uniform(1.0, 3.0) # 圆的半径
factors[i] = [z1, z2, z3] # 保存潜在因素
# 根据潜在因素生成图像（画一个圆）
cx = int(z1 * image_size) # 圆心x坐标（像素）
cy = int(z2 * image_size) # 圆心y坐标（像素）
r = z3 # 圆的半径
# 创建坐标网格
y_grid, x_grid = np.mgrid[ # 生成网格坐标
0:image_size, # y方向范围
0:image_size # x方向范围
]
# 计算每个像素到圆心的距离
distance = np.sqrt( # 欧几里得距离
(x_grid - cx) ** 2 + (y_grid - cy) ** 2
)
# 距离小于半径的像素设为1（画圆）
images[i][distance <= r] = 1.0 # 圆内像素为1
# 展平图像并添加噪声
images_flat = images.reshape(n_samples, -1) # 展平为(n_samples, image_size^2)
images_flat += np.random.normal( # 添加高斯噪声
0, 0.05, images_flat.shape # 均值0，标准差0.05
)
images_flat = np.clip(images_flat, 0, 1) # 裁剪到[0,1]范围
return ( # 返回数据
torch.FloatTensor(images_flat), # 图像张量
torch.FloatTensor(factors) # 因素张量
)
# ============================================================
# 定义β-VAE模型
# ============================================================
class BetaVAE(nn.Module):
"""
β-VAE：用于学习解纠缠潜在表示的变分自编码器
与标准VAE的区别：
- 标准VAE：L = 重建损失 + KL散度
- β-VAE： L = 重建损失 + β * KL散度 (β > 1)
更大的β迫使模型学习更独立、更解纠缠的潜在表示
代价是重建质量可能略有下降
关键概念：
- KL散度惩罚：防止潜在分布偏离先验太远
- β > 1：更强的KL惩罚，促进解纠缠
- 重建损失：确保潜在表示保留足够信息
"""
def __init__(self, input_dim, latent_dim, hidden_dims=[256, 128]):
"""
构造函数
参数:
input_dim: 输入维度（图像展平后的维度）
latent_dim: 潜在空间维度（我们希望发现的因果因素数量）
hidden_dims: 编码器/解码器的隐藏层维度列表
"""
super(BetaVAE, self).__init__()
self.latent_dim = latent_dim # 保存潜在维度
# ========== 编码器 q(z|x) ==========
encoder_layers = [] # 初始化编码器层列表
prev_dim = input_dim # 上一层的维度（初始为输入维度）
for h_dim in hidden_dims: # 遍历隐藏层维度
encoder_layers.extend([ # 添加层
nn.Linear(prev_dim, h_dim), # 全连接层
nn.LayerNorm(h_dim), # 层归一化（比BatchNorm更适合小batch）
nn.ReLU(), # ReLU激活
])
prev_dim = h_dim # 更新上一层维度
self.encoder = nn.Sequential(*encoder_layers) # 创建编码器
# 潜在分布参数（均值和对数方差）
self.fc_mu = nn.Linear(hidden_dims[-1], latent_dim) # 均值网络
self.fc_logvar = nn.Linear(hidden_dims[-1], latent_dim) # 对数方差网络
# ========== 解码器 p(x|z) ==========
decoder_layers = [] # 初始化解码器层列表
prev_dim = latent_dim # 上一层维度（初始为潜在维度）
for h_dim in reversed(hidden_dims): # 反向遍历隐藏层维度
decoder_layers.extend([ # 添加层
nn.Linear(prev_dim, h_dim), # 全连接层
nn.LayerNorm(h_dim), # 层归一化
nn.ReLU(), # ReLU激活
])
prev_dim = h_dim # 更新上一层维度
decoder_layers.append(nn.Linear(hidden_dims[0], input_dim)) # 输出层
decoder_layers.append(nn.Sigmoid()) # Sigmoid激活（输出0-1）
self.decoder = nn.Sequential(*decoder_layers) # 创建解码器
def encode(self, x):
"""
编码过程
参数:
x: 输入数据
返回:
mu: 潜在分布的均值
logvar: 潜在分布的对数方差
"""
h = self.encoder(x) # 通过编码器获取隐藏表示
mu = self.fc_mu(h) # 计算均值 μ
logvar = self.fc_logvar(h) # 计算对数方差 log σ²
return mu, logvar # 返回分布参数
def reparameterize(self, mu, logvar):
"""
重参数化采样
z = μ + σ * ε, 其中ε ~ N(0, I)
参数:
mu: 均值
logvar: 对数方差
返回:
z: 采样的潜在变量
"""
if self.training: # 训练模式下进行采样
std = torch.exp(0.5 * logvar) # 标准差
eps = torch.randn_like(std) # 标准正态噪声
return mu + eps * std # 重参数化采样
else: # 评估模式下直接返回均值
return mu # 不添加噪声
def decode(self, z):
"""
解码过程
参数:
z: 潜在变量
返回:
重建的输入
"""
return self.decoder(z) # 通过解码器重建
def forward(self, x):
"""
完整前向传播
参数:
x: 输入数据
返回:
x_recon: 重建数据
mu: 潜在分布均值
logvar: 潜在分布对数方差
z: 采样的潜在变量
"""
mu, logvar = self.encode(x) # 编码
z = self.reparameterize(mu, logvar) # 重参数化采样
x_recon = self.decode(z) # 解码重建
return x_recon, mu, logvar, z # 返回所有结果
# ============================================================
# β-VAE的损失函数
# ============================================================
def beta_vae_loss(x_recon, x, mu, logvar, beta=4.0):
"""
β-VAE损失函数
L = 重建损失 + β * KL散度
重建损失：衡量重建质量（使用二元交叉熵）
KL散度：正则化潜在空间，使其接近标准正态分布
β系数：控制解纠缠程度（β越大，解纠缠越强）
参数:
x_recon: 重建数据
x: 原始数据
mu: 潜在分布均值
logvar: 潜在分布对数方差
beta: β系数（>1促进解纠缠）
返回:
total_loss: 总损失
recon_loss: 重建损失
kl_loss: KL散度损失
"""
# 重建损失（使用二元交叉熵）
recon_loss = nn.functional.binary_cross_entropy( # 二元交叉熵
x_recon, x, # 预测值和真实值
reduction='sum' # 求和
) / x.size(0) # 除以批次大小（取平均）
# KL散度损失
# KL(q(z|x) || p(z)) = -0.5 * Σ(1 + log σ² - μ² - σ²)
# 其中 p(z) = N(0, I)
kl_loss = -0.5 * torch.sum( # 计算KL散度
1 + logvar - mu.pow(2) - logvar.exp() # KL散度公式
) / x.size(0) # 除以批次大小
# β-VAE的总损失
total_loss = recon_loss + beta * kl_loss # 加权组合
return total_loss, recon_loss, kl_loss # 返回各项损失
# ============================================================
# 训练β-VAE
# ============================================================
# 生成数据
image_size = 16 # 图像尺寸
input_dim = image_size * image_size # 输入维度（展平后的维度）
latent_dim = 5 # 潜在维度（设置为5，多于真实因素3个）
images, true_factors = generate_factor_data( # 生成数据
n_samples=5000, image_size=image_size
)
# 创建模型
model = BetaVAE( # 创建β-VAE模型
input_dim=input_dim, # 输入维度
latent_dim=latent_dim, # 潜在维度
hidden_dims=[256, 128] # 隐藏层配置
)
optimizer = optim.Adam(model.parameters(), lr=0.001) # Adam优化器
# 创建数据加载器
dataset = TensorDataset(images, true_factors) # 创建数据集
loader = DataLoader(dataset, batch_size=128, shuffle=True) # 数据加载器
# 训练
print("="*60)
print("训练β-VAE学习解纠缠表示...")
print(f"真实潜在因素: 水平位置, 垂直位置, 大小")
print(f"潜在空间维度: {latent_dim}")
print("="*60)
beta = 4.0 # β系数（大于1促进解纠缠）
epochs = 150 # 训练轮数
model.train() # 训练模式
for epoch in range(epochs): # 遍历epoch
total_loss = 0 # 总损失
total_recon = 0 # 重建损失
total_kl = 0 # KL损失
for batch_images, _ in loader: # 遍历批次（忽略真实因素）
x_recon, mu, logvar, z = model(batch_images) # 前向传播
loss, recon, kl = beta_vae_loss( # 计算β-VAE损失
x_recon, batch_images, mu, logvar, beta=beta
)
optimizer.zero_grad() # 清除梯度
loss.backward() # 反向传播
# 梯度裁剪
torch.nn.utils.clip_grad_norm_( # 裁剪梯度
model.parameters(), max_norm=1.0
)
optimizer.step() # 更新参数
total_loss += loss.item() # 累加总损失
total_recon += recon.item() # 累加重建损失
total_kl += kl.item() # 累加KL损失
if (epoch + 1) % 30 == 0: # 每30轮打印
n_batches = len(loader) # 批次数
print(f"Epoch [{epoch+1}/{epochs}], "
f"总损失: {total_loss/n_batches:.4f}, "
f"重建损失: {total_recon/n_batches:.4f}, "
f"KL损失: {total_kl/n_batches:.4f}")
# ============================================================
# 分析解纠缠程度
# ============================================================
print("\n" + "="*60)
print("分析潜在表示的解纠缠程度")
print("="*60)
model.eval() # 评估模式
with torch.no_grad(): # 关闭梯度
# 获取所有数据的潜在表示
all_mu, all_logvar = model.encode(images) # 编码所有图像
all_z = all_mu # 使用均值作为潜在表示
# 计算每个潜在维度与每个真实因素的相关性
# 高相关性说明该维度捕获了对应的真实因素
print("\n潜在维度与真实因素的Pearson相关系数:")
print("(行=潜在维度, 列=真实因素)")
print("-" * 55)
factor_names = ["水平位置", "垂直位置", "大小"] # 真实因素名称
z_numpy = all_z.numpy() # 转换为numpy
factors_numpy = true_factors.numpy() # 转换为numpy
print(f"{'':>15s}", end="") # 对齐
for fname in factor_names: # 打印表头
print(f"{fname:>12s}", end="")
print()
for d in range(latent_dim): # 遍历每个潜在维度
print(f"潜在维度 {d+1}: ", end="") # 打印行名
for f in range(3): # 遍历每个真实因素
# 计算Pearson相关系数
correlation = np.corrcoef( # 计算相关系数矩阵
z_numpy[:, d], # 潜在维度d
factors_numpy[:, f] # 真实因素f
)[0, 1] # 取相关系数
print(f"{correlation:>12.4f}", end="") # 打印相关系数
print() # 换行
# ============================================================
# 潜在空间遍历：固定其他维度，只变化一个维度
# ============================================================
print("\n" + "="*60)
print("潜在空间遍历实验")
print("（固定其他维度，只变化一个维度，观察重建结果的变化）")
print("="*60)
# 选择一个参考样本
ref_idx = 0 # 参考样本索引
ref_image = images[ref_idx:ref_idx+1] # 参考图像
ref_mu, _ = model.encode(ref_image) # 获取参考样本的潜在表示
ref_z = ref_mu.detach() # 获取潜在向量
# 对每个潜在维度进行遍历
n_steps = 7 # 遍历步数
traverse_values = torch.linspace(-2, 2, n_steps) # 遍历范围（-2到2）
for dim in range(latent_dim): # 遍历每个潜在维度
reconstructed_images = [] # 存储重建图像
for val in traverse_values: # 遍历该维度的值
z_modified = ref_z.clone() # 克隆参考潜在向量
z_modified[0, dim] = val # 修改当前维度的值
with torch.no_grad(): # 关闭梯度
recon = model.decode(z_modified) # 解码重建
# 计算重建图像中"亮像素"的位置（简单分析）
image = recon.squeeze().numpy() # 转为numpy数组
bright_pixels = np.where(image > 0.5) # 找到亮度>0.5的像素
if len(bright_pixels[0]) > 0: # 如果有亮像素
center_y = bright_pixels[0].mean() # 计算亮像素中心y
center_x = bright_pixels[1].mean() # 计算亮像素中心x
size = len(bright_pixels[0]) # 亮像素数量（代表大小）
else:
center_y, center_x, size = 0, 0, 0 # 默认值
reconstructed_images.append((center_y, center_x, size))
# 打印遍历结果
print(f"\n潜在维度 {dim+1} 的遍历:")
for i, val in enumerate(traverse_values): # 遍历每个值
cy, cx, sz = reconstructed_images[i] # 获取重建信息
print(f" z{dim+1}={val.item():>5.1f} -> "
f"位置=({cx:.1f}, {cy:.1f}), 亮像素数={sz:.0f}")
# ============================================================
# 对比实验：标准VAE (β=1) vs β-VAE (β=4)
# ============================================================
print("\n" + "="*60)
print("对比实验: 标准VAE (β=1) vs β-VAE (β=4)")
print("="*60)
for beta_val in [1.0, 4.0]: # 对比两个β值
# 创建新模型
compare_model = BetaVAE(input_dim, latent_dim, [256, 128]) # 新模型
compare_optimizer = optim.Adam(compare_model.parameters(), lr=0.001) # 优化器
compare_model.train() # 训练模式
for epoch in range(100): # 训练100轮
for batch_images, _ in loader: # 遍历批次
x_recon, mu, logvar, z = compare_model(batch_images) # 前向传播
loss, _, _ = beta_vae_loss( # 计算损失
x_recon, batch_images, mu, logvar, beta=beta_val
)
compare_optimizer.zero_grad() # 清除梯度
loss.backward() # 反向传播
compare_optimizer.step() # 更新参数
# 评估解纠缠程度
compare_model.eval() # 评估模式
with torch.no_grad(): # 关闭梯度
mu_all, _ = compare_model.encode(images) # 编码
z_all = mu_all.numpy() # 转换为numpy
# 计算每个潜在维度与最相关真实因素的最大相关性
max_correlations = [] # 存储最大相关性
for d in range(latent_dim): # 遍历每个潜在维度
corrs = [abs(np.corrcoef(z_all[:, d], factors_numpy[:, f])[0, 1])
for f in range(3)] # 计算与每个因素的相关性
max_correlations.append(max(corrs)) # 取最大相关性
avg_max_corr = np.mean(max_correlations) # 平均最大相关性
print(f"\nβ = {beta_val}:")
print(f" 各维度最大相关性: {[f'{c:.4f}' for c in max_correlations]}")
print(f" 平均最大相关性: {avg_max_corr:.4f}")
print(f" 解纠缠程度: {'好' if avg_max_corr > 0.7 else '中等' if avg_max_corr > 0.4 else '差'}")
总结表
| 知识点 | 核心思想 | 关键方法 | 应用场景 |
|---|---|---|---|
| 贪心逐层预训练 | 逐层贪心地训练每一层，用无监督方法初始化权重 | 自编码器逐层训练 + 有监督微调 | 深层网络初始化、无标签数据利用 |
| 迁移学习 | 利用源域知识帮助目标域学习 | 特征提取、微调、领域对抗训练 | 数据不足的相关任务 |
| 领域自适应 | 减小源域和目标域的分布差异 | MMD损失、对抗训练 | 跨域数据分布不同 |
| 半监督学习 | 同时利用有标签和无标签数据 | 自训练、标签传播、VAE | 标注数据昂贵的场景 |
| 分布式表示 | 多个神经元共同编码一个概念 | 词嵌入、隐藏层特征 | NLP、推荐系统 |
| 深度的指数增益 | 深度网络用更少参数表示复杂函数 | 增加深度而非宽度 | 复杂函数近似 |
| 发现潜在原因 | 学习数据背后的因果生成因素 | VAE、β-VAE、解纠缠表示 | 可解释AI、因果推理 |
更多推荐
2
0- 0
已为社区贡献49条内容
所有评论(0)