强化学习调参技巧与实战经验

1
import torch.optim as optim
2

3
# 1. 线性衰减
4
def linear_schedule(initial_lr, final_lr, max_timesteps):
5
    def lr_schedule(timestep):
6
        progress = timestep / max_timesteps
7
        return initial_lr + (final_lr - initial_lr) * progress
8
    return lr_schedule
9

10
# 2. 指数衰减
11
scheduler = optim.lr_scheduler.ExponentialLR(optimizer, gamma=0.99)
12

13
# 3. 余弦退火
14
scheduler = optim.lr_scheduler.CosineAnnealingLR(
15
    optimizer, T_max=total_steps, eta_min=1e-6
16
)
17

18
# 4. 自适应学习率（推荐使用Adam）
19
optimizer = optim.Adam(model.parameters(), lr=3e-4)

1
def find_optimal_lr(model, env, lr_min=1e-6, lr_max=1e-2, num_steps=100):
2
    """学习率查找器"""
3
    lrs = np.logspace(np.log10(lr_min), np.log10(lr_max), num_steps)
4
    losses = []
5

6
    for lr in lrs:
7
        optimizer = optim.Adam(model.parameters(), lr=lr)
8
        # 训练几步并记录损失
9
        loss = train_step(model, optimizer, env)
10
        losses.append(loss)
11

12
        if loss > 2 * min(losses):  # 损失爆炸，提前停止
13
            break
14

15
    # 绘制学习率-损失曲线
16
    import matplotlib.pyplot as plt
17
    plt.semilogx(lrs[:len(losses)], losses)
18
    plt.xlabel('Learning Rate')
19
    plt.ylabel('Loss')
20
    plt.title('Learning Rate Finder')
21
    plt.show()
22

23
    # 选择梯度最大的点（最陡峭的下降）
24
    optimal_idx = np.argmin(np.gradient(losses))
25
    return lrs[optimal_idx]

1
# 1. 线性衰减（最常用）
2
def linear_epsilon_decay(timestep, epsilon_start=1.0, epsilon_end=0.01,
3
                         decay_steps=1000000):
4
    epsilon = epsilon_start - (epsilon_start - epsilon_end) * min(1.0, timestep / decay_steps)
5
    return epsilon
6

7
# 2. 指数衰减
8
def exponential_epsilon_decay(timestep, epsilon_start=1.0, epsilon_end=0.01,
9
                              decay_rate=0.99995):
10
    epsilon = max(epsilon_end, epsilon_start * (decay_rate ** timestep))
11
    return epsilon
12

13
# 3. 分段衰减
14
def piecewise_epsilon(timestep):
15
    if timestep < 500000:
16
        return 1.0
17
    elif timestep < 1000000:
18
        return 0.5
19
    elif timestep < 2000000:
20
        return 0.1
21
    else:
22
        return 0.01
23

24
# 使用示例
25
timestep = 0
26
for episode in range(num_episodes):
27
    epsilon = linear_epsilon_decay(timestep)
28

29
    state = env.reset()
30
    done = False
31
    while not done:
32
        if np.random.rand() < epsilon:
33
            action = env.action_space.sample()  # 探索
34
        else:
35
            action = select_action(state)  # 利用
36

37
        state, reward, done, _ = env.step(action)
38
        timestep += 1

1
class PrioritizedReplayBuffer:
2
    def __init__(self, capacity, alpha=0.6, beta=0.4):
3
        self.capacity = capacity
4
        self.alpha = alpha  # 优先级指数
5
        self.beta = beta    # 重要性采样指数
6
        self.buffer = []
7
        self.priorities = np.zeros(capacity, dtype=np.float32)
8
        self.position = 0
9

10
    def add(self, state, action, reward, next_state, done):
11
        max_priority = self.priorities.max() if self.buffer else 1.0
12

13
        if len(self.buffer) < self.capacity:
14
            self.buffer.append((state, action, reward, next_state, done))
15
        else:
16
            self.buffer[self.position] = (state, action, reward, next_state, done)
17

18
        self.priorities[self.position] = max_priority
19
        self.position = (self.position + 1) % self.capacity
20

21
    def sample(self, batch_size):
22
        if len(self.buffer) == self.capacity:
23
            priorities = self.priorities
24
        else:
25
            priorities = self.priorities[:len(self.buffer)]
26

27
        # 计算采样概率
28
        probabilities = priorities ** self.alpha
29
        probabilities /= probabilities.sum()
30

31
        # 采样
32
        indices = np.random.choice(len(self.buffer), batch_size, p=probabilities)
33
        samples = [self.buffer[idx] for idx in indices]
34

35
        # 重要性采样权重
36
        total = len(self.buffer)
37
        weights = (total * probabilities[indices]) ** (-self.beta)
38
        weights /= weights.max()
39

40
        return samples, indices, weights
41

42
    def update_priorities(self, indices, priorities):
43
        for idx, priority in zip(indices, priorities):
44
            self.priorities[idx] = priority

1
from stable_baselines3 import PPO
2

3
model = PPO(
4
    "MlpPolicy",
5
    env,
6
    learning_rate=3e-4,
7
    n_steps=2048,        # 每次更新采集的步数
8
    batch_size=64,       # 小批量大小
9
    n_epochs=10,         # 训练轮数
10
    gamma=0.99,
11
    gae_lambda=0.95,
12
    clip_range=0.2,      # 裁剪范围
13
    clip_range_vf=None,  # 值函数裁剪（可选）
14
    ent_coef=0.0,        # 熵系数
15
    vf_coef=0.5,         # 值函数系数
16
    max_grad_norm=0.5,   # 梯度裁剪
17
    verbose=1
18
)

1
class SAC:
2
    def __init__(self, action_dim):
3
        # 自动调整熵温度
4
        self.target_entropy = -action_dim  # 启发式目标熵
5
        self.log_alpha = torch.zeros(1, requires_grad=True)
6
        self.alpha_optimizer = optim.Adam([self.log_alpha], lr=3e-4)
7

8
    def update_alpha(self, entropy):
9
        """更新温度参数"""
10
        alpha_loss = -(self.log_alpha * (entropy + self.target_entropy).detach()).mean()
11

12
        self.alpha_optimizer.zero_grad()
13
        alpha_loss.backward()
14
        self.alpha_optimizer.step()
15

16
        alpha = self.log_alpha.exp()
17
        return alpha

1
class RainbowDQN:
2
    def __init__(self):
3
        # 1. Double DQN：减少过估计
4
        self.use_double_dqn = True
5

6
        # 2. Dueling DQN：分离值函数和优势函数
7
        self.use_dueling = True
8

9
        # 3. Prioritized Experience Replay
10
        self.use_per = True
11
        self.per_alpha = 0.6
12
        self.per_beta = 0.4
13

14
        # 4. Multi-step Learning
15
        self.n_step = 3
16

17
        # 5. Distributional RL (C51)
18
        self.use_distributional = True
19
        self.v_min = -10
20
        self.v_max = 10
21
        self.n_atoms = 51
22

23
        # 6. Noisy Networks：替代ε-greedy
24
        self.use_noisy_net = True

1
import torch.nn.utils as nn_utils
2

3
# 方法1：按范数裁剪（最常用）
4
nn_utils.clip_grad_norm_(model.parameters(), max_norm=0.5)
5

6
# 方法2：按值裁剪
7
nn_utils.clip_grad_value_(model.parameters(), clip_value=1.0)
8

9
# 完整训练循环
10
for epoch in range(num_epochs):
11
    loss = compute_loss()
12

13
    optimizer.zero_grad()
14
    loss.backward()
15

16
    # 梯度裁剪
17
    nn_utils.clip_grad_norm_(model.parameters(), max_norm=0.5)
18

19
    optimizer.step()

1
class RewardNormalizer:
2
    def __init__(self, gamma=0.99, epsilon=1e-8):
3
        self.gamma = gamma
4
        self.epsilon = epsilon
5
        self.returns = []
6
        self.mean = 0
7
        self.var = 1
8
        self.count = 0
9

10
    def update(self, reward):
11
        """增量更新统计信息"""
12
        self.returns.append(reward)
13
        self.count += 1
14

15
        if len(self.returns) > 1000:
16
            self.returns.pop(0)
17

18
        # 计算折扣回报的均值和方差
19
        discounted_returns = []
20
        R = 0
21
        for r in reversed(self.returns):
22
            R = r + self.gamma * R
23
            discounted_returns.insert(0, R)
24

25
        self.mean = np.mean(discounted_returns)
26
        self.var = np.var(discounted_returns) + self.epsilon
27

28
    def normalize(self, reward):
29
        """归一化奖励"""
30
        return reward / np.sqrt(self.var)
31

32
# 奖励裁剪
33
def clip_reward(reward, min_value=-10, max_value=10):
34
    return np.clip(reward, min_value, max_value)
35

36
# Atari游戏常用：符号裁剪
37
def sign_clip_reward(reward):
38
    return np.sign(reward)

1
from stable_baselines3.common.vec_env import DummyVecEnv, VecNormalize
2

3
# 创建向量化环境
4
env = DummyVecEnv([lambda: gym.make("LunarLander-v2")])
5

6
# 归一化观测和奖励
7
env = VecNormalize(
8
    env,
9
    norm_obs=True,      # 归一化观测
10
    norm_reward=True,   # 归一化奖励
11
    clip_obs=10.0,      # 裁剪观测
12
    clip_reward=10.0,   # 裁剪奖励
13
    gamma=0.99
14
)
15

16
# 训练
17
model = PPO("MlpPolicy", env)
18
model.learn(total_timesteps=100000)
19

20
# 保存归一化统计信息
21
env.save("vec_normalize.pkl")
22

23
# 测试时加载
24
env = VecNormalize.load("vec_normalize.pkl", env)
25
env.training = False  # 不更新统计信息
26
env.norm_reward = False  # 测试时不归一化奖励

1
import torch.nn as nn
2

3
def orthogonal_init(module, gain=1.0):
4
    """正交初始化"""
5
    if isinstance(module, (nn.Linear, nn.Conv2d)):
6
        nn.init.orthogonal_(module.weight, gain=gain)
7
        if module.bias is not None:
8
            module.bias.data.fill_(0.0)
9

10
# 应用到网络
11
class ActorCritic(nn.Module):
12
    def __init__(self):
13
        super().__init__()
14
        self.actor = nn.Linear(state_dim, action_dim)
15
        self.critic = nn.Linear(state_dim, 1)
16

17
        # 使用正交初始化
18
        self.apply(lambda m: orthogonal_init(m, gain=np.sqrt(2)))
19

20
        # 最后一层使用小增益
21
        orthogonal_init(self.actor, gain=0.01)

1
class WarmupScheduler:
2
    def __init__(self, optimizer, warmup_steps, initial_lr, target_lr):
3
        self.optimizer = optimizer
4
        self.warmup_steps = warmup_steps
5
        self.initial_lr = initial_lr
6
        self.target_lr = target_lr
7
        self.current_step = 0
8

9
    def step(self):
10
        self.current_step += 1
11
        if self.current_step <= self.warmup_steps:
12
            lr = self.initial_lr + (self.target_lr - self.initial_lr) * \
13
                 (self.current_step / self.warmup_steps)
14
            for param_group in self.optimizer.param_groups:
15
                param_group['lr'] = lr
16

17
# 使用示例
18
optimizer = torch.optim.Adam(model.parameters(), lr=1e-6)
19
scheduler = WarmupScheduler(optimizer, warmup_steps=10000,
20
                           initial_lr=1e-6, target_lr=3e-4)
21

22
for step in range(total_steps):
23
    # 训练步骤
24
    optimizer.zero_grad()
25
    loss.backward()
26
    optimizer.step()
27

28
    # 更新学习率
29
    scheduler.step()

1
def compute_gae(rewards, values, dones, gamma=0.99, gae_lambda=0.95):
2
    """
3
    计算广义优势估计
4

5
    参数:
6
        rewards: 奖励序列 [T]
7
        values: 值函数估计 [T+1]
8
        dones: 终止标志 [T]
9
        gamma: 折扣因子
10
        gae_lambda: GAE参数
11

12
    返回:
13
        advantages: 优势估计 [T]
14
        returns: 回报估计 [T]
15
    """
16
    advantages = []
17
    gae = 0
18

19
    for t in reversed(range(len(rewards))):
20
        if t == len(rewards) - 1:
21
            next_value = values[-1]
22
        else:
23
            next_value = values[t + 1]
24

25
        # TD误差
26
        delta = rewards[t] + gamma * next_value * (1 - dones[t]) - values[t]
27

28
        # GAE
29
        gae = delta + gamma * gae_lambda * (1 - dones[t]) * gae
30
        advantages.insert(0, gae)
31

32
    advantages = np.array(advantages)
33
    returns = advantages + values[:-1]
34

35
    return advantages, returns
36

37
# 归一化优势（提高稳定性）
38
advantages = (advantages - advantages.mean()) / (advantages.std() + 1e-8)

1
class NStepReplayBuffer:
2
    def __init__(self, capacity, n_step=3, gamma=0.99):
3
        self.capacity = capacity
4
        self.n_step = n_step
5
        self.gamma = gamma
6
        self.buffer = deque(maxlen=capacity)
7
        self.n_step_buffer = deque(maxlen=n_step)
8

9
    def add(self, state, action, reward, next_state, done):
10
        # 添加到n步缓冲区
11
        self.n_step_buffer.append((state, action, reward, next_state, done))
12

13
        if len(self.n_step_buffer) < self.n_step:
14
            return
15

16
        # 计算n步回报
17
        n_step_reward = 0
18
        for i, (_, _, r, _, _) in enumerate(self.n_step_buffer):
19
            n_step_reward += (self.gamma ** i) * r
20

21
        # 获取n步后的状态
22
        state_0 = self.n_step_buffer[0][0]
23
        action_0 = self.n_step_buffer[0][1]
24
        next_state_n = self.n_step_buffer[-1][3]
25
        done_n = self.n_step_buffer[-1][4]
26

27
        # 存储n步转移
28
        self.buffer.append((state_0, action_0, n_step_reward,
29
                          next_state_n, done_n))
30

31
    def sample(self, batch_size):
32
        indices = np.random.choice(len(self.buffer), batch_size)
33
        return [self.buffer[i] for i in indices]

1
from stable_baselines3.common.vec_env import SubprocVecEnv, DummyVecEnv
2

3
def make_env(env_id, rank, seed=0):
4
    def _init():
5
        env = gym.make(env_id)
6
        env.reset(seed=seed + rank)
7
        return env
8
    return _init
9

10
if __name__ == '__main__':
11
    num_envs = 8  # 并行环境数量
12
    env_id = "CartPole-v1"
13

14
    # 使用多进程（推荐）
15
    env = SubprocVecEnv([make_env(env_id, i) for i in range(num_envs)])
16

17
    # 或使用单进程（调试时）
18
    # env = DummyVecEnv([make_env(env_id, i) for i in range(num_envs)])
19

20
    # 训练
21
    model = PPO("MlpPolicy", env, n_steps=128, batch_size=256)
22
    model.learn(total_timesteps=100000)

1
class CurriculumEnv:
2
    def __init__(self, base_env, difficulty_schedule):
3
        self.env = base_env
4
        self.difficulty_schedule = difficulty_schedule
5
        self.timestep = 0
6

7
    def reset(self):
8
        self.timestep += 1
9
        # 根据进度调整难度
10
        difficulty = self.difficulty_schedule(self.timestep)
11
        self.env.set_difficulty(difficulty)
12
        return self.env.reset()
13

14
    def step(self, action):
15
        return self.env.step(action)
16

17
# 难度调度函数
18
def linear_difficulty_schedule(timestep, max_timestep=1e6):
19
    """线性增加难度"""
20
    return min(1.0, timestep / max_timestep)
21

22
def threshold_difficulty_schedule(timestep, thresholds):
23
    """基于成功率的阶梯式难度"""
24
    for threshold, difficulty in thresholds:
25
        if success_rate > threshold:
26
            return difficulty
27
    return thresholds[0][1]

1
from stable_baselines3.common.evaluation import evaluate_policy
2
import optuna
3

4
def objective(trial):
5
    """Optuna优化目标函数"""
6
    # 定义超参数搜索空间
7
    learning_rate = trial.suggest_loguniform('learning_rate', 1e-5, 1e-3)
8
    gamma = trial.suggest_categorical('gamma', [0.9, 0.95, 0.99, 0.995])
9
    gae_lambda = trial.suggest_categorical('gae_lambda', [0.8, 0.9, 0.95, 0.99])
10
    clip_range = trial.suggest_uniform('clip_range', 0.1, 0.4)
11
    ent_coef = trial.suggest_loguniform('ent_coef', 1e-4, 0.1)
12

13
    # 创建模型
14
    model = PPO(
15
        "MlpPolicy",
16
        env,
17
        learning_rate=learning_rate,
18
        gamma=gamma,
19
        gae_lambda=gae_lambda,
20
        clip_range=clip_range,
21
        ent_coef=ent_coef,
22
        verbose=0
23
    )
24

25
    # 训练
26
    model.learn(total_timesteps=50000)
27

28
    # 评估
29
    mean_reward, _ = evaluate_policy(model, env, n_eval_episodes=20)
30

31
    return mean_reward
32

33
# 运行优化
34
study = optuna.create_study(direction='maximize')
35
study.optimize(objective, n_trials=100, timeout=3600)
36

37
print("最佳超参数:")
38
print(study.best_params)

1
# 在最佳超参数附近进行精细搜索
2
best_lr = study.best_params['learning_rate']
3

4
fine_tune_lrs = [
5
    best_lr * 0.5,
6
    best_lr * 0.75,
7
    best_lr,
8
    best_lr * 1.25,
9
    best_lr * 1.5
10
]
11

12
results = []
13
for lr in fine_tune_lrs:
14
    model = PPO("MlpPolicy", env, learning_rate=lr, **other_best_params)
15
    model.learn(total_timesteps=100000)
16
    reward = evaluate_policy(model, env)
17
    results.append((lr, reward))
18

19
best_lr_fine = max(results, key=lambda x: x[1])[0]

1
from torch.utils.tensorboard import SummaryWriter
2

3
writer = SummaryWriter(log_dir='runs/experiment1')
4

5
for episode in range(num_episodes):
6
    # 训练过程
7
    episode_reward = 0
8
    state = env.reset()
9

10
    while not done:
11
        action = select_action(state)
12
        next_state, reward, done, _ = env.step(action)
13
        episode_reward += reward
14

15
        # 记录损失
16
        loss = update_model()
17
        writer.add_scalar('Loss/policy_loss', loss, global_step)
18

19
        state = next_state
20
        global_step += 1
21

22
    # 记录每回合指标
23
    writer.add_scalar('Reward/episode', episode_reward, episode)
24
    writer.add_scalar('Length/episode', episode_length, episode)
25

26
    # 记录超参数
27
    writer.add_scalar('Hyperparameters/learning_rate', current_lr, episode)
28
    writer.add_scalar('Hyperparameters/epsilon', current_epsilon, episode)
29

30
writer.close()

1
import wandb
2

3
# 初始化
4
wandb.init(
5
    project="rl-tuning",
6
    config={
7
        "learning_rate": 3e-4,
8
        "gamma": 0.99,
9
        "architecture": "MLP",
10
        "environment": "CartPole-v1"
11
    }
12
)
13

14
# 训练过程中记录
15
for step in range(total_steps):
16
    # 训练
17
    loss = train_step()
18
    reward = evaluate()
19

20
    # 记录指标
21
    wandb.log({
22
        "loss": loss,
23
        "reward": reward,
24
        "epsilon": epsilon,
25
        "learning_rate": current_lr
26
    }, step=step)
27

28
# 保存最佳模型
29
wandb.save('best_model.pth')

1
import gym
2
from stable_baselines3 import PPO
3
from stable_baselines3.common.evaluation import evaluate_policy
4

5
# 创建环境
6
env = gym.make("CartPole-v1")
7

8
# 基线配置（默认参数）
9
baseline_model = PPO("MlpPolicy", env, verbose=0)
10
baseline_model.learn(total_timesteps=50000)
11
baseline_reward, _ = evaluate_policy(baseline_model, env, n_eval_episodes=100)
12
print(f"基线奖励: {baseline_reward:.2f}")
13

14
# 调优配置
15
tuned_model = PPO(
16
    "MlpPolicy",
17
    env,
18
    learning_rate=5e-4,       # 提高学习率加速收敛
19
    n_steps=1024,             # 增加采样步数
20
    batch_size=64,
21
    n_epochs=10,
22
    gamma=0.99,
23
    gae_lambda=0.98,
24
    clip_range=0.2,
25
    ent_coef=0.01,            # 增加探索
26
    vf_coef=0.5,
27
    max_grad_norm=0.5,
28
    verbose=0
29
)
30

31
tuned_model.learn(total_timesteps=50000)
32
tuned_reward, _ = evaluate_policy(tuned_model, env, n_eval_episodes=100)
33
print(f"调优后奖励: {tuned_reward:.2f}")
34
print(f"提升: {(tuned_reward - baseline_reward) / baseline_reward * 100:.1f}%")

1
import gym
2
from stable_baselines3 import SAC
3
from stable_baselines3.common.noise import NormalActionNoise
4

5
# 创建连续控制环境
6
env = gym.make("HalfCheetah-v3")
7

8
# 添加动作噪声
9
n_actions = env.action_space.shape[-1]
10
action_noise = NormalActionNoise(
11
    mean=np.zeros(n_actions),
12
    sigma=0.1 * np.ones(n_actions)
13
)
14

15
# SAC配置
16
model = SAC(
17
    "MlpPolicy",
18
    env,
19
    learning_rate=3e-4,
20
    buffer_size=1000000,
21
    learning_starts=10000,      # 预填充回放缓冲区
22
    batch_size=256,
23
    tau=0.005,                  # 软更新系数
24
    gamma=0.99,
25
    train_freq=1,
26
    gradient_steps=1,
27
    action_noise=action_noise,
28
    ent_coef='auto',            # 自动调整熵系数
29
    target_entropy='auto',
30
    verbose=1
31
)
32

33
# 训练
34
model.learn(total_timesteps=1000000)
35

36
# 评估
37
mean_reward, _ = evaluate_policy(model, env, n_eval_episodes=50)
38
print(f"平均奖励: {mean_reward:.2f}")