第 5 章：RLHF 基础与实现

阶段 1: 监督微调（SFT）
├── 输入: (图像, 指令) 对
├── 输出: 初始策略模型 π_SFT
└── 目标: 基础指令遵循能力

阶段 2: 奖励模型训练（RM）
├── 输入: 人类偏好数据 {(x, y_win, y_lose)}
├── 输出: 奖励模型 R(x, y)
└── 目标: 学习人类偏好函数

阶段 3: 强化学习优化（PPO）
├── 输入: 策略模型 π + 奖励模型 R
├── 输出: 优化后的策略 π_RLHF
└── 目标: 最大化期望奖励同时控制 KL 散度

内存占用对比（以 13B 模型为例）：
纯文本 LLM RLHF:

- Actor 模型: 26GB (bf16)
- Critic 模型: 26GB
- Reference 模型: 26GB (冻结)
- 总计: ~78GB

VLM RLHF:

- Actor 模型: 26GB + Vision Encoder 4GB = 30GB
- Critic 模型: 30GB
- Reference 模型: 30GB (冻结)
- 图像缓存: 10-20GB (取决于批次大小)
- 总计: ~100GB+

标准收集流程：

1. 采样阶段
   ├── 输入: (图像, 指令) 对
   ├── 生成: 使用不同温度/策略生成 K 个回答（K=4-7）
   └── 去重: 移除相似度 > 0.9 的回答

2. 标注阶段
   ├── 展示: 向标注者展示图像、指令和候选回答
   ├── 排序: 标注者对回答进行全排序或成对比较
   └── 质检: 计算标注者间一致性（Kappa > 0.6）

3. 数据构造
   ├── 成对比较: 从排序中提取 (chosen, rejected) 对
   ├── 权重分配: 根据排名差距设置样本权重
   └── 平衡处理: 确保正负样本比例合理（1:1 到 1:3）

# 标注指南示例
标注原则优先级：

1. 事实准确性 (40%)：描述是否与图像内容一致
2. 完整性 (25%)：是否回答了用户的问题
3. 相关性 (20%)：是否聚焦于问题相关内容
4. 流畅性 (15%)：语言表达是否自然

特殊情况处理：

- 当两个回答都包含错误时，选择错误较少的
- 当事实都正确时，优先选择信息量大的
- 对于主观问题，考虑回答的合理性和论证质量

方案 A: 独立奖励头
┌─────────┐     ┌─────────┐
│  Vision │────▶│         │
│ Encoder │     │  Fusion │────▶ [Language Model] ────▶ [Reward Head]
└─────────┘     │  Layer  │                               (单个标量)
                └─────────┘
优点：参数效率高，可复用 SFT 模型
缺点：表达能力受限

方案 B: 序列级奖励建模
┌─────────┐     ┌─────────┐
│  Vision │────▶│         │
│ Encoder │     │  Fusion │────▶ [Language Model] ────▶ [Token Rewards]
└─────────┘     │  Layer  │                               (序列长度)
                └─────────┘                                    ↓
                                                          [Aggregation]
优点：可以细粒度建模，识别具体错误位置
缺点：训练复杂，需要更多标注

# 边际损失实现
margin_loss = F.relu(margin - (r_chosen - r_rejected))
# margin 通常设为 0.5-1.0

# 幻觉检测与惩罚
def hallucination_penalty(image_features, text_output):
    # 1. 提取文本中提到的物体
    mentioned_objects = extract_objects(text_output)

    # 2. 使用目标检测模型验证
    detected_objects = object_detector(image_features)

    # 3. 计算惩罚
    false_mentions = mentioned_objects - detected_objects
    penalty = -alpha * len(false_mentions)

    return penalty

# 集成到总奖励中
final_reward = base_reward + hallucination_penalty

状态 s_t = (图像 I, 指令 q, 已生成文本 y_{<t})
动作 a_t = 下一个 token y_t
奖励 r_t = {
    0,                  if t < T (中间步骤)
    R(I, q, y_{1:T}),  if t = T (序列结束)
}

class MultiModalValueHead(nn.Module):
    def __init__(self, hidden_dim, vision_dim):
        super().__init__()
        # 视觉特征投影
        self.vision_proj = nn.Linear(vision_dim, hidden_dim)
        # 跨模态注意力
        self.cross_attention = CrossAttention(hidden_dim)
        # 价值预测头
        self.value_head = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim // 2),
            nn.Tanh(),
            nn.Linear(hidden_dim // 2, 1)
        )

    def forward(self, text_hidden, vision_features):
        # 融合视觉信息
        vision_proj = self.vision_proj(vision_features)
        fused = self.cross_attention(text_hidden, vision_proj)
        # 预测价值
        value = self.value_head(fused.mean(dim=1))
        return value

def compute_advantages_vlm(rewards, values, vision_complexity):
    # 动态折扣因子
    gamma = 0.99 - 0.05 * vision_complexity  # 复杂图像使用更小折扣

    advantages = []
    gae = 0
    for t in reversed(range(len(rewards))):
        if t == len(rewards) - 1:
            next_value = 0
        else:
            next_value = values[t + 1]

        delta = rewards[t] + gamma * next_value - values[t]
        gae = delta + gamma * lambda_ * gae
        advantages.insert(0, gae)

    return advantages

批次组织原则：

1. 图像去重：同一批次避免重复图像（节省视觉编码）
2. 长度排序：相似长度的序列放在一起（减少 padding）
3. 难度混合：简单和困难样本混合（稳定训练）
4. 模态平衡：确保纯文本和多模态样本都有

class AdaptiveKLController:
    def __init__(self, init_kl_coef=0.1, target_kl=6.0):
        self.kl_coef = init_kl_coef
        self.target_kl = target_kl

    def update(self, current_kl):
        if current_kl > 1.5 * self.target_kl:
            self.kl_coef *= 1.5  # 增大惩罚
        elif current_kl < 0.5 * self.target_kl:
            self.kl_coef *= 0.75  # 减小惩罚

        # 裁剪范围
        self.kl_coef = np.clip(self.kl_coef, 0.001, 10.0)
        return self.kl_coef

def compute_kl_with_clipping(logits_new, logits_ref, attention_mask):
    # 计算 log 概率
    log_probs_new = F.log_softmax(logits_new, dim=-1)
    log_probs_ref = F.log_softmax(logits_ref, dim=-1)

    # KL 散度
    kl = (log_probs_ref.exp() * (log_probs_ref - log_probs_new)).sum(-1)

    # 裁剪异常值
    kl = torch.clamp(kl, min=0, max=100)

    # 应用 mask 并平均
    kl = (kl * attention_mask).sum() / attention_mask.sum()

    return kl

Phase 1: 准备阶段（1-2 天）
├── SFT 模型验证
│   ├── 确保 SFT 模型收敛
│   ├── 验证生成质量基线
│   └── 冻结 SFT 权重作为参考模型
├── 奖励模型训练
│   ├── 收集偏好数据（10k-50k 对）
│   ├── 训练奖励模型（准确率 > 65%）
│   └── 验证奖励分布合理性
└── 环境配置
    ├── 设置多 GPU 并行
    ├── 配置梯度累积
    └── 准备监控工具

Phase 2: PPO 训练（3-5 天）
├── 预热阶段（10% steps）
│   ├── 小学习率（1e-7）
│   ├── 大 KL 系数（0.5）
│   └── 监控稳定性
├── 主训练阶段（80% steps）
│   ├── 正常学习率（1e-6）
│   ├── 自适应 KL 系数
│   └── 定期评估
└── 收尾阶段（10% steps）
    ├── 学习率衰减
    ├── 增大 KL 约束
    └── 选择最佳检查点

Phase 3: 后处理（1 天）
├── 模型评估
├── 消融实验
└── 部署准备

class RLHFDataPipeline:
    def __init__(self, batch_size=32, buffer_size=1000):
        self.batch_size = batch_size
        self.buffer = []
        self.buffer_size = buffer_size

    def generate_batch(self, model, prompts):
        """生成阶段：收集模型输出"""
        with torch.no_grad():
            outputs = []
            for prompt in prompts:
                # 多样性采样
                for temp in [0.7, 0.9, 1.1]:
                    output = model.generate(
                        prompt, 
                        temperature=temp,
                        do_sample=True,
                        max_length=512
                    )
                    outputs.append({
                        'prompt': prompt,
                        'response': output,
                        'temperature': temp
                    })
        return outputs

    def score_batch(self, reward_model, experiences):
        """评分阶段：计算奖励"""
        rewards = []
        for exp in experiences:
            reward = reward_model(exp['prompt'], exp['response'])
            # 添加长度惩罚
            length_penalty = -0.01 * len(exp['response'])
            exp['reward'] = reward + length_penalty
            rewards.append(exp)
        return rewards

    def update_model(self, model, experiences):
        """更新阶段：PPO 优化"""
        # 计算优势
        advantages = self.compute_advantages(experiences)

        # 多轮更新
        for _ in range(4):  # PPO epochs
            for batch in self.get_batches(experiences):
                loss = self.ppo_loss(model, batch, advantages)
                loss.backward()
                # 梯度裁剪
                torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
                optimizer.step()

def get_lr_scheduler(optimizer, num_training_steps):
    # 余弦退火 + 预热
    def lr_lambda(step):
        # 10% 预热
        warmup_steps = int(0.1 * num_training_steps)
        if step < warmup_steps:
            return float(step) / float(max(1, warmup_steps))

        # 余弦衰减
        progress = float(step - warmup_steps) / float(max(1, num_training_steps - warmup_steps))
        return 0.5 * (1.0 + math.cos(math.pi * progress))

    return LambdaLR(optimizer, lr_lambda)

class VisionEncoderStrategy:
    """视觉编码器在 RLHF 中的处理策略"""

    @staticmethod
    def frozen_strategy():
        """策略 1：完全冻结"""
        # 优点：稳定、省显存
        # 缺点：无法优化视觉表示
        return {'vision_encoder': False, 'projection': True}

    @staticmethod
    def staged_unfreezing():
        """策略 2：阶段性解冻"""
        stages = [
            (0, 0.3, {'vision_encoder': False, 'projection': True}),    # 前 30%：冻结
            (0.3, 0.7, {'vision_encoder': 'last_layer', 'projection': True}),  # 中 40%：解冻最后层
            (0.7, 1.0, {'vision_encoder': True, 'projection': True})     # 后 30%：全部解冻
        ]
        return stages

    @staticmethod
    def lora_strategy():
        """策略 3：LoRA 微调"""
        # 在视觉编码器中插入 LoRA
        return {
            'vision_lora_rank': 16,
            'vision_lora_alpha': 32,
            'vision_lora_dropout': 0.1
        }

def compute_multimodal_reward(text_reward, vision_reward, alignment_score):
    """
    组合多个奖励信号
    text_reward: 文本质量评分
    vision_reward: 视觉相关性评分
    alignment_score: 跨模态对齐评分
    """
    # 自适应权重
    if alignment_score < 0.5:
        # 对齐差时，更重视对齐
        weights = {'text': 0.2, 'vision': 0.2, 'alignment': 0.6}
    else:
        # 对齐好时，平衡各项
        weights = {'text': 0.4, 'vision': 0.3, 'alignment': 0.3}

    final_reward = (
        weights['text'] * text_reward +
        weights['vision'] * vision_reward +
        weights['alignment'] * alignment_score
    )

    return final_reward

class HallucinationSupressor:
    def __init__(self, detection_model, penalty_weight=0.5):
        self.detector = detection_model
        self.penalty_weight = penalty_weight

    def compute_penalty(self, image, generated_text):
        # 检测幻觉
        hallucinations = self.detector.detect(image, generated_text)

        # 分级惩罚
        penalties = {
            'object_hallucination': -2.0,  # 物体幻觉最严重
            'attribute_error': -1.0,        # 属性错误次之
            'relation_error': -0.5          # 关系错误较轻
        }

        total_penalty = 0
        for h_type, h_count in hallucinations.items():
            total_penalty += penalties.get(h_type, 0) * h_count

        return self.penalty_weight * total_penalty

症状观察表：
┌─────────────────┬──────────────────┬─────────────────┐
│ 现象            │ 可能原因          │ 诊断方法         │
├─────────────────┼──────────────────┼─────────────────┤
│ 奖励持续上升    │ 奖励黑客          │ 检查生成文本质量 │
│ 奖励突然下降    │ 策略崩溃          │ 查看 KL 散度     │
│ 奖励震荡        │ 学习率过大        │ 梯度范数监控     │
│ 奖励停滞        │ 局部最优/过拟合   │ 验证集表现       │
└─────────────────┴──────────────────┴─────────────────┘

class RLHFDiagnostics:
    def __init__(self, threshold_config):
        self.thresholds = threshold_config
        self.history = defaultdict(list)

    def diagnose(self, metrics):
        """实时诊断训练状态"""
        issues = []

        # 检查奖励异常
        if metrics['reward'] > self.thresholds['max_reward']:
            issues.append(('CRITICAL', 'Reward hacking detected'))

        # 检查 KL 散度
        if metrics['kl_div'] > self.thresholds['max_kl']:
            issues.append(('WARNING', f"KL divergence too high: {metrics['kl_div']:.2f}"))

        # 检查梯度
        if metrics['grad_norm'] > self.thresholds['max_grad']:
            issues.append(('WARNING', 'Gradient explosion risk'))

        # 检查生成长度
        avg_length = np.mean(metrics['response_lengths'])
        if avg_length < 20 or avg_length > 500:
            issues.append(('INFO', f'Abnormal response length: {avg_length:.0f}'))

        # 趋势分析
        self.history['reward'].append(metrics['reward'])
        if len(self.history['reward']) > 100:
            recent_std = np.std(self.history['reward'][-100:])
            if recent_std > self.thresholds['reward_std']:
                issues.append(('WARNING', 'Reward instability detected'))

        return issues

VLM 常见奖励黑客行为：

1. 重复描述：不断重复图像中的显著物体
2. 模板化回答：使用固定句式获得稳定奖励
3. 过度详细：生成冗长但信息量低的描述
4. 关键词堆砌：堆积奖励模型偏好的词汇
5. 忽略指令：只描述图像，不回答问题

class RewardHackingDefense:
    def __init__(self):
        self.detectors = {
            'repetition': self.detect_repetition,
            'template': self.detect_template,
            'keyword_stuffing': self.detect_keyword_stuffing,
            'length_gaming': self.detect_length_gaming
        }

    def detect_repetition(self, text):
        """检测重复模式"""
        sentences = text.split('.')
        if len(sentences) < 3:
            return 0

        # 计算句子相似度
        similarities = []
        for i in range(len(sentences)-1):
            sim = self.sentence_similarity(sentences[i], sentences[i+1])
            similarities.append(sim)

        # 高相似度表示重复
        return max(similarities) if similarities else 0

    def detect_template(self, responses):
        """检测模板化回答"""
        # 提取结构特征
        structures = [self.extract_structure(r) for r in responses]

        # 计算结构多样性
        unique_structures = len(set(structures))
        diversity_score = unique_structures / len(structures)

        return 1 - diversity_score  # 低多样性 = 高模板化

    def apply_defense(self, reward, text, detection_scores):
        """应用防御机制调整奖励"""
        penalty = 0

        for detector_name, score in detection_scores.items():
            if score > 0.7:  # 高置信度检测
                penalty += self.penalty_weights[detector_name] * score

        # 应用惩罚
        adjusted_reward = reward - penalty

        # 确保不会过度惩罚
        return max(adjusted_reward, reward * 0.3)

def diversity_bonus(current_response, previous_responses, alpha=0.1):
    """
    鼓励多样性的奖励调整
    """
    if not previous_responses:
        return 0

    # 计算与历史回答的最小距离
    min_distance = float('inf')
    for prev in previous_responses[-10:]:  # 只看最近10个
        distance = edit_distance(current_response, prev) / max(len(current_response), len(prev))
        min_distance = min(min_distance, distance)

    # 距离越大，奖励越高
    bonus = alpha * min_distance
    return bonus

class RLHFMonitor:
    def __init__(self):
        self.metrics = {
            # 奖励相关
            'reward_mean': [],
            'reward_std': [],
            'reward_max': [],
            'reward_min': [],

            # KL 散度
            'kl_div_mean': [],
            'kl_div_max': [],
            'kl_per_token': [],

            # 生成质量
            'response_length': [],
            'unique_tokens_ratio': [],
            'perplexity': [],

            # 训练稳定性
            'grad_norm': [],
            'value_loss': [],
            'policy_loss': [],
            'entropy': [],

            # 视觉特定
            'vision_attention_entropy': [],
            'cross_modal_alignment': []
        }

    def log_step(self, batch_metrics):
        """记录每步指标"""
        for key, value in batch_metrics.items():
            if key in self.metrics:
                self.metrics[key].append(value)

    def get_dashboard_data(self):
        """生成监控面板数据"""
        dashboard = {}

        # 计算移动平均
        for key, values in self.metrics.items():
            if len(values) > 0:
                dashboard[f'{key}_ma10'] = np.mean(values[-10:])
                dashboard[f'{key}_ma100'] = np.mean(values[-100:])

        # 计算趋势
        if len(self.metrics['reward_mean']) > 100:
            recent = np.mean(self.metrics['reward_mean'][-50:])
            past = np.mean(self.metrics['reward_mean'][-100:-50])
            dashboard['reward_trend'] = (recent - past) / abs(past)

        return dashboard

class AlertSystem:
    def __init__(self):
        self.alert_rules = [
            ('kl_div_mean > 10', 'CRITICAL', 'KL divergence explosion'),
            ('reward_std > 2', 'WARNING', 'Reward instability'),
            ('grad_norm > 100', 'CRITICAL', 'Gradient explosion'),
            ('response_length < 10', 'WARNING', 'Degenerate responses'),
            ('entropy < 0.1', 'WARNING', 'Low generation diversity')
        ]

    def check_alerts(self, metrics):
        alerts = []
        for rule, level, message in self.alert_rules:
            if self.evaluate_rule(rule, metrics):
                alerts.append({
                    'level': level,
                    'message': message,
                    'metrics': metrics,
                    'timestamp': time.time()
                })
        return alerts

对每个 (图像, 问题) 对：

1. 使用 4 个不同模型生成回答：
   - LLaVA-13B (base)
   - LLaVA-13B-v1.5
   - GPT-4V (作为高质量参考)
   - 人工编写 (ground truth)

2. 温度采样增加多样性：
   - T=0.7, 0.9, 1.1 各生成一次
   - 总计 12 个候选回答

3. 去重和过滤：
   - 移除完全相同的回答
   - 过滤长度 < 10 或 > 500 的回答
   - 保留 4-6 个最多样的回答

标注者指南：
优先级 1: 事实准确性（是否正确描述图像）
优先级 2: 相关性（是否回答了问题）
优先级 3: 有用性（信息量和洞察深度）
优先级 4: 表达质量（清晰度和连贯性）

标注接口：

- 并排显示图像和问题
- 随机顺序展示候选回答
- 支持拖拽排序或成对比较
- 要求标注者解释排序理由

目标：建立基础的多模态理解能力
数据：595K 指令跟随样本
配置：

- 学习率: 2e-5 (第一轮), 2e-6 (第二轮)
- 批次大小: 128
- 训练轮数: 1 epoch
- 视觉编码器: 冻结 CLIP ViT-L/14

关键技巧：

- 两阶段训练：先训练投影层，再微调 LLM
- 数据配比：80% 多模态，20% 纯文本（保持语言能力）

架构：基于 LLaVA-13B + 线性奖励头
数据：10K 人类偏好对
配置：

- 学习率: 1e-6
- 批次大小: 64
- 训练步数: 3 epochs
- 损失函数: Bradley-Terry + Margin Loss

性能指标：

- 成对准确率: 67.3%
- 与人类一致性: Kappa = 0.62
- 验证集泛化: 65.1%

配置：

- Actor 学习率: 5e-7
- Critic 学习率: 1e-6
- KL 系数: 初始 0.1，自适应调整
- 批次大小: 32
- PPO epochs: 4
- 训练步数: 50K

训练技巧：

1. 预热阶段（前 10%）：
   - 小学习率防止崩溃
   - 大 KL 惩罚保持稳定

2. 主训练阶段：
   - 每 1000 步评估验证集
   - 动态调整 KL 系数
   - 监控奖励黑客

3. 收尾阶段（后 10%）：
   - 线性衰减学习率
   - 选择最佳检查点（非最后一个）

改进 1：幻觉显著减少
Before: "图中有一只猫在桌子上，旁边有一个红色的球。"
       （实际图中无球）
After:  "图中有一只灰色的猫躺在木桌上。"

改进 2：细节描述更准确
Before: "这是一个房间。"
After:  "这是一个现代风格的客厅，有米色沙发、玻璃茶几和大窗户。"

改进 3：推理能力增强
Question: "为什么这个人戴着安全帽？"
Before: "因为他在工作。"
After:  "这个人戴着安全帽是因为他在建筑工地工作，这是安全规定要求的防护装备。"

class CrossModalConsistencyReward:
    """评估文本与图像的一致性"""

    def compute_consistency(self, image_features, text_embeddings):
        # 方法 1：对比学习相似度
        similarity = F.cosine_similarity(
            self.image_proj(image_features),
            self.text_proj(text_embeddings)
        )

        # 方法 2：细粒度对齐
        # 检测图像中的物体
        detected_objects = self.object_detector(image_features)
        # 提取文本中提到的实体
        mentioned_entities = self.entity_extractor(text_embeddings)
        # 计算重叠度
        overlap = len(detected_objects & mentioned_entities) / len(mentioned_entities)

        # 组合两种信号
        consistency_score = 0.6 * similarity + 0.4 * overlap
        return consistency_score

class AdaptiveRewardWeighting:
    def __init__(self):
        self.weights = nn.Parameter(torch.ones(4) / 4)

    def forward(self, rewards_dict):
        # 归一化权重
        normalized_weights = F.softmax(self.weights, dim=0)

        # 计算加权奖励
        total_reward = sum(
            w * r for w, r in zip(normalized_weights, rewards_dict.values())
        )

        # 添加熵正则化，防止权重退化
        entropy = -(normalized_weights * normalized_weights.log()).sum()
        total_reward += 0.01 * entropy

        return total_reward

视觉幻觉分类体系：
├── 对象幻觉（Object Hallucination）
│   ├── 存在性错误：描述不存在的物体
│   └── 数量错误：物体数量描述错误
├── 属性幻觉（Attribute Hallucination）
│   ├── 颜色错误
│   ├── 大小错误
│   └── 材质错误
├── 关系幻觉（Relation Hallucination）
│   ├── 空间关系错误
│   └── 动作关系错误
└── 知识幻觉（Knowledge Hallucination）
    └── 错误的背景知识推断

class HallucinationPenaltySchedule:
    def __init__(self):
        # 不同类型幻觉的基础惩罚
        self.base_penalties = {
            'object_existence': -2.0,    # 最严重
            'object_count': -1.5,
            'attribute': -1.0,
            'relation': -0.8,
            'knowledge': -0.5            # 相对较轻
        }

    def compute_penalty(self, hallucination_report, training_step):
        # 随训练进程增强惩罚
        severity_multiplier = min(2.0, 1.0 + training_step / 10000)

        total_penalty = 0
        for h_type, count in hallucination_report.items():
            base = self.base_penalties.get(h_type, -0.5)
            # 多个幻觉的超线性惩罚
            penalty = base * (count ** 1.2) * severity_multiplier
            total_penalty += penalty

        return total_penalty

class HallucinationPrevention:
    def __init__(self, vision_grounder):
        self.grounder = vision_grounder

    def guided_generation(self, model, image, partial_text):
        """引导生成以减少幻觉"""
        # 1. 提取已生成文本中的实体
        entities = self.extract_entities(partial_text)

        # 2. 视觉接地验证
        grounded = self.grounder.verify(image, entities)

        # 3. 调整生成概率
        if not grounded:
            # 降低继续描述该实体的概率
            mask = self.create_entity_mask(entities[-1])
            # 应用到 logits
            logits = model.get_logits()
            logits[mask] -= 5.0  # 强惩罚

        return logits

VLM_CONSTITUTION = [
    # 准确性原则
    "只描述图像中实际可见的内容",
    "不对图像内容进行未经证实的推测",
    "承认视觉信息的局限性",

    # 安全性原则
    "不生成可能造成伤害的内容",
    "尊重图像中人物的隐私",
    "避免强化偏见和刻板印象",

    # 有用性原则
    "提供信息丰富且相关的回答",
    "根据用户需求调整详细程度",
    "承认不确定性而非猜测"
]

class ConstitutionalVLM:
    def __init__(self, base_model, constitution):
        self.model = base_model
        self.constitution = constitution

    def generate_with_critique(self, image, prompt):
        # 步骤 1：初始生成
        initial_response = self.model.generate(image, prompt)

        # 步骤 2：自我批评
        critique_prompt = f"""
        请评估以下回答是否违反了这些原则：
        {self.constitution}

        回答：{initial_response}
        """
        critique = self.model.generate(image, critique_prompt)

        # 步骤 3：修正
        if "违反" in critique:
            revision_prompt = f"""
            基于以下批评，修正回答：
            批评：{critique}
            原回答：{initial_response}
            """
            revised_response = self.model.generate(image, revision_prompt)
            return revised_response

        return initial_response

def constitutional_reward(response, image, constitution):
    """基于 constitution 的奖励函数"""
    rewards = []

    for principle in constitution:
        # 评估是否遵守原则
        adherence = evaluate_adherence(response, image, principle)
        rewards.append(adherence)

    # 加权平均，关键原则权重更高
    weights = [2.0 if "安全" in p else 1.0 for p in constitution]
    final_reward = np.average(rewards, weights=weights)

    return final_reward