基于模型的离线强化学习

1. 核心思想

基于模型的离线RL方法通过学习环境的动力学模型来生成额外的合成数据，从而缓解数据分布稀疏的问题。

$${\hat{s}}_{t+1}=f(s_t,a_t;\theta )$$

其中 $f$ 是学习到的转移函数（动力学模型）。

1.1 方法论框架

┌─────────────────────────────────────────────────────────────┐
│                    基于模型的离线RL流程                        │
│                                                              │
│  ┌─────────┐    学习     ┌─────────┐    Rollout   ┌─────────┐│
│  │ 离线数据 │ ───────→ │ 动力学   │ ──────────→ │ 策略    ││
│  │   D     │           │  模型    │             │  学习   ││
│  └─────────┘           └─────────┘             └─────────┘│
│       ↑                                            │        │
│       │              悲观约束                        │        │
│       └────────────────────────────────────────────┘        │
└─────────────────────────────────────────────────────────────┘

2. MOPO：Model-based Offline Policy Optimization

2.1 核心思想

MOPO¹的创新点在于：将模型不确定性量化为不确定性奖励，然后用悲观主义处理。

2.2 不确定性奖励

对于模型预测的不确定性，我们将其作为额外奖励惩罚：

$$\hat{r}(s,a)=r(s,a)-\lambda \cdot \hat{\sigma }(s,a)$$

其中 $\hat{\sigma }(s,a)$ 是模型预测的不确定性估计。

2.3 不确定性估计方法

class EnsembleDynamicsModel:
    """
    使用集成模型估计不确定性
    多个模型预测的差异即为不确定性
    """
    def __init__(self, models: List[nn.Module]):
        self.models = models
    
    def predict(self, state, action):
        """返回预测和不确定性"""
        predictions = []
        for model in self.models:
            pred = model(state, action)
            predictions.append(pred)
        
        predictions = torch.stack(predictions)
        
        # 预测均值
        mean = predictions.mean(dim=0)
        
        # 预测标准差作为不确定性
        uncertainty = predictions.std(dim=0)
        
        return mean, uncertainty
    
    def penalty(self, state, action, penalty_coef=1.0):
        """
        计算不确定性惩罚
        """
        _, uncertainty = self.predict(state, action)
        penalty = penalty_coef * uncertainty
        return penalty

2.4 MOPO算法流程

class MOPO:
    def __init__(self, env, penalty_coef=1.0, rollouts=5):
        self.model = EnsembleDynamicsModel(n_models=7)
        self.penalty_coef = penalty_coef
        self.rollouts = rollouts
    
    def collect_data(self, real_dataset, policy):
        """
        收集合成数据
        """
        synthetic_data = []
        
        for _ in range(self.rollouts):
            # 从真实数据中采样起始状态
            batch = sample_batch(real_dataset, batch_size=1000)
            states = batch['state']
            
            for t in range(max_steps):
                # 预测下一步
                with torch.no_grad():
                    actions = policy(states)
                    next_states, uncertainty = self.model.predict(states, actions)
                    
                    # 计算惩罚奖励
                    rewards = batch['reward'] - self.penalty_coef * uncertainty
                    
                    synthetic_data.append({
                        'state': states,
                        'action': actions,
                        'next_state': next_states,
                        'reward': rewards
                    })
                    
                    states = next_states
        
        return synthetic_data
    
    def update(self, real_dataset, policy):
        # 1. 训练动力学模型
        self.model.fit(real_dataset)
        
        # 2. 收集合成数据
        synthetic = self.collect_data(real_dataset, policy)
        
        # 3. 合并数据
        combined_dataset = real_dataset + synthetic
        
        # 4. 在组合数据上训练策略
        policy.update(combined_dataset)

3. COMBO：Conservative Offline Model-based Policy Optimization

3.1 核心思想

COMBO²结合了基于模型和无模型方法的优势：

1. 使用学到的模型生成合成数据
2. 对模型预测施加保守约束（类似CQL）

3.2 目标函数

$$\underset{\theta }{\min }{\mathbb{E}}_{s,a\sim \mathcal{D}}[Q(s,a)]-{\mathbb{E}}_{s,a\sim \hat{\mathcal{D}}}[Q(s,a)]+{\mathcal{L}}_{TD}(\theta )$$

其中 $\hat{\mathcal{D}}$ 是模型生成的数据。

3.3 实现

class COMBO:
    def __init__(self, real_buffer, model_buffer):
        self.real_buffer = real_buffer
        self.model_buffer = model_buffer
    
    def update(self, policy, q1, q2, target_q):
        # 1. 从真实数据和模型数据采样
        real_batch = self.real_buffer.sample()
        model_batch = self.model_buffer.sample()
        
        # 2. CQL-style保守损失
        with torch.no_grad():
            # 模型数据上的动作（可能是OOD）
            model_actions = policy(model_batch.state)
            target_q_values = target_q(model_batch.next_state, model_actions)
            y = model_batch.reward + gamma * target_q_values
        
        # 3. TD损失
        current_q = q1(real_batch.state, real_batch.action)
        td_loss = F.mse_loss(current_q, y)
        
        # 4. 保守损失：在真实数据上Q值应该高于模型数据
        q_real = q1(real_batch.state, real_batch.action)
        q_model = q1(model_batch.state, model_actions)
        conservative_loss = (q_model.mean() - q_real.mean())
        
        total_loss = td_loss + self.alpha * conservative_loss
        return total_loss

4. 模型学习的关键技术

4.1 集成模型

集成多个模型可以提供更好的不确定性估计：

class EnsembleModel:
    def __init__(self, ensemble_size=7):
        self.models = [
            MLP(state_dim + action_dim, state_dim)
            for _ in range(ensemble_size)
        ]
    
    def train(self, dataset, epochs=50):
        for epoch in range(epochs):
            for batch in dataset:
                # 每个模型独立训练
                for model in self.models:
                    loss = F.mse_loss(
                        model(batch.state, batch.action),
                        batch.next_state
                    )
                    model.optim.zero_grad()
                    loss.backward()
                    model.optim.step()
    
    def predict(self, state, action):
        predictions = torch.stack([
            model(state, action) for model in self.models
        ])
        return predictions.mean(0), predictions.std(0)

4.2 模型正则化

为防止模型过拟合到训练数据：

def model_loss(batch, model, alpha=0.1):
    """带正则化的模型损失"""
    pred_next_state = model(batch.state, batch.action)
    
    # 重建损失
    reconstruction = F.mse_loss(pred_next_state, batch.next_state)
    
    # 奖励预测损失
    pred_reward = model_reward(batch.state, batch.action)
    reward_loss = F.mse_loss(pred_reward, batch.reward)
    
    # 拉普拉斯正则化：惩罚高曲率
    laplacian = compute_laplacian(model)
    reg_loss = alpha * laplacian
    
    return reconstruction + reward_loss + reg_loss

4.3 Rollout策略

模型展开长度选择：

短Rollout（1-3步）：
  优点：模型误差小
  缺点：数据多样性低

长Rollout（10+步）：
  优点：数据多样性高
  缺点：模型误差累积

自适应策略：
  根据模型不确定性动态调整步数

5. 理论与局限性

5.1 模型误差累积

MOPO/COMBO的一个核心问题是模型误差会随Rollout累积：

$${\hat{s}}_{t+k}=f({\hat{s}}_{t+k-1},a_{t+k-1})$$

每一步的预测误差 ${ϵ}_t$ 累积为：

$$\Vert {\hat{s}}_{t+k}-s_{t+k}\Vert \approx \sum _{i=0}^{k-1}{ϵ}_i\cdot {\gamma }^i$$

5.2 Edge-of-Reach问题³

当Rollout长度增加时，到达的状态可能超出真实数据覆盖范围：

数据分布                    模型展开后的分布
┌───────────────┐          ┌───────────────┐
│               │          │               │
│   ████████████│          │   ████████████│
│   ████████████│   ⟶     │   ██████▒▒▒▒▒▒│
│   ████████████│          │   ████████████│
│               │          │               │
│   原始数据     │          │  ↑边缘区域↑    │
└───────────────┘          └───────────────┘
                             模型预测但无真实数据

5.3 理论保证

对于COMBO，可以证明以下PAC界：

假设：

• 动力学模型误差有界：$\Vert f(s,a)-\hat{f}(s,a){\Vert }_2\le ϵ$
• Q函数是L-Lipschitz连续

则策略性能差距有界：

$$J(\pi )\ge J(\hat{\pi })-O\left(\frac{ϵ}{(1-\gamma {)}^2}\right)$$

6. 实践建议

6.1 超参数选择

参数	建议范围	说明
模型数量	5-10	更多模型→更好的不确定性估计
Rollout长度	1-5	根据任务调整
惩罚系数 λ	0.1-2.0	太大会过于保守
合成数据比例	20-50%	避免过度依赖模型

6.2 训练策略

# 渐进式增加模型依赖
def adaptive_rollout_schedule(epoch):
    """
    初期：短Rollout，数据主要来自真实数据
    后期：长Rollout，模型数据占比增加
    """
    if epoch < 100:
        return {'rollout_len': 1, 'model_ratio': 0.1}
    elif epoch < 500:
        return {'rollout_len': 3, 'model_ratio': 0.3}
    else:
        return {'rollout_len': 5, 'model_ratio': 0.5}

7. 参考文献

Footnotes

1. Yu et al. “MOPO: Model-based Offline Policy Optimization” NeurIPS 2020 ↩

2. Yu et al. “COMBO: Conservative Offline Model-Based Policy Optimization” NeurIPS 2021 ↩

3. Kidson et al. “The Edge-of-Reach Problem in Offline Model-Based Reinforcement Learning” ICLR 2024 ↩