简介：面向车间排产调度场景的深度强化学习 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) ], dtype=np.float32) remaining_time = np.array([ sum(self.instance.duration[j][self.op_idx[j]:]) for j in range(self.n_jobs) ], dtype=np.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) ], dtype=np.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, dtype=np.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 的实例一步奖励可能到 -50，Q 值很快冲到几百，训练曲线会非常难看。这个参数在配置文件里暴露出来，换实例时先调它，再调学习率。

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, replace=False) 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), m2

capacity取 50000 左右，对 20×6 的实例足够覆盖几百个 episode。ptr用循环覆盖，旧样本自然被淘汰。这里最容易漏的是mask2：如果只存s2不存下一状态的掩码，目标网络会把已完工工件也算进 max，Q 目标会虚高。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, hidden=128): 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, dtype=torch.float32, device=device).unsqueeze(0)) q = q.masked_fill(~torch.as_tensor(mask, device=device).unsqueeze(0), -1e9) return int(q.argmax(dim=1).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, dtype=np.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(f"ep={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: 7

reward_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 读 CSV；pyyaml 读配置。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,random

train.py每 50 个 episode 打一行日志，同时把当前最优 makespan 的权重存到runs/best.pt。判断“最优”不要用训练奖励，用当轮 episode 的实际 makespan，奖励里混了等待惩罚，和排产目标不完全一致。evaluate.py加载权重后把epsilon固定为 0，跑 30 次不同种子，取均值和最差值，再和 SPT、FIFO、随机策略对比。

4.3 调参顺序：奖励尺度、探索率、网络容量

调参不要一上来改学习率。调度项目的第一敏感参数是奖励尺度，第二是探索率，第三才是网络容量。顺序反了，你会把奖励问题误判成网络欠拟合，然后不断加层，最后训出一个只会选第一个工件的模型。

奖励尺度不对的典型表现是：训练奖励一直在涨，但 makespan 不动。因为智能体学会了“少让工件等待”，却不在乎总完工时间。探索率衰减太快的表现是：前 100 个 episode 就收敛到某个固定动作序列，换一个随机种子结果差别巨大。

4.4 结果怎么看：四个指标比一个 makespan 更有用

只看 makespan 容易被单张工单带偏。评估脚本至少输出四个量：makespan、平均流程时间、机器利用率、逾期工件数。makespan 看全局完工，平均流程时间看工件在车间里平均待多久，机器利用率看有没有大量空等，逾期数看交期约束有没有被满足。

def rollout(policy, env, seed=0): 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 变成 -1，masked_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全部清零，上一轮的残留在时间轴里。

解决是评估函数里显式epsilon=0，并在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只有 128，GPU 利用率也不高，可以先在 CPU 上把流程跑通，确认奖励和掩码没问题再上 GPU。不要为了“用上 GPU”而把 batch 调到 1024，调度样本的相关性很强，batch 太大反而稀释了关键转移。

6. 二次开发：把 DQN 排产接到真实工单上要改哪几处

真要把这套代码往实际工单上靠，第一处要改的不是网络，是instance.py的字段。真实工单通常带交期、释放时间、换型矩阵、设备日历，这些字段不进状态，模型永远学不会优先做急单。我的做法是先加字段、再加环境断言，最后才动动作空间。

具体改造点可以按这张表推进：

动作空间从“选工件”扩到“选机器-工件对”时，Q 网络输出维度会从n_jobs变成n_jobs × n_machines，掩码也要跟着变成二维。别急着上卷积或注意力，先把二维掩码和masked_fill调通，再考虑用工件特征做 embedding。我见过太多人一上来就改网络结构，结果环境里的换型时间算错，训了一周才发现模型学的是错误的时间轴。

验证方法我一般用三板斧：固定 30 个种子跑 DQN 和 SPT/FIFO，比较均值和最差值；把 DQN 的排产结果导出成甘特图，人工看有没有明显违反工艺顺序；最后挑三张最差的工单逐单复盘，看是状态缺特征还是奖励给错了。第三板斧最费时间，但也最容易发现“模型学会了绕过某个约束”的问题。

我以前接真实工单时，先改网络再改状态，训了两天发现状态里没放交期，模型根本学不会优先做急单。从那以后我每次动 DQN 之前，都强制先把环境断言和状态字段过一遍，再开训练。希望帮到你。

本文还有配套的精品资源，点击获取