[Python] Python ortools MILP solution for group optimization - code too clunky?

Discussão em 'Python' iniciado por Stack, Março 18, 2025.

  1. Stack

    Stack Membro Participativo

    I have a group of people that I need to split up into groups. The groups do not all have to be the same size, but can be no smaller than 5 and no larger than 10. I have created a matrix (in excel) where each possible pairing of two individuals is assigned an affinity score from 0 to 5. Individuals that have a score of 5 must be in the same group (such as for spouses/partners), while individuals that have a score of 0 cannot be in the same group under any circumstance. I need an optimized configuration of groupings.

    Below is the most recent iteration of code ChatGPT produced, which so far has failed to produce any solutions (and in fact does not even provide the requested status updates every 120 seconds). Previous iterations had moved very slowly, and when I ctrl+c'd the terminal the solution insisted on making groups of only one size (19 groups 5) when it was a set of 95 people. I have since reduced the people to 93 and changed the code, so I am not sure if the code is unable to handle a number that doesn't divide into an equal number of groups, or if there are other issues at play. I'm also wondering if PuLP would be better for this instead of ortools. Here is the script:

    from ortools.sat.python import cp_model
    import numpy as np
    import pandas as pd
    import time

    def load_affinity_matrix(file_path):
    df = pd.read_excel(file_path, index_col=0)
    return df.to_numpy(), df.index.tolist()

    def optimize_groups(affinity_matrix, min_size=5, max_size=10):
    num_people = len(affinity_matrix)
    model = cp_model.CpModel()

    # Decision Variables: x[g] is 1 if person i is in group g
    max_groups = num_people // min_size # Upper bound on the number of groups
    x = {} # Dictionary to hold variables
    for i in range(num_people):
    for g in range(max_groups):
    x[i, g] = model.NewBoolVar(f'x_{i}_{g}')

    # Constraint 1: Each person must be in exactly one group
    for i in range(num_people):
    model.Add(sum(x[i, g] for g in range(max_groups)) == 1)

    # Constraint 2: Group sizes must be within limits
    for g in range(max_groups):
    model.Add(sum(x[i, g] for i in range(num_people)) >= min_size)
    model.Add(sum(x[i, g] for i in range(num_people)) <= max_size)

    # Constraint 3: Handle affinity rules
    for i in range(num_people):
    for j in range(i + 1, num_people):
    if affinity_matrix[j] == 5: # Must be together
    for g in range(max_groups):
    model.Add(x[i, g] == x[j, g])
    elif affinity_matrix[j] == 0: # Cannot be together
    for g in range(max_groups):
    model.Add(x[i, g] + x[j, g] <= 1)

    # Objective: Maximize total affinity within groups
    objective_terms = []
    for g in range(max_groups):
    for i in range(num_people):
    for j in range(i + 1, num_people):
    if affinity_matrix[j] > 0:
    affinity_value = affinity_matrix[j]
    pair_var = model.NewBoolVar(f'pair_{i}_{j}_{g}')
    model.AddMultiplicationEquality(pair_var, [x[i, g], x[j, g]])
    objective_terms.append(affinity_value * pair_var)

    model.Maximize(sum(objective_terms))

    # Solve
    solver = cp_model.CpSolver()
    solver.parameters.num_search_workers = 8 # Optimized for Apple M2 chip

    print("Starting optimization...")
    start_time = time.time()

    class ProgressCallback(cp_model.CpSolverSolutionCallback):
    def __init__(self, start_time, log_interval=300):
    super().__init__()
    self.start_time = start_time
    self.log_interval = log_interval
    self.last_log_time = start_time

    def on_solution_callback(self):
    current_time = time.time()
    if current_time - self.last_log_time >= self.log_interval:
    print(f"Still optimizing... {round((current_time - self.start_time) / 60, 2)} minutes elapsed.")
    self.last_log_time = current_time

    callback = ProgressCallback(start_time)
    solver.parameters.log_search_progress = False # Disable continuous logging
    status = solver.Solve(model, callback)
    print("Solver finished.")

    if status in (cp_model.FEASIBLE, cp_model.OPTIMAL):
    groups = [[] for _ in range(max_groups)]
    for i in range(num_people):
    for g in range(max_groups):
    if solver.Value(x[i, g]) == 1:
    groups[g].append(i)

    # Convert indices to actual names before returning
    return [[people_names for i in group] for group in groups if group] # Remove empty groups
    else:
    return None

    # Example usage
    affinity_matrix, people_names = load_affinity_matrix('/Users/myname/Documents/affinity_matrix.xlsx')
    groups = optimize_groups(affinity_matrix)

    # Print groups with names
    if groups:
    for idx, group in enumerate(groups, start=1):
    print(f"Group {idx}: {group}")
    else:
    print("No valid grouping found.")

    Continue reading...

Compartilhe esta Página