目标检测算法完整技术报告

Faster R-CNN、SSD、CenterNet 详细技术分析与公式推导


目录

  1. 理论基础与数学准备
  2. Faster R-CNN 完整技术解析
  3. SSD 完整技术解析
  4. CenterNet 完整技术解析
  5. 性能对比与评估指标
  6. 实现细节与代码框架

理论基础与数学准备

1.1 IoU (Intersection over Union) 计算

定义

IoU(A,B) = |A ∩ B| / |A ∪ B| = |A ∩ B| / (|A| + |B| - |A ∩ B|)

边界框表示

  • 边界框A: (x₁ᴬ, y₁ᴬ, x₂ᴬ, y₂ᴬ)
  • 边界框B: (x₁ᴮ, y₁ᴮ, x₂ᴮ, y₂ᴮ)

详细计算过程

// 交集区域坐标
x₁ᴵ = max(x₁ᴬ, x₁ᴮ)
y₁ᴵ = max(y₁ᴬ, y₁ᴮ)
x₂ᴵ = min(x₂ᴬ, x₂ᴮ)
y₂ᴵ = min(y₂ᴬ, y₂ᴮ)

// 面积计算
Area_intersection = max(0, x₂ᴵ - x₁ᴵ) × max(0, y₂ᴵ - y₁ᴵ)
Area_A = (x₂ᴬ - x₁ᴬ) × (y₂ᴬ - y₁ᴬ)
Area_B = (x₂ᴮ - x₁ᴮ) × (y₂ᴮ - y₁ᴮ)
Area_union = Area_A + Area_B - Area_intersection

IoU = Area_intersection / Area_union

1.2 损失函数基础

Smooth L1 Loss

定义

smooth_L1(x) = {
    0.5x²        if |x| < 1
    |x| - 0.5    otherwise
}

导数

d/dx smooth_L1(x) = {
    x           if |x| < 1
    sign(x)     otherwise
}

Focal Loss (用于处理类别不平衡)

定义

FL(p_t) = -α_t(1 - p_t)^γ log(p_t)

其中:p_t = {
    p       if y = 1
    1 - p   if y = 0
}

Faster R-CNN 完整技术解析

2.1 网络架构详细分析

2.1.1 特征提取网络 (Backbone)

VGG-16 为例

Conv Block 1: 2×(Conv3×3-64) + MaxPool  → 112×112×64
Conv Block 2: 2×(Conv3×3-128) + MaxPool → 56×56×128
Conv Block 3: 3×(Conv3×3-256) + MaxPool → 28×28×256
Conv Block 4: 3×(Conv3×3-512) + MaxPool → 14×14×512
Conv Block 5: 3×(Conv3×3-512) + MaxPool → 7×7×512

特征图尺寸计算

Output_size = ⌊(Input_size + 2×Padding - Kernel_size)/Stride⌋ + 1

2.1.2 RPN 网络详细设计

Anchor 生成机制

多尺度Anchor生成

# 基础anchor大小
base_size = 16  # 对应原图的感受野

# 尺度因子
scales = [8, 16, 32]  # 对应128², 256², 512²像素

# 长宽比
ratios = [0.5, 1, 2]  # 对应1:2, 1:1, 2:1

# 每个特征图位置生成9个anchor
for scale in scales:
    for ratio in ratios:
        w = scale * sqrt(ratio)
        h = scale / sqrt(ratio)
        anchor = generate_anchor(w, h)

Anchor坐标变换

# Anchor中心坐标相对特征图
anchor_centers_x = arange(0, feature_width) * stride + stride/2
anchor_centers_y = arange(0, feature_height) * stride + stride/2

# 生成所有anchor
all_anchors = []
for cy in anchor_centers_y:
    for cx in anchor_centers_x:
        for (w, h) in anchor_shapes:
            anchor = [cx - w/2, cy - h/2, cx + w/2, cy + h/2]
            all_anchors.append(anchor)
RPN 损失函数推导

1. 分类损失 (二分类交叉熵)

设第i个anchor的:

  • 真实标签:pᵢ ∈ {0, 1} (0=背景, 1=前景)
  • 预测概率:p̂ᵢ ∈ [0, 1]
L_cls_i = -[pᵢ log(p̂ᵢ) + (1-pᵢ) log(1-p̂ᵢ)]

L_cls = (1/N_cls) Σᵢ L_cls_i

2. 回归损失推导

边界框编码

# 给定anchor (xₐ, yₐ, wₐ, hₐ) 和 ground truth (x, y, w, h)
t_x = (x - xₐ) / wₐ        # x方向偏移量(归一化)
t_y = (y - yₐ) / hₐ        # y方向偏移量(归一化)
t_w = log(w / wₐ)          # 宽度缩放(对数空间)
t_h = log(h / hₐ)          # 高度缩放(对数空间)

解码过程

# 从预测值 (t̂_x, t̂_y, t̂_w, t̂_h) 恢复边界框
x̂ = t̂_x × wₐ + xₐ
ŷ = t̂_y × hₐ + yₐ
ŵ = wₐ × exp(t̂_w)
ĥ = hₐ × exp(t̂_h)

回归损失

L_reg_i = Σₘ∈{x,y,w,h} smooth_L1(t̂ᵢᵐ - tᵢᵐ)

L_reg = (1/N_reg) Σᵢ pᵢ × L_reg_i  # 只对正样本计算回归损失

总RPN损失

L_RPN = L_cls + λ × L_reg

其中 λ = 10 (平衡分类和回归损失的权重)

2.1.3 Fast R-CNN 头部详细设计

ROI Pooling 数学原理

问题:将任意尺寸的ROI转换为固定尺寸的特征

算法步骤

  1. 将ROI分割为 H×W 个子区域(通常 H=W=7)
  2. 对每个子区域进行最大池化

详细计算

# 给定ROI: (r, c, h, w) 在特征图上的坐标
# 目标输出尺寸: (H, W)

# 计算每个输出bin的大小
bin_h = h / H
bin_w = w / W

# 对于输出位置(i, j)
for i in range(H):
    for j in range(W):
        # 计算对应的输入区域
        start_h = floor(i * bin_h)
        end_h = ceil((i + 1) * bin_h)
        start_w = floor(j * bin_w)
        end_w = ceil((j + 1) * bin_w)
        
        # 在该区域内进行最大池化
        output[i][j] = max(feature_map[r+start_h:r+end_h, c+start_w:c+end_w])
Fast R-CNN 损失函数

多分类损失

L_cls = -log p_u

其中:p_u 是真实类别u的预测概率

边界框回归损失

L_loc = Σᵢ∈{x,y,w,h} smooth_L1(tᵢᵘ - vᵢ)

其中:
- tᵢᵘ 是类别u的预测边界框参数
- vᵢ 是真实边界框参数

总损失

L = L_cls + λ[u ≥ 1] L_loc

其中 [u ≥ 1] 是示性函数,只有非背景类别才计算回归损失

2.2 训练过程详细算法

2.2.1 正负样本分配策略

RPN阶段

def assign_rpn_samples(anchors, gt_boxes):
    ious = compute_iou_matrix(anchors, gt_boxes)
    
    # 策略1: 与任意GT的IoU > 0.7 → 正样本
    positive_anchors = anchors[max(ious, axis=1) > 0.7]
    
    # 策略2: 与每个GT IoU最高的anchor → 正样本
    for gt_idx in range(len(gt_boxes)):
        best_anchor_idx = argmax(ious[:, gt_idx])
        positive_anchors.add(anchors[best_anchor_idx])
    
    # 负样本: IoU < 0.3
    negative_anchors = anchors[max(ious, axis=1) < 0.3]
    
    return positive_anchors, negative_anchors

2.2.2 四步训练算法详解

Step 1: 训练RPN

# 1. 初始化CNN权重(ImageNet预训练)
backbone = load_pretrained_vgg16()

# 2. 添加RPN层
rpn = RPN(backbone.features)

# 3. 训练RPN
for epoch in range(num_epochs):
    for batch in dataloader:
        # 前向传播
        features = backbone(batch.images)
        cls_scores, bbox_preds = rpn(features)
        
        # 计算损失
        rpn_loss = compute_rpn_loss(cls_scores, bbox_preds, batch.gt_boxes)
        
        # 反向传播
        rpn_loss.backward()
        optimizer.step()

Step 2: 训练Fast R-CNN

# 1. 固定RPN,生成proposals
rpn.eval()
with torch.no_grad():
    proposals = rpn.generate_proposals(images)

# 2. 训练Fast R-CNN检测头
detector = FastRCNNHead()
for epoch in range(num_epochs):
    # 使用RPN生成的proposals训练
    detection_loss = train_detector(detector, proposals, gt_boxes)

2.3 推理过程详解

2.3.1 RPN推理

def rpn_inference(features, rpn_model):
    # 1. 前向传播获得分类和回归结果
    cls_scores, bbox_deltas = rpn_model(features)
    
    # 2. 生成所有anchors
    anchors = generate_all_anchors(features.shape)
    
    # 3. 解码边界框
    proposals = []
    for i, (score, delta) in enumerate(zip(cls_scores, bbox_deltas)):
        if score > threshold:  # 前景概率阈值过滤
            # 解码边界框
            proposal = decode_bbox(anchors[i], delta)
            proposals.append((proposal, score))
    
    # 4. NMS去除重复
    proposals = non_max_suppression(proposals, nms_threshold=0.7)
    
    # 5. 保留top-K个proposals
    proposals = proposals[:top_k]  # top_k = 2000
    
    return proposals

2.3.2 Fast R-CNN推理

def fast_rcnn_inference(features, proposals, detector):
    # 1. ROI Pooling
    roi_features = roi_pooling(features, proposals)
    
    # 2. 分类和回归预测
    cls_probs, bbox_deltas = detector(roi_features)
    
    # 3. 后处理
    detections = []
    for i, (proposal, cls_prob, bbox_delta) in enumerate(zip(proposals, cls_probs, bbox_deltas)):
        for class_id in range(1, num_classes):  # 跳过背景类
            if cls_prob[class_id] > detection_threshold:
                # 解码最终边界框
                final_bbox = decode_bbox(proposal, bbox_delta[class_id])
                detections.append({
                    'bbox': final_bbox,
                    'class': class_id,
                    'confidence': cls_prob[class_id]
                })
    
    # 4. 按类别进行NMS
    final_detections = []
    for class_id in range(1, num_classes):
        class_detections = [d for d in detections if d['class'] == class_id]
        class_detections = nms(class_detections, nms_threshold=0.3)
        final_detections.extend(class_detections)
    
    return final_detections

SSD 完整技术解析

3.1 多尺度特征金字塔设计

3.1.1 网络架构详细

基础VGG-16修改

# VGG-16 原始结构修改
Conv4_3: 512 channels → L2 Normalization → 用于检测
Conv5_3: 512 channels → 去除后续pooling和fc层

# 添加的额外卷积层
Conv6: 1024 channels (3×3, stride=1, padding=1)
Conv7: 1024 channels (1×1, stride=1)
Conv8_1: 256 channels (1×1, stride=1)
Conv8_2: 512 channels (3×3, stride=2, padding=1)
Conv9_1: 128 channels (1×1, stride=1)
Conv9_2: 256 channels (3×3, stride=2, padding=1)
Conv10_1: 128 channels (1×1, stride=1)
Conv10_2: 256 channels (3×3, stride=1)
Conv11_1: 128 channels (1×1, stride=1)
Conv11_2: 256 channels (3×3, stride=1)

3.1.2 Default Box 生成算法

尺度计算详细推导

def compute_default_box_sizes(num_layers=6):
    s_min, s_max = 0.2, 0.9  # 最小和最大尺度
    
    # 为每层计算尺度
    scales = []
    for k in range(1, num_layers + 1):
        s_k = s_min + (s_max - s_min) * (k - 1) / (num_layers - 1)
        scales.append(s_k)
    
    # 添加额外尺度 s'_k = sqrt(s_k * s_{k+1})
    extra_scales = []
    for k in range(num_layers - 1):
        s_k_prime = sqrt(scales[k] * scales[k + 1])
        extra_scales.append(s_k_prime)
    
    return scales, extra_scales

Default Box坐标生成

def generate_default_boxes(feature_map_sizes, image_size=300):
    default_boxes = []
    
    for layer_idx, (fmap_size, scale) in enumerate(zip(feature_map_sizes, scales)):
        # 特征图上的步长
        step = image_size / fmap_size
        
        # 为每个特征图位置生成default boxes
        for i in range(fmap_size):
            for j in range(fmap_size):
                # 中心坐标(归一化到[0,1])
                cx = (j + 0.5) * step / image_size
                cy = (i + 0.5) * step / image_size
                
                # 为每个长宽比生成box
                for aspect_ratio in aspect_ratios[layer_idx]:
                    # 基础尺寸
                    w = scale * sqrt(aspect_ratio)
                    h = scale / sqrt(aspect_ratio)
                    
                    # 添加default box (cx, cy, w, h)
                    default_boxes.append([cx, cy, w, h])
                    
                    # 对于aspect_ratio=1,添加额外的box
                    if aspect_ratio == 1:
                        w_prime = sqrt(scale * next_scale)
                        h_prime = sqrt(scale * next_scale)
                        default_boxes.append([cx, cy, w_prime, h_prime])
    
    return default_boxes

3.2 损失函数完整推导

3.2.1 正负样本匹配策略

匹配算法

def match_default_boxes_to_groundtruth(default_boxes, gt_boxes, gt_labels, threshold=0.5):
    # 计算IoU矩阵
    ious = compute_iou_matrix(default_boxes, gt_boxes)  # [N_default, N_gt]
    
    matches = []  # 每个default box的匹配GT索引
    labels = []   # 每个default box的类别标签
    
    # 策略1: 为每个GT分配IoU最高的default box
    for gt_idx in range(len(gt_boxes)):
        best_default_idx = argmax(ious[:, gt_idx])
        matches[best_default_idx] = gt_idx
        labels[best_default_idx] = gt_labels[gt_idx]
    
    # 策略2: IoU > threshold的default box匹配到对应GT
    for default_idx in range(len(default_boxes)):
        max_iou_gt_idx = argmax(ious[default_idx, :])
        if ious[default_idx, max_iou_gt_idx] > threshold:
            matches[default_idx] = max_iou_gt_idx
            labels[default_idx] = gt_labels[max_iou_gt_idx]
        else:
            matches[default_idx] = -1  # 背景
            labels[default_idx] = 0
    
    return matches, labels

3.2.2 位置编码与解码

编码过程

def encode_boxes(matched_gt_boxes, default_boxes, variances=[0.1, 0.1, 0.2, 0.2]):
    """
    将ground truth相对于default box进行编码
    """
    # 中心点坐标编码
    g_cx = (matched_gt_boxes[:, 0] + matched_gt_boxes[:, 2]) / 2  # GT中心x
    g_cy = (matched_gt_boxes[:, 1] + matched_gt_boxes[:, 3]) / 2  # GT中心y
    g_w = matched_gt_boxes[:, 2] - matched_gt_boxes[:, 0]         # GT宽度
    g_h = matched_gt_boxes[:, 3] - matched_gt_boxes[:, 1]         # GT高度
    
    d_cx, d_cy, d_w, d_h = default_boxes[:, 0], default_boxes[:, 1], default_boxes[:, 2], default_boxes[:, 3]
    
    # 编码公式
    encoded_boxes = torch.zeros_like(matched_gt_boxes)
    encoded_boxes[:, 0] = (g_cx - d_cx) / d_w / variances[0]  # Δcx
    encoded_boxes[:, 1] = (g_cy - d_cy) / d_h / variances[1]  # Δcy  
    encoded_boxes[:, 2] = torch.log(g_w / d_w) / variances[2] # Δw
    encoded_boxes[:, 3] = torch.log(g_h / d_h) / variances[3] # Δh
    
    return encoded_boxes

解码过程

def decode_boxes(predictions, default_boxes, variances=[0.1, 0.1, 0.2, 0.2]):
    """
    将预测结果解码为实际边界框
    """
    d_cx, d_cy, d_w, d_h = default_boxes[:, 0], default_boxes[:, 1], default_boxes[:, 2], default_boxes[:, 3]
    
    # 解码公式
    decoded_boxes = torch.zeros_like(predictions)
    decoded_boxes[:, 0] = predictions[:, 0] * variances[0] * d_w + d_cx  # 预测中心x
    decoded_boxes[:, 1] = predictions[:, 1] * variances[1] * d_h + d_cy  # 预测中心y
    decoded_boxes[:, 2] = torch.exp(predictions[:, 2] * variances[2]) * d_w  # 预测宽度
    decoded_boxes[:, 3] = torch.exp(predictions[:, 3] * variances[3]) * d_h  # 预测高度
    
    # 转换为 (x1, y1, x2, y2) 格式
    decoded_boxes[:, 0] -= decoded_boxes[:, 2] / 2  # x1 = cx - w/2
    decoded_boxes[:, 1] -= decoded_boxes[:, 3] / 2  # y1 = cy - h/2
    decoded_boxes[:, 2] += decoded_boxes[:, 0]      # x2 = x1 + w
    decoded_boxes[:, 3] += decoded_boxes[:, 1]      # y2 = y1 + h
    
    return decoded_boxes

3.2.3 损失函数详细推导

总体损失函数

L(x, c, l, g) = (1/N)[L_conf(x, c) + α·L_loc(x, l, g)]

其中:

  • x: 匹配指示矩阵,x_{ij}^p = 1表示第i个default box匹配到类别p的第j个ground truth
  • c: 类别置信度预测
  • l: 位置预测
  • g: ground truth位置
  • N: 匹配的正样本数量

1. 置信度损失推导

L_conf(x, c) = -∑_{i∈Pos} x_{ij}^p log(ĉ_i^p) - ∑_{i∈Neg} log(ĉ_i^0)

Softmax计算

ĉ_i^p = exp(c_i^p) / ∑_{p'} exp(c_i^{p'})

Hard Negative Mining详细算法

def hard_negative_mining(conf_loss, pos_mask, neg_ratio=3):
    """
    选择最困难的负样本进行训练
    """
    # 正样本数量
    num_pos = pos_mask.sum()
    num_neg = neg_ratio * num_pos
    
    # 只考虑负样本的损失
    neg_conf_loss = conf_loss.clone()
    neg_conf_loss[pos_mask] = 0
    
    # 排序并选择top-k困难负样本
    _, indices = neg_conf_loss.sort(descending=True)
    hard_neg_mask = torch.zeros_like(pos_mask)
    hard_neg_mask[indices[:num_neg]] = 1
    
    return hard_neg_mask

2. 位置损失推导

L_loc(x, l, g) = ∑_{i∈Pos} ∑_{m∈\{cx,cy,w,h\}} x_{ij}^k smooth_{L1}(l_i^m - ĝ_j^m)

其中 ĝ 是编码后的ground truth:

ĝ_j^{cx} = (g_j^{cx} - d_i^{cx}) / d_i^w
ĝ_j^{cy} = (g_j^{cy} - d_i^{cy}) / d_i^h  
ĝ_j^w = log(g_j^w / d_i^w)
ĝ_j^h = log(g_j^h / d_i^h)

3.3 训练算法详细实现

3.3.1 数据增强策略

def ssd_augmentation(image, boxes, labels):
    """
    SSD专用数据增强
    """
    # 1. 随机采样策略
    sample_options = [
        None,  # 使用原图
        {'min_jaccard_overlap': 0.1},
        {'min_jaccard_overlap': 0.3}, 
        {'min_jaccard_overlap': 0.5},
        {'min_jaccard_overlap': 0.7},
        {'min_jaccard_overlap': 0.9}
    ]
    
    option = random.choice(sample_options)
    if option is not None:
        image, boxes, labels = random_crop(image, boxes, labels, **option)
    
    # 2. 随机水平翻转
    if random.random() < 0.5:
        image = horizontal_flip(image)
        boxes[:, [0, 2]] = 1.0 - boxes[:, [2, 0]]  # 翻转x坐标
    
    # 3. 颜色扭曲
    image = random_brightness(image)
    image = random_contrast(image)
    image = random_saturation(image)
    image = random_hue(image)
    
    return image, boxes, labels

def random_crop(image, boxes, labels, min_jaccard_overlap=0.5):
    """
    随机裁剪,保证与GT的IoU满足要求
    """
    height, width = image.shape[:2]
    
    while True:
        # 随机生成裁剪区域
        w = random.uniform(0.3, 1.0) * width
        h = random.uniform(0.3, 1.0) * height
        
        # 长宽比限制
        if h / w < 0.5 or h / w > 2:
            continue
            
        left = random.uniform(0, width - w)
        top = random.uniform(0, height - h)
        
        crop_box = [left, top, left + w, top + h]
        
        # 检查与所有GT box的IoU
        ious = compute_ious(crop_box, boxes)
        if ious.min() >= min_jaccard_overlap:
            # 裁剪图像和调整框坐标
            image = image[int(top):int(top+h), int(left):int(left+w)]
            boxes = adjust_boxes_after_crop(boxes, crop_box)
            return image, boxes, labels

CenterNet 完整技术解析

4.1 关键点检测数学原理

4.1.1 热力图生成详细算法

高斯核函数

G(x, y) = exp(-((x - μ_x)² + (y - μ_y)²) / (2σ²))

自适应σ计算

def gaussian_radius(det_size, min_overlap=0.7):
    """
    根据物体大小计算高斯半径
    """
    height, width = det_size
    
    a1 = 1
    b1 = (height + width)
    c1 = width * height * (1 - min_overlap) / (1 + min_overlap)
    sq1 = sqrt(b1 ** 2 - 4 * a1 * c1)
    r1 = (b1 + sq1) / 2
    
    a2 = 4
    b2 = 2 * (height + width)
    c2 = (1 - min_overlap) * width * height
    sq2 = sqrt(b2 ** 2 - 4 * a2 * c2)
    r2 = (b2 + sq2) / 2
    
    a3 = 4 * min_overlap
    b3 = -2 * min_overlap * (height + width)
    c3 = (min_overlap - 1) * width * height
    sq3 = sqrt(b3 ** 2 - 4 * a3 * c3)
    r3 = (b3 + sq3) / 2
    
    return min(r1, r2, r3)

热力图生成

def generate_heatmap(heatmap, center, radius, k=1):
    """
    在热力图上绘制高斯分布
    """
    diameter = 2 * radius + 1
    gaussian = gaussian2D((diameter, diameter), sigma=diameter / 6)
    
    x, y = int(center[0]), int(center[1])
    
    height, width = heatmap.shape[0:2]
    
    left, right = min(x, radius), min(width - x, radius + 1)
    top, bottom = min(y, radius), min(height - y, radius + 1)
    
    masked_heatmap = heatmap[y - top:y + bottom, x - left:x + right]
    masked_gaussian = gaussian[radius - top:radius + bottom, radius - left:radius + right]
    
    if min(masked_gaussian.shape) > 0 and min(masked_heatmap.shape) > 0:
        heatmap[y - top:y + bottom, x - left:x + right] = maximum(masked_heatmap, masked_gaussian)
    
    return heatmap

4.1.2 损失函数完整推导

改进的Focal Loss

标准Focal Loss

FL(p_t) = -α_t(1 - p_t)^γ log(p_t)

CenterNet改进版本

L_k = -(1/N) ∑_{xyc} {
    (1 - Ŷ_{xyc})^α log(Ŷ_{xyc})                    if Y_{xyc} = 1
    (1 - Y_{xyc})^β (Ŷ_{xyc})^α log(1 - Ŷ_{xyc})    otherwise
}

详细推导过程

  1. 正样本损失项

    • 当 Y_{xyc} = 1 (真实关键点位置)
    • (1 - Ŷ_{xyc})^α:当预测接近1时,权重接近0;预测远离1时,权重较大
    • 这确保了模型专注于困难的正样本
  2. 负样本损失项

    • 当 Y_{xyc} = 0 (非关键点位置)
    • (1 - Y_{xyc})^β:根据距离真实关键点的远近调节权重
    • (Ŷ_{xyc})^α:当预测接近0时权重小,预测接近1时权重大
    • 减少易分负样本的损失贡献

β值的设计意义

def compute_beta_weight(heatmap_gt, center_point, sigma):
    """
    计算负样本权重β
    """
    y, x = torch.meshgrid(torch.arange(heatmap_gt.shape[0]), 
                          torch.arange(heatmap_gt.shape[1]))
    
    # 计算到中心点的距离
    dist = ((x - center_point[0]) ** 2 + (y - center_point[1]) ** 2) / (2 * sigma ** 2)
    
    # β权重:距离中心越远,权重越大
    beta_weight = torch.exp(-dist)
    
    return beta_weight
回归损失详细

L1损失用于offset和size

L_off = (1/N) ∑_{p=1}^N |Ô_{p_k} - (p̃_k/R - p_k)|

L_size = (1/N) ∑_{k=1}^N |Ŝ_{p_k} - s_k|

其中:

  • p_k:第k个物体的真实中心点坐标
  • p̃_k:下采样后的中心点坐标
  • R:下采样比例(通常为4)
  • Ô_{p_k}:预测的offset
  • Ŝ_{p_k}:预测的size
  • s_k:真实物体尺寸

为什么使用L1而非L2

  • L1损失对异常值更鲁棒
  • L1损失在0点不可导,但在实践中表现更好
  • 收敛速度适中,避免梯度爆炸

4.2 网络架构详细设计

4.2.1 骨干网络选择

DLA-34 (Deep Layer Aggregation)架构

class DLA34(nn.Module):
    def __init__(self):
        super(DLA34, self).__init__()
        
        # 基础层
        self.base_layer = nn.Sequential(
            nn.Conv2d(3, 16, 7, stride=1, padding=3, bias=False),
            nn.BatchNorm2d(16),
            nn.ReLU(inplace=True)
        )
        
        # 分层聚合结构
        self.level0 = self._make_conv_level(16, 16, 1)
        self.level1 = self._make_conv_level(16, 32, 2)  # stride=2, 下采样
        self.level2 = Tree(64, 64, 2, level_root=True)  # Tree结构聚合
        self.level3 = Tree(128, 128, 2, level_root=True)
        self.level4 = Tree(256, 256, 2, level_root=True)
        self.level5 = Tree(512, 512, 2, level_root=True)
        
    def forward(self, x):
        y = []
        x = self.base_layer(x)
        
        for i in range(6):
            x = getattr(self, 'level{}'.format(i))(x)
            y.append(x)
            
        return y[-1]  # 返回最高层特征

Tree聚合模块

class Tree(nn.Module):
    def __init__(self, levels, block, in_channels, out_channels, stride=1, level_root=False):
        super(Tree, self).__init__()
        
        if level_root:
            self.root = Root(in_channels, out_channels)
        
        if levels == 1:
            self.subtree = block(in_channels, out_channels, stride)
        else:
            self.subtree = Tree(levels-1, block, in_channels, out_channels, stride)
            
        self.level_root = level_root
        self.levels = levels
        
    def forward(self, x, residual=None, children=None):
        children = [] if children is None else children
        
        if self.levels == 1:
            x1 = self.subtree(x)
            if self.level_root:
                x1 = self.root(x1, residual, *children)
            return x1
        else:
            x1 = self.subtree(x)
            return self.forward(x1, x, children)

4.2.2 上采样网络设计

转置卷积上采样

class DeformConvUpsampling(nn.Module):
    def __init__(self, in_channels, out_channels, kernel_size=4, stride=2, padding=1):
        super().__init__()
        
        # 可变形卷积用于特征对齐
        self.deform_conv = DeformConv2d(in_channels, out_channels, 3, padding=1)
        
        # 转置卷积进行上采样
        self.transpose_conv = nn.ConvTranspose2d(
            out_channels, out_channels, kernel_size, stride, padding, bias=False
        )
        
        self.bn = nn.BatchNorm2d(out_channels)
        self.relu = nn.ReLU(inplace=True)
        
    def forward(self, x):
        x = self.deform_conv(x)
        x = self.transpose_conv(x)
        x = self.bn(x)
        x = self.relu(x)
        return x

4.2.3 预测头设计

三分支预测头

class CenterNetHead(nn.Module):
    def __init__(self, in_channels=64, num_classes=80):
        super().__init__()
        
        # 热力图分支
        self.heatmap = nn.Sequential(
            nn.Conv2d(in_channels, 64, 3, padding=1),
            nn.ReLU(inplace=True),
            nn.Conv2d(64, num_classes, 1),
            nn.Sigmoid()  # 确保输出在[0,1]范围
        )
        
        # offset分支  
        self.offset = nn.Sequential(
            nn.Conv2d(in_channels, 64, 3, padding=1),
            nn.ReLU(inplace=True),
            nn.Conv2d(64, 2, 1)  # 2通道:x_offset, y_offset
        )
        
        # size分支
        self.size = nn.Sequential(
            nn.Conv2d(in_channels, 64, 3, padding=1), 
            nn.ReLU(inplace=True),
            nn.Conv2d(64, 2, 1)  # 2通道:width, height
        )
        
    def forward(self, x):
        heatmap = self.heatmap(x)
        offset = self.offset(x)
        size = self.size(x)
        
        return heatmap, offset, size

4.3 推理算法详细实现

4.3.1 峰值提取算法

def extract_peaks(heatmap, kernel_size=3, threshold=0.1):
    """
    从热力图中提取局部峰值
    """
    # 使用最大池化找局部最大值
    pad = (kernel_size - 1) // 2
    hmax = F.max_pool2d(heatmap, kernel_size, stride=1, padding=pad)
    
    # 保留等于最大值的点(峰值)
    keep = (hmax == heatmap).float()
    
    # 应用阈值过滤
    keep = keep * (heatmap > threshold).float()
    
    return keep * heatmap

def get_topk_from_heatmap(scores, K=100):
    """
    从热力图中获取top-K个检测
    """
    batch, cat, height, width = scores.size()
    
    # 获取每个类别的top-K
    topk_scores, topk_inds = torch.topk(scores.view(batch, cat, -1), K)
    
    # 转换索引为2D坐标
    topk_ys = (topk_inds // width).int().float()
    topk_xs = (topk_inds % width).int().float()
    
    # 获取全局top-K
    topk_score, topk_ind = torch.topk(topk_scores.view(batch, -1), K)
    topk_clses = (topk_ind // K).int()
    topk_inds = gather_feat(topk_inds.view(batch, -1, 1), topk_ind).view(batch, K)
    topk_ys = gather_feat(topk_ys.view(batch, -1, 1), topk_ind).view(batch, K)
    topk_xs = gather_feat(topk_xs.view(batch, -1, 1), topk_ind).view(batch, K)
    
    return topk_score, topk_inds, topk_clses, topk_ys, topk_xs

4.3.2 后处理算法

def decode_centernet_output(heatmap, offset, size, K=100, threshold=0.1):
    """
    解码CenterNet输出为最终检测结果
    """
    batch, num_classes, height, width = heatmap.shape
    
    # 1. 峰值提取
    heatmap = extract_peaks(heatmap)
    
    # 2. 获取top-K检测
    scores, inds, clses, ys, xs = get_topk_from_heatmap(heatmap, K)
    
    # 3. 收集offset和size
    offset = transpose_and_gather_feat(offset, inds)  # [batch, K, 2]
    size = transpose_and_gather_feat(size, inds)      # [batch, K, 2]
    
    # 4. 应用offset修正
    xs = xs.view(batch, K, 1) + offset[:, :, 0:1]
    ys = ys.view(batch, K, 1) + offset[:, :, 1:2]
    
    # 5. 计算边界框
    w = size[:, :, 0:1]
    h = size[:, :, 1:2]
    
    bboxes = torch.cat([
        xs - w / 2,  # x1
        ys - h / 2,  # y1  
        xs + w / 2,  # x2
        ys + h / 2   # y2
    ], dim=2)
    
    # 6. 缩放回原图尺寸
    bboxes *= 4  # 上采样比例
    
    detections = torch.cat([bboxes, scores.unsqueeze(2), clses.unsqueeze(2)], dim=2)
    
    return detections

def transpose_and_gather_feat(feat, ind):
    """
    根据索引收集特征
    """
    feat = feat.permute(0, 2, 3, 1).contiguous()  # [B, H, W, C]
    feat = feat.view(feat.size(0), -1, feat.size(3))  # [B, H*W, C]
    
    feat = gather_feat(feat, ind)
    return feat

def gather_feat(feat, ind):
    """
    根据索引收集特征
    """
    dim = feat.size(2)
    ind = ind.unsqueeze(2).expand(ind.size(0), ind.size(1), dim)
    feat = feat.gather(1, ind)
    return feat

性能对比与评估指标

5.1 评估指标详细定义

5.1.1 mAP计算详解

Average Precision (AP)计算

def compute_ap(recall, precision):
    """
    计算Average Precision
    """
    # 在recall首尾添加0和1
    mrec = np.concatenate(([0.], recall, [1.]))
    mpre = np.concatenate(([0.], precision, [0.]))
    
    # 单调化precision曲线
    for i in range(mpre.size - 1, 0, -1):
        mpre[i - 1] = np.maximum(mpre[i - 1], mpre[i])
    
    # 计算面积(积分)
    i = np.where(mrec[1:] != mrec[:-1])[0]
    ap = np.sum((mrec[i + 1] - mrec[i]) * mpre[i + 1])
    
    return ap

def compute_map(all_detections, all_annotations, iou_threshold=0.5):
    """
    计算mAP
    """
    average_precisions = {}
    
    for label in range(num_classes):
        # 获取该类别的所有检测和标注
        detections = all_detections[label]
        annotations = all_annotations[label]
        
        # 按置信度排序
        detections = sorted(detections, key=lambda x: x['confidence'], reverse=True)
        
        # 计算TP, FP
        tp = np.zeros(len(detections))
        fp = np.zeros(len(detections))
        
        for d, detection in enumerate(detections):
            # 找到同一图像的所有GT
            gt_boxes = [ann for ann in annotations if ann['image_id'] == detection['image_id']]
            
            if len(gt_boxes) > 0:
                # 计算IoU
                ious = [compute_iou(detection['bbox'], gt['bbox']) for gt in gt_boxes]
                max_iou = max(ious)
                max_iou_idx = ious.index(max_iou)
                
                if max_iou >= iou_threshold:
                    if not gt_boxes[max_iou_idx]['detected']:
                        tp[d] = 1
                        gt_boxes[max_iou_idx]['detected'] = True
                    else:
                        fp[d] = 1
                else:
                    fp[d] = 1
            else:
                fp[d] = 1
        
        # 计算累积和
        tp_cumsum = np.cumsum(tp)
        fp_cumsum = np.cumsum(fp)
        
        # 计算recall和precision
        recalls = tp_cumsum / len(annotations)
        precisions = tp_cumsum / (tp_cumsum + fp_cumsum)
        
        # 计算AP
        average_precisions[label] = compute_ap(recalls, precisions)
    
    # 计算mAP
    mean_ap = np.mean(list(average_precisions.values()))
    return mean_ap, average_precisions

5.1.2 COCO评估标准

多IoU阈值mAP

def compute_coco_map(detections, annotations):
    """
    计算COCO风格的mAP (IoU 0.5:0.95)
    """
    iou_thresholds = np.arange(0.5, 1.0, 0.05)  # [0.5, 0.55, ..., 0.95]
    
    maps = []
    for iou_thresh in iou_thresholds:
        map_at_iou, _ = compute_map(detections, annotations, iou_thresh)
        maps.append(map_at_iou)
    
    # COCO mAP是所有IoU阈值下mAP的平均
    coco_map = np.mean(maps)
    
    return {
        'mAP': coco_map,
        'mAP@0.5': maps[0],  # IoU=0.5的mAP
        'mAP@0.75': maps[5]  # IoU=0.75的mAP  
    }

5.2 详细性能对比

5.2.1 定量分析表格

模型骨干网络输入尺寸mAP@0.5mAP@0.5:0.95FPS参数量(M)FLOPs(G)
Faster R-CNNResNet-50800×133373.2%36.2%741.8180
Faster R-CNNResNet-101800×133375.6%39.4%560.3250
SSD300VGG-16300×30068.0%25.1%4626.831
SSD512VGG-16512×51271.6%28.8%2226.891
CenterNetResNet-18512×51267.8%28.1%14211.218
CenterNetResNet-101512×51273.2%34.6%4544.152
CenterNetDLA-34512×51271.3%37.4%5220.132

5.2.2 不同场景下的性能分析

小目标检测性能

目标大小分类 (COCO标准):
- 小目标: area < 32² = 1024 pixels
- 中目标: 32² < area < 96² = 9216 pixels  
- 大目标: area > 96²

小目标mAP对比:
- Faster R-CNN: 21.8%
- SSD512: 10.2%
- CenterNet-DLA34: 18.9%

推理速度详细分析

def benchmark_inference_speed(model, input_size=(3, 512, 512), num_runs=100):
    """
    测试推理速度
    """
    model.eval()
    dummy_input = torch.randn(1, *input_size).cuda()
    
    # 预热
    for _ in range(10):
        with torch.no_grad():
            _ = model(dummy_input)
    
    torch.cuda.synchronize()
    
    # 正式测试
    start_time = time.time()
    for _ in range(num_runs):
        with torch.no_grad():
            _ = model(dummy_input)
    torch.cuda.synchronize()
    end_time = time.time()
    
    avg_time = (end_time - start_time) / num_runs
    fps = 1 / avg_time
    
    return fps, avg_time

5.3 算法复杂度分析

5.3.1 时间复杂度

Faster R-CNN

总复杂度 = O(backbone) + O(RPN) + O(ROI pooling) + O(detection head)

- Backbone: O(H × W × C × K²)  # H,W:特征图尺寸, C:通道数, K:卷积核大小
- RPN: O(H × W × num_anchors × (num_classes + 4))
- ROI Pooling: O(num_proposals × pool_H × pool_W × C)
- Detection Head: O(num_proposals × (num_classes + 4))

假设 H=W=38, C=512, num_anchors=9, num_proposals=300:
复杂度约为 O(10⁸) 量级

SSD

总复杂度 = O(backbone) + Σᵢ O(detection_head_i)

- 每个检测头: O(Hᵢ × Wᵢ × num_default_boxes_i × (num_classes + 4))

SSD300总复杂度约为 O(10⁷) 量级,比Faster R-CNN快一个数量级

CenterNet

总复杂度 = O(backbone) + O(upsampling) + O(prediction_heads)

- Upsampling: O(H × W × C)
- Prediction heads: O(H × W × (num_classes + 6))  # 6 = 2(offset) + 2(size) + 2(额外)

CenterNet复杂度约为 O(10⁶) 量级,最快

5.3.2 空间复杂度

内存使用分析

def estimate_memory_usage(model_name, input_size, batch_size=1):
    """
    估算模型内存使用
    """
    if model_name == "faster_rcnn":
        # 特征图: batch_size × 512 × 38 × 38
        feature_memory = batch_size * 512 * 38 * 38 * 4  # 4 bytes per float
        
        # RPN proposals: batch_size × 2000 × 4
        proposal_memory = batch_size * 2000 * 4 * 4
        
        # ROI features: batch_size × 300 × 512 × 7 × 7  
        roi_memory = batch_size * 300 * 512 * 7 * 7 * 4
        
        total_memory = feature_memory + proposal_memory + roi_memory
        
    elif model_name == "ssd":
        # 多尺度特征图内存
        multi_scale_memory = 0
        feature_sizes = [(38, 38, 512), (19, 19, 1024), (10, 10, 512), 
                        (5, 5, 256), (3, 3, 256), (1, 1, 256)]
        
        for h, w, c in feature_sizes:
            multi_scale_memory += batch_size * h * w * c * 4
            
        total_memory = multi_scale_memory
        
    elif model_name == "centernet":
        # 下采样特征图: batch_size × 64 × 128 × 128
        feature_memory = batch_size * 64 * 128 * 128 * 4
        
        # 三个预测头输出
        heatmap_memory = batch_size * 80 * 128 * 128 * 4  # 80 classes
        offset_memory = batch_size * 2 * 128 * 128 * 4
        size_memory = batch_size * 2 * 128 * 128 * 4
        
        total_memory = feature_memory + heatmap_memory + offset_memory + size_memory
    
    return total_memory / (1024**3)  # 转换为GB

实现细节与代码框架

6.1 PyTorch实现框架

6.1.1 通用检测器基类

import torch
import torch.nn as nn
import torch.nn.functional as F
from abc import ABC, abstractmethod

class BaseDetector(nn.Module, ABC):
    """
    目标检测器基类
    """
    def __init__(self, backbone, num_classes, **kwargs):
        super(BaseDetector, self).__init__()
        self.backbone = backbone
        self.num_classes = num_classes
        self.training_mode = True
        
    @abstractmethod
    def forward(self, images, targets=None):
        """
        前向传播
        Args:
            images: 输入图像 [batch_size, 3, H, W]
            targets: 训练时的标签信息
        Returns:
            训练时返回损失字典,推理时返回检测结果
        """
        pass
    
    @abstractmethod  
    def compute_loss(self, predictions, targets):
        """
        计算损失函数
        """
        pass
    
    @abstractmethod
    def post_process(self, predictions):
        """
        后处理得到最终检测结果
        """
        pass

class DetectionLoss(nn.Module):
    """
    通用检测损失基类
    """
    def __init__(self):
        super(DetectionLoss, self).__init__()
        
    def focal_loss(self, pred, gt, alpha=2, beta=4):
        """
        Focal Loss实现
        """
        pos_inds = gt.eq(1).float()
        neg_inds = gt.lt(1).float()
        
        neg_weights = torch.pow(1 - gt, beta)
        
        loss = 0
        
        pos_loss = torch.log(pred) * torch.pow(1 - pred, alpha) * pos_inds
        neg_loss = torch.log(1 - pred) * torch.pow(pred, alpha) * neg_weights * neg_inds
        
        num_pos = pos_inds.float().sum()
        pos_loss = pos_loss.sum()
        neg_loss = neg_loss.sum()
        
        if num_pos == 0:
            loss = loss - neg_loss
        else:
            loss = loss - (pos_loss + neg_loss) / num_pos
            
        return loss
    
    def smooth_l1_loss(self, pred, gt, beta=1.0):
        """
        Smooth L1 Loss实现
        """
        diff = torch.abs(pred - gt)
        loss = torch.where(diff < beta, 0.5 * diff * diff / beta, diff - 0.5 * beta)
        return loss.sum()

6.1.2 数据加载与预处理

import cv2
import numpy as np
from torch.utils.data import Dataset, DataLoader
import albumentations as A
from albumentations.pytorch import ToTensorV2

class DetectionDataset(Dataset):
    """
    通用目标检测数据集类
    """
    def __init__(self, annotations, image_dir, transforms=None, format='coco'):
        self.annotations = annotations
        self.image_dir = image_dir  
        self.transforms = transforms
        self.format = format
        
    def __len__(self):
        return len(self.annotations)
    
    def __getitem__(self, idx):
        ann = self.annotations[idx]
        
        # 加载图像
        image_path = os.path.join(self.image_dir, ann['filename'])
        image = cv2.imread(image_path)
        image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
        
        # 获取标注
        boxes = ann['boxes']  # [[x1, y1, x2, y2], ...]
        labels = ann['labels']  # [class_id1, class_id2, ...]
        
        # 应用数据增强
        if self.transforms:
            augmented = self.transforms(image=image, bboxes=boxes, class_labels=labels)
            image = augmented['image']
            boxes = augmented['bboxes']
            labels = augmented['class_labels']
        
        # 转换格式
        target = {
            'boxes': torch.FloatTensor(boxes),
            'labels': torch.LongTensor(labels),
            'image_id': torch.tensor([ann['image_id']])
        }
        
        return image, target

def get_detection_transforms(phase='train', image_size=512):
    """
    获取数据增强变换
    """
    if phase == 'train':
        return A.Compose([
            A.Resize(image_size, image_size),
            A.HorizontalFlip(p=0.5),
            A.RandomBrightnessContrast(brightness_limit=0.2, contrast_limit=0.2, p=0.5),
            A.HueSaturationValue(hue_shift_limit=20, sat_shift_limit=30, val_shift_limit=20, p=0.5),
            A.RandomGamma(gamma_limit=(80, 120), p=0.5),
            A.CLAHE(clip_limit=2, tile_grid_size=(8, 8), p=0.5),
            A.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
            ToTensorV2()
        ], bbox_params=A.BboxParams(format='pascal_voc', label_fields=['class_labels']))
    else:
        return A.Compose([
            A.Resize(image_size, image_size),
            A.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
            ToTensorV2()
        ], bbox_params=A.BboxParams(format='pascal_voc', label_fields=['class_labels']))

def collate_fn(batch):
    """
    自定义批处理函数
    """
    images, targets = list(zip(*batch))
    images = torch.stack(images, 0)
    
    # targets保持为列表,因为每个图像的目标数量可能不同
    return images, targets

6.1.3 训练循环实现

import torch.optim as optim
from torch.optim.lr_scheduler import MultiStepLR, ReduceLROnPlateau
from tqdm import tqdm
import logging

class DetectionTrainer:
    """
    目标检测训练器
    """
    def __init__(self, model, train_loader, val_loader, device, config):
        self.model = model.to(device)
        self.train_loader = train_loader
        self.val_loader = val_loader
        self.device = device
        self.config = config
        
        # 优化器
        self.optimizer = self._get_optimizer()
        
        # 学习率调度器
        self.scheduler = self._get_scheduler()
        
        # 日志
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)
        
        # 最佳指标记录
        self.best_map = 0.0
        
    def _get_optimizer(self):
        """获取优化器"""
        if self.config.optimizer == 'sgd':
            return optim.SGD(
                self.model.parameters(), 
                lr=self.config.lr,
                momentum=self.config.momentum,
                weight_decay=self.config.weight_decay
            )
        elif self.config.optimizer == 'adam':
            return optim.Adam(
                self.model.parameters(),
                lr=self.config.lr,
                weight_decay=self.config.weight_decay
            )
        elif self.config.optimizer == 'adamw':
            return optim.AdamW(
                self.model.parameters(),
                lr=self.config.lr,
                weight_decay=self.config.weight_decay
            )
    
    def _get_scheduler(self):
        """获取学习率调度器"""
        if self.config.scheduler == 'multistep':
            return MultiStepLR(
                self.optimizer,
                milestones=self.config.milestones,
                gamma=self.config.gamma
            )
        elif self.config.scheduler == 'plateau':
            return ReduceLROnPlateau(
                self.optimizer,
                mode='max',
                factor=self.config.factor,
                patience=self.config.patience
            )
    
    def train_epoch(self, epoch):
        """训练一个epoch"""
        self.model.train()
        
        epoch_loss = 0.0
        progress_bar = tqdm(self.train_loader, desc=f'Epoch {epoch}')
        
        for batch_idx, (images, targets) in enumerate(progress_bar):
            images = images.to(self.device)
            targets = [{k: v.to(self.device) for k, v in t.items()} for t in targets]
            
            # 前向传播
            loss_dict = self.model(images, targets)
            
            # 计算总损失
            total_loss = sum(loss for loss in loss_dict.values())
            
            # 反向传播
            self.optimizer.zero_grad()
            total_loss.backward()
            
            # 梯度裁剪
            if self.config.get('grad_clip', 0) > 0:
                torch.nn.utils.clip_grad_norm_(self.model.parameters(), self.config.grad_clip)
            
            self.optimizer.step()
            
            # 记录损失
            epoch_loss += total_loss.item()
            
            # 更新进度条
            progress_bar.set_postfix({
                'loss': f'{total_loss.item():.4f}',
                'avg_loss': f'{epoch_loss/(batch_idx+1):.4f}'
            })
            
            # 详细损失日志
            if batch_idx % self.config.log_interval == 0:
                loss_str = ', '.join([f'{k}: {v.item():.4f}' for k, v in loss_dict.items()])
                self.logger.info(f'Epoch: {epoch}, Batch: {batch_idx}, {loss_str}')
        
        return epoch_loss / len(self.train_loader)
    
    def validate(self, epoch):
        """验证"""
        self.model.eval()
        
        val_loss = 0.0
        all_predictions = []
        all_targets = []
        
        with torch.no_grad():
            for images, targets in tqdm(self.val_loader, desc='Validating'):
                images = images.to(self.device)
                targets = [{k: v.to(self.device) for k, v in t.items()} for t in targets]
                
                # 获取预测结果
                if self.model.training:
                    # 如果模型在训练模式,计算损失
                    loss_dict = self.model(images, targets)
                    val_loss += sum(loss for loss in loss_dict.values()).item()
                else:
                    # 推理模式,获取检测结果
                    predictions = self.model(images)
                    all_predictions.extend(predictions)
                    all_targets.extend(targets)
        
        # 计算mAP
        if all_predictions and all_targets:
            map_score = self.compute_map(all_predictions, all_targets)
            self.logger.info(f'Epoch {epoch}: Validation mAP = {map_score:.4f}')
            
            # 保存最佳模型
            if map_score > self.best_map:
                self.best_map = map_score
                self.save_checkpoint(epoch, map_score, is_best=True)
                
            return map_score
        else:
            avg_val_loss = val_loss / len(self.val_loader)
            self.logger.info(f'Epoch {epoch}: Validation Loss = {avg_val_loss:.4f}')
            return avg_val_loss
    
    def train(self, num_epochs):
        """完整训练流程"""
        self.logger.info(f'开始训练,总共 {num_epochs} 个epoch')
        
        for epoch in range(1, num_epochs + 1):
            # 训练
            train_loss = self.train_epoch(epoch)
            
            # 验证
            val_metric = self.validate(epoch)
            
            # 更新学习率
            if isinstance(self.scheduler, ReduceLROnPlateau):
                self.scheduler.step(val_metric)
            else:
                self.scheduler.step()
            
            # 保存检查点
            if epoch % self.config.save_interval == 0:
                self.save_checkpoint(epoch, val_metric)
            
            # 日志记录
            current_lr = self.optimizer.param_groups[0]['lr']
            self.logger.info(
                f'Epoch {epoch}/{num_epochs}: '
                f'Train Loss: {train_loss:.4f}, '
                f'Val Metric: {val_metric:.4f}, '
                f'LR: {current_lr:.2e}'
            )
    
    def save_checkpoint(self, epoch, metric, is_best=False):
        """保存检查点"""
        checkpoint = {
            'epoch': epoch,
            'model_state_dict': self.model.state_dict(),
            'optimizer_state_dict': self.optimizer.state_dict(),
            'scheduler_state_dict': self.scheduler.state_dict() if self.scheduler else None,
            'best_map': self.best_map,
            'config': self.config
        }
        
        filename = f'checkpoint_epoch_{epoch}.pth'
        if is_best:
            filename = 'best_model.pth'
            
        torch.save(checkpoint, os.path.join(self.config.save_dir, filename))
        self.logger.info(f'保存检查点: {filename}')
    
    def compute_map(self, predictions, targets):
        """计算mAP(简化版本)"""
        # 这里应该实现完整的mAP计算逻辑
        # 为简化起见,返回一个占位符值
        return 0.5  # 实际应该调用前面实现的compute_map函数

# 配置类
class Config:
    def __init__(self):
        # 模型配置
        self.model_name = 'centernet'
        self.backbone = 'dla34'
        self.num_classes = 80
        
        # 训练配置
        self.batch_size = 16
        self.num_epochs = 100
        self.lr = 1e-4
        self.optimizer = 'adamw'
        self.weight_decay = 1e-4
        self.momentum = 0.9
        
        # 学习率调度
        self.scheduler = 'multistep'
        self.milestones = [60, 80]
        self.gamma = 0.1
        
        # 其他
        self.grad_clip = 10.0
        self.log_interval = 100
        self.save_interval = 10
        self.save_dir = './checkpoints'

6.2 推理和部署

6.2.1 模型推理接口

class DetectionInference:
    """
    目标检测推理类
    """
    def __init__(self, model_path, config, device='cuda'):
        self.device = device
        self.config = config
        
        # 加载模型
        self.model = self.load_model(model_path)
        self.model.eval()
        
        # 预处理
        self.transforms = get_detection_transforms('test', config.image_size)
        
        # 类别名称
        self.class_names = self.load_class_names(config.class_names_file)
    
    def load_model(self, model_path):
        """加载模型"""
        checkpoint = torch.load(model_path, map_location=self.device)
        
        # 根据配置创建模型
        if self.config.model_name == 'faster_rcnn':
            model = FasterRCNN(self.config)
        elif self.config.model_name == 'ssd':
            model = SSD(self.config)  
        elif self.config.model_name == 'centernet':
            model = CenterNet(self.config)
        
        model.load_state_dict(checkpoint['model_state_dict'])
        model.to(self.device)
        
        return model
    
    def preprocess(self, image):
        """图像预处理"""
        if isinstance(image, str):
            # 文件路径
            image = cv2.imread(image)
            image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
        elif isinstance(image, np.ndarray):
            # numpy数组
            if image.shape[2] == 3 and image.dtype == np.uint8:
                pass  # RGB格式
            else:
                raise ValueError("Unsupported image format")
        
        # 保存原始尺寸
        original_height, original_width = image.shape[:2]
        
        # 应用变换
        augmented = self.transforms(image=image)
        processed_image = augmented['image'].unsqueeze(0)  # 添加batch维度
        
        return processed_image, (original_width, original_height)
    
    def postprocess(self, predictions, original_size, confidence_threshold=0.5):
        """后处理"""
        original_width, original_height = original_size
        processed_width, processed_height = self.config.image_size, self.config.image_size
        
        # 尺度因子
        scale_x = original_width / processed_width
        scale_y = original_height / processed_height
        
        results = []
        
        for pred in predictions:
            # 过滤低置信度检测
            keep_indices = pred['scores'] > confidence_threshold
            
            boxes = pred['boxes'][keep_indices]
            scores = pred['scores'][keep_indices] 
            labels = pred['labels'][keep_indices]
            
            # 恢复原始尺寸
            boxes[:, [0, 2]] *= scale_x  # x坐标
            boxes[:, [1, 3]] *= scale_y  # y坐标
            
            # 转换为列表格式
            for i in range(len(boxes)):
                results.append({
                    'bbox': boxes[i].cpu().numpy().tolist(),
                    'score': float(scores[i].cpu()),
                    'class_id': int(labels[i].cpu()),
                    'class_name': self.class_names[int(labels[i].cpu())]
                })
        
        return results
    
    def predict(self, image, confidence_threshold=0.5, nms_threshold=0.5):
        """单张图像预测"""
        # 预处理
        processed_image, original_size = self.preprocess(image)
        processed_image = processed_image.to(self.device)
        
        # 推理
        with torch.no_grad():
            predictions = self.model(processed_image)
        
        # 后处理
        results = self.postprocess(predictions, original_size, confidence_threshold)
        
        return results
    
    def predict_batch(self, images, confidence_threshold=0.5):
        """批量预测"""
        batch_results = []
        
        for image in images:
            result = self.predict(image, confidence_threshold)
            batch_results.append(result)
            
        return batch_results

# 可视化工具
def visualize_detections(image, detections, class_names, save_path=None):
    """
    可视化检测结果
    """
    import matplotlib.pyplot as plt
    import matplotlib.patches as patches
    
    fig, ax = plt.subplots(1, figsize=(12, 8))
    ax.imshow(image)
    
    colors = plt.cm.Set3(np.linspace(0, 1, len(class_names)))
    
    for det in detections:
        bbox = det['bbox']
        class_name = det['class_name']
        score = det['score']
        class_id = det['class_id']
        
        # 创建矩形框
        rect = patches.Rectangle(
            (bbox[0], bbox[1]), bbox[2] - bbox[0], bbox[3] - bbox[1],
            linewidth=2, edgecolor=colors[class_id], facecolor='none'
        )
        
        ax.add_patch(rect)
        
        # 添加标签
        ax.text(
            bbox[0], bbox[1] - 2,
            f'{class_name}: {score:.2f}',
            color=colors[class_id],
            fontsize=12,
            bbox=dict(facecolor='white', alpha=0.8)
        )
    
    ax.axis('off')
    plt.tight_layout()
    
    if save_path:
        plt.savefig(save_path, dpi=150, bbox_inches='tight')
    plt.show()

# 使用示例
if __name__ == "__main__":
    # 配置
    config = Config()
    config.model_name = 'centernet'
    config.image_size = 512
    config.class_names_file = 'coco_classes.txt'
    
    # 创建推理器
    detector = DetectionInference('best_model.pth', config)
    
    # 预测单张图像
    image_path = 'test_image.jpg'
    detections = detector.predict(image_path, confidence_threshold=0.3)
    
    # 可视化结果
    image = cv2.imread(image_path)
    image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
    
    visualize_detections(image, detections, detector.class_names, 'result.jpg')
    
    print(f"检测到 {len(detections)} 个目标")
    for det in detections:
        print(f"类别: {det['class_name']}, 置信度: {det['score']:.3f}")

总结

本技术报告详细分析了三种主要的目标检测算法:

  1. Faster R-CNN: 两阶段检测的代表,精度高但速度慢
  2. SSD: 单阶段多尺度检测,平衡了精度和速度
  3. CenterNet: 基于关键点的检测方法,简洁高效

每种算法都有其适用场景,实际应用中应根据具体需求选择合适的算法。随着深度学习技术的发展,目标检测算法仍在不断演进,朝着更高精度、更快速度、更低计算成本的方向发展。

Logo

DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐