CFlow原理及代码解析-开发者社区

CFlow解析

1.CFlow
2 训练与推理
3. CFlow 模型结构图解：
- 3.1 整体架构概览（文字版 + 图形描述）
3.2 PyTorch 代码解读
4 如何用自己的数据训练
4.2 进阶技巧（提升性能）
快速试跑（小规模）
自定义数据训练
测试并保存图像

官方文档：https://anomalib.readthedocs.io
https://github.com/gudovskiy/cflow
GitHub 示例：https://github.com/openvinotoolkit/anomalib/tree/main/examples
开源实现：
https://github.com/gudovskiy/cflow
GitHub: https://github.com/galeone/cflow-ad
PyTorch 版本：https://github.com/chenyifei618/CFlow
参考资料
论文原文（CVPR 2023）
https://openaccess.thecvf.com/content/CVPR2023/html/Gudovskiy_CFlow_Real-Time_Pixel-Level_Anomaly_Detection_via_Conditional_Latent_Normalization_CVPR_2023_paper.html

MVTec AD 数据集
https://www.mvtec.com/company/research/datasets/mvtec-ad

1.CFlow

1.1 概念：Conditional Normalizing Flow，深度生成模型（Generative Model），无监督/弱监督异常检测（如产品表面缺陷识别）。 Google Research 在 2021 年提出（论文：CFlow-AD: Real-Time Industrial Anomaly Detection with Conditional Flows），专为工业质检设计。
1.2 背景：自动检测产品是否有划痕、污渍、变形等缺陷。缺陷样本极少（甚至没有），缺陷种类多样、不可预知，要求实时性高（毫秒级响应）。
1.3 思路：学习正常像素的概率分布，根据像素的概率密度估计识别“不符合正常模式”的区域，低似然即异常。只用“正常样本”训练一个模型，让它学会“什么是正常的”，一旦输入偏离就判定为异常。
1.4 原理：图像输入深度学习模型得到特征，学习特征的概率分布。正常样本的特征接近于训练样本的分布，负样本偏离训练样本的分布，依据以上条件判断图片上的缺陷。
1.5 缺点：标准 Flow 模型难以处理高维图像（计算开销大）。对纹理复杂的物体效果下降、显存占用较高。
1.6 训练阶段（仅使用正常样本）
[正常图像]
→ ResNet 提取多层特征
→ 条件化 Flow 模型学习特征分布
→ 最大化似然损失（MLE）
1.7 推理阶段（检测新图像）
[新图像]
→ 提取特征
→ 计算每个位置的似然值
→ 似然越低，越可能是异常
→ 输出热力图（红色=异常区域）
1.8 相关论文与资源
原始论文：
CFlow-AD: Real-Time Industrial Anomaly Detection with Conditional Flows (ICCVW 2021)

2 训练与推理

步骤 1：安装依赖

git clone https://github.com/gudovskiy/cflow.git cd cflow pip install -r requirements.txt

步骤 2：训练（使用正常图像）

python train.py --dataset mvtec --data ./data/mvtec --object_class pill --epochs 100

步骤 3：测试并生成热力图

python test.py --weights ./runs/pill/model.pth --save_dir ./results/pill

输出包含：

每张图的异常分数
可视化热力图（红=异常，蓝=正常）

3. CFlow 模型结构图解：

CFlow - Real-Time Pixel-Level Anomaly Detection via Conditional Latent Normalization
一种基于条件归一化流（Conditional Normalizing Flows）的无监督异常检测模型。

3.1 整体架构概览（文字版 + 图形描述）

输入图像 img[224x224 RGB] │ ▼ 预训练骨干网络（如 WideResNet-50） │ → 提取多尺度特征图 {F1, F2[28*28*256], F3[14,14,512]} │ （例如：layer2, layer3, layer4 输出） ▼ 特征重塑（Patch Partition） F2[28*28*256],—>[784 patches] , F2[14×14×512],—>[196 patches] │ → 将每个特征图划分为局部块（patches），如 1×1 或池化后处理。 ▼ Normalizing Flow 模块（带条件归一化） │ ← 条件信号来自同一骨干网络 ▼ 建模潜在变量 z 的概率密度 p(z) │ ▼ 计算似然得分（Likelihood Score） │ ▼ 异常热力图（Anomaly Heatmap）

3.2 PyTorch 代码解读

项目目录结构

cflow/ ├── train.py # 主训练脚本 ├── test.py # 测试与推理脚本 ├── model.py # 核心模型定义（Normalizing Flow + 条件归一化） ├── dataset.py # 数据加载器（MVTec AD 格式） ├── utils.py # 工具函数（如热力图生成、评估指标） └── requirements.txt # 依赖包

model.py

import torch import torch.nn as nn import torch.nn.functional as F # (1) 条件仿射变换层（Conditional Affine Layer） # 这是 CFlow 的核心创新：动态预测缩放和平移参数。 class ConditionalAffine(nn.Module): def __init__(self, dim, cond_dim): super().__init__() self.dim = dim self.cond_dim = cond_dim # 小型网络，根据上下文 cond 预测 s 和 t self.scale_shift_net = nn.Sequential( nn.Linear(cond_dim, 64), nn.ReLU(), nn.Linear(64, dim * 2) # 输出 scale 和 shift ) def forward(self, x, cond): # cond: [B, C_cond] → 缩小到空间维度匹配 B, C, H, W = x.shape cond = F.adaptive_avg_pool2d(cond, (H, W)) # 全局特征 → 空间对齐 cond = cond.permute(0, 2, 3, 1).reshape(-1, self.cond_dim) # [B*H*W, cond_dim] scale_shift = self.scale_shift_net(cond) # [B*H*W, 2*C] scale, shift = scale_shift.chunk(2, dim=1) # 分割为 scale 和 shift scale = scale.view(B, H, W, C).permute(0, 3, 1, 2) # [B, C, H, W] shift = shift.view(B, H, W, C).permute(0, 3, 1, 2) # 可逆变换：y = scale * x + shift y = scale * x + shift log_jac = torch.sum(torch.log(torch.abs(scale)), dim=[1,2,3]) # 对数雅可比行列式 return y, log_jac def reverse(self, y, cond): # 逆变换：用于采样或重建 B, C, H, W = y.shape cond = F.adaptive_avg_pool2d(cond, (H, W)) cond = cond.permute(0, 2, 3, 1).reshape(-1, self.cond_dim) scale_shift = self.scale_shift_net(cond) scale, shift = scale_shift.chunk(2, dim=1) scale = scale.view(B, H, W, C).permute(0, 3, 1, 2) shift = shift.view(B, H, W, C).permute(0, 3, 1, 2) x = (y - shift) / (scale + 1e-8) log_jac = -torch.sum(torch.log(torch.abs(scale)), dim=[1,2,3]) return x, log_jac # 使用 cond（来自骨干网络高层特征）控制低层特征的标准化。 # 所有操作必须是可逆的，以支持概率密度计算。 # (2) 多尺度 Flow 块（Multi-Scale Flow Block） # 每个层级使用多个条件仿射层堆叠： class FlowBlock(nn.Module): def __init__(self, dim, cond_dim, num_layers=3): super().__init__() self.layers = nn.ModuleList([ ConditionalAffine(dim, cond_dim) for _ in range(num_layers) ]) def forward(self, z, cond): total_log_jac = 0 for layer in self.layers: z, log_jac = layer(z, cond) total_log_jac += log_jac return z, total_log_jac # (3) 主干网络 + 多尺度融合 class CFlowModel(nn.Module): def __init__(self, backbone='wideresnet50', layers=['layer3', 'layer4']): super().__init__() self.backbone = torch.hub.load('pytorch/vision', backbone, pretrained=True) self.backbone.eval() # 冻结权重 self.layers = layers # 移除最后的分类头 self.feature_extractor = nn.Sequential(*list(self.backbone.children())[:-2]) # Flow blocks for each layer self.flow_blocks = nn.ModuleDict({ 'layer3': FlowBlock(dim=256, cond_dim=1024), # 示例维度 'layer4': FlowBlock(dim=512, cond_dim=2048) }) def extract_features(self, x): feat3 = None feat4 = None for name, module in self.feature_extractor._modules.items(): x = module(x) if name == 'layer3': feat3 = x.detach() # 不参与梯度更新 elif name == 'layer4': feat4 = x.detach() return {'layer3': feat3, 'layer4': feat4} def forward(self, x): # 提取多尺度特征 feats = self.extract_features(x) total_log_prob = 0 for layer_name in self.layers: feat = feats[layer_name] cond = feats['layer4'] if layer_name == 'layer3' else None # 高层作为条件 _, log_jac = self.flow_blocks[layer_name](feat, cond) total_log_prob += log_jac.mean() return -total_log_prob # 最大似然 → 最小负对数似然 # 主干网络冻结，只训练 flow 层。 # 损失函数：loss = -log p(x)，即最小化负对数似然。

4 如何用自己的数据训练

步骤 1：准备你的数据集
要求：
只需要正常图像（无异常）用于训练
测试时可以包含异常图像（用于验证效果）
图像格式：.png 或 .jpg
推荐尺寸：≥ 224×224

目录结构示例：

your_dataset/ ├── train/ │ └── good/ │ ├── img1.png │ ├── img2.png │ └── ... ├── test/ │ ├── good/ │ │ ├── normal1.png │ ├── broken/ │ │ ├── crack1.png │ │ └── scratch1.png │ └── dirty/ │ └── stain1.png 类别文件夹名（如 broken, dirty）会被自动识别为异常类型。

步骤 2：安装依赖

git clone https://github.com/gudovskiy/cflow.git cd cflow pip install -r requirements.txt 安装 PyTorch（如果未安装） pip install torch torchvision

步骤 3：启动训练

python train.py \ --dataset custom \ --data ./your_dataset \ --object_class your_product_name \ --epochs 100 \ --lr 0.001 \ --batch_size 16 \ --workers 4 \ --save_dir ./runs/your_product_name

参数说明：
–dataset custom 使用自定义数据集
–data 数据根路径
–object_class 类别名称（对应 train/good/ 下的数据）
–epochs 训练轮数（建议 50~100）
–lr 学习率（flow 层可用稍高学习率）

步骤 4：测试与可视化

python test.py \ --weights ./runs/your_product_name/model_last.pth \ --data ./your_dataset \ --dataset custom \ --object_class your_product_name \ --save_dir ./results/your_product_name \ --save_image_thresh 0.8

输出内容：

./results/your_product_name/images/: 包含原始图 + 热力图叠加
scores.txt: 每张图的异常得分
AUC 评估结果（如果有标注）

步骤 5：查看结果
打开 ./results/your_product_name/images/test/ 中的图片，你会看到类似：

红色区域 = 高异常分数（很可能是缺陷）
蓝色区域 = 符合正常模式

4.2 进阶技巧（提升性能）

技巧方法
调整输入大小修改 dataset.py 中 Resize((256,256)) 提高分辨率
增加 flow 层数在 model.py 中增大 num_layers（但会变慢）
使用更强主干替换为 resnet101 或 convnext（需修改通道数）
数据增强在训练中加入轻微旋转/颜色扰动（不影响正常分布建模）

常见问题 FAQ
1 训练时显存不足怎么办？
降低 batch_size 到 4 或 8，或使用更小输入尺寸（如 224→192）

2 模型检测不出小缺陷？
尝试启用更多中间层（如 layer2），或多尺度融合加权优化。

3 能否检测多种产品？
每类产品需单独训练一个模型。CFlow 是“单类别”模型。

4：是否支持视频流实时检测？
支持！推理速度约 30ms/帧，可用于工业相机实时质检。

快速调试命令模板

快速试跑（小规模）

python train.py --dataset mvtec --data ./data/mvtec --object_class pill --epochs 10 --batch_size 4

自定义数据训练

python train.py --dataset custom --data ./mydata --object_class circuit_board --epochs 50

测试并保存图像

python test.py --weights runs/circuit_board/model_last.pth --save_dir results/cb —save_images

CFLOW-AD由一个判别式预训练的编码器和多个多尺度生成式解码器组成，其中解码器显式估计编码特征的似然概率。该方法实现了计算与内存效率的显著提升：在相同输入设置下，CFLOW-AD的速度和模型尺寸均比当前最先进的方法快10倍、小10倍。在MVTec数据集上的实验结果表明，CFLOW-AD在检测任务中比先前方法提升了0.36%的AUROC，在定位任务中分别提升了1.12%的AUROC和2.5%的AUPRO。本文已开源全部代码，并提供完全可复现的实验设置。