如何用OR-Tools CP-SAT高效建模制药行业生产Campaign约束？

针对制药行业换产/清洁调度中的Campaign约束（同产品连续生产上限、不同产品强制换产），以下是扩展性更强的OR-Tools CP-SAT建模思路，避免预创建Campaign导致的性能瓶颈：

1. 直接在任务序列上约束，跳过预分配Campaign #

无需提前创建Campaign并分配任务，直接对生产任务的顺序、产品连续性施加约束，让求解器自动处理任务分组，大幅减少变量数量和复杂度。

2. 高效实现连续生产计数约束 #

用布尔变量和递推逻辑替代复杂的累积变量，降低求解器计算负担：

• 定义same_product[i]（布尔变量）：表示第i+1个生产任务与第i个任务产品相同
• 定义run_length[i]（整数变量）：表示到第i个任务为止的连续同产品生产数量
• 核心递推约束：
  • 若same_product[i]为真，run_length[i+1] = run_length[i] + 1
  • 若same_product[i]为假，run_length[i+1] = 1
• 全局约束：所有run_length[i] <= max_campaign_size

3. 换产/清洁工序的触发逻辑 #

通过布尔变量控制清洁任务的激活状态，避免冗余任务：

• 为每两个连续生产任务之间定义清洁任务的start/end变量
• 当same_product[i]为假时，强制清洁任务在两个生产任务之间执行（设置正常时长）
• 当same_product[i]为真时，清洁任务设为虚拟任务（时长为0，相当于跳过）

4. 性能优化关键技巧 #

• 使用AddImplication替代嵌套逻辑约束，简化求解器推理路径
• 将生产任务的时间约束与资源约束（设备、操作人员）统一用IntervalVar和NoOverlap建模
• 设置启发式搜索策略，例如优先安排同产品任务，引导求解器快速收敛到可行解
• 避免创建不必要的中间变量，尽量用原生CP-SAT约束表达业务逻辑

核心代码示例 #

from ortools.sat.python import cp_model

def build_campaign_scheduling_model(production_tasks, max_campaign_size, cleaning_duration):
    model = cp_model.CpModel()
    num_tasks = len(production_tasks)
    
    # 生产任务基础变量
    start = [model.NewIntVar(0, 10000, f"task_start_{i}") for i in range(num_tasks)]
    end = [model.NewIntVar(0, 10000, f"task_end_{i}") for i in range(num_tasks)]
    product_ids = [task['product_id'] for task in production_tasks]
    task_durations = [task['duration'] for task in production_tasks]
    
    # 生产任务时间约束
    for i in range(num_tasks):
        model.Add(end[i] == start[i] + task_durations[i])
    
    # 连续生产计数与换产约束
    if num_tasks > 1:
        same_product = [model.NewBoolVar(f"same_product_{i}") for i in range(num_tasks-1)]
        run_length = [model.NewIntVar(1, max_campaign_size, f"run_length_{i}") for i in range(num_tasks)]
        
        # 第一个任务的连续长度为1
        model.Add(run_length[0] == 1)
        
        for i in range(num_tasks-1):
            # 判断前后任务是否同产品
            model.Add(product_ids[i+1] == product_ids[i]).OnlyEnforceIf(same_product[i])
            model.Add(product_ids[i+1] != product_ids[i]).OnlyEnforceIf(same_product[i].Not())
            
            # 更新连续生产长度
            model.Add(run_length[i+1] == run_length[i] + 1).OnlyEnforceIf(same_product[i])
            model.Add(run_length[i+1] == 1).OnlyEnforceIf(same_product[i].Not())
            
            # 连续生产长度不超过上限
            model.Add(run_length[i+1] <= max_campaign_size)
            
            # 清洁任务变量
            clean_start = model.NewIntVar(0, 10000, f"clean_start_{i}")
            clean_end = model.NewIntVar(0, 10000, f"clean_end_{i}")
            
            # 不同产品时，强制执行清洁
            model.Add(end[i] <= clean_start).OnlyEnforceIf(same_product[i].Not())
            model.Add(clean_end == clean_start + cleaning_duration).OnlyEnforceIf(same_product[i].Not())
            model.Add(clean_end <= start[i+1]).OnlyEnforceIf(same_product[i].Not())
            
            # 同产品时，跳过清洁（时长为0）
            model.Add(clean_start == end[i]).OnlyEnforceIf(same_product[i])
            model.Add(clean_end == clean_start).OnlyEnforceIf(same_product[i])
    
    # 设备资源约束（示例：单台设备）
    production_intervals = [
        cp_model.IntervalVar(start[i], task_durations[i], end[i], f"task_interval_{i}")
        for i in range(num_tasks)
    ]
    model.AddNoOverlap(production_intervals)
    
    # 目标：最小化总生产+清洁时间
    makespan = model.NewIntVar(0, 10000, "makespan")
    model.Add(makespan >= end[-1])
    model.Minimize(makespan)
    
    return model

# 示例任务数据
production_tasks = [
    {"product_id": "A", "duration": 10},
    {"product_id": "A", "duration": 10},
    {"product_id": "A", "duration": 10},
    {"product_id": "A", "duration": 10},
    {"product_id": "A", "duration": 10},
    {"product_id": "B", "duration": 12},
    {"product_id": "B", "duration": 12},
]

# 构建并求解模型
model = build_campaign_scheduling_model(production_tasks, max_campaign_size=4, cleaning_duration=5)
solver = cp_model.CpSolver()
solver.parameters.max_time_in_seconds = 30.0
status = solver.Solve(model)

if status in (cp_model.OPTIMAL, cp_model.FEASIBLE):
    print("调度结果：")
    for i in range(len(production_tasks)):
        print(f"任务{i}（产品{production_tasks[i]['product_id']}）：开始时间={solver.Value(start[i])}，结束时间={solver.Value(end[i])}")
    print(f"总时长：{solver.Value(makespan)}")

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