OR-Tools:4-分配问题（Assignment）

tech2023-09-15 92

OR-Tools 解决的问题类型：

Linear optimizationConstraint optimizationMixed-integer optimizationBin packingNetwork flowsAssignmentSchedulingRouting

分配问题：

假设一组工作人员需要执行一组任务，并且对于每个工作人员和任务，将工作人员分配给该任务需要成本。问题是将每个工作人员最多分配给一个任务，没有两个工作人员执行相同的任务，同时最大限度地降低总成本。

有五个工作工（编号为 0-4）和四个任务（编号为 0-3）

012309080757013585556521259590953451109511545010090100

from ortools.linear_solver import pywraplp def main(): # Data costs = [ [90, 80, 75, 70], [35, 85, 55, 65], [125, 95, 90, 95], [45, 110, 95, 115], [50, 100, 90, 100], ] num_workers = len(costs) num_tasks = len(costs[0]) # Solver # Create the mip solver with the CBC backend. solver = pywraplp.Solver.CreateSolver('assignment_mip', 'CBC') # Variables # x[i, j] is an array of 0-1 variables, which will be 1 # if worker i is assigned to task j. x = {} for i in range(num_workers): for j in range(num_tasks): x[i, j] = solver.IntVar(0, 1, '') # Constraints # Each worker is assigned to at most 1 task. for i in range(num_workers): solver.Add(solver.Sum([x[i, j] for j in range(num_tasks)]) <= 1) # Each task is assigned to exactly one worker. for j in range(num_tasks): solver.Add(solver.Sum([x[i, j] for i in range(num_workers)]) == 1) # Objective objective_terms = [] for i in range(num_workers): for j in range(num_tasks): objective_terms.append(costs[i][j] * x[i, j]) solver.Minimize(solver.Sum(objective_terms)) # Solve status = solver.Solve() # Print solution. if status == pywraplp.Solver.OPTIMAL or status == pywraplp.Solver.FEASIBLE: print('Total cost = ', solver.Objective().Value(), '\n') for i in range(num_workers): for j in range(num_tasks): # Test if x[i,j] is 1 (with tolerance for floating point arithmetic). if x[i, j].solution_value() > 0.5: print('Worker %d assigned to task %d. Cost = %d' % (i, j, costs[i][j])) if __name__ == '__main__': main() Total cost = 265.0 Worker 0 assigned to task 3. Cost = 70 Worker 1 assigned to task 2. Cost = 55 Worker 2 assigned to task 1. Cost = 95 Worker 3 assigned to task 0. Cost = 45

或者使用CP优化的方式

def main(): # Data costs = [ [90, 80, 75, 70], [35, 85, 55, 65], [125, 95, 90, 95], [45, 110, 95, 115], [50, 100, 90, 100], ] num_workers = len(costs) num_tasks = len(costs[0]) # Model model = cp_model.CpModel() # Variables x = [] for i in range(num_workers): t = [] for j in range(num_tasks): t.append(model.NewBoolVar('x[%i,%i]' % (i, j))) x.append(t) # Constraints # Each worker is assigned to at most one task. for i in range(num_workers): model.Add(sum(x[i][j] for j in range(num_tasks)) <= 1) # Each task is assigned to exactly one worker. for j in range(num_tasks): model.Add(sum(x[i][j] for i in range(num_workers)) == 1) # Objective objective_terms = [] for i in range(num_workers): for j in range(num_tasks): objective_terms.append(costs[i][j] * x[i][j]) model.Minimize(sum(objective_terms)) # Solve solver = cp_model.CpSolver() status = solver.Solve(model) # Print solution. if status == cp_model.OPTIMAL or status == cp_model.FEASIBLE: print('Total cost = %i' % solver.ObjectiveValue()) print() for i in range(num_workers): for j in range(num_tasks): if solver.BooleanValue(x[i][j]): print('Worker ', i, ' assigned to task ', j, ' Cost = ', costs[i][j]) else: print('No solution found.')

多人合作的任务：六个工作人员被分成两个团队，每个团队最多可以执行两个任务。

from __future__ import print_function from ortools.linear_solver import pywraplp def main(): solver = pywraplp.Solver('SolveAssignmentProblemMIP', pywraplp.Solver.CBC_MIXED_INTEGER_PROGRAMMING) cost = [[90, 76, 75, 70], [35, 85, 55, 65], [125, 95, 90, 105], [45, 110, 95, 115], [60, 105, 80, 75], [45, 65, 110, 95]] team1 = [0, 2, 4] team2 = [1, 3, 5] team_max = 2 num_workers = len(cost) num_tasks = len(cost[1]) x = {} for i in range(num_workers): for j in range(num_tasks): x[i, j] = solver.BoolVar('x[%i,%i]' % (i, j)) # Objective solver.Minimize(solver.Sum([cost[i][j] * x[i,j] for i in range(num_workers) for j in range(num_tasks)])) # Constraints # Each worker is assigned to at most 1 task. for i in range(num_workers): solver.Add(solver.Sum([x[i, j] for j in range(num_tasks)]) <= 1) # Each task is assigned to exactly one worker. for j in range(num_tasks): solver.Add(solver.Sum([x[i, j] for i in range(num_workers)]) == 1) # Each team takes on two tasks. solver.Add(solver.Sum([x[i, j] for i in team1 for j in range(num_tasks)]) <= team_max) solver.Add(solver.Sum([x[i, j] for i in team2 for j in range(num_tasks)]) <= team_max) sol = solver.Solve() print('Total cost = ', solver.Objective().Value()) print() for i in range(num_workers): for j in range(num_tasks): if x[i, j].solution_value() > 0: print('Worker %d assigned to task %d. Cost = %d' % ( i, j, cost[i][j])) print() print("Time = ", solver.WallTime(), " milliseconds") if __name__ == '__main__': main()

根据任务大小的分配

每个任务都有一个大小，它表示任务需要多少时间或工作量。每个工作人员执行的任务的总大小具有固定的约束。

用CP优化的方式实现：

用线性优化的方式：

from __future__ import print_function import time from ortools.linear_solver import pywraplp def main(): solver = pywraplp.Solver('SolveAssignmentProblem', pywraplp.Solver.CBC_MIXED_INTEGER_PROGRAMMING) start = time.time() cost = [[90, 76, 75, 70, 50, 74, 12, 68], [35, 85, 55, 65, 48, 101, 70, 83], [125, 95, 90, 105, 59, 120, 36, 73], [45, 110, 95, 115, 104, 83, 37, 71], [60, 105, 80, 75, 59, 62, 93, 88], [45, 65, 110, 95, 47, 31, 81, 34], [38, 51, 107, 41, 69, 99, 115, 48], [47, 85, 57, 71, 92, 77, 109, 36], [39, 63, 97, 49, 118, 56, 92, 61], [47, 101, 71, 60, 88, 109, 52, 90]] task_sizes = [10, 7, 3, 12, 15, 4, 11, 5] # Maximum total of task sizes for any worker total_size_max = 15 num_workers = len(cost) num_tasks = len(cost[1]) # Variables x = {} for i in range(num_workers): for j in range(num_tasks): x[i, j] = solver.IntVar(0, 1, 'x[%i,%i]' % (i, j)) # Constraints # The total size of the tasks each worker takes on is at most total_size_max. for i in range(num_workers): solver.Add(solver.Sum([task_sizes[j] * x[i, j] for j in range(num_tasks)]) <= total_size_max) # 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 range(num_workers)]) >= 1) solver.Minimize(solver.Sum([cost[i][j] * x[i,j] for i in range(num_workers) for j in range(num_tasks)])) sol = solver.Solve() print('Minimum cost = ', solver.Objective().Value()) print() for i in range(num_workers): for j in range(num_tasks): if 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") if __name__ == '__main__': main() # cpsat Minimum cost = 326 Worker 0 assigned to task 6 Cost = 12 Worker 1 assigned to task 0 Cost = 35 Worker 1 assigned to task 2 Cost = 55 Worker 4 assigned to task 4 Cost = 59 Worker 5 assigned to task 5 Cost = 31 Worker 5 assigned to task 7 Cost = 34 Worker 6 assigned to task 1 Cost = 51 Worker 8 assigned to task 3 Cost = 49 Time = 11.847 seconds # MIP Minimum cost = 326.0 Worker 0 assigned to task 6 Cost = 12 Worker 1 assigned to task 0 Cost = 35 Worker 1 assigned to task 2 Cost = 55 Worker 4 assigned to task 4 Cost = 59 Worker 5 assigned to task 5 Cost = 31 Worker 5 assigned to task 7 Cost = 34 Worker 6 assigned to task 1 Cost = 51 Worker 8 assigned to task 3 Cost = 49 Time = 0.031 seconds

线性优化耗时短一些

某些允许的工作组分配给任务。在示例中有 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]]

允许的组可以是三对工作人员的任意组合，每组 1、组 2 和组 3 各一对。例如，组合 [2，3]、[6、7] 和 [10，11] 会导致允许的组 [2，3，6，7，10，11]。由于三个组都包含五个元素，因此允许的组总数为 5 * 5 * 5 = 125。

方法1 CPSAT

from __future__ import print_function 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]) 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 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()

方法2 MIP

from __future__ import print_function 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() MIP 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.8036 seconds CP-SAT 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.3831 seconds

CPSAT速度快一些

线性赋值解算器据说比CPSAT,MIP都快

from __future__ import print_function from ortools.graph import pywrapgraph import time def main(): cost = create_data_array() rows = len(cost) cols = len(cost[0]) assignment = pywrapgraph.LinearSumAssignment() for worker in range(rows): for task in range(cols): if cost[worker][task]: assignment.AddArcWithCost(worker, task, cost[worker][task]) solve_status = assignment.Solve() if solve_status == assignment.OPTIMAL: print('Total cost = ', assignment.OptimalCost()) print() for i in range(0, assignment.NumNodes()): print('Worker %d assigned to task %d. Cost = %d' % ( i, assignment.RightMate(i), assignment.AssignmentCost(i))) elif solve_status == assignment.INFEASIBLE: print('No assignment is possible.') elif solve_status == assignment.POSSIBLE_OVERFLOW: print('Some input costs are too large and may cause an integer overflow.') def create_data_array(): cost = [[90, 76, 75, 70], [35, 85, 55, 65], [125, 95, 90, 105], [45, 110, 95, 115]] return cost if __name__ == "__main__": start_time = time.clock() main() print() print("Time =", time.clock() - start_time, "seconds")

我们假设所有工作人员都可以执行所有任务。但情况并非总是如此 - 工作人员可能由于各种原因无法执行一个或多个任务。但是，很容易修改上面的程序来处理这个问题。假设辅助角色 0 无法执行任务 3。

cost = [[90, 76, 75, 'NA'], [35, 85, 55, 65], [125, 95, 90, 105], [45, 110, 95, 115]]

在将成本分配给解算器的代码部分中，添加行，如下所示。if cost[worker][task] != 'NA'

for worker in range(0, rows): for task in range(0, cols): if cost[worker][task] != 'NA': assignment.AddArcWithCost(worker, task, cost[worker][task])

图形理论--婚姻定理Hall's Marriage

婚姻定理应用的标准例子是设想两组;一个男人，和一个女人。对于每个女人，有一个子集的男人，任何一个她很乐意结婚;任何男人都会很乐意嫁给一个想嫁给他的女人考虑是否有可能（在婚姻中）配对男女，使每个人都幸福。

我们有1-N号女嘉宾，和1-N号男嘉宾。每个女嘉宾在属于自己的小卡片上写下中意的男嘉宾们的号码（没有数量限制，甚至可以全写）。把这些小卡片收集起来。

如何通过收集的小卡片来判断，是否存在一种配对，使得所有女嘉宾都选到自己中意的男嘉宾。

例如：N=3时候，女1喜欢男2和男3，可以看出男2收到了三个女士的欢迎

女1：男2，男3

女2：男1，男2

女3：男2

当女1只写男2，女3只写男2的时候就有抢男人的操作因此就配对不成功

假如任取k个女嘉宾，把她们的卡片合起来，这些卡片上写到的所有男嘉宾个数（重复的只算一次）必须至少是k，否则这些女嘉宾们就会抢起来，就不存在配对。意思是男人要比选择人也就是女人的数目多。这样才能挑嘛。

如果严格的阐述Hall's Marriage定理，记为所有女嘉宾集合, 为所有男嘉宾集合, 记为女嘉宾到所选男嘉宾集的映射，这里表示M所有子集的集合。若对任意F的子集S满足以下不等式

则存在配对

关于这个定理的证明请网上自己搜索，我现在还没有时间搞得那么详细。等我过段时间把各种有趣的算法整理好后再细分。然后加上有趣的例子和实现。

关于算法这部分学习我认为是每一个计算机领域的研发人员需要了解的。我会把很多我看到的算法集中到一个专栏《算法优化》里面。到何时方便查找和学习。

说回婚姻定理，这个和分配的关系就是，让左边的节点全都分配到右边节点。也就是都有男士接受，且各个接受的男士是不同的，不然就重婚了，哈哈。

关于婚姻问题这部分还有个The Stable Marriage Problem。留着以后写。

最新回复(0)

OR-Tools:4-分配问题（Assignment）

线性赋值解算器 据说比CPSAT,MIP都快

线性赋值解算器据说比CPSAT,MIP都快