基于CP Optimizer的复杂任务分配与排程建模咨询

现有代码的基础问题 #

• 存在语法笔误：print('Tasks',TaskList)l 行尾多了无效字符l
• 优先级约束逻辑错误：循环遍历Precedences时，x本身是存储[上游任务ID, 下游任务ID]的一维数组，不能作为数组下标取值；且Precendences存在拼写错误
• 缺失核心建模要素：没有定义调度问题必需的区间变量（interval variable），所有时间、资源约束都无法绑定生效

通用建模实现流程 #

1. 基础配置与核心变量定义 #

CP Optimizer求解调度问题的核心是区间变量，每个任务对应一个区间变量，绑定工时、可用时间范围属性：

from docplex.cp.model import CpoModel
import pandas as pd
import numpy as np

mdl = CpoModel()
# 读取源数据
Task = pd.read_csv("TaskList.csv")
Crew = pd.read_csv("CrewList.csv")

# 基础配置映射，可根据CSV实际字段调整
TASK_BID = dict(zip(Task['ConcatTask'], Task['Days']*8)) # 原数据按天统计则乘8转小时，若已是小时单位直接取工时字段即可
TASK_DEPT = dict(zip(Task['ConcatTask'], Task['Dept']))
# 部门可用时间窗：key为部门名，value为(最早可开始时间, 最晚可结束时间)
DEPT_WINDOW = {d: (
    Crew[Crew['Dpt']==d]['StartWindow'].iloc[0],
    Crew[Crew['Dpt']==d]['EndWindow'].iloc[0]
) for d in Crew['Dpt'].unique()}
# 部门固定人力数
DEPT_CAP = {d: Crew[Crew['Dpt']==d].shape[0] for d in Crew['Dpt'].unique()}
# 相邻任务默认缓冲时间，单位小时，可按部门/任务类型差异化配置
BUFFER_TIME = 0.25

# 为每个任务创建区间变量
task_intervals = {}
for task_id in Task['ConcatTask'].tolist():
    dept = TASK_DEPT[task_id]
    win_start, win_end = DEPT_WINDOW[dept]
    task_duration = TASK_BID[task_id]
    task_intervals[task_id] = mdl.interval_var(
        size=task_duration,
        start=(win_start, win_end - task_duration), # 约束任务执行全程在部门可用时间窗内
        name=f"task_{task_id}"
    )

2. 添加上下游依赖约束 #

修正原有优先级逻辑，同时加入相邻任务缓冲时间要求：

# 加载上下游依赖关系
Precedences = np.vstack((Task['UpTask'].tolist(), Task['DownTask'].tolist())).T
for up_task, down_task in Precedences:
    # 上游任务结束、预留缓冲时间后，下游任务才可启动
    mdl.add(mdl.end_before_start(
        task_intervals[up_task],
        task_intervals[down_task],
        delay=BUFFER_TIME
    ))

3. 添加部门人力容量约束 #

针对“各部门制作人员数量固定、单人同一时间只能处理一个任务”的要求，使用累积资源函数实现，不需要逐个绑定员工，建模效率和求解效率更高：

for dept in DEPT_CAP.keys():
    # 初始化部门资源累积表达式
    dept_resource = mdl.cumul_func()
    # 该部门下每个任务执行期间占用1个人力单位
    for task_id in [t for t in task_intervals.keys() if TASK_DEPT[t] == dept]:
        dept_resource += mdl.pulse(task_intervals[task_id], 1)
    # 约束任意时间点部门人力占用不超过总编制
    mdl.add(dept_resource <= DEPT_CAP[dept])

如果后续需要细化到员工个人日程（比如员工请假、特定员工仅能处理特定类型任务），只需要把上述部门级累积资源约束替换为员工维度的任务序列无重叠约束即可。

4. 设置优化目标 #

核心目标为保障所有任务尽早完成，最小化全局最晚完工时间：

all_task_ends = [mdl.end_of(task_intervals[t]) for t in task_intervals.keys()]
mdl.add(mdl.minimize(mdl.max(all_task_ends)))

如果有明确的全局计划截止日期，可以在创建区间变量时直接给end属性添加上限约束，求解器会自动校验当前人力、时间窗配置下是否存在可行解。

5. 求解与结果导出 #

# 启动求解，可根据需要设置求解时间上限
sol = mdl.solve(TimeLimit=60)
if sol:
    # 提取每个任务的排程结果
    res = []
    for task_id in task_intervals.keys():
        task_sol = sol.get_var_solution(task_intervals[task_id])
        res.append({
            "task_id": task_id,
            "start_time": task_sol.get_start(),
            "end_time": task_sol.get_end(),
            "belong_dept": TASK_DEPT[task_id]
        })
    pd.DataFrame(res).to_csv("schedule_result.csv", index=False)
else:
    print("当前约束下无可行排程，请检查部门人力、时间窗配置是否合理")

适配说明 #

• 200-300个任务的规模属于CP Optimizer的轻量求解场景，上述建模方式通常在1分钟内即可得到最优可行解
• 如果不同部门、不同任务类型的缓冲时间要求不同，只需要把约束里的固定BUFFER_TIME替换成对应任务的配置值即可
• 如果需要排除法定节假日、员工集体休假等不可用时段，只需要在部门时间窗配置中剔除对应时段即可

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