如何在Pyomo中基于CPLEX求解主-子问题分解模型？

问题背景 #

需要用Pyomo结合CPLEX求解一个主-子问题分解模型，核心逻辑是交替求解主、子问题：主问题的变量作为子问题的输入参数，子问题的变量反过来更新主问题的参数，直到目标函数变化量小于设定阈值sigma。

原始代码的核心问题 #

你提供的代码存在几个致命问题：

1. 抽象模型未声明集合i和t，Pyomo的AbstractModel必须先定义集合才能使用带索引的参数/变量
2. 主、子问题均未定义目标函数，代码中instance1.of无定义
3. 参数赋值方式错误，不能直接用instance1.z1=instance.z1批量赋值索引化参数
4. While循环缩进错误，导致迭代逻辑仅执行一次
5. 未检查求解器终止状态，无法保证每次求解成功

修正后的完整实现代码 #

from pyomo.environ import *
from pyomo.opt import SolverStatus, TerminationCondition

# 全局配置：根据实际问题修改集合与收敛阈值
sigma = 2  # 目标函数收敛阈值
TIME_SET = [1, 2, 3]  # t集合示例
ITEM_SET = [1, 2]     # i集合示例

############### 主问题构建 #########################################
def build_master():
    model = AbstractModel()
    # 声明集合
    model.t = Set(initialize=TIME_SET)
    model.i = Set(initialize=ITEM_SET)
    # 可更新参数（由子问题输出提供）
    model.z1 = Param(model.i, mutable=True, initialize=0)
    model.z2 = Param(model.i, mutable=True, initialize=39)
    # 主问题决策变量
    model.x = Var(model.t, within=Reals)
    model.X = Var(model.t, within=Reals)
    # 目标函数：替换为你的实际目标
    def obj_rule(m):
        return sum(m.x[t] + m.X[t] for t in m.t)
    model.of = Objective(rule=obj_rule, sense=minimize)
    # 主问题约束：替换为你的实际约束
    def master_constr(m, t):
        return m.x[t] <= m.X[t] + sum(m.z1[i] for i in m.i)
    model.master_constraint = Constraint(model.t, rule=master_constr)
    return model

# 初始化并求解初始主问题
master_model = build_master()
master_instance = master_model.create_instance()
opt = SolverFactory('cplex')

res_master = opt.solve(master_instance, tee=True)
if res_master.solver.status != SolverStatus.ok or res_master.solver.termination_condition != TerminationCondition.optimal:
    raise RuntimeError("主问题初始求解失败")

objective_history = [value(master_instance.of)]

######## 子问题构建 #####################
def build_subproblem():
    model = AbstractModel()
    # 声明集合
    model.t = Set(initialize=TIME_SET)
    model.i = Set(initialize=ITEM_SET)
    # 可更新参数（由主问题输出提供）
    model.x = Param(model.t, mutable=True)
    model.X = Param(model.t, mutable=True)
    # 子问题决策变量
    model.z1 = Var(model.i, within=Reals)
    model.z2 = Var(model.i, within=Reals)
    # 子问题目标函数：替换为你的实际目标
    def obj_rule(m):
        return sum(m.z1[i] + m.z2[i] for i in m.i)
    model.of = Objective(rule=obj_rule, sense=minimize)
    # 子问题约束：替换为你的实际约束
    def sub_constr(m, i):
        return m.z1[i] >= m.z2[i] - sum(m.x[t] for t in m.t)
    model.sub_constraint = Constraint(model.i, rule=sub_constr)
    return model

# 初始化子问题并传递初始主问题变量
sub_model = build_subproblem()
sub_instance = sub_model.create_instance()
for t in TIME_SET:
    sub_instance.x[t] = value(master_instance.x[t])
    sub_instance.X[t] = value(master_instance.X[t])

res_sub = opt.solve(sub_instance, tee=True)
if res_sub.solver.status != SolverStatus.ok or res_sub.solver.termination_condition != TerminationCondition.optimal:
    raise RuntimeError("子问题初始求解失败")

######## 交替迭代求解 #####################
while len(objective_history) < 2 or (objective_history[-2] - objective_history[-1]) > sigma:
    # 更新主问题参数：子问题变量 -> 主问题参数
    for i in ITEM_SET:
        master_instance.z1[i] = value(sub_instance.z1[i])
        master_instance.z2[i] = value(sub_instance.z2[i])
    
    # 求解主问题
    res_master = opt.solve(master_instance, tee=True)
    if res_master.solver.status != SolverStatus.ok or res_master.solver.termination_condition != TerminationCondition.optimal:
        print("主问题求解异常，终止迭代")
        break
    objective_history.append(value(master_instance.of))
    print(f"当前主问题目标值: {objective_history[-1]}")
    
    # 更新子问题参数：主问题变量 -> 子问题参数
    for t in TIME_SET:
        sub_instance.x[t] = value(master_instance.x[t])
        sub_instance.X[t] = value(master_instance.X[t])
    
    # 求解子问题
    res_sub = opt.solve(sub_instance, tee=True)
    if res_sub.solver.status != SolverStatus.ok or res_sub.solver.termination_condition != TerminationCondition.optimal:
        print("子问题求解异常，终止迭代")
        break
    print(f"当前子问题目标值: {value(sub_instance.of)}")

# 输出最终结果
print("\n迭代终止，主问题目标值序列:", objective_history)
print("\n最终主问题变量x值:")
for t in TIME_SET:
    print(f"x[{t}] = {round(value(master_instance.x[t]), 4)}")
print("\n最终子问题变量z1值:")
for i in ITEM_SET:
    print(f"z1[{i}] = {round(value(sub_instance.z1[i]), 4)}")

关键修正说明 #

• 为抽象模型补充了集合定义，这是Pyomo抽象模型的必要前提
• 完善了主、子问题的目标函数与约束示例，请根据你的实际模型修改对应部分
• 使用循环遍历集合的方式完成参数更新，符合Pyomo的API规范
• 修复了While循环的缩进错误，确保迭代逻辑正确执行
• 添加了求解器状态检查，避免使用无效的求解结果
• 增加了迭代过程的打印输出，便于调试与监控

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