【强化学习】结合Python实战深入分析原理

杆的倾斜角度超过12度；
小车移动超过2.4个单位长度；
回合步数达到回合最大步数。

import gym
env = gym.make("CartPole-v0")

环境对象env：它可以通过gym.make("CartPole-v0")直接获得。
智能体对象agent：我们一会儿要实现智能体类，它就是智能体类的对象。这个智能体需要实现一些成员，包括agent.reset(mode)、agent.step(observation, reward, terminated)、agent.close()。后文会介绍如何实现这些成员。
模式参数mode：字符串类型，可以是'train'和'test'。这个参数会进一步传到agent.reset(mode)中。如果是'train'那么智能体会处于训练模式，会更新参数；如果是’test‘则智能体会处于训练模式。

def play_episode(env, agent, mode=None):
    # 初始化
    observation, _ = env.reset()
    reward, terminated, truncated = 0., False, False
    agent.reset(mode=mode)
    episode_reward, elapsed_steps = 0., 0
    # 交互
    whileTrue:
        action = agent.step(observation, reward, terminated)
        if terminated or truncated:
            break
        observation, reward, terminated, truncated, _ = env.step(action)
        episode_reward += reward
        elapsed_steps += 1
    # 结束
    agent.close()
    return episode_reward, elapsed_steps

import torch
import torch.distributions as distributions
import torch.nn as nn
import torch.optim as optim
class VPGAgent:
    def __init__(self, env):
        self.action_n = env.action_space.n
        self.policy_net = nn.Sequential(
                nn.Linear(env.observation_space.shape[0], self.action_n, bias=False),
                nn.Softmax(1))
        self.optimizer = optim.Adam(self.policy_net.parameters(), lr=0.005)
    def reset(self, mode=None):
        self.mode = mode
        if self.mode == 'train':
            self.trajectory = []
    def step(self, observation, reward, terminated):
        state_tensor = torch.as_tensor(observation, dtype=torch.float).unsqueeze(0)
        prob_tensor = self.policy_net(state_tensor)
        action_tensor = distributions.Categorical(prob_tensor).sample()
        action = action_tensor.numpy()[0]
        if self.mode == 'train':
            self.trajectory += [observation, reward, terminated, action]
        return action
    def close(self):
        if self.mode == 'train':
            self.learn()
    def learn(self):
        state_tensor = torch.as_tensor(self.trajectory[0::4], dtype=torch.float)
        action_tensor = torch.as_tensor(self.trajectory[3::4], dtype=torch.long)
        return_tensor = torch.as_tensor(sum(self.trajectory[1::4]), dtype=torch.float)
        all_pi_tensor = self.policy_net(state_tensor)
        pi_tensor = torch.gather(all_pi_tensor, 1, action_tensor.unsqueeze(1)).squeeze(1)
        log_pi_tensor = torch.log(torch.clamp(pi_tensor, 1e-6, 1.))
        loss_tensor = -(return_tensor * log_pi_tensor).mean()
        self.optimizer.zero_grad()
        loss_tensor.backward()
        self.optimizer.step()
agent = VPGAgent(env)

import itertools
import numpy as np
episode_rewards = []
for episode in itertools.count():
    episode_reward, elapsed_steps = play_episode(env, agent, mode='train')
    episode_rewards.append(episode_reward)
    logging.info('训练回合 %d: 奖励 = %.2f, 步数 = %d',
            episode, episode_reward, elapsed_steps)
    if np.mean(episode_rewards[-20:]) > env.spec.reward_threshold:
        break
plt.plot(episode_rewards)

episode_rewards = []
for episode in range(100):
    episode_reward, elapsed_steps = play_episode(env, agent)
    episode_rewards.append(episode_reward)
    logging.info('测试回合%d：奖励 = %.2f，步数 = %d',
            episode, episode_reward, elapsed_steps)
logging.info('平均回合奖励 = %.2f ± %.2f',
        np.mean(episode_rewards), np.std(episode_rewards))

PyTorch版本：
https://zhiqingxiao.github.io/rl-book/en2023/code/CartPole-v0_VPG_torch.html
TensorFlow版本：
https://zhiqingxiao.github.io/rl-book/en2023/code/CartPole-v0_VPG_tf.html