ortools系列：分配问题中的分组作业
ortools系列：分配问题中的分组作业
1. 分组作业
下面描述一个分配问题，在这个问题中，只允许某些工人组分配任务，在这个例子中有12个工人，编号为0 - 11，允许的组是下列工人对的组合。
group1 = [[2, 3], # Subgroups of workers 0 - 3
[1, 3],
[1, 2],
[0, 1],
[0, 2]]
group2 = [[6, 7], # Subgroups of workers 4 - 7
[5, 7],
[5, 6],
[4, 5],
[4, 7]]
group3 = [[10, 11], # Subgroups of workers 8 - 11
[9, 11],
[9, 10],
[8, 10],
[8, 11]]
允许的组可以是三对工作者的任意组合，每对工作者来自group1、group2和group3。例如，将[2,3]，[6,7]和[10,11]组合在允许的组[2,3,6,7,10,11]中。因为这三个集合都包含5个元素，所以允许的组总数是5 * 5 * 5 = 125。
请注意，如果一组worker属于任何一个允许的组，那么它可以是问题的解。换句话说，可行集由满足任意一个约束条件的点组成。这是一个非凸问题的例子。相比之下，前面描述的MIP示例是一个凸问题：要使一个点可行，必须满足所有约束。
对于像这样的非凸问题，CP-SAT求解器通常比MIP求解器快。下面将介绍使用CP-SAT和MIP的方法，并比较这两种解决方案的解决时间。
2. CP-SAT
# CP-SAT
from ortools.sat.python import cp_model
import time
import numpy as np
def main():
model = cp_model.CpModel()
start = time.time()
# 各个工人做对应任务的成本矩阵
cost = [[90, 76, 75, 70, 50, 74],
[35, 85, 55, 65, 48, 101],
[125, 95, 90, 105, 59, 120],
[45, 110, 95, 115, 104, 83],
[60, 105, 80, 75, 59, 62],
[45, 65, 110, 95, 47, 31],
[38, 51, 107, 41, 69, 99],
[47, 85, 57, 71, 92, 77],
[39, 63, 97, 49, 118, 56],
[47, 101, 71, 60, 88, 109],
[17, 39, 103, 64, 61, 92],
[101, 45, 83, 59, 92, 27]]
num_workers = len(cost)
num_tasks = len(cost[1])
# 要为 CP-SAT 定义允许的工人组，需要创建二进位数组，以指示哪些工人属于一个组。
# 例如，对于group1 (workers 0 - 3)，二进位向量[0,0,1,1]指定包含workers 2和3的组。
# 这种指定组的方式与在MIP解决方案中指定组的方式略有不同，但实际上是一样的。
group1 = [[0, 0, 1, 1], # Workers 2, 3
[0, 1, 0, 1], # Workers 1, 3
[0, 1, 1, 0], # Workers 1, 2
[1, 1, 0, 0], # Workers 0, 1
[1, 0, 1, 0]] # Workers 0, 2
group2 = [[0, 0, 1, 1], # Workers 6, 7
[0, 1, 0, 1], # Workers 5, 7
[0, 1, 1, 0], # Workers 5, 6
[1, 1, 0, 0], # Workers 4, 5
[1, 0, 0, 1]] # Workers 4, 7
group3 = [[0, 0, 1, 1], # Workers 10, 11
[0, 1, 0, 1], # Workers 9, 11
[0, 1, 1, 0], # Workers 9, 10
[1, 0, 1, 0], # Workers 8, 10
[1, 0, 0, 1]] # Workers 8, 11
# Declare the variables.
x = []
for i in range(num_workers):
t = []
for j in range(num_tasks):
t.append(model.NewIntVar(0, 1, "x[%i,%i]" % (i, j)))
x.append(t)
x_array = [x[i][j] for i in range(num_workers) for j in range(num_tasks)]
# Constraints
# Each task is assigned to at least one worker.
[model.Add(sum(x[i][j] for i in range(num_workers)) == 1)
for j in range(num_tasks)]
# Each worker is assigned to at most one task.
[model.Add(sum(x[i][j] for j in range(num_tasks)) <= 1)
for i in range(num_workers)]
# Create variables for each worker, indicating whether they work on some task.
work = []
for i in range(num_workers):
work.append(model.NewIntVar(0, 1, "work[%i]" % i))
for i in range(num_workers):
for j in range(num_tasks):
model.Add(work[i] == sum(x[i][j] for j in range(num_tasks)))
# Define the allowed groups of worders
# 对于CP-SAT，没有必要在一个循环中创建所有这些向量的125个组合。
# CP-SAT解决程序提供了一个方法AllowedAssignments，它能够为三组工作者(0 - 3,4 - 7和8-11)中的每一组分别指定允许的组的约束。
# 变数work[i]是0-1个变数，表示工作状态或每个worker。也就是说，如果工人i被分配到一个任务，work[i]等于0，否则为0。
# 定义工人0-3的工作状态必须与group1中的一个模式匹配的约束。
model.AddAllowedAssignments([work[0], work[1], work[2], work[3]], group1)
model.AddAllowedAssignments([work[4], work[5], work[6], work[7]], group2)
model.AddAllowedAssignments([work[8], work[9], work[10], work[11]], group3)
# 创建目标
model.Minimize(sum([np.dot(x_row, cost_row) for (x_row, cost_row) in zip(x, cost)]))
# 求解问题
solver = cp_model.CpSolver()
status = solver.Solve(model)
if status == cp_model.OPTIMAL:
print(Minimum cost = %i % solver.ObjectiveValue())
print()
for i in range(num_workers):
for j in range(num_tasks):
if solver.Value(x[i][j]) == 1:
print(Worker , i, assigned to task , j, Cost = , cost[i][j])
print()
end = time.time()
print("Time = ", round(end - start, 4), "seconds")
if __name__ == __main__:
main()
# 结果
Minimum cost = 239
Worker 0 assigned to task 4 Cost = 50
Worker 1 assigned to task 2 Cost = 55
Worker 5 assigned to task 5 Cost = 31
Worker 6 assigned to task 3 Cost = 41
Worker 10 assigned to task 0 Cost = 17
Worker 11 assigned to task 1 Cost = 45
Time = 0.2188 seconds
理解了描述的问题，代码就很简单了，接下来我们看MIP方法。
3. MIP
MIP解法生成数据的代码是一样的，唯一不同就是约束条件和目标函数写法不同。
from ortools.linear_solver import pywraplp
import time
def main():
start = time.time()
cost = [[90, 76, 75, 70, 50, 74],
[35, 85, 55, 65, 48, 101],
[125, 95, 90, 105, 59, 120],
[45, 110, 95, 115, 104, 83],
[60, 105, 80, 75, 59, 62],
[45, 65, 110, 95, 47, 31],
[38, 51, 107, 41, 69, 99],
[47, 85, 57, 71, 92, 77],
[39, 63, 97, 49, 118, 56],
[47, 101, 71, 60, 88, 109],
[17, 39, 103, 64, 61, 92],
[101, 45, 83, 59, 92, 27]]
num_tasks = len(cost[1])
# Allowed groups of workers:
group1 = [[2, 3], # Subgroups of workers 0 - 3
[1, 3],
[1, 2],
[0, 1],
[0, 2]]
group2 = [[6, 7], # Subgroups of workers 4 - 7
[5, 7],
[5, 6],
[4, 5],
[4, 7]]
group3 = [[10, 11], # Subgroups of workers 8 - 11
[9, 11],
[9, 10],
[8, 10],
[8, 11]]
allowed_groups = []
for i in range(len(group1)):
for j in range(len(group2)):
for k in range(len(group3)):
allowed_groups.append(group1[i] + group2[j] + group3[k])
min_val = 1e6
total_time = 0
for i in range(len(allowed_groups)):
group = allowed_groups[i]
res = assignment(cost, group)
solver_tmp = res[0]
x_tmp = res[1]
total_time = total_time + solver_tmp.WallTime()
if solver_tmp.Objective().Value() < min_val:
min_val = solver_tmp.Objective().Value()
min_index = i
min_solver = solver_tmp
min_x = x_tmp
min_group = group
print(Minimum cost = , min_val)
print()
for i in min_group:
for j in range(num_tasks):
if min_x[i, j].solution_value() > 0:
print(Worker, i, assigned to task, j, Cost = , cost[i][j])
print()
end = time.time()
print("Time = ", round(end - start, 4), "seconds")
def assignment(cost, group):
# Solve assignment problem for given group of workers.
num_tasks = len(cost[1])
# Clear values in x
solver = None
# Instantiate a mixed-integer solver
solver = pywraplp.Solver(AssignmentProblemGroups,
pywraplp.Solver.CBC_MIXED_INTEGER_PROGRAMMING)
x = None
x = {}
for i in group:
for j in range(num_tasks):
x[i, j] = solver.IntVar(0, 1, x[%i,%i] % (i, j))
# Each worker is assigned to exactly one task.
for i in group:
solver.Add(solver.Sum([x[i, j] for j in range(num_tasks)]) <= 1)
# Each task is assigned to at least one worker.
for j in range(num_tasks):
solver.Add(solver.Sum([x[i, j] for i in group]) >= 1)
# Objective
solver.Minimize(solver.Sum([cost[i][j] * x[i, j] for i in group
for j in range(num_tasks)]))
solver.Solve()
res = [solver, x]
return res
if __name__ == __main__:
main()
# 结果
Minimum cost = 239.0
Worker 0 assigned to task 4 Cost = 50
Worker 1 assigned to task 2 Cost = 55
Worker 5 assigned to task 5 Cost = 31
Worker 6 assigned to task 3 Cost = 41
Worker 10 assigned to task 0 Cost = 17
Worker 11 assigned to task 1 Cost = 45
Time = 0.6276 seconds
4. 参考
- google_ortools_guide
大家看完记得关注点赞.
master苏.
推荐阅读：
