OR-Tools CP-SAT地理与日历约束集成及任务调度优化问询

问题描述 #

正在学习OR-Tools的CP-SAT求解器，开展一项任务调度与技能匹配优化项目——根据护理人员（operators）的可用性、专业技能、任务地点地理邻近性等多维度条件，自动分配任务。目前已掌握CP-SAT基础，但在复杂约束和数据结构实现上遇到困难，数据集示例如下：

tasks = [
    {"id": "task1", "client": "elderly lady 1", "city": "City X", "province": "Province Y", "service": "physiotherapy"},
    {"id": "task2", "client": "elderly lady 1", "city": "City X", "province": "Province Y", "service": "house cleaning"},
    {"id": "task3", "client": "elderly lady 2", "city": "City Z", "province": "Province Z", "service": "physiotherapy"},
]

operators = [
    {"id": "operator1", "address": "Address A", "province": "Province Y", "specialization": "physiotherapist", "occupied_days": ["2024-03-01", "2024-06-02"]},
    {"id": "operator2", "address": "Address B", "province": "Province Y", "specialization": "cleaner", "occupied_days": ["2024-03-05", "2024-06-06"]},
]

想确认用MongoDB计算任务与护理人员的距离是否可行，同时不清楚如何声明变量并将地理邻近性、日历可用性这类复杂约束集成到CP-SAT中，需要具体思路或示例。

解决方案 #

1. MongoDB计算地理距离的可行性：完全可行 #

CP-SAT擅长处理离散逻辑约束和数值优化，地理距离计算属于预处理步骤，无需在求解器内部执行。利用MongoDB的地理空间能力可高效完成这一步：

• 先将任务的城市/地址、护理人员的地址转换为经纬度（可通过第三方地理编码工具，或MongoDB集成的地理编码能力），存储为MongoDB的GeoJSON格式字段（如location: { type: "Point", coordinates: [lng, lat] }）。
• 使用MongoDB的$geoNear聚合查询，批量计算每个任务到所有护理人员的距离，将结果存入数据集（比如给tasks和operators新增distance_to_xxx字段，或单独维护距离矩阵表）。
• 预处理后的距离为数值类型，可直接用于CP-SAT的约束或目标函数。

2. CP-SAT核心变量声明 #

定义核心决策变量：

• 分配变量：x[t][o]，布尔变量（0或1），表示任务t是否分配给护理人员o。
• （可选）任务执行日期变量：若任务无固定执行日期，可声明d[t]为整数变量（如用日期对应的时间戳或自增序号），表示任务t的执行日期。

3. 核心约束实现 #

（1）技能匹配约束

每个任务只能分配给具备对应服务技能的护理人员：

# 技能映射：护理人员专长对应服务类型
skill_map = {
    "physiotherapist": "physiotherapy",
    "cleaner": "house cleaning"
}

for t in tasks:
    for o in operators:
        if skill_map[o["specialization"]] != t["service"]:
            # 不匹配的组合直接设为0
            model.Add(x[(t["id"], o["id"])] == 0)

（2）日历可用性约束

场景1：任务有固定执行日期

假设每个任务已确定执行日期，若护理人员在该日期被占用，则不能分配该任务：

# 给任务补充固定执行日期（示例）
tasks_with_dates = [
    {**tasks[0], "exec_date": "2024-03-02"},  # 避开operator1的占用日期
    {**tasks[1], "exec_date": "2024-03-06"},  # 避开operator2的占用日期
    {**tasks[2], "exec_date": "2024-06-01"}
]

for t in tasks_with_dates:
    for o in operators:
        if t["exec_date"] in o["occupied_days"]:
            model.Add(x[(t["id"], o["id"])] == 0)

场景2：任务执行日期为变量

若任务执行日期不固定，需用蕴含约束关联分配变量与日期变量：

# 将日期转换为整数序号（便于CP-SAT处理）
date_to_idx = {
    "2024-03-01":1, "2024-03-02":2, "2024-03-05":5, "2024-03-06":6,
    "2024-06-01":92, "2024-06-02":93, "2024-06-06":97
}

# 声明日期变量
date_vars = {}
for t in tasks:
    date_vars[t["id"]] = model.NewIntVar(
        min(date_to_idx.values()), max(date_to_idx.values()),
        f"date_{t['id']}"
    )

# 添加蕴含约束：若任务分配给某护理人员，则执行日期不能在该护理人员的占用列表中
for t in tasks:
    d_t = date_vars[t["id"]]
    for o in operators:
        for occupied_date in o["occupied_days"]:
            occupied_idx = date_to_idx[occupied_date]
            model.Add(d_t != occupied_idx).OnlyEnforceIf(x[(t["id"], o["id"])])

（3）地理邻近性约束

使用预处理好的距离矩阵，设置最大允许距离约束：

MAX_DISTANCE = 50  # 假设最大允许50公里
# 模拟MongoDB计算后的距离矩阵
distance_matrix = [
    [10, 15],  # task1到operator1、operator2的距离
    [12, 8],   # task2到operator1、operator2的距离
    [200, 210] # task3到operator1、operator2的距离（跨省份）
]

task_indices = {t["id"]: i for i, t in enumerate(tasks)}
operator_indices = {o["id"]: i for i, o in enumerate(operators)}

for t in tasks:
    t_idx = task_indices[t["id"]]
    for o in operators:
        o_idx = operator_indices[o["id"]]
        dist = distance_matrix[t_idx][o_idx]
        # 若分配该护理人员，距离不能超过最大值
        model.Add(dist <= MAX_DISTANCE).OnlyEnforceIf(x[(t["id"], o["id"])])

4. 完整示例代码 #

from ortools.sat.python import cp_model

# 原始数据集
tasks = [
    {"id": "task1", "client": "elderly lady 1", "city": "City X", "province": "Province Y", "service": "physiotherapy"},
    {"id": "task2", "client": "elderly lady 1", "city": "City X", "province": "Province Y", "service": "house cleaning"},
    {"id": "task3", "client": "elderly lady 2", "city": "City Z", "province": "Province Z", "service": "physiotherapy"},
]

operators = [
    {"id": "operator1", "address": "Address A", "province": "Province Y", "specialization": "physiotherapist", "occupied_days": ["2024-03-01", "2024-06-02"]},
    {"id": "operator2", "address": "Address B", "province": "Province Y", "specialization": "cleaner", "occupied_days": ["2024-03-05", "2024-06-06"]},
]

# 预处理：任务执行日期
tasks_with_dates = [
    {**tasks[0], "exec_date": "2024-03-02"},
    {**tasks[1], "exec_date": "2024-03-06"},
    {**tasks[2], "exec_date": "2024-06-01"}
]

# 预处理：距离矩阵（模拟MongoDB计算结果）
distance_matrix = [
    [10, 15],
    [12, 8],
    [200, 210]
]
MAX_DISTANCE = 50

# 初始化CP-SAT模型
model = cp_model.CpModel()

# 1. 声明分配变量
x = {}
task_indices = {t["id"]: i for i, t in enumerate(tasks_with_dates)}
operator_indices = {o["id"]: i for i, o in enumerate(operators)}

for t in tasks_with_dates:
    for o in operators:
        x[(t["id"], o["id"])] = model.NewBoolVar(f"x_{t['id']}_{o['id']}")

# 2. 添加约束
# 技能匹配约束
skill_map = {"physiotherapist": "physiotherapy", "cleaner": "house cleaning"}
for t in tasks_with_dates:
    for o in operators:
        if skill_map[o["specialization"]] != t["service"]:
            model.Add(x[(t["id"], o["id"])] == 0)

# 日历可用性约束
for t in tasks_with_dates:
    for o in operators:
        if t["exec_date"] in o["occupied_days"]:
            model.Add(x[(t["id"], o["id"])] == 0)

# 地理邻近性约束
for t in tasks_with_dates:
    t_idx = task_indices[t["id"]]
    for o in operators:
        o_idx = operator_indices[o["id"]]
        dist = distance_matrix[t_idx][o_idx]
        model.Add(dist <= MAX_DISTANCE).OnlyEnforceIf(x[(t["id"], o["id"])])

# 任务唯一性约束：每个任务必须分配给恰好一个护理人员
for t in tasks_with_dates:
    model.Add(sum(x[(t["id"], o["id"])] for o in operators) == 1)

# 3. 设置目标函数：最小化总行驶距离
total_distance = model.NewIntVar(0, 1000, "total_distance")
model.Add(total_distance == sum(
    distance_matrix[task_indices[t["id"]]][operator_indices[o["id"]]] * x[(t["id"], o["id"])]
    for t in tasks_with_dates
    for o in operators
))
model.Minimize(total_distance)

# 4. 求解并输出结果
solver = cp_model.CpSolver()
status = solver.Solve(model)

if status == cp_model.OPTIMAL or status == cp_model.FEASIBLE:
    print("任务分配结果：")
    for t in tasks_with_dates:
        for o in operators:
            if solver.Value(x[(t["id"], o["id"])]) == 1:
                dist = distance_matrix[task_indices[t["id"]]][operator_indices[o["id"]]]
                print(f"任务 {t['id']} -> 护理人员 {o['id']}，距离：{dist}公里，执行日期：{t['exec_date']}")
    print(f"总行驶距离：{solver.Value(total_distance)}公里")
else:
    print("无可行解")

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