1.2.1

下载 Notebook 下载完整案例

简单装配线平衡问题 (SALBP)¶

类别： 排程

使用 OptAgent 的 Python 接口描述变量、约束与目标。

问题与原始示例来源：Hexaly Code Templates。

问题描述¶

在 Simple Assembly Line Balancing Problem 中，我们考虑一组必须被分配到工作站中的任务。每个任务需要一定的处理时间。每个工作站中所有任务处理时间之和不能超过某个限制，称为节拍时间（cycle time）。任务之间还存在优先级约束。每个任务必须与其所有前驱任务位于同一工作台或更靠后的工作台。最后，目标函数是最小化工作站的数量。

学习要点¶

• 使用 OptAgent 的 set 决策变量建模工作站的内容
• 使用 lambda_function 计算每个工作站的总处理时间
• 使用 find 检索每个任务所在工作站的索引

数据¶

我们提供的 Simple Assembly Line Balancing Problem (SALBP) 实例来自 Otto et al.，格式如下：

• 任务数量
• 节拍时间限制
• 对于每个任务，其索引和处理时间
• 对于每个优先级约束，前驱任务的索引和后继任务的索引

建模思路¶

Simple Assembly Line Balancing Problem (SALBP) 的 OptAgent 模型使用集合变量分配任务。每个 set 表示分配到一个工作站的任务，借助 partition 算子确保每个任务恰好分配到一个工作站。

每个工作站的总处理时间通过 lambda_function 计算，对所有已分配任务应用 sum 算子。此求和中项的数量以及集合大小在搜索过程中会变化，并被约束为不超过节拍时间。

使用 find 算子可以检索每个任务所在工作站的索引，从而写出优先级约束：每个前驱任务所在工作站的索引不得大于其后继任务所在工作站的索引。

已使用的工作站数量等于非空 set 变量的数量，模型以此为最小化目标。

Python 实现¶

from pathlib import Path

from optagent import OptModel, solve


def read_tokens(filename):
    return Path(filename).read_text(encoding="utf-8").split()


def read_instance(instance_file):
    file_it = iter(read_tokens(instance_file))

    for _ in range(3):
        next(file_it)

    # Read number of tasks
    nb_tasks = int(next(file_it))
    max_nb_stations = nb_tasks
    for _ in range(2):
        next(file_it)

    # Read the cycle time limit
    cycle_time = int(next(file_it))
    for _ in range(5):
        next(file_it)

    # Read the processing times
    processing_time_dict = {}
    for _ in range(nb_tasks):
        task = int(next(file_it)) - 1
        processing_time_dict[task] = int(next(file_it))
    for _ in range(2):
        next(file_it)
    processing_time = [elem[1] for elem in sorted(processing_time_dict.items(), key=lambda x: x[0])]

    # Read the precedence relations.
    successors = {}
    for relation in file_it:
        if "," not in relation:
            break
        pred, succ = relation.split(",")
        pred = int(pred) - 1
        succ = int(succ) - 1
        successors.setdefault(pred, []).append(succ)
    return nb_tasks, max_nb_stations, cycle_time, processing_time, successors


def main(input_file, output_file=None, time_limit=20):
    (
        nb_tasks,
        max_nb_stations,
        cycle_time,
        processing_time_data,
        successors_data,
    ) = read_instance(input_file)

    model = OptModel()

    # station_vars[s] is the set of tasks assigned to station s.
    station_vars = [model.set(nb_tasks) for s in range(max_nb_stations)]
    stations = model.array(station_vars)
    model.constraint(model.partition(stations))

    # A station is used when at least one task is assigned to it.
    stations_used = [model.count(station) > 0 for station in station_vars]
    nb_used_stations = model.sum(stations_used)

    # Every station must respect the cycle-time limit.
    processing_time = model.array(processing_time_data)
    # Collection lambda parameters are numeric; // 1 gives the array an integer index.
    time_lambda = model.lambda_function(lambda task: processing_time[task // 1])
    time_in_station = [model.sum(station, time_lambda) for station in station_vars]
    for s, station_time in enumerate(time_in_station):
        model.constraint(
            station_time <= cycle_time,
        )

    # A predecessor must be assigned no later than any of its successors.
    task_station = [model.find(stations, task) for task in range(nb_tasks)]
    for predecessor, successors in successors_data.items():
        for successor in successors:
            model.constraint(
                task_station[predecessor] <= task_station[successor],
            )

    model.minimize(nb_used_stations)
    solution = solve(model, time_limit_s=float(time_limit))
    if not solution.feasible:
        print(f"No feasible assignment found; Status = {solution.status}")
        return solution

    lines = [
        f"Tasks = {nb_tasks}; Cycle time = {cycle_time}; "
        f"Stations used = {nb_used_stations.value}; Status = {solution.status}"
    ]
    for s in range(max_nb_stations):
        if not stations_used[s].value:
            continue
        tasks = " ".join(str(task + 1) for task in sorted(station_vars[s].value))
        lines.append(f"Station {s + 1}: time={time_in_station[s].value}/{cycle_time}; tasks={tasks}")

    result_text = "\n".join(lines)
    print(result_text)
    if output_file is not None:
        output_lines = [str(nb_used_stations.value), str(nb_tasks)]
        output_lines.extend(f"{task + 1},{task_station[task].value + 1}" for task in range(nb_tasks))
        Path(output_file).write_text(
            "\n".join(output_lines) + "\n",
            encoding="utf-8",
        )
    return solution

本地运行¶

Notebook 直接调用 main 并显式传入 .alb 实例路径。以下代码格相互独立，可以按需要单独运行；调整 time_limit 可以控制每个实例的求解时间。

INSTANCE_DIR = Path.cwd() / "instances"
print("Instances:", INSTANCE_DIR)

solution_n20_1 = main(
    INSTANCE_DIR / "instance_n20_1.alb",
    time_limit=1,
)

solution_n20_3 = main(
    INSTANCE_DIR / "instance_n20_3.alb",
    time_limit=1,
)

solution_n20_17 = main(
    INSTANCE_DIR / "instance_n20_17.alb",
    time_limit=1,
)

solution_n20_26 = main(
    INSTANCE_DIR / "instance_n20_26.alb",
    time_limit=1,
)

solution_n20_39 = main(
    INSTANCE_DIR / "instance_n20_39.alb",
    time_limit=1,
)

solution_n20_52 = main(
    INSTANCE_DIR / "instance_n20_52.alb",
    time_limit=1,
)

solution_n20_54 = main(
    INSTANCE_DIR / "instance_n20_54.alb",
    time_limit=1,
)

solution_n20_60 = main(
    INSTANCE_DIR / "instance_n20_60.alb",
    time_limit=1,
)

solution_n20_94 = main(
    INSTANCE_DIR / "instance_n20_94.alb",
    time_limit=1,
)

solution_n20_99 = main(
    INSTANCE_DIR / "instance_n20_99.alb",
    time_limit=1,
)