问题描述¶
在 Resource-Constrained Project Scheduling Problem (RCPSP) 中，一个项目由一组需要调度的任务组成。每个任务都有一个给定的持续时间，且不能被中断。任务之间存在优先级约束：每个任务必须在其所有后继任务开始之前结束。问题涉及一组可再生资源。每个任务对每种资源都有一个给定的资源需求或权重（可能为零），表示该任务在执行过程中消耗的资源量。每种资源都有一个给定的最大容量：它可以同时处理多个任务，但被处理任务的权重之和不能超过该最大容量。目标是找到一个使完工时间（makespan）最小的调度方案，即所有任务处理完成的时间。
学习要点¶
- 添加 区间决策变量 来建模任务
- 定义 符号 Lambda 函数 来建模累积资源约束
建模思路¶
Resource-Constrained Project Scheduling Problem (RCPSP) 的 OptAgent 模型使用 区间决策变量 来表示任务。每个 interval 的长度等于相应任务的持续时间。
然后我们写出优先级约束：每个任务必须在其任何后继任务开始之前结束。
累积资源约束可以表述如下：对于每种资源以及每个时间槽 t，正在处理的任务所消耗的资源量不能超过该资源的容量。为了建模这些约束，我们对每种资源和每个时间槽的所有活跃任务的权重进行求和。我们使用可变参数的 and 公式结合 符号 Lambda 函数，以确保资源容量在任何时刻都被满足。这种写法集中表达了各时间段的资源约束；实际求解开销取决于实例规模与求解配置。
需要最小化的完工时间（makespan）是所有任务结束的时间。
Python 实现¶
# Copyright (c) Hexaly. Permission is hereby granted to use, copy,
# and modify this code for applications developed with Hexaly.
from pathlib import Path
from optagent import OptModel, solve
# The input files follow the "Patterson" format
def read_instance(filename):
lines = Path(filename).read_text(encoding="utf-8").splitlines()
first_line = lines[0].split()
# Number of tasks
nb_tasks = int(first_line[0])
# Number of resources
nb_resources = int(first_line[1])
# Maximum capacity of each resource
capacity = [int(lines[1].split()[r]) for r in range(nb_resources)]
# Duration of each task
duration = [0 for i in range(nb_tasks)]
# Resource weight of resource r required for task i
weight = [[] for i in range(nb_tasks)]
# Number of successors
nb_successors = [0 for i in range(nb_tasks)]
# Successors of each task i
successors = [[] for i in range(nb_tasks)]
for i in range(nb_tasks):
line = lines[i + 2].split()
duration[i] = int(line[0])
weight[i] = [int(line[r + 1]) for r in range(nb_resources)]
nb_successors[i] = int(line[nb_resources + 1])
successors[i] = [int(line[nb_resources + 2 + s]) - 1 for s in range(nb_successors[i])]
# Trivial upper bound for the end times of the tasks
horizon = sum(duration[i] for i in range(nb_tasks))
return (nb_tasks, nb_resources, capacity, duration, weight, nb_successors, successors, horizon)
def main(instance_file, output_file=None, time_limit=60):
nb_tasks, nb_resources, capacity, duration, weight, nb_successors, successors, horizon = read_instance(
instance_file)
model = OptModel()
# Interval decisions: time range of each task
tasks = [
model.interval(0, horizon)
for i in range(nb_tasks)
]
# Task duration constraints
for i in range(nb_tasks):
model.constraint(tasks[i].length() == duration[i])
# Precedence constraints between the tasks
for i in range(nb_tasks):
for successor in successors[i]:
model.constraint(tasks[i] < tasks[successor])
# Makespan: end of the last task
makespan = model.max([task.end() for task in tasks])
# Cumulative resource constraints
for r in range(nb_resources):
capacity_respected = model.lambda_function(
lambda t: model.sum(
weight[i][r] * model.contains(tasks[i], t // 1)
for i in range(nb_tasks)
) <= capacity[r]
)
model.constraint(model.and_(model.range(makespan), capacity_respected))
# Minimize the makespan
model.minimize(makespan)
solution = solve(model, time_limit_s=float(time_limit))
if not solution.feasible:
print(f"No feasible schedule found; Status = {solution.feasible}")
return solution
print(f"Makespan = {makespan.value}; Status = {solution.feasible}")
#
# Write the solution in a file with the following format:
# - total makespan
# - for each task, the task id, the start and end times
#
if output_file is not None:
lines = [str(makespan.value)]
lines.extend(
f"{i + 1} {tasks[i].value.start()} {tasks[i].value.end()}"
for i in range(nb_tasks)
)
Path(output_file).write_text("\n".join(lines) + "\n", encoding="utf-8")
return solution
实例调用¶
下面使用仓库提供的 Patterson 格式实例运行模型。
INSTANCE_DIR = Path.cwd() / "instances"
print("Instances:", INSTANCE_DIR)
solution_rcpsp = main(INSTANCE_DIR / "Pat1.rcp", time_limit=1)
solution_rcpsp.feasible