Optapy分组调用toList函数报错：TypeError无匹配重载

问题背景 #

处理员工排班场景，目标通过OptaPlanner约束流避免员工排班出现间隙。需求为按员工resource_id关联timeslot_assignment类与availability类，将每组关联数据转为列表后检查排班间隙。原约束流代码如下：

def continuous_shifts(constraint_factory: ConstraintFactory, score = HardSoftScore.ONE_HARD):
    return constraint_factory \
        .for_each(timeslot_assignment_class) \
        .join(availability_class, [
            Joiners.equal(
                lambda timeslot_assignment : timeslot_assignment.resource.resource_id,
                lambda availability : availability.resource.resource_id
            )
        ]) \
        .group_by(lambda timeslot_assignment, availability : timeslot_assignment.resource.resource_id,
                lambda timeslot_assignment, availability : availability.resource.resource_id,
                ConstraintCollectors.to_list()) \
        .penalize("holes in schedule",score,lambda timeslot_list : holes_in_list(timeslot_list))

报错信息 #

TypeError: No matching overloads found for org.optaplanner.constraint.streams.drools.bi.DroolsAbstractBiConstraintStream.groupBy(proxy.PythonBiFunction,proxy.PythonBiFunction,org.optaplanner.core.api.score.stream.DefaultUniConstraintCollector),

问题分析与修正 #

报错核心是groupBy方法的参数格式不符合OptaPlanner BiStream（两个类join后的流）的要求：

1. 原代码重复传入两个相同逻辑的分组键函数，完全多余——join条件已经保证timeslot_assignment和availability的resource_id一致，只需一个分组键即可。
2. BiStream的groupBy方法参数应为分组键函数 + 收集器，若需收集特定对象的列表，需通过mapping收集器先提取目标对象。

修正方案1：收集配对对象列表后提取目标数据 #

def continuous_shifts(constraint_factory: ConstraintFactory, score=HardSoftScore.ONE_HARD):
    return constraint_factory \
        .for_each(timeslot_assignment_class) \
        .join(availability_class, [
            Joiners.equal(
                lambda ta: ta.resource.resource_id,
                lambda a: a.resource.resource_id
            )
        ]) \
        # 按resource_id分组，收集所有(ta, a)配对的列表
        .group_by(
            lambda ta, a: ta.resource.resource_id,
            ConstraintCollectors.to_list()
        ) \
        # 从配对列表中提取timeslot_assignment传入检查函数
        .penalize("holes in schedule", score, lambda resource_id, ta_a_list: holes_in_list([ta for ta, a in ta_a_list]))

修正方案2：直接收集目标对象列表 #

def continuous_shifts(constraint_factory: ConstraintFactory, score=HardSoftScore.ONE_HARD):
    return constraint_factory \
        .for_each(timeslot_assignment_class) \
        .join(availability_class, [
            Joiners.equal(
                lambda ta: ta.resource.resource_id,
                lambda a: a.resource.resource_id
            )
        ]) \
        # 先映射提取timeslot_assignment，再收集为列表
        .group_by(
            lambda ta, a: ta.resource.resource_id,
            ConstraintCollectors.mapping(lambda ta, a: ta, ConstraintCollectors.to_list())
        ) \
        .penalize("holes in schedule", score, lambda resource_id, ta_list: holes_in_list(ta_list))

额外检查点 #

• 确认holes_in_list函数的参数类型与传入的列表类型匹配，确保能正确处理timeslot_assignment对象的时间区间逻辑。

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