简介这份资源面向车联网与智能交通方向的学习者和研究者提供一套基于多智能体深度强化学习MADDPG优化车联网通信资源分配的Python实现适合具备一定强化学习基础、希望将理论落地到工程场景的个人学习使用。压缩包共24个文件以13个py源码为主另含6个pyc编译文件、4个zbak备份和1个md说明文档整体约75KB体量轻便便于快速阅读与二次修改。代码涵盖MADDPG、MADQN、DDPG等多种算法实现并配有环境模拟、经验回放、模型智能体等模块目录结构清晰方便对照理解策略网络与价值网络的协作机制。目前已有79人学习下载。通过该资源读者可掌握多智能体强化学习在通信资源分配中的建模思路、奖励机制设计与算法调优方法并在此基础上扩展实验或搭建原型系统。1. 车联网资源分配为什么需要多智能体深度强化学习从单车决策到车队博弈车联网V2X场景下的通信资源分配本质上是一个高动态、强耦合的分布式决策问题。当路侧单元RSU覆盖范围内同时存在几十甚至上百辆车每辆车都在争夺有限的时频资源块RB传统的集中式优化方法——比如凸优化或启发式调度——在毫秒级调度周期内根本来不及收敛。更麻烦的是车辆高速移动导致信道状态信息CSI快速变化集中式方案需要收集全局信息信令开销直接吃掉一大块频谱效率。多智能体深度强化学习MADRL之所以在这个场景里被反复提及核心原因是它把「全局优化」拆成了「每个智能体基于局部观测做决策通过共享奖励信号协同」。每辆车或每个V2V链路作为一个智能体观测自己的信道质量、队列长度、邻居干扰输出资源块选择和发射功率。训练完成后推理阶段只需要局部信息信令开销从O(N)降到O(1)级别。这套方案适合谁如果你在做车联网仿真、资源调度算法研究或者需要一套能跑通的Python代码来验证MADRL在V2X场景的效果下面的内容会从环境搭建、算法选型、代码实现到调参避坑一步步拆开讲。不适合纯理论推导的读者这里更偏工程落地。2. 环境搭建与仿真平台选型Python生态下怎么搭一个能跑MADRL的车联网仿真2.1 为什么选Python而不是MATLAB或NS-3车联网仿真常见的三条路MATLAB通信工具箱、NS-35G-LENA、Python自建。MATLAB的强化学习工具箱虽然成熟但多智能体扩展性差自定义环境接口繁琐NS-3精度高但和Python强化学习框架PyTorch/TensorFlow的桥接成本大调试周期长。Python方案的优势在于PyTorch/TensorFlow原生支持多智能体训练Gym/Gymnasium接口标准化numpy/scipy做信道建模足够用。代价是物理层精度不如NS-3但对于资源分配算法验证路径损耗快衰落干扰的统计模型已经能反映核心矛盾。我一般会这样搭环境# 创建虚拟环境避免和系统Python冲突 python -m venv v2x_madrl source v2x_madrl/bin/activate # Linux/Mac # v2x_madrl\Scripts\activate # Windows # 安装核心依赖 pip install torch numpy gymnasium matplotlib pandas pip install scipy # 用于信道建模中的统计计算注意PyTorch版本建议选2.x1.x在MADRL的梯度回传上偶发兼容问题。CUDA版本按自己显卡驱动选没有GPU也能跑只是训练慢。2.2 车联网仿真环境的核心模块拆解一个能用的V2X资源分配仿真环境至少包含四个模块车辆移动模型高速公路场景用恒定速度随机加速度城市场景用曼哈顿模型。我一般先用高速场景验证算法因为车辆间距大、干扰模式简单容易定位问题。信道模型V2V链路用WINNER II或3GPP TR 36.885的路径损耗公式加上对数正态阴影衰落和小尺度瑞利衰落。关键参数载波频率5.9GHz带宽10MHzRB数量25个每个180kHz。干扰计算每个RB上所有复用该RB的V2V链路产生累积干扰。这是整个仿真最耗时的部分用numpy向量化能提速10倍以上。奖励函数通常设计为加权和——吞吐量、时延、可靠性SINR达标率的加权。权重怎么设直接决定算法学到的策略偏向后面避坑章节会细说。import numpy as np class V2XEnv: def __init__(self, num_vehicles20, num_rbs25, area_size500): self.num_vehicles num_vehicles self.num_rbs num_rbs self.area_size area_size # 车辆位置随机初始化 self.positions np.random.uniform(0, area_size, (num_vehicles, 2)) # 速度 60-120 km/h 转换为 m/s self.velocities np.random.uniform(16.7, 33.3, num_vehicles) # 每个车辆的发射功率预算 (dBm) self.max_power_dbm 23 # 信道增益矩阵后续每步更新 self.channel_gains np.zeros((num_vehicles, num_vehicles, num_rbs)) def _path_loss(self, dist): # 3GPP TR 36.885 高速公路路径损耗模型 return 63.3 17.7 * np.log10(max(dist, 1.0)) def _update_channel(self): # 更新车辆位置 for i in range(self.num_vehicles): self.positions[i, 0] self.velocities[i] * 0.001 # 1ms步长 if self.positions[i, 0] self.area_size: self.positions[i, 0] 0 # 计算距离矩阵 diff self.positions[:, None, :] - self.positions[None, :, :] dists np.linalg.norm(diff, axis-1) # 路径损耗 阴影衰落 pl self._path_loss(dists) shadowing np.random.normal(0, 4, dists.shape) # 4dB标准差 # 小尺度瑞利衰落 rayleigh np.random.rayleigh(1, (self.num_vehicles, self.num_vehicles, self.num_rbs)) # 合并为信道增益 (线性域) gain_db -pl[:, :, None] shadowing[:, :, None] self.channel_gains (10 ** (gain_db / 10)) * rayleigh这段代码定义了环境骨架。_path_loss用的是3GPP高速公路模型常数63.3和17.7是标准值不要随意改。_update_channel里阴影衰落标准差4dB、瑞利衰落用rayleigh分布都是常见做法。步长1ms对应LTE-V2X的调度周期。参数说明num_vehicles20是起步值实际训练可以从10开始逐步加到50。num_rbs25对应10MHz带宽如果做更宽带宽按比例加。max_power_dbm23是车载终端典型值路侧单元可以到33dBm。2.3 Gymnasium接口封装与并行环境MADRL训练需要大量交互样本单环境串行跑太慢。Gymnasium的VectorEnv可以并行多个环境实例但车联网环境有车辆间耦合不能简单复制。我的做法是每个环境实例独立随机初始化车辆位置和信道训练时从不同实例采样相当于数据增强。import gymnasium as gym from gymnasium import spaces class V2XGymEnv(gym.Env): def __init__(self, num_vehicles20, num_rbs25): super().__init__() self.env V2XEnv(num_vehicles, num_rbs) self.num_vehicles num_vehicles self.num_rbs num_rbs # 动作空间每个智能体选择RB (0~num_rbs-1) 和功率等级 (0~3) # 这里用离散动作连续功率控制后面章节讲 self.action_space spaces.MultiDiscrete( [num_rbs] * num_vehicles ) # 观测空间每个智能体的局部观测 # [自身位置x, y, 速度, 队列长度, 各RB上的干扰估计] obs_dim 4 num_rbs self.observation_space spaces.Box( low-np.inf, highnp.inf, shape(num_vehicles, obs_dim), dtypenp.float32 ) def reset(self, seedNone): super().reset(seedseed) self.env V2XEnv(self.num_vehicles, self.num_rbs) obs self._get_obs() return obs, {} def _get_obs(self): obs np.zeros((self.num_vehicles, 4 self.num_rbs)) for i in range(self.num_vehicles): obs[i, :2] self.env.positions[i] / self.env.area_size obs[i, 2] self.env.velocities[i] / 33.3 obs[i, 3] 0.5 # 队列长度占位实际从业务模型获取 # 各RB上的干扰估计简化用上一时刻的干扰 obs[i, 4:] np.random.uniform(0, 1, self.num_rbs) return obs.astype(np.float32) def step(self, actions): # actions: shape (num_vehicles,) self.env._update_channel() rewards np.zeros(self.num_vehicles) # 计算每个RB上的干扰 for rb in range(self.num_rbs): users_on_rb np.where(actions rb)[0] if len(users_on_rb) 1: continue for i in users_on_rb: interference 0 for j in users_on_rb: if j ! i: interference self.env.channel_gains[j, i, rb] sinr (self.env.channel_gains[i, i, rb] * 0.2) / (interference 1e-10) rewards[i] np.log2(1 sinr) # 香农容量 # 惩罚冲突多个用户选同一RB且SINR过低 done False truncated False return self._get_obs(), rewards, done, truncated, {}MultiDiscrete动作空间让每个智能体独立选RB这是最直接的做法。观测里包含了归一化位置、速度和干扰估计。step函数里计算SINR和香农容量作为奖励注意0.2是发射功率的线性值23dBm约200mW归一化后取0.2实际应该从动作里解码。提示_get_obs里的干扰估计用了随机数占位真实实现应该从上一时刻的SINR反推。这个简化在训练初期影响不大但后期会限制策略上限。3. MADRL算法选型与实现MADDPG、QMIX还是MAPPO3.1 三种主流算法的适用边界车联网资源分配场景下算法选型取决于三个维度动作空间是离散还是连续、智能体之间是合作还是竞争、通信约束是否允许共享全局信息。MADDPGMulti-Agent DDPG适合连续动作空间比如功率控制。每个智能体有独立的Actor网络Critic网络能看到全局信息包括其他智能体的动作。缺点是Critic输入维度随智能体数量线性增长20个车以上训练不稳定。QMIX适合离散动作空间且智能体完全合作。它用混合网络把每个智能体的Q值非线性组合成全局Q值满足单调性约束。车联网里如果所有车辆共享同一个目标最大化总吞吐量QMIX很合适。但如果有竞争关系比如不同运营商的车QMIX的单调性假设就不成立。MAPPOMulti-Agent PPO是PPO的多智能体版本支持离散和连续动作训练稳定性比MADDPG好。它用集中式Critic分布式Actor和MADDPG架构类似但PPO的裁剪机制让更新更平滑。我目前最常用MAPPO因为调参少、复现性好。算法动作空间智能体关系通信需求训练稳定性MADDPG连续合作/竞争集中式Critic中等QMIX离散完全合作全局状态较高MAPPO离散/连续合作/竞争集中式Critic高3.2 MAPPO核心代码实现下面是一个简化版MAPPO实现去掉了GAE和裁剪的细节保留核心逻辑。完整训练循环在下一节。import torch import torch.nn as nn import torch.nn.functional as F class Actor(nn.Module): def __init__(self, obs_dim, action_dim, hidden128): super().__init__() self.fc1 nn.Linear(obs_dim, hidden) self.fc2 nn.Linear(hidden, hidden) self.mean nn.Linear(hidden, action_dim) self.log_std nn.Parameter(torch.zeros(action_dim)) def forward(self, obs): x F.relu(self.fc1(obs)) x F.relu(self.fc2(x)) mean self.mean(x) std self.log_std.exp().expand_as(mean) return mean, std class Critic(nn.Module): def __init__(self, global_obs_dim, hidden256): super().__init__() self.fc1 nn.Linear(global_obs_dim, hidden) self.fc2 nn.Linear(hidden, hidden) self.v nn.Linear(hidden, 1) def forward(self, global_obs): x F.relu(self.fc1(global_obs)) x F.relu(self.fc2(x)) return self.v(x) class MAPPO: def __init__(self, num_agents, obs_dim, action_dim, lr3e-4, gamma0.99, clip0.2): self.num_agents num_agents self.gamma gamma self.clip clip # 每个智能体独立Actor共享Critic self.actors [Actor(obs_dim, action_dim) for _ in range(num_agents)] self.critic Critic(obs_dim * num_agents) self.actor_optimizers [torch.optim.Adam(a.parameters(), lrlr) for a in self.actors] self.critic_optimizer torch.optim.Adam(self.critic.parameters(), lrlr) def select_actions(self, obs): # obs: (num_agents, obs_dim) actions [] log_probs [] for i, actor in enumerate(self.actors): mean, std actor(obs[i:i1]) dist torch.distributions.Normal(mean, std) action dist.sample() log_prob dist.log_prob(action).sum(dim-1) actions.append(action) log_probs.append(log_prob) return torch.cat(actions), torch.stack(log_probs) def update(self, obs_batch, actions_batch, old_log_probs, rewards, next_obs): # 简化版更新实际需要GAE和mini-batch global_obs obs_batch.reshape(obs_batch.shape[0], -1) values self.critic(global_obs).squeeze() returns rewards # 简化单步回报 advantages returns - values.detach() # Critic更新 critic_loss F.mse_loss(values, returns) self.critic_optimizer.zero_grad() critic_loss.backward() self.critic_optimizer.step() # Actor更新 for i, actor in enumerate(self.actors): mean, std actor(obs_batch[:, i]) dist torch.distributions.Normal(mean, std) new_log_probs dist.log_prob(actions_batch[:, i]).sum(dim-1) ratio (new_log_probs - old_log_probs[:, i]).exp() surr1 ratio * advantages surr2 torch.clamp(ratio, 1 - self.clip, 1 self.clip) * advantages actor_loss -torch.min(surr1, surr2).mean() self.actor_optimizers[i].zero_grad() actor_loss.backward() self.actor_optimizers[i].step()Actor输出高斯分布的均值和标准差连续动作采样后通过tanh映射到合法范围这里省略了。Critic输入是所有智能体观测的拼接这是集中式Critic的标准做法。update里用了简化版PPO裁剪实际训练需要多轮epoch和mini-batch。参数说明lr3e-4是PPO常用学习率gamma0.99对应约100步的有效视野clip0.2是PPO默认裁剪范围。hidden128对20个智能体够用50个以上建议加到256。3.3 训练循环与经验回放MADRL训练的核心循环采样→存储→更新。车联网环境每步1ms一个episode通常跑1000步1秒仿真时间。经验回放用简单队列容量10000条。from collections import deque def train_mappo(env, agent, episodes500, steps_per_ep1000): replay deque(maxlen10000) reward_history [] for ep in range(episodes): obs, _ env.reset() obs torch.FloatTensor(obs) ep_reward 0 for step in range(steps_per_ep): with torch.no_grad(): actions, log_probs agent.select_actions(obs) # 连续动作离散化取绝对值最大的维度作为RB选择 discrete_actions actions.argmax(dim-1).numpy() next_obs, rewards, done, truncated, _ env.step(discrete_actions) next_obs torch.FloatTensor(next_obs) rewards_t torch.FloatTensor(rewards) replay.append((obs, actions, log_probs, rewards_t, next_obs)) obs next_obs ep_reward rewards.mean().item() if len(replay) 256 and step % 4 0: # 采样mini-batch batch random.sample(replay, 64) obs_b torch.stack([b[0] for b in batch]) act_b torch.stack([b[1] for b in batch]) lp_b torch.stack([b[2] for b in batch]) rew_b torch.stack([b[3] for b in batch]) next_b torch.stack([b[4] for b in batch]) agent.update(obs_b, act_b, lp_b, rew_b, next_b) reward_history.append(ep_reward) if ep % 50 0: print(fEpisode {ep}, Avg Reward: {ep_reward/steps_per_ep:.4f}) return reward_historyselect_actions返回连续动作argmax转成离散RB索引。replay存的是连续动作和log概率更新时用。step % 4 0控制更新频率避免每步都更新导致计算量爆炸。注意random.sample需要import random且batch里的张量维度要一致。如果obs维度随episode变化比如车辆数动态增减需要padding或改用固定车辆数。4. 避坑与排查车联网MADRL训练中五个血泪教训4.1 奖励不收敛曲线像心电图现象训练500个episode平均奖励在-10到5之间震荡没有上升趋势。原因奖励函数里吞吐量和时延的权重没归一化。吞吐量单位是bps时延是ms数值差3个数量级Critic学到的梯度被吞吐量主导时延项形同虚设。解决所有奖励分量归一化到[0,1]再加权。吞吐量除以理论最大值带宽×log2(1SINR_max)时延除以最大容忍时延比如100ms。权重从均匀开始根据业务需求微调。4.2 智能体学会“摆烂”全部选同一个RB现象训练后期所有车辆都选RB 0其他RB空着。总吞吐量反而下降。原因奖励函数里没有冲突惩罚。如果多个用户选同一RB但SINR还行奖励是正的智能体没有动力分散。加上探索噪声衰减太快早期就锁定了局部最优。解决奖励里加冲突惩罚项——同一RB上用户数超过阈值比如3个时每个用户扣固定惩罚。同时提高探索噪声的衰减周期从1000步延到5000步。4.3 Critic损失爆炸梯度变成NaN现象训练到200 episode左右Critic loss突然从0.1跳到1e6然后所有参数变NaN。原因车联网环境里SINR计算有除零风险。当干扰为0时1e-10的epsilon不够SINR冲到1e10奖励爆炸Critic拟合目标值过大。解决SINR计算加双重保护——分母max(interference, 1e-6)SINR上限截断到100约20dB。奖励再做clip到[-10, 10]。梯度裁剪torch.nn.utils.clip_grad_norm_(params, 0.5)也加上。4.4 训练快但推理慢模型太大现象训练时每步10ms推理时每步50ms实时性不达标。原因Actor网络用了3层512维全连接参数量200万。车载计算单元算力有限。解决Actor减到2层128维参数量降到5万。推理速度提升到5ms以内。精度损失不到3%但实时性满足要求。Critic只在训练时用可以保持大网络。4.5 换场景后性能崩塌现象高速场景训练的模型直接用到城市场景吞吐量掉40%。原因观测里的位置归一化用了固定area_size500城市场景区域小归一化后数值分布偏移。信道模型参数也不同。解决观测归一化用动态统计——维护一个running mean/std每步更新。或者训练时随机化场景参数区域大小、车辆密度、速度范围做domain randomization。5. 进阶技巧用课程学习和参数共享把训练效率翻倍5.1 课程学习从简单场景逐步加难度车联网MADRL训练最大的成本是样本量。直接上20辆车25个RB收敛需要上万episode。课程学习的思路是先训5辆车10个RB策略初步成型后逐步增加车辆数和RB数Actor网络复用Critic重新初始化。def curriculum_train(agent, stages): # stages: [(num_vehicles, num_rbs, episodes), ...] for num_v, num_rb, eps in stages: env V2XGymEnv(num_vehiclesnum_v, num_rbsnum_rb) # 扩展Actor输出维度如果RB数变了 if num_rb ! agent.num_rbs: agent.expand_action_space(num_rb) train_mappo(env, agent, episodeseps) print(fStage done: {num_v} vehicles, {num_rb} RBs)stages典型配置[(5,10,200), (10,15,300), (20,25,500)]。总episode从1000降到1000但每阶段样本效率更高。实测收敛速度提升2-3倍。5.2 参数共享所有智能体共用一个Actor如果所有车辆同构相同发射功率、相同业务类型Actor网络可以共享。20个独立Actor变成1个共享Actor参数量降到1/20训练时每个智能体的经验都用来更新同一个网络样本效率提升20倍。class SharedActorMAPPO: def __init__(self, num_agents, obs_dim, action_dim): self.actor Actor(obs_dim, action_dim) # 只有一个 self.critic Critic(obs_dim * num_agents) self.actor_optimizer torch.optim.Adam(self.actor.parameters(), lr3e-4) def select_actions(self, obs): # 所有智能体用同一个Actor mean, std self.actor(obs) # obs: (num_agents, obs_dim) dist torch.distributions.Normal(mean, std) actions dist.sample() log_probs dist.log_prob(actions).sum(dim-1) return actions, log_probs共享Actor的代价是失去异构性——如果不同车辆有不同的功率预算或业务优先级共享网络学不到差异化策略。这时候可以用条件Actor输入里加一个one-hot的车辆类型编码。5.3 验证方法怎么确认策略真的学到了东西训练曲线好看不代表策略有效。我一般做三个验证基线对比随机分配、轮询调度、贪心SINR最大化。MADRL策略至少要比贪心高10%才算有效。泛化测试用训练时没见过的车辆数比如训练20辆测试30辆跑100个episode看性能下降幅度。下降超过20%说明过拟合。消融实验去掉集中式Critic改成独立Critic看性能掉多少。如果掉得不多说明智能体间耦合弱可以用更简单的独立PPO。def evaluate(env, agent, episodes100): total_throughput [] for _ in range(episodes): obs, _ env.reset() obs torch.FloatTensor(obs) ep_throughput 0 for _ in range(1000): with torch.no_grad(): actions, _ agent.select_actions(obs) discrete_actions actions.argmax(dim-1).numpy() obs, rewards, _, _, _ env.step(discrete_actions) obs torch.FloatTensor(obs) ep_throughput rewards.sum().item() total_throughput.append(ep_throughput) return np.mean(total_throughput), np.std(total_throughput)evaluate跑100个episode取均值和标准差。标准差大说明策略不稳定需要检查探索噪声是否关掉了。我自己的习惯是每次改完奖励函数或网络结构先跑50个episode看趋势趋势对了再跑完整训练。不要一上来就5000 episode电费和时间都耗不起。另外随机种子至少设3个取平均不然单次结果可能是玄学。希望帮到你。本文还有配套的精品资源点击获取