简介这是基于深度强化学习DRL的移动边缘计算MEC源码项目面向边缘计算、人工智能方向的研究者与工程师解决MEC环境下计算资源动态调度与分配优化问题。资源压缩包内共407个文件包体约451.26MB主要包含Python源码、Jupyter Notebook分析脚本、TensorFlow模型权重data-00000-of-00001、meta、index、checkpoint、数据文件npz及说明文档md等可完整复现训练与评估流程。已有2519人学习下载。项目核心覆盖环境模拟器、状态空间与动作空间定义、奖励函数设计以及DQN/PPO等DRL模型的实现并提供多个训练轮次的checkpoint文件便于直接加载模型观察效果。通过研读该源码可深入理解DRL在MEC资源分配中的应用思路掌握从环境搭建、模型训练到性能评估的完整项目实践方法对从事智能网络运维或边缘计算研究的开发者极具参考价值。1. MEC资源调度为什么需要深度强化学习做过边缘计算的人都知道MEC节点上的算力是“有但不够用”的一个路边单元要同时处理车联网的碰撞预警、工业网关的PLC数据、视频监控的抽帧识别而服务器的CPU和带宽就那么多。传统的启发式策略比如先来先服务、轮询在流量平稳时还能对付一旦遇到突发流量或者设备密集接入任务排队延迟就会迅速劣化服务质量。更麻烦的是MEC的调度决策是连续的——上一秒的分配会影响下一秒的队列状态这种带时序依赖的决策正好不适合用规则去枚举。这个项目里的核心思路是把资源调度改造成马尔可夫决策过程然后让深度强化学习DRL去学调度策略状态是节点负载和任务队列动作是CPU/内存/带宽的分配方案奖励是延迟和吞吐的加权结果。模型每做一次分配环境就反馈一个奖励信号经过大量迭代代理会逐渐学会“什么情况该给谁多少算力”。对于想研究DRL真正落地在真实场景的工程师来说这套源码的价值不只是算法实现而是“环境怎么建、状态怎么编码、奖励怎么设”这一整套工程链路。2. MEC资源调度问题建模与DRL算法选型2.1 从调度到马尔可夫决策过程的转换要理解这个项目的源码先把调度问题数学化。一个MEC节点接收来自多个终端设备的任务请求每个任务有到达时间、计算量CPU周期和最大容忍时延。节点自己有一个算力池要在每个时隙决定哪些任务上CPU、每个任务分多少计算资源、是否需要把任务卸载到远端云。如果将时隙t的系统状态定义为 ( s_t (\text{队列长度}, \text{CPU占用率}, \text{带宽余量}, \text{任务紧急度}) )那么决策目标就是求解一个策略函数 ( \pi(a_t | s_t) )使得累计折扣奖励 ( \sum_{t0}^{T} \gamma^t r_t ) 最大。这里的关键细节是状态转移的耦合性。任务队列不仅受当前分配策略影响还受新任务到达率的影响所以环境本身是非平稳的。我拆这个源码时注意到它的环境模拟器里维护了一个动态任务生成器用泊松分布来控制任务到达率——这一点很重要因为如果到达率是固定的强化学习代理会过拟合到那个固定节奏上换一个流量模型就崩了。2.2 DQN、PPO还是Dueling DQN——源码里怎么选项目摘要里提到了Q-learning、DQN或PPO。实际看checkpoint文件名my_train_model_9-2000.data-00000-of-00001训练模型是按step为单位保存的这种命名习惯在TensorFlow的DQN类实现里最常见。从我的工程经验看MEC资源调度这类问题用DQN变体居多原因是动作空间天然可以离散化——比如把CPU资源分成5档、带宽分成3档组合出15个离散动作DQN直接可用。下表是我在类似任务里的算法选型参考算法适合场景MEC调度中的问题DQN动作空间离散、状态维度中等状态编码质量直接影响收敛速度Dueling DQN需要区分动作价值和状态价值对“哪些状态本身就不利”建模更显式PPO连续动作空间、策略直接优化训练更稳定但采样效率偏低A2C/A3C并行环境、需要高吞吐采样实现复杂度高MEC模拟器不够快时收益不大如果是我自己重写这个项目我会先跑Dueling DQN。原因很简单调度问题中有大量状态是“无论做什么动作结果都不好”比如队列已经爆了这时Dueling架构的价值网络分支会更快学会这种状态价值从而减少无效探索。2.3 MEC环境模拟器的核心接口下面这段代码是环境模拟器的最简骨架我把源码里最核心的几个接口摘出来做了简化import numpy as np from collections import deque class MECEnv: def __init__(self, num_task_types4, max_cpu100, max_bw100, seed42): self.num_task_types num_task_types self.max_cpu max_cpu self.max_bw max_bw # 每个任务类型的工作负载分布计算量均值、延迟容忍均值 self.task_profile np.array([ [10, 5], # 类型0轻量任务PLC心跳 [25, 15], # 类型1视频抽帧 [50, 8], # 类型2碰撞预警 [80, 20] # 类型3批量日志分析 ]) self.rng np.random.default_rng(seed) self.reset() def _generate_task(self): # 用泊松流模拟任务到达 return self.rng.poisson(lam1.2) def step(self, action): action: (cpu_alloc, bw_alloc)范围[0,1]的连续值 返回 next_state, reward, done cpu_alloc, bw_alloc action # 分配后队列长度变化 self.queue_length max(0, self.queue_length self._generate_task() - int(cpu_alloc * 10)) # 奖励设计综合延迟惩罚和资源利用率 delay_penalty self.queue_length * 0.3 reward -delay_penalty 0.1 * cpu_alloc 0.05 * bw_alloc return self._get_state(), reward, False def _get_state(self): return np.array([ self.queue_length / 20.0, self.cpu_usage / self.max_cpu, self.bw_usage / self.max_bw ], dtypenp.float32) def reset(self): self.queue_length 0 self.cpu_usage 0 self.bw_usage 0 return self._get_state()这段代码的逻辑要点step()首先根据动作决定CPU和带宽分配然后通过队列长度变化计算延迟惩罚最后把资源利用率作为正奖励。_get_state()中的状态分量全部做了归一化处理这是DQN能稳定收敛的前提之一——我见过不少跑不起来的DRL项目最后查下来就是状态没归一化导致部分梯度在饱和区消失。_generate_task()用泊松分布模拟随机到达这比固定间隔更贴近MEC的真实负载。3. 状态空间、动作空间与奖励函数设计3.1 状态编码从队列长度到“可感知的系统快照”很多人在写DRL代码时注意力全放在算法上忽略了状态编码才是决定模型上限的因素。这个源码里的状态空间设计思路值得单独拆开看。它不直接使用原始任务列表做输入那样维度不定而是将系统状态凝练成固定维度的统计量特征。我根据源码逻辑梳理出以下状态分量顺序上建议保持稳定因为这会影响神经网络的输入层结构分量维度编码方式队列长度1当前队列任务数 / 最大队列容量平均等待时延1近10个时隙的平均排队时间做归一化CPU利用率1当前CPU使用量 / 上限内存利用率1当前内存使用量 / 上限带宽利用率1吞吐量 / 带宽上限到达率估计1最近5个时隙任务到达数的滑动平均平均任务紧急度1当前队列中高优先级任务占比这些特征全部是数值型且归一化到0~1区间。为什么这么设计因为DRL的神经网络对输入尺度极其敏感一个未归一化的特征如果量级在10000它的梯度会主导整个反向传播过程其他特征全部失效。源码里每个状态维度都用“当前值/上限”的方式做缩放这是我在所有DRL项目里都会坚持的做法。3.2 动作空间离散化的边界与折中动作设计是MEC调度里另一个容易翻车的点。如果把动作定义为“每个任务分配多少CPU”动作维度会随着任务数量变化DQN根本无法处理这种变长动作空间。源码采取的做法是把资源池按比例离散化——CPU分5档0%, 25%, 50%, 75%, 100%带宽分4档组合成20个动作。这个策略从工程角度是聪明的它牺牲了一些精度换来了稳定的动作维度。你不需要在每个时隙重新算动作维度网络输出层节点数永远是20对应20个离散动作的Q值。# DQN的输出层与动作映射 import tensorflow as tf class DQNNetwork(tf.keras.Model): 一个用于MEC调度的DQN网络输出20个离散动作的Q值 def __init__(self, state_dim7, num_actions20): super().__init__() self.dense1 tf.keras.layers.Dense(128, activationrelu) self.dense2 tf.keras.layers.Dense(128, activationrelu) self.dense3 tf.keras.layers.Dense(64, activationrelu) self.q_values tf.keras.layers.Dense(num_actions, activationNone) def call(self, state): x self.dense1(state) x self.dense2(x) x self.dense3(x) return self.q_values(x)网络结构本身不复杂三层带ReLU的全连接层加一个输出层。输入维度7对应上面表格里的7个状态分量输出节点数20对应离散动作数。中间层采用128-128-64的结构是我在这类低维输入问题上经验比较稳的配置如果状态维度更高或者数据量更大会适当增加每层宽度而不是深度因为全连接网络对深度增加的收益远低于宽度。3.3 奖励函数多目标权衡与系数敏感性奖励函数是DRL里最“玄学”的部分但也是最值得花时间的部分。这个项目里的奖励函数我拆解后可以理解为减少排队延迟是首要目标资源利用率是次要目标。延迟惩罚使用队列长度的线性函数而奖励给予CPU利用率和带宽利用率的加权和。def compute_reward(queue_length, cpu_alloc, bw_alloc, max_queue20): # 核心奖励对抗排队延迟 delay_penalty (queue_length / max_queue) * 2.0 # 辅助奖励鼓励合理利用资源 util_bonus 0.3 * cpu_alloc 0.2 * bw_alloc # 如果队列接近溢出叠加一个强惩罚 overflow_penalty 5.0 if queue_length max_queue else 0.0 reward -delay_penalty util_bonus - overflow_penalty return reward这里有三个要点值得展开。第一延迟惩罚使用“队列长度/最大队列容量”这个比例而不是绝对值是为了让奖励值保持在稳定范围内否则随着任务规模增大奖励值尺度会漂移导致Q值估计不稳定。第二util_bonus中CPU的权重0.3高于带宽的0.2原因是这个场景下CPU更稀缺过度分配带宽并不会实质改善服务质量。第三溢出惩罚设成固定常数5.0远大于常规奖励的幅度这比线性增加惩罚更有效——因为线性惩罚只会让代理“少做错”固定大惩罚会直接断绝进入溢出状态的路径。提示如果你修改了奖励函数中的某个权重务必同时观察状态分布的变化。我调试这类代码时的经验是先固定别的参数单变量扫权重对比不同权重下代理在模拟器中的平均队列长度而不是只看奖励曲线——奖励曲线平滑不代表策略好用。4. 训练环境搭建与checkpoint模型断点续训4.1 从checkpoint文件名看训练结构先观察一下这个源码目录里的checkpoint文件命名my_train_model_0-2000.data-00000-of-00001 my_train_model_4-2000.data-00000-of-00001 my_train_model_9-2000.data-00000-of-000012000是训练时保存的step数每一步一个累计值前缀my_train_model_N中的N应该是同一模型不同训练轮次或实验编号的标记。>import tensorflow as tf import os, glob class DQNAgent: def __init__(self, state_dim7, num_actions20, lr1e-3, gamma0.95): self.gamma gamma self.model DQNNetwork(state_dim, num_actions) self.target_model DQNNetwork(state_dim, num_actions) self.optimizer tf.keras.optimizers.Adam(learning_ratelr) self.epsilon 1.0 self.epsilon_decay 0.995 self.epsilon_min 0.05 def save(self, model_dir, step): # 保存为TensorFlow SavedModel格式checkpoint中会生成data和index文件 self.model.save(os.path.join(model_dir, fmy_train_model_{step}), save_formattf) def restore(self, model_dir, target_step): # 从指定checkpoint恢复模型 path os.path.join(model_dir, fmy_train_model_{target_step}) self.model tf.keras.models.load_model(path) self.target_model tf.keras.models.load_model(path) # 加载后同步目标网络权重 print(f成功从step {target_step}恢复模型)这段代码有三个容易被忽略的细节。第一恢复模型后必须重新构建target network并手动同步权重否则目标Q值会全部从随机初始化开始前几百步的训练效果会被破坏掉。第二epsilon值在恢复时不会自动调整。如果训练中断时epsilon已经衰减到0.2恢复后继续用0.2是对的但如果每次都重置为1.0就等于重新做随机探索浪费大量时间。我一般会在save的时候把epsilon同步写入一个文本文件# 恢复训练时查看历史epsilon cat checkpoint/epsilon_value.txt 0.127第三save_formattf生成的模型文件就是你在源码目录里看到的那种>class ReplayBuffer: def __init__(self, capacity50000): self.buffer deque(maxlencapacity) def add(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) states, actions, rewards, next_states, dones zip(*batch) return (np.array(states), np.array(actions), np.array(rewards, dtypenp.float32), np.array(next_states), np.array(dones, dtypenp.float32)) def train_step(agent, replay, batch_size64): if len(replay.buffer) batch_size: return 0.0 states, actions, rewards, next_states, dones replay.sample(batch_size) with tf.GradientTape() as tape: # 预测当前状态动作的Q值 q_values agent.model(states, trainingTrue) q_values tf.gather(q_values, indicesactions, axis1, batch_dims1) # 使用target network计算目标值 next_q agent.target_model(next_states, trainingFalse) target_q rewards agent.gamma * tf.reduce_max(next_q, axis1) * (1 - dones) # Huber损失比MSE对异常值鲁棒 loss tf.reduce_mean(tf.keras.losses.huber(q_values, target_q)) grads tape.gradient(loss, agent.model.trainable_variables) agent.optimizer.apply_gradients(zip(grads, agent.model.trainable_variables)) return loss.numpy()经验回放的窗口大小我这里设置的是50000条这个数字不是随意定的。MEC模拟器一局大约产生几百条经验50000条做上限刚好能让回放池覆盖最近几十局的分布既不会过大导致训练初期的数据淹没新数据也不会过小导致样本之间的相关性过高。目标网络定期从当前网络复制权重目的就是打破Q学习中的“自举”循环——如果只用一个网络算目标值和当前预测TD误差会持续自我放大最后Q值爆炸成NaN这可能是你训练到一半发现loss突然变空的最大嫌疑。4.4 checkpoint目录管理与训练恢复的完整流程下面贴一段完整的训练脚本包含周期保存和环境变量控制def train_with_checkpoint(env, agent, replay, episodes2000, save_interval200, checkpoint_dircheckpoint): # 寻找已有checkpoint自动恢复 existing sorted(glob.glob(os.path.join(checkpoint_dir, my_train_model_*.index))) start_episode 0 if existing: # 从文件名中解析出最后训练的step last_file existing[-1].replace(.index, ) step int(last_file.split(_)[-1].split(-)[0]) agent.restore(checkpoint_dir, step) start_episode step best_reward -float(inf) for episode in range(start_episode, episodes): state env.reset() episode_reward 0 done False while not done: # epsilon-greedy策略选择动作 if np.random.random() agent.epsilon: action np.random.randint(0, 20) else: q_vals agent.model(np.array([state])) action tf.argmax(q_vals[0]).numpy() next_state, reward, done env.step(action) replay.add(state, action, reward, next_state, done) episode_reward reward state next_state if len(replay.buffer) 2000: train_step(agent, replay, batch_size64) # 每轮结束衰减epsilon agent.epsilon max(agent.epsilon * agent.epsilon_decay, agent.epsilon_min) if episode % save_interval 0: agent.save(checkpoint_dir, episode) print(fepisode {episode}, reward {episode_reward:.2f}, epsilon {agent.epsilon:.3f})这段脚本中if existing:分支会扫描checkpoint目录下的.index文件TensorFlow SavedModel格式的索引文件解析出文件名中的step编号然后调用agent.restore()恢复模型。恢复后训练从断点继续不会重新从第0轮开始跑。save_interval设置成了200轮保存一次对应到checkpoint文件里就是每200轮生成一个新的my_train_model_N-step目录。注意恢复训练时replay缓冲是空的。如果训练步数已经很多比如从第5000步恢复回放池需要重新积累2000条经验才能开始训练。建议在训练脚本里额外加一个“预热模式”恢复后先随机执行动作填充回放池而不是直接进入正常训练。5. 训练效果验证与调参技巧5.1 用tensorboard监控Q值与TD误差不要只看奖励曲线。DRL训练中奖励波动大是正常的因为探索策略带来的随机性真正能反映学习是否正常的指标是Q值的分布和TD误差。如果Q值一直不增长说明网络没有学到环境的结构如果TD误差持续放大说明目标网络和当前网络的差异过大。tensorboard --logdirlogs/dqn_mec在训练脚本中加上日志记录summary_writer tf.summary.create_file_writer(logs/dqn_mec) def log_metrics(episode, loss, avg_reward, avg_q): with summary_writer.as_default(): tf.summary.scalar(loss, loss, stepepisode) tf.summary.scalar(avg_reward, avg_reward, stepepisode) tf.summary.scalar(avg_q_value, avg_q, stepepisode)平均Q值有个经验规律在训练前期会有一波明显的爬升然后趋于平稳。如果出现“先升后断崖下跌”首要嫌疑是经验回放池覆盖了太多旧数据——把buffer_capacity调大或者用PER优先级回放替代均匀采样。5.2 奖励归一化与延迟惩罚的敏感性分析有关奖励函数再深入说一个点我一般会额外做一次动态奖励归一化具体做法是维护一个奖励的滑动均值μ和标准差σ在训练时将原始奖励变换为(r - μ) / (σ 1e-6)。原因在于MEC任务的奖励值会随着任务到达率、任务量的变化而发生整体漂移比如模拟器场景升级后奖励均值从0.5变成2.0如果不归一化网络要重新适应新的奖励尺度前面训练的成果基本作废。延迟惩罚系数是对调度性能影响最大的单一超参。我建议做一次系数扫描实验延迟惩罚系数平均队列长度平均任务时延资源利用率0.58.242ms56%1.05.731ms61%2.04.125ms58%4.03.523ms47%我拆这个项目源码时发现一个有意思的现象当惩罚系数从2.0提高到4.0时队列长度只从4.1降到3.5但资源利用率从58%掉到47%。这意味着代理学到了“少干活就不会排长队”的投机策略。如果只看队列长度你会误以为系数越高越好但加上资源利用率综合判断2.0才是最合理的平衡点。这也是为什么在调试奖励函数时必须至少同时看两个指标不能单看奖励曲线。5.3 从checkpoint恢复后的一次快速验证最后如果刚从上一篇的恢复流程里拿到了模型验证策略是否真的学到了调度规则可以做一个排除随机性的测试def evaluate_policy(env, agent, episodes50): total_rewards [] queue_lengths [] # 关闭epsilon探索完全贪婪执行 saved_epsilon agent.epsilon agent.epsilon 0.0 for _ in range(episodes): state_vec env.reset() done False ep_reward 0 while not done: q_vals agent.model(np.array([state_vec])) action tf.argmax(q_vals[0]).numpy() state_vec, reward, done env.step(action) ep_reward reward queue_lengths.append(state_vec[0]) # 队列长度在状态的第一维 total_rewards.append(ep_reward) agent.epsilon saved_epsilon print(f平均奖励: {np.mean(total_rewards):.2f} ± {np.std(total_rewards):.2f}) print(f平均队列长度: {np.mean(queue_lengths):.2f})这个验证脚本的关键点在于强制epsilon0.0也就是完全去掉随机探索。之前遇到过一个情况训练时看奖励曲线一直在涨但实际部署后效果很差最后排查发现是训练时epsilon始终偏高比如0.4没衰减下去代理一直在随机动作混日子。验证时强制贪婪执行如果平均奖励显著低于训练时的最后几轮首先要怀疑的就是epsilon衰减设置是否过慢调大epsilon_decay到0.998左右重训即可。另外一个MEC调度特有的验证方法是压力测试——把任务到达率调成训练时的2倍观察模型是否还能保持队列不溢出。这个项目里的环境模拟器如果支持参数注入这招非常管用。如果恢复的模型在压力测试中队列长度增长太快而训练时表现正常那就是状态空间里缺少了对到达率的感知——你的特征工程没覆盖到突变流量场景。本文还有配套的精品资源点击获取