简介：基于DQN解决的柔性作业车间动态调度问题，属于深度强化学习与传统运筹优化的交叉方向。项目针对生产过程中突发订单插入导致的扰动，将车间状态、可用机器和工序约束建模为智能体交互环境，通过深度Q网络学习最优调度策略，适合人工智能、自动化、工业工程、智能制造等专业学生用于毕业设计、课程设计、大作业或竞赛初期立项。压缩包共9个文件，约503KB，其中5个Python脚本构成核心代码，分别实现DQN算法、作业车间环境、实例生成器与对象封装；另有3个Python缓存文件和1篇参考论文PDF，便于直接运行和对照理解算法原理。目前已有167人学习或下载，项目经过本地运行验证，可快速复现调度决策流程。代码采用模块化结构，训练入口、环境交互、样本实例和网络定义层次清晰，可在现有基础上调整车间规模、插单策略或奖励函数进行二次开发，配套PDF文档也提供了相关研究背景，适合希望深入探索深度强化学习在调度领域应用的读者。
1. 带插单的柔性作业车间调度，是时候把决策交给DQN了
柔性作业车间调度（FJSP）比传统作业车间多一层选择：每道工序可以在多台机器上加工，加工时间随机器不同而不同；再加上“带插单”这个前提，难度立刻翻倍——新订单在任意时刻插入，刚排好的甘特图必须被局部推翻，重新回答“接下来哪个工件优先、放到哪台机器”。传统优先规则如SPT、MWKR响应快，但容易把局部忙乱放大成全局排队；元启发式算法能得到接近最优的解，可每次重调度都要跑上几十分钟，插单节奏快时根本等不起。DQN把调度建模成序贯决策问题，在每个决策点输入车间状态，输出“该调度哪个工序-机器对”，一次前向传播就能给出动作，推理成本低，且能从历史调度交互中学习长期收益。这套源码把DQN完整落到FJSP动态仿真上，并显式支持插单事件，是读强化学习调度方向时值得拆开看的第一份工程实现。
2. 定义FJSP的MDP：状态空间、动作空间与奖励函数
在动DQN.py之前，先把MDP四要素对应到仿真代码上。强化学习在调度问题里的大部分失败都不是网络没调好，而是动作空间设计不当或奖励信号太稀疏。下面按状态、动作、奖励三个维度拆解本项目的一般设计思路，后续读代码时可以直接对号入座。
2.1 状态特征：把车间编码成定长向量
DQN要求输入维度固定，但调度问题天然是变长的：插单导致工件数不定，机器空闲状态随时间切换，每台机器的负载也不一样。常见做法是先设定环境容量上限，包括最大机器数、最大工件数、最大工序数，再按块拼接定长向量。我在类似项目里使用的编码方式如下：
def get_state_vector(self): state = [] # 机器块：每台机器两个特征，空闲标志与剩余加工时间占比 for m in self.machines: state.append(1.0 if m.is_idle() else 0.0) state.append(m.remaining_time() / self.max_span()) # 工件块：每个工件的已完成工序比例和剩余工序占比 for j in self.jobs: state.append(j.done_ratio()) state.append(j.remaining_ops() / j.total_ops()) # 全局块：就绪工序数、当前时间占比与插单标记 state.append(len(self.ready_ops) / self.max_ready_ops()) state.append(self.current_time / self.horizon()) state.append(1.0 if self.new_order_arrived else 0.0) return np.asarray(state, dtype=np.float32)编码的逻辑是分层描述：机器块回答“设备是否可用”，工件块回答“每个工件进度如何”，全局块回答“系统整体紧不紧张”。插单发生后，新工件进入jobs列表，工件块尾部出现新工件的进度特征，同时ready_ops集合变大，所以智能体输入几乎立刻反映扰动，不需要等调度结果出来才知道有插单。这里有一个容易踩的坑：状态向量里不要放绝对加工时间，因为插单到达时刻不同，同样的机器占用时长在不同时间点含义完全不同，必须转成占比或相对量，训练策略才可能跨场景泛化。luo2020.pdf作为配套论文，其中对每个特征维度的消融实验验证了这类归一化方式的有效性。
另外注意DQN.py输入层的维度必须和get_state_vector返回的长度严格一致。改Instances配置增大了容量上限，第一件事就是去检查state_shape，否则训练启动时就会报维度不匹配错误。
2.2 动作空间：直接选择工序-机器对，还是间接输出优先级
FJSP动作空间有两种常见设计。第一种把每个“就绪工序 × 可选机器”组合作为动作，动作空间大小等于所有可选组合数，随调度推进不断变化。第二种只输出工件编号，由仿真器内部选定“最早可加工工序 + 最早可用机器”来执行，动作空间小、收敛快，但策略上限受内部规则限制。
本项目源码更接近第一种：DQN输出所有动作的Q值，但训练时用掩码屏蔽非法动作。就绪工序指前序工序已完成且所属工件尚未完工的工序，只有这些工序与机器组成的对可以执行。
def build_action_mask(self): mask = np.zeros(self.n_actions_total, dtype=np.float32) valid = self.env.get_valid_action_ids() # 返回[(工序id, 机器id), ...] mask[valid] = 1.0 return mask在choose_action里对这个mask做处理：无效动作的Q值置为-1e9，再走argmax。这样无论epsilon-greedy还是贪心选择，随机探索也只会在合法工序-机器对里发生。这个掩码是动态调度实现的重点，插单之后就绪工序集合变大，mask的合法位置立即增加，避免了“新订单来了但模型不知道它可被调度”的问题。
动作空间还有一个细节：不同机器加工同一工序的时间不同，动作本身包含机器选择，DQN输出的Q值必须能够区分“把工序放到这台机器”和“放到另一台机器”的差别。若某道工序可选机器数非常多，动作空间会变得很大，训练初期需要更多探索，这也是前面epsilon要设高的原因。
2.3 奖励函数：完工时间差分与空闲惩罚
调度问题的奖励设计有个典型坑，直接拿回合结束时的makespan做奖励会导致信号太稀疏，插单发生在中段，直到回合结束才能拿到一次反馈，模型基本学不动。常见做法是使用“执行动作前后预估完工时间的变化量”作为即时奖励，让每一步都携带梯度信息。
def compute_reward(self, pre_estimate, post_estimate): r = pre_estimate - post_estimate if self.is_idle_but_ops_ready(): r -= self.idle_penalty return rpre_estimate是动作执行前所有工件最晚预计完成时间，post_estimate是将工序放到指定机器新位置之后的预计完成时间。若动作把紧急工序放到了关键机器上，完工时间估计下降，奖励为正；若把工序排到末尾导致整条产线拖延，奖励为负。空闲惩罚解决的是“机器空转但就绪工序没人处理”的异常被低估的问题，常见的idle_penalty取0.1到0.2：取值过大时正奖励会被淹没，累计回报一直为负；过小则模型会学成“让机器闲着”，甘特图上出现大量空洞。
奖励函数定义通常独立放在Object_for_FJSP.py里，训练时环境只负责推进时间，目标计算单独封装，后续替换目标函数不用改动环境。读源码时建议先看这个文件里的预估完工时间是怎么算的，它决定了整个训练信号的语义。
3. 源码拆解：实例生成、车间仿真、DQN主循环的衔接
这套源码的工程结构适合按数据流顺序读：Instance_Generator.py生成工件实例，Job_Shop.py推进车间仿真，Object_for_FJSP.py计算调度目标，DQN.py维护网络和训练逻辑，main.py把插单事件和训练循环串起来。这个分层递进清晰，改模块时不容易互相污染，也是拿来做毕设、课程设计或期末大作业时应该保留的骨架。
3.1 Instance_Generator.py：可控的工件实例与插单序列生成
Instance_Generator.py负责生成FJSP实例，包括各工件的工序数、每道工序可选机器集合和加工时间。作为整套实验的可重复性来源，生成器通常会暴露随机种子参数：
class InstanceGenerator: def __init__(self, seed=42, n_machines=6, n_initial_jobs=10): self.rng = np.random.default_rng(seed) self.n_machines = n_machines self.n_initial_jobs = n_initial_jobs def gen_job(self, release_time=None): n_ops = self.rng.integers(3, 8) job = { 'id': self.next_id, 'ops': [{ 'machine_options': self.rng.choice( self.n_machines, size=self.rng.integers(1, 3), replace=False ).tolist(), 'processing_time': self.rng.integers(5, 20) } for _ in range(n_ops)], 'release_time': release_time } self.next_id += 1 return job这段逻辑生成了3到7道工序的工件，每道工序在1到2台机器上可选，加工时间均匀分布在5到20之间。release_time为None表示初始工件在0时刻全部就绪，插单工件则会被赋上到达时刻。把插单时间点与初始工件总加工时间挂钩，比完全随机更公平。例如在总工时窗口的30%、60%处各插入一个高优先级短工件，模型能稳定观察到扰动影响，若完全随机，中段插单和末尾插单难度差异太大会让训练曲线剧烈震荡。
Instance_Generator.py还有一个作用是为训练和测试提供同分布数据。测试时固定种子，训练时换种子，验证模型是否真的学到了调度策略而不是背下了某组工件排列。
3.2 Job_Shop.py和Object_for_FJSP.py：仿真环境与目标函数封装
Job_Shop.py是离散事件仿真环境，负责推进时间、管理机器、维护每个工序的完工状态。核心是step(action)：
def step(self, action): op_id, machine_id = action op = self.ops[op_id] # 开工时间取机器空闲时间和前序工序完成时间的较大者 start = max(self.machines[machine_id].free_at, op.release_at) finish = start + op.processing_time # 排入机器时间轴，标记工序完成 self.machines[machine_id].schedule(op_id, start, finish) op.mark_scheduled(finish) # 从就绪集合移除，并激活该工件的下一道工序 self.ready_ops.remove(op_id) nxt = op.successor() if nxt: nxt.release_at = finish self.ready_ops.add(nxt.id) self.current_time = finish return finish这里把当前时间推进到本次工序完工时刻，属于事件驱动推进。机器的free_at记录这台机器上所有已排工序的累计完工时间，新工序开工不会早于它；工序的release_at来自前序工序完工时间，两条时间约束合在一起，保证任何动作序列都不会排出不可行的甘特图。这个设计决定了动作的执行不依赖真实时间流逝，而是一次跳到一个决策点，训练采样效率远高于固定步长仿真。
Object_for_FJSP.py把目标函数从环境中拆出来，接口通常包括get_makespan、get_machine_load_list等。环境只负责状态迁移，评价交给这个模块，后期把makespan换成加权延迟、最大延迟或能耗目标，Job_Shop.py完全不用动。工程上这一层隔离对实验对比帮助很大。
3.3 DQN.py：Q网络、经验回放、目标网络的实现层级
DQN.py是大脑，网络本身不复杂，将前面得到的状态向量映射到所有动作的Q值：
class QNet(nn.Module): def __init__(self, n_state, n_action, hidden=128): super().__init__() self.fc = nn.Sequential( nn.Linear(n_state, hidden), nn.ReLU(), nn.Linear(hidden, hidden), nn.ReLU(), nn.Linear(hidden, n_action) ) def forward(self, s): return self.fc(s)真正影响训练稳定性的是经验回放和目标网络。回放池里存储(s, a, r, s', done)五元组，采样时打乱顺序，打断相邻状态下决策的强相关性。目标网络有两种更新方式，硬更新每C步直接把online参数复制过来，软更新target = tau * online + (1 - tau) * target。FJSP状态变化连续，硬更新会让Q目标跳变明显，我一般优先选软更新，tau取0.005。
DQN.py里还需要结合动作掩码做选择：
def choose_action(self, state, mask): if np.random.rand() < self.epsilon: valid = np.where(mask == 1.0)[0] return int(np.random.choice(valid)) with torch.no_grad(): q = self.q_net(torch.FloatTensor(state).unsqueeze(0)) q = q.masked_fill(torch.from_numpy(mask == 0), -1e9) return int(q.argmax(dim=1).item())随机探索只在合法动作中选取，贪心动作则先把非法Q值压到极小再取最大值。这两个分支都依赖mask，否则探索会撞上非法调度。
3.4 main.py：主循环如何把插单事件与训练串起来
main.py把以上模块串成完整训练流，主干如下：
for episode in range(args.episodes): env = JobShopEnv(instance_gen.generate_initial_set()) agent = DQN(state_dim, action_dim) state = env.reset() while not env.is_finished(): # 时间推进过程中检查插单事件 if env.has_new_arrival(): new_job = instance_gen.gen_job(release_time=env.current_time) env.insert_job(new_job) state = env.get_state_vector() # 插单后立即刷新状态 mask = env.build_action_mask() action = agent.choose_action(state, mask) next_state, reward = env.step(action) agent.store(state, action, reward, next_state, mask) agent.update() state = next_state这里最容易忽视的是插单后要重新取一次状态向量。若新工件已经插入jobs列表但state还是插单前采集的，智能体给出的动作不会考虑新工件，表现为对插单反应滞后。另外mask也要在插单后重新构建，否则新工件的工序不会被纳入可选动作，插单等于没发生。
main.py还承担超参数入口职责，命令行传入episodes、lr、gamma等，方便批量实验。训练过程中建议每隔一定episode计算一次当前策略在固定测试集上的makespan，单独记录，避免只在训练环境里自评。
4. 训练与调参：让DQN在FJSP上真正收敛的工程经验
神经网络跑通不代表学到调度策略。常见现象是loss曲线一直震荡，或者累计回报长期为负不回升。对FJSP这类组合优化问题，超参数的有效区间其实很窄，下面给出经过多轮实验验证的起点区间，然后再谈诊断方法。
4.1 超参数组合：学习率、批量、折扣因子的相互作用
| 参数 | 建议范围 | 说明 |
|---|---|---|
| 学习率 lr | 1e-4 ~ 5e-4 | 过大Q值震荡明显，曲线呈锯齿状 |
| 折扣因子 gamma | 0.90 ~ 0.95 | 调度序列长，过大容易带进远端噪声 |
| batch_size | 64 ~ 128 | 过小Q值方差大，过大更新频率下降 |
| replay buffer | 50000 ~ 200000 | FJSP状态空间大，池子太小学不到分布 |
| 目标网络更新 | tau=0.005 软更新 | 硬更新会导致Q目标跳变 |
| epsilon初始 | 0.9 ~ 1.0 | 前期必须充分探索工序-机器组合 |
| epsilon终止 | 0.03 ~ 0.05 | 后期保留少量探索应对插单变化 |
其中gamma的选择常被忽视。插单场景下动作影响往往在几十步之后才在makespan上体现，gamma太低会让模型只顾眼前机器空闲；但gamma过高，每一步的噪声都会累积，Q值收敛慢。实际调试中以0.95起步，如果发现训练后期Q值方差大，降到0.9再跑一轮。
4.2 经验回放和目标网络更新的节奏
FJSP相邻状态高度相关，尤其插单事件发生后，连续几步产生的高负反馈样本会在批次中占比过大，把参数往一个方向猛推。回放池的作用就是打散这种时间相关性。如果训练中Q值突然爆掉，可以降低更新频率，常见做法是每收集5步样本才做一次梯度更新，而不是每步更新。buffer size太小会导致老样本过早被淘汰，模型忘掉早期学到的调度知识。
探索率epsilon的衰减也值得单列。推荐按episode比例衰减而不是按step衰减。FJSP一个episode步数随实例变化很大，按step衰减会出现不同实例探索程度不一致。例如总800个episode，前200个保持epsilon在0.9以上，中间400个线性降到0.1，最后200个固定0.05。若训练后期调度质量仍然波动，把最小epsilon提高到0.1，给插单场景留出更多随机试探空间。
4.3 用收敛曲线与makespan双重指标判断训练状态
训练日志里同时记录累计回报和平均makespan。累计回报受奖励函数缩放影响，不同idle_penalty下数值大小没有可比性；makespan是绝对时间指标，能直接反映调度质量。只有两者一起看才能避免“回报曲线好看但调度结果变差”的假象。
python main.py --episodes 800 --lr 2e-4 --gamma 0.95 \ --batch-size 64 --buffer-size 100000 --log-every 50导出的训练日志可以做简单对比：取前200个episode的平均makespan和后200个episode的平均makespan。后段比前段低10%以上，基本说明策略学到了应对插单的调度经验。如果后段反而升高，多半是奖励函数设计问题，优先检查idle_penalty是不是过大。模型学到的是“宁让机器空转也不接插单”，因为接了高紧急性工序会带来更大的完工时间差分负奖励。这种情况把idle_penalty调低到0.05，或者改为只在机器空闲超过阈值时才施加惩罚。
5. 插单场景的验证技巧：从训练日志到动态扰动测试
训练完成后需要验证DQN对插单的实际应对能力，不能只盯着训练集表现。推荐做一个固定测试协议：同一组初始工件，三种扰动模式，每种跑20个随机种子取平均。
| 场景 | 扰动设置 | 评估指标 |
|---|---|---|
| 无插单 | 不注入新工件 | makespan |
| 中段插单 | 在初始完工时间估计的50%处插入1个工件 | makespan、延迟增量 |
| 末段插单 | 在80%处插入1个高优先级短工件 | makespan、延迟增量 |
测试时把agent.eval模式打开，关闭探索，使用贪心动作。逐行检查插单注入代码时，重点关注insert_job之后是否刷新了状态向量和动作掩码。项目里main.py的event检查位置，我建议每次插单后打印当前时间、就绪工序数、机器空闲数，确认新工件真实进入了调度决策范围。
另一个容易被忽略的细节是目标网络在验证时的状态。验证阶段不更新参数，DQN.py里不应继续向回放池写入样本，也不应做梯度更新，否则验证结果会被学习过程干扰。常见做法是把update函数的总开关在测试循环里置为False，这样训练和验证走同一套choose_action逻辑，只是少了探索和回放写入。
最后对照luo2020.pdf里的曲线检查复现程度：reward曲线形状是否同趋势，插单后DQN是否倾向把剩余工序最少或剩余加工时间最短的工件优先放到关键机器上。如果观察不到这种偏好，先把验证样本的工件规模调小到6台机器、8个初始工件，再逐步放大。这一检查过程能准确定位问题出在状态编码、奖励设置还是参数区间，比盲目调网络宽度有效得多。
本文还有配套的精品资源，点击获取
