Dual Co-Train框架实战：解决极端数据稀缺下的跨域超声舌体分割

在医学影像分析领域，超声舌体分割是一个关键但极具挑战性的任务，它对于语音病理学研究、发音辅助治疗以及人机交互等应用至关重要。然而，现实中的困境是：标注数据极度稀缺，且不同设备、不同采集协议下的超声图像存在显著的域差异（Domain Shift），这使得在一个数据集上训练好的模型，直接应用到另一个数据集时性能会急剧下降。近期，一种名为Dual Co-Train的框架为解决这一“极端数据稀缺下的跨数据集超声舌体分割”难题提供了新思路。本文将深入拆解这一技术的核心原理，并提供一个从理论到代码实现的完整实战指南，帮助读者理解如何利用极少量标注数据，实现模型在不同数据域间的有效迁移与泛化。

本文适合对医学图像分割、域适应（Domain Adaptation）和半监督学习感兴趣的研究者与开发者。无论你是刚入门的新手，希望了解如何处理数据稀缺问题，还是有一定经验的工程师，寻求跨域分割的工程化解决方案，都能从本文中获得清晰的路径和可运行的代码示例。

1. 背景与核心概念：为何跨数据集舌体分割如此困难？

在深入技术细节之前，我们首先要理解问题的本质。

超声舌体分割的目标是从超声图像中精确地勾勒出舌头的轮廓。超声成像因其无创、实时、低成本的优势，成为观察舌部运动的首选方式。但超声图像通常噪声大、对比度低、边界模糊（特别是舌体与周围组织的交界处），这给自动分割带来了巨大挑战。

数据稀缺性是医学AI领域的普遍痛点。获取医学影像本身成本高昂，而由专业医师进行像素级标注更是费时费力。因此，我们往往只能获得非常有限的标注数据（例如，仅几十张有标注的图像）。

域差异是跨数据集应用中的“拦路虎”。即使都是舌部超声图像，不同数据集可能来源于：

1. 不同的超声设备：探头频率、成像算法不同导致纹理和分辨率差异。
2. 不同的采集协议：探头放置位置、角度、受试者状态（如发不同元音）不同。
3. 不同的人群分布：年龄、性别、病理状况等差异会影响舌部形态。

一个在数据集A（源域）上训练得非常好的分割模型，在数据集B（目标域）上表现可能很差，因为模型学习到的是源域特有的图像特征和分布，无法泛化到目标域。传统的解决思路是域适应，但大多数域适应方法假设目标域有大量无标注数据。而在“极端数据稀缺”的设定下，目标域可能只有极少量（如1-5张）甚至没有标注图像，同时有少量无标注图像。这几乎堵死了传统监督学习和主流域适应方法的路径。

Dual Co-Train框架的核心思想，正是在这种“左右为难”的困境中，开辟一条新路。它通过双模型协同训练的机制，巧妙地利用源域丰富的标注数据、目标域极少的标注数据以及相对较多的无标注数据，让两个模型相互教学、共同进步，最终实现强大的跨域泛化能力。

2. 环境准备与版本说明

为了复现和实验Dual Co-Train框架，我们需要搭建一个标准的深度学习开发环境。以下配置是一个通用性较强的起点，你可以根据实际拥有的硬件资源进行调整。

操作系统: Ubuntu 20.04 LTS 或 Windows 10/11 (WSL2推荐) 或 macOSPython: 3.8 或 3.9 (这是多数深度学习库兼容性较好的版本)深度学习框架: PyTorch 1.9+ 或 1.12+

核心Python库:

• torch&torchvision: 模型定义与训练的核心。
• numpy,scipy: 数值计算。
• opencv-python(cv2),Pillow(PIL): 图像处理。
• scikit-learn(sklearn): 评估指标计算。
• tqdm: 训练进度条。
• tensorboard或wandb: 实验跟踪与可视化（可选但推荐）。

版本管理建议: 强烈建议使用conda或venv创建独立的虚拟环境，以避免包依赖冲突。

# 使用 conda 创建环境的示例 conda create -n dual_co_train python=3.8 conda activate dual_co_train # 安装 PyTorch (请根据你的CUDA版本访问官网获取对应命令) # 例如，对于CUDA 11.3 pip install torch==1.12.1+cu113 torchvision==0.13.1+cu113 --extra-index-url https://download.pytorch.org/whl/cu113 # 安装其他依赖 pip install numpy opencv-python pillow scikit-learn tqdm tensorboard

项目结构: 一个清晰的项目结构有助于管理代码和数据。

dual_co_train_project/ ├── data/ │ ├── source/ # 源域数据集 │ │ ├── images/ # 源域超声图像 │ │ └── masks/ # 对应的分割标注（舌体mask） │ └── target/ # 目标域数据集 │ ├── images/ # 目标域超声图像 │ ├── masks_labeled/ # 极少量有标注的mask（可选，用于验证） │ └── masks_unlabeled/ # 无标注数据（实际为空文件夹，仅占位） ├── src/ │ ├── models/ # 模型定义 │ │ ├── __init__.py │ │ ├── segmentation.py # 分割网络（如UNet, DeepLab） │ │ └── discriminator.py # 域判别器（如果用到对抗学习） │ ├── datasets.py # 自定义Dataset类，处理源域和目标域数据 │ ├── losses.py # 损失函数定义（分割损失、一致性损失等） │ ├── trainers.py # 核心训练逻辑，实现Dual Co-Train │ └── utils.py # 工具函数（指标计算、可视化等） ├── configs/ # 配置文件（YAML或JSON） │ └── default.yaml ├── scripts/ # 运行脚本 │ ├── train.py │ └── evaluate.py ├── outputs/ # 训练输出（模型、日志） │ ├── checkpoints/ │ └── logs/ └── requirements.txt

3. 核心原理拆解：Dual Co-Train 如何工作？

Dual Co-Train 不是一个单一的算法，而是一个训练范式。其核心在于维护两个结构相同但初始化不同的分割模型，让它们在训练过程中相互提供“伪标签”作为监督信号，特别是在目标域的无标注数据上。

3.1 整体训练流程

假设我们拥有：

• 源域 (Source Domain): 大量标注数据(Xs, Ys)
• 目标域 (Target Domain): 极少量标注数据(Xt_l, Yt_l)+ 一些无标注数据Xt_u

1. 初始化: 创建两个分割网络F1和F2（例如两个UNet），它们结构相同但参数随机初始化不同。
2. 监督学习: 在每个训练批次（Batch）中，F1和F2都独立地在源域标注数据(Xs, Ys)和目标域极少量标注数据(Xt_l, Yt_l)上进行有监督训练，最小化标准的分割损失（如Dice Loss + Cross-Entropy Loss）。这确保了模型具备基础的分割能力。

   # 伪代码示意 loss_supervised = DiceCE_Loss(F1(Xs), Ys) + DiceCE_Loss(F1(Xt_l), Yt_l) # 对F2同理

3. 协同训练 - 生成伪标签: 对于目标域的无标注数据Xt_u，我们用其中一个模型（如F1）的预测结果，作为另一个模型（F2）的监督信号（即“伪标签”），反之亦然。但并非所有预测都可靠。
4. 一致性筛选: 为了过滤掉噪声大的伪标签，我们引入一个一致性筛选机制。具体来说，对于同一张无标注图像x_t_u，我们通过数据增强（如旋转、缩放、颜色抖动）生成两个不同的视图v1和v2。分别输入到F1中，得到两个预测p1和p2。如果p1和p2的差异很小（例如，计算Dice系数很高），说明F1对这个样本的预测是稳定、置信度高的，那么这个预测就可以作为高质量的伪标签给F2学习。

   # 伪代码示意：为F2筛选伪标签 v1, v2 = strong_augment(x_t_u), weak_augment(x_t_u) # 两种增强 p1, p2 = F1(v1), F1(v2) # F1的预测 consistency = dice_coefficient(p1, p2) if consistency > threshold: pseudo_label_for_F2 = (p1 > 0.5).float() # 将高置信度预测二值化作为伪标签 # 将 (x_t_u, pseudo_label_for_F2) 加入F2的无监督损失计算

5. 无监督损失: 利用筛选后的高质量伪标签，计算无监督损失（如交叉熵损失），鼓励模型F2在目标域无标注数据上的预测与伪标签一致。F1也从F2那里以同样方式获取伪标签进行学习。

   loss_unsupervised_F2 = CrossEntropyLoss(F2(x_t_u), pseudo_label_for_F2)

6. 总损失与优化: 每个模型的总损失是其有监督损失和无监督损失的加权和。通过反向传播和优化器（如Adam）同时更新两个模型的参数。

   total_loss_F1 = loss_supervised_F1 + lambda_u * loss_unsupervised_F1 total_loss_F2 = loss_supervised_F2 + lambda_u * loss_unsupervised_F2 # lambda_u 是无监督损失的权重，随时间增长（课程学习策略）

7. 迭代: 重复步骤2-6，两个模型在源域监督信号和彼此提供的目标域伪标签信号下共同进化，逐渐适应目标域的数据分布。

3.2 为何有效？—— 视角差异与误差纠正

Dual Co-Train 有效的关键在于两个模型的视角差异。由于初始化不同，F1和F2学习到的特征表示和决策边界会略有不同。这种差异使得：

• 当一个模型对某个样本预测错误时，另一个模型可能预测正确。
• 通过一致性筛选，我们只选取两个模型各自“内部一致”（即对增强视图预测稳定）的预测作为伪标签。这大概率是正确或接近正确的预测。
• 模型之间相互提供高质量的、多样化的伪标签，相当于为目标域引入了额外的、可靠的监督信号，有效缓解了目标域标注稀缺的问题。
• 这个过程也是一种高效的数据增强，因为模型是在学习如何对经过扰动的数据做出稳定预测，提升了泛化能力。

4. 完整实战案例：实现一个简化的 Dual Co-Train

下面我们将用PyTorch实现一个简化版的Dual Co-Train框架，用于演示核心流程。我们假设使用一个公开的超声模拟数据集和一个简单的UNet作为分割网络。

4.1 数据准备与Dataset类

首先，我们需要一个能同时加载源域和目标域数据的Dataset。

# file: src/datasets.py import os from PIL import Image import torch from torch.utils.data import Dataset import torchvision.transforms as T import numpy as np class DualDomainDataset(Dataset): """ 同时加载源域和目标域数据的Dataset。 假设图像为灰度图，mask为二值图。 """ def __init__(self, source_img_dir, source_mask_dir, target_img_dir, target_mask_dir=None, # target_mask_dir可能为空或只有少量标注 is_train=True, target_has_label=False): self.source_img_paths = sorted([os.path.join(source_img_dir, f) for f in os.listdir(source_img_dir) if f.endswith('.png')]) self.source_mask_paths = sorted([os.path.join(source_mask_dir, f) for f in os.listdir(source_mask_dir) if f.endswith('.png')]) self.target_img_paths = sorted([os.path.join(target_img_dir, f) for f in os.listdir(target_img_dir) if f.endswith('.png')]) self.target_has_label = target_has_label if target_has_label and target_mask_dir: self.target_mask_paths = sorted([os.path.join(target_mask_dir, f) for f in os.listdir(target_mask_dir) if f.endswith('.png')]) else: self.target_mask_paths = None self.is_train = is_train # 基础转换：转为Tensor并归一化 self.img_transform = T.Compose([ T.Grayscale(num_output_channels=1), # 确保是单通道 T.ToTensor(), T.Normalize(mean=[0.5], std=[0.5]) # 归一化到[-1,1] ]) self.mask_transform = T.Compose([ T.Grayscale(num_output_channels=1), T.ToTensor(), ]) # 用于无监督数据增强的强增强和弱增强 self.strong_aug = T.Compose([ T.RandomHorizontalFlip(p=0.5), T.RandomRotation(degrees=10), T.ColorJitter(brightness=0.2, contrast=0.2), T.RandomAffine(degrees=0, translate=(0.1, 0.1)), ]) self.weak_aug = T.Compose([ T.RandomHorizontalFlip(p=0.5), ]) def __len__(self): # 返回源域和目标域中较大的长度，便于采样 return max(len(self.source_img_paths), len(self.target_img_paths)) def __getitem__(self, idx): # 获取源域数据 s_idx = idx % len(self.source_img_paths) s_img = Image.open(self.source_img_paths[s_idx]) s_mask = Image.open(self.source_mask_paths[s_idx]) s_img_t = self.img_transform(s_img) s_mask_t = self.mask_transform(s_mask) # 获取目标域数据 t_idx = idx % len(self.target_img_paths) t_img = Image.open(self.target_img_paths[t_idx]) t_img_t = self.img_transform(t_img) item = { 'source_img': s_img_t, 'source_mask': s_mask_t, 'target_img': t_img_t, 'target_has_label': self.target_has_label, } # 如果目标域有标注（极少量情况），则加载 if self.target_has_label and self.target_mask_paths is not None: t_mask = Image.open(self.target_mask_paths[t_idx]) t_mask_t = self.mask_transform(t_mask) item['target_mask'] = t_mask_t # 如果是训练阶段，为目标域图像生成增强视图，用于一致性计算 if self.is_train: t_img_pil = Image.open(self.target_img_paths[t_idx]).convert('L') # 注意：增强是在PIL Image上进行的，然后再转换 t_img_strong = self.strong_aug(t_img_pil) t_img_weak = self.weak_aug(t_img_pil) item['target_img_strong'] = self.img_transform(t_img_strong) item['target_img_weak'] = self.img_transform(t_img_weak) return item

4.2 模型定义：分割网络

我们使用一个轻量化的UNet。

# file: src/models/segmentation.py import torch import torch.nn as nn import torch.nn.functional as F class DoubleConv(nn.Module): """(卷积 => BN => ReLU) * 2""" def __init__(self, in_channels, out_channels): super().__init__() self.double_conv = nn.Sequential( nn.Conv2d(in_channels, out_channels, kernel_size=3, padding=1), nn.BatchNorm2d(out_channels), nn.ReLU(inplace=True), nn.Conv2d(out_channels, out_channels, kernel_size=3, padding=1), nn.BatchNorm2d(out_channels), nn.ReLU(inplace=True) ) def forward(self, x): return self.double_conv(x) class UNet(nn.Module): def __init__(self, n_channels=1, n_classes=1): super(UNet, self).__init__() self.n_channels = n_channels self.n_classes = n_classes self.inc = DoubleConv(n_channels, 64) self.down1 = nn.Sequential(nn.MaxPool2d(2), DoubleConv(64, 128)) self.down2 = nn.Sequential(nn.MaxPool2d(2), DoubleConv(128, 256)) self.down3 = nn.Sequential(nn.MaxPool2d(2), DoubleConv(256, 512)) self.down4 = nn.Sequential(nn.MaxPool2d(2), DoubleConv(512, 1024)) self.up1 = nn.ConvTranspose2d(1024, 512, kernel_size=2, stride=2) self.conv1 = DoubleConv(1024, 512) # 1024 = 512(up1) + 512(skip) self.up2 = nn.ConvTranspose2d(512, 256, kernel_size=2, stride=2) self.conv2 = DoubleConv(512, 256) self.up3 = nn.ConvTranspose2d(256, 128, kernel_size=2, stride=2) self.conv3 = DoubleConv(256, 128) self.up4 = nn.ConvTranspose2d(128, 64, kernel_size=2, stride=2) self.conv4 = DoubleConv(128, 64) self.outc = nn.Conv2d(64, n_classes, kernel_size=1) def forward(self, x): x1 = self.inc(x) x2 = self.down1(x1) x3 = self.down2(x2) x4 = self.down3(x3) x5 = self.down4(x4) x = self.up1(x5) # 拼接跳跃连接，需要确保尺寸匹配，这里假设尺寸是2的倍数 x = torch.cat([x, x4], dim=1) x = self.conv1(x) x = self.up2(x) x = torch.cat([x, x3], dim=1) x = self.conv2(x) x = self.up3(x) x = torch.cat([x, x2], dim=1) x = self.conv3(x) x = self.up4(x) x = torch.cat([x, x1], dim=1) x = self.conv4(x) logits = self.outc(x) return logits # 输出logits，在损失函数中处理sigmoid

4.3 损失函数定义

我们需要有监督的Dice损失和用于无监督训练的伪标签交叉熵损失。

# file: src/losses.py import torch import torch.nn as nn import torch.nn.functional as F class DiceLoss(nn.Module): def __init__(self, smooth=1e-6): super(DiceLoss, self).__init__() self.smooth = smooth def forward(self, logits, targets): # logits: [B, 1, H, W], targets: [B, 1, H, W] probs = torch.sigmoid(logits) num = 2. * (probs * targets).sum(dim=(2,3)) den = probs.sum(dim=(2,3)) + targets.sum(dim=(2,3)) dice = (num + self.smooth) / (den + self.smooth) return 1 - dice.mean() class DiceBCELoss(nn.Module): """常用的分割损失，Dice Loss + BCE Loss""" def __init__(self, smooth=1e-6, bce_weight=0.5): super(DiceBCELoss, self).__init__() self.dice = DiceLoss(smooth) self.bce_weight = bce_weight def forward(self, logits, targets): dice_loss = self.dice(logits, targets) bce_loss = F.binary_cross_entropy_with_logits(logits, targets) return dice_loss + self.bce_weight * bce_loss def consistency_loss(pred1, pred2, threshold=0.9): """ 计算两个预测之间的一致性。 用于筛选伪标签：如果一致性高，则认为预测可靠。 """ # pred1, pred2: [B, 1, H, W] 经过sigmoid的概率 dice = 2 * (pred1 * pred2).sum(dim=(2,3)) / (pred1.sum(dim=(2,3)) + pred2.sum(dim=(2,3)) + 1e-6) # 返回平均Dice系数和一致性掩码（哪些样本是可靠的） reliable_mask = (dice > threshold).float() return dice.mean(), reliable_mask

4.4 核心训练器：Dual Co-Train 逻辑

这是整个框架的核心，实现了两个模型协同训练的循环。

# file: src/trainers.py import torch import torch.nn as nn from tqdm import tqdm import numpy as np class DualCoTrainer: def __init__(self, model1, model2, optimizer1, optimizer2, device, supervised_loss_fn, lambda_u=0.1, consistency_threshold=0.9): self.model1 = model1.to(device) self.model2 = model2.to(device) self.optimizer1 = optimizer1 self.optimizer2 = optimizer2 self.device = device self.supervised_loss_fn = supervised_loss_fn self.lambda_u = lambda_u # 无监督损失权重 self.consistency_threshold = consistency_threshold def train_epoch(self, dataloader, epoch): self.model1.train() self.model2.train() total_loss1, total_loss2 = 0, 0 pbar = tqdm(dataloader, desc=f'Epoch {epoch}') for batch in pbar: # 将数据移动到设备 s_img = batch['source_img'].to(self.device) s_mask = batch['source_mask'].to(self.device) t_img = batch['target_img'].to(self.device) t_img_s = batch['target_img_strong'].to(self.device) t_img_w = batch['target_img_weak'].to(self.device) has_target_label = batch['target_has_label'][0] # 假设batch内一致 batch_size = s_img.size(0) # ============ 有监督损失 ============ # 模型1在源域和目标域（如果有标签）的监督损失 pred_s1 = self.model1(s_img) loss_sup1 = self.supervised_loss_fn(pred_s1, s_mask) # 模型2的监督损失 pred_s2 = self.model2(s_img) loss_sup2 = self.supervised_loss_fn(pred_s2, s_mask) # 如果目标域有极少量标注，也加入监督损失 if has_target_label: t_mask = batch['target_mask'].to(self.device) pred_t1 = self.model1(t_img) pred_t2 = self.model2(t_img) loss_sup1 += self.supervised_loss_fn(pred_t1, t_mask) loss_sup2 += self.supervised_loss_fn(pred_t2, t_mask) # ============ 无监督协同训练 ============ # 步骤1: 为模型2生成伪标签（使用模型1） with torch.no_grad(): # 模型1对强增强和弱增强视图的预测 pred1_strong = torch.sigmoid(self.model1(t_img_s)) pred1_weak = torch.sigmoid(self.model1(t_img_w)) # 计算一致性 dice_consistency, reliable_mask = consistency_loss( pred1_strong, pred1_weak, self.consistency_threshold ) # 生成伪标签：使用强增强预测的二值化结果 pseudo_label_for_m2 = (pred1_strong > 0.5).float() # 只保留高一致性样本的伪标签 reliable_mask = reliable_mask.view(-1, 1, 1, 1) # 扩展维度用于mask pseudo_label_for_m2 = pseudo_label_for_m2 * reliable_mask # 步骤2: 计算模型2在目标域的无监督损失（仅对可靠样本） if reliable_mask.sum() > 0: # 如果有可靠样本 pred_t2_u = self.model2(t_img_s) # 模型2对强增强视图的预测 # 只计算可靠样本的损失 loss_unsup2 = F.binary_cross_entropy_with_logits( pred_t2_u, pseudo_label_for_m2, reduction='none' ) loss_unsup2 = (loss_unsup2 * reliable_mask).sum() / (reliable_mask.sum() + 1e-6) else: loss_unsup2 = 0.0 # 步骤3: 为模型1生成伪标签（使用模型2） - 同理 with torch.no_grad(): pred2_strong = torch.sigmoid(self.model2(t_img_s)) pred2_weak = torch.sigmoid(self.model2(t_img_w)) dice_consistency2, reliable_mask2 = consistency_loss( pred2_strong, pred2_weak, self.consistency_threshold ) pseudo_label_for_m1 = (pred2_strong > 0.5).float() reliable_mask2 = reliable_mask2.view(-1, 1, 1, 1) pseudo_label_for_m1 = pseudo_label_for_m1 * reliable_mask2 if reliable_mask2.sum() > 0: pred_t1_u = self.model1(t_img_s) loss_unsup1 = F.binary_cross_entropy_with_logits( pred_t1_u, pseudo_label_for_m1, reduction='none' ) loss_unsup1 = (loss_unsup1 * reliable_mask2).sum() / (reliable_mask2.sum() + 1e-6) else: loss_unsup1 = 0.0 # ============ 总损失与反向传播 ============ # 总损失 = 有监督损失 + λ * 无监督损失 total_loss1 = loss_sup1 + self.lambda_u * loss_unsup1 total_loss2 = loss_sup2 + self.lambda_u * loss_unsup2 # 分别更新两个模型 self.optimizer1.zero_grad() total_loss1.backward() self.optimizer1.step() self.optimizer2.zero_grad() total_loss2.backward() self.optimizer2.step() # 记录损失 total_loss1_item = total_loss1.item() total_loss2_item = total_loss2.item() total_loss1 += total_loss1_item total_loss2 += total_loss2_item pbar.set_postfix({ 'Loss1': f'{total_loss1_item:.4f}', 'Loss2': f'{total_loss2_item:.4f}', 'Reliable%': f'{(reliable_mask.sum()/(batch_size + 1e-6)*100):.1f}%' }) avg_loss1 = total_loss1 / len(dataloader) avg_loss2 = total_loss2 / len(dataloader) return avg_loss1, avg_loss2 def save_models(self, path1, path2): torch.save(self.model1.state_dict(), path1) torch.save(self.model2.state_dict(), path2)

4.5 主训练脚本

将以上模块组合起来，形成完整的训练流程。

# file: scripts/train.py import sys import os sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import torch from torch.utils.data import DataLoader from src.datasets import DualDomainDataset from src.models.segmentation import UNet from src.losses import DiceBCELoss from src.trainers import DualCoTrainer import argparse def main(): parser = argparse.ArgumentParser() parser.add_argument('--source_img_dir', type=str, required=True) parser.add_argument('--source_mask_dir', type=str, required=True) parser.add_argument('--target_img_dir', type=str, required=True) parser.add_argument('--target_mask_dir', type=str, default=None) parser.add_argument('--epochs', type=int, default=100) parser.add_argument('--batch_size', type=int, default=4) parser.add_argument('--lr', type=float, default=1e-4) parser.add_argument('--lambda_u', type=float, default=0.1) parser.add_argument('--device', type=str, default='cuda' if torch.cuda.is_available() else 'cpu') args = parser.parse_args() # 1. 准备数据 target_has_label = args.target_mask_dir is not None train_dataset = DualDomainDataset( source_img_dir=args.source_img_dir, source_mask_dir=args.source_mask_dir, target_img_dir=args.target_img_dir, target_mask_dir=args.target_mask_dir, is_train=True, target_has_label=target_has_label ) train_loader = DataLoader(train_dataset, batch_size=args.batch_size, shuffle=True, num_workers=2) # 2. 初始化两个模型和优化器 model1 = UNet(n_channels=1, n_classes=1) model2 = UNet(n_channels=1, n_classes=1) optimizer1 = torch.optim.Adam(model1.parameters(), lr=args.lr) optimizer2 = torch.optim.Adam(model2.parameters(), lr=args.lr) # 3. 损失函数和训练器 supervised_loss = DiceBCELoss() trainer = DualCoTrainer( model1=model1, model2=model2, optimizer1=optimizer1, optimizer2=optimizer2, device=args.device, supervised_loss_fn=supervised_loss, lambda_u=args.lambda_u ) # 4. 训练循环 for epoch in range(1, args.epochs + 1): avg_loss1, avg_loss2 = trainer.train_epoch(train_loader, epoch) print(f'Epoch {epoch} finished. Avg Loss1: {avg_loss1:.4f}, Avg Loss2: {avg_loss2:.4f}') # 每隔一定epoch保存模型 if epoch % 20 == 0: os.makedirs('outputs/checkpoints', exist_ok=True) trainer.save_models( f'outputs/checkpoints/model1_epoch{epoch}.pth', f'outputs/checkpoints/model2_epoch{epoch}.pth' ) print("Training completed.") if __name__ == '__main__': main()

4.6 运行与验证

假设你的数据已按项目结构放置，可以运行以下命令开始训练：

python scripts/train.py \ --source_img_dir ./data/source/images \ --source_mask_dir ./data/source/masks \ --target_img_dir ./data/target/images \ --target_mask_dir ./data/target/masks_labeled \ # 如果目标域有少量标签 --epochs 100 \ --batch_size 8 \ --lr 1e-4 \ --lambda_u 0.1

结果说明: 训练过程中，你会看到两个模型的损失在下降，同时“Reliable%”（可靠伪标签的百分比）会逐渐上升，这表明两个模型对目标域数据的预测越来越稳定、一致。训练结束后，你可以使用训练好的模型（例如取两个模型的预测平均值）在目标域的测试集上进行评估，通常会比直接在源域训练或简单微调（Fine-tuning）有显著的性能提升。

5. 常见问题与排查思路

在实际实现和训练Dual Co-Train框架时，你可能会遇到以下典型问题：

6. 最佳实践与工程建议

将Dual Co-Train从实验代码应用到实际项目或研究中，需要注意以下工程细节：

1. 数据预处理与标准化：

   • 统一图像尺寸：将源域和目标域图像缩放到相同分辨率。超声图像通常较小（如 640x480），保持原始宽高比进行中心裁剪或填充。
   • 域特定的标准化：不要对两域数据使用相同的均值和标准差进行归一化。应分别计算源域和目标域训练集的均值和标准差，并在各自数据上应用。这有助于模型更好地适应各自的强度分布。

   # 分别计算统计量 source_mean, source_std = compute_mean_std(source_image_list) target_mean, target_std = compute_mean_std(target_image_list) # 在Dataset中应用不同的归一化

2. 模型架构选择：

   • 骨干网络：UNet是医学分割的经典选择，但对于更复杂的域差异，可以考虑使用带有预训练编码器（如ResNet, EfficientNet）的UNet变体（如UNet++， DeepLabv3+），以利用在大型自然图像数据集上学到的通用特征。
   • 共享与独立参数：一种进阶策略是让两个模型共享编码器（特征提取器），但使用独立的解码器。这可以减少参数量，同时保留一定的视角差异。

3. 伪标签质量优化：

   • 置信度校准：除了基于一致性的筛选，还可以结合预测的置信度（如最大softmax概率或熵）。只选择高一致性且高置信度的预测作为伪标签。
   • 时间集成：不使用当前模型的瞬时预测作为伪标签，而是使用其过去一段时间内预测的指数移动平均作为更稳定的伪标签源。
   • 锐化伪标签：对于分割任务，可以对伪标签概率图进行锐化操作（如温度缩放），使其更接近0或1，提供更明确的监督信号。

4. 损失函数设计：

   • 自适应权重：无监督损失权重lambda_u不应是固定的。可以采用课程学习策略，随着训练进行，逐渐增加lambda_u，让模型先打好有监督基础，再逐步依赖伪标签。
   • 对抗性损失：在特征层面引入域判别器，通过对抗训练让特征提取器学习域不变的特征表示，可以作为有监督和无监督损失之外的补充。

5. 训练策略与超参数调优：

   • 学习率调度：使用余弦退火或带热重启的余弦退火（CosineAnnealingWarmRestarts）学习率调度器，有助于模型跳出局部最优。
   • 早停机制：在目标域的一个极小验证集（如果有的话）上监控性能，当性能不再提升时提前停止训练，防止过拟合。
   • 模型集成：训练结束后，不要只使用其中一个模型。将两个模型的预测结果进行平均（或加权平均）作为最终输出，通常能获得更稳定、更准确的结果。

6. 实验记录与可复现性：

   • 配置管理：使用YAML或JSON文件记录所有超参数（学习率、批大小、增强参数、损失权重等），确保实验可复现。
   • 版本控制：对代码、配置和数据集划分使用Git进行版本控制。
   • 实验跟踪：使用TensorBoard或Weights & Biases (WandB) 记录训练损失、验证指标、预测可视化图等，方便分析和比较不同实验设置的效果。

通过系统地应用这些最佳实践，你可以显著提升Dual Co-Train框架在实际跨域超声舌体分割任务中的鲁棒性和性能，使其从一个研究概念转化为一个可靠的工程解决方案。记住，处理极端数据稀缺问题的核心思想是最大化利用有限信息和引导模型进行自我改进，Dual Co-Train正是这一思想的优雅实现。