如何在Python CPLEX API中使用MIP回调记录MILP问题的可行解

我刚好做过类似的需求，原生CPLEX API里通过自定义MIP回调类，完全可以实时捕获并保存求解过程中的可行解，避免长时间运行中途中断导致数据丢失。下面是具体的实现步骤和代码：

1. 定义自定义回调类 #

我们需要继承cplex.callbacks.MIPInfoCallback——这个回调类专门用于MIP求解过程中的信息回调，能让我们访问到每一个新找到的可行解。

import cplex
import json

class FeasibleSolutionLogger(cplex.callbacks.MIPInfoCallback):
    def __init__(self, env, log_file_path="feasible_solutions.log", is_minimize=True):
        # 调用父类初始化方法
        super().__init__(env)
        # 初始化最优解记录（根据目标类型设置初始值）
        self.best_obj = float("inf") if is_minimize else float("-inf")
        self.is_minimize = is_minimize
        # 打开日志文件，用追加模式避免覆盖历史数据
        self.log_file = open(log_file_path, "a", encoding="utf-8")
        # 可选：记录初始化信息
        self.log_file.write(f"=== 求解开始时间: {cplex.util.get_time()} ===\n")
        self.log_file.flush()

    def invoke(self):
        # 检查是否有新的可行解可用
        if self.get_num_incumbent() > 0:
            current_obj = self.get_incumbent_objective_value()
            # 只记录比当前最优解更优的解（避免重复记录同一水平的解）
            if (self.is_minimize and current_obj < self.best_obj) or (not self.is_minimize and current_obj > self.best_obj):
                self.best_obj = current_obj
                # 获取所有变量的取值（顺序与模型中变量定义顺序一致）
                var_values = self.get_incumbent_values()
                # 获取变量名称，方便后续对应取值
                var_names = self.get_problem_data().variables.get_names()
                # 整理解信息
                solution_data = {
                    "timestamp": cplex.util.get_time(),
                    "objective_value": current_obj,
                    "variables": dict(zip(var_names, var_values))
                }
                # 写入文件，用JSON格式便于后续解析
                self.log_file.write(json.dumps(solution_data) + "\n")
                # 强制刷新缓冲区，确保数据实时写入磁盘
                self.log_file.flush()

    def __del__(self):
        # 对象销毁时关闭文件，避免资源泄漏
        if hasattr(self, "log_file") and not self.log_file.closed:
            self.log_file.write(f"=== 求解结束/中断时间: {cplex.util.get_time()} ===\n")
            self.log_file.close()

2. 注册回调并启动求解 #

在你构建好CPLEX模型之后，把自定义回调注册到模型中，然后启动求解即可：

# 假设你已经完成了模型的构建（变量、约束、目标函数等）
model = cplex.Cplex()
# ... 这里是你的模型构建代码 ...

# 注册回调
model.register_callback(FeasibleSolutionLogger)

# 启动求解（即使中途中断，已经找到的可行解也已经写入文件）
try:
    model.solve()
except KeyboardInterrupt:
    print("求解被手动中断，已记录的可行解保存在日志文件中")
except Exception as e:
    print(f"求解过程中出现错误: {e}")

关键细节说明 #

• 实时写入保障：一定要调用flush()方法，否则数据可能留在内存缓冲区，中途中断时不会写入磁盘。
• 只记录更优解：通过对比best_obj，我们只保留比当前最优解更好的可行解，避免日志文件过大。
• 变量名称对应：get_incumbent_values()返回的是变量值列表，顺序和模型中变量添加的顺序一致，用get_names()可以把值和变量名对应起来，方便后续分析。
• 资源清理：通过__del__方法在回调对象销毁时关闭文件，避免资源泄漏；如果是手动中断，也能保证文件正常关闭。
• 性能平衡：如果你的模型会产生大量可行解，可以适当调整记录逻辑，比如每N个解写入一次，或者只记录目标值变化超过阈值的解，减少IO开销对求解速度的影响。

内容的提问来源于stack exchange，提问作者Nguyen Quang Anh