Pulp优化中确保活动执行时间连续的调度问题

问题背景 #

我有一个基于优先级的调度优化需求，需在给定时间范围内安排活动执行。当前时间足够完成所有活动，但优化结果中部分活动被拆分为非连续时间槽执行，而非按完整时长连续进行。

现有Pulp实现代码 #

import pulp

# 定义活动：包含时长、资源需求、优先级
# 资源需求为每个时间槽的用量（例如每小时需要1台机器，值为1而非1*小时）
activities = {
    "Activity A": {
        "duration": 2,
        "resources": {"resource1": 1, "resource2": 1},
        "priority": 3,
    },
    "Activity B": {
        "duration": 3,
        "resources": {"resource1": 2, "resource2": 1},
        "priority": 2,
    },
    "Activity C": {
        "duration": 1,
        "resources": {"resource1": 1, "resource2": 2},
        "priority": 1,
    },
    "Activity D": {
        "duration": 2,
        "resources": {"resource1": 1, "resource2": 1},
        "priority": 4,
    },
}

# 将优先级转换为单位时长优先级
for name, value in activities.items():
    activities[name]["priority"] = (
        activities[name]["priority"] / activities[name]["duration"]
    )

# 定义时间范围
time_horizon = 10

# 创建线性规划问题（最大化目标）
problem = pulp.LpProblem("ScheduleOptimization", pulp.LpMaximize)

# 创建二进制变量：表示活动在某个时间槽是否执行
activity_vars = {}
for activity in activities:
    for t in range(time_horizon):
        activity_vars[(activity, t)] = pulp.LpVariable(
            f"{activity}_{t}", 0, 1, pulp.LpInteger
        )

# 定义总优先级变量
total_priority = pulp.LpVariable("TotalPriority", cat=pulp.LpContinuous)

# 目标函数：最大化总优先级
problem += total_priority

# 约束1：每个活动执行的总时长不超过其设定时长
for activity_name, activity in activities.items():
    problem += (
        pulp.lpSum(activity_vars[(activity_name, tt)] for tt in range(time_horizon))
        <= activity["duration"]
    )

# 关联总优先级变量与实际执行的优先级总和
problem += total_priority == pulp.lpSum(
    activity["priority"] * activity_vars[(activity_name, t)]
    for activity_name, activity in activities.items()
    for t in range(time_horizon)
)

# 求解问题
problem.solve(pulp.PULP_CBC_CMD(msg=1))

# 输出调度结果
schedule = {}
for activity_name in activities:
    for t in range(time_horizon):
        print(
            activity_vars[(activity_name, t)],
            "___",
            pulp.value(activity_vars[(activity_name, t)]),
        )
        if (
            pulp.value(activity_vars[(activity_name, t)]) == 1
            and schedule.get(activity_name) is None
        ):
            schedule[activity_name] = t

print("Optimal Schedule:")
for activity, start_time in schedule.items():
    print(f"{activity} starts at time {start_time}")
print(f"Total Priority: {pulp.value(total_priority)}")

问题现象 #

预期所有活动的时间槽二进制变量（1表示执行）是连续排列的，例如Activity B的变量向量为[0,0,0,0,0,1,1,1,0,0]。但实际运行后，Activity A、D的变量出现非连续的1：

Activity_A_0 ___ 1.0 # 非连续
Activity_A_1 ___ 0.0 # 非连续
Activity_A_2 ___ 1.0 # 非连续
Activity_A_3 ___ 0.0
Activity_A_4 ___ 0.0
Activity_A_5 ___ 0.0
Activity_A_6 ___ 0.0
Activity_A_7 ___ 0.0
Activity_A_8 ___ 0.0
Activity_A_9 ___ 0.0
Activity_B_0 ___ 0.0
Activity_B_1 ___ 0.0
Activity_B_2 ___ 0.0
Activity_B_3 ___ 0.0
Activity_B_4 ___ 0.0
Activity_B_5 ___ 1.0 # 连续
Activity_B_6 ___ 1.0 # 连续
Activity_B_7 ___ 1.0 # 连续
Activity_B_8 ___ 0.0
Activity_B_9 ___ 0.0
Activity_C_0 ___ 0.0
Activity_C_1 ___ 0.0
Activity_C_2 ___ 0.0
Activity_C_3 ___ 0.0
Activity_C_4 ___ 0.0
Activity_C_5 ___ 0.0
Activity_C_6 ___ 0.0
Activity_C_7 ___ 0.0
Activity_C_8 ___ 1.0
Activity_C_9 ___ 0.0
Activity_D_0 ___ 0.0
Activity_D_1 ___ 1.0 # 非连续
Activity_D_2 ___ 0.0 # 非连续
Activity_D_3 ___ 1.0 # 非连续
Activity_D_4 ___ 0.0
Activity_D_5 ___ 0.0
Activity_D_6 ___ 0.0
Activity_D_7 ___ 0.0
Activity_D_8 ___ 0.0
Activity_D_9 ___ 0.0

解决方案：添加连续执行约束 #

要确保活动连续执行，需要引入开始时间变量并添加关联约束，保证活动一旦开始就连续运行完整个时长。修改后的完整代码如下：

import pulp

# 定义活动：包含时长、资源需求、优先级
activities = {
    "Activity A": {
        "duration": 2,
        "resources": {"resource1": 1, "resource2": 1},
        "priority": 3,
    },
    "Activity B": {
        "duration": 3,
        "resources": {"resource1": 2, "resource2": 1},
        "priority": 2,
    },
    "Activity C": {
        "duration": 1,
        "resources": {"resource1": 1, "resource2": 2},
        "priority": 1,
    },
    "Activity D": {
        "duration": 2,
        "resources": {"resource1": 1, "resource2": 1},
        "priority": 4,
    },
}

# 将优先级转换为单位时长优先级
for name, value in activities.items():
    activities[name]["priority"] = (
        activities[name]["priority"] / activities[name]["duration"]
    )

# 定义时间范围
time_horizon = 10

# 创建线性规划问题（最大化目标）
problem = pulp.LpProblem("ScheduleOptimization", pulp.LpMaximize)

# 创建二进制变量：表示活动在某个时间槽是否执行
activity_vars = {}
for activity in activities:
    for t in range(time_horizon):
        activity_vars[(activity, t)] = pulp.LpVariable(
            f"{activity}_{t}", 0, 1, pulp.LpInteger
        )

# 创建二进制变量：表示活动是否在某个时间点开始
start_vars = {}
for activity in activities:
    for t in range(time_horizon):
        start_vars[(activity, t)] = pulp.LpVariable(
            f"start_{activity}_{t}", 0, 1, pulp.LpInteger
        )

# 定义总优先级变量
total_priority = pulp.LpVariable("TotalPriority", cat=pulp.LpContinuous)

# 目标函数：最大化总优先级
problem += total_priority

# 约束1：每个活动必须执行满设定时长（题目说明时间足够完成所有活动）
for activity_name, activity in activities.items():
    problem += (
        pulp.lpSum(activity_vars[(activity_name, tt)] for tt in range(time_horizon))
        == activity["duration"]
    )

# 约束2：每个活动最多只能在一个时间点开始
for activity_name in activities:
    problem += pulp.lpSum(start_vars[(activity_name, t)] for t in range(time_horizon)) <= 1

# 约束3：活动执行时间槽与开始时间的关联
for activity_name, activity in activities.items():
    duration = activity["duration"]
    for t in range(time_horizon):
        # 如果活动在t开始，那么t到t+duration-1的时间槽必须为1
        for offset in range(duration):
            if t + offset < time_horizon:
                problem += activity_vars[(activity_name, t + offset)] >= start_vars[(activity_name, t)]
        # 如果活动在t时间槽执行，那么它必须在t-duration+1到t之间的某个时间点开始
        problem += activity_vars[(activity_name, t)] <= pulp.lpSum(
            start_vars[(activity_name, max(0, t - duration + 1)) : (t + 1)]
        )

# 关联总优先级变量与实际执行的优先级总和
problem += total_priority == pulp.lpSum(
    activity["priority"] * activity_vars[(activity_name, t)]
    for activity_name, activity in activities.items()
    for t in range(time_horizon)
)

# 求解问题
problem.solve(pulp.PULP_CBC_CMD(msg=1))

# 输出调度结果
schedule = {}
for activity_name in activities:
    print(f"\n{activity_name} time slots:")
    for t in range(time_horizon):
        val = pulp.value(activity_vars[(activity_name, t)])
        print(f"Time {t}: {val}")
        if val == 1 and schedule.get(activity_name) is None:
            schedule[activity_name] = t

print("\nOptimal Schedule:")
for activity, start_time in schedule.items():
    print(f"{activity} starts at time {start_time}, runs until {start_time + activities[activity]['duration'] - 1}")
print(f"Total Priority: {pulp.value(total_priority)}")

关键约束说明 #

1. 开始时间唯一性约束：每个活动最多只能在一个时间点开始，避免多次启动。
2. 执行时间槽关联约束：
   • 如果活动在t开始，那么t到t+duration-1的所有时间槽必须执行该活动。
   • 如果活动在t时间槽执行，那么它的开始时间必须在t-duration+1到t的范围内，确保执行是连续的。
3. 满时长执行约束：将原有的<=改为==，确保所有活动都执行完整时长（符合题目中时间足够的前提）。

修改后，所有活动的执行时间槽都会连续排列，满足预期需求。

内容的提问来源于stack exchange，提问作者user22818073