基于PULP的工厂需求优化规划求解咨询

工厂内的机器可生产一种或多种产品，单位产品的生产耗时定义为Cycle-Time，例如在Cycle-Time为5的机器上生产10单位产品共需5*10=50秒。仅部分产品适配特定机器，需选择适配机器以最高效的方式完成生产，同时单台机器的总运行时长不可超过其可用时长上限。

优化需求 #

• 计算机器生产产品的运行成本
• 设定产品需求未满足的惩罚成本
• 优化目标为最小化上述两项总成本

示例数据 #

• 工厂机器配置：
• 产品需求：

原有PULP代码问题说明 #

现有代码运行异常、高需求下未用尽机器时长的核心原因如下：

1. 需求设置为硬约束：强制总产量等于需求，当需求超过机器总产能时模型直接不可行，不会触发产能用尽的逻辑
2. 缺少单台机器总运行时长约束：仅限制了单产品在对应机器的产能上下限，没有约束同一机器生产所有产品的总耗时不超过可用上限
3. 目标函数不符合要求：仅统计了总生产耗时，未包含需求未满足的惩罚成本
4. 需求求和硬编码机器列表：固定写死IP01-IP05五台机器，当产品不匹配对应机器时会直接报错

修正后完整代码 #

import pandas as pd
import pulp

# 读取数据
factories = pd.read_csv('factory_machines_small.csv', index_col=['Component', 'Machine'])
demand = pd.read_csv('component_demand_small.csv', index_col=['Component'])

# 定义变量
# 生产数量变量：(产品, 机器)对应产量
production = pulp.LpVariable.dicts(
    "production",
    ((component, machine) for component, machine in factories.index),
    lowBound=0,
    cat='Integer'
)
# 机器生产对应产品的启停状态变量
factory_status = pulp.LpVariable.dicts(
    "factory_status",
    ((component, machine) for component, machine in factories.index),
    cat='Binary'
)
# 新增：需求未满足量变量
unmet_demand = pulp.LpVariable.dicts(
    "unmet_demand",
    demand.index,
    lowBound=0,
    cat='Integer'
)

# 定义模型：最小化总成本
model = pulp.LpProblem("Cost_minimising_scheduling_problem", pulp.LpMinimize)

# 单位时间运行成本可根据实际调整，惩罚成本设为远高于运行成本的数值，优先满足需求
run_cost_per_sec = 1
unmet_punish_per_unit = 1000

# 目标函数：总运行成本 + 总缺货惩罚成本
model += pulp.lpSum(
    [production[component, machine] * factories.loc[(component, machine), 'Cycle_Time'] * run_cost_per_sec 
     for component, machine in factories.index]
) + pulp.lpSum(
    [unmet_demand[component] * unmet_punish_per_unit for component in demand.index]
)

# 约束1：产量+未满足量 = 总需求
for component in demand.index:
    # 动态获取该产品适配的所有机器，避免硬编码报错
    matched_machines = [m for c, m in factories.index if c == component]
    model += pulp.lpSum([production[(component, m)] for m in matched_machines]) + unmet_demand[component] == demand.loc[component, 'Demand']

# 约束2：单产品在对应机器的产量上下限约束
for component, machine in factories.index:
    min_prod = factories.loc[(component, machine), 'Min_Capacity']
    max_prod = factories.loc[(component, machine), 'Max_Capacity']
    model += production[(component, machine)] >= min_prod * factory_status[(component, machine)]
    model += production[(component, machine)] <= max_prod * factory_status[(component, machine)]

# 约束3：新增单台机器总运行时长不超过可用上限（假设factories表中有Available_Time字段存储机器可用时长，没有的话可单独配置）
all_machines = factories.index.get_level_values('Machine').unique()
for machine in all_machines:
    # 获取该机器适配的所有产品
    matched_components = [c for c, m in factories.index if m == machine]
    # 总耗时 = 各产品产量*对应CycleTime之和 <= 机器可用时长
    model += pulp.lpSum(
        [production[(c, machine)] * factories.loc[(c, machine), 'Cycle_Time'] for c in matched_components]
    ) <= factories.loc[(matched_components[0], machine), 'Available_Time'], f"Machine_{machine}_time_limit"

# 求解
model.solve()
print("求解状态：", pulp.LpStatus[model.status])

# 输出结果
output = []
for component, machine in production:
    output.append({
        'Component': component,
        'Machine': machine,
        'Production': production[(component, machine)].varValue,
        'Machine_Status': factory_status[(component, machine)].varValue
    })
output_df = pd.DataFrame.from_records(output).sort_values(['Component', 'Machine'])
output_df.set_index(['Component', 'Machine'], inplace=True)
print(output_df)
output_df.to_csv('OUTPUT.csv')

# 输出未满足需求
print("\n未满足需求：")
for component in demand.index:
    print(f"{component}: {unmet_demand[component].varValue}")

# 输出总成本
print("\n总成本：", pulp.value(model.objective))

补充说明 #

代码中的run_cost_per_sec（单位时间运行成本）和unmet_punish_per_unit（单位缺货惩罚）可根据实际业务调整数值，惩罚成本需要高于生产单位产品的成本，保证模型优先用尽产能再触发缺货。若你的机器配置表中可用时长字段名不是Available_Time，可自行修改为对应字段名即可。

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