☰
DQN车间排产实战:Gym环境、动作掩码与训练调优
2026/10/26 18:26:28 网站建设 项目流程

简介:面向车间排产调度研究与工程实践的深度强化学习代码包,围绕DQN在动态生产环境中的策略学习展开,适合具备Python与强化学习基础、希望把调度问题建模为序贯决策的读者。压缩包共3个Python文件、约5KB,按核心算法、车间仿真环境与训练运行入口三个模块组织,分别承载深度Q网络更新、状态动作与奖励设计、训练评估及可视化流程。已有848人学习,说明该主题在工业排产与智能优化方向具有一定关注度。读者可据此理解经验回放、目标网络等DQN关键机制的落地方式,并基于gym式环境接口改写状态、动作和奖励,复现从环境搭建到策略训练的完整链路,为最小化成本、缩短周期或提升资源利用率提供可扩展的实验底座。

1. 车间排产为什么值得用 DQN 试一次

车间排产在工程上更常见的做法是挂一套优先级规则——SPT 先短后长、EDD 先紧后松、MWKR 剩余工时优先,跑一遍仿真看 makespan。规则可解释、上线快,但订单一换、机器一坏、换型时间一加进来,规则之间的组合就开始变得难以预测。这份 Python 代码走的是另一条路:把带工艺路线约束的作业车间封装成标准 Gym 环境,用深度强化学习和 DQN 学“每一步把哪台机器的下一个空档让给哪个工件”。它适合两类人:想拿强化学习练手、又找不到比平衡杆更贴近业务场景的开发者,以及做 APS 排产模块、想验证 AI 调度能不能压过规则调度的工程同学。跑通之后你手里会有一条训练曲线、一份甘特图,以及一个能直接对标的规则基线。

2. 工程骨架与技术选型:从工件数据到 Gym 环境的四层拆分

拿到一份强化学习项目,我会先看它把“环境 / 智能体 / 配置 / 入口”切成了几块。切得清楚,后面改奖励、换数据、加动作掩码才不用满仓库翻文件;切得糊,训练脚本里塞几百行环境代码,改一个 reward 得跑十分钟才能验证一次。这套代码属于前者,结构克制,该有的分层都有。

2.1 目录结构:一个能跑的最小工程长什么样

整个工程压在十来个文件里,没有多余的抽象层,也没套什么框架,就是纯 Python 加几个第三方库。这种体量对二次开发反而友好——你不用理解一套自研框架的设计哲学,打开文件就是逻辑本身。

文件/目录职责
env/shop_env.pyGym 环境主体,reset/step/ 状态构造 / 奖励计算
env/job_data.py工件工艺路线、工时、交期的生成与解析
agent/dqn.pyQ 网络定义、动作选择、损失计算、参数更新
agent/replay_buffer.py环形经验回放缓冲区
agent/mask.py合法动作掩码的生成与广播
configs/default.yaml工件数、机器数、学习率、折扣因子等超参
train.py训练入口,串起环境与智能体
evaluate.py加载权重,用贪心策略跑评估
baselines/rules.pySPT、EDD、FIFO 等规则调度对照
requirements.txt依赖清单

env和agent之间只通过 Gym 的四个接口交互:action_space、observation_space、step()、reset()。这条边界卡住了,你后面把 DQN 换成 Double DQN、Dueling DQN 甚至 PPO,都不需要动环境代码;反过来把车间换成流水车间、并行机,智能体侧也一行不用改。我第一次拆这类项目时没在意这条边界,结果环境里直接 import 了网络的输出维度,后面换模型结构整个工程重写了一遍,血泪经验。

job_data.py拆出来是第二个好习惯。排产数据的形状决定了状态张量的形状,把生成逻辑独立出去,你才能在不碰环境的前提下,塞一份真实车间的工艺路线 CSV 进去做对比实验。

2.2 技术栈拆解:gym、numpy、torch 各管什么

技术选型非常朴素,没有一样是“为了显得高级”加的:

  • gym / gymnasium:只提供接口规范。spaces.Discrete定义动作,spaces.Box定义观测,真正值钱的是它把“环境”这个概念标准化了,你写的step返回结构必须符合约定,逼着你想清楚状态转移和终止条件。
  • numpy:状态构造、掩码计算、指标统计全在 numpy 里做。别小看这一层,排产环境每步都要重算工件剩余工时、机器可用时间,用 Python 原生列表跑几千步就开始肉了。
  • PyTorch:只用来搭一个三层 MLP,加经验回放里的张量搬运。CPU 完全够用,状态维度通常几十到几百,网络参数量在十万级别,用不上 GPU。
  • PyYAML:超参外置。这一条看着不起眼,但排产实验的特点是“同一份代码要跑十几个参数组合”,超参写死在脚本里,你迟早会把某个改过的版本覆盖回去。

为什么不用现成的强化学习库?关键在动作掩码。排产问题里大量动作是非法的——工件已经完工、机器还没空出来、工艺路线规定下一道不在那台机器上。非法动作如果让网络硬选,轻则 episode 提前结束,重则策略永远学不到有效梯度。主流库的策略网络内部不开放“选动作前先屏蔽一批 logits”的钩子,自己写一个两百行的 DQN 反而更省事。这套代码就是这么干的:动作选择、目标 Q 值计算、损失回传全在自己手里,想在哪一步插掩码都行。

2.3 依赖安装与第一条跑通命令

环境用虚拟环境隔离,Python 3.9 以上。torch 装 CPU 版就行,官方源里选对应平台的 CPU wheel,几百 MB,比 CUDA 版省事得多。

# 建环境,隔离依赖,避免污染系统 Python python -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate # 核心依赖:环境接口 + 数值计算 + 网络 + 配置 pip install "gym==0.26.2" numpy torch pyyaml matplotlib

装完之后先别急着训练,跑一次环境的随机策略冒烟测试,确认状态维度、动作数量、episode 长度都对得上:

# 冒烟测试:随机策略跑 3 个 episode,只看形状和终止逻辑 python -c " import gym, numpy as np from env.shop_env import ShopEnv env = ShopEnv(n_jobs=6, n_machines=4) for ep in range(3): obs = env.reset() done, steps = False, 0 while not done: a = env.action_space.sample() obs, r, done, info = env.step(a) steps += 1 print(f'episode {ep}: steps={steps}, makespan={info[\"makespan\"]}') "

reset()返回初始观测,step()返回四元组,info里带当轮的 makespan,这是这套代码的约定。冒烟通过之后再进训练:

# 正式训练,超参全部从 yaml 读 python train.py --config configs/default.yaml # 训练完用贪心策略评估,和规则基线对比 python evaluate.py --ckpt runs/dqn_best.pt --episodes 20

提示:如果你装的是 gymnasium 而不是 gym,step()返回的是五元组(obs, reward, terminated, truncated, info),reset()返回(obs, info)。直接套上面的冒烟脚本会报解包错误。这两个库的 API 不兼容,先确认pip show gym和pip show gymnasium装了哪个,再决定改代码还是改依赖。

3. Gym 环境封装:状态、动作、奖励三件套怎么落地

强化学习项目能不能训出东西,八成取决于环境设计,网络结构反而是次要的。排产场景最容易翻车的地方是状态表达——你把车间拍成什么形状的张量,直接决定了网络能不能从里面读出“哪台机器快空了、哪个工件快拖期了”这类信息。

3.1 状态张量的构造:把排产进度压成固定形状

车间的真实状态是个动态图:工件、工序、机器、时间轴,长度随进度变化。网络只吃定长向量,所以必须做一次压缩。这套代码的做法是“工件维度 + 机器维度”拼接,工件侧每行四个特征,机器侧每台一个特征,最后补一个全局时钟。

def _build_state(self): # 工件侧特征矩阵:(n_jobs, 4) job_feat = np.zeros((self.n_jobs, 4), dtype=np.float32) for j in range(self.n_jobs): remain_ops = self.route_len[j] - self.op_idx[j] # 剩余工序数 remain_time = self.remain_work[j] # 剩余总工时 ready = max(0.0, self.ready_time[j] - self.now) # 到当前时刻还差多久可开工 finished = 1.0 if self.op_idx[j] >= self.route_len[j] else 0.0 job_feat[j] = [remain_ops, remain_time, ready, finished] # 机器侧特征:(n_machines,),距下一次空闲还有多久 machine_feat = np.maximum(0.0, self.machine_free - self.now) # 全局时钟一维,帮助网络建立“整体进度”的概念 clock = np.array([self.now], dtype=np.float32) state = np.concatenate([ job_feat.ravel() / self.norm_scale, # 归一化,统一到 0~1 量级 machine_feat / self.norm_scale, clock / self.norm_scale, ]) return state.astype(np.float32)

四个工件特征各有分工:remain_ops告诉网络这个工件离完工还有几道工序,remain_time是剩余工作量的绝对量,ready是就绪约束(上一道工序还没做完就不能开工),finished是终止信号。机器侧的machine_free - now很关键,它表达的是“这台机器还要忙多久”,没有这一维,网络根本无法判断某台机器当前能不能接活。

norm_scale这个归一化系数别省。原始工时动辄几百上千,直接喂进网络,第一层权重的梯度量级会失控,训练曲线前几百轮基本是平的,看着像玄学,其实是数值问题。我一般取“所有工件工时总和”,让每一维都落在 0~1 之间。

observation_space要跟这个形状严格对上:

state_dim = n_jobs * 4 + n_machines + 1 self.observation_space = spaces.Box( low=0.0, high=1.0, shape=(state_dim,), dtype=np.float32 )

形状写错不会立刻报错,而是训练几十轮之后 loss 突然变 NaN,排查起来很痛苦。改完_build_state之后,第一件事就是打印env.observation_space.shape和env.reset().shape做对拍。

3.2 动作空间与合法动作掩码

动作定义成“工件 × 机器”的笛卡尔积,Discrete(n_jobs * n_machines),解码时divmod(action, n_machines)就能拆回(job, machine)。这个设计的好处是通用:一条工艺路线允许同一道工序在多台并行机上选,动作空间天然覆盖了;坏处是大部分组合是非法的,必须靠掩码压住。

def action_mask(self): """返回长度 n_jobs * n_machines 的 0/1 掩码,1 表示当前可执行""" mask = np.zeros(self.n_jobs * self.n_machines, dtype=np.int8) for j in range(self.n_jobs): if self.op_idx[j] >= self.route_len[j]: continue # 工件已完工,全部非法 if self.ready_time[j] > self.now + self.EPS: continue # 上一道工序没做完 m = self.route[j][self.op_idx[j]] # 工艺路线指定的下一台机器 if self.machine_free[m] <= self.now + self.EPS: mask[j * self.n_machines + m] = 1 # 机器空闲才能排 return mask

掩码的生成逻辑跟工艺路线绑死:route[j][op_idx[j]]取出工件 j 当前该做的那道工序对应的机器,只有这一台合法。如果你的车间允许并行机柔性选择,把这一行换成“遍历可选机器集合”即可,动作空间不用动。

掩码在智能体侧有两次使用,一次是选动作:

def select_action(self, state, mask, epsilon): if np.random.rand() < epsilon: legal = np.flatnonzero(mask) # 探索也只从合法动作里挑 return int(np.random.choice(legal)) with torch.no_grad(): q = self.online(torch.as_tensor(state).unsqueeze(0)) q = q.masked_fill(torch.as_tensor(mask == 0).unsqueeze(0), -1e9) return int(q.argmax(dim=1).item())

另一处是算目标 Q 值时,next_state的掩码也要一起参与,否则 target 会收敛到一个“物理上做不到的最优值”,网络学到的策略在真实环境里执行时全是非法动作。这一点是排产 DQN 区别于 Atari DQN 的核心差异,后面第 5 章还会展开讲。

注意:masked_fill里填的是-1e9而不是-inf,因为-inf一旦参与反向传播会污染梯度,出现整批 loss 为 NaN。

3.3 奖励函数:makespan 增量与拖期惩罚

奖励设计有两个流派:稀疏的“整轮结束给总分”,和稠密的“每步给增量”。这套代码选后者,因为排产一轮动辄几十上百步,稀疏奖励下 DQN 几乎不可能在合理时间内学到东西。

def _calc_reward(self, prev_makespan, job, machine): # 1) makespan 增量:调度让预计完工时间变短就正奖励,变长就负奖励 cur_makespan = self._estimate_makespan() delta = cur_makespan - prev_makespan reward = -delta / self.norm_scale # 2) 拖期惩罚:本次排产导致工件超出交期,额外扣分 finish = self._job_finish_time(job) tardiness = max(0.0, finish - self.due[job]) reward -= self.tardiness_weight * tardiness / self.norm_scale # 3) 换型惩罚:机器上一个工件与本工件类型不同,加一次 setup 成本 if self.last_type[machine] != self.job_type[job]: reward -= self.setup_weight return reward

prev_makespan是执行动作前的预计总完工时间,_estimate_makespan()用“所有机器可用时间的最大值”近似,这一步不需要真跑完所有工序,只要单调可比即可。-delta的符号很关键:makespan 变小是好事,所以取负。

tardiness_weight和setup_weight是两个旋钮。前者调大,策略会偏向“先做紧急订单”,代价是整体 makespan 变长;后者调大,策略倾向于把同类型工件排在一起,减少换型。这两个权重没有标准答案,取决于你的车间更在乎交期还是更在乎产能,我一般从 0.1 开始往上试,每次只动一个。

3.4 reset() 与 step() 的完整骨架

reset()要清干净所有状态,这是排产环境最容易埋雷的地方——任何一个数组没重置,第二条 episode 就会带着上一条的残留数据跑,指标看着虚高,排查起来能折腾一整天。

def reset(self, seed=None, options=None): super().reset(seed=seed) # gym 0.26 的种子机制 self.now = 0.0 self.op_idx = np.zeros(self.n_jobs, dtype=np.int32) # 每个工件做到第几道工序 self.ready_time = np.zeros(self.n_jobs, dtype=np.float32) self.remain_work = self.total_work.copy() # 剩余工时,必须 copy self.machine_free = np.zeros(self.n_machines, dtype=np.float32) self.last_type = -np.ones(self.n_machines, dtype=np.int32) self.done = False return self._build_state()

remain_work用.copy()而不是直接赋值,否则第一轮训练就会把原始工时数组改掉,第二轮开始所有数据都是错的。这个坑我踩过不止一次,后来养成习惯:reset里凡是数组赋值,一律.copy()。

step()的骨架要做三件事:校验动作合法性、推进状态、判断终止。

def step(self, action): mask = self.action_mask() if mask[action] == 0: # 非法动作直接终止并给惩罚,不要静默忽略 return self._build_state(), -1.0, True, False, {"illegal": True} job, machine = divmod(action, self.n_machines) prev_makespan = self._estimate_makespan() # 排产:机器要等工件就绪,工件要等机器空闲,取两者最大值作为开工时间 start = max(self.now, self.machine_free[machine], self.ready_time[job]) proc = self.proc_time[job][self.op_idx[job]] end = start + proc self.machine_free[machine] = end self.ready_time[job] = end self.remain_work[job] -= proc self.last_type[machine] = self.job_type[job] self.op_idx[job] += 1 reward = self._calc_reward(prev_makespan, job, machine) self.done = bool(np.all(self.op_idx >= self.route_len)) info = {"makespan": self._estimate_makespan(), "now": self.now} return self._build_state(), reward, self.done, False, info

start那个max是排产的核心约束:机器空闲时间、工件就绪时间、全局时钟三者取最大,才是一道工序真正的开工时刻。写反了或者漏了一项,仿真出来的甘特图会出现“工件在上一道工序没做完时就上了下一台机器”这种物理上不可能的结果,而训练照样能跑,指标照样漂亮,属于最有欺骗性的 bug。

4. DQN 主体拆解:网络、回放与目标网络的工程细节

环境定好之后,智能体这一侧其实是一套相对固定的模板:Q 网络、经验回放、目标网络、epsilon 探索。差异全在细节里——掩码怎么进损失、目标 Q 怎么算、什么时候同步参数。

4.1 Q 网络结构:为什么是三层 MLP 而不是 CNN

排产的状态是拼接出来的定长向量,没有空间结构,用 CNN 没有意义。三层 MLP 足够:输入维度n_jobs*4 + n_machines + 1,两个 256 的隐藏层,输出维度等于动作数。

class QNet(nn.Module): def __init__(self, state_dim, n_actions, hidden=256): super().__init__() self.net = nn.Sequential( nn.Linear(state_dim, hidden), nn.ReLU(), nn.Linear(hidden, hidden), nn.ReLU(), nn.Linear(hidden, n_actions), # 每个动作一个 Q 值,不加 softmax ) def forward(self, x): return self.net(x)

输出层不加激活函数,Q 值可以是任意实数,这是 DQN 的标准做法。隐藏层用 ReLU 而不是 tanh,收敛快一些,代价是偶尔会出现“某些神经元永久失活”,如果发现训练后期 loss 卡住不动,换成 LeakyReLU(负斜率 0.01)通常能缓解。

hidden=256是经验值。工件数在 10 以内、机器数在 5 以内时,128 也够;工件数上到 50、机器上到 20,状态维度接近 250,256 会显得偏窄,往上调到 512,同时把学习率降一档。

4.2 经验回放与目标网络

回放缓冲区用环形数组实现,固定容量,写满之后覆盖最老的样本。用namedtuple存一条转移,取 batch 的时候一次转成张量,比存 Python 对象再逐条转换快很多。

Transition = namedtuple("Transition", ["s", "a", "r", "ns", "done", "mask", "nmask"]) class ReplayBuffer: def __init__(self, capacity): self.buf = deque(maxlen=capacity) # 满了自动丢弃最老的 def push(self, *args): self.buf.append(Transition(*args)) def sample(self, batch_size): batch = random.sample(self.buf, batch_size) s = torch.as_tensor(np.stack([b.s for b in batch]), dtype=torch.float32) a = torch.as_tensor([b.a for b in batch], dtype=torch.long).unsqueeze(1) r = torch.as_tensor([b.r for b in batch], dtype=torch.float32).unsqueeze(1) ns = torch.as_tensor(np.stack([b.ns for b in batch]), dtype=torch.float32) done = torch.as_tensor([b.done for b in batch], dtype=torch.float32).unsqueeze(1) mask = torch.as_tensor(np.stack([b.mask for b in batch]), dtype=torch.bool) nmask = torch.as_tensor(np.stack([b.nmask for b in batch]), dtype=torch.bool) return s, a, r, ns, done, mask, nmask def __len__(self): return len(self.buf)

注意Transition里多带了两个字段mask和nmask,这是排产场景特有的。普通 DQN 的转移是五元组,这里必须把“当前步哪些动作合法”“下一步哪些动作合法”一起存进去,否则从回放里采样的时候无法复现当时的约束条件。这个改动看着小,但它是整份代码里最容易漏掉的一处——漏了不会报错,只会让策略慢慢退化成“经常输出非法动作”。

目标网络是另一份结构完全相同的网络,参数不参与梯度更新,每隔target_update步从在线网络硬拷贝一次:

def sync_target(self): self.target.load_state_dict(self.online.state_dict())

硬更新比 soft update(θ_target = τθ_online + (1-τ)θ_target)更简单,排产这种状态规模不大的任务里两者差别不明显,硬更新调起来参数更少。target_update一般设 200 到 500,太小目标网络抖动厉害,太大会让 Q 值估计长期滞后。

4.3 损失计算中的掩码处理

损失函数这一段是整套代码最需要小心的地方。标准的 DQN 损失是MSE(Q(s,a), r + γ·max_a' Q_target(s',a')),但在排产里,max必须只对合法动作取,否则网络会朝着一个现实中做不到的目标去拟合。

def compute_loss(self, batch, gamma): s, a, r, ns, done, _, nmask = batch q_sa = self.online(s).gather(1, a) # 当前状态-动作的 Q 值 with torch.no_grad(): # Double DQN:用在线网络选动作,用目标网络估值,缓解过估计 next_q_online = self.online(ns).masked_fill(nmask, -1e9) next_a = next_q_online.argmax(dim=1, keepdim=True) next_q_target = self.target(ns).gather(1, next_a) target = r + gamma * (1.0 - done) * next_q_target loss = F.smooth_l1_loss(q_sa, target) return loss

两个细节。第一,nmask是从回放里取出的“下一步合法动作掩码”,注意它和当前步的mask不是一回事,存的时候要分清楚。第二,用的smooth_l1_loss(也就是 Huber)而不是 MSE,排产奖励里偶尔会出现比较大的负值(比如严重拖期),MSE 会让这些样本的梯度主导整个 batch,Huber 在误差大的区间退化成线性,稳得多。

提示:(1.0 - done)里的done必须是“真终止”,不能把“步数超限被截断”也算进去。截断的 episode 不是真正的终局,把它的next_q归零会让网络误以为任务已完成,价值估计整体偏低。如果你用的环境有截断逻辑,这里要拆成terminated和truncated两个布尔量分别处理。

4.4 训练主循环与 epsilon 衰减

主循环的结构没什么花样,但几个执行顺序不能乱:先采样动作、再执行、再入回放、再从回放取 batch 更新。

for ep in range(total_episodes): s = env.reset() mask = env.action_mask() ep_reward, done = 0.0, False while not done: eps = max(eps_end, eps_start * (eps_decay ** ep)) # 按 episode 衰减 a = agent.select_action(s, mask, eps) ns, r, terminated, truncated, info = env.step(a) # gymnasium 五元组 done = terminated or truncated nmask = env.action_mask() if not done else np.zeros_like(mask) buffer.push(s, a, r, ns, float(terminated), mask, nmask) s, mask = ns, nmask ep_reward += r if len(buffer) >= warmup: # 先攒够样本再开始学 batch = buffer.sample(batch_size) loss = agent.compute_loss(batch, gamma) optimizer.zero_grad() loss.backward() nn.utils.clip_grad_norm_(agent.online.parameters(), 5.0) # 梯度裁剪 optimizer.step() if global_step % target_update == 0: agent.sync_target() global_step += 1 # 每 20 轮评估一次,保存 makespan 最优的权重 if ep % 20 == 0: score = evaluate(agent, env) if score < best_makespan: best_makespan = score torch.save(agent.online.state_dict(), "runs/dqn_best.pt")

warmup这一步别省。回放里只有几十条样本的时候就开始更新,网络会被同一小批数据反复冲刷,Q 值直接发散。我一般设warmup = 500,工件规模大的时候设到 2000。

eps_decay按 episode 衰减而不是按 step 衰减,是因为排产的每一步决策强相关,step 级衰减会让后期的探索严重不足。eps_start=1.0、eps_end=0.05、eps_decay=0.995是这套代码的默认值,工件数上到 20 以上时把eps_decay调到 0.999,不然还没怎么探索就退化成贪心了。

clip_grad_norm_是保命的。排产奖励的方差比游戏大得多,不裁剪的话总有那么几个 batch 会导致梯度爆炸,loss 曲线突然竖起来。

5. 跑起来以后才发现的坑:5 条排产 DQN 常见故障排查

这一章写的都是训练能跑起来、但结果不对的情况。好消息是它们都不难定位,坏消息是如果不提前知道该看哪个指标,很容易在错误的方向上改半天。

5.1 训练 loss 稳定下降,但评估 makespan 不降反升

现象:loss 从 2.0 一路降到 0.3,看着很健康,但每 20 轮评估的 makespan 和随机策略差不多,甚至更差。

原因:损失下降只说明网络拟合当前回放里的 Q 值拟合得好,不代表策略变好。最常见的两个诱因是奖励尺度失衡和状态缺关键维度——如果状态里没有“机器剩余忙时”,网络学到的 Q 值本质上是在猜,把训练集上的噪声记住了。

解决:先关掉训练,用固定种子做一次对照——同一个初始状态,分别用规则调度和训练后的策略各跑一遍,把甘特图打印出来逐行比对,看策略是在哪些决策点上做错的。如果错在“机器明显空闲却不去排”,那就是状态缺维度;如果错在“顺序反了”,那是奖励里 makespan 增量的符号或归一化系数有问题。我遇到过一次是norm_scale用了固定值 1000,而实际总工时是 300 左右,奖励全部被压到 1e-3 量级,Q 值近似常数,loss 当然降得漂亮。

5.2 大量 episode 因 illegal action 提前终止

现象:训练日志里illegal=True的比例超过 10%,episode 长度普遍只有正常值的一半。

原因:探索动作没有走掩码。epsilon分支里直接从action_space.sample()随机采样,采到非法动作的概率是1 - 合法动作数/总动作数,在n_jobs=20, n_machines=10的设置下这个比例高得吓人。

解决:探索也必须从合法动作集合里挑,用np.flatnonzero(mask)取索引再np.random.choice。另外检查select_action里masked_fill的布尔张量方向——mask == 0表示非法,写反了会把合法动作屏蔽掉,症状一模一样。

5.3 第二条 episode 指标突然变好,之后又回落

现象:第一条 episode 的 makespan 是 380,第二条突然掉到 210,第三条又回到 370 左右,来回震荡。

原因:reset()没把状态清干净。最常见的是self.remain_work = self.total_work直接赋值而不是.copy(),第一条 episode 训练时把total_work改掉了,后面所有轮次用的都是被污染的初始数据。

解决:reset()里所有数组赋值一律加.copy(),包括op_idx、ready_time、machine_free、last_type。改完跑三遍同样的随机种子,三条曲线应该完全重合,重合了就说明状态确实清干净了。

5.4 训练集表现很好,换一批工件数据直接崩

现象:在默认生成的 6×4 数据上 makespan 能压到规则调度的 85%,把工件数换成 10、机器数换成 5,策略退化得比 FIFO 还差。

原因:状态维度跟着n_jobs和n_machines变,网络输入层尺寸也变,旧权重根本加载不了;即便重新训练,如果归一化系数norm_scale还是按旧数据算的,状态分布整体偏移,学到的策略自然失效。

解决:换规模时把norm_scale改成从当批数据动态计算(sum(proc_time) / n_jobs),并且重新初始化网络。更进一步的做法是把状态设计成与工件数无关的统计量(比如“平均剩余工时”“最晚完工工件的剩余工时”),这样同一份权重能跨规模复用,代价是信息有损。我在 20 个工件以内试过,损失大概 3% 到 5% 的 makespan,换来的是不用为每个规模单独训一遍。

5.5 评估脚本加载权重后输出全是一样的动作

现象:evaluate.py跑 20 轮,每轮的 makespan 完全一致,甘特图长得一模一样。

原因:评估时epsilon没有置零,或者更隐蔽——load_state_dict之后忘了agent.online.eval()。BatchNorm 和 Dropout 在训练模式下的行为与评估模式不同,这套代码的网络里虽然没有 BN,但如果二次开发时加了,就会踩到这个坑。

解决:评估函数开头固定写agent.online.eval(),epsilon传 0,torch.no_grad()包住推理。另外确认load_state_dict的返回值没有被忽略——它返回_IncompatibleKeys,里面有missing_keys和unexpected_keys,非空就说明权重没对上,但 PyTorch 默认不报错,只是静默地保留随机初始化。

6. 二次开发:把玩具车间改成你自己的产线

这份代码拿来改,最值钱的地方是环境那一层,智能体侧基本可以当模板用。改的方向无非三个:换数据、换奖励、换约束。

换数据最直接。job_data.py里现在是随机生成工艺路线和工时,把它换成读 CSV 就行,字段按job_id, op_seq, machine_id, proc_time, due_date组织,读进来之后填到self.route、self.proc_time、self.due三个数组里。注意route要是二维的(n_jobs, max_ops),工序数不一致的工件用-1补齐,action_mask里遇到-1直接判非法。

换奖励要克制。我见过有人一上来就往奖励里塞七八项——makespan、拖期、换型、机器利用率、在制品数量,结果训练曲线完全不可控,调一个权重另外几个全乱。稳妥的做法是一次只加一项,加完先跑 200 轮看曲线有没有明显劣化,没有再加下一项。换型时间这类硬约束,更合适的做法是放进环境的状态转移里(start = max(..., last_end + setup_time)),而不是放进奖励里让网络去学——约束是确定性的,交给环境处理比交给网络学靠谱得多。

动作空间如果要扩,比如从“选工件”扩到“选工件 + 选机器 + 选优先级”,Discrete就不够用了,得换成MultiDiscrete或者干脆用连续动作配 PPO。这时候环境侧的改动很小,action_mask拆成几段拼接即可,智能体侧要重写,但env和agent的边界还在,重写的范围是可控的。

验证方法我固定走一套流程:先用 5 个不同的随机种子各训一遍,看 makespan 的均值和方差,方差大的说明策略不稳定,大概率是探索衰减太快;再拿 SPT 和 EDD 两个规则基线跑同一批数据,比的是平均值不是最好值;最后挑一轮把甘特图画出来,人工检查有没有“机器空转却不去排”这种明显不合理的空档。这三步走完,一个改动到底有没有用基本就有结论了,比盯着 loss 曲线猜靠谱得多。

从那以后我每次动奖励函数或者状态维度,都会先把随机种子固定住,跑一个只改一处的对照实验,确认差异超出种子本身的波动范围再往下走——排产这类组合优化问题,噪声大到能吞掉 5% 的改进,不控制变量的话,你永远不知道自己是在优化还是在掷骰子。希望帮到你。

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

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询