job-shop-scheduling

作业车间调度问题（JSP）

你是作业车间调度与生产排序优化领域的专家。你的目标是帮助确定机器上操作的最优顺序，以尽量缩短完工时间（最大完工时间）、延误时间或其他目标，同时满足优先约束和机器可用性要求。

初步评估

在解决 JSP 实例之前，请先了解：

1. 问题特征

   • 需要安排多少个作业？
   • 有多少台可用机器？
   • 固定工艺路线（经典 JSP）还是柔性工艺路线（FJSP）？
   • 是否允许循环加工？（作业可以重复访问同一台机器）

2. 操作详情

   • 每个作业包含多少道操作？
   • 加工时间是否已知且确定？
   • 每个作业内是否存在优先约束？
   • 操作之间是否需要设置时间？

3. 目标

   • 尽量缩短最大完工时间（总完工时间）？
   • 尽量减少总延误时间？
   • 尽量减少最大逾期时间？
   • 采用加权组合？

4. 约束

   • 是否无等待（某道操作必须在前一道操作结束后立即开始）？
   • 机器之间的缓冲区是否有限？
   • 是否存在机器故障或维护时段？
   • 作业是否有交期？

5. 问题规模

   • 小型（< 10 个作业，< 10 台机器）：可采用精确方法
   • 中型（10-20 个作业）：采用高级算法
   • 大型（20+ 个作业）：需要元启发式算法

---

数学建模

作业车间调度（析取图模型）

集合：

• J = {1, ..., n}：作业
• M = {1, ..., m}：机器
• O_j：作业 j 的操作集合

参数：

• p_{ij}：作业 j 的操作 i 的加工时间
• μ(i,j)：作业 j 的操作 i 所需的机器
• Prec_j：作业 j 的优先约束

决策变量：

• C_{ij}：作业 j 的操作 i 的完工时间
• S_{ij}：作业 j 的操作 i 的开始时间
• y_{ij,kl} ∈ {0,1}：若操作 (i,j) 在同一台机器上先于操作 (k,l) 加工，则为 1

目标函数：

Minimize makespan: C_max = max_{j∈J, i∈O_j} C_{ij}

或最小化总延误时间：

Minimize: Σ_{j∈J} max(0, C_{last(j)} - d_j)

约束：

1. Completion time relationship:
   C_{ij} = S_{ij} + p_{ij},  ∀j ∈ J, ∀i ∈ O_j

2. Precedence within jobs:
   C_{i,j} ≤ S_{i+1,j},  ∀j ∈ J, ∀i ∈ O_j\{last}

3. Disjunctive constraints (no machine overlap):
   For operations (i,j) and (k,l) on same machine μ:

   Either: S_{ij} + p_{ij} ≤ S_{kl}  (i,j before k,l)
   Or:     S_{kl} + p_{kl} ≤ S_{ij}  (k,l before i,j)

   Linearized as:
   S_{ij} + p_{ij} ≤ S_{kl} + M*(1 - y_{ij,kl})
   S_{kl} + p_{kl} ≤ S_{ij} + M*y_{ij,kl}

4. Non-negativity:
   S_{ij}, C_{ij} ≥ 0

5. Binary variables:
   y_{ij,kl} ∈ {0,1}

---

精确算法

1. JSP 分支定界法

import numpy as np
from collections import defaultdict

class JobShopProblem:
    """Job Shop Scheduling Problem representation"""

    def __init__(self, jobs, machines):
        """
        Args:
            jobs: list of jobs, each job is list of (machine, time) tuples
                  Example: [[(0, 3), (1, 2), (2, 2)],  # Job 0
                           [(0, 2), (2, 1), (1, 4)]]   # Job 1
            machines: number of machines
        """
        self.jobs = jobs
        self.n_jobs = len(jobs)
        self.n_machines = machines
        self.n_operations = sum(len(job) for job in jobs)

    def calculate_lower_bound(self, partial_schedule):
        """
        Calculate lower bound on makespan

        Uses machine-based and job-based bounds
        """
        # Machine lower bound: max completion time for each machine
        machine_times = [0] * self.n_machines

        for job_id, job in enumerate(self.jobs):
            for op_idx, (machine, time) in enumerate(job):
                if (job_id, op_idx) in partial_schedule:
                    continue
                machine_times[machine] += time

        machine_lb = max(machine_times)

        # Job lower bound: max total time for each job
        job_times = []
        for job_id, job in enumerate(self.jobs):
            total_time = sum(time for _, time in job)
            job_times.append(total_time)

        job_lb = max(job_times)

        return max(machine_lb, job_lb)


def jsp_branch_and_bound_simple(jobs, machines, time_limit=60):
    """
    Simple branch-and-bound for small JSP instances

    Args:
        jobs: list of jobs (each job is list of (machine, processing_time))
        machines: number of machines
        time_limit: time limit in seconds

    Returns:
        best schedule found
    """
    import time

    problem = JobShopProblem(jobs, machines)

    best_makespan = float('inf')
    best_schedule = None

    # Current partial schedule
    schedule = []  # List of (job_id, operation_idx, start_time, machine)
    machine_available = [0] * machines  # Next available time for each machine
    job_next_op = [0] * problem.n_jobs  # Next operation index for each job
    job_available = [0] * problem.n_jobs  # Next available time for each job

    start_time = time.time()

    def backtrack(depth):
        nonlocal best_makespan, best_schedule

        if time.time() - start_time > time_limit:
            return

        # All operations scheduled
        if depth == problem.n_operations:
            makespan = max(machine_available)
            if makespan < best_makespan:
                best_makespan = makespan
                best_schedule = schedule.copy()
            return

        # Pruning: check lower bound
        # (simplified - just check current makespan)
        if max(machine_available) >= best_makespan:
            return

        # Try scheduling next operation for each job
        for job_id in range(problem.n_jobs):
            op_idx = job_next_op[job_id]

            # Check if this job has more operations
            if op_idx >= len(jobs[job_id]):
                continue

            machine, proc_time = jobs[job_id][op_idx]

            # Calculate start time
            start = max(job_available[job_id], machine_available[machine])

            # Schedule this operation
            schedule.append((job_id, op_idx, start, machine))
            old_machine_avail = machine_available[machine]
            old_job_avail = job_available[job_id]

            machine_available[machine] = start + proc_time
            job_available[job_id] = start + proc_time
            job_next_op[job_id] += 1

            # Recurse
            backtrack(depth + 1)

            # Undo
            schedule.pop()
            machine_available[machine] = old_machine_avail
            job_available[job_id] = old_job_avail
            job_next_op[job_id] -= 1

    backtrack(0)

    return {
        'makespan': best_makespan,
        'schedule': best_schedule
    }

---

经典启发式算法

1. 优先级调度规则

def jsp_dispatch_rule(jobs, machines, rule='SPT'):
    """
    Dispatch rule heuristic for JSP

    Priority rules:
    - SPT: Shortest Processing Time
    - LPT: Longest Processing Time
    - FCFS: First Come First Served
    - EDD: Earliest Due Date
    - MWR: Most Work Remaining

    Args:
        jobs: list of jobs (each job is list of (machine, time))
        machines: number of machines
        rule: dispatch rule to use

    Returns:
        schedule and makespan
    """
    n_jobs = len(jobs)

    # Track state
    machine_available = [0] * machines
    job_available = [0] * n_jobs
    job_next_op = [0] * n_jobs
    schedule = []

    # Total operations
    n_ops = sum(len(job) for job in jobs)

    for _ in range(n_ops):
        # Find all eligible operations (next operation for each job)
        eligible = []

        for job_id in range(n_jobs):
            op_idx = job_next_op[job_id]

            if op_idx < len(jobs[job_id]):
                machine, proc_time = jobs[job_id][op_idx]

                # Calculate earliest start time
                earliest_start = max(job_available[job_id],
                                   machine_available[machine])

                eligible.append({
                    'job_id': job_id,
                    'op_idx': op_idx,
                    'machine': machine,
                    'time': proc_time,
                    'start': earliest_start
                })

        if not eligible:
            break

        # Select operation based on rule
        if rule == 'SPT':
            selected = min(eligible, key=lambda x: x['time'])
        elif rule == 'LPT':
            selected = max(eligible, key=lambda x: x['time'])
        elif rule == 'FCFS':
            selected = min(eligible, key=lambda x: x['start'])
        elif rule == 'MWR':
            # Most work remaining for this job
            def remaining_work(op):
                job_id = op['job_id']
                op_idx = op['op_idx']
                remaining = sum(t for _, t in jobs[job_id][op_idx:])
                return remaining

            selected = max(eligible, key=remaining_work)
        else:
            selected = eligible[0]

        # Schedule selected operation
        job_id = selected['job_id']
        machine = selected['machine']
        start_time = selected['start']
        proc_time = selected['time']

        schedule.append({
            'job': job_id,
            'operation': selected['op_idx'],
            'machine': machine,
            'start': start_time,
            'end': start_time + proc_time
        })

        machine_available[machine] = start_time + proc_time
        job_available[job_id] = start_time + proc_time
        job_next_op[job_id] += 1

    makespan = max(machine_available)

    return {
        'schedule': schedule,
        'makespan': makespan,
        'rule': rule
    }

2. 移动瓶颈启发式算法

def shifting_bottleneck_jsp(jobs, machines):
    """
    Shifting Bottleneck Heuristic for JSP

    One of the best constructive heuristics for JSP

    Args:
        jobs: list of jobs
        machines: number of machines

    Returns:
        schedule and makespan
    """
    n_jobs = len(jobs)

    # Initialize
    scheduled_machines = set()
    machine_schedules = {m: [] for m in range(machines)}

    while len(scheduled_machines) < machines:
        # Find bottleneck machine
        max_delay = 0
        bottleneck = None

        for machine in range(machines):
            if machine in scheduled_machines:
                continue

            # Calculate delay caused by this machine
            # (simplified: sum of processing times)
            delay = sum(
                proc_time for job in jobs
                for m, proc_time in job
                if m == machine
            )

            if delay > max_delay:
                max_delay = delay
                bottleneck = machine

        if bottleneck is None:
            break

        # Schedule operations on bottleneck machine
        # using one-machine scheduling (EDD or other rule)
        ops_on_machine = []

        for job_id, job in enumerate(jobs):
            for op_idx, (m, proc_time) in enumerate(job):
                if m == bottleneck:
                    ops_on_machine.append((job_id, op_idx, proc_time))

        # Sort by processing time (SPT)
        ops_on_machine.sort(key=lambda x: x[2])

        # Schedule these operations
        current_time = 0
        for job_id, op_idx, proc_time in ops_on_machine:
            machine_schedules[bottleneck].append({
                'job': job_id,
                'operation': op_idx,
                'start': current_time,
                'end': current_time + proc_time
            })
            current_time += proc_time

        scheduled_machines.add(bottleneck)

    # Combine schedules and calculate makespan
    all_schedule = []
    for machine, ops in machine_schedules.items():
        all_schedule.extend(ops)

    makespan = max(op['end'] for ops in machine_schedules.values()
                  for op in ops) if all_schedule else 0

    return {
        'schedule': all_schedule,
        'makespan': makespan
    }

---

元启发式算法

1. JSP 遗传算法

import random

def jsp_genetic_algorithm(jobs, machines, population_size=50,
                         generations=200, mutation_rate=0.1):
    """
    Genetic Algorithm for JSP

    Chromosome representation: operation-based encoding

    Args:
        jobs: list of jobs
        machines: number of machines
        population_size: GA population size
        generations: number of generations
        mutation_rate: mutation probability

    Returns:
        best schedule found
    """
    n_jobs = len(jobs)

    # Create operation list (job_id repeated by number of operations)
    operations = []
    for job_id, job in enumerate(jobs):
        operations.extend([job_id] * len(job))

    def decode_chromosome(chromosome):
        """
        Decode chromosome to schedule

        Chromosome is a permutation of operations
        """
        job_next_op = [0] * n_jobs
        machine_available = [0] * machines
        job_available = [0] * n_jobs
        schedule = []

        for job_id in chromosome:
            op_idx = job_next_op[job_id]

            if op_idx >= len(jobs[job_id]):
                continue

            machine, proc_time = jobs[job_id][op_idx]

            start = max(job_available[job_id], machine_available[machine])

            schedule.append({
                'job': job_id,
                'operation': op_idx,
                'machine': machine,
                'start': start,
                'end': start + proc_time
            })

            machine_available[machine] = start + proc_time
            job_available[job_id] = start + proc_time
            job_next_op[job_id] += 1

        makespan = max(machine_available)
        return schedule, makespan

    def fitness(chromosome):
        """Fitness = 1 / makespan"""
        _, makespan = decode_chromosome(chromosome)
        return 1.0 / (1.0 + makespan)

    def precedence_crossover(parent1, parent2):
        """Precedence Preserving Crossover (PPX)"""
        child = []
        remaining1 = parent1.copy()
        remaining2 = parent2.copy()

        while remaining1 or remaining2:
            if random.random() < 0.5 and remaining1:
                job = remaining1.pop(0)
                child.append(job)
                if job in remaining2:
                    remaining2.remove(job)
            elif remaining2:
                job = remaining2.pop(0)
                child.append(job)
                if job in remaining1:
                    remaining1.remove(job)

        return child

    def mutate(chromosome):
        """Swap mutation"""
        if random.random() < mutation_rate:
            i, j = random.sample(range(len(chromosome)), 2)
            chromosome[i], chromosome[j] = chromosome[j], chromosome[i]
        return chromosome

    # Initialize population
    population = []
    for _ in range(population_size):
        individual = operations.copy()
        random.shuffle(individual)
        population.append(individual)

    best_chromosome = None
    best_makespan = float('inf')

    for generation in range(generations):
        # Evaluate fitness
        fitnesses = [fitness(ind) for ind in population]

        # Track best
        for ind, fit in zip(population, fitnesses):
            _, makespan = decode_chromosome(ind)
            if makespan < best_makespan:
                best_makespan = makespan
                best_chromosome = ind.copy()

        # Selection and reproduction
        new_population = []

        # Elitism
        elite_count = int(0.1 * population_size)
        elite_indices = sorted(range(len(fitnesses)),
                              key=lambda i: fitnesses[i],
                              reverse=True)[:elite_count]
        new_population = [population[i].copy() for i in elite_indices]

        # Create offspring
        while len(new_population) < population_size:
            # Tournament selection
            parent1 = max(random.sample(list(zip(population, fitnesses)), 3),
                         key=lambda x: x[1])[0]
            parent2 = max(random.sample(list(zip(population, fitnesses)), 3),
                         key=lambda x: x[1])[0]

            child = precedence_crossover(parent1, parent2)
            child = mutate(child)

            new_population.append(child)

        population = new_population

    best_schedule, _ = decode_chromosome(best_chromosome)

    return {
        'schedule': best_schedule,
        'makespan': best_makespan,
        'chromosome': best_chromosome
    }


# Example usage
if __name__ == "__main__":
    # Classic 3x3 JSP instance (FT03)
    jobs = [
        [(0, 1), (1, 3), (2, 6)],  # Job 0
        [(0, 8), (2, 5), (1, 10)], # Job 1
        [(1, 5), (0, 4), (2, 8)]   # Job 2
    ]

    machines = 3

    print("Job Shop Scheduling Problem")
    print(f"Jobs: {len(jobs)}, Machines: {machines}")
    print("\nJob routing and times:")
    for i, job in enumerate(jobs):
        print(f"  Job {i}: {job}")

    # Try different methods
    print("\n" + "="*60)
    print("Dispatch Rules:")
    print("="*60)

    for rule in ['SPT', 'LPT', 'FCFS', 'MWR']:
        result = jsp_dispatch_rule(jobs, machines, rule)
        print(f"\n{rule}: Makespan = {result['makespan']}")

    print("\n" + "="*60)
    print("Genetic Algorithm:")
    print("="*60)

    ga_result = jsp_genetic_algorithm(jobs, machines,
                                     population_size=50,
                                     generations=100)
    print(f"\nGA: Makespan = {ga_result['makespan']}")

    print("\nBest schedule:")
    for op in sorted(ga_result['schedule'], key=lambda x: x['start']):
        print(f"  Job {op['job']} Op {op['operation']} on Machine {op['machine']}: "
              f"[{op['start']}, {op['end']}]")

    # Visualize Gantt chart
    print("\nGantt Chart:")
    print("-" * 60)

    for machine in range(machines):
        print(f"Machine {machine}: ", end="")
        ops_on_machine = [op for op in ga_result['schedule']
                         if op['machine'] == machine]
        ops_on_machine.sort(key=lambda x: x['start'])

        for op in ops_on_machine:
            print(f"J{op['job']}[{op['start']}-{op['end']}] ", end="")
        print()

---

工具与库

Python 库

• OR-Tools (Google)：用于 JSP 的 CP-SAT 求解器
• Pyomo：MIP/CP 建模
• simpy：离散事件仿真
• matplotlib：甘特图可视化

专用软件

• CPLEX CP Optimizer：约束规划
• Gurobi：MIP 求解器
• OptaPlanner：基于 Java 的调度软件

---

常见挑战与解决方案

挑战：搜索空间庞大

问题：

• n 个作业 × m 台机器会产生庞大的解空间
• 复杂度呈指数级（NP-hard）

解决方案：

• 使用元启发式算法（GA、禁忌搜索）
• 使用调度规则生成较好的初始解
• 采用分解方法

挑战：柔性作业车间（FJSP）

问题：

• 操作可以在多台机器中的任意一台上执行
• 比经典 JSP 更复杂

解决方案：

• 两级优化：机器分配 + 排序
• 分层 GA
• OR-Tools 可直接处理

挑战：动态到达

问题：

• 执行期间会有新作业到达
• 需要重新调度

解决方案：

• 滚动时域方法
• 右移式重新调度
• 通过缓冲区生成鲁棒调度方案

---

输出格式

JSP 解决方案报告

问题：

• 作业：10
• 机器：5
• 操作总数：47
• 目标：尽量缩短最大完工时间

解决方案：

指标	值
最大完工时间	243 分钟
机器利用率	平均 78%
空闲时间	总计 187 分钟

甘特图：

Machine 0: J2[0-15] J5[15-32] J1[35-48] ...
Machine 1: J1[0-22] J3[22-40] J7[45-67] ...
Machine 2: J4[0-18] J2[20-35] J6[35-52] ...
...

---

需要询问的问题

1. 有多少个作业和机器？
2. 路由是固定还是灵活的（FJSP）？
3. 目标是什么？（最大完工时间、延迟时间、流程时间）
4. 有截止日期吗？
5. 工序之间有准备时间吗？
6. 工序可以中断吗（抢占）？
7. 机器是否始终可用？
8. 这是静态还是动态场景（是否会有新作业到达）？

---

相关技能

• flow-shop-scheduling：适用于线性路由
• production-scheduling：适用于更广泛的制造场景
• master-production-scheduling：用于集成规划
• constraint-programming：用于 CP 方法
• optimization-modeling：用于 MIP 建模