尧图网站开发 
简介本资源为基于DQN解决带插单的柔性作业车间动态调度问题项目Python源码面向计算机、人工智能、自动化、机械电子等专业在校学生及行业从业人员可作为毕业设计、课程设计、期末大作业或比赛初期立项演示的参考方案。压缩包共9个文件约503KB以5个py源码文件为主涵盖主程序入口、DQN算法实现、FJSP对象建模、实例生成器与作业车间调度等核心模块另含3个pyc编译缓存文件及1份luo2020.pdf参考文献便于理解算法背景与实现细节。目前已有167人学习下载。项目将深度强化学习与带插单的动态调度场景结合代表性与创新性较强读者可从中获取完整的算法建模思路、网络结构设计与训练流程并在此基础上进行二次开发或迁移到其他调度问题适合入门进阶与实战借鉴。1. 带插单的柔性作业车间调度为什么传统方法一到动态场景就翻车柔性作业车间调度FJSP本身就是 NP-hard 问题工序可以在多台机器上选择加工解空间比经典 JSP 大了一个量级。而“带插单”意味着生产过程中随时可能有紧急订单插入原有调度方案必须在线调整。传统方法在这里会遇到两个硬伤一是遗传算法、禁忌搜索这类元启发式方法每次重调度都要重新迭代响应时间从秒级跳到分钟级车间等不起二是基于规则的启发式如最短加工时间优先虽然快但缺乏全局优化能力插单后容易把整体交期拖垮。DQNDeep Q-Network之所以被引入这个场景核心逻辑是把调度决策建模为马尔可夫决策过程用神经网络学习“当前车间状态下选哪个工序-机器组合最优”的策略。训练完成后推理阶段只需一次前向传播毫秒级输出决策天然适合动态插单场景。这个项目标题指向的是一套 Python 源码实现适合运筹优化方向的研究生、智能制造方向的工程师以及想用深度强化学习落地车间调度的从业者。它解决的不是“调度理论”问题而是“插单来了怎么在秒级内给出可执行方案”的工程问题。2. 把车间调度翻译成 DQN 能吃的四元组状态、动作、奖励、终止2.1 状态设计用什么特征描述一个车间DQN 的输入是状态向量状态设计直接决定模型能不能学到有效策略。在带插单的 FJSP 中我一般会把状态拆成四个部分工序状态每个待加工工序的剩余工序数、当前工序在工艺路线中的位置、是否已被调度。机器状态每台机器的当前可用时间、正在加工的工序剩余时间、历史利用率。订单状态每个订单的交期紧迫度交期减去当前时间再除以剩余工时、是否为插单、插单优先级。全局状态当前完工时间makespan、平均机器利用率、已插单数量。这些特征拼成一个定长向量。注意工序数和机器数在不同算例中可能不同所以状态向量要么做归一化后补齐到固定长度要么用注意力机制处理变长输入。源码项目里通常采用固定最大工序数的做法简单直接。import numpy as np def build_state(jobs, machines, current_time, max_ops20, max_machines10): 构建 DQN 状态向量 jobs: 订单列表每个订单包含工序列表和交期 machines: 机器列表每台机器有 available_time current_time: 当前调度时刻 state [] # 工序特征剩余工序数、当前工序索引、是否已调度 for job in jobs[:max_ops]: remaining len(job[ops]) - job[current_op_idx] state.extend([ remaining / max_ops, job[current_op_idx] / max_ops, 1.0 if job[scheduled] else 0.0 ]) # 补齐 while len(state) max_ops * 3: state.extend([0.0, 0.0, 0.0]) # 机器特征可用时间归一化、利用率 for m in machines[:max_machines]: state.append(m[available_time] / (current_time 1e-6)) while len(state) max_ops * 3 max_machines: state.append(0.0) # 订单交期紧迫度 for job in jobs[:max_ops]: urgency (job[due_date] - current_time) / (sum(job[proc_times]) 1e-6) state.append(np.clip(urgency, -1, 1)) while len(state) max_ops * 3 max_machines max_ops: state.append(0.0) return np.array(state, dtypenp.float32)这段代码的关键参数是max_ops和max_machines它们决定了状态向量的维度。设太小会截断信息设太大会让网络输入稀疏。我一般按算例中最大工序数的 1.2 倍来设。归一化用current_time 1e-6防止除零交期紧迫度用clip限制在 [-1,1] 避免梯度爆炸。2.2 动作空间选工序还是选机器动作空间有两种常见建模方式方式一联合动作。每个动作是一个(工序, 机器)对。动作数量等于待调度工序数乘以可选机器数。优点是决策直接缺点是动作空间随规模线性增长插单后动作空间会变化。方式二分层动作。先选工序再选机器。两个网络或一个网络两个输出头。优点是动作空间小缺点是训练时信用分配更复杂。源码项目里通常用方式一因为实现简单且插单场景下动作空间变化可以通过掩码处理。具体做法是对当前不可选的动作工序已调度或机器不可用输出一个极大的负值softmax 后概率为 0。def get_action_mask(jobs, machines): 返回动作掩码1 表示可选0 表示不可选 mask [] for job in jobs: if job[scheduled] or job[current_op_idx] len(job[ops]): mask.extend([0] * len(machines)) else: op job[ops][job[current_op_idx]] for m_idx, m in enumerate(machines): # 机器能加工该工序且当前可用 if m_idx in op[eligible_machines]: mask.append(1) else: mask.append(0) return np.array(mask, dtypenp.float32)掩码机制是 DQN 做调度时最容易被忽略的细节。没有掩码网络会学到大量无效动作训练效率极低。掩码在每步调度前重新计算因为机器可用状态和工序状态都在变。2.3 奖励函数让 DQN 学会“插单不拖交期”奖励设计是 DQN 调度最玄学的地方。常见做法有三种稀疏奖励只在完工时给-makespan。缺点是中间步骤没有反馈训练慢。稠密奖励每步给-加工时间或-等待时间。优点是反馈快缺点是可能短视。混合奖励每步给-加工时间完工时额外给-max(0, makespan - due_date)惩罚交期延误。我一般用混合奖励并在插单发生时给一个额外惩罚项让模型学会优先处理插单。具体公式def compute_reward(job, machine, current_time, is_inserted, due_date): proc_time job[ops][job[current_op_idx]][proc_times][machine] reward -proc_time * 0.01 # 加工时间惩罚 if is_inserted: reward - 0.5 # 插单额外惩罚鼓励优先处理 finish_time current_time proc_time if finish_time due_date: reward - (finish_time - due_date) * 0.1 # 交期延误惩罚 return reward参数0.01、0.5、0.1需要根据算例规模调。加工时间惩罚系数太小模型不在乎效率太大模型会忽略交期。插单惩罚系数一般设成加工时间惩罚的 10 到 50 倍具体看插单频率。2.4 终止条件与经验回放终止条件很简单所有工序都被调度完毕。但插单场景下终止条件要加一条如果插单导致原方案完全不可行比如交期已过提前终止并给大负奖励。经验回放池存(state, action, reward, next_state, done)五元组。插单场景下回放池要保证新旧数据混合采样否则模型会遗忘旧策略。我一般设回放池大小 10000 到 50000batch size 64 到 128。3. 用 Python 把 DQN 调度器跑起来网络结构、训练循环与插单注入3.1 网络结构三层全连接够不够调度问题的状态向量维度通常在几百到几千动作空间在几十到几百。三层全连接网络输入层、两个隐藏层、输出层足够。隐藏层神经元数一般设 256 和 128激活函数用 ReLU。import torch import torch.nn as nn class DQN(nn.Module): def __init__(self, state_dim, action_dim, hidden1256, hidden2128): super(DQN, self).__init__() self.net nn.Sequential( nn.Linear(state_dim, hidden1), nn.ReLU(), nn.Linear(hidden1, hidden2), nn.ReLU(), nn.Linear(hidden2, action_dim) ) def forward(self, x): return self.net(x)如果状态向量超过 2000 维建议加一层 BatchNorm 或 LayerNorm否则训练初期 loss 震荡严重。动作空间超过 500 时输出层可以加一个 dueling 结构把 Q 值拆成状态价值和优势函数提升稳定性。3.2 训练循环epsilon 衰减和 target network 更新DQN 训练的核心是两件事epsilon-greedy 探索和 target network 软更新。import random from collections import deque def train_dqn(env, episodes5000, gamma0.95, lr1e-3, epsilon_start1.0, epsilon_end0.05, epsilon_decay0.995, target_update10, batch_size64): state_dim env.state_dim action_dim env.action_dim policy_net DQN(state_dim, action_dim) target_net DQN(state_dim, action_dim) target_net.load_state_dict(policy_net.state_dict()) optimizer torch.optim.Adam(policy_net.parameters(), lrlr) replay_buffer deque(maxlen50000) epsilon epsilon_start for episode in range(episodes): state env.reset() done False while not done: # epsilon-greedy 选择动作 if random.random() epsilon: action env.sample_valid_action() else: with torch.no_grad(): q_values policy_net(torch.FloatTensor(state).unsqueeze(0)) mask torch.FloatTensor(env.get_action_mask()) q_values q_values (1 - mask) * (-1e9) action q_values.argmax().item() next_state, reward, done, info env.step(action) replay_buffer.append((state, action, reward, next_state, done)) state next_state # 经验回放 if len(replay_buffer) batch_size: batch random.sample(replay_buffer, batch_size) states, actions, rewards, next_states, dones zip(*batch) states torch.FloatTensor(states) actions torch.LongTensor(actions).unsqueeze(1) rewards torch.FloatTensor(rewards).unsqueeze(1) next_states torch.FloatTensor(next_states) dones torch.FloatTensor(dones).unsqueeze(1) q_values policy_net(states).gather(1, actions) with torch.no_grad(): next_q target_net(next_states).max(1)[0].unsqueeze(1) target_q rewards gamma * next_q * (1 - dones) loss nn.MSELoss()(q_values, target_q) optimizer.zero_grad() loss.backward() optimizer.step() epsilon max(epsilon_end, epsilon * epsilon_decay) if episode % target_update 0: target_net.load_state_dict(policy_net.state_dict()) return policy_net关键参数说明gamma0.95表示模型关注未来 20 步左右的奖励调度问题一般设 0.9 到 0.99。epsilon_decay0.995表示探索率每轮衰减 0.5%5000 轮后降到 0.05 以下。target_update10表示每 10 轮同步一次 target network太频繁会失去稳定性太慢会滞后。3.3 插单注入怎么模拟动态事件插单注入要在训练和测试时都做。训练时随机在某个调度步插入新订单测试时按固定概率插入。def inject_order(env, prob0.1): 以 prob 概率插入新订单 if random.random() prob: new_job generate_random_job(env.num_machines) env.jobs.append(new_job) env.state_dim env.compute_state_dim() # 状态维度可能变化 return True return False注意插单后状态维度可能变化如果网络输入维度固定需要把新订单特征映射到已有维度。常见做法是预留插单槽位或者用固定最大订单数。源码项目里一般用后者简单但浪费一些维度。3.4 训练效果验证看 makespan 和交期延误率训练过程中要监控两个指标平均 makespan 和交期延误率。前者衡量效率后者衡量插单响应能力。def evaluate(policy_net, env, episodes100): makespans [] tardiness [] for _ in range(episodes): state env.reset() done False while not done: with torch.no_grad(): q_values policy_net(torch.FloatTensor(state).unsqueeze(0)) mask torch.FloatTensor(env.get_action_mask()) q_values q_values (1 - mask) * (-1e9) action q_values.argmax().item() state, _, done, info env.step(action) makespans.append(info[makespan]) tardiness.append(info[tardiness]) return np.mean(makespans), np.mean(tardiness)如果 makespan 在 1000 轮后还在震荡检查奖励尺度是否太大或者学习率是否太高。如果交期延误率居高不下检查插单惩罚系数是否太小。4. 避坑带插单 FJSP 做 DQN 时最容易翻车的五个地方4.1 状态维度不固定导致网络报错现象插单后state_dim变化policy_net输入维度不匹配直接抛异常。原因状态向量按实际订单数动态拼接插单后长度变了。解决固定最大订单数和最大工序数不足补零。或者用nn.Linear之前加一个自适应池化层。我一般用固定维度简单可靠。4.2 动作掩码忘记在 target Q 值上也加现象训练 loss 正常下降但推理时选出的动作大量无效调度器卡死。原因计算 target Q 值时只用了max(1)[0]没有加掩码导致 target 值被无效动作拉高。解决target Q 值计算也要加掩码with torch.no_grad(): next_q_all target_net(next_states) next_mask torch.FloatTensor(np.array([env.get_action_mask() for _ in range(batch_size)])) next_q_all next_q_all (1 - next_mask) * (-1e9) next_q next_q_all.max(1)[0].unsqueeze(1)4.3 奖励尺度太大导致梯度爆炸现象loss 变成 NaN网络权重全是 nan。原因makespan 通常在几百到几千直接用作奖励Q 值会到几千MSE loss 轻松到百万级。解决奖励归一化。加工时间除以最大加工时间makespan 除以基准 makespan。或者用 Huber loss 替代 MSE。4.4 插单频率太高导致模型只学“应付插单”现象插单响应快了但整体 makespan 比规则调度还差。原因插单惩罚系数太大模型过度优先插单牺牲了整体效率。解决插单惩罚系数从 0.1 开始调逐步增加观察 makespan 和延误率的 trade-off。一般插单惩罚系数是加工时间惩罚的 5 到 20 倍比较合理。4.5 回放池被新经验淹没导致灾难性遗忘现象训练后期模型在旧算例上表现突然变差。原因插单产生的新经验不断覆盖回放池旧经验被挤出。解决回放池设大一点50000 以上或者用优先经验回放给旧经验更高采样权重。我一般还会保留一个“旧策略测试集”每 100 轮跑一次掉点就降低学习率。5. 让 DQN 调度器真正可用的三个进阶技巧5.1 用 Double DQN 消除 Q 值高估标准 DQN 的max操作会系统性高估 Q 值在调度问题里表现为模型对某些动作过度自信。Double DQN 用 policy_net 选动作target_net 算 Q 值with torch.no_grad(): next_actions policy_net(next_states).argmax(1, keepdimTrue) next_q target_net(next_states).gather(1, next_actions) target_q rewards gamma * next_q * (1 - dones)改动很小但训练稳定性提升明显。我一般默认用 Double DQN。5.2 用优先经验回放加速收敛调度问题中插单发生前后的经验价值很高但随机采样可能漏掉。优先经验回放按 TD 误差采样TD 误差大的经验被采概率高。class PrioritizedReplayBuffer: def __init__(self, capacity, alpha0.6): self.capacity capacity self.alpha alpha self.buffer [] self.priorities np.zeros(capacity, dtypenp.float32) self.pos 0 def push(self, transition, td_error): self.buffer.append(transition) self.priorities[self.pos] (np.abs(td_error) 1e-6) ** self.alpha self.pos (self.pos 1) % self.capacity def sample(self, batch_size, beta0.4): probs self.priorities[:len(self.buffer)] / self.priorities[:len(self.buffer)].sum() indices np.random.choice(len(self.buffer), batch_size, pprobs) weights (len(self.buffer) * probs[indices]) ** (-beta) weights weights / weights.max() return [self.buffer[i] for i in indices], indices, weightsalpha0.6控制优先级程度beta0.4控制重要性采样权重。这两个参数不用太纠结默认值就能用。5.3 用规则调度做 warm start纯 DQN 从零训练前几千轮基本在随机试错。用规则调度如 SPT生成一批经验先灌进回放池再开始 DQN 训练收敛速度能快一倍。def warm_start(env, policy_net, episodes100): 用 SPT 规则生成经验 for _ in range(episodes): state env.reset() done False while not done: action env.spt_action() # 最短加工时间优先 next_state, reward, done, _ env.step(action) replay_buffer.append((state, action, reward, next_state, done)) state next_statewarm start 之后epsilon 可以从 0.5 开始不用从 1.0 开始省掉大量无效探索。5.4 验证方法和规则调度、遗传算法对比训练完不能只看 loss要跑对比实验。我一般设三组基线SPT 规则、遗传算法迭代 1000 代、随机调度。DQN 在 makespan 上应该比 SPT 好 10% 到 20%比遗传算法差 5% 以内但推理时间快两个数量级。如果 DQN 比 SPT 还差检查奖励设计和掩码如果比遗传算法差太多检查网络容量和训练轮数。最后说个血泪教训我最早做这个方向时花了两个月调网络结构从三层加到七层效果提升不到 3%。后来发现把奖励函数从稀疏改成混合效果直接提升 15%。在调度问题里奖励设计比网络结构重要十倍。希望帮到你。本文还有配套的精品资源点击获取 
简介本资源为基于DQN解决带插单的柔性作业车间动态调度问题项目Python源码面向计算机、人工智能、自动化、机械电子等专业在校学生及行业从业人员可作为毕业设计、课程设计、期末大作业或比赛初期立项演示的参考方案。压缩包共9个文件约503KB以5个py源码文件为主涵盖主程序入口、DQN算法实现、FJSP对象建模、实例生成器与作业车间调度等核心模块另含3个pyc编译缓存文件及1份luo2020.pdf参考文献便于理解算法背景与实现细节。目前已有167人学习下载。项目将深度强化学习与带插单的动态调度场景结合代表性与创新性较强读者可从中获取完整的算法建模思路、网络结构设计与训练流程并在此基础上进行二次开发或迁移到其他调度问题适合入门进阶与实战借鉴。1. 带插单的柔性作业车间调度为什么传统方法一到动态场景就翻车柔性作业车间调度FJSP本身就是 NP-hard 问题工序可以在多台机器上选择加工解空间比经典 JSP 大了一个量级。而“带插单”意味着生产过程中随时可能有紧急订单插入原有调度方案必须在线调整。传统方法在这里会遇到两个硬伤一是遗传算法、禁忌搜索这类元启发式方法每次重调度都要重新迭代响应时间从秒级跳到分钟级车间等不起二是基于规则的启发式如最短加工时间优先虽然快但缺乏全局优化能力插单后容易把整体交期拖垮。DQNDeep Q-Network之所以被引入这个场景核心逻辑是把调度决策建模为马尔可夫决策过程用神经网络学习“当前车间状态下选哪个工序-机器组合最优”的策略。训练完成后推理阶段只需一次前向传播毫秒级输出决策天然适合动态插单场景。这个项目标题指向的是一套 Python 源码实现适合运筹优化方向的研究生、智能制造方向的工程师以及想用深度强化学习落地车间调度的从业者。它解决的不是“调度理论”问题而是“插单来了怎么在秒级内给出可执行方案”的工程问题。2. 把车间调度翻译成 DQN 能吃的四元组状态、动作、奖励、终止2.1 状态设计用什么特征描述一个车间DQN 的输入是状态向量状态设计直接决定模型能不能学到有效策略。在带插单的 FJSP 中我一般会把状态拆成四个部分工序状态每个待加工工序的剩余工序数、当前工序在工艺路线中的位置、是否已被调度。机器状态每台机器的当前可用时间、正在加工的工序剩余时间、历史利用率。订单状态每个订单的交期紧迫度交期减去当前时间再除以剩余工时、是否为插单、插单优先级。全局状态当前完工时间makespan、平均机器利用率、已插单数量。这些特征拼成一个定长向量。注意工序数和机器数在不同算例中可能不同所以状态向量要么做归一化后补齐到固定长度要么用注意力机制处理变长输入。源码项目里通常采用固定最大工序数的做法简单直接。import numpy as np def build_state(jobs, machines, current_time, max_ops20, max_machines10): 构建 DQN 状态向量 jobs: 订单列表每个订单包含工序列表和交期 machines: 机器列表每台机器有 available_time current_time: 当前调度时刻 state [] # 工序特征剩余工序数、当前工序索引、是否已调度 for job in jobs[:max_ops]: remaining len(job[ops]) - job[current_op_idx] state.extend([ remaining / max_ops, job[current_op_idx] / max_ops, 1.0 if job[scheduled] else 0.0 ]) # 补齐 while len(state) max_ops * 3: state.extend([0.0, 0.0, 0.0]) # 机器特征可用时间归一化、利用率 for m in machines[:max_machines]: state.append(m[available_time] / (current_time 1e-6)) while len(state) max_ops * 3 max_machines: state.append(0.0) # 订单交期紧迫度 for job in jobs[:max_ops]: urgency (job[due_date] - current_time) / (sum(job[proc_times]) 1e-6) state.append(np.clip(urgency, -1, 1)) while len(state) max_ops * 3 max_machines max_ops: state.append(0.0) return np.array(state, dtypenp.float32)这段代码的关键参数是max_ops和max_machines它们决定了状态向量的维度。设太小会截断信息设太大会让网络输入稀疏。我一般按算例中最大工序数的 1.2 倍来设。归一化用current_time 1e-6防止除零交期紧迫度用clip限制在 [-1,1] 避免梯度爆炸。2.2 动作空间选工序还是选机器动作空间有两种常见建模方式方式一联合动作。每个动作是一个(工序, 机器)对。动作数量等于待调度工序数乘以可选机器数。优点是决策直接缺点是动作空间随规模线性增长插单后动作空间会变化。方式二分层动作。先选工序再选机器。两个网络或一个网络两个输出头。优点是动作空间小缺点是训练时信用分配更复杂。源码项目里通常用方式一因为实现简单且插单场景下动作空间变化可以通过掩码处理。具体做法是对当前不可选的动作工序已调度或机器不可用输出一个极大的负值softmax 后概率为 0。def get_action_mask(jobs, machines): 返回动作掩码1 表示可选0 表示不可选 mask [] for job in jobs: if job[scheduled] or job[current_op_idx] len(job[ops]): mask.extend([0] * len(machines)) else: op job[ops][job[current_op_idx]] for m_idx, m in enumerate(machines): # 机器能加工该工序且当前可用 if m_idx in op[eligible_machines]: mask.append(1) else: mask.append(0) return np.array(mask, dtypenp.float32)掩码机制是 DQN 做调度时最容易被忽略的细节。没有掩码网络会学到大量无效动作训练效率极低。掩码在每步调度前重新计算因为机器可用状态和工序状态都在变。2.3 奖励函数让 DQN 学会“插单不拖交期”奖励设计是 DQN 调度最玄学的地方。常见做法有三种稀疏奖励只在完工时给-makespan。缺点是中间步骤没有反馈训练慢。稠密奖励每步给-加工时间或-等待时间。优点是反馈快缺点是可能短视。混合奖励每步给-加工时间完工时额外给-max(0, makespan - due_date)惩罚交期延误。我一般用混合奖励并在插单发生时给一个额外惩罚项让模型学会优先处理插单。具体公式def compute_reward(job, machine, current_time, is_inserted, due_date): proc_time job[ops][job[current_op_idx]][proc_times][machine] reward -proc_time * 0.01 # 加工时间惩罚 if is_inserted: reward - 0.5 # 插单额外惩罚鼓励优先处理 finish_time current_time proc_time if finish_time due_date: reward - (finish_time - due_date) * 0.1 # 交期延误惩罚 return reward参数0.01、0.5、0.1需要根据算例规模调。加工时间惩罚系数太小模型不在乎效率太大模型会忽略交期。插单惩罚系数一般设成加工时间惩罚的 10 到 50 倍具体看插单频率。2.4 终止条件与经验回放终止条件很简单所有工序都被调度完毕。但插单场景下终止条件要加一条如果插单导致原方案完全不可行比如交期已过提前终止并给大负奖励。经验回放池存(state, action, reward, next_state, done)五元组。插单场景下回放池要保证新旧数据混合采样否则模型会遗忘旧策略。我一般设回放池大小 10000 到 50000batch size 64 到 128。3. 用 Python 把 DQN 调度器跑起来网络结构、训练循环与插单注入3.1 网络结构三层全连接够不够调度问题的状态向量维度通常在几百到几千动作空间在几十到几百。三层全连接网络输入层、两个隐藏层、输出层足够。隐藏层神经元数一般设 256 和 128激活函数用 ReLU。import torch import torch.nn as nn class DQN(nn.Module): def __init__(self, state_dim, action_dim, hidden1256, hidden2128): super(DQN, self).__init__() self.net nn.Sequential( nn.Linear(state_dim, hidden1), nn.ReLU(), nn.Linear(hidden1, hidden2), nn.ReLU(), nn.Linear(hidden2, action_dim) ) def forward(self, x): return self.net(x)如果状态向量超过 2000 维建议加一层 BatchNorm 或 LayerNorm否则训练初期 loss 震荡严重。动作空间超过 500 时输出层可以加一个 dueling 结构把 Q 值拆成状态价值和优势函数提升稳定性。3.2 训练循环epsilon 衰减和 target network 更新DQN 训练的核心是两件事epsilon-greedy 探索和 target network 软更新。import random from collections import deque def train_dqn(env, episodes5000, gamma0.95, lr1e-3, epsilon_start1.0, epsilon_end0.05, epsilon_decay0.995, target_update10, batch_size64): state_dim env.state_dim action_dim env.action_dim policy_net DQN(state_dim, action_dim) target_net DQN(state_dim, action_dim) target_net.load_state_dict(policy_net.state_dict()) optimizer torch.optim.Adam(policy_net.parameters(), lrlr) replay_buffer deque(maxlen50000) epsilon epsilon_start for episode in range(episodes): state env.reset() done False while not done: # epsilon-greedy 选择动作 if random.random() epsilon: action env.sample_valid_action() else: with torch.no_grad(): q_values policy_net(torch.FloatTensor(state).unsqueeze(0)) mask torch.FloatTensor(env.get_action_mask()) q_values q_values (1 - mask) * (-1e9) action q_values.argmax().item() next_state, reward, done, info env.step(action) replay_buffer.append((state, action, reward, next_state, done)) state next_state # 经验回放 if len(replay_buffer) batch_size: batch random.sample(replay_buffer, batch_size) states, actions, rewards, next_states, dones zip(*batch) states torch.FloatTensor(states) actions torch.LongTensor(actions).unsqueeze(1) rewards torch.FloatTensor(rewards).unsqueeze(1) next_states torch.FloatTensor(next_states) dones torch.FloatTensor(dones).unsqueeze(1) q_values policy_net(states).gather(1, actions) with torch.no_grad(): next_q target_net(next_states).max(1)[0].unsqueeze(1) target_q rewards gamma * next_q * (1 - dones) loss nn.MSELoss()(q_values, target_q) optimizer.zero_grad() loss.backward() optimizer.step() epsilon max(epsilon_end, epsilon * epsilon_decay) if episode % target_update 0: target_net.load_state_dict(policy_net.state_dict()) return policy_net关键参数说明gamma0.95表示模型关注未来 20 步左右的奖励调度问题一般设 0.9 到 0.99。epsilon_decay0.995表示探索率每轮衰减 0.5%5000 轮后降到 0.05 以下。target_update10表示每 10 轮同步一次 target network太频繁会失去稳定性太慢会滞后。3.3 插单注入怎么模拟动态事件插单注入要在训练和测试时都做。训练时随机在某个调度步插入新订单测试时按固定概率插入。def inject_order(env, prob0.1): 以 prob 概率插入新订单 if random.random() prob: new_job generate_random_job(env.num_machines) env.jobs.append(new_job) env.state_dim env.compute_state_dim() # 状态维度可能变化 return True return False注意插单后状态维度可能变化如果网络输入维度固定需要把新订单特征映射到已有维度。常见做法是预留插单槽位或者用固定最大订单数。源码项目里一般用后者简单但浪费一些维度。3.4 训练效果验证看 makespan 和交期延误率训练过程中要监控两个指标平均 makespan 和交期延误率。前者衡量效率后者衡量插单响应能力。def evaluate(policy_net, env, episodes100): makespans [] tardiness [] for _ in range(episodes): state env.reset() done False while not done: with torch.no_grad(): q_values policy_net(torch.FloatTensor(state).unsqueeze(0)) mask torch.FloatTensor(env.get_action_mask()) q_values q_values (1 - mask) * (-1e9) action q_values.argmax().item() state, _, done, info env.step(action) makespans.append(info[makespan]) tardiness.append(info[tardiness]) return np.mean(makespans), np.mean(tardiness)如果 makespan 在 1000 轮后还在震荡检查奖励尺度是否太大或者学习率是否太高。如果交期延误率居高不下检查插单惩罚系数是否太小。4. 避坑带插单 FJSP 做 DQN 时最容易翻车的五个地方4.1 状态维度不固定导致网络报错现象插单后state_dim变化policy_net输入维度不匹配直接抛异常。原因状态向量按实际订单数动态拼接插单后长度变了。解决固定最大订单数和最大工序数不足补零。或者用nn.Linear之前加一个自适应池化层。我一般用固定维度简单可靠。4.2 动作掩码忘记在 target Q 值上也加现象训练 loss 正常下降但推理时选出的动作大量无效调度器卡死。原因计算 target Q 值时只用了max(1)[0]没有加掩码导致 target 值被无效动作拉高。解决target Q 值计算也要加掩码with torch.no_grad(): next_q_all target_net(next_states) next_mask torch.FloatTensor(np.array([env.get_action_mask() for _ in range(batch_size)])) next_q_all next_q_all (1 - next_mask) * (-1e9) next_q next_q_all.max(1)[0].unsqueeze(1)4.3 奖励尺度太大导致梯度爆炸现象loss 变成 NaN网络权重全是 nan。原因makespan 通常在几百到几千直接用作奖励Q 值会到几千MSE loss 轻松到百万级。解决奖励归一化。加工时间除以最大加工时间makespan 除以基准 makespan。或者用 Huber loss 替代 MSE。4.4 插单频率太高导致模型只学“应付插单”现象插单响应快了但整体 makespan 比规则调度还差。原因插单惩罚系数太大模型过度优先插单牺牲了整体效率。解决插单惩罚系数从 0.1 开始调逐步增加观察 makespan 和延误率的 trade-off。一般插单惩罚系数是加工时间惩罚的 5 到 20 倍比较合理。4.5 回放池被新经验淹没导致灾难性遗忘现象训练后期模型在旧算例上表现突然变差。原因插单产生的新经验不断覆盖回放池旧经验被挤出。解决回放池设大一点50000 以上或者用优先经验回放给旧经验更高采样权重。我一般还会保留一个“旧策略测试集”每 100 轮跑一次掉点就降低学习率。5. 让 DQN 调度器真正可用的三个进阶技巧5.1 用 Double DQN 消除 Q 值高估标准 DQN 的max操作会系统性高估 Q 值在调度问题里表现为模型对某些动作过度自信。Double DQN 用 policy_net 选动作target_net 算 Q 值with torch.no_grad(): next_actions policy_net(next_states).argmax(1, keepdimTrue) next_q target_net(next_states).gather(1, next_actions) target_q rewards gamma * next_q * (1 - dones)改动很小但训练稳定性提升明显。我一般默认用 Double DQN。5.2 用优先经验回放加速收敛调度问题中插单发生前后的经验价值很高但随机采样可能漏掉。优先经验回放按 TD 误差采样TD 误差大的经验被采概率高。class PrioritizedReplayBuffer: def __init__(self, capacity, alpha0.6): self.capacity capacity self.alpha alpha self.buffer [] self.priorities np.zeros(capacity, dtypenp.float32) self.pos 0 def push(self, transition, td_error): self.buffer.append(transition) self.priorities[self.pos] (np.abs(td_error) 1e-6) ** self.alpha self.pos (self.pos 1) % self.capacity def sample(self, batch_size, beta0.4): probs self.priorities[:len(self.buffer)] / self.priorities[:len(self.buffer)].sum() indices np.random.choice(len(self.buffer), batch_size, pprobs) weights (len(self.buffer) * probs[indices]) ** (-beta) weights weights / weights.max() return [self.buffer[i] for i in indices], indices, weightsalpha0.6控制优先级程度beta0.4控制重要性采样权重。这两个参数不用太纠结默认值就能用。5.3 用规则调度做 warm start纯 DQN 从零训练前几千轮基本在随机试错。用规则调度如 SPT生成一批经验先灌进回放池再开始 DQN 训练收敛速度能快一倍。def warm_start(env, policy_net, episodes100): 用 SPT 规则生成经验 for _ in range(episodes): state env.reset() done False while not done: action env.spt_action() # 最短加工时间优先 next_state, reward, done, _ env.step(action) replay_buffer.append((state, action, reward, next_state, done)) state next_statewarm start 之后epsilon 可以从 0.5 开始不用从 1.0 开始省掉大量无效探索。5.4 验证方法和规则调度、遗传算法对比训练完不能只看 loss要跑对比实验。我一般设三组基线SPT 规则、遗传算法迭代 1000 代、随机调度。DQN 在 makespan 上应该比 SPT 好 10% 到 20%比遗传算法差 5% 以内但推理时间快两个数量级。如果 DQN 比 SPT 还差检查奖励设计和掩码如果比遗传算法差太多检查网络容量和训练轮数。最后说个血泪教训我最早做这个方向时花了两个月调网络结构从三层加到七层效果提升不到 3%。后来发现把奖励函数从稀疏改成混合效果直接提升 15%。在调度问题里奖励设计比网络结构重要十倍。希望帮到你。本文还有配套的精品资源点击获取