把Q-Learning换到PPO时踩过的坑

强化学习 Q-Learning PPO PyTorch 算法实践
作者: 醉月思 发布日期: 2021-02-15 分类: 学习笔记

Agent在环境里试错，跟人在公司里摸鱼本质上是一回事——都在寻找最优策略，只不过一个靠探索利用，一个靠踩红线与打擦边球。

强化学习这一路走来，我从Q-Learning的表格查法走到PPO的神经网络策略，踩了不少坑，也认清了一个现实：没有银弹，只有合适场景的合适方法。

为什么折腾强化学习

起因是个游戏：Gym里的CartPole，让小车平衡杆子。看起来简单，但写规则太麻烦。这类场景恰好是强化学习的用武之地——你定义环境、奖惩机制，让模型自己学策略。

中间又试了几次：让Agent学会走迷宫、优化资源调度、甚至在某个游戏里自动挂机。有些成功了，有些失败了。这个过程里最大的感受是：强化学习不像监督学习那样可以线性堆算力，更多是跟环境设计、超参数调整和训练稳定性较劲。

先从Q-Learning开始

Q-Learning是最早的强化学习算法之一，原理粗暴直接：用一张表格记下每个状态和动作的价值，然后更新。

import numpy as np
import gymnasium as gym

class QLearningAgent:
    def __init__(self, state_size, action_size, learning_rate=0.1, discount_factor=0.95, epsilon=0.1):
        self.q_table = np.zeros((state_size, action_size))
        self.learning_rate = learning_rate
        self.discount_factor = discount_factor
        self.epsilon = epsilon
        self.action_size = action_size

    def act(self, state):
        if np.random.rand() < self.epsilon:
            return np.random.randint(self.action_size)
        return np.argmax(self.q_table[state])

    def learn(self, state, action, reward, next_state):
        best_next_action = np.argmax(self.q_table[next_state])
        td_target = reward + self.discount_factor * self.q_table[next_state][best_next_action]
        td_error = td_target - self.q_table[state][action]
        self.q_table[state][action] += self.learning_rate * td_error

# 训练
env = gym.make('FrozenLake-v1', is_slippery=False)
state_size = env.observation_space.n
action_size = env.action_space.n
agent = QLearningAgent(state_size, action_size)

episodes = 1000
for episode in range(episodes):
    state, _ = env.reset()
    done = False
    while not done:
        action = agent.act(state)
        next_state, reward, done, truncated, info = env.step(action)
        agent.learn(state, action, reward, next_state)
        state = next_state

这套东西在FrozenLake这种离散小环境里效果还行，但问题很快就来了：状态空间一变大，表格就爆炸。CartPole还好，但换个复杂点的情况，比如连续状态空间，直接就炸了。

训练过程中遇到的第一个坑：exploration和exploitation的平衡。epsilon太大全是瞎试，太小容易陷入局部最优。常见做法是epsilon decay，从1.0慢慢降到0.01，但衰减节奏得配合环境复杂度。

另一个坑是reward design。CartPole里给-1到1的即时奖励没问题，但复杂任务里需要稀疏奖励，这时候模型根本学不到东西。解决办法通常是reward shaping，比如给中间步骤一些引导信号，但引导太多又会改了原始目标。

从表格到神经网络：DQN

状态空间一大，Q-Learning的表格就顶不住了。DQN用神经网络拟合Q函数，把状态输入进去，输出每个动作的价值。

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

class DQNNetwork(nn.Module):
    def __init__(self, state_size, action_size):
        super(DQNNetwork, self).__init__()
        self.fc1 = nn.Linear(state_size, 64)
        self.fc2 = nn.Linear(64, 64)
        self.fc3 = nn.Linear(64, action_size)

    def forward(self, x):
        x = torch.relu(self.fc1(x))
        x = torch.relu(self.fc2(x))
        x = self.fc3(x)
        return x

class DQNAgent:
    def __init__(self, state_size, action_size):
        self.state_size = state_size
        self.action_size = action_size
        self.memory = deque(maxlen=10000)
        self.gamma = 0.95
        self.epsilon = 1.0
        self.epsilon_min = 0.01
        self.epsilon_decay = 0.995
        self.learning_rate = 0.001

        self.model = DQNNetwork(state_size, action_size)
        self.target_model = DQNNetwork(state_size, action_size)
        self.update_target_model()
        self.optimizer = optim.Adam(self.model.parameters(), lr=self.learning_rate)

    def update_target_model(self):
        self.target_model.load_state_dict(self.model.state_dict())

    def remember(self, state, action, reward, next_state, done):
        self.memory.append((state, action, reward, next_state, done))

    def act(self, state):
        if np.random.rand() <= self.epsilon:
            return random.randrange(self.action_size)
        act_values = self.model(torch.FloatTensor(state))
        return np.argmax(act_values.data.numpy())

    def replay(self, batch_size):
        if len(self.memory) < batch_size:
            return
        minibatch = random.sample(self.memory, batch_size)
        for state, action, reward, next_state, done in minibatch:
            target = self.model(torch.FloatTensor(state))
            if done:
                target[0][action] = reward
            else:
                t = self.target_model(torch.FloatTensor(next_state)).max()
                target[0][action] = reward + self.gamma * t.item()

            pred = self.model(torch.FloatTensor(state))
            loss = nn.MSELoss()(pred, target)
            self.optimizer.zero_grad()
            loss.backward()
            self.optimizer.step()

        if self.epsilon > self.epsilon_min:
            self.epsilon *= self.epsilon_decay

DQN比Q-Learning强在能处理连续状态空间，但新坑又来了：训练不稳定、收敛慢。解决办法有几个：

  1. Experience Replay：存一批样本，训练时随机取，打破样本间的相关性
  2. Target Network：用一个目标网络计算Q值，定期同步，避免自我训练产生的发散
  3. Double DQN：用主网络选动作，目标网络算Q值，减少Q值高估
graph LR A[状态输入] --> B[主网络] B --> C[动作选择] B --> D[Q值计算] E[目标网络] --> D D --> F[目标Q值] C --> G[经验回放缓冲区] G --> H[损失计算] F --> H H --> I[网络更新] I --> E

这张图想解释DQN里主网络和目标网络的关系：主网络负责做决策和梯度更新，目标网络提供稳定的Q值估计，避免训练过程中的震荡。实际用的时候，每隔几步就把主网络的权重拷给目标网络。

DQN在CartPole上能收敛，但学习曲线很抖。有时候训练几千步突然垮了，得重新来。这种不稳定性是后续很多改进算法要解决的问题。

Policy Gradient：直接学策略

DQN还在学Q函数，然后从Q函数推导策略。Policy Gradient更直接：学一个策略网络，输入状态，输出动作概率。

class PolicyGradientNetwork(nn.Module):
    def __init__(self, state_size, action_size):
        super(PolicyGradientNetwork, self).__init__()
        self.fc1 = nn.Linear(state_size, 128)
        self.fc2 = nn.Linear(128, 128)
        self.fc3 = nn.Linear(128, action_size)

    def forward(self, x):
        x = torch.relu(self.fc1(x))
        x = torch.relu(self.fc2(x))
        x = torch.softmax(self.fc3(x), dim=-1)
        return x

class PolicyGradientAgent:
    def __init__(self, state_size, action_size):
        self.policy = PolicyGradientNetwork(state_size, action_size)
        self.optimizer = optim.Adam(self.policy.parameters(), lr=0.001)
        self.gamma = 0.99
        self.episodes = []

    def act(self, state):
        probs = self.policy(torch.FloatTensor(state))
        action_dist = torch.distributions.Categorical(probs)
        action = action_dist.sample()
        log_prob = action_dist.log_prob(action)
        return action.item(), log_prob

    def store_transition(self, reward, log_prob):
        self.episodes.append((reward, log_prob))

    def update_policy(self):
        rewards = [e[0] for e in self.episodes]
        log_probs = [e[1] for e in self.episodes]

        # 计算discounted rewards
        returns = []
        R = 0
        for r in reversed(rewards):
            R = r + self.gamma * R
            returns.insert(0, R)
        returns = torch.FloatTensor(returns)
        returns = (returns - returns.mean()) / (returns.std() + 1e-8)

        policy_loss = []
        for log_prob, R in zip(log_probs, returns):
            policy_loss.append(-log_prob * R)

        self.optimizer.zero_grad()
        policy_loss = torch.stack(policy_loss).sum()
        policy_loss.backward()
        self.optimizer.step()

        self.episodes = []

Policy Gradient的好处是能处理连续动作空间，比如控制机器人的扭矩。但问题是方差太大，训练不稳定，样本效率低。一个episode里的运气好坏直接影响梯度方向。

实际训练Policy Gradient的时候，常常遇到reward scale的问题。如果reward太大或者太小，梯度会异常。通常做法是把reward归一化，或者用advantage代替原始reward。

Actor-Critic：结合值函数和策略

Policy Gradient方差大，Actor-Critic的思路是：用Critic网络估计状态价值，减少方差。Actor网络负责输出策略，Critic网络负责估计状态值。

class Actor(nn.Module):
    def __init__(self, state_size, action_size):
        super(Actor, self).__init__()
        self.fc1 = nn.Linear(state_size, 128)
        self.fc2 = nn.Linear(128, 128)
        self.fc3 = nn.Linear(128, action_size)

    def forward(self, x):
        x = torch.relu(self.fc1(x))
        x = torch.relu(self.fc2(x))
        x = torch.softmax(self.fc3(x), dim=-1)
        return x

class Critic(nn.Module):
    def __init__(self, state_size):
        super(Critic, self).__init__()
        self.fc1 = nn.Linear(state_size, 128)
        self.fc2 = nn.Linear(128, 128)
        self.fc3 = nn.Linear(128, 1)

    def forward(self, x):
        x = torch.relu(self.fc1(x))
        x = torch.relu(self.fc2(x))
        x = self.fc3(x)
        return x

class ActorCriticAgent:
    def __init__(self, state_size, action_size):
        self.actor = Actor(state_size, action_size)
        self.critic = Critic(state_size)
        self.actor_optimizer = optim.Adam(self.actor.parameters(), lr=0.001)
        self.critic_optimizer = optim.Adam(self.critic.parameters(), lr=0.001)
        self.gamma = 0.99

    def act(self, state):
        probs = self.actor(torch.FloatTensor(state))
        action_dist = torch.distributions.Categorical(probs)
        action = action_dist.sample()
        return action.item()

    def train(self, state, action, reward, next_state, done):
        state = torch.FloatTensor(state)
        next_state = torch.FloatTensor(next_state)
        action = torch.tensor(action)
        reward = torch.tensor(reward, dtype=torch.float)

        # Critic update
        value = self.critic(state)
        next_value = self.critic(next_state)
        target = reward + self.gamma * next_value * (1 - done)
        critic_loss = nn.MSELoss()(value, target)

        self.critic_optimizer.zero_grad()
        critic_loss.backward()
        self.critic_optimizer.step()

        # Actor update
        probs = self.actor(state)
        action_dist = torch.distributions.Categorical(probs)
        log_prob = action_dist.log_prob(action)
        advantage = target - value.detach()
        actor_loss = -log_prob * advantage

        self.actor_optimizer.zero_grad()
        actor_loss.backward()
        self.actor_optimizer.step()

Actor-Critic比Policy Gradient稳定一些，但还是有坑：如果Critic训练不好，会误导Actor。实践中常加上entropy regularization，防止策略过早收敛到确定性动作。

这个阶段的训练曲线平滑了一些，但收敛还是很慢。跟DQN比，样本效率是明显短板。

PPO：现代算法的收敛

PPO（Proximal Policy Optimization）是目前最常用的算法之一，关键是在policy update时加上trust region限制，避免策略更新过大导致训练崩掉。

class PPOAgent:
    def __init__(self, state_size, action_size, clip_epsilon=0.2):
        self.actor = Actor(state_size, action_size)
        self.critic = Critic(state_size)
        self.actor_optimizer = optim.Adam(self.actor.parameters(), lr=3e-4)
        self.critic_optimizer = optim.Adam(self.critic.parameters(), lr=1e-3)
        self.gamma = 0.99
        self.clip_epsilon = clip_epsilon
        self.epochs = 10
        self.batch_size = 64

        self.memory = []

    def act(self, state):
        state = torch.FloatTensor(state)
        probs = self.actor(state)
        action_dist = torch.distributions.Categorical(probs)
        action = action_dist.sample()
        log_prob = action_dist.log_prob(action)
        value = self.critic(state)

        return action.item(), log_prob.item(), value.item()

    def store(self, transition):
        self.memory.append(transition)

    def update(self):
        states = torch.FloatTensor([t[0] for t in self.memory])
        actions = torch.LongTensor([t[1] for t in self.memory])
        old_log_probs = torch.FloatTensor([t[2] for t in self.memory])
        rewards = torch.FloatTensor([t[3] for t in self.memory])
        next_states = torch.FloatTensor([t[4] for t in self.memory])
        dones = torch.FloatTensor([t[5] for t in self.memory])
        old_values = torch.FloatTensor([t[6] for t in self.memory])

        # 计算advantages
        with torch.no_grad():
            next_values = self.critic(next_states).squeeze()
            returns = rewards + self.gamma * next_values * (1 - dones)
            advantages = returns - old_values
            advantages = (advantages - advantages.mean()) / (advantages.std() + 1e-8)

        # 多次更新
        for _ in range(self.epochs):
            indices = torch.randperm(len(self.memory))
            for start in range(0, len(self.memory), self.batch_size):
                end = start + self.batch_size
                batch_indices = indices[start:end]

                batch_states = states[batch_indices]
                batch_actions = actions[batch_indices]
                batch_old_log_probs = old_log_probs[batch_indices]
                batch_advantages = advantages[batch_indices]
                batch_returns = returns[batch_indices]

                # Actor update with clipping
                probs = self.actor(batch_states)
                action_dist = torch.distributions.Categorical(probs)
                new_log_probs = action_dist.log_prob(batch_actions)
                ratio = torch.exp(new_log_probs - batch_old_log_probs)

                surr1 = ratio * batch_advantages
                surr2 = torch.clamp(ratio, 1 - self.clip_epsilon, 1 + self.clip_epsilon) * batch_advantages
                actor_loss = -torch.min(surr1, surr2).mean()

                # Critic update
                values = self.critic(batch_states).squeeze()
                critic_loss = nn.MSELoss()(values, batch_returns)

                self.actor_optimizer.zero_grad()
                actor_loss.backward()
                self.actor_optimizer.step()

                self.critic_optimizer.zero_grad()
                critic_loss.backward()
                self.critic_optimizer.step()

        self.memory = []

PPO的优势在于样本效率和稳定性平衡得不错，训练过程中能持续进步而不是突然崩掉。实际用的时候，关键是调好clip_epsilon和learning rate。clip_epsilon太大等于没约束，太小策略更新太慢。

另一个容易踩的坑是advantage normalization。不归一化的话，训练过程会非常抖，特别是reward scale不稳定的时候。

实际项目里的踩坑记录

去年做一个机器人路径规划的项目，用强化学习让机器人自动避障。第一个版本用的DQN，结果在仿真环境里还行，一上真实机器人就崩了。问题有两个：真实环境的物理参数跟仿真不一致，以及真实环境的延迟比仿真大。

后来换了PPO，并在仿真里加了domain randomization——随机改变摩擦系数、传感器噪声、执行器延迟等参数，让策略更鲁棒。最后迁移效果好了不少，但训练时间从一天变成了三天。

# Domain randomization示例
def create_randomized_env():
    friction = np.random.uniform(0.5, 1.0)
    noise = np.random.uniform(0.0, 0.1)
    latency = np.random.uniform(0.0, 0.1)

    env = RobotNavigationEnv(
        friction=friction,
        sensor_noise=noise,
        action_latency=latency
    )
    return env

另一个项目是资源调度，用强化学习动态分配服务器资源。这个坑在reward design：最初设计的reward是响应时间，结果模型学会了拒绝请求来降低延迟。后来改成了综合指标，考虑了吞吐量和响应时间，模型才正常工作。

# 错误的reward design
def reward_wrong(response_time):
    return -response_time  # 模型会拒绝请求

# 改进后的reward design
def reward_improved(response_time, throughput):
    return 10 * throughput - response_time

环境设计比算法更重要

实践中发现，算法选型其实不是最关键的问题。环境设计、reward shaping、状态空间表示这些，往往决定成败。

一个经典的反面教材：想让Agent学会玩游戏，结果reward只给"游戏结束时的胜负"。这种稀疏reward导致Agent随机探索到能赢的策略需要的时间太长。后来加了几个引导reward，比如"距离目标越近奖励越高"，训练时间直接从几天变成几小时。

# 稀疏reward（不推荐）
def reward_sparse(done, won):
    if won:
        return 100
    elif done:
        return -100
    else:
        return 0

# 引导reward（推荐）
def reward_shaped(distance_to_goal, done, won):
    if won:
        return 100
    elif done:
        return -100
    else:
        return -distance_to_goal * 0.1  # 鼓励靠近目标

状态空间的表示也很关键。如果给模型喂了一堆无用的信息，训练效率会大打折扣。实践经验是：先人工筛选，只保留跟任务强相关的特征，然后再逐步增加信息量。

什么时候用什么

Q-Learning：状态空间小且离散的情况，比如FrozenLake这类。表格能存得下，而且训练快。

DQN：状态空间较大但离散，或者连续状态但动作空间离散。比如Atari游戏、CartPole。但要准备好应对训练不稳定的问题。

Policy Gradient：连续动作空间，比如控制机器人的关节角度。但样本效率低，适合仿真环境，不适合真实世界高频交互。

Actor-Critic：想在策略梯度基础上提升稳定性的情况。比纯Policy Gradient稳定，但调参复杂。

PPO：通用性强，样本效率和稳定性都不错。适合大多数现代RL应用，是默认选择。

Reward-based learning vs. Imitation learning：如果有expert demo，优先考虑模仿学习或者跟强化学习结合。纯强化学习从头训练成本太高。

代码库与工具

折腾过程中用过的几个工具：

  • Gymnasium：标准环境库，跟旧版Gym不兼容，注意依赖版本
  • Stable-Baselines3：现成的RL算法实现，PPO、SAC、TD3都有，适合快速验证
  • Ray RLLib：分布式训练，适合大规模任务，但学习成本高
  • CleanRL：代码简洁，适合学习算法原理，但工程化程度一般

安装命令：

pip install gymnasium
pip install stable-baselines3[extra]
pip install "ray[rllib]"
pip install cleanrl

这一路走下来的感受

从Q-Learning到PPO，算法越来越复杂，但核心思想没变：试错、反馈、调整。不同的是，现在的算法在"如何试"“如何反馈"“如何调整"这几个环节上做了大量工程优化。

最大的感受是：强化学习不像监督学习那样容易复现。同一个算法换个人实现，训练过程可能完全不同。代码实现细节、超参数选择、随机种子，都会影响结果。

另一个感受是：reward design比算法选择更重要。一个好的reward能让你用简单算法达到不错效果，一个差的reward再高级算法也救不回来。

最后，强化学习不是万能药。它能处理很多传统算法搞不定的问题，但训练成本、调试难度、工程复杂度都摆在那里。遇到问题时，先想想传统方法能不能解决，再考虑强化学习。

参考资源

  • Sutton & Barto, “Reinforcement Learning: An Introduction”
  • Schulman et al., “Proximal Policy Optimization Algorithms”
  • Mnih et al., “Playing Atari with Deep Reinforcement Learning”
  • Stable-Baselines3文档
  • CleanRL代码仓库

这些东西折腾到现在，差不多能说明白了。强化学习这一路从Q-Learning的表格查法走到PPO的神经网络策略，核心是试错与反馈，工程上则一直在解决训练稳定性与样本效率的问题。选算法前先想清楚环境特性、reward设计和工程约束，比盲目追求最新模型重要得多。

可用性说明：本文发布于 2021 年 2 月，距今已超过五年。文中涉及的软件版本、接口、下载地址、命令参数和操作界面可能已经发生变化，部分方案在当前环境下可能失效。请结合官方最新文档核对后再操作，生产环境使用前务必先行验证。

版权声明: 本文首发于 指尖魔法屋-把Q-Learning换到PPO时踩过的坑（https://blog.thinkmoon.cn/post/171-rl-deep-dive-qlearning-to-ppo/） 转载或引用必须申明原指尖魔法屋来源及源地址！

相关阅读

根据标签、分类和主题为你挑选的相近文章