简介:面向移动边缘计算(MEC)场景的Python深度强化学习源码包,围绕计算卸载与资源分配两大核心问题展开,适用于通信工程、人工智能、计算机等专业的毕设项目、课程设计或科研复现。源码共19个文件,压缩包约113KB,主要包含5个Python脚本、4个Shell运行脚本、6个txt日志文件、3张结果图和1个MD说明文档。其中核心算法脚本基于深度Q网络(DQN)实现卸载与分配决策,环境建模脚本构建MEC仿真环境,绘图脚本用于输出收敛曲线与对比图,Shell脚本支持Q-learning与DQN两种方案的运行切换,整体覆盖训练、对比、绘图全流程。目前已有153人学习下载。代码内置Q-learning基线,可复现多组实验数据,方便读者对比深度强化学习与传统方法的性能差异,也可在现有状态空间、动作空间或奖励函数基础上做二次开发;日志与图片文件有助于快速验证效果、制作论文插图。下载后建议先阅读说明文档了解目录结构,源码已测试运行成功,适合需要快速入门的MEC研究者及算法复现者。
1. MEC 计算卸载为什么要用深度强化学习,而不是贪心
做移动边缘计算(MEC)计算卸载的同学,最后多少都会走到同一个路口:任务往不往边缘卸载、卸载多大比例、边缘算力给谁多分一点,这些决策既是动态的,又是连续的。传统优化方法在信道和任务到达一变的时候就要重新求解,等解出来环境已经变了。于是很多人转向深度强化学习,把 MEC 计算卸载与资源分配做成一个 Python 仿真的训练闭环,用策略网络直接输出卸载比例和资源分配比例。这个标题对应的源码,典型形态就是仿真环境、DRL 智能体、训练循环、评估脚本四件套。它适合已经会 Python,但不知道怎么把 MEC 问题写成可训练 MDP 的研究生和工程师。
2. 建模是第一步:MEC 卸载与资源分配如何写成可训练的状态、动作和奖励
2.1 单边缘节点的计算卸载场景:任务可拆分假设与动态信道
把计算卸载变成强化学习问题,第一步不是写神经网络,而是写 MDP。绝大多数深度强化学习计算卸载的源码都默认一个经典场景:一个边缘节点覆盖多个移动用户,每个用户在每个时隙产生一个计算任务,任务可以选择留在本地执行,也可以卸载到边缘服务器执行。边缘服务器算力有限,所以多个用户来抢时就需要分配。
这里有一个关键假设:任务是否可拆分。如果任务不可拆分,动作只能是“卸载/不卸载”的离散选择,适合用 DQN、PPO 的离散动作版本。如果任务可以按比例拆分,比如视频渲染、数据分析这类延迟容忍型任务,动作就是连续卸载比例,可以直接用 DDPG、SAC 这类连续控制算法。标题里同时出现“计算卸载”和“资源分配”,我一般会先做可拆分任务加连续动作,否则资源分配比例这个动作会非常不自然。
信道状态不能写成静态的。MEC 场景的价值就在于环境动态变化,所以每时隙要重新采样信道增益、任务大小和计算密度。最简做法是用np.random.exponential生成小尺度衰落,再叠加一个固定的路径损耗系数。不要把问题一开始就复杂化,例如上 3GPP 完整信道模型,会让收敛排查变得很难。
常见环境参数可以这样设定:
| 参数 | 取值范围 | 说明 |
|---|---|---|
| 用户数 | 4~20 | 用户越多,资源分配冲突越明显 |
| 任务数据量 | 200 KB ~ 2 MB | 决定传输时延和执行时延 |
| 任务计算密度 | 200~800 cycles/bit | 每 bit 需要的 CPU 周期数 |
| 本地算力 | 0.5~2 GHz | 移动设备本地执行速度 |
| 边缘节点算力 | 20~50 GHz | 所有用户共同分配 |
| 信道带宽 | 5~20 MHz | 固定带宽,先不让网络学习带宽分配 |
| 路径损耗指数 | 3.5 | 简化信道模型 |
这些值不需要和论文完全一致,但数量级要对。如果奖励函数里同时出现毫秒级时延和焦耳级能耗,不做归一化,训练大概率会翻车。
2.2 状态空间:信道、任务、队列缺一不可
状态空间承担着马尔可夫性的责任。网络在时刻 t 看到的 state,必须足以推断出下一步环境会怎么变。很多初学源码的人把状态只设成“当前任务大小”,结果策略完全不收敛,因为信道增益一变,同样的任务在边缘执行的成本就完全不同。
我常用的状态向量是每个用户五个值:
- 任务数据量
task_size - 计算密度
cpu_density - 当前信道增益
channel_gain - 边缘节点上一帧排队负载
edge_queue - 本地上一帧排队负载
local_queue
在多用户环境下,状态形状是[n_users, 5],不要把用户维度展开成一个大向量,除非你真的理解每个特征的位置含义。Actor 和 Critic 输入都会用到这个维度结构,后面代码会按特征维拼接。
有些源码会把上一时刻的卸载动作也放进状态,这相当于给网络一个“历史记忆”,在静置环境里有点用,但不建议默认加。MEC 环境通常是逐时隙独立生成任务的,上一动作与当前最优动作没有强相关,加了反而增加输入维度。
2.3 动作空间:卸载比例和资源分配比例如何避免失效约束
连续动作设计是这个标题里最容易被低估的部分。每个用户至少有两个动作:卸载比例 λ,以及从边缘节点分配到该用户的算力比例 μ。λ 的自然范围是 [0,1],μ 不能独立取 [0,1],因为所有用户的 μ 求和必须等于 1,否则边缘算力被超额分配。
常见的处理方式有两种。第一种是在 Actor 输出后对 μ 做 softmax,保证求和为 1;第二种是在环境 step 内部做 softmax,网络输出的只是“原始分配分数”。我强烈建议在环境外部、也就是智能体动作映射时就把 softmax 做掉,否则存储进经验回放的动作和真正送往环境的动作不是同一个分布,Critic 学到的 Q 值是乱的。
动作维度可以这样定义:
| 动作分量 | 维度 | 取值范围 | 含义 |
|---|---|---|---|
| unload_ratio | n_users | [0,1] | 每个用户卸载到边缘的任务比例 |
| compute_alloc | n_users | 和为 1 | 边缘算力资源分配给每个用户的比例 |
如果边缘节点还区分 CPU 池和 GPU 池,可以继续扩展成[λ, cpu_alloc, gpu_alloc],在动作映射里分别做 softmax。标题里只写了“资源分配”,我一版先做 CPU 算力分配,够用了。
2.4 奖励函数:时延、能耗、队列惩罚的加权尺度
奖励函数决定了网络最终在优化什么。MEC 计算卸载的典型优化目标是“在满足任务时延的前提下降低设备能耗”,但工业场景里更重要的是不能让边缘队列无限积压。所以奖励一般写成三个部分的加权组合:
时延项用每个任务的实际完成时间,包括本地执行时间、传输时间和边缘执行时间。因为任务可拆分,本地部分和边缘部分是并行进行的,所以单个任务总时延取两端执行时间中的最大值,而不是求和。这一点很多初次实现会写错。
能耗项只统计移动设备侧:本地执行能耗加上传输能耗。边缘服务器能耗一般由运营商承担,不放进用户奖励,否则网络会为了降低边缘能耗而把所有任务留在本地。
队列惩罚项可以加到奖励中,强制网络避免把全部任务都卸载到一个边缘节点。最简单的做法是看edge_queue的均值,超过阈值就扣分。也可以直接对每个用户的total_time / total_time_baseline做惩罚。
我常用的奖励公式是:
$$r_t = -\alpha \cdot \frac{T_{total}}{T_{local}} - \beta \cdot \frac{E_{total}}{E_{local}} - \gamma \cdot \frac{Q_{edge}}{Q_{threshold}}$$
其中分母都使用“任务全部在本地执行”的时延和能耗作为归一化基线,这样各个项的尺度接近,不会出现能量项把时延项淹没的情况。参数 α、β、γ 在 config 里配置,先取 α=0.5、β=0.5、γ=0.1,再逐步调。
奖励设计完,再回头看“为什么不用贪心”:贪心只能看到当前时隙最优,看不到卸载对边缘队列的下一时刻影响;深度强化学习源码的价值,就是让策略网络自己学到这种跨时隙权衡。
3. 代码结构先行:用 Python 搭一套可复现的 MEC-DRL 训练框架
3.1 项目文件划分:环境、智能体、训练、配置四模块
有了 MDP 设计,就可以组织 Python 源码了。很多网上流传的 MEC+DRL 源码是单文件几百行,环境、网络、训练全混在一个脚本里,能跑,但换环境参数时要改十几个地方。如果你要自己复现或改造,我建议先按四个模块拆:
mec_drl/ ├── config.py # 训练参数和环境参数集中管理 ├── envs/ │ └── mec_env.py # MEC 仿真环境,实现 reset 和 step ├── algos/ │ └── ddpg.py # Actor、Critic、ReplayBuffer、update ├── train.py # 训练入口 └── evaluate.py # 加载模型并评估这个结构不是唯一答案,但它能让“环境逻辑”和“算法逻辑”解耦。后面你想把 DDPG 换成 SAC,只需要改algos/目录,不需要动环境。
3.2 配置管理:先把 seed 和所有参数写进 config.py
复现问题一半是随机种子引起的。我会把 seed、环境参数、网络参数全部放到 config.py 里,训练脚本只读取配置,不允许散落魔法数。
# config.py SEED = 42 N_USERS = 10 EPISODES = 500 EPISODE_LEN = 100 BATCH_SIZE = 128 BUFFER_SIZE = 200000 GAMMA = 0.99 TAU = 0.005 LR_ACTOR = 1e-4 LR_CRITIC = 1e-3 NOISE_STD = 0.2 HIDDEN_DIM = 256 # 环境参数 PAYLOAD_RANGE = (200 * 1024, 2 * 1024 * 1024) # byte CPU_DENSITY_RANGE = (200, 800) # cycles/byte F_EDGE = 30e9 # Hz F_LOCAL = 1e9 # Hz BANDWIDTH = 10e6 # Hz NOISE_POWER = 1e-13 # 简化噪声功率 P_LOCAL = 1.5 # 本地执行功率,单位 W P_TRANS = 0.5 # 传输功率,单位 W REWARD_ALPHA = 0.5 REWARD_BETA = 0.5 REWARD_GAMMA = 0.1注意数据单位。PAYLOAD_RANGE如果用 bit,传输速率也用 bit/s,传输时延公式才成立。我习惯统一用 byte 和 Hz,计算执行时延时用payload * cpu_density / f,只要单位一致就不会出问题。
3.3 环境类:Numpy 即可,别把仿真搬到 GPU
MEC 环境的特点是每个 step 都要生成随机任务和信道,这个过程用 Numpy 在 CPU 上跑是最高效的。只有神经网络的前向和反向才应该用 GPU。很多人把环境也写进 tensor,每个 step 都在 GPU 上做大数组拼接,训练速度反而慢。
下面是一个最小但可训练的环境实现:
# envs/mec_env.py import numpy as np def softmax(x): e_x = np.exp(x - np.max(x, axis=-1, keepdims=True)) return e_x / np.sum(e_x, axis=-1, keepdims=True) class MECEnv: def __init__(self, cfg): self.cfg = cfg self.n_users = cfg["N_USERS"] self.f_edge = cfg["F_EDGE"] self.f_local = cfg["F_LOCAL"] self.bandwidth = cfg["BANDWIDTH"] self.noise_power = cfg["NOISE_POWER"] def reset(self, seed): rng = np.random.default_rng(seed) # 每个用户独立产生一个计算任务 self.task_size = rng.uniform(*self.cfg["PAYLOAD_RANGE"], size=self.n_users) self.cpu_density = rng.uniform(*self.cfg["CPU_DENSITY_RANGE"], size=self.n_users) # 简化信道:指数衰落乘以路径损耗 self.channel_gain = rng.exponential(scale=1.0, size=self.n_users) / (1.0 ** 3.5) self.edge_queue = np.zeros(self.n_users) self.local_queue = np.zeros(self.n_users) return self._get_state().astype(np.float32) def _get_state(self): # 每个用户 5 维状态:任务量、计算密度、信道、边缘队列、本地队列 return np.stack([ self.task_size, self.cpu_density, self.channel_gain, self.edge_queue, self.local_queue, ], axis=1) def _transmission_rate(self): snr = self.channel_gain / self.noise_power return self.bandwidth * np.log2(1.0 + snr) def step(self, actions): # actions shape [n_users, 2] = [卸载比例, 算力分配分数] unload_ratio = np.clip(actions[:, 0], 0.0, 1.0) compute_alloc = softmax(actions[:, 1]) rate = self._transmission_rate() # 本地执行部分:任务量 * (1 - 卸载比例) local_cycles = self.task_size * self.cpu_density * (1.0 - unload_ratio) local_time = self.local_queue + local_cycles / self.f_local local_energy = self.cfg["P_LOCAL"] * (local_cycles / self.f_local) # 卸载部分:传输时延 + 边缘执行时延 trans_time = self.task_size * unload_ratio / rate edge_cycles = self.task_size * self.cpu_density * unload_ratio edge_time = trans_time + edge_cycles / (compute_alloc * self.f_edge + 1e-8) total_time = np.maximum(local_time, edge_time) total_energy = local_energy + self.cfg["P_TRANS"] * trans_time # 以全本地执行为归一化基线 t_baseline = self.task_size * self.cpu_density / self.f_local e_baseline = self.cfg["P_LOCAL"] * t_baseline reward = - self.cfg["REWARD_ALPHA"] * np.mean(total_time / t_baseline) \ - self.cfg["REWARD_BETA"] * np.mean(total_energy / e_baseline) \ - self.cfg["REWARD_GAMMA"] * np.mean(self.edge_queue / (edge_cycles / self.f_edge + 1e-8)) # 更新队列状态,供下一帧使用 self.edge_queue = edge_cycles / self.f_edge self.local_queue = local_cycles / self.f_local next_state = self._get_state().astype(np.float32) return next_state, reward, False, {"time": total_time, "energy": total_energy}这段代码里有几个关键点:softmax放在环境内部,是为了让算力分配比例总和为 1;np.maximum(local_time, edge_time)对应任务并行处理;归一化奖励让reward数量级在-(alpha + beta)附近。done 恒为 False,训练循环用EPISODE_LEN控制结束。
3.4 经验回放和训练循环:先采一段再更新
DDPG 是 off-policy 算法,必须配经验回放。经验回放就是固定容量队列,每一条记录是(state, action, reward, next_state, done)。
# train.py from collections import deque import random import numpy as np class ReplayBuffer: def __init__(self, max_size): self.buf = deque(maxlen=max_size) def push(self, *transition): self.buf.append(transition) def sample(self, batch_size): batch = random.sample(self.buf, batch_size) return tuple(np.stack(x) for x in zip(*batch)) def __len__(self): return len(self.buf) def train(): env = MECEnv(cfg) agent = DDPG(state_dim=5, action_dim=2 * cfg["N_USERS"], cfg=cfg) buffer = ReplayBuffer(cfg["BUFFER_SIZE"]) for episode in range(cfg["EPISODES"]): obs = env.reset(cfg["SEED"] + episode * 100) episode_reward = 0 for step in range(cfg["EPISODE_LEN"]): action = agent.select_action(obs, add_noise=True) next_obs, reward, done, info = env.step(action) buffer.push(obs, action, reward, next_obs, float(done)) if len(buffer) > cfg["BATCH_SIZE"]: batch = buffer.sample(cfg["BATCH_SIZE"]) agent.update(batch) obs = next_obs episode_reward += reward if episode % 20 == 0: print(f"Episode {episode}, Reward {episode_reward:.2f}")select_action中加噪声是为了探索。后面在 DDPG 源码里会看到,动作噪声一般加在网络输出上,而不是加到已经经过 sigmoid 的比例上,否则探索会被约束扭曲。
4. 深度强化学习核心源码:DDPG 如何把状态映射成卸载动作和资源分配
4.1 为什么选 DDPG 而不是 DQN:连续动作和样本效率
当前深度强化学习算法有很多,标题里的“深度强化学习”没有限定算法,但做连续卸载比例和资源分配,最稳妥的默认选择是 DDPG。DQN 只能输出离散动作,要把卸载比例离散成 0、0.5、1 三档,边缘算力分配更是不好离散。PPO 也能处理连续动作,但它是 on-policy 算法,每个 epoch 收集的数据用一次就丢,在 MEC 仿真这种样本生成成本相对高的环境里,训练效率明显低于 DDPG 这类 off-policy 算法。
DDPG 的问题在于对超参数敏感,后面避坑章会细说。SAC 是 DDPG 的强化版本,但要调的熵温度、双 Q 网络参数更多。我一般先跑通 DDPG,再逐步加复杂度。
4.2 Actor-Critic 网络:动作映射里暗藏约束
Actor 网络输入状态,输出原始动作向量。注意不要直接输出卸载比例和算力分配比例,而是输出 raw 值,再通过 sigmoid 和 softmax 做可微约束映射。
# algos/ddpg.py import torch import torch.nn as nn import torch.nn.functional as F class Actor(nn.Module): def __init__(self, state_dim, n_users, hidden_dim=256): super().__init__() self.n_users = n_users self.net = nn.Sequential( nn.Linear(state_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, 2 * n_users), ) # 最后一层权重小,避免初始动作过于饱和 self.net[-1].weight.data.mul_(0.1) self.net[-1].bias.data.fill_(0.0) def forward(self, s): return self.net(s) def get_action(self, s): raw = self.net(s) unload = torch.sigmoid(raw[..., :self.n_users]) alloc = torch.softmax(raw[..., self.n_users:], dim=-1) return torch.cat([unload, alloc], dim=-1) class Critic(nn.Module): def __init__(self, state_dim, action_dim, hidden_dim=256): super().__init__() self.net = nn.Sequential( nn.Linear(state_dim + action_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, 1), ) def forward(self, s, a): return self.net(torch.cat([s, a], dim=-1))get_action是整条动作约束的核心。raw[..., :n_users]经过 sigmoid 得到卸载比例,天然落在 [0,1] 区间;raw[..., n_users:]经过 softmax 得到资源分配比例,天然满足所有用户比例求和为 1。这样环境里的sim.step不需要再做 softmax,训练和推理路径完全一致。
4.3 更新流程:目标网络、软更新与经验回放
DDPG 更新时,Critic 的目标值依赖目标 Actor 计算下一状态动作,目标网络参数通过软更新缓慢逼近在线网络,这是稳定性的核心。
class DDPG: def __init__(self, state_dim, n_users, cfg): self.n_users = n_users self.cfg = cfg self.actor = Actor(state_dim, n_users, cfg["HIDDEN_DIM"]) self.critic = Critic(state_dim, 2 * n_users, cfg["HIDDEN_DIM"]) self.actor_target = Actor(state_dim, n_users, cfg["HIDDEN_DIM"]) self.critic_target = Critic(state_dim, 2 * n_users, cfg["HIDDEN_DIM"]) self.actor_target.load_state_dict(self.actor.state_dict()) self.critic_target.load_state_dict(self.critic.state_dict()) self.actor_opt = torch.optim.Adam(self.actor.parameters(), lr=cfg["LR_ACTOR"]) self.critic_opt = torch.optim.Adam(self.critic.parameters(), lr=cfg["LR_CRITIC"]) self.noise_std = cfg["NOISE_STD"] def select_action(self, s, add_noise=True): s = torch.FloatTensor(s).unsqueeze(0) with torch.no_grad(): raw = self.actor(s) if add_noise: raw = raw + torch.randn_like(raw) * self.noise_std # 噪声加在 raw 上,再映射到比例 unload = torch.sigmoid(raw[..., :self.n_users]) alloc = torch.softmax(raw[..., self.n_users:], dim=-1) return torch.cat([unload, alloc], dim=-1).squeeze(0).numpy() def update(self, batch): s, a, r, s_next, done = batch s = torch.FloatTensor(s) a = torch.FloatTensor(a) r = torch.FloatTensor(r).unsqueeze(-1) s_next = torch.FloatTensor(s_next) done = torch.FloatTensor(done).unsqueeze(-1) with torch.no_grad(): a_next = self.actor_target.get_action(s_next) q_target = self.critic_target(s_next, a_next) y = r + self.cfg["GAMMA"] * (1 - done) * q_target q = self.critic(s, a) critic_loss = F.mse_loss(q, y) self.critic_opt.zero_grad() critic_loss.backward() self.critic_opt.step() # 用在线 Actor 输出动作,让 Critic Q 值最大 a_policy = self.actor.get_action(s) actor_loss = -self.critic(s, a_policy).mean() self.actor_opt.zero_grad() actor_loss.backward() self.actor_opt.step() # 软更新 tau = self.cfg["TAU"] for target_param, param in zip(self.critic_target.parameters(), self.critic.parameters()): target_param.data.copy_(tau * param + (1 - tau) * target_param) for target_param, param in zip(self.actor_target.parameters(), self.actor.parameters()): target_param.data.copy_(tau * param + (1 - tau) * target_param)这里最关键的是get_action参与了 actor loss 的梯度传播。因为 sigmoid 和 softmax 都可微,Critic 对动作的梯度能一路传到 Actor 网络,Actor 才会知道“增大卸载比例会让 Q 值变高还是变低”。如果你把动作映射写在 Numpy 环境里,不让 PyTorch 看到,Actor 就失去梯度信号。
4.4 DDPG 在 MEC 场景的常见参数与调参方向
| 参数 | 常见值 | 调参说明 |
|---|---|---|
| LR_ACTOR | 1e-4 | 太大容易震荡,太小收敛慢 |
| LR_CRITIC | 1e-3 | Critic 一般可以比 Actor 学得快一点 |
| TAU | 0.005 | 目标网络软更新系数,太大的话目标值不稳 |
| GAMMA | 0.99 | MEC 每时隙任务独立,可以适度降到 0.95 |
| BATCH_SIZE | 128 | 经验回放采样批量大小 |
| NOISE_STD | 0.2 | 探索噪声强度,后期可以衰减到 0.05 |
| HIDDEN_DIM | 256 | 两隐层,用户多时加到 512 |
如果你是首次跑这个标题的源码,先把种子固定到 42,跑 100 个 episode,看奖励是否从负值缓慢上升。只要能上升,说明 MDP 和代码结构没有大问题;如果完全不动,优先查环境 step 的返回值和奖励尺度,而不是调网络层数。
如果你的 MEC 场景里边缘节点还有 GPU 算力池,资源分配维度可以扩展成[unload_ratio, cpu_alloc, gpu_alloc],Actor 输出维度变成3 * n_users,环境里对 cpu 和 gpu 分别做 softmax。其余训练结构不变。
5. 训练避坑与常见问题:不收敛、动作退化、分配比例畸形的排查顺序
5.1 奖励曲线下降,但平均时延反而变高
现象:每几十个 episode 记录一次 reward,曲线确实在上升,但把训练好的模型拉出来看平均时延,比“全本地执行”还高。
原因:reward 函数里时延项和能耗项没有做归一化。时延是毫秒级,能耗是焦耳级,数值上能耗可能比时延大几十倍,网络实际上只优化了能耗,牺牲了时延。更隐蔽的情况是队列惩罚项权重过大,模型学会了用低卸载率压队列,但任务全留在本地。
解决:在 reward 里不要直接使用时延和能耗的原始值,而是除以“任务全本地执行”对应的基线值,让每项量纲变成 1 左右。然后在 evaluate 脚本里分别统计平均时延、平均能耗、卸载比例三项指标,而不是只看总 reward。总 reward 是训练信号,不是落地指标。
5.2 卸载比例总是 0 或 1,几乎没有中间值
现象:训练一段时间后,Actor 输出的卸载比例全部贴到边界,有的用户恒为 0,有的恒为 1,看不到连续分布的 0.3、0.6 这类折中方案。
原因:Actor 最后一层权重太大,raw 输出经过 sigmoid 后落到了饱和区。饱和区梯度极小,Actor 很难继续优化。另外,Critic 对动作的 Q 值面如果太平坦,Actor 也会被推着往边界跑。
解决:把 Actor 最后一层权重初始化乘 0.1,让初始 raw 输出在 [-1,1] 附近。探索噪声加在 raw 上,再走 sigmoid,而不是加在 sigmoid 之后。如果已经跑到边界,可以降低 NOISE_STD,同时把 Critic 学习率调低,让 Q 值先稳定下来。
5.3 算力分配比例总和不是 1,资源被超额分配
现象:打印每个 step 的 compute_alloc,每个用户的值都在 [0,1],但求和是 1.7,明显不可能。
原因:对资源分配比例的处理不一致。例如在环境 step 内部做了 softmax,但 Actor 输出返回给 Critic 的动作没有经过同一套 softmax,经验回放里存的是原始分数。Critic 看到的动作和真正输入环境的动作分布不一致,训练自然发散。
解决:把动作映射统一封装在Actor.get_action里,卸载比例走 sigmoid,资源分配走 softmax;训练、探索、目标网络、评估全部走同一个方法。不要在环境内部单独做一次 softmax,也不要在 eval 时重新写一套动作映射。
5.4 多用户场景下,某个用户被“饿死”
现象:系统平均时延和能耗都还行,但看每个用户单独统计,2 号用户平均时延是其他用户的 3 倍,算力资源长期分配为 0.05。
原因:reward 函数用的是所有用户的平均值,网络只要牺牲一个用户就能换来整体指标提升,在资源总量受限时尤其容易发生。这在 MEC 资源分配里属于公平性问题。
解决:在 reward 中加入“最差用户惩罚”,例如把np.mean(total_time / t_baseline)改成np.mean(...) + np.max(total_time / t_baseline)。或者限制每个用户的最小资源分配比例,在环境 step 里做compute_alloc = (1 - epsilon) * softmax(raw) + epsilon / n_users,保证每个人至少拿到一点资源。
5.5 换一台机器或换一个 seed,结果完全对不上
现象:同一份源码,在别人机器上训练 500 episode 的 reward 已经到 -0.8,自己跑还是 -1.5。代码没有任何改动。
原因:没有固定随机种子。np.random.default_rng(seed)只控制了环境里的随机数,但 Actor 初始化、经验回放采样、探索噪声用的分别是 PyTorch 和 Python 的全局随机源,只要其中一个没固定,结果就不可复现。
解决:在训练入口统一固定np.random.seed(seed)、random.seed(seed)、torch.manual_seed(seed)。如果用的是 CUDA,再加torch.backends.cudnn.deterministic = True。最后把 seed 连同训练参数一起写进日志文件,这才是深度强化学习源码能“复现”的基础。
6. 从仿真到落地:四个验证指标、基线对比和部署经验
6.1 训练完先看这四个数字,别只看 reward
仿真的终点不是模型装进onnx,而是能用指标说服自己这个方案值得投入。我每次训练完会生成一张表:
| 指标 | 正常信号 | 异常信号 |
|---|---|---|
| 平均时延 | 低于全本地执行,且曲线平稳 | 后期反弹上涨 |
| 平均能耗 | 比全本地低 30% 以上 | 无下降趋势 |
| 卸载率均值 | 0.3~0.7 之间,有分布 | 恒为 0 或 1 |
| 边缘队列积压 | 不上涨 | 持续增大 |
如果前两个指标明显优于基线,再谈“动态计算卸载层”的可部署性。否则不要继续加新功能,先修环境或奖励。
6.2 基线怎么设置,结果才有说服力
最低要求是跟“全本地执行”和“全量卸载”比,因为这两个策略不需要任何智能决策。更严格一点,还应该加随机卸载、贪心卸载两个基线。贪心策略可以设为“信道好的用户全卸载,信道差的用户不卸载”,这已经比随机策略强。如果 DRL 方案连贪心都打不过,问题大概率出在状态或奖励设计,而不是算法本身。
6.3 模型导出与部署:把 sigmoid/softmax 留在模型里
训练完的 Actor 要部署到边缘服务器,常见做法是导出 ONNX,用 ONNX Runtime 做推理。导出前最好把动作映射写进 Actor 的 forward,让卸载比例和资源分配比例在导出前后保持一致。
# 导出时使用包装后的 Actor class DeployActor(nn.Module): def __init__(self, actor): super().__init__() self.actor = actor def forward(self, s): raw = self.actor(s) unload = torch.sigmoid(raw[:, :N]) alloc = torch.softmax(raw[:, N:], dim=-1) return torch.cat([unload, alloc], dim=-1) dummy = torch.randn(1, 5) torch.onnx.export(DeployActor(agent.actor), dummy, "mec_actor.onnx", input_names=["state"], output_names=["action"])如果部署端不使用 Python,也可以把导出的 ONNX 交给 C++ 调用。MEC 场景里策略模型通常集中训练、边缘节点在线推理,模型体积不大,不需要跑到边缘节点上训练。
我自己的习惯是训练结束后把 config、seed、最终 reward 一起写入 json,和 checkpoint 放在同目录。下次任何人拿到模型,都能先复现配置,再去改网络结构。模型再复杂,也怕环境定义和随机源不清。希望帮到你。
本文还有配套的精品资源,点击获取