flow-shop-scheduling

流水车间调度问题（FSP）

你是流水车间调度和装配线优化方面的专家。你的目标是帮助确定作业经过一系列机器的最优顺序，在所有作业按相同路线经过所有机器的前提下，尽量缩短完工时间（最大完工时间）、延迟时间或流程时间。

初步评估

在求解 FSP 实例之前，请先了解：

1. 问题特征

   • 要安排多少个作业？
   • 顺序排列了多少台机器？
   • 是否为置换流水车间（所有机器上的顺序相同）？
   • 是否为无等待流水车间（机器之间不能等待）？
   • 是否为阻塞流水车间（缓冲区有限）？

2. 加工时间

   • 加工时间是否已知且确定？
   • 准备时间是独立的还是与顺序相关？
   • 加工时间是否因机器而异？

3. 目标

   • 是否要最小化最大完工时间（总完工时间）？
   • 是否要最小化总流程时间（完工时间之和）？
   • 是否要最小化总延迟时间？
   • 是否要最小化最大逾期时间？

4. 约束

   • 是否有置换约束（所有机器上的顺序相同）？
   • 是否无等待（在上一台机器加工完成后立即加工）？
   • 缓冲空间是否有限？
   • 是否有交期？

5. 问题规模

   • 小型（< 10 个作业，2–3 台机器）：可使用精确方法
   • 中型（10–50 个作业）：可使用启发式方法
   • 大型（50 个以上作业）：可使用元启发式方法

---

数学建模

置换流水车间调度

参数：

• n：作业数量
• m：机器数量
• p_{ij}：作业 i 在机器 j 上的加工时间
• J = {1, ..., n}：作业集合
• M = {1, ..., m}：机器集合

决策变量：

• π：作业的排列（顺序）
• C_{i,j}：作业 π(i) 在机器 j 上的完工时间

目标函数：

Minimize makespan: C_max = C_{n,m}

或最小化总流程时间：

Minimize: Σ_{i=1}^n C_{i,m}

约束：

1. First machine:
   C_{1,1} = p_{π(1),1}
   C_{i,1} = C_{i-1,1} + p_{π(i),1},  i = 2,...,n

2. First job:
   C_{1,j} = C_{1,j-1} + p_{π(1),j},  j = 2,...,m

3. Other jobs and machines:
   C_{i,j} = max(C_{i-1,j}, C_{i,j-1}) + p_{π(i),j},
             i = 2,...,n; j = 2,...,m

---

精确算法

1. Johnson 算法（双机 FSP）

import numpy as np

def johnsons_algorithm(processing_times):
    """
    Johnson's Algorithm for 2-machine flow shop

    Optimal algorithm for m=2

    Args:
        processing_times: n x 2 array where
                         processing_times[i][0] = time on machine 1
                         processing_times[i][1] = time on machine 2

    Returns:
        optimal sequence and makespan
    """
    n = len(processing_times)
    jobs = list(range(n))

    # Separate into two sets
    set_1 = []  # Jobs where machine 1 time < machine 2 time
    set_2 = []  # Jobs where machine 1 time >= machine 2 time

    for job_id in jobs:
        m1_time = processing_times[job_id][0]
        m2_time = processing_times[job_id][1]

        if m1_time < m2_time:
            set_1.append((job_id, m1_time))
        else:
            set_2.append((job_id, m2_time))

    # Sort set_1 by machine 1 time (ascending)
    set_1.sort(key=lambda x: x[1])

    # Sort set_2 by machine 2 time (descending)
    set_2.sort(key=lambda x: x[1], reverse=True)

    # Combine: set_1 first, then set_2
    sequence = [job_id for job_id, _ in set_1] + [job_id for job_id, _ in set_2]

    # Calculate makespan
    makespan = calculate_makespan_2machine(sequence, processing_times)

    return {
        'sequence': sequence,
        'makespan': makespan,
        'algorithm': 'Johnson'
    }


def calculate_makespan_2machine(sequence, processing_times):
    """Calculate makespan for 2-machine flow shop"""
    n = len(sequence)

    m1_completion = 0
    m2_completion = 0

    for job_id in sequence:
        m1_time = processing_times[job_id][0]
        m2_time = processing_times[job_id][1]

        # Machine 1
        m1_completion += m1_time

        # Machine 2 (must wait for both machine 1 and previous job on machine 2)
        m2_completion = max(m2_completion, m1_completion) + m2_time

    return m2_completion

2. 动态规划（小型实例）

def flowshop_dp_small(processing_times):
    """
    Dynamic programming for small flow shop instances

    Args:
        processing_times: n x m array (n jobs, m machines)

    Returns:
        optimal sequence and makespan
    """
    n, m = processing_times.shape

    # For very small instances only (n <= 12)
    if n > 12:
        raise ValueError("DP only for n <= 12 due to exponential complexity")

    import itertools

    best_sequence = None
    best_makespan = float('inf')

    # Try all permutations
    for sequence in itertools.permutations(range(n)):
        makespan = calculate_makespan(sequence, processing_times)

        if makespan < best_makespan:
            best_makespan = makespan
            best_sequence = sequence

    return {
        'sequence': list(best_sequence),
        'makespan': best_makespan,
        'algorithm': 'Dynamic Programming (enumerate)'
    }


def calculate_makespan(sequence, processing_times):
    """
    Calculate makespan for a given sequence

    Args:
        sequence: job sequence
        processing_times: n x m processing time matrix

    Returns:
        makespan (completion time of last job on last machine)
    """
    n = len(sequence)
    m = processing_times.shape[1]

    # Completion time matrix
    C = np.zeros((n, m))

    # First job
    C[0][0] = processing_times[sequence[0]][0]
    for j in range(1, m):
        C[0][j] = C[0][j-1] + processing_times[sequence[0]][j]

    # First machine
    for i in range(1, n):
        C[i][0] = C[i-1][0] + processing_times[sequence[i]][0]

    # Other positions
    for i in range(1, n):
        for j in range(1, m):
            job = sequence[i]
            C[i][j] = max(C[i-1][j], C[i][j-1]) + processing_times[job][j]

    return C[n-1][m-1]

---

经典启发式方法

1. NEH（Nawaz-Enscore-Ham）启发式

def neh_heuristic(processing_times):
    """
    NEH Heuristic for permutation flow shop

    One of the best constructive heuristics for FSP

    Args:
        processing_times: n x m array

    Returns:
        sequence and makespan
    """
    n, m = processing_times.shape

    # Step 1: Sort jobs by total processing time (descending)
    total_times = processing_times.sum(axis=1)
    sorted_jobs = np.argsort(-total_times)

    # Step 2: Build sequence iteratively
    sequence = [sorted_jobs[0]]

    for k in range(1, n):
        job = sorted_jobs[k]

        # Try inserting job at each position
        best_position = 0
        best_makespan = float('inf')

        for pos in range(len(sequence) + 1):
            # Create temporary sequence
            temp_sequence = sequence[:pos] + [job] + sequence[pos:]

            # Calculate makespan
            makespan = calculate_makespan(temp_sequence, processing_times)

            if makespan < best_makespan:
                best_makespan = makespan
                best_position = pos

        # Insert job at best position
        sequence.insert(best_position, job)

    final_makespan = calculate_makespan(sequence, processing_times)

    return {
        'sequence': sequence,
        'makespan': final_makespan,
        'algorithm': 'NEH'
    }

2. Palmer 启发式

def palmer_heuristic(processing_times):
    """
    Palmer's Heuristic for flow shop

    Simple slope-based heuristic

    Args:
        processing_times: n x m array

    Returns:
        sequence and makespan
    """
    n, m = processing_times.shape

    # Calculate slope index for each job
    slopes = []

    for job_id in range(n):
        slope = 0
        for machine in range(m):
            weight = m - 2*machine - 1
            slope += weight * processing_times[job_id][machine]
        slopes.append((slope, job_id))

    # Sort by slope (descending)
    slopes.sort(reverse=True)
    sequence = [job_id for _, job_id in slopes]

    makespan = calculate_makespan(sequence, processing_times)

    return {
        'sequence': sequence,
        'makespan': makespan,
        'algorithm': 'Palmer'
    }

3. CDS（Campbell-Dudek-Smith）启发式

def cds_heuristic(processing_times):
    """
    CDS Heuristic for flow shop

    Applies Johnson's algorithm m-1 times on aggregated machines

    Args:
        processing_times: n x m array

    Returns:
        best sequence and makespan
    """
    n, m = processing_times.shape

    best_sequence = None
    best_makespan = float('inf')

    # Apply Johnson's algorithm for each aggregation level
    for k in range(1, m):
        # Aggregate first k machines and last k machines
        aggregated = np.zeros((n, 2))

        for job in range(n):
            # First k machines (sum)
            aggregated[job][0] = processing_times[job][:k].sum()

            # Last k machines (sum)
            aggregated[job][1] = processing_times[job][-k:].sum()

        # Apply Johnson's algorithm
        result = johnsons_algorithm(aggregated)
        sequence = result['sequence']

        # Calculate actual makespan with full schedule
        makespan = calculate_makespan(sequence, processing_times)

        if makespan < best_makespan:
            best_makespan = makespan
            best_sequence = sequence

    return {
        'sequence': best_sequence,
        'makespan': best_makespan,
        'algorithm': 'CDS'
    }

---

改进型启发式方法

1. 局部搜索（FSP 的 2-Opt）

def flowshop_local_search(initial_sequence, processing_times, max_iterations=100):
    """
    Local search (2-opt) for flow shop

    Args:
        initial_sequence: initial job sequence
        processing_times: n x m processing time matrix
        max_iterations: maximum iterations

    Returns:
        improved sequence and makespan
    """
    current_sequence = initial_sequence.copy()
    current_makespan = calculate_makespan(current_sequence, processing_times)

    for iteration in range(max_iterations):
        improved = False

        # Try all pairwise swaps
        for i in range(len(current_sequence)):
            for j in range(i + 1, len(current_sequence)):
                # Swap
                new_sequence = current_sequence.copy()
                new_sequence[i], new_sequence[j] = new_sequence[j], new_sequence[i]

                # Evaluate
                new_makespan = calculate_makespan(new_sequence, processing_times)

                if new_makespan < current_makespan:
                    current_sequence = new_sequence
                    current_makespan = new_makespan
                    improved = True
                    break

            if improved:
                break

        if not improved:
            break

    return {
        'sequence': current_sequence,
        'makespan': current_makespan
    }

---

元启发式方法

1. FSP 遗传算法

import random

def flowshop_genetic_algorithm(processing_times, population_size=50,
                               generations=200, mutation_rate=0.1):
    """
    Genetic Algorithm for Flow Shop Scheduling

    Args:
        processing_times: n x m processing time matrix
        population_size: population size
        generations: number of generations
        mutation_rate: mutation probability

    Returns:
        best sequence and makespan
    """
    n = processing_times.shape[0]

    def fitness(sequence):
        makespan = calculate_makespan(sequence, processing_times)
        return 1.0 / (1.0 + makespan)

    def order_crossover(parent1, parent2):
        """Order Crossover (OX)"""
        size = len(parent1)
        start, end = sorted(random.sample(range(size), 2))

        child = [-1] * size
        child[start:end] = parent1[start:end]

        pos = end
        for gene in parent2[end:] + parent2[:end]:
            if gene not in child:
                if pos >= size:
                    pos = 0
                child[pos] = gene
                pos += 1

        return child

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

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

    best_sequence = None
    best_makespan = float('inf')

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

        # Track best
        for ind in population:
            makespan = calculate_makespan(ind, processing_times)
            if makespan < best_makespan:
                best_makespan = makespan
                best_sequence = 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 = order_crossover(parent1, parent2)
            child = mutate(child)

            new_population.append(child)

        population = new_population

    return {
        'sequence': best_sequence,
        'makespan': best_makespan,
        'algorithm': 'Genetic Algorithm'
    }

---

可视化

def visualize_flowshop_schedule(sequence, processing_times, save_path=None):
    """
    Visualize flow shop schedule as Gantt chart

    Args:
        sequence: job sequence
        processing_times: processing time matrix
        save_path: path to save figure
    """
    import matplotlib.pyplot as plt
    import matplotlib.patches as mpatches

    n = len(sequence)
    m = processing_times.shape[1]

    # Calculate completion times
    C = np.zeros((n, m))

    # First job
    C[0][0] = processing_times[sequence[0]][0]
    for j in range(1, m):
        C[0][j] = C[0][j-1] + processing_times[sequence[0]][j]

    # First machine
    for i in range(1, n):
        C[i][0] = C[i-1][0] + processing_times[sequence[i]][0]

    # Other positions
    for i in range(1, n):
        for j in range(1, m):
            job = sequence[i]
            start_time = max(C[i-1][j], C[i][j-1])
            C[i][j] = start_time + processing_times[job][j]

    # Create Gantt chart
    fig, ax = plt.subplots(figsize=(14, 6))

    colors = plt.cm.Set3(np.linspace(0, 1, n))

    for j in range(m):
        for i in range(n):
            job = sequence[i]
            proc_time = processing_times[job][j]

            if i == 0 and j == 0:
                start = 0
            elif j == 0:
                start = C[i-1][j]
            elif i == 0:
                start = C[i][j-1]
            else:
                start = max(C[i-1][j], C[i][j-1])

            # Draw rectangle
            rect = mpatches.Rectangle((start, j - 0.4), proc_time, 0.8,
                                     facecolor=colors[job],
                                     edgecolor='black', linewidth=1)
            ax.add_patch(rect)

            # Add job label
            ax.text(start + proc_time/2, j, f'J{job}',
                   ha='center', va='center', fontweight='bold')

    ax.set_xlabel('Time')
    ax.set_ylabel('Machine')
    ax.set_yticks(range(m))
    ax.set_yticklabels([f'M{i}' for i in range(m)])
    ax.set_xlim(0, C[n-1][m-1] * 1.05)
    ax.set_ylim(-0.5, m - 0.5)
    ax.set_title(f'Flow Shop Schedule (Makespan: {C[n-1][m-1]:.1f})')
    ax.grid(True, axis='x', alpha=0.3)

    plt.tight_layout()

    if save_path:
        plt.savefig(save_path, dpi=300, bbox_inches='tight')

    plt.show()


# Complete example
if __name__ == "__main__":
    np.random.seed(42)
    random.seed(42)

    # Generate random flow shop problem
    n_jobs = 10
    n_machines = 5

    processing_times = np.random.randint(5, 50, size=(n_jobs, n_machines))

    print("Flow Shop Scheduling Problem")
    print(f"Jobs: {n_jobs}, Machines: {n_machines}")
    print("\nProcessing Times:")
    print(processing_times)

    # Compare different algorithms
    print("\n" + "="*60)
    print("Algorithm Comparison:")
    print("="*60)

    algorithms = [
        ('NEH', neh_heuristic),
        ('Palmer', palmer_heuristic),
        ('CDS', cds_heuristic)
    ]

    results = []

    for name, algorithm in algorithms:
        result = algorithm(processing_times)
        results.append((name, result))
        print(f"\n{name}:")
        print(f"  Makespan: {result['makespan']:.1f}")
        print(f"  Sequence: {result['sequence']}")

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

    ga_result = flowshop_genetic_algorithm(processing_times,
                                          population_size=50,
                                          generations=100)
    print(f"\nGA Makespan: {ga_result['makespan']:.1f}")
    print(f"GA Sequence: {ga_result['sequence']}")

    # Find best result
    all_results = results + [('GA', ga_result)]
    best_name, best_result = min(all_results, key=lambda x: x[1]['makespan'])

    print("\n" + "="*60)
    print(f"Best Algorithm: {best_name}")
    print(f"Best Makespan: {best_result['makespan']:.1f}")
    print("="*60)

    # Visualize best schedule
    visualize_flowshop_schedule(best_result['sequence'], processing_times)

    # Calculate machine utilization
    makespan = best_result['makespan']
    total_processing = processing_times.sum()
    utilization = total_processing / (makespan * n_machines) * 100

    print(f"\nMachine Utilization: {utilization:.1f}%")
    print(f"Idle Time: {makespan * n_machines - total_processing:.1f}")

---

工具和库

Python 库

• NumPy：数组操作和计算
• OR-Tools：用于调度的 CP-SAT
• matplotlib：甘特图可视化
• pandas：数据处理

专用软件

• Lekin：教学用调度系统
• CPLEX：MIP 求解器
• Gurobi：MIP 求解器

---

常见挑战与解决方案

挑战：问题规模较大

问题：

• 数百个作业会使精确方法难以实际应用
• 即使启发式方法也可能运行缓慢

解决方案：

• 使用 NEH 启发式（质量很高）
• 对较大实例使用元启发式方法
• 并行评估不同序列

挑战：无等待流水车间

问题：

• 机器之间没有缓冲区
• 作业必须立即进入下一台机器

解决方案：

• 修改完工时间计算方法
• 使用专用的 NEH 变体
• 在元启发式方法中加入无等待约束

挑战：与顺序相关的准备时间

问题：

• 准备时间取决于前一个作业
• 问题复杂度显著增加

解决方案：

• 修改最大完工时间计算，将准备时间纳入其中
• 使用旅行商问题建模
• 使用高级元启发式方法

---

输出格式

Flow Shop 解决方案报告

问题：

• 作业：20
• 机器：6（线性顺序）
• 目标：最小化完工时间

解决方案：

指标	值
完工时间	487 分钟
机器利用率	82%
总流动时间	8,945 分钟
平均流动时间	447 分钟

顺序：

J5 → J12 → J3 → J18 → J7 → J15 → J2 → J10 → ...

甘特图：

M0: J5[0-23] J12[23-48] J3[48-71] ...
M1: [idle] J5[23-45] J12[48-72] ...
M2: [idle] J5[45-63] J12[72-94] ...
...

机器统计数据：

机器	加工时间	空闲时间	利用率
M0	423	64	87%
M1	401	86	82%
M2	389	98	80%
[...]

---

要提出的问题

1. 有多少个作业和机器？
2. 这是置换流水车间（所有机器上的顺序相同）吗？
3. 目标是什么？（完工时间、流动时间、拖期）
4. 机器之间有缓冲区吗？
5. 是否有无等待约束？
6. 换型时间（与顺序有关还是无关）？
7. 是否有交期？
8. 可以使用元启发式算法，还是需要保证最优？

---

相关技能

• job-shop-scheduling：用于柔性路径规划
• production-scheduling：用于更广泛的制造排程
• assembly-line-balancing：用于生产线设计
• master-production-scheduling：用于计划制定
• optimization-modeling：用于数学建模