PPO代码的第5部分包含训练循环、测试评估和主程序逻辑。让我详细拆解每个功能模块 5.1 训练函数train_ppo()pythondef train_ppo(env_nameCartPole-v1, num_episodes500, hidden_dim128, save_pathNone):训练流程架构text┌─────────────────────────────────────────────────┐ │ 训练循环 │ ├─────────────────────────────────────────────────┤ │ Phase 1 (episodes 1-50) Phase 2 (51-100) │ │ ┌──────────────┐ ┌──────────────┐ │ │ │ 探索阶段 │ ──────→ │ 利用阶段 │ │ │ │ 随机性大 │ │ 策略趋稳 │ │ │ └──────────────┘ └──────────────┘ │ │ ↓ ↓ │ │ ┌─────────────────────────────────────────┐ │ │ │ 经验回放缓冲区 │ │ │ │ 存储 (s,a,r,s,done) 元组 │ │ │ └─────────────────────────────────────────┘ │ │ ↓ ↓ │ │ ┌─────────────────────────────────────────┐ │ │ │ PPO更新模块 │ │ │ │ 1. 计算GAE优势 │ │ │ │ 2. 计算裁剪损失 │ │ │ │ 3. 更新网络参数 │ │ │ └─────────────────────────────────────────┘ │ │ ↓ │ │ ┌─────────────────────────────────────────┐ │ │ │ 模型保存检查点 │ │ │ │ 如果 reward best_reward: 保存 │ │ │ └─────────────────────────────────────────┘ │ └─────────────────────────────────────────────────┘详细代码解析python# 1. 环境初始化 env gym.make(env_name) state_dim env.observation_space.shape[0] # CartPole: 4 action_dim env.action_space.n # CartPole: 2 # 2. 设备设置自动选择GPU/CPU device torch.device(cuda if torch.cuda.is_available() else cpu) print(f使用设备: {device}) # 3. 创建PPO智能体 agent PPO( state_dimstate_dim, hidden_dimhidden_dim, # 128 action_dimaction_dim, actor_lr1e-3, # 策略网络学习率 critic_lr1e-2, # 价值网络学习率 lmbda0.95, # GAE参数 epochs10, # 每批数据复用次数 eps0.2, # 裁剪阈值 gamma0.98, # 折扣因子 devicedevice )阶段式训练循环pythonreturn_list [] best_reward -float(inf) # 记录最佳奖励 # 10个训练阶段 for i in range(10): # 10个阶段 # tqdm进度条显示 with tqdm(totalint(num_episodes/10), descf阶段 {i1}/10) as pbar: # 每个阶段训练 num_episodes/10 个episode for i_episode in range(int(num_episodes/10)): episode_return 0 # 4. 经验收集 transition_dict { states: [], actions: [], next_states: [], rewards: [], dones: [] } observation, info env.reset() done False # 单个episode循环 while not done: # 4.1 选择动作 action agent.take_action(observation) # 4.2 执行动作 next_observation, reward, terminated, truncated, info env.step(action) done terminated or truncated # 4.3 存储经验 transition_dict[states].append(observation) transition_dict[actions].append(action) transition_dict[next_states].append(next_observation) transition_dict[rewards].append(reward) transition_dict[dones].append(done) observation next_observation episode_return reward # 5. 记录回报 return_list.append(episode_return) # 6. PPO更新核心 agent.update(transition_dict) # 7. 保存最佳模型 if episode_return best_reward: best_reward episode_return if save_path: best_path save_path.replace(.pth, _best.pth) agent.save_model(best_path) # 8. 更新进度条 if (i_episode 1) % 10 0: pbar.set_postfix({ 平均奖励: f{np.mean(return_list[-10:]):.2f} }) pbar.update(1) 5.2 测试函数test_trained_ppo()pythondef test_trained_ppo(agent, env_nameCartPole-v1, num_episodes5, renderTrue):测试流程text┌─────────────────────────────────────────────────┐ │ 测试循环 │ ├─────────────────────────────────────────────────┤ │ Episode 1 Episode 2 Episode 3 │ │ ┌──────────────┐ ┌──────────────┐ ┌──────┐ │ │ │ 环境重置 │ │ 环境重置 │ │ ... │ │ │ │ ↓ │ │ ↓ │ │ │ │ │ │ 动作选择 │ │ 动作选择 │ │ │ │ │ │ (无探索) │ │ (无探索) │ │ │ │ │ │ ↓ │ │ ↓ │ │ │ │ │ │ 环境步进 │ │ 环境步进 │ │ │ │ │ │ ↓ │ │ ↓ │ │ │ │ │ │ 渲染显示 │ │ 渲染显示 │ │ │ │ │ │ ↓ │ │ ↓ │ │ │ │ │ │ 记录步数/奖励│ │ 记录步数/奖励│ │ │ │ │ └──────────────┘ └──────────────┘ └──────┘ │ │ ↓ ↓ ↓ │ │ ┌─────────────────────────────────────────────┐ │ │ │ 统计结果平均/最大/最小奖励 │ │ │ └─────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────┘测试与训练的区别方面训练模式测试模式动作选择概率采样探索贪婪策略利用渲染通常关闭开启可视化网络更新更新参数冻结参数数据记录详细记录用于训练只记录性能指标python# 测试模式下的动作选择实际代码中使用的是同样的take_action # 但注意训练时take_action使用概率采样测试时也使用同样的方法 # 如果想要确定性动作可以修改为 def test_action(self, state): state torch.tensor([state], dtypetorch.float).to(self.device) probs self.actor(state) action torch.argmax(probs, dim1) # 选择概率最大的而不是采样 return action.item() 5.3 模型加载函数load_and_test_model()pythondef load_and_test_model(model_path, env_nameCartPole-v1, num_episodes5):模型恢复流程text┌─────────────────────────────────────────────────┐ │ 模型加载流程 │ ├─────────────────────────────────────────────────┤ │ 1. 读取保存的checkpoint │ │ ├── actor_state_dict (策略网络权重) │ │ ├── critic_state_dict (价值网络权重) │ │ └── optimizer状态 (优化器内部变量) │ │ ↓ │ │ 2. 创建新的智能体实例 │ │ ├── 相同的网络结构 │ │ └── 移动到相同设备 │ │ ↓ │ │ 3. 加载状态字典 │ │ ├── actor.load_state_dict() │ │ └── critic.load_state_dict() │ │ ↓ │ │ 4. 测试验证 │ │ └── run test_trained_ppo() │ └─────────────────────────────────────────────────┘ 5.4 策略对比函数随机策略pythondef random_policy_demo(): action env.action_space.sample() # 随机选择特点无记忆无学习纯随机探索性能最差平均20-30步简单规则策略pythondef simple_policy_demo(): if angle 0: action 1 # 杆向右倾 → 向右移动 else: action 0 # 杆向左倾 → 向左移动特点基于领域知识确定性规则性能中等平均50-80步PPO策略python# 通过学习得到的策略 def ppo_policy(observation): return trained_agent.take_action(observation)特点数据驱动学习自适应策略性能最优可达500步️ 5.5 主程序逻辑pythonif __name__ __main__:执行流程图text┌─────────────────────────────────────────────────────────┐ │ 主程序入口 │ └─────────────────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────────────────┐ │ 1. 随机策略演示 │ │ random_policy_demo(5 episodes) │ └─────────────────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────────────────┐ │ 2. 简单策略演示 │ │ simple_policy_demo(5 episodes) │ └─────────────────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────────────────┐ │ 3. 模型决策 │ ├─────────────────────────────────────────────────────────┤ │ 用户输入: 是否重新训练? (y/n) │ │ ↓ ↓ │ │ 是: 训练新模型 否: 检查已有模型 │ │ ↓ ↓ │ │ train_ppo() 文件存在? → 是 → load_model() │ │ ↓ ↓ ↓ │ │ 保存模型 否 → train_ppo() │ └─────────────────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────────────────┐ │ 4. 测试PPO模型 │ │ test_trained_ppo(renderTrue) │ └─────────────────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────────────────┐ │ 5. 三种策略可视化对比 │ │ ┌──────────────┬──────────────┬──────────────┐ │ │ │ 随机策略 │ 简单策略 │ PPO策略 │ │ │ │ 实时渲染 │ 实时渲染 │ 实时渲染 │ │ │ └──────────────┴──────────────┴──────────────┘ │ └─────────────────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────────────────┐ │ 6. 总结输出 │ │ 性能对比 模型保存位置 │ └─────────────────────────────────────────────────────────┘交互式用户输入处理python# 训练决策 train_new input(\n是否要重新训练模型(y/n默认y): ).lower() ! n # 测试回合数设置 test_episodes int(input(请输入要测试的回合数默认5: ) or 5) # 对比回合数设置 compare_episodes int(input(请输入每个策略对比的回合数默认2: ) or 2) 5.6 性能监控与统计python# 滑动平均计算 if len(ppo_rewards) 50: moving_avg np.convolve(ppo_rewards, np.ones(50)/50, modevalid) # 统计指标 print(fPPO模型测试平均奖励: {np.mean(episode_rewards):.2f}) print(f最大奖励: {np.max(episode_rewards)}) print(f最小奖励: {np.min(episode_rewards)})训练曲线解读text奖励 ↑ 500┤ ╱╲ ╱╲ ╱╲ ╱╲ ╱╲ │ ╱ ╲ ╱ ╲ ╱ ╲ ╱ ╲ ╱ ╲ │ ╱ ╲ ╱ ╲╱ ╲╱ ╲╱ ╲ │ ╱ ╱ ╱ ╱ ╱ ╲ │ ╱ ╱ ╱ ╱ ╱ ╲ │╱______╱______╱______╱______╱__________╲___→ 回合 0 50 100 150 200 250 300 区域说明 0-50回合探索期波动大性能低 50-150回合快速学习期性能提升 150-300回合收敛期稳定在高分 最佳实践建议1. 训练参数调优python# 根据环境复杂度调整 if env_name CartPole-v1: num_episodes 300 # 简单环境 elif env_name LunarLander-v2: num_episodes 2000 # 复杂环境 hidden_dim 256 # 更大的网络2. 模型保存策略python# 定期保存检查点 if i_episode % 100 0: agent.save_model(fcheckpoint_episode_{total_episodes}.pth)3. 早停机制python# 如果连续50个episode都达到500分提前停止 if np.mean(return_list[-50:]) 495: print(环境已解决提前停止训练) break4. 学习率衰减python# 随着训练进行降低学习率 if total_episodes 200: for param_group in agent.actor_optimizer.param_groups: param_group[lr] 1e-4 # 降低学习率这个第5部分代码实现了完整的训练-测试-评估流程使得PPO算法不仅是一个理论模型更是一个实用的强化学习解决方案