数据¶
劳动力调度问题的数据文件格式如下:
- 第一行:员工数量
- 第二行:任务数量
- 第三行:天数
后续行由若干节段组成,分别包含以下信息:
- 每个任务的持续时长
- 每项任务每天每小时所需的员工数
- 每位员工可工作的天数
- 工作日内员工可工作的小时数
- 每位员工能执行的任务集
建模思路¶
用于劳动力调度问题的 OptAgent 模型使用布尔决策变量来建模每位员工的排班。对于每位员工、每个任务和每个时间,决策变量表示该员工是否在该时刻开始该任务。
为了降低模型的复杂度,我们进行一些预计算。在读取数据时,我们将每位员工的可用性信息与技能信息合并,生成一个指示变量,用于评估该员工是否能在给定时刻开始任务、执行任务,并在当天结束前完成。我们利用该指示变量来书写模型中的主要约束。为确保没有员工在同一时间被分配多项任务,我们考察所有可能导致冲突的决策变量对,并施加约束以防止它们同时为 true。
接下来,我们估算每位员工每天的工作时间。我们对其施加上下界,以更贴近现实的工作场景。这样可以避免出现一位员工一天工作 12 小时而另一位一周仅工作 1 小时的情形。
对于每个时间步,我们计算所需员工数与分配到各项任务的员工数之间的差值,从而得到每个任务的人手不足与人手过剩的指示变量。
我们定义以下三个目标来求解劳动力调度问题:
- 最小化任务的人手不足
- 最小化任务的人手过剩
- 最小化员工工作日的长度
三个目标按字典序进行优化:声明的顺序即其重要性顺序。将人手不足和人手过剩分成两个独立目标,可以优先满足需求,即便这意味着某些时刻人员偏多。相比于某些时刻严重短缺而另一些时刻过剩,这更可取。该方法同时还能避免人手不足与过剩之间的相互抵消效应。
第三个目标旨在使员工的工作日紧凑,尤其是在仅被分配少量任务时。通常,我们希望避免出现员工在早上 7:00 执行一项一小时任务,又在傍晚 6:00 执行另一项任务,中间长时间空闲的情形。
Python 实现¶
from pathlib import Path
from optagent import OptModel, solve
# Function used to extract the information from the data file
def read_data(filename):
lines = iter(Path(filename).read_text(encoding="utf-8").splitlines())
# Dimensions of the problem
nb_agents = int(next(lines))
nb_tasks = int(next(lines))
nb_days = int(next(lines))
horizon = nb_days * 24
min_working_time = 4
max_working_time = 8
# Tasks duration
next(lines)
task_duration = [0 for i in range(nb_tasks)]
for i in range(nb_tasks):
task_duration[i] = int(next(lines))
# Number of agents needed for the task i at time t
# agents_needed[i][t] contains the number of agents needed
# for the task i at time t
next(lines)
next(lines)
agents_needed = [[0 for t in range(horizon)] for i in range(nb_tasks)]
for d in range(nb_days):
for i in range(nb_tasks):
elements = next(lines).split(" ")
for h in range(24):
t = d * 24 + h
agents_needed[i][t] = int(elements[h])
next(lines)
# Agent disponibility
# day_dispo[a][d] = 1 if agent a is available on day d
day_dispo = [[0 for d in range(nb_days)] for a in range(nb_agents)]
for a in range(nb_agents):
elements = next(lines).split(" ")
for d in range(nb_days):
day_dispo[a][d] = int(elements[d])
# hour_dispo[a][h] = 1 if agent a is available on hour h
next(lines)
hour_dispo = [[0 for h in range(24)] for a in range(nb_agents)]
start_day = [0 for a in range(nb_agents)]
end_day = [0 for a in range(nb_agents)]
for a in range(nb_agents):
elements = next(lines).split(" ")
start_day[a] = int(elements[0])
end_day[a] = int(elements[1])
hour_dispo[a][0 : start_day[a]] = [0] * start_day[a]
hour_dispo[a][start_day[a] : end_day[a]] = [1] * (end_day[a] - start_day[a])
hour_dispo[a][end_day[a] : 24] = [0] * (24 - end_day[a])
# We can concatenate these two informations
# into a global indicator of agent disponibility
# agent_dispo[a][t] = 1 if agent a is available
# on time t
agent_dispo = [[0 for t in range(horizon)] for a in range(nb_agents)]
for a in range(nb_agents):
for d in range(nb_days):
for h in range(24):
t = d * 24 + h
agent_dispo[a][t] = day_dispo[a][d] * hour_dispo[a][h]
# Agent skills
# agent_skills[a][i] = 1 if agent a is able to perform the task i
next(lines)
agent_skills = [[0 for i in range(nb_tasks)] for a in range(nb_agents)]
for a in range(nb_agents):
elements = next(lines).split(" ")
for i in range(nb_tasks):
agent_skills[a][i] = int(elements[i])
# We can use all these informations and task length to get an indicator
# telling us if the agent a can begin the task i at time t,
# and complete it before the end of day
agent_can_start = [[[0 for t in range(horizon)] for i in range(nb_tasks)] for a in range(nb_agents)]
for a in range(nb_agents):
for i in range(nb_tasks):
for d in range(nb_days):
for h in range(0, start_day[a]):
t = d * 24 + h
agent_can_start[a][i][t] = 0
for h in range(start_day[a], end_day[a] - task_duration[i] + 1):
t = d * 24 + h
agent_can_start[a][i][t] = agent_dispo[a][t] * agent_skills[a][i]
for h in range(end_day[a] - task_duration[i] + 1, 24):
t = d * 24 + h
agent_can_start[a][i][t] = 0
return (
nb_agents,
nb_tasks,
nb_days,
horizon,
task_duration,
agents_needed,
day_dispo,
agent_can_start,
min_working_time,
max_working_time,
)
# Returns the time steps corresponding to day d
def times_of_day(d):
return range(d * 24, (d + 1) * 24)
# Returns the time steps s such that
# [s, s + duration) intersects time t
def intersecting_starts(t, duration):
return range(max(t - duration + 1, 0), t + 1)
def main(instance_file, output_file=None, time_limit=30):
(
nb_agents,
nb_tasks,
nb_days,
horizon,
task_duration,
agents_needed,
day_dispo,
agent_can_start,
min_working_time,
max_working_time,
) = read_data(instance_file)
def build_and_solve():
#
# Declare the optimization model
#
model = OptModel()
# Definition of the decision variable :
# agent_start[a][i][t] = 1 if agent a starts the task i at time t
agent_start = [[[model.bool() for t in range(horizon)] for i in range(nb_tasks)] for a in range(nb_agents)]
# An agent can only work if he is able to complete his task
# before the end of the day
for a in range(nb_agents):
for i in range(nb_tasks):
for t in range(horizon):
model.constraint(agent_start[a][i][t] <= agent_can_start[a][i][t])
# Each agent can only work on one task at a time
for a in range(nb_agents):
for i1 in range(nb_tasks):
for i2 in range(nb_tasks):
for t1 in range(horizon):
for t2 in intersecting_starts(t1, task_duration[i2]):
if (i1 != i2) or (t1 != t2):
model.constraint(agent_start[a][i1][t1] + agent_start[a][i2][t2] <= 1)
# Working time per agent per day
agent_working_time = [
[
sum(agent_start[a][i][t] * task_duration[i] for i in range(nb_tasks) for t in times_of_day(d))
for d in range(nb_days)
]
for a in range(nb_agents)
]
# Minimum and maximum amount of time worked
for a in range(nb_agents):
for d in range(nb_days):
if day_dispo[a][d] == 1:
model.constraint(agent_working_time[a][d] >= min_working_time)
model.constraint(agent_working_time[a][d] <= max_working_time)
# Difference between needed and attributed number of agents
# for each task
agent_attributed = [
[
sum(agent_start[a][i][t] for a in range(nb_agents) for t in intersecting_starts(t0, task_duration[i]))
for t0 in range(horizon)
]
for i in range(nb_tasks)
]
agent_diff = [[agents_needed[i][t] - agent_attributed[i][t] for t in range(horizon)] for i in range(nb_tasks)]
# Indicators for lack and excess of agents
agent_lack = sum(model.pow(model.max(agent_diff[i][t], 0), 2) for t in range(horizon) for i in range(nb_tasks))
agent_excess = sum(
model.pow(model.max(-agent_diff[i][t], 0), 2) for t in range(horizon) for i in range(nb_tasks)
)
agent_day_start = [
[
model.min(t + (1 - agent_start[a][i][t]) * horizon for i in range(nb_tasks) for t in times_of_day(d))
for d in range(nb_days)
]
for a in range(nb_agents)
]
agent_day_end = [
[
model.max(
(t + task_duration[i]) * agent_start[a][i][t] for i in range(nb_tasks) for t in times_of_day(d)
)
for d in range(nb_days)
]
for a in range(nb_agents)
]
# Difference between first and last hour of agents each day
agent_day_length = sum(
model.max(agent_day_end[a][d] - agent_day_start[a][d], 0) for a in range(nb_agents) for d in range(nb_days)
)
# Objectives
model.minimize(agent_lack)
model.minimize(agent_excess)
model.minimize(agent_day_length)
solution = solve(model, time_limit_s=float(time_limit))
if not solution.feasible:
print(f"No feasible schedule found; Status = {solution.status}")
return solution
print(
f"Agent lack = {agent_lack.value}; Agent excess = {agent_excess.value}; "
f"Day length = {agent_day_length.value}; Status = {solution.status}"
)
# Outputs if output file specified
if output_file is not None:
lines = [str(agent_lack.value), str(agent_excess.value), str(agent_day_length.value)]
for a in range(nb_agents):
for i in range(nb_tasks):
lines.append(" ".join(str(agent_start[a][i][t].value) for t in range(horizon)) + " ")
lines.append("")
Path(output_file).write_text("\n".join(lines) + "\n", encoding="utf-8")
print("Solution written in file", output_file)
return solution
return build_and_solve()
INSTANCE_DIR = Path.cwd() / "instances"
print("Instances:", INSTANCE_DIR)
solution_2tasks = main(INSTANCE_DIR / "2tasks.txt", time_limit=1)