问题描述¶
在批调度问题中,一组任务必须在一组可用资源上以批次形式进行处理。同一批次中的任务会被一起处理,因此它们必须具有相同的时长、相同的类型以及相同的资源类型。资源之间互斥,同一时间只能处理一个批次。每个任务都有给定的资源类型、类型、时长以及若干后续任务。目标是最小化所有任务完成的总体时间(完工时间)。
学习要点¶
- 使用集合决策变量建模批次
- 使用 区间决策变量 建模每个批次的时间范围
- 使用 distinct 算子 与 lambda 函数 确保同一批次内的所有任务具有相同的类型
建模思路¶
批调度问题的 OptAgent 模型使用集合决策变量来建模在每个资源上调度的批次。每个集合中的元素对应于该批次内任务的索引。使用 partition 算子,可以确保一个任务不会被分配到多个批次,且不会遗漏任何任务。
同一批次内的所有任务必须具有相同的类型。使用 distinct 算子与 lambda 函数,我们可以约束同一批次内的所有任务具有相同的类型。
除集合外,我们还使用区间决策变量来建模每个批次的时间范围。为了确保每个资源上的批次互不重叠,我们通过优先约束对这些资源的批次区间进行排序。
借助 find 算子,我们可以检索包含每个任务的批次区间。然后我们可以约束该区间的长度等于任务的时长,并强制其后续任务的开始时间不早于该批次的结束时间。
目标是最小化完工时间(makespan),即所有任务完成的时间。
Python 实现¶
from pathlib import Path
from optagent import OptModel, solve
def read_instance(filename):
# The import files follow the "Taillard" format
with open(filename, encoding="utf-8") as f:
lines = f.readlines()
first_line = lines[0].split()
nb_tasks = int(first_line[0])
nb_resources = int(first_line[1])
second_line = lines[1].split()
capacity = [int(value) for value in second_line]
nb_tasks_per_resource = [0 for _ in range(nb_resources)]
types_in_resource = [[] for _ in range(nb_resources)]
tasks_in_resource = [[] for _ in range(nb_resources)]
task_index_in_resource = []
types, resources, duration, nb_successors = [], [], [], []
successors = [[] for _ in range(nb_tasks)]
for i in range(nb_tasks):
task_line = i + 2
task_information = lines[task_line].split()
types.append(int(task_information[0]))
resources.append(int(task_information[1]))
task_index_in_resource.append(nb_tasks_per_resource[resources[i]])
types_in_resource[resources[i]].append(types[i])
tasks_in_resource[resources[i]].append(i)
nb_tasks_per_resource[resources[i]] += 1
duration.append(int(task_information[2]))
nb_successors.append(int(task_information[3]))
for succeeding_task in task_information[4:]:
successors[i].append(int(succeeding_task))
time_horizon = sum(duration[t] for t in range(nb_tasks))
return (
nb_tasks,
nb_resources,
capacity,
types,
resources,
duration,
nb_successors,
successors,
nb_tasks_per_resource,
task_index_in_resource,
types_in_resource,
tasks_in_resource,
time_horizon,
)
def main(instance_file, output_file=None, time_limit=60):
(
nb_tasks,
nb_resources,
capacity,
types,
resources,
duration,
nb_successors,
successors,
nb_tasks_per_resource,
task_index_in_resource,
types_in_resource,
tasks_in_resource,
time_horizon,
) = read_instance(instance_file)
model = OptModel()
batch_content = [
[
model.set(nb_tasks_per_resource[r])
for b in range(nb_tasks_per_resource[r])
]
for r in range(nb_resources)
]
batch_content_arrays = [model.array(batch_content[r]) for r in range(nb_resources)]
for r in range(nb_resources):
model.constraint(
model.partition(batch_content_arrays[r])
)
types_in_resource_array = model.array(types_in_resource)
def resource_type_lambda(resource):
return model.lambda_function(lambda i: types_in_resource_array[resource][i])
for r in range(nb_resources):
resource_type_mapper = resource_type_lambda(r)
for batch in batch_content[r]:
model.constraint(
model.count(model.distinct(batch, resource_type_mapper)) <= 1,
)
for r in range(nb_resources):
for batch in batch_content[r]:
model.constraint(
model.count(batch) <= capacity[r]
)
batch_interval = [
[model.interval(0, time_horizon) for b in range(nb_tasks_per_resource[r])]
for r in range(nb_resources)
]
batch_interval_arrays = [
model.array(batch_interval[r]) for r in range(nb_resources)
]
for r in range(nb_resources):
for b in range(1, nb_tasks_per_resource[r]):
model.constraint(
batch_interval[r][b - 1] < batch_interval[r][b],
)
task_interval = [None for _ in range(nb_tasks)]
for t in range(nb_tasks):
r = resources[t]
b = model.find(batch_content_arrays[r], task_index_in_resource[t])
task_interval[t] = batch_interval_arrays[r][b]
for t in range(nb_tasks):
model.constraint(
task_interval[t].length() == duration[t]
)
for t in range(nb_tasks):
for s in successors[t]:
model.constraint(
task_interval[t] < task_interval[s]
)
makespan = model.max(
*(
batch_interval_arrays[r][i].end()
for r in range(nb_resources)
for i in range(nb_tasks_per_resource[r])
)
)
model.minimize(makespan)
solution = solve(model, time_limit_s=float(time_limit))
result_values = {'makespan': makespan.value, **{f'task_{t}': task_interval[t].value for t in range(nb_tasks)}}
lines = [
f"Status = {solution.feasible}",
f"Makespan = {result_values['makespan']}",
]
if solution.feasible:
for r in range(nb_resources):
lines.append(f"Resource {r}")
for task in tasks_in_resource[r]:
interval = result_values[f"task_{task}"]
lines.append(f"{task} {interval['start']} {interval['end']}")
result_text = "\n".join(lines)
print(result_text)
if output_file is not None:
Path(output_file).write_text(result_text + "\n", encoding="utf-8")
return solution
本地运行¶
Notebook 直接调用 main 并显式传入实例路径。以下代码格相互独立，可以按需要单独运行；调整 time_limit 可以控制每个实例的求解时间。
from pathlib import Path
INSTANCE_DIR = Path.cwd() / "instances"
print("Instances:", INSTANCE_DIR)
solution_5_2_000 = main(INSTANCE_DIR / "5_2-000.nfb", time_limit=10)