有口碑的武进网站建设,临沂住房和城乡建设局网站,手机建设银行网站首页,广州的一起做网店app下载前言の碎碎念
由于我做的模仿学习#xff0c;可能由于没有完全模仿#xff0c;可以说效果很烂……后来用强化学习优化#xff0c;这个倒是不用自己做数据集了#xff0c;为方便大家只搞代码#xff0c;这里只说这部分的经历和方法。 实践基础介绍
1-动作
先介绍一个强化… 前言の碎碎念
由于我做的模仿学习可能由于没有完全模仿可以说效果很烂……后来用强化学习优化这个倒是不用自己做数据集了为方便大家只搞代码这里只说这部分的经历和方法。 实践基础介绍
1-动作
先介绍一个强化学习中的“行为空间”的概念。 玩过游戏机的都知道游戏按键就那几个顶多加上组合所以你玩游戏可以操作的空间组合就那么大你与游戏世界连接的输入方式只有这几个按钮。 比如坦克大战你操纵的坦克只能上下左右和攻击那你只能在这几个里进行操作就好比你的手柄你不能做出手柄按钮以外的操作来作为游戏的输入。
很自然的智能体能做的只有行为空间中的行为如果根据概率来选择行为来行动是很自然的一种方式。
2-状态 我们玩游戏还得看游戏画面才能采取下一个决策这就是反馈给智能体的状态。在实际工程中需要将状态中的各种要素摘取出来比如坦克大战需要知道敌人的位置自己的位置哪里有墙其他的信息就无所谓了。
3-奖惩
我们打游戏如果失误掉血了肯定以后就会尽量避免这个操作如果试出来的组合绝招管用那以后肯定要记下来好好利用。这就是奖励和惩罚。衡量该奖还是惩罚的函数就叫奖励函数。
4-智能体
就类比成玩游戏的人不同人玩游戏的技术进步也不一样这就涉及到神经网络设计或者简单的函数套路策略。
5-环境
游戏本身毕竟玩坦克大战的经验在玩宝可梦的时候用不上所以环境是根本。 范式 环境初始化定义游戏规则和状态表示。 智能体设计选择合适的算法如 DQN设计要被优化的目标神经网络。 训练循环 初始化环境获取初始状态。 选择动作epsilon-greedy策略。 执行动作获取新的状态、奖励和是否终止。 存储经验到回放缓存。 从回放缓存中采样一批数据更新目标网络参数。 更新目标网络每隔一定频率步。 保存模型每隔一定局数因为一局的奖励不平均所以以几局为单位。 评估定期评估智能体的性能记录关键指标。每隔一定轮数评估智能体评估时禁止对目标网络的改动输出模型得到的动作然后评估 原理的简要介绍 1. 强化学习的基本概念 智能体Agent学习者和决策者负责根据当前状态选择动作。 环境Environment智能体所处的外部世界接收智能体的动作并返回新的状态和奖励。 状态State环境的当前状况描述了智能体所处的环境信息。 动作Action智能体在某个状态下可以采取的行为。 奖励Reward环境对智能体动作的反馈通常是一个标量值用于衡量动作的好坏。 策略Policy智能体的行为规则决定了在给定状态下选择哪个动作。 价值函数Value Function衡量某个状态或动作的价值通常用 V(s) 或 Q(s,a) 表示。 回报Return从某个时间步开始到结束的所有奖励的累积和通常用 Gt 表示。 2. 马尔科夫决策过程Markov Decision Process, MDP 强化学习的核心是马尔科夫决策过程它是一个数学框架用于描述智能体与环境的交互过程。 马尔科夫性Markov Property当前状态包含了所有历史信息即未来的状态只依赖于当前状态而与之前的状态无关。 P(st1∣st)P(st1∣st,st−1,…,s1) MDP 的组成 状态集合 S 动作集合 A 转移概率 P(s′∣s,a)在状态 s 下采取动作 a 转移到状态 s′ 的概率。 奖励函数 R(s,a,s′)在状态 s 下采取动作 a 转移到状态 s′ 时获得的奖励。 3. 贝尔曼方程Bellman Equation 贝尔曼方程是强化学习中的核心方程用于递归地定义价值函数。 贝尔曼期望方程Bellman Expectation Equation V(s)a∈A∑π(a∣s)[R(s,a)γs′∈S∑P(s′∣s,a)V(s′)]Q(s,a)R(s,a)γs′∈S∑P(s′∣s,a)a′∈A∑π(a′∣s′)Q(s′,a′) 其中π(a∣s) 是策略γ 是折扣因子0≤γ1。 贝尔曼最优方程Bellman Optimality Equation V∗(s)a∈Amax[R(s,a)γs′∈S∑P(s′∣s,a)V∗(s′)]Q∗(s,a)R(s,a)γs′∈S∑P(s′∣s,a)a′∈AmaxQ∗(s′,a′) 这些方程用于定义最优价值函数和最优策略。 4. 价值函数与策略 价值函数Value Function 状态价值函数 V(s)在策略 π 下从状态 s 开始的期望回报。 动作价值函数 Q(s,a)在策略 π 下从状态 s 开始并采取动作 a 的期望回报。 策略Policy 确定性策略在每个状态下选择一个固定的动作。 随机性策略在每个状态下根据概率分布选择动作。 5. 策略迭代与价值迭代 策略迭代Policy Iteration 策略评估Policy Evaluation计算当前策略的价值函数。 策略改进Policy Improvement根据价值函数更新策略。 交替进行策略评估和策略改进直到策略收敛。 价值迭代Value Iteration 直接求解贝尔曼最优方程更新价值函数直到收敛。 从初始价值函数开始逐步逼近最优价值函数。 6. 探索与利用Exploration vs. Exploitation 探索Exploration尝试新的动作以获取更多信息。 利用Exploitation使用当前已知的最佳动作以获取最大回报。 平衡方法 ϵ-贪心策略以概率 ϵ 随机选择动作以概率 1−ϵ 选择当前最优动作。 软最大化策略Softmax Policy根据动作的价值以概率分布选择动作。 7. 深度强化学习Deep Reinforcement Learning 将深度学习与强化学习结合使用神经网络来近似价值函数或策略函数。 DQNDeep Q-Network 使用神经网络近似 Q(s,a)。 引入经验回放缓存Experience Replay和目标网络Target Network以提高训练稳定性。 PPOProximal Policy Optimization 基于策略梯度的方法通过截断概率比来限制策略更新的幅度提高训练的稳定性和效率。 8. 必须补充的基础 概率论与数理统计理解随机过程、概率分布、期望值等概念。 线性代数掌握向量、矩阵运算用于表示状态和动作。 动态规划理解贝尔曼方程和策略迭代、价值迭代等概念。 神经网络了解神经网络的基本结构和训练方法用于近似价值函数或策略函数。 优化算法掌握梯度下降、Adam 等优化算法用于更新网络参数。 总结 强化学习通过智能体与环境的交互来学习最优策略。其核心包括马尔科夫决策过程、贝尔曼方程、价值函数、策略迭代和价值迭代等。深度强化学习结合了深度学习技术进一步提升了强化学习的性能。理解这些基础知识是掌握强化学习的关键。 最后代码。
由于大家没有数据集和模仿学习的部分的神经网络这里放一个基本的梯度下降来优化目标网络的强化学习的样例代码需要装一些包。
这部分由于设计的依赖比较多都放出来放不下这部分只做参考
1-环境准备
def init_wargame(redAgent , blueAgent):run demo in single agent modeprint(running in single agent mode...)# instantiate agents and envred1 redAgent #MyAgent1()#自己的agent,因为样本数据都来自红方blue1 blueAgentenv1 TrainEnv()begin time.time()# get data ready, data can from files, web, or any other sourceswith open(gopt.scenpath201033029601.json, encodingutf8) as f:scenario_data json.load(f)with open(gopt.mappathbasic.json, encodingutf8) as f:basic_data json.load(f)with open(gopt.mappathcost.pickle, rb) as file:cost_data pickle.load(file)see_data numpy.load(gopt.mappathsee.npz)[data]# varialbe to build replayall_states []# player setup infoplayer_info [{seat: 1,faction: 0,role: 1,user_name: myAgentname,user_id: 0},{seat: 11,faction: 1,role: 1,user_name: demo,user_id: 0}]# env setup infoenv_step_info {scenario_data: scenario_data,basic_data: basic_data,cost_data: cost_data,see_data: see_data,player_info: player_info}# setup env - 初始化环境并获取指针state env1.setup(env_step_info)import copyall_states.append(copy.deepcopy(state[GREEN]))#初始帧print(Environment is ready.)# setup AIs - 注册AI信息red1.setup({scenario: scenario_data,basic_data: basic_data,cost_data: cost_data,see_data: see_data,seat: 1,faction: 0,role: 0,user_name: myAgentname,user_id: 0,state: state,})blue1.setup({scenario: scenario_data,basic_data: basic_data,cost_data: cost_data,see_data: see_data,seat: 11,faction: 1,role: 0,user_name: demo,user_id: 0,state: state,})print(agents are ready.)return red1, blue1, env1, state, all_states, begin
2-训练脚本
def train():智能体-主训练函数# 初始化环境和智能体# 日志记录writer create_writer(LOG_DIR)# TensorBoardbest_reward -float(inf)for episode in tqdm(range(EPOCH), descTraining Episodes):print(fEpisode {episode 1}/{EPOCH} started...)red1, blue1, env1, state, all_states, begin init_wargame(MyAgent56(), Agent())rewards []losses []# 训练阶段episode_reward 0episode_loss 0step 0# loop until the end of gameprint(steping)done False#传入actions-处理step-记录all_states循环while not done :actions []actions red1.step(state[RED])actions blue1.step(state[BLUE])state, done env1.step(actions) #更新环境状态print(step_, state[-1][time][cur_step])print(stage, state[-1][time][stage])all_states.append(copy.deepcopy(state[GREEN]))#原因可能是 state 和 all_states 中的元素共享相同的引用# # 方便验证-cq20250421# if state[-1][time][cur_step] 29:# breakepisode_reward red1.trainer.outrewardepisode_loss red1.trainer.outlossstep 1if done:break# 一局结束了如果你的任务是分局进行的例如一局有1800帧step记录了并且每局的奖励是独立计算的那么每局结束后保存一次模型是一个合理的选择。# 记录训练指标rewards.append(episode_reward)losses.append(episode_loss / (step 1))writer.add_scalar(Train/Reward, episode_reward, episode)writer.add_scalar(Train/Loss, episode_loss / (step 1), episode)mean_eval_reward np.mean(rewards)writer.add_scalar(Eval/Reward, mean_eval_reward , episode)# 保存模型的策略if episode_reward best_reward:best_reward episode_rewardred1.trainer.save(gopt.modelpathbest_rl_model.pth) # 保存最佳模型print(f回合:{episode}/{EPOCH}奖励:{episode_reward:.2f}, \评估奖励:{mean_eval_reward:.2f}最佳评估奖励:{best_reward:.2f})# if episode % save_interval 0:# agent.save_model(fmodel_episode_{episode}.pth) # 定期保存# 探索率衰减red1.trainer.epsilon max(red1.trainer.epsilon * 0.995, red1.trainer.epsilon_min)# # 评估阶段,每5局评估一次# if episode % 5 0:# # 在训练过程中定期冻结策略暂停梯度更新让智能体在固定环境中运行多个回合计算平均奖励以衡量当前策略的优劣。# # 保存最佳模型# if mean_eval_reward best_reward:# best_reward mean_eval_reward# save_model_lite(red1.trainer.policy_net, f{gopt.rlmodelsavepath}best_model.pth)# print(f回合:{episode}/{1800}奖励:{episode_reward:.2f}, \# 评估奖励:{mean_eval_reward:.2f}最佳评估奖励:{best_reward:.2f})print(fTotal time: {time.time() - begin:.3f}s)begin_formatted time.strftime(%Y%m%d%H%M%S, time.localtime(begin)) #年月日时分秒的字符串# 旧的连续方式zip_name gopt.logdirreplays/freplay_{begin_formatted}_old.zipwith zipfile.ZipFile(zip_name, w, zipfile.ZIP_DEFLATED, compresslevel9) as z:z.writestr(freplay_{begin_formatted}.json, json.dumps(all_states, ensure_asciiFalse, indentNone, separators(,, :)).encode(gbk))print(fReplay saved to {zip_name})# 添加资源释放# env1.reset()# red1.reset()# blue1.reset()del red1, blue1, env1, state, all_statesif torch.cuda.is_available():torch.cuda.empty_cache()# 训练结束writer.close()print(fTraining completed. Best eval reward: {best_reward:.2f})
3-强化学习部分优化的目标网络是self.policy_net def get_pred(policy_net ,obs_tensor: torch.Tensor) - List[int]:预测动作的函数# 这里可以直接调用模型进行预测# 假设模型输出的是一个概率分布with torch.no_grad():probs policy_net(obs_tensor) # 输出概率分布[1, 3, 8]# 在最后一个维度上取【概率最大】的类别索引max_indices torch.argmax(probs, dim-1) # 结果形状为 [1, 3]# 如果需要去掉第一个维度batch 维度可以使用 squeezemax_indices max_indices.squeeze(0) # 结果形状为 [3]# 将 max_indices 转换为整数列表actionLists max_indices.tolist()return actionLists# 可插拔强化学习类
class RLTrainer:def __init__(self, policy_net: nn.Module, lr: float 1e-3):# 加载模仿学习的预训练模型self.policy_net policy_netself.target_update_freq 20 # 目标网络更新频率(TARGET_UPDATE)# self.batch_size 32 # 训练批量大小(BATCH_SIZE)self.epsilon_min 0.1 # 最小探索率self.epsilon 0.1 # 探索率self.gamma0.99 # 折扣因子self.step_counter 0 # 全局步数计数器用于调试和监控训练进度self.optimizer optim.Adam(self.policy_net.parameters(), lrlr)self.memory [] # 简单 bufferself.outreward 0 # 初始化输出奖励(对外可获取)self.outloss 0 # 初始化输出损失(对外可获取)def select_action(self, obs_tensor: torch.Tensor) - int:actions get_pred(self.policy_net, obs_tensor)return actionsdef store_transition(self, obs, action, reward, next_obs): if len(self.memory) BUFFERSIZE:self.memory.pop(0) # 移除最旧的经验self.memory.append((obs, action, reward, next_obs))def loss_fn(self, obs_batch_dict, next_obs_batch_dict, action_batch, reward_batch):# 计算损失函数的示例# 这里可以使用 PPO 或其他算法的损失函数# 这里只是一个简单的示例# 将输入数据移动到模型所在的设备# device next(self.policy_net.parameters()).device # 获取模型所在的设备action_batch.to(device) # 将 action_batch 移动到同一设备reward_batch reward_batch.to(device) # 将 reward_batch 移动到同一设备i PolicyGradientif i PolicyGradient:# 策略梯度 log_probs torch.log(self.policy_net(obs_batch_dict))log_probs log_probs.squeeze(0) # 去掉第 0 维形状变为 (3, 8)action_batch action_batch.squeeze(0) # 去掉第 0 维形状变为 (3,)selected_log_probs log_probs[range(len(action_batch)), action_batch]loss -torch.mean(selected_log_probs * reward_batch) # 计算策略梯度损失elif i DQN:if next_obs_batch_dict is None:raise ValueError(next_obs_batch_dict must be provided for DQN loss calculation)# 假设有 Q 网络 q_net 和目标 Q 网络 target_q_net if not hasattr(self, q_net): # 首次调用时初始化self.q_net Noneq_values self.q_net(obs_batch_dict)next_q_values self.target_q_net(next_obs_batch_dict).max(dim1)[0].detach()target_q_values reward_batch self.gamma * next_q_valuesloss torch.mean((q_values[range(len(action_batch)), action_batch] - target_q_values) ** 2)else:raise ValueError(Unsupported loss function type)return lossdef train_step(self ):# 经验回放训练if not self.memory:return# 简单策略梯度损失示意可替换成 PPOobs_batch, action_batch, reward_batch, _ zip(*self.memory) # zip 函数将多个可迭代对象的对应元素打包成元组。 操作符*将列表解包为独立的参数即拆开 self.memory 的外层列表。# 将 obs_batch经过memory后会变成元组tmd还得转回来 -cq20250522obs_batch_temp obs_batch[-1]obs_batch_dict {game_stats: obs_batch_temp[game_stats],bop_features: obs_batch_temp[bop_features],spatial_features: obs_batch_temp[spatial_features],bop_embeddings: obs_batch_temp[bop_embeddings]}if len(self.memory) 20: # 使用len()获取列表长度 obs_batch_temp obs_batch[-20]else:obs_batch_temp obs_batch[-1] # 如果只有一条记录使用最后一条记录next_obs_batch_dict {game_stats: obs_batch_temp[game_stats],bop_features: obs_batch_temp[bop_features],spatial_features: obs_batch_temp[spatial_features],bop_embeddings: obs_batch_temp[bop_embeddings]}action_batch torch.tensor(action_batch)reward_batch torch.tensor(reward_batch, dtypetorch.float32)# 构建loss --可替换的策略优化器PPO, A2C, BCQ, DQNloss self.loss_fn(obs_batch_dict, next_obs_batch_dict, action_batch, reward_batch)# 更新参数self.optimizer.zero_grad()loss.backward()self.optimizer.step()self.memory.clear()# 监控训练进度self.step_counter 1if self.step_counter % 100 0: # 每100步打印一次损失,保存一次网络模型print(fStep: {self.step_counter}, Loss: {loss.item()})# # 定期更新目标网络# if self.count % self.target_update_freq 0:# self.policy_net.load_state_dict(self.policy_net.state_dict())# 监控与保存import ossvaepath RL_MODEL_PATH f_{current_date}/if not os.path.exists(svaepath):os.makedirs(svaepath)self.save(svaepath ftrainer_step_{self.step_counter}.pth) # 保存模型
4-智能体网络设计 (比较复杂考虑了各种信息作为了向量输入)
这部分我没全放出来需要大家自己设计
而且这部分要搭配数据向量化处理的代码神经网络的形状也要考虑十分繁琐…… class troopsNNv2(nn.Module):def __init__(self, game_stat_dim12,bop_feat_dim43,enemy_num3,ally_num3,spatial_channels6, # 适配精简版空间通道数enemy(3) exposure(1) see(1) fire(1)output_dim8,use_opponent_modelingFalse):super(troopsNNv2, self).__init__()self.ally_num ally_num# 游戏状态信息全局统计self.mlp_game nn.Linear(game_stat_dim, 32)# 单位特征处理self.unit_fc nn.Linear(bop_feat_dim, 128)self.unit_gru nn.GRU(input_size128, hidden_size128, batch_firstTrue)......................................# 最终动作预测头7方向 停止 8个动作self.policy_head nn.Sequential(nn.Linear(512, 256),nn.ReLU(),nn.Linear(256, output_dim))def forward(self, inputs):# print(type(inputs)) # 打印类型# 将输入张量移动到设备device torch.device(DEVICE)game_stats inputs[game_stats].to(device)bop_features inputs[bop_features].to(device)spatial_features inputs[spatial_features].to(device)bop_embeddings inputs[bop_embeddings].to(device)# 对敌的张量没有t步的事情所以不用进入下面的循环enemy_strategy inputs[enemy_strategy].to(device) if self.is_using_opponent else torch.empty(0, dtypetorch.float32, devicedevice)bop_enemy_embeddings inputs[bop_enemy_embeddings].to(device) if self.is_using_opponent else torch.empty(0, dtypetorch.float32, devicedevice)batch_size, seq_len game_stats.shape[:2]outputs []for t in range(seq_len):# 当前帧特征提取g game_stats[:, t].float()bop bop_features[:, t].float()spatial spatial_features[:, t].float() # shape (B, 6, 13, 23)embeddings bop_embeddings[:, t].float() # shape (B, 3, 13, 23)f_game self.mlp_game(g) # (B, 32)# 单位特征编码bop_embed self.unit_fc(bop)_, bop_hidden self.unit_gru(bop_embed)bop_hidden bop_hidden.squeeze(0) # (B, 128)
......................................temporal_input torch.stack(outputs, dim1) # (B, T, D)_, final_hidden self.temporal_gru(temporal_input) # final_hidden: (1, B, 512)final_feature final_hidden.squeeze(0).unsqueeze(1).repeat(1, self.ally_num, 1) # (B, 3, 512)logits self.policy_head(final_feature) # (B, 3, 8)# logits_all []# logits_all.append(logits)return logits在实际编码时我曾经把3-4-1都放到一个python中但是调试起来自己都乱了所以分成这么4大块挺好的这是自己的经验。 参考
深度强化学习之模仿学习Imitation Learning-腾讯云开发者社区-腾讯云
