简介:本资源面向通信、人工智能与边缘计算方向的学生及研究人员,提供一套基于无人机辅助移动边缘计算的计算卸载优化完整实现,采用深度确定性策略梯度(DDPG)方法求解卸载决策问题。压缩包共17个文件,以16个Python源码与1份Markdown说明为主,整体约44KB,代码含详细注释,新手也能理解。内容涵盖UAV_env.py环境建模、ddpg_algo.py算法实现、state_normalization.py状态归一化,并附Actor-Critic、DQN、Edge_only、Local_only等对比方案,便于横向比较不同卸载策略的性能差异。已有130人学习下载,适合作为毕业设计、期末大作业或课程设计的高分参考,也可用于复现实验、理解强化学习在移动边缘计算中的落地思路,部署简单,调试完善,具备较高实际应用价值。
1. 无人机辅助移动边缘计算卸载:为什么 DDPG 是当前最稳的落地选择
城区巡检场景里,一架四旋翼悬停在高楼侧面,机载视觉模型每秒吐出 30 帧待推理图像,而机载算力只有几 TOPS,本地跑一帧要 200ms 以上。把任务全部丢给地面基站边缘服务器,回传链路又因为遮挡和距离抖动,时延忽高忽低。这种「本地算不动、全传又不稳」的处境,正是无人机辅助移动边缘计算(UAV-assisted MEC)要解决的核心问题:让无人机在飞行过程中,动态决定每个计算任务是本地执行、卸载到边缘服务器,还是按比例拆分并行执行。
计算卸载本身不新鲜,难的是无人机一直在动。信道增益随位置连续变化,任务队列随到达率波动,电池余量还在持续下降,这是一个典型的高维连续状态、连续动作决策问题。传统凸优化需要每帧重解,Q-learning 又扛不住连续动作空间。深度确定性策略梯度(DDPG)用 Actor-Critic 结构直接输出连续卸载比例,配合经验回放和目标网络,成了目前公开方案里复现成本最低、效果最稳的一条路。这篇笔记面向想用 Python 把整套仿真跑起来的工程师,从环境建模、网络搭建到训练调参和踩坑,一步步拆开讲。
2. 把 MEC 卸载建模成 MDP:状态、动作、奖励怎么定
2.1 无人机 MEC 系统的四个关键量
在写任何代码之前,必须先把物理模型落到公式上,否则 DDPG 训练出来的策略没有物理意义。一个最小可用的无人机辅助 MEC 系统包含四个量:无人机位置、任务队列、信道状态、能耗预算。
设无人机在第 $t$ 个时隙的位置为 $q_t=(x_t,y_t,H)$,$H$ 是飞行高度。地面边缘服务器固定在原点。无人机到服务器的距离 $d_t=\sqrt{x_t^2+y_t^2+H^2}$,视距链路下的信道增益用自由空间模型:
$$h_t = \frac{\beta_0}{d_t^2}$$
$\beta_0$ 是参考距离 1m 处的信道增益,典型取值 $10^{-3}$ 到 $10^{-4}$ 之间,取决于载波频率。这个值直接决定卸载速率上限,设错了后面所有时延数字都是假的。
任务队列用 $Q_t$ 表示,单位是比特。每个时隙新到达任务 $A_t$,服从泊松分布,均值 $\lambda$。队列演化:
$$Q_{t+1} = \max(Q_t - D_t, 0) + A_t$$
$D_t$ 是本时隙实际处理掉的比特数,由卸载决策决定。能耗方面,本地计算功耗 $P_{loc}$,传输功耗 $P_{tx}$,飞行推进功耗 $P_{fly}$,三者相加不能超过电池剩余能量 $E_t$。
2.2 状态空间与动作空间的定义
状态向量我一般设计成 6 维:
import numpy as np def build_state(uav_pos, queue, channel_gain, energy, task_size, deadline): """ 构造 DDPG 的状态向量 uav_pos: (x, y) 归一化到 [-1, 1] queue: 当前队列长度,归一化到 [0, 1] channel_gain: 归一化信道增益 energy: 剩余电量比例 [0, 1] task_size: 当前任务大小归一化 deadline: 剩余时延容忍归一化 """ state = np.array([ uav_pos[0] / 500.0, # x 坐标,假设活动半径 500m uav_pos[1] / 500.0, # y 坐标 min(queue / 1e7, 1.0), # 队列,上限 10Mbit min(channel_gain / 1e-3, 1.0), energy, # 已经是比例 min(task_size / 5e6, 1.0), ], dtype=np.float32) return state归一化不是可选项。DDPG 的 Actor 网络最后一层通常是 tanh,输出范围 [-1,1],如果状态量纲差几个数量级,Critic 的 Q 值估计会剧烈震荡,训练曲线就是一条心电图。我见过太多人卡在这里,以为是算法不行,其实是状态没归一化。
动作空间设计成 2 维连续量:
def build_action(raw_action): """ raw_action: Actor 网络输出,范围 [-1, 1] 返回:卸载比例 offload_ratio, 飞行方向角 heading """ offload_ratio = (raw_action[0] + 1.0) / 2.0 # 映射到 [0, 1] heading = raw_action[1] * np.pi # 映射到 [-pi, pi] return offload_ratio, heading卸载比例 0 表示全本地,1 表示全卸载。飞行方向角控制无人机往哪个方向飞,这决定了下一时隙的信道质量。把飞行也纳入动作空间,是无人机 MEC 区别于固定边缘节点的关键,也是训练难度上升的主要原因。
2.3 奖励函数:时延、能耗、队列的加权博弈
奖励函数是整个方案里最需要反复调的部分。常见做法是加权和:
$$r_t = -(\omega_1 T_t + \omega_2 E_t + \omega_3 Q_t)$$
$T_t$ 是总时延,$E_t$ 是总能耗,$Q_t$ 是队列积压惩罚。三个权重不能拍脑袋定。我的经验是先把 $\omega_3$ 设大一点(比如 0.5),让智能体先学会别把队列堆爆,再逐步降低,去优化时延和能耗。
def compute_reward(delay, energy, queue, w1=0.4, w2=0.3, w3=0.5): """ delay: 归一化时延 energy: 归一化能耗 queue: 归一化队列长度 """ reward = -(w1 * delay + w2 * energy + w3 * queue) # 队列溢出直接给大惩罚 if queue > 0.95: reward -= 10.0 return reward提示:奖励里的队列惩罚项如果一开始设太小,智能体会学出「一直卸载、不管队列」的短视策略,后期很难纠正。先重罚队列,再放开优化目标,收敛更稳。
3. DDPG 网络搭建:Actor、Critic 与目标网络的 Python 实现
3.1 Actor 与 Critic 的网络结构选择
DDPG 的核心是两套网络。Actor 输入状态、输出动作,Critic 输入状态加动作、输出 Q 值。对于上面 6 维状态、2 维动作的问题,不需要很深。我一般用两个隐藏层,每层 256 个神经元,ReLU 激活。层数再深,训练时间翻倍,效果提升有限,还容易过拟合到某条特定轨迹。
import torch import torch.nn as nn import torch.nn.functional as F class Actor(nn.Module): def __init__(self, state_dim=6, action_dim=2, hidden=256): super().__init__() self.fc1 = nn.Linear(state_dim, hidden) self.fc2 = nn.Linear(hidden, hidden) self.out = nn.Linear(hidden, action_dim) def forward(self, state): x = F.relu(self.fc1(state)) x = F.relu(self.fc2(x)) return torch.tanh(self.out(x)) # 输出限制在 [-1, 1] class Critic(nn.Module): def __init__(self, state_dim=6, action_dim=2, hidden=256): super().__init__() self.fc1 = nn.Linear(state_dim, hidden) self.fc2 = nn.Linear(hidden + action_dim, hidden) self.out = nn.Linear(hidden, 1) def forward(self, state, action): x = F.relu(self.fc1(state)) x = torch.cat([x, action], dim=1) # 动作在第二层才拼接 x = F.relu(self.fc2(x)) return self.out(x)Critic 里动作拼接的位置有讲究。放在第一层之后拼接,是 DDPG 原论文的做法,能让网络先提取状态特征再融合动作。如果放在输入层直接拼,状态和动作量纲不一致时,Q 值估计会偏。
3.2 经验回放池与目标网络软更新
经验回放池解决样本相关性,目标网络解决 Q 值过估计。这两个机制缺一不可。
import random from collections import deque class ReplayBuffer: def __init__(self, capacity=100000): self.buffer = deque(maxlen=capacity) def push(self, state, action, reward, next_state, done): self.buffer.append((state, action, reward, next_state, done)) def sample(self, batch_size): batch = random.sample(self.buffer, batch_size) state, action, reward, next_state, done = zip(*batch) return (np.array(state), np.array(action), np.array(reward), np.array(next_state), np.array(done)) def __len__(self): return len(self.buffer)目标网络的软更新系数 $\tau$ 一般取 0.001 到 0.005。太大,目标网络跟得太快,训练不稳;太小,Q 值更新滞后,收敛慢。
def soft_update(target_net, online_net, tau=0.005): for target_param, param in zip(target_net.parameters(), online_net.parameters()): target_param.data.copy_(tau * param.data + (1 - tau) * target_param.data)3.3 训练主循环与探索噪声
DDPG 是离线策略算法,但训练初期必须加探索噪声,否则 Actor 输出几乎恒定,经验回放池里全是相似样本。常用 Ornstein-Uhlenbeck 噪声或高斯噪声。工程上高斯噪声更简单,衰减策略也更好控。
def train_step(agent, buffer, batch_size=128, gamma=0.99): if len(buffer) < batch_size: return state, action, reward, next_state, done = buffer.sample(batch_size) state = torch.FloatTensor(state) action = torch.FloatTensor(action) reward = torch.FloatTensor(reward).unsqueeze(1) next_state = torch.FloatTensor(next_state) done = torch.FloatTensor(done).unsqueeze(1) # Critic 更新 with torch.no_grad(): next_action = agent.actor_target(next_state) target_q = agent.critic_target(next_state, next_action) target_q = reward + gamma * (1 - done) * target_q current_q = agent.critic(state, action) critic_loss = F.mse_loss(current_q, target_q) agent.critic_optimizer.zero_grad() critic_loss.backward() agent.critic_optimizer.step() # Actor 更新 actor_loss = -agent.critic(state, agent.actor(state)).mean() agent.actor_optimizer.zero_grad() actor_loss.backward() agent.actor_optimizer.step() soft_update(agent.actor_target, agent.actor) soft_update(agent.critic_target, agent.critic)gamma取 0.99 是默认值,但如果你的时隙间隔是秒级、任务时延容忍只有几百毫秒,折扣因子要调小到 0.9 左右,让智能体更关注即时回报。这个参数和物理时隙长度强相关,不能照抄。
4. 训练不收敛、奖励震荡:DDPG 在 MEC 场景的避坑清单
4.1 现象:奖励曲线前期上升后突然崩掉
原因通常是 Critic 过估计导致 Actor 输出饱和。DDPG 的 Q 值容易越估越大,Actor 为了追高 Q 值,输出直接顶到 tanh 的 ±1 边界,动作失去调节能力。
解决办法有三个:一是给 Critic 加 L2 正则,权重衰减设 1e-4;二是把 Actor 的学习率降到 Critic 的十分之一,常见配置是 Actor 1e-4、Critic 1e-3;三是引入动作噪声裁剪,限制单步动作变化幅度。
# 动作平滑:限制相邻时隙动作变化 def smooth_action(prev_action, new_action, max_delta=0.2): delta = np.clip(new_action - prev_action, -max_delta, max_delta) return prev_action + delta4.2 现象:队列长度一直降不下来
先检查奖励函数里队列惩罚的权重。如果 $\omega_3$ 小于 0.2,智能体基本无视队列。其次检查状态里的队列归一化上限,如果实际队列经常超过你设的 1e7,归一化后恒等于 1,智能体分辨不出队列严重程度。
还有一种隐蔽情况:卸载速率算错了。信道增益 $\beta_0$ 取值偏大,导致智能体以为卸载很快,实际仿真里传输时延被低估。把 $\beta_0$ 从 $10^{-3}$ 调到 $10^{-4}$,策略会明显更保守。
4.3 现象:训练到后期奖励方差依然很大
这是经验回放池采样策略的问题。均匀采样会让早期差策略的样本和后期好策略的样本混在一起。可以改用优先经验回放,按 TD 误差加权采样。
class PrioritizedBuffer: def __init__(self, capacity=100000, alpha=0.6): self.capacity = capacity self.alpha = alpha self.buffer = [] self.priorities = np.zeros(capacity, dtype=np.float32) self.pos = 0 def push(self, transition, td_error): priority = (np.abs(td_error) + 1e-5) ** self.alpha if len(self.buffer) < self.capacity: self.buffer.append(transition) else: self.buffer[self.pos] = transition self.priorities[self.pos] = priority self.pos = (self.pos + 1) % self.capacity优先回放能把收敛速度提上来,但实现复杂度高,如果只是验证方案可行性,均匀采样加足够大的池子(10 万条以上)也能跑出稳定结果。
4.4 现象:换一组任务到达率,策略完全失效
DDPG 训练出来的是针对特定环境参数的策略,不是通用策略。任务到达率 $\lambda$ 从 1Mbps 变到 5Mbps,状态分布整体偏移,原策略直接崩。工程上的做法是把 $\lambda$ 也放进状态向量,训练时随机化 $\lambda$,让策略学会适应不同负载。这叫域随机化,是 sim-to-real 的常用手段。
注意:域随机化的范围要覆盖你实际部署时的负载波动区间,范围太窄没用,太宽会稀释策略在典型工况下的表现。
5. 从仿真到可复现:一套能跑通的训练脚本与验证方法
5.1 环境封装与训练入口
把物理模型封装成标准 Gym 接口,训练脚本和算法解耦,换算法不用改环境。
class UAVMECEnv: def __init__(self, area=500, height=100, beta0=1e-4, task_arrival=2e6, max_steps=200): self.area = area self.height = height self.beta0 = beta0 self.task_arrival = task_arrival self.max_steps = max_steps self.reset() def reset(self): self.uav_pos = np.random.uniform(-self.area, self.area, size=2) self.queue = 0.0 self.energy = 1.0 self.step_count = 0 return self._get_state() def step(self, action): offload_ratio, heading = self._parse_action(action) # 更新位置 self.uav_pos[0] += 10 * np.cos(heading) self.uav_pos[1] += 10 * np.sin(heading) # 计算信道、时延、能耗 dist = np.sqrt(self.uav_pos[0]**2 + self.uav_pos[1]**2 + self.height**2) channel = self.beta0 / dist**2 delay, energy_cost = self._compute_metrics(offload_ratio, channel) # 更新队列和能量 self.queue = max(self.queue - offload_ratio * 5e6, 0) + self.task_arrival self.energy -= energy_cost self.step_count += 1 done = self.step_count >= self.max_steps or self.energy <= 0 reward = compute_reward(delay / 0.5, energy_cost / 0.01, min(self.queue / 1e7, 1.0)) return self._get_state(), reward, done, {}_compute_metrics里把本地计算时延、传输时延、飞行能耗分别算出来再加权,具体公式按第 2 章的模型填。这里的关键是每一步都要检查能量是否耗尽,否则训练到后期智能体会学出「耗尽电池换低时延」的极端策略。
5.2 训练超参数配置表
| 参数 | 推荐值 | 说明 |
|---|---|---|
| Actor 学习率 | 1e-4 | 比 Critic 低一个量级 |
| Critic 学习率 | 1e-3 | 主更新网络 |
| 折扣因子 gamma | 0.95 | 时隙 1s 场景 |
| 软更新 tau | 0.005 | 目标网络跟随速度 |
| 回放池容量 | 100000 | 低于 5 万条收敛慢 |
| 批大小 | 128 | 显存够可上 256 |
| 探索噪声 sigma | 0.2 | 每 50 轮衰减 0.99 |
| 隐藏层宽度 | 256 | 两层足够 |
这张表是我在 6 维状态、2 维动作场景下反复试出来的起点。换场景不要照抄,先按这个跑通,再根据奖励曲线调。
5.3 验证策略是否真的学到了东西
训练完不能只看奖励曲线,要做三组对照。第一组,固定任务到达率,对比 DDPG 策略和「全本地」「全卸载」「随机卸载」三个基线的平均时延和能耗。第二组,改变无人机初始位置,看策略能否适应不同距离。第三组,把训练好的 Actor 网络冻结,在测试环境跑 100 个 episode,统计队列溢出次数。
def evaluate(env, actor, episodes=100): total_delay, total_energy, overflow = 0, 0, 0 for _ in range(episodes): state = env.reset() done = False while not done: with torch.no_grad(): action = actor(torch.FloatTensor(state)).numpy() state, reward, done, info = env.step(action) total_delay += info['delay'] total_energy += info['energy'] if info['queue'] > 0.95: overflow += 1 return total_delay / episodes, total_energy / episodes, overflow如果 DDPG 策略的平均时延比全卸载基线还差,说明训练没到位,或者奖励函数权重有问题。如果队列溢出次数超过总步数的 5%,说明队列惩罚还不够重。
5.4 一个容易被忽略的技巧:动作输出后处理
Actor 输出的动作经过 tanh 后是 [-1,1],映射到卸载比例 [0,1] 和方向角 [-π,π]。但实际部署时,无人机转向有物理限制,不能瞬间掉头。在动作后处理里加一个一阶低通滤波,让方向角平滑变化,策略在仿真里的表现会更接近真实飞行。
class ActionFilter: def __init__(self, alpha=0.3): self.alpha = alpha self.prev_heading = 0.0 def filter(self, offload_ratio, heading): heading = self.alpha * heading + (1 - self.alpha) * self.prev_heading self.prev_heading = heading return offload_ratio, headingalpha越小,转向越平滑,但响应越慢。0.3 是我在仿真里试出来的折中值,实际机型要根据最大角速度重新标定。
这套方案我从建模到跑通大概花了两周,其中一半时间耗在奖励函数调参上。DDPG 本身不复杂,难的是把物理约束准确地翻译成状态、动作和奖励。如果你也在做无人机辅助 MEC 的卸载优化,建议先把第 2 章的模型在纸上推一遍,确认每个量的量纲和取值范围,再动手写代码。训练不收敛的时候,先查状态归一化和奖励权重,十有八九问题出在这两处。希望帮到你。
本文还有配套的精品资源,点击获取