在CP-SAT护士排班模型中添加软/硬连续工作日约束的技术咨询

一、Negated Bounded Span函数工作原理 #

AddNegatedBoundedSpan是OR-Tools CP-SAT专门用于约束连续布尔变量序列的工具，核心作用是禁止序列中出现连续k个为真的变量。它会自动生成一组隐含约束，确保任意连续k个位置里至少有一个变量为假，正好适配「限制最大连续工作日」的需求。

二、参数传递说明 #

调用该函数时需要传入两个参数：

• sequence：布尔变量列表，按时间顺序排列，每个变量代表对应日期护士是否工作（1=工作，0=休息）。
• k：整数，代表禁止出现的连续工作天数+1。比如要限制最多连续4天工作，就传入k=5——禁止连续5天工作，等价于最多连续4天。

三、在你的排班系统中实现约束 #

1. 全局硬最大连续工作日限制 #

先为每个护士生成每日工作状态的布尔序列，再用AddNegatedBoundedSpan添加硬约束：

# 生成每个护士的每日工作状态变量
daily_work = {}
for n in all_nurses:
    daily_work[n] = []
    for d in all_days:
        worked = model.new_bool_var(f"daily_work_n{n}_d{d}")
        if is_nurse_available(n, d):
            # 当天有任何班次被安排即视为工作
            model.add(worked == cp_model.sum(shifts[(n, d, s)] for s in all_shifts) >= 1)
        else:
            # 调休日强制不工作
            model.add(worked == 0)
        daily_work[n].append(worked)

# 全局硬约束：所有护士连续工作天数不超过max_cwork
for n in all_nurses:
    model.add_negated_bounded_span(daily_work[n], max_cwork + 1)

2. 个人软连续工作日约束 #

软约束通过惩罚项实现，违反时降低目标函数值，让求解器尽量避免：

• 软最大连续约束：若护士连续工作超过个人设定的最大值，添加惩罚。
• 软最小连续约束：若护士工作段长度不足个人设定的最小值，添加惩罚。

# 软约束惩罚权重（可根据业务需求调整）
SOFT_MAX_PENALTY = 100
SOFT_MIN_PENALTY = 50

# 处理每个护士的个人软连续约束
for n in all_nurses:
    if "cwork" not in nurse_preferences[n]:
        continue
    min_consec, max_consec = nurse_preferences[n]["cwork"]
    
    # 软最大连续约束：禁止连续max_consec+1天工作，违反则惩罚
    for i in range(num_days - max_consec):
        window = daily_work[n][i:i+max_consec+1]
        violation = model.new_bool_var(f"violate_max_cwork_n{n}_i{i}")
        # 当窗口内所有天都工作时，标记为违反
        model.add(cp_model.sum(window) == max_consec + 1).only_enforce_if(violation)
        # 目标函数中减去惩罚值
        model.maximize(cp_model.LinearExpr.term(violation, -SOFT_MAX_PENALTY))
    
    # 软最小连续约束：若工作段长度不足min_consec，添加惩罚
    for d in all_days:
        if not is_nurse_available(n, d):
            continue
        # 判断当前是否为工作段的起始日（当天工作且前一天不工作，或为第一天）
        is_start = model.new_bool_var(f"start_seq_n{n}_d{d}")
        if d == 0:
            model.add(is_start == daily_work[n][d])
        else:
            model.add(is_start == cp_model.And(daily_work[n][d], cp_model.Not(daily_work[n][d-1])))
        
        # 检查从起始日开始的min_consec天是否都工作
        end_idx = min(d + min_consec, num_days)
        required_days = daily_work[n][d:end_idx]
        violation = model.new_bool_var(f"violate_min_cwork_n{n}_d{d}")
        # 若起始日之后的天数未全部工作，标记为违反
        model.add(cp_model.sum(required_days) < len(required_days)).only_enforce_if(is_start)
        model.add(violation == 1).only_enforce_if(cp_model.And(is_start, cp_model.sum(required_days) < len(required_days)))
        # 目标函数中减去惩罚值
        model.maximize(cp_model.LinearExpr.term(violation, -SOFT_MIN_PENALTY))

四、修改后的完整代码 #

from ortools.sat.python import cp_model

max_cwork = 4  # 全局硬约束：最大连续工作日
num_nurses = 4
num_shifts = 2
num_days = 9
all_nurses = range(num_nurses)
all_shifts = range(num_shifts)
all_days = range(num_days)

# "off" : 硬调休（当天绝对不能工作）
# "prefer": 软偏好（优先安排这些天工作）
# "cwork": 软连续工作日约束（min, max）
nurse_preferences = {
    0: { "off": { 8 }, "prefer": { 0, 1, 2 }, "cwork": { 3, 3 } },
    1: { "off": { 5 }, "prefer": { 0, 1, 2, 6, 7, 8 } },
    2: { "off": { 2 }, "prefer": { 3, 4, 5, 6, 7, 8 } },
    3: { "off": { 0 }, "prefer": { 3, 4, 5 }, "cwork": { 1, 2 } },
}

nurse_prefer_requests = { kvp[0]: len(kvp[1]["prefer"]) if "prefer" in kvp[1] else 0 for kvp in nurse_preferences.items() }
max_nurse_prefer_requests = sum(nurse_prefer_requests.values())

nurse_prefer_factor = { kvp[0]: 1 if kvp[1] == 0 else max_nurse_prefer_requests / kvp[1] for kvp in nurse_prefer_requests.items() }

def is_nurse_available(n, d):
    return (n not in nurse_preferences or "off" not in nurse_preferences[n] or d not in nurse_preferences[n]["off"])

def get_nurse_day_preference(n, d):
    if (n not in nurse_preferences or "prefer" not in nurse_preferences[n] or d not in nurse_preferences[n]["prefer"]):
        return 1
    return nurse_prefer_factor[n]

model = cp_model.CpModel()

shifts = {}
for n in all_nurses:
    for d in all_days:
        for s in all_shifts:
            if is_nurse_available(n,d):
                shifts[(n, d, s)] = model.new_bool_var(f"shift_n{n}_d{d}_s{s}")

# 每天每个班次必须安排恰好一名护士
for d in all_days:
    for s in all_shifts:
        model.add_exactly_one(shifts[(n, d, s)] for n in all_nurses if (n, d, s) in shifts)

# 每个护士每天最多安排一个班次
for n in all_nurses:
    for d in all_days:
        model.add_at_most_one(shifts[(n, d, s)] for s in all_shifts if (n, d, s) in shifts)

# 每个护士的排班数量上下限
min_shifts_per_nurse = (num_shifts * num_days) // num_nurses
max_shifts_per_nurse = min_shifts_per_nurse + 1 if (num_shifts * num_days) % num_nurses != 0 else min_shifts_per_nurse
for n in all_nurses:
    shifts_worked = []
    for d in all_days:
        for s in all_shifts:
            if (n, d, s) in shifts:
                shifts_worked.append(shifts[(n, d, s)])
    model.add(min_shifts_per_nurse <= sum(shifts_worked))
    model.add(sum(shifts_worked) <= max_shifts_per_nurse)

# -------------------------- 新增约束部分 --------------------------
# 生成每个护士的每日工作状态变量
daily_work = {}
for n in all_nurses:
    daily_work[n] = []
    for d in all_days:
        worked = model.new_bool_var(f"daily_work_n{n}_d{d}")
        if is_nurse_available(n, d):
            model.add(worked == cp_model.sum(shifts[(n, d, s)] for s in all_shifts) >= 1)
        else:
            model.add(worked == 0)
        daily_work[n].append(worked)

# 全局硬约束：所有护士连续工作天数不超过max_cwork
for n in all_nurses:
    model.add_negated_bounded_span(daily_work[n], max_cwork + 1)

# 软约束惩罚权重
SOFT_MAX_PENALTY = 100
SOFT_MIN_PENALTY = 50

# 处理个人软连续工作日约束
for n in all_nurses:
    if "cwork" not in nurse_preferences[n]:
        continue
    min_consec, max_consec = nurse_preferences[n]["cwork"]
    
    # 软最大连续约束
    for i in range(num_days - max_consec):
        window = daily_work[n][i:i+max_consec+1]
        violation = model.new_bool_var(f"violate_max_cwork_n{n}_i{i}")
        model.add(cp_model.sum(window) == max_consec + 1).only_enforce_if(violation)
        model.maximize(cp_model.LinearExpr.term(violation, -SOFT_MAX_PENALTY))
    
    # 软最小连续约束
    for d in all_days:
        if not is_nurse_available(n, d):
            continue
        is_start = model.new_bool_var(f"start_seq_n{n}_d{d}")
        if d == 0:
            model.add(is_start == daily_work[n][d])
        else:
            model.add(is_start == cp_model.And(daily_work[n][d], cp_model.Not(daily_work[n][d-1])))
        
        end_idx = min(d + min_consec, num_days)
        required_days = daily_work[n][d:end_idx]
        violation = model.new_bool_var(f"violate_min_cwork_n{n}_d{d}")
        model.add(cp_model.sum(required_days) < len(required_days)).only_enforce_if(is_start)
        model.add(violation == 1).only_enforce_if(cp_model.And(is_start, cp_model.sum(required_days) < len(required_days)))
        model.maximize(cp_model.LinearExpr.term(violation, -SOFT_MIN_PENALTY))
# -------------------------- 新增约束结束 --------------------------

# 最大化偏好满足度
model.maximize(
    sum(
        get_nurse_day_preference(n,d) * shifts[(n, d, s)]
        for n in all_nurses
        for d in all_days
        for s in all_shifts
        if (n,d,s) in shifts
    )
)

solver = cp_model.CpSolver()
solver.parameters.enumerate_all_solutions = True
status = solver.solve(model)

if status == cp_model.OPTIMAL:
    print("最优解：")
    for d in all_days:
        print(f"第{d}天")
        for n in all_nurses:
            for s in all_shifts:
                if (n,d,s) in shifts and solver.value(shifts[(n, d, s)]) == 1:
                    if get_nurse_day_preference(n,d) > 1:
                        print(f"护士{n} 排班{s}（偏好日）")
                    else:
                        print(f"护士{n} 排班{s}（非偏好日）")
        print()
    print(f"偏好满足得分：{solver.objective_value}")
else:
    print("未找到最优解！")

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