基于CP SAT Solver的车间调度动态库存管理问题求助

API 探路者

2026-6-20

车间制造调度的库存时间维度约束解决方案

问题背景

已通过Python的CP SAT Solver实现车间制造调度中的任务依赖、目标日期及最早/最晚调度等约束，但在库存随时间动态变化的环节遇到瓶颈：无法关联任务的开始/结束时间与库存消耗、产出的时序关系，导致无法保证调度过程中无库存短缺，同时要实现最大化任务调度量（支持任务并行）。

输入的调度数据如下：

{"stocks":{"toto": 600,"tata": 6000},"tasks": [{"id": 1,"duration": 50,"stocks_required":{"tata": 600},"stocks_produced":{"toto" : 10}},{"id": 2,"duration": 50,"stocks_required":{"tata": 600},"stocks_produced":{"toto" : 1}},{"id" : 3,"duration": 100,"stocks_required":{"toto": 660},"stocks_produced":{"tata" : 300}},{"id" : 4,"duration": 5,"stocks_required":{"tata" : 600,"toto" : 8},"stocks_produced":{"toto" : 60}}]}

当前已实现无时间维度的任务最大化调度代码（仅考虑静态库存总和约束）：

from ortools.sat.python import cp_model
import json

# Définition des données
json_brut = """
    {
      "stocks": 
        {
          "toto": 600,
          "tata": 6000
        },
      "taches": [
        {
          "id": 1,
          "stocks_requis": 
            {
              "tata" : 600
            }
        },
        {
          "id": 2,
          "stocks_requis": 
            {
              "tata" : 600
            }
        },
        {
          "id": 3,
          "stocks_requis": 
            {
              "toto" : -600
            }
        },
        {
          "id": 4,
          "stocks_requis": 
            {
              "tata" : 600,
              "toto" : 8
            }
        }
      ]
    }
"""

data = json.loads(json_brut)

stocks = data["stocks"]
taches = data["taches"]

# Création du modèle
model = cp_model.CpModel()

# Définition des variables
execute_vars = {}
for tache in taches:
    execute_vars[tache["id"]] = model.NewBoolVar("execute_%i" % tache["id"])

# Définition des contraintes
for tache in taches:
    for stock, quantite in tache["stocks_requis"].items():
        model.Add(sum(execute_vars[t["id"]] * t["stocks_requis"].get(stock, 0) for t in taches) <= stocks[stock])

# Définition de l'objectif
objectif = sum(execute_vars.values())
model.Maximize(objectif)

# Résolution du modèle
solver = cp_model.CpSolver()
status = solver.Solve(model)

# Affichage des résultats
if status == cp_model.OPTIMAL:
    print("Solution optimale trouvée:")
    for tache in taches:
        if solver.Value(execute_vars[tache["id"]]):
            print("Tâche %i: exécutée" % tache["id"])
        else:
            print("Tâche %i: non exécutée" % tache["id"])
else:
    print("Aucune solution trouvée")

解决方案：时间维度与库存动态约束建模

核心思路是通过事件驱动的库存变化约束，将任务的时间变量（开始/结束时间）与库存的消耗、产出动作绑定，确保任意时间点库存水平非负。

步骤1：定义核心变量

每个任务的布尔执行变量（execute_var）
每个任务的开始时间（start_var）、结束时间（end_var），使用整数变量表示时间单位
设定时间上限（如MAX_TIME = 1000，根据任务总工期预估）

步骤2：任务时间约束

若任务执行：end_var = start_var + duration
若任务不执行：强制start_var和end_var为0（或不占用时间资源），用蕴含约束实现

步骤3：动态库存约束

对于每个物料，跟踪库存的时序变化：

初始库存：给定的初始值
任务开始事件：消耗对应库存，此时库存 = 当前库存 - 需求数量 ≥ 0
任务结束事件：增加对应产出库存，库存 = 当前库存 + 产出数量

利用CP SAT的线性约束+蕴含关系，结合时间点的先后顺序来实现库存的动态流转。

完整实现代码

from ortools.sat.python import cp_model
import json

# 加载输入数据
input_json = """{"stocks":{"toto": 600,"tata": 6000},"tasks": [{"id": 1,"duration": 50,"stocks_required":{"tata": 600},"stocks_produced":{"toto" : 10}},{"id": 2,"duration": 50,"stocks_required":{"tata": 600},"stocks_produced":{"toto" : 1}},{"id" : 3,"duration": 100,"stocks_required":{"toto": 660},"stocks_produced":{"tata" : 300}},{"id" : 4,"duration": 5,"stocks_required":{"tata" : 600,"toto" : 8},"stocks_produced":{"toto" : 60}}]}"""
data = json.loads(input_json)

stocks_initial = data["stocks"]
tasks = data["tasks"]
MAX_TIME = 1000  # 根据实际场景调整时间上限

# 创建模型
model = cp_model.CpModel()

# 定义变量
execute_vars = {}
start_vars = {}
end_vars = {}

for task in tasks:
    task_id = task["id"]
    execute_vars[task_id] = model.NewBoolVar(f"execute_{task_id}")
    start_vars[task_id] = model.NewIntVar(0, MAX_TIME, f"start_{task_id}")
    end_vars[task_id] = model.NewIntVar(0, MAX_TIME, f"end_{task_id}")
    
    # 任务执行的时间约束：若执行，则end = start + duration；否则start=end=0
    model.Add(end_vars[task_id] == start_vars[task_id] + task["duration"]).OnlyEnforceIf(execute_vars[task_id])
    model.Add(start_vars[task_id] == 0).OnlyEnforceIf(execute_vars[task_id].Not())
    model.Add(end_vars[task_id] == 0).OnlyEnforceIf(execute_vars[task_id].Not())

# 动态库存约束：对每个物料，跟踪所有时间点的库存水平
for stock_name, initial_qty in stocks_initial.items():
    # 定义每个任务对该物料的消耗/产出事件的时间点和数量
    events = []
    # 初始库存作为时间0的事件
    events.append((0, initial_qty))
    
    for task in tasks:
        task_id = task["id"]
        # 任务开始时：消耗库存（负数表示消耗）
        req_qty = task["stocks_required"].get(stock_name, 0)
        if req_qty != 0:
            # 仅当任务执行时，该事件生效
            event_start = model.NewIntVar(0, MAX_TIME, f"event_start_{task_id}_{stock_name}")
            model.Add(event_start == start_vars[task_id]).OnlyEnforceIf(execute_vars[task_id])
            model.Add(event_start == 0).OnlyEnforceIf(execute_vars[task_id].Not())
            events.append((event_start, -req_qty))
        
        # 任务结束时：产出库存
        prod_qty = task["stocks_produced"].get(stock_name, 0)
        if prod_qty != 0:
            event_end = model.NewIntVar(0, MAX_TIME, f"event_end_{task_id}_{stock_name}")
            model.Add(event_end == end_vars[task_id]).OnlyEnforceIf(execute_vars[task_id])
            model.Add(event_end == 0).OnlyEnforceIf(execute_vars[task_id].Not())
            events.append((event_end, prod_qty))
    
    # 对所有事件按时间排序，计算累积库存并确保非负
    n_events = len(events)
    for i in range(n_events):
        for j in range(i+1, n_events):
            time_i, delta_i = events[i]
            time_j, delta_j = events[j]
            # 定义两个事件的先后关系
            is_before = model.NewBoolVar(f"before_{i}_{j}")
            model.Add(time_i <= time_j).OnlyEnforceIf(is_before)
            model.Add(time_i > time_j).OnlyEnforceIf(is_before.Not())
            
            # 计算到i事件后的库存，以及到j事件后的库存
            stock_i = sum(d for t, d in events[:i+1])
            stock_j = stock_i + delta_j
            
            # 如果i在j之前，那么j事件发生后的库存不能为负；同时i事件后的库存也不能为负
            model.Add(stock_i >= 0).OnlyEnforceIf(is_before)
            model.Add(stock_j >= 0).OnlyEnforceIf(is_before)
            
            # 如果j在i之前，同理
            model.Add(sum(d for t, d in events[:j+1]) >= 0).OnlyEnforceIf(is_before.Not())
            model.Add(sum(d for t, d in events[:j+1]) + delta_i >= 0).OnlyEnforceIf(is_before.Not())

# 目标：最大化执行的任务数量
model.Maximize(sum(execute_vars.values()))

# 求解
solver = cp_model.CpSolver()
status = solver.Solve(model)

# 输出结果
if status == cp_model.OPTIMAL:
    print("最优调度方案：")
    for task in tasks:
        task_id = task["id"]
        if solver.Value(execute_vars[task_id]):
            print(f"任务{task_id}：执行，开始时间={solver.Value(start_vars[task_id])}，结束时间={solver.Value(end_vars[task_id])}")
        else:
            print(f"任务{task_id}：不执行")
else:
    print("无可行调度方案")

关键说明

代码中通过事件点的时间顺序约束来模拟库存的动态变化，确保任意事件发生后库存水平非负
使用OnlyEnforceIf实现任务执行状态与事件生效的绑定，避免未执行任务影响库存
时间上限MAX_TIME可根据实际任务总工期调整，平衡求解效率与覆盖范围

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

基于CP SAT Solver的车间调度动态库存管理问题求助

问题背景 #

解决方案：时间维度与库存动态约束建模 #

步骤1：定义核心变量 #

步骤2：任务时间约束 #

步骤3：动态库存约束 #

完整实现代码 #

关键说明 #