4. 基于自训练的域适应
任务说明
本案例面向语义分割的Sim2Real域适应问题：在合成数据（源域）上训练的分割模型，直接应用于真实图像（目标域）时精度显著下降。目标是通过自训练（Self-training）方法，利用模型自身对真实图像生成的伪标签进行迭代微调，从而提升模型在真实数据上的分割精度。案例使用轻量级UNet作为分割网络，在简化版合成与真实数据集上完成完整闭环。
实现方案
整体流程分为三个阶段：源域预训练 → 伪标签生成 → 目标域自训练微调。核心思想是：先用合成数据训练一个初始模型；随后用该模型对无标注的真实图像进行预测，取高置信度像素作为伪标签；最后用伪标签以监督学习方式微调模型，重复该过程以逐步提升域适应能力。为控制复杂度，本案例使用合成数据模拟器生成带噪矩形图像作为源域，使用带风格偏移的同类图像作为目标域，并仅对“矩形区域”这一类进行二分类分割。
代码实现
import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
from torch.utils.data import Dataset, DataLoader
# ---------- 1. 合成数据生成器 ----------
def generate_synthetic_images(num_samples, img_size=64, noise_level=0.1, style_shift=0.0):
"""生成带矩形前景的合成图像及对应二值掩码。
style_shift>0时模拟目标域风格偏移（亮度/对比度变化）。"""
images, masks = [], []
for _ in range(num_samples):
img = np.random.rand(img_size, img_size, 3).astype(np.float32)
mask = np.zeros((img_size, img_size), dtype=np.int64)
# 随机矩形
x0, y0 = np.random.randint(0, img_size//2, 2)
w = np.random.randint(img_size//4, img_size//2)
h = np.random.randint(img_size//4, img_size//2)
x1, y1 = min(x0+w, img_size), min(y0+h, img_size)
img[y0:y1, x0:x1] = 0.8 # 矩形区域设为亮色
mask[y0:y1, x0:x1] = 1
# 添加噪声与风格偏移
img += np.random.normal(0, noise_level, img.shape).astype(np.float32)
if style_shift > 0:
img = img * (1 + style_shift) + 0.1 * style_shift
images.append(np.clip(img, 0, 1))
masks.append(mask)
return np.stack(images), np.stack(masks)
# ---------- 2. 数据集封装 ----------
class SegDataset(Dataset):
def __init__(self, images, masks=None):
self.images = torch.FloatTensor(images).permute(0, 3, 1, 2)
self.masks = torch.LongTensor(masks) if masks is not None else None
def __len__(self):
return len(self.images)
def __getitem__(self, idx):
if self.masks is not None:
return self.images[idx], self.masks[idx]
return self.images[idx]
# ---------- 3. 轻量UNet ----------
class LightUNet(nn.Module):
def __init__(self, in_ch=3, out_ch=2):
super().__init__()
self.enc1 = nn.Sequential(nn.Conv2d(in_ch, 16, 3, padding=1), nn.ReLU(), nn.MaxPool2d(2))
self.enc2 = nn.Sequential(nn.Conv2d(16, 32, 3, padding=1), nn.ReLU(), nn.MaxPool2d(2))
self.dec = nn.Sequential(
nn.ConvTranspose2d(32, 16, 2, stride=2),
nn.ReLU(),
nn.ConvTranspose2d(16, out_ch, 2, stride=2)
)
def forward(self, x):
x = self.enc1(x)
x = self.enc2(x)
return self.dec(x)
# ---------- 4. 训练与评估函数 ----------
def train_one_epoch(model, loader, optimizer, criterion, device):
model.train()
total_loss = 0
for imgs, masks in loader:
imgs, masks = imgs.to(device), masks.to(device)
optimizer.zero_grad()
out = model(imgs)
loss = criterion(out, masks)
loss.backward()
optimizer.step()
total_loss += loss.item()
return total_loss / len(loader)
def compute_iou(model, loader, device):
"""计算mIoU（仅对前景类）。"""
model.eval()
intersection, union = 0, 0
with torch.no_grad():
for imgs, masks in loader:
imgs, masks = imgs.to(device), masks.to(device)
pred = model(imgs).argmax(dim=1)
inter = ((pred == 1) & (masks == 1)).sum().item()
un = ((pred == 1) | (masks == 1)).sum().item()
intersection += inter
union += un
return intersection / (union + 1e-6)
# ---------- 5. 主流程：自训练域适应 ----------
def main():
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
torch.manual_seed(42)
np.random.seed(42)
# 数据准备
src_imgs, src_masks = generate_synthetic_images(200, noise_level=0.05, style_shift=0.0)
tgt_imgs, tgt_masks = generate_synthetic_images(100, noise_level=0.15, style_shift=0.5) # 目标域带偏移
# 目标域仅使用图像（无标注），掩码仅用于最终评估
src_loader = DataLoader(SegDataset(src_imgs, src_masks), batch_size=16, shuffle=True)
tgt_unlabeled = DataLoader(SegDataset(tgt_imgs), batch_size=16, shuffle=False)
tgt_eval_loader = DataLoader(SegDataset(tgt_imgs, tgt_masks), batch_size=16, shuffle=False)
# 模型与优化器
model = LightUNet().to(device)
optimizer = optim.Adam(model.parameters(), lr=1e-3)
criterion = nn.CrossEntropyLoss()
# 阶段1：源域预训练
print("=== 阶段1: 源域预训练 ===")
for epoch in range(10):
loss = train_one_epoch(model, src_loader, optimizer, criterion, device)
src_iou = compute_iou(model, src_loader, device)
tgt_iou_before = compute_iou(model, tgt_eval_loader, device)
print(f"源域mIoU: {src_iou:.4f} | 目标域初始mIoU: {tgt_iou_before:.4f}")
# 阶段2-3：伪标签生成 + 自训练微调（迭代2轮）
print("=== 阶段2/3: 伪标签自训练 ===")
for iteration in range(2):
# 生成伪标签
model.eval()
pseudo_masks = []
with torch.no_grad():
for imgs in tgt_unlabeled:
imgs = imgs.to(device)
probs = torch.softmax(model(imgs), dim=1)
conf, pred = probs.max(dim=1)
# 仅保留高置信度像素（阈值0.8），其余设为忽略索引-100
pseudo = torch.where(conf > 0.8, pred, torch.full_like(pred, -100))
pseudo_masks.append(pseudo.cpu())
pseudo_masks = torch.cat(pseudo_masks, dim=0).numpy()
# 用伪标签微调
pseudo_dataset = SegDataset(tgt_imgs, pseudo_masks)
pseudo_loader = DataLoader(pseudo_dataset, batch_size=16, shuffle=True)
for epoch in range(5):
train_one_epoch(model, pseudo_loader, optimizer, criterion, device)
tgt_iou_after = compute_iou(model, tgt_eval_loader, device)
print(f"迭代{iteration+1}后目标域mIoU: {tgt_iou_after:.4f}")
print("自训练完成")
if __name__ == "__main__":
main()关键实现说明
- 伪标签置信度过滤：代码中通过
torch.where(conf > 0.8, pred, -100)实现，低于阈值的像素在CrossEntropyLoss中默认被忽略（ignore_index=-100），防止低质量伪标签误导训练。 - 迭代自训练：主循环执行2轮“生成伪标签→微调”过程，每轮微调5个epoch，在保持训练效率的同时验证精度提升趋势。
- 数据划分：源域200张带噪图像，目标域100张带风格偏移图像。目标域掩码仅用于最终评估，训练过程中模型从未接触目标域真实标注。
运行与验证
运行方式：将代码保存为 self_training_da.py，在安装有PyTorch（≥1.8）和NumPy的环境中执行 python self_training_da.py 即可。
验证方法：观察控制台输出的mIoU变化。判定标准为：目标域初始mIoU（阶段1后）应明显低于源域mIoU（通常低0.2以上），而经过2轮自训练后，目标域mIoU应逐步提升并接近源域水平。由于随机种子固定为42，每次运行结果可复现。若目标域初始mIoU与源域差距过小，可增大style_shift至0.8增强域偏移；若提升不明显，可增加自训练迭代轮数至3-4轮。
结果判定与说明
运行脚本后，控制台会依次输出源域mIoU、目标域初始mIoU以及每轮自训练后的目标域mIoU。由于代码中固定了随机种子，每次运行结果一致。典型的成功表现为：源域mIoU在0.85以上，目标域初始mIoU因风格偏移降至0.5-0.7，经过2轮自训练后回升至0.75以上。若未达到该趋势，请检查目标域风格偏移参数style_shift是否生效，或适当增加自训练轮次与微调epoch数。本案例完整实现了“合成数据预训练→伪标签生成→置信度过滤→目标域微调”的自训练闭环，可直接作为Sim2Real语义分割域适应的最小可复现基线。