如何在OptaPy员工调度中添加最短距离匹配约束？

1. 预处理距离数据（避免实时调用API） #

OptaPy约束会频繁执行，实时调用谷歌地图API会严重拖慢求解速度并消耗配额，必须提前批量计算所有员工-班次地址对的距离，存储为问题事实类：

定义Distance问题事实类 #

@optapy.problem_fact
class Distance:
    employee_address: str
    shift_address: str
    distance_in_meters: int  # 使用数值类型便于分数计算，而非文本

    def __init__(self, employee_address: str, shift_address: str, distance_in_meters: int):
        self.employee_address = employee_address
        self.shift_address = shift_address
        self.distance_in_meters = distance_in_meters

批量预计算距离 #

在加载员工和班次数据后，调用谷歌API计算所有地址对的距离，加入问题事实集合：

import googlemaps
from typing import List
from .models import Employee, Shift, Distance

api_key = '你的API_KEY'
gmaps = googlemaps.Client(api_key)

def precompute_distances(employees: List[Employee], shifts: List[Shift]) -> List[Distance]:
    distance_cache = {}  # 缓存重复地址对，避免重复调用API
    distance_list = []
    for emp in employees:
        emp_addr = emp.address
        if not emp_addr:
            continue
        for shift in shifts:
            shift_addr = shift.location
            if not shift_addr:
                continue
            # 用地址对作为缓存键
            cache_key = (emp_addr, shift_addr)
            if cache_key not in distance_cache:
                try:
                    route = gmaps.directions(emp_addr, shift_addr, mode='driving')
                    distance_meters = route[0]['legs'][0]['distance']['value']
                    distance_cache[cache_key] = distance_meters
                except Exception as e:
                    # 处理API调用失败，设置默认大距离
                    print(f"获取路线失败: {e}")
                    distance_cache[cache_key] = 100000  # 默认100公里
            distance_list.append(Distance(emp_addr, shift_addr, distance_cache[cache_key]))
    return distance_list

在创建求解问题时，将预计算的distance_list添加到problem_facts中：

# 假设已加载employees和shifts数据
distance_list = precompute_distances(employees, shifts)
problem = optapy.create_problem(
    employee_list=employees,
    shift_list=shifts,
    # 其他原有问题事实...
    distance_list=distance_list
)

2. 编写正确的距离约束 #

针对已分配员工的班次，关联对应的距离数据，通过惩罚远距或奖励近距来引导求解器选择最优分配：

约束实现（惩罚远距版本） #

from optapy import ConstraintFactory
from optapy.types import HardSoftScore
from .models import Shift, Distance

def minimize_employee_shift_distance(constraint_factory: ConstraintFactory):
    return constraint_factory.for_each(Shift) \
        # 只处理已分配员工的班次
        .filter(lambda shift: shift.employee is not None) \
        # 关联对应的Distance记录（通过员工地址和班次地址匹配）
        .join(Distance,
              Joiners.equal(lambda shift: shift.employee.address, lambda dist: dist.employee_address),
              Joiners.equal(lambda shift: shift.location, lambda dist: dist.shift_address)) \
        # 距离每1000米扣1个软分，距离越远惩罚越重，引导求解器选近距员工
        .penalize('员工到班次距离惩罚', HardSoftScore.ONE_SOFT,
                  lambda shift, dist: dist.distance_in_meters // 1000)

可选：奖励近距版本 #

如果更倾向于奖励近距离分配，可替换为奖励逻辑（确保奖励值非负）：

def reward_closest_employee_to_shift(constraint_factory: ConstraintFactory):
    return constraint_factory.for_each(Shift) \
        .filter(lambda shift: shift.employee is not None) \
        .join(Distance,
              Joiners.equal(lambda shift: shift.employee.address, lambda dist: dist.employee_address),
              Joiners.equal(lambda shift: shift.location, lambda dist: dist.shift_address)) \
        # 距离越近奖励越高，这里设定10公里以内有效，超过则无奖励
        .reward('员工到班次距离奖励', HardSoftScore.ONE_SOFT,
                lambda shift, dist: max(0, 10000 - dist.distance_in_meters) // 1000)

3. 原有错误说明 #

你之前的约束写法存在三个核心问题：

• 关联条件错误：用Joiners.equal匹配员工地址和班次地址，只筛选地址完全相同的对，不符合“按距离匹配”的需求；
• 参数传递错误：调用find_distance_and_duration_of_route时传入了shift和employee对象，而非地址字符串；
• 实时API调用：约束执行时频繁调用API，导致求解性能极差且消耗大量API配额。

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