简介基于ExpertNet与ResNet50构建的多任务学习网络无监督自适应模型源码面向医疗图像分析与深度学习开发者旨在解决标注数据稀缺场景下的多疾病特征识别与建模问题。项目通过ResNet50的残差模块提取深层图像特征残差连接有效缓解深层网络梯度消失借助ExpertNet的专家分支为不同识别任务分配专属子网络并采用无监督学习策略降低对人工标注的依赖可应用于肺部CT肿瘤检测、眼底病变诊断等场景。压缩包共16个文件以Python脚本为主10个py涵盖数据预处理、模型构建、训练与评估等完整流程另含requirements.txt与README说明整体仅27KB结构清晰。源码中提供自动编码器训练、模型生成与测试等工具脚本便于理解多任务学习与无监督自适应的工程实现。已有254人学习浏览适合作为医疗图像多任务学习方向的项目参考与二次开发基础。1. 给无标注医疗图像做自适应ExpertNet Resnet50能顶什么用一套在公开胸部X光集上训练好的多任务模型换到本地医院PACS导出的设备数据上AUC掉一截而目标域又往往拿不到标注。标题里这套基于ExpertNet Resnet50的多任务学习网络无监督自适应模型就是用来解决这种“源域很准、目标域拉胯”的现实问题的。ExpertNet当门控专家Resnet50当共享骨干多任务同时学疾病分类与掩码回归无监督自适应阶段不依赖目标域标签只靠统计量对齐和特征分布拉近把模型拉回正轨。这套方案适合手里已有预训练模型、但本地影像没有标注的工程师也适合做跨设备迁移的医疗AI研究者。下面按落地顺序展开先拆架构再讲数据然后给训练代码最后把常见踩坑点集中说透。2. ExpertNet Resnet50的架构拆解门控专家与共享backbone的取舍2.1 门控专家网络不是简单加多个head多任务学习最省钱的做法是共享Resnet50后接多个全连接head一个head分类、一个head回归各自算loss。但医疗图像里的任务差异往往比想象中大疾病分类要的是全局语义器官分割要的是空间细节两点同时压在同一个特征向量上梯度方向会互相拉扯。更麻烦的是不同影像科室的数据分布差异很大一套weight去拟合所有任务和所有域经常是哪头都顾不好。我一般会在这类场景里给backbone后面接一个门控专家模块也就是标题里的ExpertNet。它的核心思路是准备若干个小专家网络各自学到不一样的“偏好”特征再由一个门控网络根据当前输入动态地给每个任务分配专家权重。训练时每个任务用的不是同一份特征而是专家特征的加权组合等于给任务之间装了缓冲层。import torch import torch.nn as nn class ExpertModule(nn.Module): def __init__(self, in_dim2048, num_experts4, num_tasks2, hidden_dim512): super().__init__() self.num_experts num_experts self.num_tasks num_tasks # 每个专家是一个小 MLP负责提取一种任务/域偏好的特征 self.experts nn.ModuleList([ nn.Sequential( nn.Linear(in_dim, hidden_dim), nn.ReLU(inplaceTrue), nn.Linear(hidden_dim, in_dim // 2) ) for _ in range(num_experts) ]) # 门控网络对每个任务输出一组专家权重 self.gate nn.Sequential( nn.Linear(in_dim, hidden_dim), nn.ReLU(inplaceTrue), nn.Linear(hidden_dim, num_experts * num_tasks) ) def forward(self, x): # x: [B, in_dim]Resnet50 backbone 输出的全局特征 gate_logits self.gate(x) # [B, num_experts * num_tasks] gate_logits gate_logits.view(-1, self.num_tasks, self.num_experts) gate_weight torch.softmax(gate_logits, dim-1) # 每个任务自己的专家权重 expert_outs torch.stack([e(x) for e in self.experts], dim1) # [B, E, out_dim] task_embedding torch.einsum(bte,btd-bed, gate_weight, expert_outs) return task_embedding, gate_weight参数上num_experts我默认给4落地上26都可以超过6后训练时间涨得明显但收益递减。in_dim必须跟backbone输出维度对齐用Resnet50的avgpool特征就是2048。task_embedding的维度是[B, num_tasks, 1024]后面接任务head时按任务索引取即可。einsum这行如果看得别扭等价操作是先expert_outs.transpose(0, 1)再用torch.bmm效果一样。门控网络的参数初始化不用特别处理但训练时有个容易翻车的细节门控必须加一点熵正则或者专家dropout否则训练后期权重容易坍缩到某一个专家上这个问题放到第5章展开。2.2 Resnet50在无监督自适应里的正确姿势Resnet50网络结构示意图大家都很熟作为backbone时只需要砍掉尾部。具体来说torchvision的resnet50可以拆成features和fc两部分我们保留到layer4输出前面接一个AdaptiveAvgPool给分类和全局特征layer4的特征图则留给分割类任务。无监督自适应阶段backbone的BatchNorm统计量是个隐形坑。源域训练完成后BN里的running_mean和running_var记录的是源域图像统计量目标域图像一进来如果这些统计量不更新特征分布会被错误拉伸导致后续MMD对齐和门控都跟着偏。所以我会把前两三个stage冻结但让BN层保持track_running_statsTrue并继续更新或者干脆在目标域上用一小批无标签数据预热几步。import torch import torch.nn as nn import torchvision.models as models def build_backbone(pretrainedTrue, freeze_stages2): resnet50 models.resnet50(pretrainedpretrained) # 去掉 resnet50 末尾的 avgpool 与 fc保留到 layer4 features nn.Sequential(*list(resnet50.children())[:-2]) pool nn.AdaptiveAvgPool2d((1, 1)) if freeze_stages 0: # 冻结前几个 stage通常是 conv1/bn1/layer1/layer2 blocks [features[i] for i in range(len(features))] for block in blocks[:freeze_stages * 2 3]: # 保守冻结到 layer1 附近 for p in block.parameters(): p.requires_grad False return features, pool这段代码的冻结范围写得比较保守freeze_stages * 2 3是为了把conv1、bn1、relu、maxpool、layer1都覆盖到。如果数据量很小、目标域跟源域差异又极大我会少冻一层让layer2也参与自适应如果目标域只有几百张无标注图建议只冻结conv1和layer1否则backbone很快被目标域噪声带偏。这里还有个常见的取舍ImageNet预训练权重在医疗图像上到底有没有用我的经验是有用Resnet50的前几层学到的是边缘、纹理这种通用结构对X光、CT同样有效但最后几层语义偏离医疗特征所以后两三层要放开训练不要整条backbone锁死。3. 医疗图像数据管线DICOM转张量与无监督数据划分3.1 DICOM读取、窗宽窗位与归一化医疗图像落地第一步往往是处理DICOM。直接拿pixel_array喂给Resnet50是必翻车的因为DICOM像素值不是常规0255灰度而是带RescaleSlope和RescaleIntercept的原始值有的是CT的HU值有的是DR的线性响应。不先做窗宽窗位调整模型会一直去学习无意义的极值和噪声。import numpy as np import cv2 import pydicom def load_dicom_to_rgb(path, target_size(224, 224)): ds pydicom.dcmread(path) img ds.pixel_array.astype(np.float32) # DICOM 里偶尔会混入 nan/inf先清理不然后面 loss 直接翻车 img np.nan_to_num(img, nan0.0, posinf0.0, neginf0.0) slope float(getattr(ds, RescaleSlope, 1.0)) intercept float(getattr(ds, RescaleIntercept, 0.0)) img img * slope intercept # 只要是 CT就用软组织窗 [-500, 1000]DR/CR 则直接线性拉伸 lower, upper -500.0, 1000.0 img np.clip(img, lower, upper) img (img - lower) / (upper - lower) img (img * 255).astype(np.uint8) img cv2.resize(img, target_size, interpolationcv2.INTER_CUBIC) rgb cv2.cvtColor(img, cv2.COLOR_GRAY2RGB) # 复制三通道对齐 ImageNet 输入 return rgb窗宽窗位这里写的是软组织窗如果项目是肺结节或者胸部DR窗口应该改成[0, 255]或[-300, 1200]。更好做法是在DICOM tag里读取WindowCenter和WindowWidth但很多设备写入不规范兜底方案就是硬编码一个合理范围。你把这份代码直接接上torchvision.transforms.Normalize(mean[0.485,0.456,0.406], std[0.229,0.224,0.225])整条图像管线就通了。目标域和源域要使用完全相同的窗位处理否则自适应等于在一个不对齐的输入空间上操作。3.2 有标签源域与无标签目标域的Dataset设计无监督自适应的训练循环需要同时从两个数据流取数据源域有标签目标域只有图像。如果只做一个Dataset看起来省事但训练时容易把目标域标签当成零张量误传播严重污染loss。我一般分两个Dataset源域返回(image, label, mask)目标域只返回(image,)。from torch.utils.data import Dataset class SourceDataset(Dataset): def __init__(self, file_list, transformNone): self.file_list file_list self.transform transform def __len__(self): return len(self.file_list) def __getitem__(self, idx): path, label, mask self.file_list[idx] image load_dicom_to_rgb(path) if self.transform: image self.transform(image) return image, label, mask class TargetDataset(Dataset): def __init__(self, file_list, transformNone): self.file_list file_list self.transform transform def __len__(self): return len(self.file_list) def __getitem__(self, idx): path self.file_list[idx] image load_dicom_to_rgb(path) if self.transform: image self.transform(image) return image,注意TargetDataset最后返回一个只有一个元素的tuple目的是防止dataloader自动把image本身当作可迭代目标。实际业务里目标域文件列表可能有一万多张全量参与MMD计算会很慢我建议训练时每个epoch随机抽一个子集比如10003000张让MMD估计既稳定又不拖慢训练。数据划分上最容易被忽略的是病人级别的数据泄露。同一个病人的多次检查既出现在源域又出现在目标域会让自适应效果虚高。我会按patient_id做group split保证训练集、验证集里不出现同一个人的不同影像。4. 搭建多任务学习网络与无监督自适应训练脚本4.1 MultiTaskExpertNet的完整前向结构把前面两个模块拼起来就得到项目的主体网络。这里用分类加分割两个任务做演示分类走全局特征和门控专家分割直接吃layer4特征图。分割不经过专家模块听起来不够“多任务”但落地时我会把专家输出的task embedding用广播方式拼接到分割特征图的每个像素位置作为全局上下文注入这样两个任务都利用了门控专家。import torch import torch.nn as nn class MultiTaskExpertNet(nn.Module): def __init__(self, num_classes14, num_experts4): super().__init__() self.features, self.pool build_backbone(pretrainedTrue) self.expert ExpertModule(in_dim2048, num_expertsnum_experts, num_tasks2) self.cls_head nn.Linear(1024, num_classes) self.seg_head nn.Conv2d(2048, 1, kernel_size1) self.gate_entropy None # 训练时记录便于观察门控是否坍缩 def forward(self, x): ftr self.features(x) # [B, 2048, H, W] pooled self.pool(ftr).flatten(1) # [B, 2048] task_embed, gate_weight self.expert(pooled) # [B, 2, 1024] # 分类任务用门控专家的第 0 个输出 cls_feat task_embed[:, 0, :] cls_logits self.cls_head(cls_feat) # 分割任务用 layer4 特征图 全局专家特征 seg_feat ftr global_ctx task_embed[:, 1, :].unsqueeze(-1).unsqueeze(-1) seg_feat seg_feat global_ctx.expand_as(seg_feat) seg_logit self.seg_head(seg_feat) self.gate_entropy -( gate_weight * torch.log(gate_weight 1e-8) ).sum(dim-1).mean() return {cls: cls_logits, seg: seg_logit, gate: gate_weight}这个global_ctx.expand_as(seg_feat)是把1024维向量广播到每个空间位置再逐像素相加相当于给每个像素都注入了全图的任务偏好。实现上会占用一点显存H、W在Resnet50 layer4处通常是7×7或者14×14问题不大。如果输入分辨率是512×512H、W变成16×16显存开销会涨一些建议实验室里先用224验证流程。损失设计上分类用BCEWithLogitsLoss因为医疗多标签任务里一张片子可能同时有多个病灶。分割用DiceLoss加上一点L1正则防止mask输出全黑或全白。这两个loss直接相加往往不行因为BCE的值通常在0.10.5量级而DiceLoss在0.1以下二者数量级不同后面训练循环里我会用不确定性加权或固定系数处理。4.2 训练循环源域监督 目标域MMD对齐 门控正则无监督自适应的常见路线有对抗域判别器、最大均值差异MMD、伪标签自训练。我在这套方案里优先用MMD因为实现稳定、不需要额外的判别器网络去平衡GAN训练。核心思想是拉近源域和目标域在backbone特征空间上的中心距离让模型在目标域上不靠标签也能把特征分布对齐。import torch def rbf_mmd(x_s, x_t, bandwidths[1.0, 5.0, 10.0]): # 输入都是 [N, D]建议每次取相同 batch_size x_s x_s.detach() if x_s.size(0) 1 else x_s x_t x_t.detach() if x_t.size(0) 1 else x_t loss 0.0 ns, nt x_s.size(0), x_t.size(0) for bw in bandwidths: k_ss torch.cdist(x_s, x_s) ** 2 / (2 * bw ** 2) k_tt torch.cdist(x_t, x_t) ** 2 / (2 * bw ** 2) k_st torch.cdist(x_s, x_t) ** 2 / (2 * bw ** 2) loss (-k_ss).exp().mean() loss (-k_tt).exp().mean() loss -2 * (-k_st).exp().mean() return loss / len(bandwidths) def dice_loss(pred, target, smooth1.0): pred torch.sigmoid(pred) inter (pred * target).sum(dim(2, 3)) union pred.sum(dim(2, 3)) target.sum(dim(2, 3)) return 1 - (2 * inter smooth) / (union smooth)MMD用多带宽是为了在特征尺度不确定时更稳实际跑下来bandwidths[1.0, 5.0]也够。注意cdist在特征维度大时计算快但如果batch里有一张图尺寸异常导致特征维度不一致会直接报维度错。所以数据管线里必须保证所有图像经过resize到相同分辨率。主训练循环需要同时从源域和目标域加载数据from torch.utils.data import DataLoader model MultiTaskExpertNet(num_classes14) optimizer torch.optim.AdamW(model.parameters(), lr1e-4, weight_decay1e-4) bce torch.nn.BCEWithLogitsLoss() src_loader DataLoader(SourceDataset(src_files), batch_size16, shuffleTrue) tgt_loader DataLoader(TargetDataset(tgt_files), batch_size16, shuffleTrue) src_iter iter(src_loader) tgt_iter iter(tgt_loader) for step in range(2000): try: src_batch next(src_iter) except StopIteration: src_iter iter(src_loader) src_batch next(src_iter) try: tgt_batch next(tgt_iter) except StopIteration: tgt_iter iter(tgt_loader) tgt_batch next(tgt_iter) src_img, src_label, src_mask src_batch tgt_img tgt_batch[0] out_s model(src_img) loss_cls bce(out_s[cls], src_label) loss_seg dice_loss(out_s[seg], src_mask) loss_task loss_cls loss_seg * 0.5 # 目标域只参与特征对齐不参与任何监督loss ftr_s model.pool(model.features(src_img)).flatten(1) ftr_t model.pool(model.features(tgt_img)).flatten(1) loss_align rbf_mmd(ftr_s, ftr_t) loss_gate 0.02 * model.gate_entropy # 抑制门控坍缩 total_loss loss_task 0.1 * loss_align loss_gate optimizer.zero_grad() total_loss.backward() optimizer.step() if step % 100 0: print(fstep {step} cls{loss_cls.item():.4f} fseg{loss_seg.item():.4f} align{loss_align.item():.4f} fgate_entropy{model.gate_entropy.item():.4f})这里的几个系数不是玄学是有实测意义的。loss_seg * 0.5是因为DiceLoss数值偏小乘以系数让分类和分割保持同一量级。loss_align * 0.1是让MMD作为正则项而不是主导项否则模型会过度追求分布一致丢掉任务语义。loss_gate * 0.02是门控熵正则门控全部坍缩到单一专家时熵为0惩罚会让门控尽量保持一定分散度。如果想要更精细的控制可以每500步动态调整loss_align的权重比如从0.05升到0.2模拟课程学习。不过追求省心的版本就直接固定0.1也能收敛。5. 避坑让ExpertNet在医疗图像上翻车的五个细节5.1 label都是nanloss也跟着变nan现象训练跑到二三百步打印出来的loss突然变成nan且从此之后不恢复于是只能回滚checkpoint重训。原因源头多半是DICOM读出来的像素里有Inf或NaNpixel_array.astype(np.float32)不会报错但带进计算图后反向传播直接爆掉。另一个原因是BN层在极小的batch下方差为0除以0产生nan。解决在load_dicom_to_rgb里加np.nan_to_num并把输入图像clip到合法范围。batch_size小于8时把BN替换成GroupNorm或者用梯度累积把等效batch拉大到16以上。这个坑最气人的是它随机出现有时候跑一整晚才炸一次所以数据装载函数里务必加异常兜底。5.2 门控坍缩所有样本选了同一个专家现象训练后打印gate_weight发现绝大多数样本在任务上的权重分布都接近one-hot且每次都是同一个专家。专家之间的参数差越来越小等于门控机制失效退化成普通共享head。原因多任务网络刚初始化时某个专家恰好梯度更新快门控就偏向它门控越偏向其他专家梯度越小进入“富者愈富”的循环。如果不加约束几乎必然发生。解决我在代码里加了门控熵正则loss_gate 0.02 * entropy就是把门控权重的分布熵拉高。另外可以在ExpertModule的gate前面加一个温度参数训练初期温度高让权重平滑后期温度降低让门控更果断。专家dropout也有效每个专家输出有10%概率被置零迫使其他专家保持竞争力。5.3 分类loss降了分割loss不动现象训练曲线里分类loss一路下滑分割loss却像死鱼一样横盘最终只学到了分类能力mask输出接近全零。原因两个loss数值量级差太多BCE是0.30.5DiceLoss通常是0.050.2。优化器把所有梯度都花在大的分类loss上分割任务形同虚设。解决先打印两个loss的实际数值计算出量级差再定权重系数。比如我代码里写的loss_seg * 0.5就是根据实际比例拍的。更稳妥的办法是使用不确定性加权把每个任务的标准差作为可学习参数自动调节权重但那是另一个话题项目早期手工调就够了。5.4 BN统计量漂移让自适应变成负优化现象目标域验证AUC比源域直接推理还低甚至接近0.5但训练loss正常下降特征MMD也在减小。黑匣子一样怎么看都找不出原因。原因无监督自适应过程中虽然我们更新了模型参数但目标域batch size常设得比较小BN的running_mean/var没有足够样本更新仍然停留在源域统计量附近。目标域图像在推理时被归一化到完全错误的范围特征分布自然对不上。解决在无监督训练启动前专门用目标域数据不需要标签跑几个前向并让BN统计量自适应更新即TENT的思路。如果目标域batch太小就做统计量EMA或者干脆用GroupNorm替换backbone的BN层。我早期就是没意识到这个问题自适应做了两星期全在空转。5.5 伪标签自训练把自己带偏现象给目标域打伪标签再拿伪标签做监督训练模型看起来越训越自信但留出集上真实AUC不升反降。原因源域模型在目标域上本身就有系统性偏差伪标签会把错误预测当成真值形成确认偏差。置信度阈值再高也只是筛选出模型“自信但不一定对”的样本错误被持续放大。解决在无监督阶段优先用MMD或对抗域对齐这类不依赖标签的损失。如果确实需要伪标签必须配合三类措施置信度阈值提到0.9以上、按类别重采样让样本均衡、每轮只保留与上一轮预测一致的那部分伪标签。即便如此也不要让伪标签loss权重超过监督任务的三分之一。6. 验证与进阶t-SNE可视化加最小指标闭环6.1 最小验证协议看自适应有没有生效跑完无监督训练后拿目标域上少量留出的真实标签做验证同时对比三组结果源域模型直接推理、不做无监督自适应直接微调、做了MMD自适应后的模型。每组都算目标域AUC如果第二组比第一组还差说明BN统计量坑踩中了如果第三组没有明显提升说明MMD权重或门控约束有问题。from sklearn.metrics import roc_auc_score model.eval() with torch.no_grad(): preds torch.sigmoid(model(tgt_val_img)[cls]).numpy() auc roc_auc_score(tgt_val_label, preds) print(ftarget AUC after adaptation: {auc:.4f})顺便把backbone最后输出的特征拿来做t-SNE可视化目标域颜色按真实类别标上。如果训练有效两个域的特征点会纠缠在一起而不是两团分明如果门控坍缩任务embedding的t-SNE上会看到一个巨大的主团。这个可视化比任何loss曲线都直观适合拿去跟团队同步进展。6.2 进阶用对比自监督代替伪标签如果MMD对齐之后瓶颈很明显比如同类别内部方差大但类别间重叠我会在专家模块后加一个对比头把同一张目标域图像的两种强度增强当正样本把不同病人图像当负样本拉近正样本特征、推远负样本。对比学习不依赖标签天然契合无监督自适应还能逼着专家学到更细粒度的域不变特征。def contrastive_loss(feat_t1, feat_t2, temperature0.07): # feat_t1, feat_t2 是同一 batch 目标域图像两种增强的专家输出 feat_t1 torch.nn.functional.normalize(feat_t1, dim-1) feat_t2 torch.nn.functional.normalize(feat_t2, dim-1) logits torch.matmul(feat_t1, feat_t2.T) / temperature labels torch.arange(logits.size(0)) return torch.nn.functional.cross_entropy(logits, labels)把这个contrastive loss加到总loss里权重从0.1开始调一般效果会比MMD更稳。但代价是训练时间几乎翻倍因为目标域每个batch要做两次前向。如果项目周期紧建议先跑MMD版本确认管线没问题再换对比自监督。我自己在这类项目上吃过最多亏的是一开始太急着上高级方案把对比学习、伪标签、域判别器全塞进去结果排查问题时根本分不清是哪个loss在拖后腿。后来养成一个习惯无监督自适应阶段每次只调一个损失项先跑通MMD再逐步叠加。希望帮到你。本文还有配套的精品资源点击获取
ExpertNet与ResNet50多任务无监督自适应：医疗影像跨域迁移落地指南
简介基于ExpertNet与ResNet50构建的多任务学习网络无监督自适应模型源码面向医疗图像分析与深度学习开发者旨在解决标注数据稀缺场景下的多疾病特征识别与建模问题。项目通过ResNet50的残差模块提取深层图像特征残差连接有效缓解深层网络梯度消失借助ExpertNet的专家分支为不同识别任务分配专属子网络并采用无监督学习策略降低对人工标注的依赖可应用于肺部CT肿瘤检测、眼底病变诊断等场景。压缩包共16个文件以Python脚本为主10个py涵盖数据预处理、模型构建、训练与评估等完整流程另含requirements.txt与README说明整体仅27KB结构清晰。源码中提供自动编码器训练、模型生成与测试等工具脚本便于理解多任务学习与无监督自适应的工程实现。已有254人学习浏览适合作为医疗图像多任务学习方向的项目参考与二次开发基础。1. 给无标注医疗图像做自适应ExpertNet Resnet50能顶什么用一套在公开胸部X光集上训练好的多任务模型换到本地医院PACS导出的设备数据上AUC掉一截而目标域又往往拿不到标注。标题里这套基于ExpertNet Resnet50的多任务学习网络无监督自适应模型就是用来解决这种“源域很准、目标域拉胯”的现实问题的。ExpertNet当门控专家Resnet50当共享骨干多任务同时学疾病分类与掩码回归无监督自适应阶段不依赖目标域标签只靠统计量对齐和特征分布拉近把模型拉回正轨。这套方案适合手里已有预训练模型、但本地影像没有标注的工程师也适合做跨设备迁移的医疗AI研究者。下面按落地顺序展开先拆架构再讲数据然后给训练代码最后把常见踩坑点集中说透。2. ExpertNet Resnet50的架构拆解门控专家与共享backbone的取舍2.1 门控专家网络不是简单加多个head多任务学习最省钱的做法是共享Resnet50后接多个全连接head一个head分类、一个head回归各自算loss。但医疗图像里的任务差异往往比想象中大疾病分类要的是全局语义器官分割要的是空间细节两点同时压在同一个特征向量上梯度方向会互相拉扯。更麻烦的是不同影像科室的数据分布差异很大一套weight去拟合所有任务和所有域经常是哪头都顾不好。我一般会在这类场景里给backbone后面接一个门控专家模块也就是标题里的ExpertNet。它的核心思路是准备若干个小专家网络各自学到不一样的“偏好”特征再由一个门控网络根据当前输入动态地给每个任务分配专家权重。训练时每个任务用的不是同一份特征而是专家特征的加权组合等于给任务之间装了缓冲层。import torch import torch.nn as nn class ExpertModule(nn.Module): def __init__(self, in_dim2048, num_experts4, num_tasks2, hidden_dim512): super().__init__() self.num_experts num_experts self.num_tasks num_tasks # 每个专家是一个小 MLP负责提取一种任务/域偏好的特征 self.experts nn.ModuleList([ nn.Sequential( nn.Linear(in_dim, hidden_dim), nn.ReLU(inplaceTrue), nn.Linear(hidden_dim, in_dim // 2) ) for _ in range(num_experts) ]) # 门控网络对每个任务输出一组专家权重 self.gate nn.Sequential( nn.Linear(in_dim, hidden_dim), nn.ReLU(inplaceTrue), nn.Linear(hidden_dim, num_experts * num_tasks) ) def forward(self, x): # x: [B, in_dim]Resnet50 backbone 输出的全局特征 gate_logits self.gate(x) # [B, num_experts * num_tasks] gate_logits gate_logits.view(-1, self.num_tasks, self.num_experts) gate_weight torch.softmax(gate_logits, dim-1) # 每个任务自己的专家权重 expert_outs torch.stack([e(x) for e in self.experts], dim1) # [B, E, out_dim] task_embedding torch.einsum(bte,btd-bed, gate_weight, expert_outs) return task_embedding, gate_weight参数上num_experts我默认给4落地上26都可以超过6后训练时间涨得明显但收益递减。in_dim必须跟backbone输出维度对齐用Resnet50的avgpool特征就是2048。task_embedding的维度是[B, num_tasks, 1024]后面接任务head时按任务索引取即可。einsum这行如果看得别扭等价操作是先expert_outs.transpose(0, 1)再用torch.bmm效果一样。门控网络的参数初始化不用特别处理但训练时有个容易翻车的细节门控必须加一点熵正则或者专家dropout否则训练后期权重容易坍缩到某一个专家上这个问题放到第5章展开。2.2 Resnet50在无监督自适应里的正确姿势Resnet50网络结构示意图大家都很熟作为backbone时只需要砍掉尾部。具体来说torchvision的resnet50可以拆成features和fc两部分我们保留到layer4输出前面接一个AdaptiveAvgPool给分类和全局特征layer4的特征图则留给分割类任务。无监督自适应阶段backbone的BatchNorm统计量是个隐形坑。源域训练完成后BN里的running_mean和running_var记录的是源域图像统计量目标域图像一进来如果这些统计量不更新特征分布会被错误拉伸导致后续MMD对齐和门控都跟着偏。所以我会把前两三个stage冻结但让BN层保持track_running_statsTrue并继续更新或者干脆在目标域上用一小批无标签数据预热几步。import torch import torch.nn as nn import torchvision.models as models def build_backbone(pretrainedTrue, freeze_stages2): resnet50 models.resnet50(pretrainedpretrained) # 去掉 resnet50 末尾的 avgpool 与 fc保留到 layer4 features nn.Sequential(*list(resnet50.children())[:-2]) pool nn.AdaptiveAvgPool2d((1, 1)) if freeze_stages 0: # 冻结前几个 stage通常是 conv1/bn1/layer1/layer2 blocks [features[i] for i in range(len(features))] for block in blocks[:freeze_stages * 2 3]: # 保守冻结到 layer1 附近 for p in block.parameters(): p.requires_grad False return features, pool这段代码的冻结范围写得比较保守freeze_stages * 2 3是为了把conv1、bn1、relu、maxpool、layer1都覆盖到。如果数据量很小、目标域跟源域差异又极大我会少冻一层让layer2也参与自适应如果目标域只有几百张无标注图建议只冻结conv1和layer1否则backbone很快被目标域噪声带偏。这里还有个常见的取舍ImageNet预训练权重在医疗图像上到底有没有用我的经验是有用Resnet50的前几层学到的是边缘、纹理这种通用结构对X光、CT同样有效但最后几层语义偏离医疗特征所以后两三层要放开训练不要整条backbone锁死。3. 医疗图像数据管线DICOM转张量与无监督数据划分3.1 DICOM读取、窗宽窗位与归一化医疗图像落地第一步往往是处理DICOM。直接拿pixel_array喂给Resnet50是必翻车的因为DICOM像素值不是常规0255灰度而是带RescaleSlope和RescaleIntercept的原始值有的是CT的HU值有的是DR的线性响应。不先做窗宽窗位调整模型会一直去学习无意义的极值和噪声。import numpy as np import cv2 import pydicom def load_dicom_to_rgb(path, target_size(224, 224)): ds pydicom.dcmread(path) img ds.pixel_array.astype(np.float32) # DICOM 里偶尔会混入 nan/inf先清理不然后面 loss 直接翻车 img np.nan_to_num(img, nan0.0, posinf0.0, neginf0.0) slope float(getattr(ds, RescaleSlope, 1.0)) intercept float(getattr(ds, RescaleIntercept, 0.0)) img img * slope intercept # 只要是 CT就用软组织窗 [-500, 1000]DR/CR 则直接线性拉伸 lower, upper -500.0, 1000.0 img np.clip(img, lower, upper) img (img - lower) / (upper - lower) img (img * 255).astype(np.uint8) img cv2.resize(img, target_size, interpolationcv2.INTER_CUBIC) rgb cv2.cvtColor(img, cv2.COLOR_GRAY2RGB) # 复制三通道对齐 ImageNet 输入 return rgb窗宽窗位这里写的是软组织窗如果项目是肺结节或者胸部DR窗口应该改成[0, 255]或[-300, 1200]。更好做法是在DICOM tag里读取WindowCenter和WindowWidth但很多设备写入不规范兜底方案就是硬编码一个合理范围。你把这份代码直接接上torchvision.transforms.Normalize(mean[0.485,0.456,0.406], std[0.229,0.224,0.225])整条图像管线就通了。目标域和源域要使用完全相同的窗位处理否则自适应等于在一个不对齐的输入空间上操作。3.2 有标签源域与无标签目标域的Dataset设计无监督自适应的训练循环需要同时从两个数据流取数据源域有标签目标域只有图像。如果只做一个Dataset看起来省事但训练时容易把目标域标签当成零张量误传播严重污染loss。我一般分两个Dataset源域返回(image, label, mask)目标域只返回(image,)。from torch.utils.data import Dataset class SourceDataset(Dataset): def __init__(self, file_list, transformNone): self.file_list file_list self.transform transform def __len__(self): return len(self.file_list) def __getitem__(self, idx): path, label, mask self.file_list[idx] image load_dicom_to_rgb(path) if self.transform: image self.transform(image) return image, label, mask class TargetDataset(Dataset): def __init__(self, file_list, transformNone): self.file_list file_list self.transform transform def __len__(self): return len(self.file_list) def __getitem__(self, idx): path self.file_list[idx] image load_dicom_to_rgb(path) if self.transform: image self.transform(image) return image,注意TargetDataset最后返回一个只有一个元素的tuple目的是防止dataloader自动把image本身当作可迭代目标。实际业务里目标域文件列表可能有一万多张全量参与MMD计算会很慢我建议训练时每个epoch随机抽一个子集比如10003000张让MMD估计既稳定又不拖慢训练。数据划分上最容易被忽略的是病人级别的数据泄露。同一个病人的多次检查既出现在源域又出现在目标域会让自适应效果虚高。我会按patient_id做group split保证训练集、验证集里不出现同一个人的不同影像。4. 搭建多任务学习网络与无监督自适应训练脚本4.1 MultiTaskExpertNet的完整前向结构把前面两个模块拼起来就得到项目的主体网络。这里用分类加分割两个任务做演示分类走全局特征和门控专家分割直接吃layer4特征图。分割不经过专家模块听起来不够“多任务”但落地时我会把专家输出的task embedding用广播方式拼接到分割特征图的每个像素位置作为全局上下文注入这样两个任务都利用了门控专家。import torch import torch.nn as nn class MultiTaskExpertNet(nn.Module): def __init__(self, num_classes14, num_experts4): super().__init__() self.features, self.pool build_backbone(pretrainedTrue) self.expert ExpertModule(in_dim2048, num_expertsnum_experts, num_tasks2) self.cls_head nn.Linear(1024, num_classes) self.seg_head nn.Conv2d(2048, 1, kernel_size1) self.gate_entropy None # 训练时记录便于观察门控是否坍缩 def forward(self, x): ftr self.features(x) # [B, 2048, H, W] pooled self.pool(ftr).flatten(1) # [B, 2048] task_embed, gate_weight self.expert(pooled) # [B, 2, 1024] # 分类任务用门控专家的第 0 个输出 cls_feat task_embed[:, 0, :] cls_logits self.cls_head(cls_feat) # 分割任务用 layer4 特征图 全局专家特征 seg_feat ftr global_ctx task_embed[:, 1, :].unsqueeze(-1).unsqueeze(-1) seg_feat seg_feat global_ctx.expand_as(seg_feat) seg_logit self.seg_head(seg_feat) self.gate_entropy -( gate_weight * torch.log(gate_weight 1e-8) ).sum(dim-1).mean() return {cls: cls_logits, seg: seg_logit, gate: gate_weight}这个global_ctx.expand_as(seg_feat)是把1024维向量广播到每个空间位置再逐像素相加相当于给每个像素都注入了全图的任务偏好。实现上会占用一点显存H、W在Resnet50 layer4处通常是7×7或者14×14问题不大。如果输入分辨率是512×512H、W变成16×16显存开销会涨一些建议实验室里先用224验证流程。损失设计上分类用BCEWithLogitsLoss因为医疗多标签任务里一张片子可能同时有多个病灶。分割用DiceLoss加上一点L1正则防止mask输出全黑或全白。这两个loss直接相加往往不行因为BCE的值通常在0.10.5量级而DiceLoss在0.1以下二者数量级不同后面训练循环里我会用不确定性加权或固定系数处理。4.2 训练循环源域监督 目标域MMD对齐 门控正则无监督自适应的常见路线有对抗域判别器、最大均值差异MMD、伪标签自训练。我在这套方案里优先用MMD因为实现稳定、不需要额外的判别器网络去平衡GAN训练。核心思想是拉近源域和目标域在backbone特征空间上的中心距离让模型在目标域上不靠标签也能把特征分布对齐。import torch def rbf_mmd(x_s, x_t, bandwidths[1.0, 5.0, 10.0]): # 输入都是 [N, D]建议每次取相同 batch_size x_s x_s.detach() if x_s.size(0) 1 else x_s x_t x_t.detach() if x_t.size(0) 1 else x_t loss 0.0 ns, nt x_s.size(0), x_t.size(0) for bw in bandwidths: k_ss torch.cdist(x_s, x_s) ** 2 / (2 * bw ** 2) k_tt torch.cdist(x_t, x_t) ** 2 / (2 * bw ** 2) k_st torch.cdist(x_s, x_t) ** 2 / (2 * bw ** 2) loss (-k_ss).exp().mean() loss (-k_tt).exp().mean() loss -2 * (-k_st).exp().mean() return loss / len(bandwidths) def dice_loss(pred, target, smooth1.0): pred torch.sigmoid(pred) inter (pred * target).sum(dim(2, 3)) union pred.sum(dim(2, 3)) target.sum(dim(2, 3)) return 1 - (2 * inter smooth) / (union smooth)MMD用多带宽是为了在特征尺度不确定时更稳实际跑下来bandwidths[1.0, 5.0]也够。注意cdist在特征维度大时计算快但如果batch里有一张图尺寸异常导致特征维度不一致会直接报维度错。所以数据管线里必须保证所有图像经过resize到相同分辨率。主训练循环需要同时从源域和目标域加载数据from torch.utils.data import DataLoader model MultiTaskExpertNet(num_classes14) optimizer torch.optim.AdamW(model.parameters(), lr1e-4, weight_decay1e-4) bce torch.nn.BCEWithLogitsLoss() src_loader DataLoader(SourceDataset(src_files), batch_size16, shuffleTrue) tgt_loader DataLoader(TargetDataset(tgt_files), batch_size16, shuffleTrue) src_iter iter(src_loader) tgt_iter iter(tgt_loader) for step in range(2000): try: src_batch next(src_iter) except StopIteration: src_iter iter(src_loader) src_batch next(src_iter) try: tgt_batch next(tgt_iter) except StopIteration: tgt_iter iter(tgt_loader) tgt_batch next(tgt_iter) src_img, src_label, src_mask src_batch tgt_img tgt_batch[0] out_s model(src_img) loss_cls bce(out_s[cls], src_label) loss_seg dice_loss(out_s[seg], src_mask) loss_task loss_cls loss_seg * 0.5 # 目标域只参与特征对齐不参与任何监督loss ftr_s model.pool(model.features(src_img)).flatten(1) ftr_t model.pool(model.features(tgt_img)).flatten(1) loss_align rbf_mmd(ftr_s, ftr_t) loss_gate 0.02 * model.gate_entropy # 抑制门控坍缩 total_loss loss_task 0.1 * loss_align loss_gate optimizer.zero_grad() total_loss.backward() optimizer.step() if step % 100 0: print(fstep {step} cls{loss_cls.item():.4f} fseg{loss_seg.item():.4f} align{loss_align.item():.4f} fgate_entropy{model.gate_entropy.item():.4f})这里的几个系数不是玄学是有实测意义的。loss_seg * 0.5是因为DiceLoss数值偏小乘以系数让分类和分割保持同一量级。loss_align * 0.1是让MMD作为正则项而不是主导项否则模型会过度追求分布一致丢掉任务语义。loss_gate * 0.02是门控熵正则门控全部坍缩到单一专家时熵为0惩罚会让门控尽量保持一定分散度。如果想要更精细的控制可以每500步动态调整loss_align的权重比如从0.05升到0.2模拟课程学习。不过追求省心的版本就直接固定0.1也能收敛。5. 避坑让ExpertNet在医疗图像上翻车的五个细节5.1 label都是nanloss也跟着变nan现象训练跑到二三百步打印出来的loss突然变成nan且从此之后不恢复于是只能回滚checkpoint重训。原因源头多半是DICOM读出来的像素里有Inf或NaNpixel_array.astype(np.float32)不会报错但带进计算图后反向传播直接爆掉。另一个原因是BN层在极小的batch下方差为0除以0产生nan。解决在load_dicom_to_rgb里加np.nan_to_num并把输入图像clip到合法范围。batch_size小于8时把BN替换成GroupNorm或者用梯度累积把等效batch拉大到16以上。这个坑最气人的是它随机出现有时候跑一整晚才炸一次所以数据装载函数里务必加异常兜底。5.2 门控坍缩所有样本选了同一个专家现象训练后打印gate_weight发现绝大多数样本在任务上的权重分布都接近one-hot且每次都是同一个专家。专家之间的参数差越来越小等于门控机制失效退化成普通共享head。原因多任务网络刚初始化时某个专家恰好梯度更新快门控就偏向它门控越偏向其他专家梯度越小进入“富者愈富”的循环。如果不加约束几乎必然发生。解决我在代码里加了门控熵正则loss_gate 0.02 * entropy就是把门控权重的分布熵拉高。另外可以在ExpertModule的gate前面加一个温度参数训练初期温度高让权重平滑后期温度降低让门控更果断。专家dropout也有效每个专家输出有10%概率被置零迫使其他专家保持竞争力。5.3 分类loss降了分割loss不动现象训练曲线里分类loss一路下滑分割loss却像死鱼一样横盘最终只学到了分类能力mask输出接近全零。原因两个loss数值量级差太多BCE是0.30.5DiceLoss通常是0.050.2。优化器把所有梯度都花在大的分类loss上分割任务形同虚设。解决先打印两个loss的实际数值计算出量级差再定权重系数。比如我代码里写的loss_seg * 0.5就是根据实际比例拍的。更稳妥的办法是使用不确定性加权把每个任务的标准差作为可学习参数自动调节权重但那是另一个话题项目早期手工调就够了。5.4 BN统计量漂移让自适应变成负优化现象目标域验证AUC比源域直接推理还低甚至接近0.5但训练loss正常下降特征MMD也在减小。黑匣子一样怎么看都找不出原因。原因无监督自适应过程中虽然我们更新了模型参数但目标域batch size常设得比较小BN的running_mean/var没有足够样本更新仍然停留在源域统计量附近。目标域图像在推理时被归一化到完全错误的范围特征分布自然对不上。解决在无监督训练启动前专门用目标域数据不需要标签跑几个前向并让BN统计量自适应更新即TENT的思路。如果目标域batch太小就做统计量EMA或者干脆用GroupNorm替换backbone的BN层。我早期就是没意识到这个问题自适应做了两星期全在空转。5.5 伪标签自训练把自己带偏现象给目标域打伪标签再拿伪标签做监督训练模型看起来越训越自信但留出集上真实AUC不升反降。原因源域模型在目标域上本身就有系统性偏差伪标签会把错误预测当成真值形成确认偏差。置信度阈值再高也只是筛选出模型“自信但不一定对”的样本错误被持续放大。解决在无监督阶段优先用MMD或对抗域对齐这类不依赖标签的损失。如果确实需要伪标签必须配合三类措施置信度阈值提到0.9以上、按类别重采样让样本均衡、每轮只保留与上一轮预测一致的那部分伪标签。即便如此也不要让伪标签loss权重超过监督任务的三分之一。6. 验证与进阶t-SNE可视化加最小指标闭环6.1 最小验证协议看自适应有没有生效跑完无监督训练后拿目标域上少量留出的真实标签做验证同时对比三组结果源域模型直接推理、不做无监督自适应直接微调、做了MMD自适应后的模型。每组都算目标域AUC如果第二组比第一组还差说明BN统计量坑踩中了如果第三组没有明显提升说明MMD权重或门控约束有问题。from sklearn.metrics import roc_auc_score model.eval() with torch.no_grad(): preds torch.sigmoid(model(tgt_val_img)[cls]).numpy() auc roc_auc_score(tgt_val_label, preds) print(ftarget AUC after adaptation: {auc:.4f})顺便把backbone最后输出的特征拿来做t-SNE可视化目标域颜色按真实类别标上。如果训练有效两个域的特征点会纠缠在一起而不是两团分明如果门控坍缩任务embedding的t-SNE上会看到一个巨大的主团。这个可视化比任何loss曲线都直观适合拿去跟团队同步进展。6.2 进阶用对比自监督代替伪标签如果MMD对齐之后瓶颈很明显比如同类别内部方差大但类别间重叠我会在专家模块后加一个对比头把同一张目标域图像的两种强度增强当正样本把不同病人图像当负样本拉近正样本特征、推远负样本。对比学习不依赖标签天然契合无监督自适应还能逼着专家学到更细粒度的域不变特征。def contrastive_loss(feat_t1, feat_t2, temperature0.07): # feat_t1, feat_t2 是同一 batch 目标域图像两种增强的专家输出 feat_t1 torch.nn.functional.normalize(feat_t1, dim-1) feat_t2 torch.nn.functional.normalize(feat_t2, dim-1) logits torch.matmul(feat_t1, feat_t2.T) / temperature labels torch.arange(logits.size(0)) return torch.nn.functional.cross_entropy(logits, labels)把这个contrastive loss加到总loss里权重从0.1开始调一般效果会比MMD更稳。但代价是训练时间几乎翻倍因为目标域每个batch要做两次前向。如果项目周期紧建议先跑MMD版本确认管线没问题再换对比自监督。我自己在这类项目上吃过最多亏的是一开始太急着上高级方案把对比学习、伪标签、域判别器全塞进去结果排查问题时根本分不清是哪个loss在拖后腿。后来养成一个习惯无监督自适应阶段每次只调一个损失项先跑通MMD再逐步叠加。希望帮到你。本文还有配套的精品资源点击获取