深度强化学习求解机组组合：MDP建模与开源实现

简介这份开源项目面向电力系统调度、能源管理与强化学习方向的研究生、算法工程师及科研人员聚焦机组组合这一经典优化难题。机组组合需在满足负荷需求与设备约束的前提下安排各类型机组的启停与出力计划以最小化运行成本而深度强化学习凭借处理高维状态空间与复杂序贯决策的能力为该问题提供了新解法。资源包共4个文件包含Python实现脚本、Markdown说明文档、文本说明及Word文档压缩后约37KB其中脚本对应DQN求解机组组合的核心逻辑文档则交代算法思路与使用方式。项目围绕状态空间、动作空间、奖励函数与策略设计展开涉及马尔可夫决策过程、神经网络训练、经验回放与目标网络等关键环节并与动态规划思路形成对照。目前已有100人学习适合希望快速理解深度强化学习如何落地电力调度、并在此基础上复现与改进的读者参考。1. 机组组合问题的深度强化学习建模从马尔可夫决策过程到可复现的开源方案火电机组组合Unit Commitment, UC是电力调度里最让人头疼的优化问题之一。一台机组从冷态启动到并网带负荷中间要烧油、要暖机、要满足最小启停时间而调度员第二天早上八点前必须把 24 小时的启停计划交出去。传统做法是混合整数规划MIP用分支定界硬啃规模一上来求解时间就爆炸。深度强化学习Deep Reinforcement Learning, DRL这几年被反复拿来试这条路核心思路是把 UC 的时序决策过程建模成马尔可夫决策过程MDP让智能体在“开机/停机/保持”的动作空间里学出一套策略网络推理时毫秒级出结果。这个开源项目方向适合两类人一类是电力系统优化背景、想找 MIP 之外替代方案的调度算法工程师另一类是强化学习背景、想找一个有真实物理约束的工业级 benchmark 练手的研究者。下面按“建模—环境—训练—调参—避坑—进阶”的顺序把这条路径拆到能照着复现的程度。2. 把 UC 写成 MDP状态空间、动作空间与奖励函数怎么定2.1 状态空间设计哪些量必须进哪些量进去就是噪声UC 的 MDP 状态设计直接决定策略网络能不能学到可行解。常见做法是把状态分成三类系统级负荷与备用需求、机组级物理状态、时间耦合量。系统级至少要有未来若干时段的负荷预测值归一化后、旋转备用需求、当前时段索引机组级要有每台机组的出力上下限、最小启停时间计数、当前在线状态、上一时段出力、爬坡速率。时间耦合量容易被忽略——比如某台机组已经连续运行了 5 小时而最小运行时间是 6 小时这个“还差 1 小时”的信息如果不进状态策略网络就会在边界上反复翻车。我一般会把状态向量组织成固定长度的扁平数组机组数量 N 固定时维度是N * 单机特征数 系统特征数。如果机组数量可变就得用集合型网络如 Deep Sets或者把机组排序后 padding。新手最容易犯的错是把原始负荷值直接喂进去量纲从几百到几万网络第一层就饱和了。归一化用训练集的最大最小值推理时沿用同一套 scaler不要重新拟合。import numpy as np class UCStateBuilder: def __init__(self, num_units, load_max, reserve_ratio0.05): self.num_units num_units self.load_max load_max self.reserve_ratio reserve_ratio # 单机特征出力、在线状态、已运行/已停机时长、最小启停剩余 self.unit_feat_dim 5 self.sys_feat_dim 3 # 负荷、备用、时段索引 def build(self, load_forecast, unit_states, t): # load_forecast: 未来 T 步负荷这里取当前步和下一步 load_now load_forecast[t] / self.load_max load_next load_forecast[min(t1, len(load_forecast)-1)] / self.load_max reserve load_now * self.reserve_ratio sys_feat np.array([load_now, load_next, reserve], dtypenp.float32) unit_feats [] for u in unit_states: # 出力归一化到 [0,1] p_norm u[p] / u[p_max] if u[p_max] 0 else 0.0 on float(u[on]) # 已运行或已停机时长截断到最小启停时间上限 run_hours min(u[run_hours], u[min_up] u[min_down]) # 距离满足最小启停还差多少 remain max(0, u[min_up] - u[run_hours]) if u[on] else max(0, u[min_down] - u[off_hours]) unit_feats.extend([p_norm, on, run_hours / 24.0, remain / 24.0, u[ramp_rate]]) return np.concatenate([sys_feat, np.array(unit_feats, dtypenp.float32)])这段代码里unit_feat_dim5对应出力、在线状态、运行时长、最小启停剩余、爬坡率。remain这个特征是我踩过坑之后加上的没有它的时候策略网络在最小启停边界上会输出振荡动作导致大量不可行解。load_max用训练集历史最大负荷不要用单日最大否则跨季节泛化会崩。2.2 动作空间与奖励函数稀疏奖励是训练不收敛的头号原因动作空间最直接的定义是每台机组一个离散动作0 停机、1 开机、2 保持。N 台机组就是3^N的组合空间N10 时已经 59049直接做多离散动作的 Q-learning 维度爆炸。常见做法有两种一是用自回归策略网络逐台机组输出动作把联合动作概率分解成条件概率乘积二是用连续动作空间加阈值离散化输出 N 维连续值再映射到启停。我倾向第一种因为电力系统对可行性要求高自回归结构可以在每台机组决策时把前面机组的决策作为条件输入减少不可行组合。奖励函数是 DRL 做 UC 最玄学的地方。纯经济性奖励燃料成本启停成本在早期几乎全是负的智能体随机探索时频繁触发不可行解奖励信号被惩罚项淹没。我的做法是分三段基础经济性奖励、可行性惩罚、平滑项。可行性惩罚不要一上来就给大常数先用软惩罚比如违反最小启停时间每小时罚一个中等值等策略能稳定输出可行解后再加大惩罚力度。平滑项是给相邻时段出力变化加一个小惩罚抑制策略网络输出锯齿状功率曲线。def compute_reward(dispatch, unit_params, penalty_scale1.0): fuel_cost 0.0 startup_cost 0.0 for u, p in zip(unit_params, dispatch[p]): if dispatch[on][u[id]]: # 二次燃料成本曲线 fuel_cost u[a] * p**2 u[b] * p u[c] if dispatch[startup][u[id]]: startup_cost u[startup_cost] # 可行性惩罚 penalty 0.0 for u in unit_params: if dispatch[on][u[id]]: if dispatch[run_hours][u[id]] u[min_up] and dispatch[just_started][u[id]]: penalty penalty_scale * (u[min_up] - dispatch[run_hours][u[id]]) else: if dispatch[off_hours][u[id]] u[min_down] and dispatch[just_stopped][u[id]]: penalty penalty_scale * (u[min_down] - dispatch[off_hours][u[id]]) # 功率平衡惩罚 balance_gap abs(sum(dispatch[p]) - dispatch[load]) penalty penalty_scale * balance_gap * 10 return -(fuel_cost startup_cost penalty)penalty_scale是训练中最需要调的参数。太小则策略无视约束太大则早期奖励全负、梯度消失。我一般从 0.1 开始每 500 个 episode 检查可行解比例低于 60% 就乘 1.5高于 95% 就乘 0.9。balance_gap的系数 10 是经验值因为功率不平衡在物理上直接对应频率偏差必须重罚。3. 训练环境搭建从数据到 Gym 接口的完整链路3.1 数据准备负荷曲线、机组参数与场景生成开源项目里通常会给一份标准测试系统数据比如 10 机系统或 IEEE RTS-96。你需要确认三样东西负荷时序至少 8760 小时、每台机组的成本系数和物理约束、启停成本。如果数据里没有爬坡率按额定容量的 20%/小时补一个保守值。场景生成用历史负荷加高斯噪声噪声标准差取负荷的 3%5%太小则策略过拟合太大则训练不稳定。数据格式建议统一成 CSV YAMLCSV 存时序负荷YAML 存机组静态参数。这样换系统时只改 YAML代码不用动。下面是一个最小数据加载器。import pandas as pd import yaml def load_uc_data(load_csv, unit_yaml): load_df pd.read_csv(load_csv, parse_dates[timestamp]) load_series load_df[load_mw].values.astype(np.float32) with open(unit_yaml, r, encodingutf-8) as f: unit_cfg yaml.safe_load(f) units [] for u in unit_cfg[units]: units.append({ id: u[id], p_min: u[p_min], p_max: u[p_max], a: u[cost_a], b: u[cost_b], c: u[cost_c], min_up: u[min_up_hours], min_down: u[min_down_hours], ramp_rate: u.get(ramp_rate, u[p_max] * 0.2), startup_cost: u[startup_cost], }) return load_series, unitsramp_rate缺省值用p_max * 0.2是保守估计实际燃煤机组爬坡率通常在 1%3%/分钟折算到小时约 60%180%但为了训练稳定先用 20%。min_up_hours和min_down_hours必须显式给出不能默认 1否则策略会学到频繁启停的退化解。3.2 Gym 环境封装reset、step 与 done 的边界条件把 UC 封装成 Gym 接口时reset要返回初始状态step要返回(next_state, reward, done, info)。done的判定有两个条件一是到达调度周期末尾比如 24 小时二是出现严重不可行比如功率缺额超过总负荷的 10%。第二个条件要谨慎太早触发 done 会让智能体学不到恢复策略我一般只在连续 3 个时段功率缺额都超标时才终止。import gym from gym import spaces class UCEnv(gym.Env): def __init__(self, load_series, units, horizon24): super().__init__() self.load_series load_series self.units units self.horizon horizon self.num_units len(units) self.action_space spaces.MultiDiscrete([3] * self.num_units) # 状态维度与 StateBuilder 对齐 self.observation_space spaces.Box(low-10, high10, shape(3 5*self.num_units,), dtypenp.float32) def reset(self): self.t 0 self.unit_states [{p: 0, on: 0, run_hours: 0, off_hours: 0, p_max: u[p_max], min_up: u[min_up], min_down: u[min_down], ramp_rate: u[ramp_rate]} for u in self.units] return self._get_state() def step(self, action): # action: 每台机组 0/1/2 dispatch self._apply_action(action) reward compute_reward(dispatch, self.units) self.t 1 done self.t self.horizon or dispatch[balance_gap] 0.1 * dispatch[load] info {balance_gap: dispatch[balance_gap], feasible: dispatch[balance_gap] 1e-3} return self._get_state(), reward, done, infoaction_space用MultiDiscrete而不是Discrete(3**N)因为后者在 N8 时整数溢出风险高且策略网络输出层不好设计。observation_space的low/high设成 -10 到 10 是给归一化后的特征留余量实际值都在 01 附近。_apply_action里要做经济调度ED二次分配确定哪些机组在线后用等微增率法或简单按容量比例分配负荷不要用随机分配否则奖励噪声太大。4. 策略网络与训练循环PPO 在 UC 上的参数怎么设4.1 网络结构共享底层 双头输出UC 的观测是扁平向量用 MLP 就够不需要 Transformer。我一般用 3 层 MLP每层 256 个神经元激活函数用 ReLU。输出分两个头策略头输出N * 3个 logits价值头输出一个标量。策略头用自回归方式逐机组采样价值头直接回归状态价值。共享底层可以让价值估计帮策略学习但要注意梯度冲突——如果价值损失波动大就把共享层的学习率调低一半。import torch import torch.nn as nn class UCPolicyNet(nn.Module): def __init__(self, state_dim, num_units, hidden256): super().__init__() self.num_units num_units self.shared nn.Sequential( nn.Linear(state_dim, hidden), nn.ReLU(), nn.Linear(hidden, hidden), nn.ReLU(), ) self.policy_head nn.Linear(hidden, num_units * 3) self.value_head nn.Linear(hidden, 1) def forward(self, state): feat self.shared(state) logits self.policy_head(feat).view(-1, self.num_units, 3) value self.value_head(feat) return logits, value def act(self, state, deterministicFalse): logits, value self.forward(state) actions [] for i in range(self.num_units): dist torch.distributions.Categorical(logitslogits[:, i, :]) a dist.probs.argmax(dim-1) if deterministic else dist.sample() actions.append(a) return torch.stack(actions, dim-1), valuelogits的 shape 是(batch, num_units, 3)逐机组采样时每台机组独立一个 Categorical 分布。deterministicTrue用于推理阶段直接取 argmax。注意act里没有做动作掩码实际训练时要加如果某台机组处于最小停机时间内开机动作的 logit 要设成负无穷强制不可选。这个掩码是 UC 训练能收敛的关键之一。4.2 PPO 关键参数clip 范围、GAE 与学习率调度PPO 在 UC 上的超参和常规 Atari 任务差别很大。clip_range我一般设 0.1 而不是 0.2因为 UC 的奖励尺度大clip 太宽会导致策略更新过猛。gamma设 0.99lambdaGAE设 0.95。学习率用 3e-4 起步每 2000 步线性衰减到 1e-5。entropy_coef从 0.01 开始如果策略过早收敛到“全部开机”或“全部停机”的退化解就加到 0.05。def train_ppo(env, policy, epochs5000, steps_per_epoch2048, clip0.1): optimizer torch.optim.Adam(policy.parameters(), lr3e-4) scheduler torch.optim.lr_scheduler.LinearLR(optimizer, start_factor1.0, end_factor0.01, total_itersepochs) for epoch in range(epochs): states, actions, rewards, dones, log_probs, values [], [], [], [], [], [] state env.reset() for _ in range(steps_per_epoch): s_tensor torch.FloatTensor(state).unsqueeze(0) with torch.no_grad(): a, v policy.act(s_tensor) logits, _ policy(s_tensor) log_prob sum(torch.log_softmax(logits[:, i, :], dim-1)[0, a[0, i]] for i in range(env.num_units)) next_state, reward, done, info env.step(a[0].numpy()) states.append(state); actions.append(a[0].numpy()) rewards.append(reward); dones.append(done) log_probs.append(log_prob.item()); values.append(v.item()) state next_state if not done else env.reset() # GAE 计算 advantages [] gae 0 for t in reversed(range(len(rewards))): if t len(rewards) - 1: next_value 0 else: next_value values[t1] delta rewards[t] 0.99 * next_value * (1 - dones[t]) - values[t] gae delta 0.99 * 0.95 * (1 - dones[t]) * gae advantages.insert(0, gae) advantages torch.FloatTensor(advantages) returns advantages torch.FloatTensor(values) # PPO 更新 old_states torch.FloatTensor(np.array(states)) old_actions torch.LongTensor(np.array(actions)) old_log_probs torch.FloatTensor(log_probs) for _ in range(10): logits, new_values policy(old_states) new_log_probs [] for i in range(env.num_units): dist torch.distributions.Categorical(logitslogits[:, i, :]) new_log_probs.append(dist.log_prob(old_actions[:, i])) new_log_probs torch.stack(new_log_probs, dim-1).sum(dim-1) ratio torch.exp(new_log_probs - old_log_probs) surr1 ratio * advantages surr2 torch.clamp(ratio, 1-clip, 1clip) * advantages policy_loss -torch.min(surr1, surr2).mean() value_loss nn.MSELoss()(new_values.squeeze(), returns) loss policy_loss 0.5 * value_loss optimizer.zero_grad(); loss.backward(); optimizer.step() scheduler.step()steps_per_epoch2048是经验值UC 的 episode 长度 24约 85 个完整 episode 做一次更新。clip0.1比默认 0.2 更保守因为电力系统奖励的方差大。advantages计算里dones[t]的处理要注意如果 done 是因为到达 horizonnext_value应该为 0如果是因为不可行提前终止next_value也应该为 0但要在 info 里记录终止原因方便排查。5. 避坑与排查UC 强化学习训练中最容易翻车的 5 个点5.1 现象奖励曲线震荡不收敛可行解比例长期低于 30%原因通常是惩罚项系数太大或太小。太大时早期奖励全负策略梯度方向被惩罚主导太小时策略无视约束功率平衡长期不满足。解决方法是把惩罚项拆成两部分硬约束功率平衡用大系数软约束最小启停用小系数并且每 500 episode 动态调整一次。我一般会打印每个 episode 的balance_gap均值和min_up_violation次数前者应该在前 1000 episode 内降到负荷的 1% 以下后者降到 5 次以内。5.2 现象策略学到“全部开机”或“全部停机”的退化解这是奖励函数设计问题。全部开机时燃料成本高但可行性好如果惩罚项权重远大于经济性奖励策略就会摆烂。解决方法是给经济性奖励加一个基准线用“全部开机”的成本作为 baseline奖励减去 baseline这样策略有动力去优化启停组合。另一个原因是 entropy 系数太低策略过早收敛把entropy_coef从 0.01 提到 0.05 通常能缓解。5.3 现象训练时可行推理时不可行大概率是归一化不一致。训练时用训练集的最大最小值做 scaler推理时如果重新拟合了 scaler状态分布就变了。解决方法是把 scaler 的参数保存到 checkpoint 里推理时直接加载。另一个原因是动作掩码在推理时没加训练时加了掩码所以可行推理时忘了加就会输出违反最小启停的动作。检查act函数里deterministicTrue分支是否也走了掩码逻辑。5.4 现象换一个测试系统后性能暴跌UC 的 DRL 策略对机组数量敏感。训练时 10 台机组测试时 20 台策略网络输出维度对不上。解决方法是训练时随机化机组数量比如每 episode 从 812 台里随机采样用 padding 对齐到最大数量。或者用集合型网络但实现复杂度高。我一般先用随机化机组数量做课程学习从 5 台开始逐步加到 15 台。5.5 现象训练速度慢单卡一天跑不完 1000 episodeUC 环境的step里如果做了完整的 economic dispatch 优化每步耗时可能超过 10ms。优化方法是把 ED 用查表法近似离线生成“在线机组组合 → 最优出力分配”的查找表训练时直接查表。或者用 GPU 并行多个环境实例把 batch 维度利用起来。我一般用 8 个并行环境每个环境独立 resetPPO 的 batch 从 8 个环境里各采 256 步这样 GPU 利用率能从 20% 提到 70%。6. 进阶技巧用课程学习与动作掩码把可行解率推到 99%课程学习在 UC 上效果非常明显。第一阶段只训练 4 台机组、12 小时调度周期策略很快能学到可行解第二阶段加到 8 台机组、24 小时第三阶段随机化负荷波动和机组数量。每个阶段训练 1000 episode阶段切换时把学习率重置到 1e-4让策略适应新难度。动作掩码的实现要放在策略网络输出之后、采样之前对每台机组检查如果当前在线且运行时长小于最小运行时间停机动作的 logit 设为 -1e9如果当前停机且停机时长小于最小停机时间开机动作的 logit 设为 -1e9。def apply_action_mask(logits, unit_states): # logits: (batch, num_units, 3) for i, u in enumerate(unit_states): if u[on] and u[run_hours] u[min_up]: logits[:, i, 0] -1e9 # 禁止停机 if not u[on] and u[off_hours] u[min_down]: logits[:, i, 1] -1e9 # 禁止开机 return logits验证可行解率的方法跑 100 个测试场景统计info[feasible]为 True 的比例。如果低于 95%先检查动作掩码是否生效再检查功率平衡惩罚系数是否够大。我自己的习惯是每次修改奖励函数后先跑 50 个 episode 的短训练看可行解率曲线再决定要不要跑完整训练。这个习惯帮我省了很多次通宵跑废模型的后悔药。希望帮到你。本文还有配套的精品资源点击获取