简介面向车间排产调度场景的深度强化学习 DQN 实现代码包适合具备 Python 与强化学习基础的高校研究者、算法工程师和调度方向学生用于在动态生产环境中学习并优化排产策略缓解传统优化方法在 NP 难问题和大规模约束下适应性不足的困扰。压缩包为 rar 格式共 3 个 py 文件、约 5KB分别承担算法核心、环境建模与流程运行三类职责网络结构与参数更新、经验回放和目标网络等 DQN 关键机制基于 gym 的状态空间、动作空间与奖励函数定义及环境更新训练、评估与可视化的一体化入口。目前已有 848 人浏览学习。读者可据此搭建可复现的排产仿真环境理解奖励设计与状态表示对调度效果的影响并替换车间规模、工序约束与成本目标形成自己的实验对照与调参思路也可作为从算法入门到课题落地的练习。1. 从一张 20×6 的加工单说起DQN 排产到底在解什么问题深度强化学习 DQN 做车间排产调度很多人第一反应是“状态、动作、奖励一搭就能跑”真把 gym 环境写出来才发现麻烦全在建模工件有先后工序机器有占用时间动作选错还会把环境卡死。这个 Python 项目把 DQN 排产拆成三层gym 环境负责推进时间轴智能体负责选工件实验脚本负责对比规则基线。它适合两类人想用深度强化学习练手调度问题的算法同学以及手里有工单表、想先把排产原型跑起来的 Python 开发。读完至少能判断一件事——你的车间适不适合把规则排产换成 DQN以及换之前要补哪些状态特征。2. 项目结构与 gym 环境建模状态、动作、奖励三件套怎么落地调度类强化学习项目最怕一上来就堆网络。工件、工序、机器、交期这些实体没被翻译成固定长度的向量之前Q 网络再深也学不到东西。这个项目的骨架很朴素环境层只依赖 numpy智能体层只依赖 torch训练脚本把两边接起来。先看目录怎么分再看状态和动作怎么定最后看奖励怎么给。2.1 目录结构环境层不碰 torch智能体层不碰工单我一般把调度环境当成一个“时间轴推进器”它只负责三件事回答当前哪些动作合法、执行一个动作后时间走到哪、返回这一刻的观测。智能体不应该知道工单表长什么样训练脚本也不应该手写状态拼接。这个项目的目录大致是这样分的dqn_scheduling/ ├── envs/ │ ├── __init__.py │ ├── jobshop_env.py # gym 风格环境reset/step/action_mask │ └── instance.py # 工单表解析、字段校验、基础统计 ├── agents/ │ ├── dqn.py # Q 网络、选动作、训练一步 │ └── replay_buffer.py # 经验回放 ├── configs/ │ └── jobshop_20x6.yaml # 实例路径、超参、随机种子 ├── data/ │ └── jobshop_20x6.csv # 工件-工序-机器-工时 ├── train.py # 训练入口 ├── evaluate.py # 规则基线对比与指标输出 └── utils/ ├── metrics.py # makespan、流程时间、利用率 └── seed.py # 固定随机种子envs 目录里没有import torch这是刻意为之。环境要能单独跑随机策略也要能在没有 GPU 的机器上做单元测试agents 目录里没有读 CSV 的代码智能体只接收obs、mask和reward。configs 把实例路径和超参抽出去换一张 30×8 的工单只需要新增一个 yaml不用改训练脚本。instance.py常见做法是用一个轻量 dataclass 存三类数组machine[j][k]表示工件 j 的第 k 道工序在哪台机器duration[j][k]表示对应工时n_ops[j]表示每个工件有几道工序。读取 CSV 后立刻校验三件事工序号是否连续、机器编号是否越界、工时是否为正。很多排产环境跑到一半报索引错根子都在工单表里有一行工序号跳号。2.2 状态设计把工单表压成固定长度向量状态设计的第一原则是“网络输入维度固定”但车间里每个工件的剩余工序数不一样。这个项目用“按工件对齐、按机器对齐、再加两个全局量”的方式拼向量维度是n_jobs × 4 n_machines 2。对 20 个工件、6 台机器输入就是 88 维不算大MLP 完全吃得下。def _obs(self): # 每个工件的剩余工序数、剩余总工时、当前工序工时、可开工时间 remaining_ops np.array([ self.instance.n_ops[j] - self.op_idx[j] for j in range(self.n_jobs) ], dtypenp.float32) remaining_time np.array([ sum(self.instance.duration[j][self.op_idx[j]:]) for j in range(self.n_jobs) ], dtypenp.float32) cur_duration np.array([ self.instance.duration[j][self.op_idx[j]] if self.op_idx[j] self.instance.n_ops[j] else 0.0 for j in range(self.n_jobs) ], dtypenp.float32) # 用当前最大剩余工时做归一化避免维度之间量纲差太大 scale max(1.0, float(remaining_time.max())) return np.concatenate([ remaining_ops / max(1, self.instance.max_ops), remaining_time / scale, cur_duration / scale, self.job_ready / scale, self.machine_free / scale, [self.done_jobs / self.n_jobs, self.makespan() / scale], ]).astype(np.float32)这段代码里op_idx[j]是工件 j 当前做到第几道工序job_ready[j]是这个工件最早能开工的时间machine_free[m]是机器 m 下一次空闲的时间点。前四组按工件排列第五组按机器排列最后两个是全局进度和当前 makespan 估计。归一化用remaining_time.max()而不是固定常数换实例时不用重新调缩放系数。参数上最容易被忽略的是max_ops。如果工单里最多 8 道工序就用 8 做分母如果直接除以当前工件的工序数不同工件之间就没法横向比较。另一个细节是已完工工件的cur_duration要补 0不能让索引越界也不能让已完工工件在状态里继续“占位”产生误导。2.3 动作空间与奖励函数让 DQN 学会不拖延动作空间有两种常见建模一种是选“工件”环境自动把它安排到当前可用的机器上另一种是选“机器-工件对”适合柔性车间。这个项目用第一种动作数等于工件数20 个工件就是 20 个离散动作Q 网络输出 20 维简单直接。代价是它假设每道工序的机器是固定的遇到并行机或可替代设备要改动作层。def action_mask(self): # 未完工的工件都可选机器占用通过时间轴推进不需要在这里排除 mask np.zeros(self.n_jobs, dtypenp.bool_) for j in range(self.n_jobs): if self.op_idx[j] self.instance.n_ops[j]: mask[j] True return mask def step(self, action): assert self.action_mask()[action], 非法动作工件已完工或不可选 j int(action) op self.op_idx[j] machine self.instance.machine[j][op] duration self.instance.duration[j][op] # 开工时间取“工件就绪”和“机器空闲”的较大值 start max(self.job_ready[j], self.machine_free[machine]) end start duration idle start - self.machine_free[machine] # 这次决策让机器空等了多久 wait start - self.job_ready[j] # 工件在缓冲区等了多久 self.machine_free[machine] end self.job_ready[j] end self.op_idx[j] 1 if self.op_idx[j] self.instance.n_ops[j]: self.done_jobs 1 # 稠密奖励压空闲和等待终局再压一次总完工时间 reward -(idle 0.2 * wait) * self.reward_scale done self.done_jobs self.n_jobs if done: reward - self.makespan() * self.reward_scale return self._obs(), reward, done, {makespan: self.makespan()}step里没有“机器当前是否被占用”的判断因为machine_free存的是时间点不是布尔状态。这样写的好处是环境可以一次推进到未来不需要按小时模拟。奖励由两部分组成idle惩罚机器空等wait惩罚工件在缓冲区排队系数 0.2 是让机器利用率优先于单个工件的等待。终局再减一次makespan是给智能体一个全局目标避免它只盯着每一步的小便宜。reward_scale建议先用1 / 平均工时做缩放让每步奖励落在 -1 到 0 附近。如果直接用工时原值20×6 的实例一步奖励可能到 -50Q 值很快冲到几百训练曲线会非常难看。这个参数在配置文件里暴露出来换实例时先调它再调学习率。3. DQN 训练链路拆解经验回放、目标网络和动作掩码怎么配合环境写对之后DQN 这一侧的反而不复杂但有三处必须对齐回放池里要存“下一个状态的动作掩码”目标网络计算下一状态 Q 值时要过滤非法动作训练循环里探索率要按 episode 衰减而不是按 step。这三处任何一处漏了训练都会出现“损失下降但排产结果不降”的怪现象。3.1 经验回放先解决样本相关性问题调度环境是串行决策连续几步的状态高度相关。如果每走一步就立刻用这条样本更新网络会在一张工单上反复过拟合。经验回放的作用是把不同 episode、不同时间点的转移打散batch 里既有开局样本也有收尾样本。class ReplayBuffer: def __init__(self, capacity): self.capacity capacity self.data [] self.ptr 0 def push(self, s, a, r, s2, done, mask2): # mask2 必须一起存目标网络算下一状态 Q 值时要用它过滤非法动作 item (s, a, r, s2, done, mask2) if len(self.data) self.capacity: self.data.append(item) else: self.data[self.ptr] item self.ptr (self.ptr 1) % self.capacity def sample(self, batch_size): idx np.random.choice(len(self.data), batch_size, replaceFalse) batch [self.data[i] for i in idx] s, a, r, s2, d, m2 map(np.stack, zip(*batch)) return s, a, r, s2, d.astype(np.float32), m2capacity取 50000 左右对 20×6 的实例足够覆盖几百个 episode。ptr用循环覆盖旧样本自然被淘汰。这里最容易漏的是mask2如果只存s2不存下一状态的掩码目标网络会把已完工工件也算进 maxQ 目标会虚高。done存成 float计算 TD 目标时用1 - done截断避免把下一 episode 的奖励串进来。3.2 网络与动作掩码Q 值输出后先做一次过滤网络结构不需要花哨两层 128 的 MLP 足够。输入是 2.2 节拼出来的 88 维向量输出是n_jobs维 Q 值。真正要小心的是选动作和算目标时的掩码处理非法动作的 Q 值要压到极小而不是置 0因为 0 可能比合法动作的负 Q 值还大。class QNet(nn.Module): def __init__(self, obs_dim, n_actions, hidden128): super().__init__() self.net nn.Sequential( nn.Linear(obs_dim, hidden), nn.ReLU(), nn.Linear(hidden, hidden), nn.ReLU(), nn.Linear(hidden, n_actions), ) def forward(self, x): return self.net(x) def select_action(qnet, obs, mask, epsilon, device): if np.random.rand() epsilon: valid np.flatnonzero(mask) # 只在合法动作里随机探索 return int(np.random.choice(valid)) with torch.no_grad(): q qnet(torch.as_tensor(obs, dtypetorch.float32, devicedevice).unsqueeze(0)) q q.masked_fill(~torch.as_tensor(mask, devicedevice).unsqueeze(0), -1e9) return int(q.argmax(dim1).item())masked_fill的第二个参数用-1e9不用-inf是为了避免后续计算出现 NaN。探索阶段也要用np.flatnonzero(mask)限制在合法动作里否则随机动作会直接触发环境断言。epsilon从 1.0 退到 0.05按 episode 衰减不要按 step 衰减因为每个 episode 的步数随着排产方案变化按 step 衰减会让不同实例的探索进度不一致。3.3 训练循环一次 episode 从随机走到贪心训练循环把环境、回放池、在线网络和目标网络串起来。每个 step 做四件事选动作、推进环境、存样本、抽 batch 更新。目标网络每 200 步同步一次梯度裁剪设 5.0防止个别大奖励把参数打飞。for ep in range(num_episodes): obs env.reset() done False epsilon max(eps_end, eps_start * (eps_decay ** ep)) while not done: mask env.action_mask() action select_action(qnet, obs, mask, epsilon, device) next_obs, reward, done, info env.step(action) next_mask env.action_mask() if not done else np.zeros(env.n_jobs, dtypenp.bool_) buffer.push(obs, action, reward, next_obs, done, next_mask) obs next_obs loss train_step(qnet, target, opt, buffer, batch_size, gamma, device) if total_steps % target_update 0: target.load_state_dict(qnet.state_dict()) total_steps 1 if ep % 50 0: print(fep{ep} eps{epsilon:.3f} makespan{info[makespan]:.1f})train_step里用smooth_l1_loss而不是 MSE调度奖励里有少量大负值MSE 会放大离群样本的影响。gamma取 0.95因为排产 episode 通常几十到几百步折扣太狠会让智能体只看眼前几步。target_update取 200 步太频繁等于没有目标网络太慢又会让目标滞后。这几个值在 20×6 的实例上比较稳换到 50 个工件要适当加大hidden和buffer_size但不要把网络堆到三四层调度状态没有图像那种空间结构深了反而难训。4. 跑通与调参从依赖安装到规则基线对比的四步拿到代码后不要急着改网络先把随机策略跑通再跑规则基线最后才开训练。顺序反了后面出了问题时你分不清是环境写错、奖励给错还是 DQN 本身没学会。下面按“实例文件 → 配置 → 训练 → 评估”四步走。4.1 实例文件与配置先固定输入格式再谈训练工单表用 CSV一行一个工件列名固定为job_id, op_seq, machine, duration。同一个工件的多道工序按op_seq递增排列。读取时按job_id分组组内按op_seq排序再填到machine[j][k]和duration[j][k]两个二维数组里。空值、重复工序号、机器号从 0 开始还是从 1 开始这些都要在instance.py里统一不要让环境层去猜。instance: data/jobshop_20x6.csv n_jobs: 20 n_machines: 6 episodes: 3000 gamma: 0.95 lr: 0.0001 batch_size: 128 buffer_size: 50000 eps_start: 1.0 eps_end: 0.05 eps_decay_episodes: 1500 target_update: 200 hidden: 128 reward_scale: 0.05 seed: 7reward_scale先设 0.05对应平均工时 20 左右的实例。如果换到工时普遍 200 的实例把它降到 0.005让每步奖励仍然落在 -1 附近。eps_decay_episodes不是直接传给epsilon的衰减系数训练脚本里用eps_decay (eps_end / eps_start) ** (1 / eps_decay_episodes)算出来这样 1500 个 episode 后正好降到 0.05比手写 0.995 更好控。4.2 训练与评估命令三条命令跑完一轮依赖不多gym 用来对齐接口习惯真正用到的只是reset/step的返回格式torch 负责网络pandas 读 CSVpyyaml 读配置。Python 3.9 以上、torch 2.0 以上就够。python -m pip install -r requirements.txt python train.py --config configs/jobshop_20x6.yaml --seed 7 python evaluate.py --config configs/jobshop_20x6.yaml --ckpt runs/best.pt --baselines spt,fifo,randomtrain.py每 50 个 episode 打一行日志同时把当前最优 makespan 的权重存到runs/best.pt。判断“最优”不要用训练奖励用当轮 episode 的实际 makespan奖励里混了等待惩罚和排产目标不完全一致。evaluate.py加载权重后把epsilon固定为 0跑 30 次不同种子取均值和最差值再和 SPT、FIFO、随机策略对比。4.3 调参顺序奖励尺度、探索率、网络容量调参不要一上来改学习率。调度项目的第一敏感参数是奖励尺度第二是探索率第三才是网络容量。顺序反了你会把奖励问题误判成网络欠拟合然后不断加层最后训出一个只会选第一个工件的模型。顺序参数判断标准常见调整1reward_scale每步奖励是否落在 -1 附近按平均工时倒数缩放2eps_decay_episodes前 200 个 episode 是否还有探索实例越大衰减越慢3target_update损失曲线是否出现周期性尖峰尖峰频繁就调大4hidden损失降不下去且状态维度很大128 不够再上 2565lr损失震荡或下降过慢1e-4 上下一个数量级试奖励尺度不对的典型表现是训练奖励一直在涨但 makespan 不动。因为智能体学会了“少让工件等待”却不在乎总完工时间。探索率衰减太快的表现是前 100 个 episode 就收敛到某个固定动作序列换一个随机种子结果差别巨大。4.4 结果怎么看四个指标比一个 makespan 更有用只看 makespan 容易被单张工单带偏。评估脚本至少输出四个量makespan、平均流程时间、机器利用率、逾期工件数。makespan 看全局完工平均流程时间看工件在车间里平均待多久机器利用率看有没有大量空等逾期数看交期约束有没有被满足。def rollout(policy, env, seed0): np.random.seed(seed) obs env.reset() done False total_r 0.0 while not done: mask env.action_mask() action policy(obs, mask) obs, reward, done, info env.step(action) total_r reward return { makespan: info[makespan], reward: total_r, util: env.machine_utilization(), tardiness: env.total_tardiness(), }policy可以是 DQN 的贪心策略也可以是 SPT最短加工时间优先、FIFO先到先做、随机策略。对比时固定同一批种子每个策略跑 30 次。如果 DQN 只在某一张工单上赢换一张就输说明它记住了实例而不是学到了调度规则这时候要回头检查状态里是不是漏了交期或换型信息。5. 避坑与排查调度环境里最容易翻车的五个点调度强化学习的坑很集中一半在环境的时间轴一半在奖励和掩码。下面五条是这类项目里出现频率最高的每条按“现象 → 原因 → 解决”拆开写。遇到问题时先按顺序排查不要一上来就怀疑网络结构。5.1 损失在降makespan 不动现象是训练日志里 loss 从 2.0 降到 0.3但每 50 个 episode 打印的 makespan 一直在 180 到 200 之间晃。原因通常是稠密奖励权重过大智能体学会了“让当前这一步的空等最小”而不是“让整张工单早点完工”。idle和wait的系数把终局奖励淹没了。解决分两步先把终局奖励的权重从 1.0 提到 2.0 到 5.0让总完工时间在回报里占主导再把wait的系数从 0.2 降到 0.05避免智能体为了少等待而频繁切换工件。改完重训如果 makespan 开始随 episode 下降说明方向对了。如果还是不降检查done时奖励有没有真的加进去很多实现里done分支写在了return后面等于没生效。5.2 Q 值越训越大最后输出几万现象是打印 Q 值均值从几十涨到几千甚至几万动作选择变成固定选某一个工件。原因有三个常见组合奖励没有缩放、gamma太高、目标网络同步太频繁。三者叠加会让 TD 目标每一步都比当前 Q 大一点误差被反复放大。解决顺序是先把reward_scale调到每步奖励绝对值小于 1再把gamma从 0.99 降到 0.95最后把target_update从 50 步拉到 200 步以上。梯度裁剪保留 5.0不要因为 Q 值大就裁到 0.5裁太狠会让网络学不动。改完观察 Q 值均值正常应该在 -20 到 20 之间波动随 episode 缓慢上升而不是指数爆炸。5.3 动作全被 mask环境直接卡死现象是训练到某个 episode 突然报“非法动作”或者while not done死循环。原因通常是action_mask在done之后还被调用或者掩码用了int而不是bool~mask按位取反后把 0 变成 -1masked_fill的条件全乱。解决是在step入口加断言先把非法动作拦下来done为 True 时直接break不要再去取next_mask掩码统一用np.bool_并在select_action里打印一次mask.sum()确认每一步至少有一个合法动作。如果mask.sum()在某一步变成 0说明op_idx推进逻辑写错了检查是不是有工件被重复计数或漏加。5.4 同一张工单评估两次结果不一样现象是evaluate.py跑两次同一个权重、同一张工单makespan 差 10 以上。原因通常是评估时epsilon没设 0或者reset没有把machine_free、job_ready、op_idx全部清零上一轮的残留在时间轴里。解决是评估函数里显式epsilon0并在reset里用np.zeros重建所有数组不要用fill复用旧数组。随机种子也要在每次 rollout 前重设np.random.seed(seed)和torch.manual_seed(seed)都要。如果还复现不了检查action_mask的遍历顺序是不是依赖了字典或集合set的迭代顺序在不同进程里可能不同。5.5 GPU 上训练快评估反而慢现象是训练时每秒几十步评估时一个 episode 要好几秒。原因是评估循环里每一步都把单个obs搬到 GPU再取回 CPU 做argmax传输开销远大于计算。20 维动作的 MLP 在 CPU 上跑只要微秒级放 GPU 反而多一次同步。解决是评估阶段把模型留在 CPU或者把多个 episode 的观测拼成 batch 一起前向。训练阶段如果batch_size只有 128GPU 利用率也不高可以先在 CPU 上把流程跑通确认奖励和掩码没问题再上 GPU。不要为了“用上 GPU”而把 batch 调到 1024调度样本的相关性很强batch 太大反而稀释了关键转移。6. 二次开发把 DQN 排产接到真实工单上要改哪几处真要把这套代码往实际工单上靠第一处要改的不是网络是instance.py的字段。真实工单通常带交期、释放时间、换型矩阵、设备日历这些字段不进状态模型永远学不会优先做急单。我的做法是先加字段、再加环境断言最后才动动作空间。具体改造点可以按这张表推进改造点改哪里验证方式交期与释放时间instance.py、_obs()状态里能看到交期余量换型时间step()的start计算换型后机器空闲时间正确设备日历machine_free初始值非工作时段不开工动作空间action_mask()、Q 网络输出维度非法组合被过滤多目标奖励step()的 reward逾期数和 makespan 同时下降动作空间从“选工件”扩到“选机器-工件对”时Q 网络输出维度会从n_jobs变成n_jobs × n_machines掩码也要跟着变成二维。别急着上卷积或注意力先把二维掩码和masked_fill调通再考虑用工件特征做 embedding。我见过太多人一上来就改网络结构结果环境里的换型时间算错训了一周才发现模型学的是错误的时间轴。验证方法我一般用三板斧固定 30 个种子跑 DQN 和 SPT/FIFO比较均值和最差值把 DQN 的排产结果导出成甘特图人工看有没有明显违反工艺顺序最后挑三张最差的工单逐单复盘看是状态缺特征还是奖励给错了。第三板斧最费时间但也最容易发现“模型学会了绕过某个约束”的问题。我以前接真实工单时先改网络再改状态训了两天发现状态里没放交期模型根本学不会优先做急单。从那以后我每次动 DQN 之前都强制先把环境断言和状态字段过一遍再开训练。希望帮到你。本文还有配套的精品资源点击获取