如何在pymoo中定义含动态列表决策变量的并行机调度问题

针对你遇到的变长决策变量无法适配pymoo固定长度数组要求的问题，以下提供两种可行的编码方案，将调度方案转化为固定长度的数值数组，同时完整保留作业分配与执行顺序信息。

方案1：作业全排列+分割点编码 #

核心思路 #

用固定长度数组存储两类信息：

• 前n_jobs位：所有作业的全排列（每个作业编号唯一，代表全局作业顺序）
• 后n_machines-1位：分割点（取值范围1到n_jobs-1），将全排列拆分为n_machines个连续子序列，每个子序列对应一台机器的作业执行顺序

完整代码实现 #

import numpy as np
from pymoo.core.problem import ElementwiseProblem

def get_tardiness(machine_jobs, processing_times, release_dates, due_dates):
    """计算单台机器上各作业的 tardiness 值"""
    tardiness = []
    current_time = 0
    for job in machine_jobs:
        start_time = max(current_time, release_dates[job])
        finish_time = start_time + processing_times[job]
        tard = max(0, finish_time - due_dates[job])
        tardiness.append(tard)
        current_time = finish_time
    return np.array(tardiness)

def get_intime_jobs(machine_jobs, processing_times, release_dates, due_dates):
    """计算单台机器上按时完成的作业标识（1=按时，0=延误）"""
    intime = []
    current_time = 0
    for job in machine_jobs:
        start_time = max(current_time, release_dates[job])
        finish_time = start_time + processing_times[job]
        intime.append(1 if finish_time <= due_dates[job] else 0)
        current_time = finish_time
    return np.array(intime)

class ParallelScheduling(ElementwiseProblem):
    def __init__(self, n_machines, processing_times, release_dates, due_dates, **kwargs):
        self.n_jobs = processing_times.shape[0]
        self.n_machines = n_machines
        self.processing_times = processing_times
        self.release_dates = release_dates
        self.due_dates = due_dates
        
        # 决策变量总数：作业全排列长度 + 分割点数量
        n_var = self.n_jobs + (self.n_machines - 1)
        # 定义变量上下界：全排列部分0~n_jobs-1，分割点部分1~n_jobs-1
        xl = np.concatenate([np.zeros(self.n_jobs, int), np.ones(self.n_machines-1, int)])
        xu = np.concatenate([np.full(self.n_jobs, self.n_jobs-1, int), np.full(self.n_machines-1, self.n_jobs-1, int)])
        
        super().__init__(
            n_var=n_var,
            n_obj=2,
            n_constr=self.n_machines-1,  # 分割点递增约束
            xl=xl,
            xu=xu,
            vtype=int,
            **kwargs
        )

    def _decode(self, X):
        """将固定长度决策变量解码为各机器的作业执行顺序"""
        job_perm = X[:self.n_jobs]
        split_points = X[self.n_jobs:]
        
        # 处理遗传算法可能生成的无序分割点，确保递增且无重复
        split_points = np.sort(np.unique(split_points))
        # 补充分割首尾，确保拆分为n_machines段
        splits = np.concatenate([[0], split_points, [self.n_jobs]])
        
        machine_schedules = []
        for i in range(self.n_machines):
            start, end = splits[i], splits[i+1]
            machine_schedules.append(job_perm[start:end].tolist())
        return machine_schedules

    def _evaluate(self, X, out, *args, **kwargs):
        machine_schedules = self._decode(X)
        
        total_tardiness = 0
        total_intime = 0
        for schedule in machine_schedules:
            if not schedule:
                continue
            total_tardiness += np.sum(get_tardiness(schedule, self.processing_times, self.release_dates, self.due_dates))
            total_intime += np.sum(get_intime_jobs(schedule, self.processing_times, self.release_dates, self.due_dates))
        
        # 目标函数：最小化总延误，最大化按时完成作业数（用负值转换为最小化问题）
        out["F"] = [total_tardiness, -total_intime]
        
        # 添加分割点递增约束：确保split_points[i+1] > split_points[i]
        split_points = X[self.n_jobs:]
        constr = []
        for i in range(self.n_machines-2):
            # 约束值<=0表示满足split[i+1] >= split[i]+1
            constr.append(split_points[i] - split_points[i+1] + 1)
        out["G"] = np.array(constr)

# 示例调用
if __name__ == "__main__":
    n_machines = 3
    processing_times = np.array([3, 2, 4, 1, 5])
    release_dates = np.array([0, 1, 0, 2, 1])
    due_dates = np.array([5, 3, 6, 3, 7])
    
    problem = ParallelScheduling(n_machines, processing_times, release_dates, due_dates)
    
    # 生成随机决策变量并解码输出调度方案
    from pymoo.core.variable import get_random_variable
    X = get_random_variable(problem.vars)
    schedules = problem._decode(X)
    print("各机器作业执行顺序：")
    for idx, sch in enumerate(schedules):
        print(f"机器{idx+1}: {sch}")

方案2：作业分配+优先级编码 #

核心思路 #

用2*n_jobs长度的数组存储：

• 前n_jobs位：每个作业分配的机器编号（0到n_machines-1）
• 后n_jobs位：每个作业的优先级（0到n_jobs-1），同一机器内按优先级升序排列作业（优先级相同则按作业编号排序）

关键解码逻辑 #

def _decode(self, X):
    machine_assign = X[:self.n_jobs]
    priorities = X[self.n_jobs:]
    
    machine_schedules = [[] for _ in range(self.n_machines)]
    # 按机器分组作业
    for job_idx in range(self.n_jobs):
        machine_id = machine_assign[job_idx]
        machine_schedules[machine_id].append((priorities[job_idx], job_idx))
    
    # 每组按优先级升序排序，提取作业编号
    for i in range(self.n_machines):
        machine_schedules[i] = [job for _, job in sorted(machine_schedules[i])]
    return machine_schedules

该方案无需分割点约束，编码更灵活，但需处理优先级重复的情况。

内容的提问来源于stack exchange，提问作者wittn