2025-6-8

总之，今天提交了两个大作业！。

吐槽一下神秘的计科办事效率，宿管阿姨上门来催：17号之前滚蛋。结果班上说6.5端午过后就会出安排，结果这都三四天了，也还没有。急急急，到了17号就要离开雁栖湖、不知何去何从……估计只能回家？或者是申请一下宿舍。

rl_detr/
├── configs/               # 配置文件
│   ├── train.yaml         # 训练超参数
│   └── model.yaml         # 模型结构配置
├── data/                  # 数据管理
│   ├── datasets/          # 原始数据
│   ├── processed/         # 预处理后的数据
│   └── data_loader.py     # 数据加载器
├── models/                 # 模型定义
│   ├── __init__.py
│   ├── backbone.py         # 骨干网络（ViT）
│   ├── transformer.py      # Transformer编码器和解码器
│   ├── rl_detr.py          # RL-DETR主模型
│   └── heads.py            # 预测头和Value网络
├── training/              # 训练模块
│   ├── train.py           # 主训练脚本
│   ├── optimizer.py       # 优化器配置
│   └── scheduler.py       # 学习率调整
├── evaluation/            # 评估模块
│   ├── eval.py            # 评估脚本
│   ├── metrics.py         # mAP计算
│   └── visualize.py       # 结果可视化
├── inference/             # 推理模块
│   ├── detect.py          # 单张图推理
│   └── webcam_demo.py     # 实时检测
├── utils/                 # 通用工具
│   ├── logger.py          # 日志记录
│   ├── config_parser.py   # 配置解析
│   ├── metrics.py         # 计算mAP等指标
│   └── augmentations.py   # 数据增强
│   └── augmentations.py   # 数据增强
├── requirements.txt       # 依赖列表
└── main.py                 # 主入口

Configs

configs/               # 配置文件
├── train.yaml         # 训练超参数
└── model.yaml         # 模型结构配置

Data

data/                  # 数据管理
├── datasets/          # 原始数据
├── processed/         # 预处理后的数据
└── data_loader.py     # 数据加载器

Dataset:FoodDet100K+搜索的自助餐照片用来迁移学习2k

Models

models/                 # 模型定义
├── __init__.py
├── backbone.py         # 骨干网络（ViT）
├── transformer.py      # Transformer编码器和解码器
├── rl_detr.py          # RL-DETR主模型
└── heads.py            # 预测头和Value网络

1. 搭建模型框架，包括：

- 图像编码器（ViT）

- Transformer编码器-解码器

- 强化学习增强的查询更新机制（在解码器中实现）

- 预测头（分类和回归）

- Value网络（用于计算优势函数）

3. 实现训练循环，包括：

• 前向传播（采样动作）

• 奖励计算

• GPRO损失计算

• 反向传播

1. 骨干网络（backbone.py）

使用预训练的ViT作为骨干网络。

import torch
import torchvision
from torch import nn

class ViTBackbone(nn.Module):
    def __init__(self, name='vit_b_16', pretrained=True):
        super().__init__()
        # 使用torchvision中的ViT
        weights = torchvision.models.ViT_B_16_Weights.DEFAULT if pretrained else None
        vit = torchvision.models.vit_b_16(weights=weights)
        # 去掉分类头
        self.vit = nn.Sequential(*list(vit.children())[:-2])
        # 获取输出特征维度
        self.out_channels = vit.hidden_dim

    def forward(self, x):
        # 输入x: [B, C, H, W]
        x = self.vit(x)  # [B, num_patches+1, hidden_dim]
        # 去掉cls token，只保留patch tokens
        x = x[:, 1:, :]  # [B, num_patches, hidden_dim]
        return x

2. Transformer编码器和解码器（transformer.py）

我们使用PyTorch自带的Transformer实现。注意，我们需要修改解码器以支持RL的查询更新。

import torch
import torch.nn as nn
from torch.nn import TransformerEncoder, TransformerEncoderLayer
from torch.nn import TransformerDecoder, TransformerDecoderLayer

class TransformerEncoder(nn.Module):
    def __init__(self, d_model, nhead, num_layers, dim_feedforward=2048, dropout=0.1):
        super().__init__()
        encoder_layer = TransformerEncoderLayer(d_model, nhead, dim_feedforward, dropout, batch_first=True)
        self.encoder = TransformerEncoder(encoder_layer, num_layers)
        def forward(self, src):
        return self.encoder(src)

class RLTransformerDecoder(nn.Module):
"""
强化学习增强的Transformer解码器。
每层解码器在更新查询时，会生成高斯分布的参数，并采样新的查询。
"""
def __init__(self, d_model, nhead, num_layers, dim_feedforward=2048, dropout=0.1):
    super().__init__()
    self.layers = nn.ModuleList([
        RLTransformerDecoderLayer(d_model, nhead, dim_feedforward, dropout)
        for _ in range(num_layers)
    ])
    self.num_layers = num_layers

def forward(self, tgt, memory, tgt_mask=None, memory_mask=None, tgt_key_padding_mask=None, memory_key_padding_mask=None):
    # tgt: [B, N, d]
    # memory: [B, T, d]
    intermediate = []
    for i, layer in enumerate(self.layers):
        tgt, mu, log_var = layer(tgt, memory, tgt_mask, memory_mask, tgt_key_padding_mask, memory_key_padding_mask)
        intermediate.append((tgt, mu, log_var))

    return tgt, intermediate

class RLTransformerDecoderLayer(nn.Module):
    """
    单层解码器，包含：
    - 自注意力
    - 交叉注意力
    - FFN
    然后生成高斯分布的参数（mu, log_var），并采样新的查询。
    """

    def __init__(self, d_model, nhead, dim_feedforward=2048, dropout=0.1):
        super().__init__()
        self.self_attn = nn.MultiheadAttention(d_model, nhead, dropout=dropout, batch_first=True)
        self.multihead_attn = nn.MultiheadAttention(d_model, nhead, dropout=dropout, batch_first=True)
        # FFN

        self.linear1 = nn.Linear(d_model, dim_feedforward)
        self.dropout = nn.Dropout(dropout)
        self.linear2 = nn.Linear(dim_feedforward, d_model)
        self.norm1 = nn.LayerNorm(d_model)
        self.norm2 = nn.LayerNorm(d_model)
        self.norm3 = nn.LayerNorm(d_model)
        self.dropout1 = nn.Dropout(dropout)
        self.dropout2 = nn.Dropout(dropout)
        self.dropout3 = nn.Dropout(dropout)
        self.activation = nn.ReLU()

        # 生成高斯分布参数的层
        self.mu_head = nn.Linear(d_model, d_model)
        self.log_var_head = nn.Linear(d_model, d_model)

    def forward(self, tgt, memory, tgt_mask=None, memory_mask=None, tgt_key_padding_mask=None, memory_key_padding_mask=None):
        # 自注意力
        tgt2 = self.self_attn(tgt, tgt, tgt, attn_mask=tgt_mask, key_padding_mask=tgt_key_padding_mask)[0]
        tgt = tgt + self.dropout1(tgt2)
        tgt = self.norm1(tgt)
        # 交叉注意力
        tgt2 = self.multihead_attn(tgt, memory, memory, attn_mask=memory_mask, key_padding_mask=memory_key_padding_mask)[0]
        tgt = tgt + self.dropout2(tgt2)
        tgt = self.norm2(tgt)
        # FFN
        tgt2 = self.linear2(self.dropout(self.activation(self.linear1(tgt))))
        tgt = tgt + self.dropout3(tgt2)
        tgt = self.norm3(tgt)
        # 生成高斯分布参数
        mu = self.mu_head(tgt)
        log_var = self.log_var_head(tgt)
        # 重参数化采样
        if self.training:
            std = torch.exp(0.5 * log_var)
            eps = torch.randn_like(std)
            new_tgt = mu + eps * std
        else:
            new_tgt = mu

    return new_tgt, mu, log_var

3. 预测头和Value网络（heads.py）

import torch.nn as nn

class DetectionHead(nn.Module):
    """
    预测边界框和类别。
    """
    def __init__(self, d_model, num_classes):
        super().__init__()
        self.cls_head = nn.Linear(d_model, num_classes)
        self.bbox_head = nn.Sequential(
            nn.Linear(d_model, d_model),
            nn.ReLU(),
            nn.Linear(d_model, 4)  # (cx, cy, w, h)
        )

    def forward(self, x):
        cls_logits = self.cls_head(x)  # [B, N, num_classes]
        bbox = self.bbox_head(x)  # [B, N, 4]
    return cls_logits, bbox


class ValueHead(nn.Module):
    """
    用于计算状态值V(q_k)
    """
    def __init__(self, d_model, hidden_dim=256):
        super().__init__()
            self.value_head = nn.Sequential(
            nn.Linear(d_model, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, 1)
        )

    def forward(self, x):
        # x: [B, N, d]
        # 先池化：平均池化
        x_pooled = x.mean(dim=1)  # [B, d]
    return self.value_head(x_pooled)  # [B, 1]

4. RL-DETR主模型（rl_detr.py）

import torch
import torch.nn as nn
from .backbone import ViTBackbone
from .transformer import TransformerEncoder, RLTransformerDecoder
from .heads import DetectionHead, ValueHead

class RL_DETR(nn.Module):
    def __init__(self, num_classes, d_model=768, nhead=8, num_encoder_layers=6, num_decoder_layers=6, dim_feedforward=2048, dropout=0.1):
        super().__init__()
        # 骨干网络
        self.backbone = ViTBackbone()
        # 调整通道数（如果需要）
        if self.backbone.out_channels != d_model:
            self.proj = nn.Linear(self.backbone.out_channels, d_model)
        else:
            self.proj = nn.Identity()
        # 位置编码（可学习）
        self.pos_embed = nn.Parameter(torch.randn(1, 196, d_model))  # 假设图像被分为14x14=196个patch
        # Transformer编码器
        self.encoder = TransformerEncoder(d_model, nhead, num_encoder_layers, dim_feedforward, dropout)
        # 初始查询向量
        self.query_embed = nn.Embedding(100, d_model)  # 100个查询
        # 强化学习解码器
        self.decoder = RLTransformerDecoder(d_model, nhead, num_decoder_layers, dim_feedforward, dropout)
        # 预测头（每一层解码器后都要预测）
        self.heads = nn.ModuleList([DetectionHead(d_model, num_classes) for _ in range(num_decoder_layers)])
        # Value网络（每一层都需要一个Value网络？或者共享？这里我们共享）
        self.value_head = ValueHead(d_model)
        self.num_decoder_layers = num_decoder_layers
        self.d_model = d_model

    def forward(self, images):
        # 1. 骨干网络提取特征
        features = self.backbone(images)  # [B, num_patches, out_channels]
        features = self.proj(features)    # [B, num_patches, d_model]

        # 2. 添加位置编码
        features = features + self.pos_embed

        # 3. 编码器
        memory = self.encoder(features)  # [B, num_patches, d_model]

        # 4. 初始查询
        batch_size = images.shape[0]
        query_embed = self.query_embed.weight.unsqueeze(0).repeat(batch_size, 1, 1)  # [B, N, d_model]

        # 5. 解码器
        tgt = query_embed
        all_mu = []
        all_log_var = []
        all_hidden = []  # 保存每一层的隐藏状态（用于预测）
        for i in range(self.num_decoder_layers):
            # 注意：这里简化了，实际解码器每一层都需要memory
            tgt, mu, log_var = self.decoder.layers[i](tgt, memory)
            all_mu.append(mu)
            all_log_var.append(log_var)
            all_hidden.append(tgt)
        # 6. 每一层都进行预测

        outputs = []

        for i in range(self.num_decoder_layers):
            cls_logits, bbox = self.heads[i](all_hidden[i])
            outputs.append((cls_logits, bbox))

        # 7. 返回：每一层的预测结果、均值和方差（用于RL训练）
        return outputs, all_mu, all_log_var

训练循环（training/train.py）

由于强化学习的训练循环比较复杂，这里只展示关键步骤。

注意事项

1. 上述代码是概念性的，实际实现需要处理很多细节，例如：

• 匈牙利匹配器的实现（参考DETR）

• 损失函数的计算（分类损失、边界框损失）

• 优势函数的计算（GAE）

• 旧策略参数的保存和更新（PPO）

• 价值函数的训练

2. 由于强化学习的训练不稳定，需要仔细调整超参数。

3. 数据加载部分需要根据实际数据集实现。

4. 为了简化，上述代码中省略了部分实现（如compute_ratio、GAE等）。