强化学习 5.PPO 算法（实验部分）

发表于 2026-03-30 更新于 2026-04-10

前言

从零开始学习ai文章系列计划是个人在《动手学深度学习》和《磨菇书》两本书的学习中的个人笔记，文章也会以课本中的章节分开，即每个章节一片笔记。我会尽量的把主要内容以及遇到的难点进行记录与解决，如果哪里有错误的欢迎指正。或者不清晰的可以直接查看原文部分。

《蘑菇书》原文（课本）：https://datawhalechina.github.io/easy-rl/#/

（由于有时候公式太多，可能会直接贴图片）

蘑菇书的文章结构不会跟之前《动手学深度学习》按照原文章节进行，个人会适当调节。

1. 近端策略优化裁剪（PPO-clip）

1.1 环境介绍

我们和之前策略梯度一样，使用的环境是CartPole-v1。以方便我们了解从策略梯度到PPO我们做了什么。

状态空间：[ 小车位置，小车速度，杆子的角度，杆子的角速度 ]，4个连续值

动作空间： [左，右]，2个离散值

接下来我们创建环境并设置需要的超参数：

env = gym.make("CartPole-v1")
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.n

policy = PolicyNet(state_dim, action_dim).to(device)
num_episodes = 1000

接下来是我们的策略网络（其中的网络更新部分我们后面讲）

能看到其实跟上一节一样，也是隐藏层为128个神经元的3层感知机。

class PolicyNet(nn.Module):
    def __init__(self, state_dim, action_dim):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(state_dim, 128),
            nn.ReLU(),
            nn.Linear(128, action_dim)
        )
        self.optimizer = optim.Adam(self.parameters(), lr=3e-4, weight_decay=1e-4)
        self.ppo_epochs = 10
        self.gamma = 0.99
        self.clip_eps = 0.2

    def forward(self, x):
        logits = self.net(x)
        return torch.softmax(logits, dim=-1)

    def update(self,rewards,states,actions,log_probs_old):
        # ===== Monte Carlo Return =====
        def compute_returns(rewards, gamma):
            G = 0
            returns = []
            for r in reversed(rewards):
                G = r + gamma * G
                returns.insert(0, G)
            return torch.tensor(returns, dtype=torch.float32)

        # ===== Advantage = Return =====
        returns = compute_returns(rewards, self.gamma).to(device)
        advantages = (returns - returns.mean()) / (returns.std() + 1e-8)

        states = torch.stack(states)
        actions = torch.stack(actions)
        log_probs_old = torch.stack(log_probs_old)


        # ===== PPO update =====
        for _ in range(self.ppo_epochs):
            probs = self.forward(states)
            dist = torch.distributions.Categorical(probs)
            log_probs = dist.log_prob(actions)

            ratios = torch.exp(log_probs - log_probs_old)

            surr1 = ratios * advantages
            surr2 = torch.clamp(ratios, 1 - self.clip_eps, 1 + self.clip_eps) * advantages
            loss = -torch.min(surr1, surr2).mean()


            self.optimizer.zero_grad()
            loss.backward()
            self.optimizer.step()

1.2 算法流程

跟上一节一样，我们第一步收集轨迹链，第二步利用收集的轨迹链进行更新。

def train():
    for episode in range(num_episodes):
        states = []
        actions = []
        rewards = []
        log_probs_old = []
        state, _ = env.reset()

        # 1. 收集我们需要的sar轨迹链
        collect_datas(env,state, states, actions, rewards, log_probs_old)

        # 2. ppo 更新
        policy.update( rewards, states, actions, log_probs_old)



        if episode % 50 == 0:
            print(f"Episode {episode}, Reward: {sum(rewards)}")

    env.close()
    torch.save(policy.state_dict(), "ppo_pure_cartpole.pth")
    print("Saved to ppo_pure_cartpole.pth")

    return policy

1.3 轨迹链收集

我们在之前1.2中能看到，我们轨迹中主要就是收集这几个列表

states = []
actions = []
rewards = []
log_probs_old = []

def collect_datas(env,init_state,states,actions,rewards,log_probs_old):
    state=init_state
    done = False
    while not done:
        state_tensor = torch.tensor(state, dtype=torch.float32).to(device)
        probs = policy(state_tensor)
        dist = torch.distributions.Categorical(probs)
        action = dist.sample()
        log_prob = dist.log_prob(action)

        next_state, reward, terminated, truncated, _ = env.step(action.item())
        done = terminated or truncated

        states.append(state_tensor)
        actions.append(action)
        rewards.append(reward)
        log_probs_old.append(log_prob.detach())

        state = next_state

1.4 网络更新

其实也就是之前我们网络定义里面update的部分

def update(self,rewards,states,actions,log_probs_old):
    # ===== Monte Carlo Return =====
    def compute_returns(rewards, gamma):
        G = 0
        returns = []
        for r in reversed(rewards):
            G = r + gamma * G
            returns.insert(0, G)
        return torch.tensor(returns, dtype=torch.float32)

    # ===== Advantage = Return =====
    returns = compute_returns(rewards, self.gamma).to(device)
    advantages = (returns - returns.mean()) / (returns.std() + 1e-8)

    states = torch.stack(states)
    actions = torch.stack(actions)
    log_probs_old = torch.stack(log_probs_old)


    # ===== PPO update =====
    for _ in range(self.ppo_epochs):
        probs = self.forward(states)
        dist = torch.distributions.Categorical(probs)
        log_probs = dist.log_prob(actions)

        ratios = torch.exp(log_probs - log_probs_old)

        surr1 = ratios * advantages
        surr2 = torch.clamp(ratios, 1 - self.clip_eps, 1 + self.clip_eps) * advantages
        loss = -torch.min(surr1, surr2).mean()


        self.optimizer.zero_grad()
        loss.backward()
        self.optimizer.step()

第一步，回报的计算以及将回报转换成优势

1 2	returns = compute_returns(rewards, self.gamma).to(device) advantages = (returns - returns.mean()) / (returns.std() + 1e-8)

第二步，我们要将原来的列表转换成 nx1的矩阵形式（后面才能放入到网络中计算）

1
2
3

states = torch.stack(states)
actions = torch.stack(actions)
log_probs_old = torch.stack(log_probs_old)

第三步，利用这些数据进行更新

# ===== PPO update =====
for _ in range(self.ppo_epochs):
    probs = self.forward(states)
    dist = torch.distributions.Categorical(probs)
    log_probs = dist.log_prob(actions)

    ratios = torch.exp(log_probs - log_probs_old)

    surr1 = ratios * advantages
    surr2 = torch.clamp(ratios, 1 - self.clip_eps, 1 + self.clip_eps) * advantages
    loss = -torch.min(surr1, surr2).mean()


    self.optimizer.zero_grad()
    loss.backward()
    self.optimizer.step()

首先根据相同的state，算出action的分布，然后根据旧策略action中的值，获取对应的概率，转换成log_probs。

1
2
3

probs = self.forward(states)
dist = torch.distributions.Categorical(probs)
log_probs = dist.log_prob(actions)

第四步，是算出我们的重要性

1	ratios = torch.exp(log_probs - log_probs_old)

即 \[ \frac{p_\theta(a_t|s_t)}{p_{\theta^k}(a_t|s_t)} \] 通常我们会将该除法利用以下公式进行转换（变成我们代码中的式子）

我们做的不单单是公式的转换，我们还有其他许多好处，其中最重要的就是用更稳定的方式计算“新旧策略的概率比值（ratio）”，避免数值问题。因为强化学习里概率往往非常小（接近 0）。直接相除容易数值不稳定、下溢（变成 0）、梯度爆炸。

第五步，将重要性和优势相乘，然后根据最大值最小值进行裁剪，获取所有结果的平均值。

1
2
3

surr1 = ratios * advantages
surr2 = torch.clamp(ratios, 1 - self.clip_eps, 1 + self.clip_eps) * advantages
loss = -torch.min(surr1, surr2).mean()

最后一步就是进行求导，然后更新

1
2
3

self.optimizer.zero_grad()
loss.backward()
self.optimizer.step()

我们反复进行此过程一共self.ppo_epochs次（这里我用的10次）。用同一组数据更新完10次后，我们再从环境中抓取新的数据。

1.5 查看效果

main 函数和展示函数

def render(policy, episodes=5):
    env = gym.make("CartPole-v1", render_mode="human")

    for ep in range(episodes):
        state, _ = env.reset()
        done = False
        total_reward = 0

        while not done:
            state_tensor = torch.tensor(state, dtype=torch.float32).to(device)
            with torch.no_grad():
                probs = policy(state_tensor)
                action = torch.argmax(probs).item()

            state, reward, terminated, truncated, _ = env.step(action)
            done = terminated or truncated
            total_reward += reward

        print(f"Episode {ep}, Reward: {total_reward}")

    env.close()


def load_policy(model_path):
    state_dim = 4  # CartPole 固定
    action_dim = 2

    policy = PolicyNet(state_dim, action_dim)
    policy.load_state_dict(torch.load(model_path))
    policy.to(device)
    policy.eval()  # 重要：进入评估模式

    return policy


if __name__ == "__main__":
    # ① 训练并保存
    policy = train()

    # ② 加载并展示
    policy = load_policy("ppo_pure_cartpole.pth")
    render(policy)

我们能发现再150的时候就出现了500分的一次。后续大概600 episode的时候才稳点了点。最终测试也是5次都满分，其实效果还行。