Colossus框架使用指南
Colossus — 通用列生成求解框架
作者:syuansheng(大连海事大学)| 一个基于 Gurobi 的轻量通用列生成(Column Generation)框架
1. 这是什么
列生成是求解”变量数量巨大、但绝大多数用不上”的(混合)整数规划问题的经典方法。思路只有三步,反复循环:
- 限制主问题(RMP):用当前已有的列(变量)组成一个小 LP,求解,得到每条约束的对偶值;
- 定价子问题(Pricing SP):根据对偶值,找一条”加入后能让目标值下降”的新列(改善列);
- 找到 → 加进 RMP,回到第 1 步;找不到 → 已经最优,结束。
Colossus 把这个循环写死了。你只需要写”和你的问题绑定”的 3 个函数,剩下(迭代控制、对偶提取、列池与模型同步、终止判断、结果输出)全部由框架完成:
| 你要写的 | 框架替你做的 |
|---|---|
generateInitialCols():造初始列 |
求解 RMP,提取对偶值 y |
buildRestrictedMP(columnpool):建 RMP 模型 |
新列自动加入模型、删列自动同步 |
updateAndSolveSP_Heuristic/Exact(y, fingerprint, omega_next_col):解定价子问题 |
迭代控制、终止判断、输出表格和文件 |
2. 安装
pip install gurobipy numpy wcwidth
需要 Gurobi 求解器及许可证(学生/学术可免费申请)。把项目目录加入 sys.path 后即可使用:
from Colossus import ToolBox, Environment, Column, ColumnPool, Engine
3. 上手:跑通示例
cd Colossus/examples
python CSP.py
示例是切割下料问题(Cutting Stock):原材料长 L=1000,有 15 种需求(长度 l、数量 d),问最少用几根原料。运行后控制台输出迭代表格(每行一轮迭代,数值仅为示意):
+-----+---------+---------+--------------+----------+------------+----------+------------+
| Iter| Time(s)| Cols| RMP Obj| NoBetter| RMP Time| Method | SPs Time|
+-----+---------+---------+--------------+----------+------------+----------+------------+
| 1| 0.52| 100| 1,234.56| 0| 0.10|heuristic | 0.42|
| 2| 0.44| 145| 1,230.00| 1| 0.08|heuristic | 0.35|
| 3| 0.40| 168| 1,229.99| 2| 0.09| exact | 0.31|
+-----+---------+---------+--------------+----------+------------+----------+------------+
4. 照葫芦画瓢:跟着 CSP 写你的问题
下面把 CSP 的代码拆开讲。你写自己的问题时,照这个模板把内容换成你的问题即可。
4.1 先定义你的数据
I = [1,2,3,4,5,6,7,8,9,10,11,12,13,14,15] # 15 种需求
l = {1:80, 2:95, ..., 15:850} # 每种需求的长度
d = {1:5, 2:6, ..., 15:3} # 每种需求的数量
L = 1000 # 每根原材料的长度
4.2 generateInitialCols() —— 造初始列
干什么:给列生成一个”起步列池”。列池里至少要有一组能覆盖全部需求的可行列,否则 RMP 无解。这里为每种需求造一列”整根原料只切这一种料”的初始方案。
要返回:一个 ColumnPool 对象,里面每个 Column 必须填三个属性:
| 属性 | 含义 | CSP 例子 |
|---|---|---|
omega |
列编号(int,必须唯一) | 1, 2, ..., 15(每种需求一列,直接沿用需求编号) |
tau |
这列在目标函数里的系数 | 每根原料成本都是 1,所以 tau=1 |
coefficients |
这列在各约束里的系数,顺序必须和后面建模型时加约束的顺序一致 | [12,0,0,...] 表示”切 12 件需求 1” |
def generateInitialCols():
columnpool = ColumnPool()
for idx in I: # 每种需求造一列初始方案:整根原料只切这一种料
column = Column()
column.omega = idx # 列编号(示例直接沿用需求编号,只要唯一即可)
column.tau = 1 # 目标系数:一根原料
column.coefficients = [floor(L / l[i]) if i == idx else 0 for i in I]
columnpool.addCol(column)
return columnpool
说明:
examples/CSP.py中的正式实现为了覆盖”需求数量超过一根原料能切的数量”的情况,会为同一需求生成多列(小算例共 34 列),原理与本例相同。初始列不要设reduced_cost(保持None),它只属于定价子问题产出的列。
4.3 buildRestrictedMP(columnpool) —— 建限制主问题模型
干什么:把初始列装进一个 Gurobi LP 模型,返回 Model 对象。模型只包含初始列,以后的新列由框架自动加进去,你不用管。
两个必须遵守的规则:
- 变量命名:每个列变量名必须是
relaxed_variables_name[0] + '[' + str(omega) + ']',例如labda[1]。框架靠这个格式在模型里查找/删除列。 - 约束写成
>=(集覆盖):不要用<=。这样对偶值更稳定,列生成不容易震荡。CSP 里每条约束是”需求 i 的总生产量 ≥ 需求量 d[i]”。
def buildRestrictedMP(columnpool):
rmp_model = Model()
labda = rmp_model.addVars(
[omega for omega in range(1, columnpool.col_num + 1)],
vtype=GRB.CONTINUOUS, name='labda') # 变量名自动是 labda[1], labda[2], ...
rmp_model.setObjective(labda.sum('*'), sense=GRB.MINIMIZE) # 目标:原料根数最少
rmp_model.addConstrs(
(quicksum(a[omega, i] * labda[omega] for omega in range(1, columnpool.col_num + 1)) >= d[i] for i in I),
name='cons_rmp')
return rmp_model
4.4 updateAndSolveSP_Heuristic / _Exact(...) —— 解定价子问题(核心)
干什么:根据 RMP 当前的对偶值,找一条”加入后能让目标值下降”的新列。CSP 的定价子问题恰好是一个背包问题:把对偶值当”物品价格”、需求长度当”物品重量”,在容量 L 的背包里装出最大价值。
三个参数逐个讲:
| 参数 | 是什么 | 具体例子 |
|---|---|---|
y |
RMP 所有约束的对偶值,numpy 数组。顺序 = 你建模型时加约束的顺序 |
CSP 有 15 条约束(每种需求一条),所以 y 长度 15;y[i-1] 就是需求 i 那条约束的对偶值(”每多切一件需求 i 能省下多少根原料”) |
fingerprint |
当前要定价的子问题的标识,来自 Environment.fingerprint_lst |
CSP 只有一个子问题,fingerprint_lst=[1],传进来的恒为 1,用不上。如果你的问题有多个子问题(如 VRP 每个客户集合一个),框架会每轮按顺序把每个标识传进来,你根据它决定解哪个 |
omega_next_col |
下一个新列应该用的编号(框架的列池计数器维护好的) | 初始列用到 15,那这里就是 16,直接赋给新列 omega 即可 |
返回值:改善列的 list(没有就返回空列表 [])。每个改善列填 4 个属性,其中 reduced_cost 是关键:
reduced_cost(检验数):把这列加入 RMP 后,目标值能下降多少。最小化问题里为负就是改善列。CSP 中:原料成本为 1,背包价值为obj,所以reduced_cost = 1 - obj;例如背包解出obj=1.3,则reduced_cost=-0.3,说明加入这根”混合切割方案”能让总根数再降 0.3。
def updateAndSolveSP_Exact(y, fingerprint, omega_next_col):
improve_column_lst = []
item_price = {i: y[i-1] for i in I} # 对偶值 → 物品价格
item_weight = {i: l[i] for i in I} # 长度 → 物品重量
solution, obj = _solve_knapsack(L, I, item_price, item_weight) # 动态规划解背包(见示例文件)
if (1 - obj) > -1e-6: # reduced_cost 不为负 → 没有改善列
return improve_column_lst
column = Column()
column.omega = omega_next_col # 用框架给的编号
column.reduced_cost = 1 - obj # 检验数
column.tau = 1 # 目标系数
column.coefficients = list(solution.values()) # 这个切割方案每种需求切几件
improve_column_lst.append(column)
return improve_column_lst
updateAndSolveSP_Heuristic 签名和返回值完全一样,只是求解算法换成贪婪策略(优先装单位价值高的物品),代码见示例文件。
框架只认”返回的 list 是不是空”来判断有没有改善列,不检查符号,所以”算不算改善”由你自己在函数里判断。
4.5 coefficients_formatter(coefficients) —— 可选
开了 col_file_output_on=True 后,框架会把每个改善列写进 columns/Column[omega].col 文件,这时必须注册本函数,负责把 coefficients 变成可读字符串:
def coefficients_formatter(coefficients):
return ''.join('a[{}]={}\n'.format(i, coefficients[i-1]) for i in I)
4.6 组装起来,运行
toolbox = ToolBox()
toolbox.addTool('generateInitialCols', generateInitialCols)
toolbox.addTool('buildRestrictedMP', buildRestrictedMP)
toolbox.addTool('updateAndSolveSP_Exact', updateAndSolveSP_Exact)
toolbox.addTool('updateAndSolveSP_Heuristic', updateAndSolveSP_Heuristic)
toolbox.addTool('coefficients_formatter', coefficients_formatter)
environment = Environment(['labda'], ['INTEGER'], [1])
# 三个位置参数:被松弛的变量名 / 松弛前类型(INTEGER或BINARY) / 子问题标识列表
environment.result_file_output_on = True
engine = Engine('csp_engine', environment, toolbox)
engine.runEngine()
Engine 构造时会打印当前配置,运行前检查一下即可。结束后结果在 engine.result 里。
5. Environment 参数速查
| 参数 | 默认值 | 一句话说明 |
|---|---|---|
relaxed_variables_name |
必填 | 被松弛的变量名列表,第一个就是要动态生成的列(如 ['labda']),名字首字母别相同 |
relaxed_variables_type |
必填 | 对应变量松弛前的类型:'INTEGER' 或 'BINARY',列生成结束后恢复 |
fingerprint_lst |
必填 | 所有定价子问题的标识列表(如 [1]),框架按顺序逐个定价 |
heuristic_solve_on / exact_solve_on |
True / True |
是否用启发式 / 精确解定价。两个都开 = 先用启发式、不行再切精确解兜底 |
mini_batch_percent |
1.0 |
每轮有 round(该比例×子问题数) 个子问题找到改善列就提前结束本轮定价 |
dual_alpha |
0 |
对偶平滑系数:y=α·y_上轮+(1-α)·y_本轮,0=不平滑。目标值震荡时设 0.3~0.5 |
truncate_threshold |
1e-6 |
相邻两轮最优值差小于它就算”没改进” |
max_not_bertter |
inf |
连续多少轮没改进就终止(混合定价时先切精确解再判断) |
max_time_limit |
3600.0 |
迭代总耗时上限(秒) |
barrier_method_on |
True |
是否用内点法求解 RMP(对偶值通常更稳) |
lp_file_ouput_on / ilp_file_output_on |
True / True |
是否输出 .lp 模型 / 不可行时的 .ils 文件 |
col_file_output_on |
True |
是否把改善列写入 columns/Column[omega].col(需注册 coefficients_formatter) |
result_file_output_on |
False |
是否把最终结果写入 {engine_name}.result |
rmp_log_to_console |
False |
是否在控制台打印 Gurobi 求解日志 |
table_header_language |
'ascii' |
迭代表格表头:'ascii'(英文,任何终端对齐)/ 'chinese'(中文,需终端中文为双宽) |
6. 输出怎么看
迭代表格(每轮一行):
| 列 | 含义 |
|---|---|
Iter 轮次 / Time(s) 本轮耗时 / Cols 当前列数 |
迭代进度 |
RMP Obj 本轮 RMP 最优值 / NoBetter 连续未改进次数 |
收敛情况 |
RMP Time RMP 求解时长 / SPs Time 子问题求解时长 / Method 定价方式(heuristic/exact) |
耗时分布 |
engine.result(Result 对象)关键属性:
best_objective:最终主问题(整数规划)最优值;total_time:总耗时;history_incumbent:每轮 RMP 最优值列表(可画收敛曲线);history_rmp_time/history_sps_time:每轮耗时。
生成的文件(写入当前工作目录):initial/final_rmp_model.lp(模型)、rmp_model.ils(不可行分析)、columns/Column[omega].col(改善列明细)、{engine_name}.result(一行最终结果:最优值 总耗时 平均RMP耗时 平均子问题耗时 列数量 主问题耗时)。
7. 常见问题
- 开了
col_file_output_on却没有.col文件:忘了注册coefficients_formatter。 - 迭代很久不收敛 / 目标值来回跳:试试
dual_alpha=0.3~0.5;确认主问题约束是>=(集覆盖);确认coefficients顺序与约束顺序一致。
8. 开源许可
开发中,还未开源~







