AI强化学习实战完全教程

强化学习（Reinforcement Learning, RL）是机器学习三大范式之一，它让智能体通过与环境的交互来学习最优行为策略。从AlphaGo击败世界冠军，到ChatGPT通过RLHF对齐人类偏好，强化学习正在深刻改变AI的能力边界。

1. 强化学习概述与核心概念

1.1 马尔可夫决策过程（MDP）

强化学习的数学基础是马尔可夫决策过程，用一个五元组 (S, A, P, R, γ) 描述：

• S：状态空间，环境所有可能状态的集合
• A：动作空间，智能体可执行的所有动作
• P：状态转移概率 P(s'|s, a)，执行动作a后从状态s转移到s'的概率
• R：奖励函数 R(s, a, s')，执行动作后获得的即时奖励
• γ：折扣因子（0 ≤ γ < 1），衡量未来奖励的重要性

核心的马尔可夫性质：未来状态仅依赖于当前状态和动作，与历史无关。

import numpy as np

class SimpleMDP:
    """一个简单的网格世界MDP示例"""
    def __init__(self, grid_size=4, gamma=0.99):
        self.grid_size = grid_size
        self.gamma = gamma
        self.n_states = grid_size * grid_size
        self.n_actions = 4  # 上下左右
        self.goal = (grid_size - 1, grid_size - 1)
        
    def get_next_state(self, state, action):
        row, col = divmod(state, self.grid_size)
        if action == 0:    row = max(0, row - 1)       # 上
        elif action == 1:  row = min(self.grid_size-1, row + 1)  # 下
        elif action == 2:  col = max(0, col - 1)       # 左
        elif action == 3:  col = min(self.grid_size-1, col + 1)  # 右
        return row * self.grid_size + col
    
    def get_reward(self, state):
        return 1.0 if divmod(state, self.grid_size) == self.goal else -0.01

1.2 策略、价值函数与贝尔曼方程

策略 π(a|s) 定义了在状态s下选择动作a的概率分布。

状态价值函数 V^π(s) 表示从状态s出发，遵循策略π所能获得的期望累积回报：

V^π(s) = E_π[Σ(γ^t · r_t) | s_0 = s]

动作价值函数 Q^π(s, a) 表示在状态s执行动作a后，遵循策略π的期望累积回报。

贝尔曼最优方程 描述了最优价值函数的递推关系：

V*(s) = max_a [R(s,a) + γ · Σ P(s'|s,a) · V*(s')]
Q*(s,a) = R(s,a) + γ · Σ P(s'|s,a) · max_a' Q*(s',a')

2. 经典算法详解

2.1 Q-Learning

Q-Learning是一种无模型（model-free）的离策略（off-policy）算法，通过时序差分（TD）方法直接学习最优动作价值函数。

更新公式：

Q(s,a) ← Q(s,a) + α · [r + γ · max_a' Q(s',a') - Q(s,a)]

import numpy as np
from collections import defaultdict

class QLearningAgent:
    def __init__(self, n_actions, alpha=0.1, gamma=0.99, epsilon=1.0, epsilon_decay=0.995):
        self.q_table = defaultdict(lambda: np.zeros(n_actions))
        self.alpha = alpha
        self.gamma = gamma
        self.epsilon = epsilon
        self.epsilon_decay = epsilon_decay
        self.n_actions = n_actions
    
    def select_action(self, state):
        if np.random.random() < self.epsilon:
            return np.random.randint(self.n_actions)
        return int(np.argmax(self.q_table[state]))
    
    def update(self, state, action, reward, next_state, done):
        current_q = self.q_table[state][action]
        if done:
            target = reward
        else:
            target = reward + self.gamma * np.max(self.q_table[next_state])
        self.q_table[state][action] += self.alpha * (target - current_q)
    
    def decay_epsilon(self):
        self.epsilon = max(0.01, self.epsilon * self.epsilon_decay)

# 训练循环
def train_q_learning(env, agent, episodes=1000):
    rewards_history = []
    for ep in range(episodes):
        state, _ = env.reset()
        total_reward = 0
        done = False
        while not done:
            action = agent.select_action(state)
            next_state, reward, terminated, truncated, _ = env.step(action)
            done = terminated or truncated
            agent.update(state, action, reward, next_state, done)
            state = next_state
            total_reward += reward
        agent.decay_epsilon()
        rewards_history.append(total_reward)
    return rewards_history

2.2 SARSA

SARSA是在策略（on-policy）算法，与Q-Learning的关键区别在于：它使用实际执行的下一个动作来计算TD目标，而非最优动作。

class SARSAAgent:
    def __init__(self, n_actions, alpha=0.1, gamma=0.99, epsilon=0.1):
        self.q_table = defaultdict(lambda: np.zeros(n_actions))
        self.alpha = alpha
        self.gamma = gamma
        self.epsilon = epsilon
        self.n_actions = n_actions
    
    def select_action(self, state):
        if np.random.random() < self.epsilon:
            return np.random.randint(self.n_actions)
        return int(np.argmax(self.q_table[state]))
    
    def update(self, state, action, reward, next_state, next_action, done):
        current_q = self.q_table[state][action]
        if done:
            target = reward
        else:
            target = reward + self.gamma * self.q_table[next_state][next_action]
        self.q_table[state][action] += self.alpha * (target - current_q)

# SARSA训练循环（注意：next_action在循环中选取）
def train_sarsa(env, agent, episodes=1000):
    for ep in range(episodes):
        state, _ = env.reset()
        action = agent.select_action(state)
        done = False
        while not done:
            next_state, reward, terminated, truncated, _ = env.step(action)
            done = terminated or truncated
            next_action = agent.select_action(next_state) if not done else 0
            agent.update(state, action, reward, next_state, next_action, done)
            state, action = next_state, next_action

Q-Learning vs SARSA 核心区别：

2.3 DQN（Deep Q-Network）

当状态空间很大时（如图像输入），表格方法不再适用。DQN用深度神经网络近似Q函数，引入了两个关键技术突破：

1. 经验回放（Experience Replay）：打破数据相关性
2. 目标网络（Target Network）：稳定训练

import torch
import torch.nn as nn
import torch.optim as optim
import random
from collections import deque

class DQNetwork(nn.Module):
    def __init__(self, state_dim, action_dim, hidden_dim=128):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, action_dim)
        )
    
    def forward(self, x):
        return self.net(x)

class ReplayBuffer:
    def __init__(self, capacity=100000):
        self.buffer = deque(maxlen=capacity)
    
    def push(self, state, action, reward, next_state, done):
        self.buffer.append((state, action, reward, next_state, done))
    
    def sample(self, batch_size):
        batch = random.sample(self.buffer, batch_size)
        states, actions, rewards, next_states, dones = zip(*batch)
        return (torch.FloatTensor(np.array(states)),
                torch.LongTensor(actions),
                torch.FloatTensor(rewards),
                torch.FloatTensor(np.array(next_states)),
                torch.FloatTensor(dones))
    
    def __len__(self):
        return len(self.buffer)

class DQNAgent:
    def __init__(self, state_dim, action_dim, lr=1e-3, gamma=0.99,
                 epsilon=1.0, epsilon_decay=0.995, target_update=10):
        self.action_dim = action_dim
        self.gamma = gamma
        self.epsilon = epsilon
        self.epsilon_decay = epsilon_decay
        self.target_update = target_update
        
        self.q_net = DQNetwork(state_dim, action_dim)
        self.target_net = DQNetwork(state_dim, action_dim)
        self.target_net.load_state_dict(self.q_net.state_dict())
        self.target_net.eval()
        
        self.optimizer = optim.Adam(self.q_net.parameters(), lr=lr)
        self.buffer = ReplayBuffer()
        self.step_count = 0
    
    def select_action(self, state):
        if random.random() < self.epsilon:
            return random.randint(0, self.action_dim - 1)
        with torch.no_grad():
            state_t = torch.FloatTensor(state).unsqueeze(0)
            return int(self.q_net(state_t).argmax(dim=1).item())
    
    def learn(self, batch_size=64):
        if len(self.buffer) < batch_size:
            return 0.0
        states, actions, rewards, next_states, dones = self.buffer.sample(batch_size)
        
        # 计算当前Q值
        q_values = self.q_net(states).gather(1, actions.unsqueeze(1)).squeeze(1)
        
        # 计算目标Q值（使用目标网络）
        with torch.no_grad():
            next_q = self.target_net(next_states).max(dim=1)[0]
            target_q = rewards + self.gamma * next_q * (1 - dones)
        
        loss = nn.MSELoss()(q_values, target_q)
        self.optimizer.zero_grad()
        loss.backward()
        self.optimizer.step()
        
        self.step_count += 1
        if self.step_count % self.target_update == 0:
            self.target_net.load_state_dict(self.q_net.state_dict())
        
        return loss.item()

DQN的重要改进变体：

• Double DQN：用在线网络选择动作，目标网络评估动作，减少Q值过估计
• Dueling DQN：将Q值分解为状态价值V(s)和优势函数A(s,a)
• Prioritized Experience Replay：按TD误差优先级采样经验

3. 策略梯度方法

3.1 REINFORCE算法

REINFORCE直接参数化策略 π_θ(a|s)，通过蒙特卡洛采样估计策略梯度：

∇_θ J(θ) = E_π[Σ ∇_θ log π_θ(a_t|s_t) · G_t]

其中 G_t 是从时刻t开始的累积回报。

import torch
import torch.nn as nn
import torch.optim as optim
from torch.distributions import Categorical

class PolicyNetwork(nn.Module):
    def __init__(self, state_dim, action_dim, hidden_dim=128):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, action_dim),
            nn.Softmax(dim=-1)
        )
    
    def forward(self, x):
        return self.net(x)

class REINFORCEAgent:
    def __init__(self, state_dim, action_dim, lr=1e-3, gamma=0.99):
        self.policy = PolicyNetwork(state_dim, action_dim)
        self.optimizer = optim.Adam(self.policy.parameters(), lr=lr)
        self.gamma = gamma
        self.log_probs = []
        self.rewards = []
    
    def select_action(self, state):
        state_t = torch.FloatTensor(state).unsqueeze(0)
        probs = self.policy(state_t)
        dist = Categorical(probs)
        action = dist.sample()
        self.log_probs.append(dist.log_prob(action))
        return action.item()
    
    def update(self):
        # 计算折扣累积回报
        returns = []
        G = 0
        for r in reversed(self.rewards):
            G = r + self.gamma * G
            returns.insert(0, G)
        returns = torch.FloatTensor(returns)
        returns = (returns - returns.mean()) / (returns.std() + 1e-8)  # 归一化
        
        # 计算策略梯度损失
        policy_loss = []
        for log_prob, G in zip(self.log_probs, returns):
            policy_loss.append(-log_prob * G)
        
        self.optimizer.zero_grad()
        loss = torch.stack(policy_loss).sum()
        loss.backward()
        self.optimizer.step()
        
        self.log_probs.clear()
        self.rewards.clear()
        return loss.item()

3.2 PPO（Proximal Policy Optimization）

PPO是目前最流行的策略梯度算法，核心思想是限制每次策略更新的幅度，避免灾难性的大步更新。

PPO-Clip目标函数：

L^CLIP(θ) = E[min(r_t(θ) · A_t, clip(r_t(θ), 1-ε, 1+ε) · A_t)]

其中 r_t(θ) = π_θ(a|s) / π_θ_old(a|s) 是新旧策略的概率比。

class PPONetwork(nn.Module):
    """Actor-Critic网络"""
    def __init__(self, state_dim, action_dim, hidden_dim=256):
        super().__init__()
        self.shared = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU()
        )
        self.actor = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, action_dim),
            nn.Softmax(dim=-1)
        )
        self.critic = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, 1)
        )
    
    def forward(self, x):
        shared = self.shared(x)
        return self.actor(shared), self.critic(shared)

class PPOAgent:
    def __init__(self, state_dim, action_dim, lr=3e-4, gamma=0.99,
                 clip_epsilon=0.2, epochs=10, batch_size=64, gae_lambda=0.95):
        self.network = PPONetwork(state_dim, action_dim)
        self.optimizer = optim.Adam(self.network.parameters(), lr=lr)
        self.gamma = gamma
        self.clip_epsilon = clip_epsilon
        self.epochs = epochs
        self.batch_size = batch_size
        self.gae_lambda = gae_lambda
        
        # 存储轨迹
        self.states = []
        self.actions = []
        self.log_probs = []
        self.rewards = []
        self.dones = []
        self.values = []
    
    def select_action(self, state):
        state_t = torch.FloatTensor(state).unsqueeze(0)
        probs, value = self.network(state_t)
        dist = Categorical(probs)
        action = dist.sample()
        self.log_probs.append(dist.log_prob(action))
        self.values.append(value.squeeze())
        return action.item()
    
    def compute_gae(self, next_value):
        """广义优势估计（GAE）"""
        values = self.values + [next_value]
        gae = 0
        advantages = []
        for step in reversed(range(len(self.rewards))):
            delta = self.rewards[step] + self.gamma * values[step + 1] * (1 - self.dones[step]) - values[step]
            gae = delta + self.gamma * self.gae_lambda * (1 - self.dones[step]) * gae
            advantages.insert(0, gae)
        returns = [adv + val for adv, val in zip(advantages, self.values)]
        return advantages, returns
    
    def update(self, next_state):
        # 计算next_value和GAE
        with torch.no_grad():
            next_value = self.network(torch.FloatTensor(next_state).unsqueeze(0))[1].squeeze()
        advantages, returns = self.compute_gae(next_value)
        
        states = torch.FloatTensor(np.array(self.states))
        old_log_probs = torch.stack(self.log_probs).detach()
        advantages = torch.FloatTensor(advantages)
        advantages = (advantages - advantages.mean()) / (advantages.std() + 1e-8)
        returns = torch.FloatTensor(returns)
        
        total_loss = 0
        for _ in range(self.epochs):
            # Mini-batch更新
            indices = np.arange(len(self.states))
            np.random.shuffle(indices)
            for start in range(0, len(self.states), self.batch_size):
                end = start + self.batch_size
                idx = indices[start:end]
                
                probs, values = self.network(states[idx])
                dist = Categorical(probs)
                new_log_probs = dist.log_prob(torch.LongTensor(np.array(self.actions))[idx])
                
                ratio = (new_log_probs - old_log_probs[idx]).exp()
                surr1 = ratio * advantages[idx]
                surr2 = torch.clamp(ratio, 1 - self.clip_epsilon, 1 + self.clip_epsilon) * advantages[idx]
                
                actor_loss = -torch.min(surr1, surr2).mean()
                critic_loss = nn.MSELoss()(values.squeeze(), returns[idx])
                entropy = dist.entropy().mean()
                
                loss = actor_loss + 0.5 * critic_loss - 0.01 * entropy
                self.optimizer.zero_grad()
                loss.backward()
                nn.utils.clip_grad_norm_(self.network.parameters(), 0.5)
                self.optimizer.step()
                total_loss += loss.item()
        
        # 清空轨迹
        self.states.clear()
        self.actions.clear()
        self.log_probs.clear()
        self.rewards.clear()
        self.dones.clear()
        self.values.clear()
        
        return total_loss

3.3 A3C（Asynchronous Advantage Actor-Critic）

A3C通过多个并行的worker异步更新共享的全局网络，大幅提升训练效率。核心思想是用多个独立的环境实例产生多样化的经验，减少数据相关性。

A3C的关键实现要点：

• 每个worker维护独立的环境和本地网络副本
• Worker定期将梯度推送到全局网络
• 全局网络异步应用梯度更新

在实践中，PPO + 同步向量化环境（如Stable-Baselines3的实现）通常比A3C更容易调参且效果相当。

4. 模型基强化学习（Model-Based RL）

4.1 核心思想

与无模型方法直接从交互数据学习策略不同，模型基方法先学习环境的动态模型（转移函数和奖励函数），再利用学到的模型进行规划或策略优化。样本效率通常高出10-100倍。

4.2 MBPO（Model-Based Policy Optimization）

MBPO的核心思路：用学到的环境模型生成"虚拟"轨迹来增强真实数据，同时用短horizon的模型预测避免误差累积。

import torch.nn as nn

class EnsembleDynamicsModel:
    """集成动力学模型，用多个模型减少过拟合"""
    def __init__(self, state_dim, action_dim, hidden_dim=200, n_ensemble=5):
        self.models = []
        for _ in range(n_ensemble):
            model = nn.Sequential(
                nn.Linear(state_dim + action_dim, hidden_dim),
                nn.ReLU(),
                nn.Linear(hidden_dim, hidden_dim),
                nn.ReLU(),
                nn.Linear(hidden_dim, state_dim + 1)  # 预测delta_state和reward
            )
            self.models.append(model)
    
    def predict(self, state, action, model_idx=None):
        """使用集成模型的某个或随机模型进行预测"""
        if model_idx is None:
            model_idx = np.random.randint(len(self.models))
        x = torch.cat([state, action], dim=-1)
        output = self.models[model_idx](x)
        delta_state = output[..., :-1]
        reward = output[..., -1]
        next_state = state + delta_state  # 残差预测
        return next_state, reward

class MBPOAgent:
    def __init__(self, state_dim, action_dim):
        self.dynamics = EnsembleDynamicsModel(state_dim, action_dim)
        self.policy = PPONetwork(state_dim, action_dim)  # 使用PPO作为策略优化器
        self.real_buffer = ReplayBuffer(capacity=100000)
        self.model_buffer = ReplayBuffer(capacity=500000)
    
    def train_dynamics(self, epochs=50):
        """用真实数据训练环境模型"""
        # 训练集成模型...
        pass
    
    def generate_virtual_data(self, rollout_horizon=5, n_rollouts=1000):
        """用环境模型生成虚拟轨迹"""
        real_samples = self.real_buffer.sample(n_rollouts)
        states = real_samples[0]
        for step in range(rollout_horizon):
            actions = self.policy.select_action(states)  # 使用当前策略采样动作
            next_states, rewards = self.dynamics.predict(states, actions)
            # 存入模型buffer
            for s, a, r, ns in zip(states, actions, rewards, next_states):
                self.model_buffer.push(s, a, r, ns, False)
            states = next_states

4.3 Dreamer

Dreamer在潜在空间中学习世界模型（World Model），然后在"梦境"中直接进行策略学习。其架构包含三个核心组件：

1. 表示模型：将观测编码为潜在状态
2. 转移模型：在潜在空间中预测下一状态
3. 奖励模型：预测潜在状态对应的奖励

Dreamer v3已在多个基准测试中取得SOTA结果，特别适合视觉控制任务。

5. 多智能体强化学习（MARL）

5.1 挑战与分类

多智能体环境的核心挑战：非平稳性——每个智能体的策略在变化，导致其他智能体面对的环境也在不断变化。

MARL的常见设置：

• 完全合作：所有智能体共享全局奖励（如团队协作）
• 完全竞争：零和博弈（如棋类对弈）
• 混合动机：既有合作又有竞争（如足球比赛）

5.2 独立学习 vs 联合学习

Independent Q-Learning（IQL）：每个智能体独立学习自己的Q函数，将其他智能体视为环境的一部分。简单但受限于非平稳性。

QMIX：一种流行的值分解方法，将全局Q值分解为各智能体局部Q值的单调混合：

class QMIXNetwork(nn.Module):
    """QMIX: 将各agent的Q值通过单调混合网络组合"""
    def __init__(self, n_agents, state_dim, hidden_dim=64):
        super().__init__()
        self.n_agents = n_agents
        # 超网络：用state生成混合权重
        self.hyper_w1 = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, n_agents * hidden_dim)
        )
        self.hyper_b1 = nn.Linear(state_dim, hidden_dim)
        self.hyper_w2 = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim)
        )
        self.hyper_b2 = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, 1)
        )
    
    def forward(self, agent_qs, state):
        """
        agent_qs: [batch, n_agents] 各agent的Q值
        state: [batch, state_dim] 全局状态
        """
        batch_size = agent_qs.size(0)
        
        # 第一层混合
        w1 = torch.abs(self.hyper_w1(state))  # 保证非负 → 单调性
        w1 = w1.view(batch_size, self.n_agents, -1)
        b1 = self.hyper_b1(state).unsqueeze(1)
        hidden = torch.bmm(agent_qs.unsqueeze(1), w1) + b1
        hidden = torch.relu(hidden)
        
        # 第二层
        w2 = torch.abs(self.hyper_w2(state))
        w2 = w2.view(batch_size, -1, 1)
        b2 = self.hyper_b2(state).unsqueeze(1)
        q_total = torch.bmm(hidden, w2) + b2
        
        return q_total.squeeze(-1)

5.3 MAPPO

MAPPO将PPO扩展到多智能体场景，每个智能体有独立的Actor，共享一个全局Critic。在星际争霸II（SMAC）等基准测试中表现优异。

6. 离线强化学习

6.1 动机与挑战

离线RL（Offline RL）仅从固定的历史数据集中学习策略，无需与环境交互。适用于在线交互成本高或风险大的场景（如医疗、自动驾驶）。

核心挑战：分布偏移（Distribution Shift）——学习到的策略可能选择数据集中未覆盖的状态-动作对，导致价值估计严重偏离。

6.2 CQL（Conservative Q-Learning）

CQL通过在Q函数训练中添加保守正则化项，防止对未见过的动作给予过高的Q值估计：

class CQLAgent:
    def __init__(self, state_dim, action_dim, alpha=1.0):
        self.q_net = DQNetwork(state_dim, action_dim)
        self.target_net = DQNetwork(state_dim, action_dim)
        self.alpha = alpha  # CQL正则化系数
    
    def compute_loss(self, states, actions, rewards, next_states, dones):
        # 标准TD损失
        q_values = self.q_net(states).gather(1, actions.unsqueeze(1)).squeeze(1)
        with torch.no_grad():
            next_q = self.target_net(next_states).max(dim=1)[0]
            target_q = rewards + 0.99 * next_q * (1 - dones)
        td_loss = nn.MSELoss()(q_values, target_q)
        
        # CQL正则化：惩罚数据集中未出现的动作的Q值
        all_q = self.q_net(states)
        cql_penalty = (torch.logsumexp(all_q, dim=1).mean() - q_values.mean())
        
        return td_loss + self.alpha * cql_penalty

其他重要的离线RL算法包括：

• BCQ：用行为克隆约束策略，只选择数据分布内支持的动作
• Decision Transformer：将RL重新建模为序列预测问题，利用Transformer架构

7. 强化学习框架对比

7.1 Stable-Baselines3（SB3）

最易上手的RL库，基于PyTorch，提供开箱即用的算法实现。

import gymnasium as gym
from stable_baselines3 import PPO
from stable_baselines3.common.vec_env import DummyVecEnv

# 创建向量化环境
env = DummyVecEnv([lambda: gym.make("CartPole-v1")])

# 创建PPO智能体
model = PPO(
    "MlpPolicy",
    env,
    learning_rate=3e-4,
    n_steps=2048,
    batch_size=64,
    n_epochs=10,
    gamma=0.99,
    gae_lambda=0.95,
    clip_range=0.2,
    verbose=1
)

# 训练
model.learn(total_timesteps=100_000)

# 评估
eval_env = gym.make("CartPole-v1")
obs, _ = eval_env.reset()
for _ in range(1000):
    action, _ = model.predict(obs, deterministic=True)
    obs, reward, terminated, truncated, _ = eval_env.step(action)
    if terminated or truncated:
        obs, _ = eval_env.reset()

7.2 Ray RLlib

适合分布式训练，支持大规模并行和多GPU。适合生产环境和大规模实验。

import ray
from ray.rllib.algorithms.ppo import PPOConfig

ray.init()

config = (
    PPOConfig()
    .environment("CartPole-v1")
    .training(
        lr=3e-4,
        train_batch_size=4000,
        sgd_minibatch_size=64,
        num_sgd_iter=10,
    )
    .rollouts(num_rollout_workers=4)
    .framework("torch")
)

algo = config.build()
for i in range(50):
    result = algo.train()
    print(f"Iter {i}: reward = {result['episode_reward_mean']:.2f}")

7.3 CleanRL

采用单文件实现哲学，每个算法一个完整、可读的Python文件。非常适合学习算法原理和快速原型开发。

框架选择建议：

• 学习原理 → CleanRL
• 快速实验 → Stable-Baselines3
• 生产部署/大规模训练 → Ray RLlib

8. 环境设计与奖励工程

8.1 奖励工程的核心原则

奖励函数是RL成功的关键。设计不当的奖励会导致奖励黑客（Reward Hacking）——智能体找到意想不到的方式来获取高奖励，而非完成预期任务。

# 反例：奖励设计不当导致的hack
class BadRewardEnv:
    def step(self, action):
        # 设计目标：让机器人走到目标位置
        # 错误奖励：只奖励移动距离
        distance_moved = abs(self.robot_pos - self.prev_pos)
        reward = distance_moved  # 机器人学会原地抖动！
        return reward

# 正确的奖励设计
class GoodRewardEnv:
    def step(self, action):
        # 多维度奖励组合
        distance_to_goal = np.linalg.norm(self.robot_pos - self.goal)
        reward = 0.0
        reward += -distance_to_goal * 0.1          # 距离惩罚
        reward += 10.0 if distance_to_goal < 0.5 else 0.0  # 到达奖励
        reward += -0.01  # 时间惩罚（鼓励快速到达）
        reward += -1.0 if self.collided else 0.0    # 碰撞惩罚
        return reward

8.2 奖励塑形（Reward Shaping）

通过添加启发式的中间奖励来引导学习，但需保持策略不变性：

def shaped_reward(state, next_state, reward, goal, gamma=0.99):
    """
    基于势函数的奖励塑形，保证最优策略不变
    F(s, s') = γΦ(s') - Φ(s)
    """
    def potential(s):
        return -np.linalg.norm(s - goal)  # 势函数：负距离
    
    shaping_bonus = gamma * potential(next_state) - potential(state)
    return reward + shaping_bonus

9. 从人类反馈中学习（RLHF）

9.1 RLHF的核心流程

RLHF是将大语言模型与人类偏好对齐的关键技术，其流程分为三个阶段：

1. 监督微调（SFT）：用高质量数据微调预训练模型
2. 训练奖励模型（RM）：从人类对模型输出的偏好排序中学习奖励函数
3. RL优化：用PPO等算法优化策略模型，使其输出最大化奖励模型评分

import torch
import torch.nn as nn

class RewardModel(nn.Module):
    """从人类偏好中学习奖励函数"""
    def __init__(self, base_model_dim):
        super().__init__()
        self.backbone = nn.Linear(base_model_dim, 256)
        self.reward_head = nn.Linear(256, 1)
    
    def forward(self, hidden_states):
        features = torch.relu(self.backbone(hidden_states))
        return self.reward_head(features).squeeze(-1)
    
    def compute_preference_loss(self, chosen_hidden, rejected_hidden):
        """Bradley-Terry偏好模型"""
        chosen_reward = self.forward(chosen_hidden)
        rejected_reward = self.forward(rejected_hidden)
        # 最大化 chosen > rejected 的对数似然
        loss = -torch.log(torch.sigmoid(chosen_reward - rejected_reward)).mean()
        return loss

class PPOForRLHF:
    """RLHF中PPO的简化实现"""
    def __init__(self, policy_model, ref_model, reward_model, kl_coeff=0.1):
        self.policy = policy_model
        self.ref = ref_model    # 参考模型（冻结的SFT模型）
        self.reward_model = reward_model
        self.kl_coeff = kl_coeff
    
    def compute_reward(self, prompt, response, hidden_states):
        # 奖励模型评分
        rm_score = self.reward_model(hidden_states)
        # KL散度惩罚（防止偏离参考模型太远）
        with torch.no_grad():
            ref_logprobs = self.ref.get_log_probs(prompt, response)
        policy_logprobs = self.policy.get_log_probs(prompt, response)
        kl_penalty = (policy_logprobs - ref_logprobs).sum(dim=-1)
        # 最终奖励 = RM评分 - KL惩罚
        return rm_score - self.kl_coeff * kl_penalty

9.2 DPO（Direct Preference Optimization）

DPO绕过了显式训练奖励模型和RL训练的步骤，直接从偏好数据中优化策略：

def dpo_loss(policy_chosen_logps, policy_rejected_logps,
             ref_chosen_logps, ref_rejected_logps, beta=0.1):
    """DPO损失函数"""
    chosen_logratios = policy_chosen_logps - ref_chosen_logps
    rejected_logratios = policy_rejected_logps - ref_rejected_logps
    logits = beta * (chosen_logratios - rejected_logratios)
    loss = -torch.log(torch.sigmoid(logits)).mean()
    return loss

10. 实战案例：游戏AI训练

10.1 Atari游戏（DQN）

import gymnasium as gym
from stable_baselines3 import DQN
from stable_baselines3.common.atari_wrappers import AtariWrapper

def train_atari(game="BreakoutNoFrameskip-v4", total_steps=2_000_000):
    env = gym.make(game, render_mode=None)
    
    model = DQN(
        "CnnPolicy",
        env,
        learning_rate=1e-4,
        buffer_size=1_000_000,
        learning_starts=50000,
        batch_size=32,
        tau=1.0,
        gamma=0.99,
        train_freq=4,
        target_update_interval=1000,
        exploration_fraction=0.1,
        exploration_final_eps=0.01,
        verbose=1,
        tensorboard_log="./atari_logs/"
    )
    
    model.learn(total_timesteps=total_steps, progress_bar=True)
    model.save("dqn_breakout")
    return model

10.2 MuJoCo连续控制

from stable_baselines3 import SAC  # 连续动作空间推荐SAC

def train_mujoco(env_name="HalfCheetah-v4", total_steps=1_000_000):
    env = gym.make(env_name)
    
    model = SAC(
        "MlpPolicy",
        env,
        learning_rate=3e-4,
        buffer_size=1_000_000,
        batch_size=256,
        tau=0.005,
        gamma=0.99,
        train_freq=1,
        gradient_steps=1,
        learning_starts=10000,
        verbose=1,
        tensorboard_log="./mujoco_logs/"
    )
    
    model.learn(total_timesteps=total_steps, progress_bar=True)
    return model

11. 工业应用

11.1 推荐系统

RL用于推荐系统的优势在于优化长期用户满意度，而非仅优化单次点击率。

class RecoEnv:
    """简化的推荐环境"""
    def __init__(self, n_items, user_feature_dim):
        self.n_items = n_items
        self.user_features = None  # 用户特征
    
    def reset(self):
        self.user_features = np.random.randn(user_feature_dim)
        return self.user_features
    
    def step(self, item_id):
        # 模拟用户反馈
        relevance = self._compute_relevance(item_id)
        click = np.random.random() < self._sigmoid(relevance)
        diversity_bonus = self._compute_diversity(item_id)
        
        reward = click * 1.0 + diversity_bonus * 0.2 - 0.01  # 长期价值设计
        # 返回下一个推荐请求的状态
        next_state = self._update_user_state(item_id, click)
        return next_state, reward, False, False, {}

11.2 广告竞价

在实时竞价（RTB）场景中，RL智能体学习最优出价策略，在预算约束下最大化转化量：

class BiddingAgent:
    def __init__(self, budget, n_price_buckets=100):
        self.budget = budget
        self.remaining_budget = budget
        self.q_table = {}  # 状态: (剩余预算比例, 剩余时间比例) → 出价
    
    def get_state(self, time_remaining, total_time):
        budget_ratio = round(self.remaining_budget / self.budget, 1)
        time_ratio = round(time_remaining / total_time, 1)
        return (budget_ratio, time_ratio)
    
    def bid(self, state, floor_price=0.1):
        # 基于Q表选择出价倍率
        if state not in self.q_table:
            self.q_table[state] = np.ones(10) / 10
        action = np.argmax(self.q_table[state])
        bid_price = floor_price * (1 + action * 0.5)
        return min(bid_price, self.remaining_budget)

11.3 资源调度

RL在云计算资源调度中的应用：根据当前负载状态（CPU/内存/网络使用率、队列长度等），动态决定资源分配策略，优化延迟和成本。

class ResourceScheduler:
    def __init__(self, n_servers, n_task_types):
        self.n_servers = n_servers
        # 状态: [各服务器负载, 队列长度, 任务类型分布]
        state_dim = n_servers * 3 + n_task_types + 1
        self.agent = PPOAgent(state_dim, n_servers)
    
    def schedule(self, task, server_loads, queue_info):
        state = np.concatenate([
            server_loads.flatten(),
            queue_info,
            task.feature_vector
        ])
        server_id = self.agent.select_action(state)
        return server_id

总结

强化学习的核心工作流程可以概括为：

1. 明确问题形式化：定义状态空间、动作空间、奖励函数
2. 选择合适的算法范式：基于模型/无模型，基于值/策略梯度/Actor-Critic
3. 从简单环境验证：在标准benchmark（CartPole/MountainCar）上验证实现
4. 逐步扩展复杂度：增加环境复杂度、引入分布式训练
5. 奖励工程迭代：持续观察和调整奖励函数

实用建议：

• 新项目优先尝试PPO/SAC，它们是各自领域的通用基线
• 样本效率重要时考虑模型基方法（MBPO/Dreamer）
• 离线场景使用CQL或Decision Transformer
• 生产环境注意奖励函数的鲁棒性和安全性
• RLHF/DPO是大语言模型对齐的必经之路

强化学习仍在快速发展中，从单智能体到多智能体，从模拟到真实世界，从纯数值到语言模型——每一次突破都在拓展AI解决问题的边界。