OptaPy约束优化与类重构咨询：员工调度场景需求实现

项目背景 #

基于OptaPy官方员工调度示例开发排班系统，核心场景：

• 双地点班次配置：
  • 地点1：周一至周五（08:00-16:00）每日5个2小时班次，每班最少3人、最多4人
  • 地点2：周一至周五（08:00-18:00）每日6个2小时班次，周六（08:00-12:00）2个2小时班次，每班最少2人、最多4人
• 核心规则：先满足每班最低人员要求，再填充可选名额
• 当前实现：按时段生成多份Shift实例（如地点1某时段生成4个Shift，3个标记required=true，1个required=false），尚未利用required字段实现优先级逻辑

已实现约束 #

1. 员工可用性校验
   员工通过timeslots_unique_hashes存储可胜任时段，若分配的班次不在可用列表则按时长惩罚：

   def required_timeslot(constraint_factory: ConstraintFactory):
       return constraint_factory \
           .for_each(Shift) \
           .filter(lambda shift: shift.unique_hash not in shift.employee.timeslots_unique_hashes) \
           .penalize("Employee missing timeslot", HardSoftScore.ONE_HARD, lambda shift: get_shift_duration_in_minutes(shift))
   

2. 员工每日最多1班次
   限制同一员工单日仅能承担1个班次：

   def one_shift_per_day(constraint_factory: ConstraintFactory):
       return constraint_factory \
           .for_each_unique_pair(Shift,
                 Joiners.equal(lambda shift: shift.employee),
                 Joiners.equal(lambda shift: shift.day_name)
           ) \
           .penalize("Max one shift per day", HardSoftScore.ONE_HARD)
   

待解决需求及实现方案 #

1. 可选：实现必填班次优先填充逻辑 #

通过软约束权重差异实现优先级：对未分配员工的必填班次施加高权重惩罚，可选班次施加低权重或不惩罚，引导OptaPy优先填充必填班次。

def prioritize_required_shifts(constraint_factory: ConstraintFactory):
    # 惩罚未分配员工的必填班次（高权重）
    constraint_factory \
        .for_each(Shift) \
        .filter(lambda shift: shift.required and shift.employee is None) \
        .penalize("Unfilled required shift", HardSoftScore.ofSoft(100))
    # 可选：惩罚未分配的可选班次（低权重，仅作为优化项）
    constraint_factory \
        .for_each(Shift) \
        .filter(lambda shift: not shift.required and shift.employee is None) \
        .penalize("Unfilled optional shift", HardSoftScore.ofSoft(10))

2. 必填：员工配对约束（关联人员同班次） #

调整约束逻辑：按班次哈希分组所有已分配的Shift，遍历每组员工的关联关系，检查关联人员是否全部在当前班次中，按缺失数量惩罚。

首先假设员工类有required_partners字段存储必须同班次的人员列表：

class Employee:
    # 其他字段
    required_partners: List[Employee]

约束实现代码：

def required_partners_in_same_shift(constraint_factory: ConstraintFactory):
    return constraint_factory \
        .for_each(Shift) \
        .groupBy(lambda shift: shift.unique_hash, to_list(lambda shift: shift.employee)) \
        .flatMap(lambda shift_hash, employee_list: employee_list) \
        .filter(lambda employee: employee.required_partners) \
        .penalize("Missing required partners", 
                  HardSoftScore.ONE_HARD,
                  lambda employee, shift_hash, employee_list: 
                      sum(1 for partner in employee.required_partners if partner not in employee_list))

3. 必填：员工至少分配1个班次 #

通过软约束惩罚无班次的员工，设置较高权重确保OptaPy优先为这类员工分配班次：

def at_least_one_shift_per_employee(constraint_factory: ConstraintFactory):
    return constraint_factory \
        .for_each(Employee) \
        .filter(lambda employee: not any(shift.employee == employee for shift in constraint_factory.getProblemFactCollection(Shift))) \
        .penalize("Employee has no shifts", HardSoftScore.ofSoft(200))

同时，为了避免过度分配，可补充员工最多班次限制（根据业务需求调整）：

def max_shifts_per_employee(constraint_factory: ConstraintFactory):
    return constraint_factory \
        .for_each(Employee) \
        .groupBy(lambda employee: employee, count(lambda shift: shift.employee == employee)) \
        .filter(lambda employee, shift_count: shift_count > 5)  # 假设最多5个班次
        .penalize("Too many shifts for employee", HardSoftScore.ofSoft(50), lambda employee, count: count - 5)

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