Hongru
6b8156eceb
- 新增 RL_Algothrithms 模块，包含 A2C、QAC 智能体 - 添加 SAC 章节笔记和 C10 笔记 - 上传训练结果图片 - 完善 README 与 .gitignore
26 KiB
26 KiB
In [8]:
import torch
# 打印 PyTorch 版本
print(f"PyTorch 版本: {torch.__version__}")
# 检查 GPU 是否可用 (如果输出 True，说明大功告成！)
print(f"GPU 是否可用: {torch.cuda.is_available()}")
if torch.cuda.is_available():
# 打印当前使用的显卡型号
print(f"当前显卡: {torch.cuda.get_device_name(0)}")PyTorch 版本: 2.5.1+cu121 GPU 是否可用: True 当前显卡: NVIDIA GeForce RTX 5060 Ti
In [9]:
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
import torch.distributions as distributions
# 定义超参数
# 学习率
ACTOR_LR = 0.001
CRITIC_LR = 0.002
# 折扣因子
GAMMA = 0.99
# 训练迭代次数
NUM_EPISODES = 1000
# 1. 定义 Actor 网络 (演员：输出策略)
class Actor(nn.Module):
def __init__(self, state_dim, action_dim):
super(Actor, self).__init__()
# 定义两层全连接网络提取状态特征
self.fc1 = nn.Linear(state_dim, 128)
self.fc2 = nn.Linear(128, action_dim)
def forward(self, state):
# 使用 ReLU 激活函数
x = F.relu(self.fc1(state))
# 使用 Softmax 输出离散动作的概率分布
action_probs = F.softmax(self.fc2(x), dim=-1)
return action_probs
# 2. 定义 Critic 网络 (评论家：输出动作价值 Q 值)
class Critic(nn.Module):
def __init__(self, state_dim, action_dim):
super(Critic, self).__init__()
# 定义全连接网络，输入是状态，输出是各个动作的 Q 值
self.fc1 = nn.Linear(state_dim, 128)
self.fc2 = nn.Linear(128, action_dim)
def forward(self, state):
x = F.relu(self.fc1(state))
q_values = self.fc2(x)
return q_values
# 3. 核心训练逻辑 (对应 Algorithm 10.1)
def train_step(actor, critic, actor_optimizer, critic_optimizer,
state, action, reward, next_state, next_action, done):
"""
执行一步 QAC 算法的参数更新
"""
# 转换数据格式为 Tensor
state = torch.FloatTensor(state)
next_state = torch.FloatTensor(next_state)
reward = torch.FloatTensor([reward])
# -----------------------------------------
# Critic 更新 (价值更新)
# -----------------------------------------
# 计算当前状态动作的 Q(s_t, a_t, w_t)
q_values = critic(state) # 选取某个state对应的所有动作的Q值
current_q = q_values[action] # 选取当前动作对应的Q值
# 计算下一状态动作的 Q(s_{t+1}, a_{t+1}, w_t)
# 使用 .detach() 截断梯度，因为目标值不需要传递梯度回网络
next_q_values = critic(next_state).detach()
next_q = next_q_values[next_action]
# 计算 TD 目标：r_{t+1} + gamma * Q(s_{t+1}, a_{t+1}) (如果是终止状态则没有下一步的Q)
td_target = reward + GAMMA * next_q * (1 - int(done))
# 计算 TD 误差并更新 Critic 参数
# 对应公式: w_{t+1} = w_t + alpha_w * TD_Error * grad(Q)
critic_loss = F.mse_loss(current_q, td_target)
critic_optimizer.zero_grad()
critic_loss.backward()
critic_optimizer.step()
# -----------------------------------------
# Actor 更新 (策略更新)
# -----------------------------------------
# 获取当前状态下所有动作的概率分布
action_probs = actor(state)
# 构建概率分布对象，方便计算对数概率
dist = distributions.Categorical(action_probs)
# 计算 ln(pi(a_t | s_t, theta_t))
log_prob = dist.log_prob(torch.tensor(action))
# Actor 梯度上升目标：ln(pi) * Q(s, a)
# 在 PyTorch 中优化器默认执行梯度下降，所以加个负号变成最小化损失
# 注意这里使用的是刚刚更新前算出的 current_q，为了阻断梯度传到 Critic，使用 .detach()
actor_loss = -log_prob * current_q.detach()
actor_optimizer.zero_grad()
actor_loss.backward()
actor_optimizer.step()
return actor_loss.item(), critic_loss.item()In [10]:
# 设置设备 - 使用 CPU
device = torch.device("cpu")
print(f"使用设备: {device}")
import matplotlib.pyplot as plt
import numpy as np
import gymnasium as gym
# 训练的主循环
def main():
# 初始化环境
env = gym.make('CartPole-v1')
state_dim = env.observation_space.shape[0] # type: ignore
action_dim = int(env.action_space.n) # type: ignore
# 实例化 Actor 和 Critic 网络
actor = Actor(state_dim, action_dim).to(device)
critic = Critic(state_dim, action_dim).to(device)
actor_optimizer = optim.Adam(actor.parameters(), lr=ACTOR_LR)
critic_optimizer = optim.Adam(critic.parameters(), lr=CRITIC_LR)
# 记录训练数据
episode_rewards = []
actor_losses = []
critic_losses = []
for episode in range(NUM_EPISODES):
state, _ = env.reset()
episode_reward = 0
episode_actor_loss = 0
episode_critic_loss = 0
steps = 0
# 预先生成一个初始动作 a_0
state_tensor = torch.FloatTensor(state).to(device)
action_probs = actor(state_tensor)
action = distributions.Categorical(action_probs).sample().item()
while True:
# 与环境交互，获取当前状态
next_state, reward, terminated, truncated, _ = env.step(action)
done = terminated or truncated
# 预先生成下一个动作 a_{t+1}，供 Critic 更新使用
next_state_tensor = torch.FloatTensor(next_state).to(device)
next_action_probs = actor(next_state_tensor)
next_action = distributions.Categorical(next_action_probs).sample().item()
# 执行单步的学习
actor_loss, critic_loss = train_step(actor, critic, actor_optimizer, critic_optimizer,
state, action, reward, next_state, next_action, done)
# 记录loss
episode_actor_loss += actor_loss
episode_critic_loss += critic_loss
steps += 1
# 更新状态和动作
state = next_state
action = next_action
episode_reward += reward # type: ignore
if done:
break
# 记录本回合数据
episode_rewards.append(episode_reward)
actor_losses.append(episode_actor_loss / steps)
critic_losses.append(episode_critic_loss / steps)
# 打印训练进度
if (episode + 1) % 100 == 0:
avg_reward = np.mean(episode_rewards[-100:])
print(f"Episode {episode + 1}/{NUM_EPISODES} completed. 近100回合平均reward: {avg_reward:.2f}")
env.close()
# 绘制训练曲线
plot_qac_results(episode_rewards, actor_losses, critic_losses)
def plot_qac_results(rewards, actor_losses, critic_losses):
"""绘制QAC训练结果（保存到文件）"""
fig, axes = plt.subplots(1, 3, figsize=(15, 4))
# 绘制 reward 曲线
axes[0].plot(rewards, alpha=0.6, label='Episode Reward')
# 添加移动平均线
window = min(50, len(rewards) // 10)
if window > 1:
moving_avg = np.convolve(rewards, np.ones(window)/window, mode='valid')
axes[0].plot(np.arange(window-1, len(rewards)), moving_avg, 'r-', label=f'{window}-Episode Moving Avg')
axes[0].set_xlabel('Episode')
axes[0].set_ylabel('Reward')
axes[0].set_title('QAC Training Rewards')
axes[0].legend()
axes[0].grid(True, alpha=0.3)
# 绘制 Actor loss
axes[1].plot(actor_losses, alpha=0.6, color='orange')
axes[1].set_xlabel('Episode')
axes[1].set_ylabel('Loss')
axes[1].set_title('QAC Actor Loss')
axes[1].grid(True, alpha=0.3)
# 绘制 Critic loss
axes[2].plot(critic_losses, alpha=0.6, color='green')
axes[2].set_xlabel('Episode')
axes[2].set_ylabel('Loss')
axes[2].set_title('QAC Critic Loss')
axes[2].grid(True, alpha=0.3)
plt.tight_layout()
# 保存图片到文件
plt.savefig('qac_training_results.png', dpi=150)
plt.close()
# 打印统计信息
print(f"\n=== QAC 训练统计 ===")
print(f"总回合数: {len(rewards)}")
print(f"平均Reward: {np.mean(rewards):.2f}")
print(f"最高Reward: {np.max(rewards):.2f}")
print(f"最后100回合平均Reward: {np.mean(rewards[-100:]):.2f}")
print(f"图片已保存到: qac_training_results.png")
if __name__ == "__main__":
main()使用设备: cpu
/tmp/ipykernel_77052/2107052713.py:73: UserWarning: Using a target size (torch.Size([1])) that is different to the input size (torch.Size([])). This will likely lead to incorrect results due to broadcasting. Please ensure they have the same size. critic_loss = F.mse_loss(current_q, td_target)
Episode 100/1000 completed. 近100回合平均reward: 15.25 Episode 200/1000 completed. 近100回合平均reward: 33.00 Episode 300/1000 completed. 近100回合平均reward: 53.27 Episode 400/1000 completed. 近100回合平均reward: 34.57 Episode 500/1000 completed. 近100回合平均reward: 59.50 Episode 600/1000 completed. 近100回合平均reward: 51.37 Episode 700/1000 completed. 近100回合平均reward: 77.24 Episode 800/1000 completed. 近100回合平均reward: 77.66 Episode 900/1000 completed. 近100回合平均reward: 78.07 Episode 1000/1000 completed. 近100回合平均reward: 51.60 === QAC 训练统计 === 总回合数: 1000 平均Reward: 53.15 最高Reward: 336.00 最后100回合平均Reward: 51.60 图片已保存到: qac_training_results.png
In [11]:
# 设置设备 - 使用 CPU
device = torch.device("cpu")
print(f"使用设备: {device}")
# 定义超参数
ACTOR_LR = 0.001
CRITIC_LR = 0.002
GAMMA = 0.99
NUM_EPISODES = 1000
# 1. 定义 Actor 网络 (演员：输出离散动作的概率分布)
class ActorA2C(nn.Module):
def __init__(self, state_dim, action_dim):
super(ActorA2C, self).__init__()
self.fc1 = nn.Linear(state_dim, 128)
self.fc2 = nn.Linear(128, action_dim)
def forward(self, state):
x = F.relu(self.fc1(state))
action_probs = F.softmax(self.fc2(x), dim=-1)
return action_probs
# 2. 定义 Critic 网络 (评论家：输出标量 V 值)
class CriticA2C(nn.Module):
def __init__(self, state_dim):
super(CriticA2C, self).__init__()
self.fc1 = nn.Linear(state_dim, 128)
self.fc2 = nn.Linear(128, 1)
def forward(self, state):
x = F.relu(self.fc1(state))
v_value = self.fc2(x)
return v_value
# 3. 核心训练逻辑：基于优势函数 (Advantage)
def train_step_a2c(actor, critic, actor_optimizer, critic_optimizer,
state, action, reward, next_state, done):
state = torch.FloatTensor(state).unsqueeze(0).to(device)
next_state = torch.FloatTensor(next_state).unsqueeze(0).to(device)
reward = torch.FloatTensor([reward]).unsqueeze(0).to(device)
# Critic 更新
v_value = critic(state)
next_v_value = critic(next_state).detach()
td_target = reward + GAMMA * next_v_value * (1 - int(done))
advantage = td_target - v_value
critic_loss = F.mse_loss(v_value, td_target)
critic_optimizer.zero_grad()
critic_loss.backward()
critic_optimizer.step()
# Actor 更新
action_probs = actor(state)
dist = distributions.Categorical(action_probs)
log_prob = dist.log_prob(torch.tensor([action]).to(device))
actor_loss = -(log_prob * advantage.detach()).mean()
actor_optimizer.zero_grad()
actor_loss.backward()
actor_optimizer.step()
return actor_loss.item(), critic_loss.item()
def plot_a2c_results(rewards, actor_losses, critic_losses):
"""绘制A2C训练结果（保存到文件）"""
fig, axes = plt.subplots(1, 3, figsize=(15, 4))
axes[0].plot(rewards, alpha=0.6, label='Episode Reward')
window = min(50, len(rewards) // 10)
if window > 1:
moving_avg = np.convolve(rewards, np.ones(window)/window, mode='valid')
axes[0].plot(np.arange(window-1, len(rewards)), moving_avg, 'r-', label=f'{window}-Episode Moving Avg')
axes[0].set_xlabel('Episode')
axes[0].set_ylabel('Reward')
axes[0].set_title('A2C Training Rewards')
axes[0].legend()
axes[0].grid(True, alpha=0.3)
axes[1].plot(actor_losses, alpha=0.6, color='orange')
axes[1].set_xlabel('Episode')
axes[1].set_ylabel('Loss')
axes[1].set_title('A2C Actor Loss')
axes[1].grid(True, alpha=0.3)
axes[2].plot(critic_losses, alpha=0.6, color='green')
axes[2].set_xlabel('Episode')
axes[2].set_ylabel('Loss')
axes[2].set_title('A2C Critic Loss')
axes[2].grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig('a2c_training_results.png', dpi=150)
plt.close()
print(f"\n=== A2C 训练统计 ===")
print(f"总回合数: {len(rewards)}")
print(f"平均Reward: {np.mean(rewards):.2f}")
print(f"最高Reward: {np.max(rewards):.2f}")
print(f"最后100回合平均Reward: {np.mean(rewards[-100:]):.2f}")
print(f"图片已保存到: a2c_training_results.png")
# 4. 主循环函数
def main_a2c():
env = gym.make('CartPole-v1')
state_dim = env.observation_space.shape[0] # type: ignore
action_dim = int(env.action_space.n) # type: ignore
actor = ActorA2C(state_dim, action_dim).to(device)
critic = CriticA2C(state_dim).to(device)
actor_optimizer = optim.Adam(actor.parameters(), lr=ACTOR_LR)
critic_optimizer = optim.Adam(critic.parameters(), lr=CRITIC_LR)
episode_rewards = []
actor_losses = []
critic_losses = []
print("开始 A2C 训练...")
for episode in range(NUM_EPISODES):
state, _ = env.reset()
episode_reward = 0
episode_actor_loss = 0
episode_critic_loss = 0
steps = 0
while True:
state_tensor = torch.FloatTensor(state).unsqueeze(0).to(device)
action_probs = actor(state_tensor)
action = distributions.Categorical(action_probs).sample().item()
next_state, reward, terminated, truncated, _ = env.step(action)
done = terminated or truncated
actor_loss, critic_loss = train_step_a2c(actor, critic, actor_optimizer, critic_optimizer,
state, action, reward, next_state, done)
episode_actor_loss += actor_loss
episode_critic_loss += critic_loss
steps += 1
state = next_state
episode_reward += reward
if done:
break
episode_rewards.append(episode_reward)
actor_losses.append(episode_actor_loss / steps)
critic_losses.append(episode_critic_loss / steps)
if (episode + 1) % 100 == 0:
avg_reward = np.mean(episode_rewards[-100:])
print(f"回合 {episode + 1}/{NUM_EPISODES} 完成. 近 100 回合平均 Reward: {avg_reward:.2f}")
env.close()
plot_a2c_results(episode_rewards, actor_losses, critic_losses)
if __name__ == "__main__":
main_a2c()使用设备: cpu 开始 A2C 训练... 回合 100/1000 完成. 近 100 回合平均 Reward: 13.11 回合 200/1000 完成. 近 100 回合平均 Reward: 65.78 回合 300/1000 完成. 近 100 回合平均 Reward: 59.20 回合 400/1000 完成. 近 100 回合平均 Reward: 9.48 回合 500/1000 完成. 近 100 回合平均 Reward: 65.32 回合 600/1000 完成. 近 100 回合平均 Reward: 67.78 回合 700/1000 完成. 近 100 回合平均 Reward: 9.36 回合 800/1000 完成. 近 100 回合平均 Reward: 9.26 回合 900/1000 完成. 近 100 回合平均 Reward: 9.68 回合 1000/1000 完成. 近 100 回合平均 Reward: 51.43 === A2C 训练统计 === 总回合数: 1000 平均Reward: 36.04 最高Reward: 500.00 最后100回合平均Reward: 51.43 图片已保存到: a2c_training_results.png