数据¶
劳动力轮班调度问题的数据文件格式如下:
第一行:员工数量
第二行:活动数量
第三行:规划周期(以天为单位)
第四行:时间增量(以小时为单位)
后续行:
每个活动:活动的时间窗口(以小时为单位的时间区间)
每个员工:其可用性时间窗口(以小时为单位的时间区间)
建模思路¶
该 OptAgent 模型使用布尔决策变量。每个变量表示某个员工是否被分配到特定的轮班。对于每个活动,模型生成一组可能的轮班。每个轮班代表员工可执行该活动的时间区间。我们通过使用固定时间增量在规划周期内离散化可能起始时间来生成这些轮班。
为防止排班冲突,模型强制施加不重叠约束。如果两个轮班在时间上重叠,则同一员工不能同时承担这两个轮班。该规则适用于所有不相容的轮班对。或者,模型也可以在离散化的时间线上施加该约束,即确保任意员工在任意时间槽(例如每 15 分钟或每小时)中都不会同时进行多个轮班。该方法避免了枚举所有冲突轮班对,并在轮班数量较多时可减少约束数量。两种方法的选择取决于轮班数量和总时间步数。
模型还要求同一员工的任意两个轮班之间至少休息一小时。这一约束与不重叠规则类似。我们将违反该休息时间的轮班对标记为不相容。为实现这一点,我们在检查冲突时将每个轮班的结束时间加上 1 小时。
每位员工每天必须工作恰好两个轮班。模型通过将员工的决策变量求和并令总和等于 2 来强制实施该约束。
为保证充分覆盖,模型将规划周期划分为相等的时间区间。对于每个区间和每个活动,我们检查哪些轮班与该区间重叠,然后统计这些轮班所分配的员工数。若数量不足,则记录人手不足。我们对所有活动与区间的人手不足进行求和,并将其最小化。
为减少不必要的工作量,我们最小化所有员工的总工作时间。两个目标按字典序进行优化:最小化人手不足是首要优先级,总工时则其次。
Python 实现¶
from dataclasses import dataclass
from itertools import combinations
from pathlib import Path
from optagent import OptModel, solve
@dataclass
class Activity:
id: int
min_start: int
max_end: int
duration: int
@dataclass
class Agent:
id: int
availability_start: int
availability_end: int
@dataclass
class Shift:
id: int
activity_id: int
shift_start: int
shift_end: int
def generate_shifts(activities, shift_increment):
shifts = []
for activity in activities:
current_time = activity.min_start
while current_time + activity.duration <= activity.max_end:
shifts.append(
Shift(
len(shifts),
activity.id,
current_time,
current_time + activity.duration,
)
)
current_time += shift_increment
return shifts
def incompatible_shifts(shift_1, shift_2):
return not (shift_1.shift_start >= shift_2.shift_end or shift_1.shift_end <= shift_2.shift_start)
def not_enough_break(shift_1, shift_2):
return not (shift_1.shift_start >= shift_2.shift_end + 3600 or shift_1.shift_end + 3600 <= shift_2.shift_start)
def overlapping_intervals(start_1, end_1, start_2, end_2):
return not (start_1 >= end_2 or end_1 <= start_2)
def read_elements(filename):
elements = []
for line in Path(filename).read_text(encoding="utf-8").splitlines():
content = line.split("#", maxsplit=1)[0].strip()
if content:
elements.extend(content.split())
return elements
def read_instance(filename):
elements = iter(read_elements(filename))
nb_agents = int(next(elements))
nb_activities = int(next(elements))
time_horizon = int(next(elements))
shift_increment = round(float(next(elements)) * 3600)
activities = [Activity(activity, 0, 0, int(next(elements)) * 3600) for activity in range(nb_activities)]
for activity in activities:
activity.min_start = int(next(elements)) * 3600
activity.max_end = int(next(elements)) * 3600
agents = [
Agent(
agent,
int(next(elements)) * 3600,
int(next(elements)) * 3600,
)
for agent in range(nb_agents)
]
return time_horizon, shift_increment, activities, agents
def validate_instance(activities, agents):
for activity in activities:
if activity.duration > activity.max_end - activity.min_start:
raise ValueError(f"Activity {activity.id} duration ({activity.duration / 3600}h) exceeds its time window")
for agent in agents:
if agent.availability_start >= agent.availability_end:
raise ValueError(f"Agent {agent.id} has an invalid availability window")
earliest_start = min(activity.min_start for activity in activities)
latest_end = max(activity.max_end for activity in activities)
for agent in agents:
if agent.availability_start > earliest_start or agent.availability_end < latest_end:
print(f"Warning: Agent {agent.id} availability might not cover all tasks")
def main(instance_file, output_file=None, time_limit=5):
_, shift_increment, activities, agents = read_instance(instance_file)
validate_instance(activities, agents)
shifts = generate_shifts(activities, shift_increment)
nb_agents = len(agents)
model = OptModel()
agent_shift = [
[model.bool() for shift in shifts] for agent in range(nb_agents)
]
# Each pair is enumerated once instead of twice as in the Hexaly loops.
for agent in range(nb_agents):
for shift_1, shift_2 in combinations(shifts, 2):
if incompatible_shifts(shift_1, shift_2):
model.constraint(
agent_shift[agent][shift_1.id] + agent_shift[agent][shift_2.id] <= 1,
)
if not_enough_break(shift_1, shift_2):
model.constraint(
agent_shift[agent][shift_1.id] + agent_shift[agent][shift_2.id] <= 1,
)
for agent in range(nb_agents):
model.constraint(
sum(agent_shift[agent][shift.id] for shift in shifts) == 2,
)
total_understaffing = 0
for activity in activities:
current_time = activity.min_start
while current_time < activity.max_end:
interval_end = min(current_time + shift_increment, activity.max_end)
shifts_at_current = [
shift
for shift in shifts
if shift.activity_id == activity.id
and overlapping_intervals(
current_time,
interval_end,
shift.shift_start,
shift.shift_end,
)
]
total_agents_working = sum(
agent_shift[agent][shift.id] for shift in shifts_at_current for agent in range(nb_agents)
)
total_understaffing += model.max(0, 1 - total_agents_working)
current_time += shift_increment
total_understaffing *= shift_increment
total_working_time = sum(
agent_shift[agent][shift.id] * activities[shift.activity_id].duration
for agent in range(nb_agents)
for shift in shifts
)
model.minimize(total_understaffing)
model.minimize(total_working_time)
solution = solve(model, time_limit_s=float(time_limit))
if not solution.feasible:
print(f"No feasible schedule found; Status = {solution.status}")
return solution
selected_shifts = [
(agent, shift) for agent in range(nb_agents) for shift in shifts if agent_shift[agent][shift.id].value
]
lines = [f"{agent} {shift.activity_id} {shift.shift_start} {shift.shift_end}" for agent, shift in selected_shifts]
print(
f"Understaffing = {total_understaffing.value}; "
f"Working time = {total_working_time.value}; Status = {solution.status}"
)
if lines:
print("Agent Activity Start End\n" + "\n".join(lines))
if output_file is not None:
Path(output_file).write_text("\n".join(lines) + "\n", encoding="utf-8")
return solution
INSTANCE_DIR = Path.cwd() / "instances"
print("Instances:", INSTANCE_DIR)
solution = main(INSTANCE_DIR / "1agents6tasks_n5.txt", time_limit=1)