Table of Contents
OptAgent — 智能优化求解器框架¶
统一建模 · 自动策略选择 · 多求解路径
OptAgent 是一个智能优化求解器框架,提供统一的建模接口,自动识别问题结构并选择求解策略,支持启发式、精确求解(CP-SAT / MILP)以及混合求解。
文档阅读路径
- 第一次访问:从“使用指南”进入
- 想看对象字段:继续读“API 参考”
- 想看完整问题:进入“示例”,再跳转到公开示例仓库
核心特性¶
| 特性 | 描述 |
|---|---|
| 统一建模 API | 一套接口描述线性、调度、黑盒等各类优化问题 |
| 自动策略选择 | 可自动推荐内置求解策略,也可通过 API 显式配置 |
| 精确解兼容说明 | API 参考标明哪些建模接口可走 CP-SAT 或 MILP |
| 公开示例入口 | 示例栏目链接到公开示例仓库和可运行问题说明 |
| 统一输出 | 启发式、CP-SAT、MILP 和混合路径输出统一结果对象 |
使用指南¶
本栏目面向首次接入 OptAgent 的用户,按“安装 -> 建模 -> 求解 -> 配置策略 -> 查看结果”的顺序组织公开使用文档。
推荐阅读顺序¶
- 安装指南:确认支持环境,获取安装包并完成安装。
- 建模接口:用
ModelBuilder描述变量、约束和目标。 - 求解接口:用
Orchestrator().run(...)发起求解。 - 策略配置:在自动选择、内置预设和显式配置之间做选择。
- 结果查看:读取
final_solution、阶段 trace 与诊断信息。
最短闭环¶
from optagent import ModelBuilder, Orchestrator
# 1. 创建 builder,并用 metadata 标记案例,方便后续 trace 和日志识别。
builder = ModelBuilder(metadata={"case": "minimal_usage"})
# 2. 创建两个 0/1 决策变量,表示是否选择对应方案。
x = builder.int_var(default=0, lb=0, ub=1, name="x")
y = builder.int_var(default=0, lb=0, ub=1, name="y")
# 3. 登记容量约束:两个方案最多选择一个。
builder.constraint(x + y <= 1, name="capacity")
# 4. 登记收益最大化目标。
builder.maximize((x * 3) + (y * 2), name="profit")
# 5. 冻结模型,得到可交给编排器求解的 program。
program = builder.freeze()
# 6. 运行编排器;不传配置时会自动选择内置策略。
result = Orchestrator().run(program)
solution = result.final_solution
# 7. 输出最终状态、目标值和变量赋值。
print(solution.status.value)
print(solution.objective_values)
print(solution.variable_values)
与其他栏目关系¶
使用指南
安装指南¶
本页说明 OptAgent wheel 安装包的适用环境、获取方式与安装步骤。
支持的环境¶
- Linux
- macOS
- Windows
Python 环境要求¶
- Python 3.9 及更高版本
- 页面中的命令示例默认使用当前虚拟环境的 Python 作为演示基线
- 建议使用独立虚拟环境安装
支持的安装工具¶
pipuv
安装包获取¶
OptAgent 当前发布形式为 wheel 文件。
如需安装包,请联系项目经理获取当前版本的 .whl 文件。
安装步骤¶
- 确认当前操作系统为 Linux、macOS 或 Windows,且本机 Python 版本为 3.9 或更高版本。
- 联系项目经理获取当前发布的 wheel 文件,例如
optagent-0.2-py3-none-any.whl。 - 在目标环境中准备 Python 虚拟环境。
- 使用
pip或uv安装 wheel 文件。
使用 pip 安装¶
# 1. 使用当前 Python 环境的 pip 安装本地 wheel 文件。
python -m pip install /path/to/optagent-0.2-py3-none-any.whl
使用 uv 安装¶
额外依赖安装¶
如果需要安装带额外依赖的本地 wheel,可使用 extras 形式安装:
# 1. 安装 OR-Tools 相关额外依赖,适合 CP-SAT 路径。
python -m pip install '/path/to/optagent-0.2-py3-none-any.whl[ortools]'
# 2. 安装 HiGHS 相关额外依赖,适合 HiGHS 原生 MILP 路径。
python -m pip install '/path/to/optagent-0.2-py3-none-any.whl[highs]'
# 3. 同时安装 OR-Tools 和 HiGHS 额外依赖。
python -m pip install '/path/to/optagent-0.2-py3-none-any.whl[ortools,highs]'
安装后验证¶
安装完成后,可执行以下命令确认包已成功安装:
说明¶
pip与uv都支持从本地 wheel 文件安装。- 安装命令适用于 Python 3.9 及更高版本。
- 上方
Wheel Python Matrix展示的是当前 wheel 已验证的 Python 版本范围。 - 请优先使用项目经理提供的最新版 wheel 文件。
建模接口¶
OptAgent 使用 ModelBuilder 描述优化问题。建模阶段的目标是把业务问题转成一个可执行的 program,后续求解器和编排器都围绕这个 program 工作。
基本流程¶
- 创建
ModelBuilder。 - 添加变量。
- 用表达式组合约束和目标。
- 调用
freeze()生成可求解的program。
from optagent import ModelBuilder
# 1. 创建建模入口,并记录案例名。
builder = ModelBuilder(metadata={"case": "knapsack_small"})
# 2. 准备业务数据:每个物品的重量和价值。
weights = [2, 3, 4]
values = [4, 5, 7]
# 3. 为每个物品创建一个 0/1 决策变量,1 表示选择该物品。
picks = [
builder.int_var(default=0, lb=0, ub=1, name=f"pick_{idx}")
for idx in range(len(weights))
]
# 4. 用表达式计算总重量和总价值。
total_weight = builder.sum(*(pick * weight for pick, weight in zip(picks, weights, strict=True)))
total_value = builder.sum(*(pick * value for pick, value in zip(picks, values, strict=True)))
# 5. 登记背包容量约束和收益最大化目标。
builder.constraint(total_weight <= 5, name="capacity")
builder.maximize(total_value, name="profit")
# 6. 冻结为可求解的 ExecutableProgram。
program = builder.freeze()
常用变量¶
| 方法 | 用途 |
|---|---|
bool_var(...) |
0/1 或布尔决策 |
int_var(...) |
整数决策,可设置 lb / ub |
float_var(...) |
连续数值决策 |
set_var(...) |
集合型选择 |
sequence_var(...) |
排列、路径、访问顺序 |
interval_var(...) |
调度任务的开始、结束、长度 |
表达式与约束¶
变量和常量可以直接组合为表达式:
# 1. 组合变量和常量,得到一个成本表达式。
cost = (x * 3) + (y * 5)
# 2. 调用 constraint 后,该表达式才会成为正式约束。
builder.constraint(cost <= 20, name="budget")
常见表达式工具包括:
| 方法 | 用途 |
|---|---|
sum(...) |
多个表达式求和 |
min(...) / max(...) |
聚合最小值或最大值 |
abs(...) |
绝对值 |
all_different(...) |
互异约束 |
element(array, index) |
根据索引取数组值 |
external_call(...) |
接入黑盒目标或外部函数 |
约束是单独登记的¶
表达式本身只是在图中构造条件,只有调用 constraint(...) 后才会被注册为正式约束:
# 1. 登记容量约束。
builder.constraint(total_weight <= 5, name="capacity")
# 2. 登记二选一约束。
builder.constraint(x + y == 1, name="choose_one")
如果只写了 total_weight <= 5,但没有传给 constraint(...),这个条件不会进入模型的约束列表。
目标函数¶
使用 minimize(...) 或 maximize(...) 声明目标:
# 1. 最小化总成本。
builder.minimize(total_cost, name="total_cost")
# 2. 或最大化总收益。
builder.maximize(total_profit, name="profit")
一个模型可以记录多个目标,但实际求解能力取决于当前求解路径和后端支持范围。公开示例中建议先从单目标问题开始。
特殊建模方法¶
除了通用数值建模,ModelBuilder 还提供了几类针对优化结构的专用方法。
集合与序列¶
| 方法 | 用途 |
|---|---|
set_len(set_expr) |
集合大小 |
set_contains(set_expr, value_expr) |
元素是否在集合中 |
sequence_contains(sequence_expr, value_expr) |
元素是否在序列中 |
all_different(...) |
互异约束 |
table(exprs, allowed_tuples) |
允许元组表约束 |
调度¶
| 方法 | 用途 |
|---|---|
interval_start(interval_expr) |
读取 interval 开始时间 |
interval_end(interval_expr) |
读取 interval 结束时间 |
interval_length(interval_expr) |
读取 interval 长度 |
cumulative(intervals, demands, capacity) |
累积资源约束 |
no_overlap(sequence_expr, *interval_exprs) |
不重叠约束 |
precedence(before, after, lag=0) |
前后工序约束 |
条件与逻辑¶
| 方法 | 用途 |
|---|---|
and_(lhs, rhs) |
逻辑与 |
or_(lhs, rhs) |
逻辑或 |
not_(expr) |
逻辑非 |
iif(cond, when_true, when_false) |
条件表达式 |
黑盒与外部函数¶
| 方法 | 用途 |
|---|---|
external_call(fn, *inputs, ...) |
将 Python 函数登记为外部调用节点 |
def score(order):
# 1. 外部函数接收求解器给出的候选序列,并返回评分。
return len(order)
# 2. route 是待优化的排列型决策变量。
route = builder.sequence_var(5, name="route")
# 3. external_call 把 Python 函数接入模型图。
blackbox_score = builder.external_call(score, route, name="route_score")
# 4. 将黑盒评分登记为目标。
builder.minimize(blackbox_score, name="score")
freeze() 的含义¶
freeze() 会把当前 builder 中的图结构编译为 ExecutableProgram:
生成 program 后,建模图会被冻结,后续应把它交给 Orchestrator().run(...) 求解。
完整公开方法范围¶
当前公开建模接口可按下面分组理解:
| 分组 | 方法 |
|---|---|
| 常量与变量 | const, bool_var, int_var, float_var, set_var, sequence_var, interval_var |
| 聚合与逻辑 | abs, sum, min, max, and_, or_, not_, iif |
| 集合与序列 | set_len, set_contains, sequence_contains, all_different, element, table |
| 调度 | interval_start, interval_end, interval_length, cumulative, no_overlap, precedence |
| 黑盒 | external_call |
| 模型登记 | constraint, minimize, maximize |
| 输出 | to_program_spec, freeze |
下一步¶
- 发起求解:阅读 求解接口。
- 查完整对象成员:阅读 Builder API。
- 看完整线性示例:从 示例入口 跳转到公开示例仓库。
求解接口¶
OptAgent 的公开求解入口是 Orchestrator().run(...)。它接收建模阶段生成的 program,返回 OrchestratorResult。
最简单的求解方式¶
不传 preset 或 config 时,系统会根据问题特征自动选择内置策略:
from optagent import Orchestrator
# 1. 使用建模阶段生成的 program 发起求解。
result = Orchestrator().run(program)
# 2. final_solution 是读取状态、目标值和变量值的主入口。
solution = result.final_solution
这种方式适合快速验证模型是否能跑通。
使用内置预设¶
如果你知道问题类型,可以显式选择内置策略:
from optagent import BuiltInStrategyPreset, Orchestrator
# 1. 显式选择内置预设,避免每次都走自动推荐。
result = Orchestrator().run(
program,
preset=BuiltInStrategyPreset.SCHEDULING_EVOLUTIONARY_REPAIR,
)
也可以使用字符串形式:
使用显式配置¶
需要固定求解器、后端或阶段时,传入 OrchestratorConfig:
from optagent import (
ExactBackendName,
Orchestrator,
OrchestratorConfig,
OrchestratorSolver,
PhaseConfig,
)
# 1. 构造显式配置,强制使用 HiGHS 原生后端。
# 2. strict_backend=True 表示后端不满足时直接失败,而不是静默降级。
result = Orchestrator().run(
program,
config=OrchestratorConfig(
required_backend=ExactBackendName.HIGHS_NATIVE,
strict_backend=True,
phases=[
# 3. 定义一个 MILP 阶段,并给出阶段预算。
PhaseConfig(
name="milp_phase",
solver=OrchestratorSolver.MILP,
budget_iterations=20,
)
],
),
)
preset 和 config 不能同时传入。预设适合标准场景,显式配置适合需要固定求解路径的场景。
常见求解器类型¶
OrchestratorSolver |
用途 |
|---|---|
HEURISTIC |
启发式求解,适合大规模组合、黑盒或快速近似 |
CP_SAT |
CP-SAT 精确求解,适合调度和组合约束 |
MILP |
MILP 精确求解,适合线性、整数规划 |
如果需要固定具体后端,可使用 ExactBackendName:
ExactBackendName |
含义 |
|---|---|
CP_SAT_NATIVE |
OR-Tools CP-SAT 路径 |
MATHOPT_MP |
MathOpt 桥接路径 |
HIGHS_NATIVE |
HiGHS 原生路径 |
求解结果入口¶
run(...) 返回的结果对象包含:
| 字段 | 含义 |
|---|---|
final_solution |
最终统一解对象 |
solver_traces |
求解器阶段记录 |
phase_traces |
编排阶段记录 |
diagnostics |
后端或求解过程诊断 |
selected_preset_name |
实际选中的预设名 |
selected_preset_source |
预设来源 |
下一步¶
- 查看解对象字段:阅读 结果查看。
- 学习策略选择:阅读 策略配置。
- 查完整对象成员:阅读 Orchestrator API。
策略配置¶
OptAgent 提供三层策略入口:自动选择、内置预设、显式配置。选择哪一层,取决于你是否需要控制求解路径。
三种入口¶
| 入口 | 写法 | 适用场景 |
|---|---|---|
| 自动选择 | Orchestrator().run(program) |
快速试跑、初次接入 |
| 内置预设 | Orchestrator().run(program, preset=...) |
已知道问题类型,想复用推荐策略 |
| 显式配置 | Orchestrator().run(program, config=...) |
需要固定求解器、后端、阶段和预算 |
自动选择¶
自动选择会分析 program 的变量、约束、目标和结构特征,然后选择一个可用预设:
# 1. 自动选择会先分析 program,再选一个匹配的内置预设。
result = Orchestrator().run(program)
# 2. 读取实际选中的预设,确认求解路径。
print(result.selected_preset_name)
print(result.selected_preset_source)
适合用来验证模型和获得第一版结果。
内置预设¶
内置预设可以通过枚举或字符串指定:
from optagent import BuiltInStrategyPreset, Orchestrator
# 1. 枚举写法适合业务代码长期维护。
result = Orchestrator().run(
program,
preset=BuiltInStrategyPreset.LINEAR_EXACT_FIRST,
)
常见预设族包括:
| 预设族 | 典型场景 |
|---|---|
scheduling_* |
调度问题,通常包含 interval_var 或顺序约束 |
sequence_* |
路径、排列、访问顺序 |
routing_* |
路由或序列黑盒目标 |
linear_* |
线性、整数规划 |
general_* |
通用启发式或进化式路径 |
hybrid_* |
启发式与精确求解组合 |
显式配置¶
当你需要控制阶段、预算或后端时,使用 OrchestratorConfig:
from optagent import OrchestratorConfig, OrchestratorSolver, PhaseConfig
# 1. 总预算控制整次编排的粗粒度运行成本。
config = OrchestratorConfig(
total_budget_iterations=80,
phases=[
# 2. 第一阶段用启发式快速产生可行解。
PhaseConfig(
name="first_pass",
solver=OrchestratorSolver.HEURISTIC,
budget_iterations=40,
),
# 3. 第二阶段用 MILP 在已有解基础上精化。
PhaseConfig(
name="exact_polish",
solver=OrchestratorSolver.MILP,
budget_iterations=40,
),
],
)
# 4. 显式配置会固定求解路径,不再走自动预设推荐。
result = Orchestrator().run(program, config=config)
如何选择配置层级¶
建议按以下顺序决策:
- 第一次跑模型:使用自动选择。
- 已确认问题类型:使用内置预设。
- 需要复现实验或固定后端:使用显式配置。
- 需要多阶段组合:使用显式
PhaseConfig。 - 需要调启发式内部行为:再进入
HeuristicOrchestrationConfig等高级配置。
注意事项¶
preset与config是互斥参数,不能同时传入。- 预设会携带完整编排配置和执行计划。
- 显式配置会覆盖预设路径,适合可控性优先的场景。
- 自动选择的结果可通过
selected_preset_name和selected_preset_source查看。
下一步¶
- 查看预设对象字段:阅读 Presets API。
- 查预设对象:阅读 Presets API。
- 查看求解阶段配置:阅读 求解接口。
结果查看¶
Orchestrator().run(...) 返回 OrchestratorResult。日常使用时,最常读取的是 result.final_solution。
最常用字段¶
# 1. final_solution 是业务读取最终解的主入口。
solution = result.final_solution
# 2. 先看求解器、状态和可行性。
print(solution.solver_name)
print(solution.status.value)
print(solution.feasible)
# 3. 再看目标值和变量赋值。
print(solution.objective_values)
print(solution.variable_values)
| 字段 | 含义 |
|---|---|
solver_name |
产出最终解的求解器或阶段 |
status |
求解状态 |
feasible |
是否满足约束 |
objective_values |
目标函数值,按目标节点 id 组织 |
variable_values |
变量取值,按变量节点 id 组织 |
metadata |
后端或求解器附加信息 |
读取变量名¶
variable_values 以内部节点 id 为 key。建模时建议给变量设置 name,这样在调试时更容易把节点和业务含义对应起来:
如果需要把节点 id 映射回名称,可以结合建模阶段保存的变量引用,或在调试时查看 program 图中的节点 metadata。
查看预设选择¶
自动选择或预设求解后,可以查看实际命中的预设:
# 1. 查看自动推荐或显式预设最终命中了哪个 preset。
print(result.selected_preset_name)
print(result.selected_preset_source)
常见来源包括显式字符串、枚举预设或自动推荐。
查看阶段记录¶
多阶段求解或混合求解时,建议查看 trace:
# 1. 遍历每个顶层求解阶段的摘要。
for trace in result.solver_traces:
print(trace.phase_name, trace.solver_name, trace.status, trace.feasible)
常用记录包括:
| 字段 | 含义 |
|---|---|
solver_traces |
每个求解阶段的结果摘要 |
phase_traces |
阶段合并和状态交接记录 |
heuristic_subphase_traces |
启发式内部子阶段记录 |
evolutionary_generation_traces |
进化式策略的代际记录 |
diagnostics |
后端选择、降级或求解诊断 |
判断结果是否可用¶
通常按以下顺序检查:
solution.feasible是否为True。solution.status.value是否符合预期。objective_values是否存在目标值。solver_traces是否有跳过、降级或失败记录。diagnostics是否包含后端不可用或求解失败信息。
下一步¶
- 发起求解:阅读 求解接口。
- 调整策略:阅读 策略配置。
- 查完整返回对象:阅读 Orchestrator API。
API 参考¶
本栏目提供当前公开 API 的对象参考。首次接入时,建议先阅读 使用指南,再回到本栏目查对象成员和配置字段。
阅读入口¶
- Builder API:
ModelBuilder建模入口。 - Orchestrator API:求解入口、阶段配置、结果对象。
- Presets API:问题特征、内置预设与策略对象。
- 精确求解器接口兼容性:CP-SAT 和 MILP 分别支持的建模接口。
API 参考
Builder API¶
ModelBuilder 是公开建模入口。教程式说明见 建模接口。
本页标题只显示方法名,便于右侧目录快速定位。每个方法下面给出签名、参数、返回值和带注释示例。
Exact Solver Compatibility¶
ModelBuilder 是统一建模入口,但精确解后端不是通用解释器。CP-SAT 和 MILP 只接收各自可降阶的 DAG 子集;如果模型包含某个后端不支持的接口,需要改用启发式、混合策略,或调整建模形式。
按后端查看支持清单,见 精确求解器接口兼容性。
标签含义:
| 标签 | 含义 |
|---|---|
| HEURISTIC | 启发式路径可评估或搜索该接口表达的模型语义。 |
| CP-SAT | 当前 CP-SAT 精确路径可 lowering 到 cp_sat_native。 |
| MILP | 当前 MILP 精确路径可 lowering 到 mathopt_mp / highs_native。 |
| MODEL | 建模、登记或导出接口;不直接代表求解语义。 |
精确解支持矩阵:
| 接口 | 标签 | CP-SAT | MILP | 说明 |
|---|---|---|---|---|
const |
HEURISTIC CP-SAT MILP | 支持整数/布尔常量 | 支持数值/布尔常量 | CP-SAT 要求参与精确解的常量为整数或布尔值。 |
bool_var |
HEURISTIC CP-SAT MILP | 支持 | 支持 | MILP 会降为 0/1 变量。 |
int_var |
HEURISTIC CP-SAT MILP | 支持有界整数 | 支持 | CP-SAT 要求有限 lb / ub。 |
float_var |
HEURISTIC MILP | 不支持 | 支持 | CP-SAT 当前只构建整数域精确模型。 |
set_var |
HEURISTIC | 不支持 | 不支持 | 当前 exact canonical model 没有集合变量表示。 |
sequence_var |
HEURISTIC CP-SAT | 仅通过 no_overlap 绑定 interval 后支持 |
不支持 | CP-SAT 用 rank 变量表达序列;单独序列或黑盒路线不属于 exact 支持面。 |
interval_var |
HEURISTIC CP-SAT | 支持 | 不支持 | MILP 当前只做线性 MP lowering,不接收 interval 结构。 |
abs |
HEURISTIC CP-SAT | 支持 | 不支持 | MILP 当前未线性化绝对值,需要用户手动改写为线性约束。 |
sum |
HEURISTIC CP-SAT MILP | 支持 | 支持线性求和 | MILP 要求所有输入保持线性。 |
min / max |
HEURISTIC CP-SAT | 支持 | 不支持 | MILP 当前未自动引入辅助变量和 epigraph 约束。 |
and_ / or_ / not_ |
HEURISTIC CP-SAT | 支持布尔表达式 | 不支持 | MILP 当前不做逻辑重写或 big-M 线性化。 |
iif |
HEURISTIC | 不支持 | 不支持 | 两个 exact 路径都未实现条件表达式 lowering。 |
set_len / set_contains |
HEURISTIC | 不支持 | 不支持 | 依赖集合变量,当前 exact canonical model 不包含集合语义。 |
sequence_contains |
HEURISTIC | 不支持 | 不支持 | CP-SAT 当前 sequence 支持仅限 no_overlap 绑定的排序语义。 |
all_different |
HEURISTIC CP-SAT | 支持 | 不支持 | MILP 当前不自动构造 pairwise 或 assignment 线性化。 |
element / table |
HEURISTIC | 不支持 | 不支持 | 当前 exact backend 没有 element/table 原生或线性化实现。 |
interval_start / interval_end / interval_length |
HEURISTIC CP-SAT | 支持 | 不支持 | 只在 CP-SAT interval 模型中可用。 |
cumulative / no_overlap / precedence |
HEURISTIC CP-SAT | 支持 | 不支持 | 调度约束当前走 CP-SAT;MILP 不做调度结构线性化。 |
external_call |
HEURISTIC | 不支持 | 不支持 | 精确解后端不能调用任意 Python 黑盒函数并证明全局最优。 |
constraint / minimize / maximize |
MODEL | 取决于内部表达式 | 取决于内部表达式 | 登记接口本身可用,精确解兼容性由包裹的表达式决定。 |
to_program_spec / freeze |
MODEL | 可用 | 可用 | 输出模型结构;不改变 solver 支持范围。 |
不能只按变量类型选择精确解
即使模型用了 int_var,只要目标或约束里包含 external_call、iif、element、集合接口,当前精确解路径仍会拒绝 lowering。选择 CP-SAT 或 MILP 时,以完整表达式图是否兼容为准。
接口分组¶
| 分组 | 方法 |
|---|---|
| 常量与变量 | const、bool_var、int_var、float_var、set_var、sequence_var、interval_var |
| 数值与逻辑表达式 | abs、sum、min、max、and_、or_、not_、iif |
| 集合与序列 | set_len、set_contains、sequence_contains、all_different、element、table |
| 调度 | interval_start、interval_end、interval_length、cumulative、no_overlap、precedence |
| 黑盒 | external_call |
| 登记 | constraint、minimize、maximize |
| 输出 | to_program_spec、freeze |
const¶
标签: HEURISTIC CP-SAT MILP
签名:const(value: bool | int | float | list[int] | dict[str, int]) -> Expr
| 参数 | 类型 | 含义 |
|---|---|---|
value |
bool | int | float | list[int] | dict[str, int] |
要接入表达式图的常量值。 |
返回值:Expr,常量表达式节点。
# 1. 把业务中的固定罚分转成 builder 表达式。
penalty = builder.const(10)
# 2. 常量表达式可以和变量表达式一起构造约束。
builder.constraint(x + penalty <= 20, name="budget")
bool_var¶
标签: HEURISTIC CP-SAT MILP
签名:bool_var(default: bool = False, *, name: str | None = None) -> Expr
| 参数 | 类型 | 含义 |
|---|---|---|
default |
bool |
初始布尔值,默认 False。 |
name |
str | None |
变量名,用于输出、trace 和调试。 |
返回值:Expr,布尔决策变量。
# 1. 创建一个是否选择某个方案的 0/1 决策。
pick = builder.bool_var(default=False, name="pick_item")
# 2. 把业务规则登记成正式约束。
builder.constraint(pick == 1, name="must_pick")
int_var¶
标签: HEURISTIC CP-SAT MILP
签名:int_var(default: int = 0, *, lb: int | None = None, ub: int | None = None, name: str | None = None) -> Expr
CP-SAT 边界
CP-SAT 精确路径要求整数变量有有限 lb / ub。如果省略边界,建模仍然有效,但 CP-SAT lowering 会拒绝该模型。
| 参数 | 类型 | 含义 |
|---|---|---|
default |
int |
初始整数值。 |
lb |
int | None |
下界,None 表示不显式设置。 |
ub |
int | None |
上界,None 表示不显式设置。 |
name |
str | None |
变量名。 |
返回值:Expr,整数决策变量。
# 1. 建立一个范围为 [0, 10] 的整数变量。
quantity = builder.int_var(default=0, lb=0, ub=10, name="quantity")
# 2. 在后续表达式里使用该变量。
builder.constraint(quantity <= 6, name="stock_limit")
float_var¶
标签: HEURISTIC MILP
签名:float_var(default: float = 0.0, *, lb: float | None = None, ub: float | None = None, name: str | None = None) -> Expr
| 参数 | 类型 | 含义 |
|---|---|---|
default |
float |
初始连续值。 |
lb |
float | None |
下界。 |
ub |
float | None |
上界。 |
name |
str | None |
变量名。 |
返回值:Expr,连续决策变量。
# 1. 创建比例变量,限制在 0 到 1 之间。
ratio = builder.float_var(default=0.0, lb=0.0, ub=1.0, name="ratio")
# 2. 用比例变量表达连续成本。
builder.minimize(ratio * 100.0, name="scaled_cost")
set_var¶
标签: HEURISTIC
签名:set_var(size: int, *, default: list[int] | set[int] | None = None, name: str | None = None) -> Expr
精确解边界
当前 CP-SAT / MILP canonical model 都没有集合变量表示;集合模型通常需要走启发式,或手动改写为 0/1 变量和线性约束。
| 参数 | 类型 | 含义 |
|---|---|---|
size |
int |
候选元素数量,元素索引通常为 0..size-1。 |
default |
list[int] | set[int] | None |
初始选中集合。 |
name |
str | None |
变量名。 |
返回值:Expr,集合变量。
# 1. 表示从 5 个任务中选择一个子集,默认选择 0 和 2。
selected = builder.set_var(size=5, default={0, 2}, name="selected_jobs")
# 2. 限制最多选择 3 个任务。
builder.constraint(builder.set_len(selected) <= 3, name="max_selected")
sequence_var¶
标签: HEURISTIC CP-SAT
签名:sequence_var(size: int, *, default: list[int] | None = None, name: str | None = None) -> Expr
CP-SAT 边界
CP-SAT 当前只支持绑定到 no_overlap(sequence, *intervals) 的序列变量,并要求序列大小与 interval 数量一致。路线黑盒或单独序列约束仍应视为启发式语义。
| 参数 | 类型 | 含义 |
|---|---|---|
size |
int |
序列元素数量。 |
default |
list[int] | None |
初始排列;不传时默认为 list(range(size))。 |
name |
str | None |
变量名。 |
返回值:Expr,序列变量。
# 1. 用序列变量表示 4 个地点的访问顺序。
route = builder.sequence_var(size=4, name="route")
# 2. 要求地点 2 必须在路线中出现。
builder.constraint(builder.sequence_contains(route, 2) == 1, name="visit_2")
interval_var¶
标签: HEURISTIC CP-SAT
签名:interval_var(*, start: int = 0, length: int = 0, lb_start: int | None = None, ub_start: int | None = None, lb_length: int | None = 0, ub_length: int | None = None, name: str | None = None) -> Expr
| 参数 | 类型 | 含义 |
|---|---|---|
start |
int |
默认开始时间。 |
length |
int |
默认持续时间。 |
lb_start / ub_start |
int | None |
开始时间上下界。 |
lb_length / ub_length |
int | None |
持续时间上下界。 |
name |
str | None |
interval 名称。 |
返回值:Expr,调度 interval 变量。
# 1. 建立一个长度为 3、开始时间范围为 [0, 10] 的任务。
job = builder.interval_var(start=0, length=3, lb_start=0, ub_start=10, name="job_a")
# 2. 限制任务必须在时间 12 之前结束。
builder.constraint(builder.interval_end(job) <= 12, name="job_a_deadline")
abs¶
标签: HEURISTIC CP-SAT
签名:abs(expr: Expr | Any) -> Expr
MILP 边界
MILP 当前未自动把绝对值改写为辅助变量和线性约束。需要 MILP 精确解时,应在模型中显式线性化。
| 参数 | 类型 | 含义 |
|---|---|---|
expr |
Expr | Any |
要取绝对值的表达式或字面量。 |
返回值:Expr,绝对值表达式。
# 1. 计算变量距离目标值 5 的偏差。
distance = builder.abs(x - 5)
# 2. 最小化该偏差。
builder.minimize(distance, name="distance_to_target")
sum¶
标签: HEURISTIC CP-SAT MILP
签名:sum(*exprs: Expr | Any) -> Expr
| 参数 | 类型 | 含义 |
|---|---|---|
*exprs |
Expr | Any |
要求和的表达式或字面量。 |
返回值:Expr,求和表达式。
# 1. 把多个装载量变量聚合为总装载量。
total_load = builder.sum(load_a, load_b, load_c)
# 2. 登记容量约束。
builder.constraint(total_load <= 100, name="capacity")
min¶
标签: HEURISTIC CP-SAT
签名:min(*exprs: Expr | Any) -> Expr
| 参数 | 类型 | 含义 |
|---|---|---|
*exprs |
Expr | Any |
要取最小值的表达式或字面量。 |
返回值:Expr,最小值表达式。
# 1. 表示三个资源池中的最小剩余容量。
least_slack = builder.min(slack_a, slack_b, slack_c)
# 2. 最大化最小剩余容量,表达均衡目标。
builder.maximize(least_slack, name="balance_slack")
max¶
标签: HEURISTIC CP-SAT
签名:max(*exprs: Expr | Any) -> Expr
MILP 边界
min / max 当前不会自动降为 MILP 的辅助变量约束。需要 MILP 时,应手动写成线性 epigraph / hypograph 形式。
| 参数 | 类型 | 含义 |
|---|---|---|
*exprs |
Expr | Any |
要取最大值的表达式或字面量。 |
返回值:Expr,最大值表达式。
# 1. 找到多台机器里的峰值负载。
peak_load = builder.max(load_a, load_b, load_c)
# 2. 最小化峰值负载。
builder.minimize(peak_load, name="minimize_peak_load")
and_¶
标签: HEURISTIC CP-SAT
签名:and_(lhs: Expr | Any, rhs: Expr | Any) -> Expr
| 参数 | 类型 | 含义 |
|---|---|---|
lhs |
Expr | Any |
左侧逻辑表达式。 |
rhs |
Expr | Any |
右侧逻辑表达式。 |
返回值:Expr,逻辑与表达式。
# 1. 同时检查两个条件是否成立。
both_ready = builder.and_(machine_ready == 1, material_ready == 1)
# 2. 要求两个条件都成立后才能生产。
builder.constraint(both_ready == 1, name="ready_to_run")
or_¶
标签: HEURISTIC CP-SAT
签名:or_(lhs: Expr | Any, rhs: Expr | Any) -> Expr
| 参数 | 类型 | 含义 |
|---|---|---|
lhs |
Expr | Any |
左侧逻辑表达式。 |
rhs |
Expr | Any |
右侧逻辑表达式。 |
返回值:Expr,逻辑或表达式。
# 1. 至少选择一种加急方式。
fast_mode = builder.or_(use_air == 1, use_express == 1)
# 2. 把可选规则登记为约束。
builder.constraint(fast_mode == 1, name="need_fast_mode")
not_¶
标签: HEURISTIC CP-SAT
签名:not_(expr: Expr | Any) -> Expr
| 参数 | 类型 | 含义 |
|---|---|---|
expr |
Expr | Any |
要取反的逻辑表达式。 |
返回值:Expr,逻辑非表达式。
# 1. 表示不能同时禁用某个安全开关。
disabled = builder.not_(safety_enabled == 1)
# 2. 禁止 disabled 为真。
builder.constraint(disabled == 0, name="safety_required")
iif¶
标签: HEURISTIC
签名:iif(cond: Expr | Any, when_true: Expr | Any, when_false: Expr | Any) -> Expr
精确解边界
iif 当前没有 CP-SAT 或 MILP lowering。需要精确解时,应把条件逻辑改写为后端可接收的布尔约束或线性化形式。
| 参数 | 类型 | 含义 |
|---|---|---|
cond |
Expr | Any |
条件表达式。 |
when_true |
Expr | Any |
条件为真时的值。 |
when_false |
Expr | Any |
条件为假时的值。 |
返回值:Expr,条件表达式。
# 1. 如果使用加急模式,成本为 100,否则成本为 30。
cost = builder.iif(use_fast == 1, 100, 30)
# 2. 把条件成本作为目标。
builder.minimize(cost, name="mode_cost")
set_len¶
标签: HEURISTIC
签名:set_len(set_expr: Expr | Any) -> Expr
| 参数 | 类型 | 含义 |
|---|---|---|
set_expr |
Expr | Any |
集合变量或集合表达式。 |
返回值:Expr,集合大小表达式。
# 1. 计算当前选择集合的元素数量。
selected_count = builder.set_len(selected)
# 2. 限制最多选择 2 个元素。
builder.constraint(selected_count <= 2, name="selection_limit")
set_contains¶
标签: HEURISTIC
签名:set_contains(set_expr: Expr | Any, value_expr: Expr | Any) -> Expr
| 参数 | 类型 | 含义 |
|---|---|---|
set_expr |
Expr | Any |
集合变量或集合表达式。 |
value_expr |
Expr | Any |
要检查的元素值。 |
返回值:Expr,元素是否属于集合的 0/1 表达式。
# 1. 检查任务 3 是否被选中。
has_job_3 = builder.set_contains(selected, 3)
# 2. 要求任务 3 必须被选中。
builder.constraint(has_job_3 == 1, name="need_job_3")
sequence_contains¶
标签: HEURISTIC
签名:sequence_contains(sequence_expr: Expr | Any, value_expr: Expr | Any) -> Expr
CP-SAT 边界
sequence_contains 当前不属于 CP-SAT sequence lowering 支持面。CP-SAT 只支持 sequence_var 与 no_overlap 组合出的排序约束。
| 参数 | 类型 | 含义 |
|---|---|---|
sequence_expr |
Expr | Any |
序列变量或序列表达式。 |
value_expr |
Expr | Any |
要检查的元素值。 |
返回值:Expr,元素是否出现在序列中的 0/1 表达式。
# 1. 检查地点 2 是否在路线中。
visits_2 = builder.sequence_contains(route, 2)
# 2. 要求路线必须访问地点 2。
builder.constraint(visits_2 == 1, name="visit_2")
all_different¶
标签: HEURISTIC CP-SAT
签名:all_different(*exprs: Expr | Any) -> Expr
| 参数 | 类型 | 含义 |
|---|---|---|
*exprs |
Expr | Any |
要求互不相同的一组表达式。 |
返回值:Expr,互异约束表达式。
# 1. 三个任务必须分配到不同工位。
different_stations = builder.all_different(station_a, station_b, station_c)
# 2. 将互异关系登记为正式约束。
builder.constraint(different_stations, name="all_different_stations")
element¶
标签: HEURISTIC
签名:element(array_values: list[int | float], index_expr: Expr | Any) -> Expr
| 参数 | 类型 | 含义 |
|---|---|---|
array_values |
list[int | float] |
常量数组。 |
index_expr |
Expr | Any |
取值索引表达式。 |
返回值:Expr,array_values[index_expr] 对应的表达式。
# 1. choice 决定选择哪个成本项。
choice = builder.int_var(default=0, lb=0, ub=2, name="choice")
# 2. 根据 choice 从常量数组中取成本。
picked_cost = builder.element([5, 8, 13], choice)
# 3. 最小化被选中的成本。
builder.minimize(picked_cost, name="picked_cost")
table¶
标签: HEURISTIC
签名:table(exprs: list[Expr | Any], allowed_tuples: list[list[int]]) -> Expr
精确解边界
element / table 当前未接入 CP-SAT 原生约束,也未线性化到 MILP。
| 参数 | 类型 | 含义 |
|---|---|---|
exprs |
list[Expr | Any] |
需要联合限制的一组表达式。 |
allowed_tuples |
list[list[int]] |
允许出现的元组列表。 |
返回值:Expr,表约束表达式。
# 1. x 和 y 只能取 (0, 1) 或 (1, 0),等价于二选一。
allowed_pair = builder.table([x, y], [[0, 1], [1, 0]])
# 2. 登记允许元组表约束。
builder.constraint(allowed_pair, name="xor_pairs")
interval_start¶
标签: HEURISTIC CP-SAT
签名:interval_start(interval_expr: Expr | Any) -> Expr
| 参数 | 类型 | 含义 |
|---|---|---|
interval_expr |
Expr | Any |
interval 变量或表达式。 |
返回值:Expr,interval 的开始时间表达式。
# 1. 读取任务开始时间。
start = builder.interval_start(job)
# 2. 要求任务不能早于时间 2 开始。
builder.constraint(start >= 2, name="release_time")
interval_end¶
标签: HEURISTIC CP-SAT
签名:interval_end(interval_expr: Expr | Any) -> Expr
| 参数 | 类型 | 含义 |
|---|---|---|
interval_expr |
Expr | Any |
interval 变量或表达式。 |
返回值:Expr,interval 的结束时间表达式。
# 1. 读取任务结束时间。
end = builder.interval_end(job)
# 2. 要求任务在工期窗口内完成。
builder.constraint(end <= 10, name="deadline")
interval_length¶
标签: HEURISTIC CP-SAT
签名:interval_length(interval_expr: Expr | Any) -> Expr
| 参数 | 类型 | 含义 |
|---|---|---|
interval_expr |
Expr | Any |
interval 变量或表达式。 |
返回值:Expr,interval 的持续时间表达式。
# 1. 读取任务持续时间。
duration = builder.interval_length(job)
# 2. 限制持续时间不能超过 4。
builder.constraint(duration <= 4, name="duration_limit")
cumulative¶
标签: HEURISTIC CP-SAT
签名:cumulative(intervals: list[Expr | Any], demands: list[Expr | Any], capacity: Expr | Any) -> Expr
| 参数 | 类型 | 含义 |
|---|---|---|
intervals |
list[Expr | Any] |
占用资源的 interval 列表。 |
demands |
list[Expr | Any] |
每个 interval 的资源需求。 |
capacity |
Expr | Any |
资源容量。 |
返回值:Expr,累积资源约束表达式。
# 1. 两个任务都占用同一台容量为 1 的机器。
machine_capacity = builder.cumulative([job_a, job_b], [1, 1], 1)
# 2. 登记容量约束,避免同时超载。
builder.constraint(machine_capacity, name="machine_capacity")
no_overlap¶
标签: HEURISTIC CP-SAT
签名:no_overlap(sequence_expr: Expr | Any, *interval_exprs: Expr | Any) -> Expr
| 参数 | 类型 | 含义 |
|---|---|---|
sequence_expr |
Expr | Any |
表示任务顺序的序列变量。 |
*interval_exprs |
Expr | Any |
需要互不重叠的 interval。 |
返回值:Expr,不重叠约束表达式。
# 1. sequence_var 表示同一机器上的任务顺序。
order = builder.sequence_var(2, name="machine_order")
# 2. no_overlap 约束 job_a 和 job_b 不能同时占用机器。
builder.constraint(builder.no_overlap(order, job_a, job_b), name="no_overlap")
precedence¶
标签: HEURISTIC CP-SAT
签名:precedence(before_interval: Expr | Any, after_interval: Expr | Any, *, lag: Expr | int = 0) -> Expr
MILP 边界
cumulative、no_overlap、precedence 属于调度结构约束。当前 MILP 路径不自动生成对应线性 formulation。
| 参数 | 类型 | 含义 |
|---|---|---|
before_interval |
Expr | Any |
必须先完成的 interval。 |
after_interval |
Expr | Any |
后续 interval。 |
lag |
Expr | int |
两个 interval 之间的最小间隔。 |
返回值:Expr,前后工序约束表达式。
# 1. job_a 完成后至少间隔 1 个时间单位才能开始 job_b。
a_before_b = builder.precedence(job_a, job_b, lag=1)
# 2. 登记工序先后关系。
builder.constraint(a_before_b, name="a_before_b")
external_call¶
标签: HEURISTIC
签名:external_call(fn: Callable[..., Any], *inputs: Expr | Any, name: str | None = None, value_kind: ValueKind = ValueKind.SCALAR, timeout_ms: int | None = None, cache: bool = True) -> Expr
精确解边界
CP-SAT / MILP 不能调用任意 Python 函数并证明全局最优。包含 external_call 的目标或约束会使 exact lowering 拒绝该路径。
| 参数 | 类型 | 含义 |
|---|---|---|
fn |
Callable[..., Any] |
Python 外部评估函数。 |
*inputs |
Expr | Any |
传给外部函数的输入表达式。 |
name |
str | None |
外部调用名;不传时使用函数名。 |
value_kind |
ValueKind |
外部调用返回值类型,默认标量。 |
timeout_ms |
int | None |
单次调用超时毫秒数。 |
cache |
bool |
是否缓存相同输入的评估结果。 |
返回值:Expr,外部调用结果表达式。
def route_score(route: list[int]) -> int:
# 1. 在外部函数中实现暂时无法线性化的业务评分。
return len(route)
# 2. 用序列变量表达待优化的路线。
route = builder.sequence_var(5, name="route")
# 3. 将 Python 函数接入模型图,作为可被求解器调用的黑盒节点。
score = builder.external_call(route_score, route, name="route_score", timeout_ms=100)
# 4. 把黑盒评分登记为目标。
builder.minimize(score, name="score")
constraint¶
标签: MODEL
签名:constraint(expr: Expr | Any, *, name: str | None = None) -> Expr
| 参数 | 类型 | 含义 |
|---|---|---|
expr |
Expr | Any |
布尔表达式或关系表达式。 |
name |
str | None |
约束名称。 |
返回值:Expr,已登记的约束节点。
# 1. 先构造容量关系表达式。
capacity_ok = x + y <= 10
# 2. 再调用 constraint,把关系加入模型约束列表。
builder.constraint(capacity_ok, name="capacity")
约束必须登记
单独写 x + y <= 10 只会创建表达式,不会进入模型约束列表。需要显式调用 constraint(...)。
minimize¶
标签: MODEL
签名:minimize(expr: Expr | Any, *, name: str | None = None) -> Expr
| 参数 | 类型 | 含义 |
|---|---|---|
expr |
Expr | Any |
要最小化的目标表达式。 |
name |
str | None |
目标名称。 |
返回值:Expr,已登记的目标节点。
# 1. 构造总成本表达式。
total_cost = fixed_cost + variable_cost
# 2. 登记最小化目标。
builder.minimize(total_cost, name="total_cost")
maximize¶
标签: MODEL
签名:maximize(expr: Expr | Any, *, name: str | None = None) -> Expr
精确解边界
constraint、minimize、maximize 本身只是登记接口。能否走 CP-SAT 或 MILP,取决于被登记表达式里的变量、算子和约束类型。
| 参数 | 类型 | 含义 |
|---|---|---|
expr |
Expr | Any |
要最大化的目标表达式。 |
name |
str | None |
目标名称。 |
返回值:Expr,已登记的目标节点。
to_program_spec¶
标签: MODEL
签名:to_program_spec() -> ProgramSpec
参数:无。
返回值:ProgramSpec,可序列化的模型结构描述。
# 1. 导出中间结构,适合调试或持久化检查。
spec = builder.to_program_spec()
# 2. 查看变量、约束、目标节点数量。
print(len(spec.variable_ids), len(spec.constraint_ids), len(spec.objective_ids))
freeze¶
标签: MODEL
签名:freeze() -> ExecutableProgram
参数:无。
返回值:ExecutableProgram,可交给 Orchestrator().run(...) 的可执行模型。
Orchestrator API¶
Orchestrator 是公开求解入口。教程式说明见 求解接口,策略选择说明见 策略配置。
本页标题只显示对象名或方法名,便于右侧目录扫描。签名、字段类型、参数含义和返回值放在正文中。
接口分组¶
| 分组 | 对象 |
|---|---|
| 求解入口 | Orchestrator、run |
| 顶层配置 | OrchestratorConfig、PhaseConfig |
| 启发式配置 | HeuristicOrchestrationConfig、HeuristicPhaseConfig、HeuristicTerminationConfig、EvolutionaryConfig |
| 结果对象 | OrchestratorResult、SolverTrace、PhaseTrace |
| 常用枚举 | OrchestratorSolver、ExactBackendName、ExecutionMode、HeuristicPhaseKind |
Orchestrator¶
Orchestrator 封装启发式、CP-SAT、MILP 和多阶段编排逻辑。通常直接创建实例后调用 run。
from optagent import ModelBuilder, Orchestrator
# 1. 先用 ModelBuilder 建立最小可求解模型。
builder = ModelBuilder(metadata={"case": "orchestrator_demo"})
x = builder.int_var(default=0, lb=0, ub=1, name="x")
y = builder.int_var(default=0, lb=0, ub=1, name="y")
builder.constraint(x + y <= 1, name="capacity")
builder.maximize((x * 3) + (y * 2), name="profit")
# 2. 冻结为 Orchestrator 可接收的 ExecutableProgram。
program = builder.freeze()
# 3. 不传 preset/config 时,系统会自动推荐内置预设。
result = Orchestrator().run(program)
run¶
签名:run(program: ExecutableProgram, *, config: OrchestratorConfig | None = None, preset: StrategyPreset | BuiltInStrategyPreset | str | None = None, decomposition_plan: DecompositionPlan | None = None) -> OrchestratorResult
| 参数 | 类型 | 含义 |
|---|---|---|
program |
ExecutableProgram |
ModelBuilder.freeze() 生成的可执行模型。 |
config |
OrchestratorConfig | None |
显式编排配置;与 preset 互斥。 |
preset |
StrategyPreset | BuiltInStrategyPreset | str | None |
内置或自定义策略预设;与 config 互斥。 |
decomposition_plan |
DecompositionPlan | None |
可选分解计划;不传时由系统分析生成。 |
返回值:OrchestratorResult,包含最终解、阶段 trace、诊断信息和选中预设信息。
# 1. 自动推荐模式:适合首次验证模型。
auto_result = Orchestrator().run(program)
# 2. 预设模式:适合已知问题类型,希望复用内置策略。
preset_result = Orchestrator().run(program, preset="linear_exact_first")
# 3. 显式配置模式:适合需要固定求解阶段、求解器或后端。
configured_result = Orchestrator().run(
program,
config=OrchestratorConfig(total_budget_iterations=120, seed=7),
)
互斥参数
config 和 preset 不能同时传入。需要固定求解细节时用 config;需要复用策略包时用 preset。
OrchestratorConfig¶
签名:OrchestratorConfig(total_budget_iterations: int = 100, phases: list[PhaseConfig] = ..., seed: int = 0, execution_mode: str | ExecutionMode = ExecutionMode.SEQUENTIAL, trace_path: str | None = None, required_backend: str | ExactBackendName | None = None, allowed_backends: tuple[str | ExactBackendName, ...] = (), strict_backend: bool = False)
| 字段 | 类型 | 含义 |
|---|---|---|
total_budget_iterations |
int |
总迭代预算,供没有单独预算的阶段参考。 |
phases |
list[PhaseConfig] |
顶层求解阶段列表;为空时使用默认流程。 |
seed |
int |
随机种子,用于可重复的启发式过程。 |
execution_mode |
str | ExecutionMode |
阶段执行模式,例如顺序执行。 |
trace_path |
str | None |
可选 trace 输出路径。 |
required_backend |
str | ExactBackendName | None |
强制使用的精确后端。 |
allowed_backends |
tuple[str | ExactBackendName, ...] |
允许使用的精确后端白名单。 |
strict_backend |
bool |
后端不可用或不匹配时是否直接失败。 |
返回值:构造函数返回 OrchestratorConfig 实例。
from optagent import ExactBackendName, OrchestratorConfig
# 1. 固定随机种子,便于复现实验结果。
# 2. 要求精确路径必须使用 HiGHS 原生后端。
config = OrchestratorConfig(
total_budget_iterations=120,
seed=7,
required_backend=ExactBackendName.HIGHS_NATIVE,
strict_backend=True,
)
PhaseConfig¶
签名:PhaseConfig(name: str, solver: str | OrchestratorSolver = OrchestratorSolver.HEURISTIC, budget_iterations: int = 50, fallback_solver: str | OrchestratorSolver = OrchestratorSolver.HEURISTIC, fallback_on_failure: bool = True, fallback_on_stall: bool = True, warm_start: bool = True, strategy: str | HeuristicStrategy = HeuristicStrategy.LOCAL_SEARCH, heuristic_plan: HeuristicOrchestrationConfig | None = None, restart_limit: int = 0, lock_objectives: bool = False, metadata: dict[str, str] = ...)
| 字段 | 类型 | 含义 |
|---|---|---|
name |
str |
阶段名称,会出现在 trace 中。 |
solver |
str | OrchestratorSolver |
当前阶段使用的求解器类型。 |
budget_iterations |
int |
当前阶段迭代预算。 |
fallback_solver |
str | OrchestratorSolver |
失败或停滞时的回退求解器。 |
fallback_on_failure |
bool |
当前阶段失败时是否尝试回退。 |
fallback_on_stall |
bool |
当前阶段停滞时是否尝试回退。 |
warm_start |
bool |
是否继承前一阶段状态。 |
strategy |
str | HeuristicStrategy |
启发式阶段的搜索策略。 |
heuristic_plan |
HeuristicOrchestrationConfig | None |
启发式内部子阶段计划;仅 solver=HEURISTIC 时可用。 |
restart_limit |
int |
阶段重启次数上限。 |
lock_objectives |
bool |
是否锁定前序阶段已取得的目标值。 |
metadata |
dict[str, str] |
自定义阶段元数据。 |
返回值:构造函数返回 PhaseConfig 实例。
from optagent import OrchestratorConfig, OrchestratorSolver, PhaseConfig
# 1. 第一阶段用启发式快速构造可行解。
# 2. 第二阶段用 MILP 做精确 polish。
config = OrchestratorConfig(
phases=[
PhaseConfig(
name="heuristic_seed",
solver=OrchestratorSolver.HEURISTIC,
budget_iterations=40,
),
PhaseConfig(
name="exact_polish",
solver=OrchestratorSolver.MILP,
budget_iterations=20,
warm_start=True,
),
]
)
HeuristicOrchestrationConfig¶
签名:HeuristicOrchestrationConfig(phases: list[HeuristicPhaseConfig], preserve_best_solution: bool = True, parallel_phases: bool = False, max_parallel_workers: int = 1, portfolio_rounds: int = 1, adaptive_budget: bool = False, adaptive_phase_skipping: bool = False, evolutionary_plan: EvolutionaryConfig | None = None)
| 字段 | 类型 | 含义 |
|---|---|---|
phases |
list[HeuristicPhaseConfig] |
启发式内部子阶段列表。 |
preserve_best_solution |
bool |
是否跨子阶段保留全局最优解。 |
parallel_phases |
bool |
是否并行执行子阶段。 |
max_parallel_workers |
int |
并行 worker 上限。 |
portfolio_rounds |
int |
portfolio 重复轮数。 |
adaptive_budget |
bool |
是否根据阶段表现自适应分配预算。 |
adaptive_phase_skipping |
bool |
是否允许跳过低收益阶段。 |
evolutionary_plan |
EvolutionaryConfig | None |
可选进化式搜索计划。 |
返回值:构造函数返回 HeuristicOrchestrationConfig 实例。
# 1. 先 repair,快速修复不可行解。
# 2. 再 intensify,在当前可行区域附近继续改进。
heuristic_plan = HeuristicOrchestrationConfig(
phases=[
HeuristicPhaseConfig(name="repair", kind=HeuristicPhaseKind.REPAIR),
HeuristicPhaseConfig(name="intensify", kind=HeuristicPhaseKind.INTENSIFY),
],
preserve_best_solution=True,
)
HeuristicPhaseConfig¶
签名:HeuristicPhaseConfig(name: str, kind: str | HeuristicPhaseKind = HeuristicPhaseKind.INTENSIFY, strategy: str | HeuristicStrategy | None = None, start_policy: str | HeuristicStartPolicy | None = None, publish_policy: str | HeuristicPublishPolicy | None = None, warm_start: bool = True, termination: HeuristicTerminationConfig | None = None, restart_limit: int = 0, enable_lns: bool | None = None, lns_every: int | None = None, lns_destroy_count: int | None = None, enable_schedule_propagation: bool | None = None, metadata: dict[str, str] = ...)
| 字段 | 类型 | 含义 |
|---|---|---|
name |
str |
子阶段名称。 |
kind |
str | HeuristicPhaseKind |
子阶段语义,例如 REPAIR、INTENSIFY、DIVERSIFY、LNS。 |
strategy |
str | HeuristicStrategy | None |
搜索策略;不传时按 kind 使用默认值。 |
start_policy |
str | HeuristicStartPolicy | None |
子阶段起点来源。 |
publish_policy |
str | HeuristicPublishPolicy | None |
子阶段结果如何发布到全局状态。 |
warm_start |
bool |
是否使用前序状态热启动。 |
termination |
HeuristicTerminationConfig | None |
停止条件;不传时按 kind 使用默认值。 |
restart_limit |
int |
子阶段重启次数上限。 |
enable_lns |
bool | None |
是否启用 LNS 行为。 |
lns_every |
int | None |
LNS 触发间隔。 |
lns_destroy_count |
int | None |
每次 LNS 破坏变量数量。 |
enable_schedule_propagation |
bool | None |
是否启用调度传播。 |
metadata |
dict[str, str] |
自定义元数据。 |
返回值:构造函数返回 HeuristicPhaseConfig 实例。
# 1. 明确子阶段最多运行 20 次迭代。
termination = HeuristicTerminationConfig(iteration_limit=20)
# 2. 创建 repair 子阶段,找到第一个可行解后可提前停止。
repair_phase = HeuristicPhaseConfig(
name="repair",
kind=HeuristicPhaseKind.REPAIR,
termination=termination,
)
HeuristicTerminationConfig¶
签名:HeuristicTerminationConfig(mode: str | HeuristicTerminationMode = HeuristicTerminationMode.ITERATIONS, iteration_limit: int | None = None, time_limit_seconds: float | None = None, unimproved_iterations: int | None = None, solution_limit: int | None = None, stop_on_first_feasible: bool = False, stop_on_target_score: float | None = None)
| 字段 | 类型 | 含义 |
|---|---|---|
mode |
str | HeuristicTerminationMode |
主要终止模式。 |
iteration_limit |
int | None |
迭代上限;mode=ITERATIONS 时必填。 |
time_limit_seconds |
float | None |
时间上限;mode=TIME 时必填。 |
unimproved_iterations |
int | None |
连续无改进上限;mode=UNIMPROVED_ITERATIONS 时必填。 |
solution_limit |
int | None |
解数量上限;mode=SOLUTION_LIMIT 时必填。 |
stop_on_first_feasible |
bool |
找到第一个可行解后是否停止。 |
stop_on_target_score |
float | None |
达到目标分数后是否停止。 |
返回值:构造函数返回 HeuristicTerminationConfig 实例。
# 1. 用迭代次数作为主停止条件。
# 2. repair 阶段找到可行解后立即停止,节省预算。
termination = HeuristicTerminationConfig(
iteration_limit=30,
stop_on_first_feasible=True,
)
EvolutionaryConfig¶
签名:EvolutionaryConfig(population_size: int = 8, elite_size: int = 2, generation_limit: int | None = None, stagnation_generations: int | None = None, selection: str | SelectionStrategy = SelectionStrategy.TOURNAMENT, crossover: str | CrossoverStrategy = CrossoverStrategy.UNIFORM, mutation: str | MutationStrategy = MutationStrategy.RANDOM_RESET, mutation_portfolio: tuple[str | MutationStrategy, ...] = (), adaptive_mutation: bool = False, parallel_evaluation: bool = False, max_parallel_workers: int = 1, island_count: int = 1, migration_interval: int | None = None, migration_size: int = 1, repair_plan: HeuristicOrchestrationConfig | None = None, repair_budget_iterations: int | None = None, local_improvement_plan: HeuristicOrchestrationConfig | None = None, local_improvement_budget_iterations: int | None = None, local_improvement_trigger: str | LocalImprovementTrigger = LocalImprovementTrigger.ALWAYS, local_improvement_top_k: int = 1, exact_polish_solver: str | OrchestratorSolver | None = None, exact_polish_backend: str | ExactBackendName | None = None)
| 字段 | 类型 | 含义 |
|---|---|---|
population_size |
int |
种群规模,至少为 2。 |
elite_size |
int |
精英保留数量,范围为 1..population_size。 |
generation_limit |
int | None |
最大代数;与 stagnation_generations 至少提供一个。 |
stagnation_generations |
int | None |
连续无改进代数上限。 |
selection / crossover / mutation |
str | Enum |
选择、交叉和变异策略。 |
mutation_portfolio |
tuple[str | MutationStrategy, ...] |
多变异算子组合。 |
adaptive_mutation |
bool |
是否根据表现调整变异。 |
parallel_evaluation |
bool |
是否并行评估个体。 |
max_parallel_workers |
int |
并行 worker 上限。 |
island_count |
int |
island model 的岛数量。 |
migration_interval |
int | None |
岛间迁移间隔。 |
migration_size |
int |
每次迁移个体数量。 |
repair_plan |
HeuristicOrchestrationConfig | None |
个体 repair 计划。 |
repair_budget_iterations |
int | None |
repair 预算。 |
local_improvement_plan |
HeuristicOrchestrationConfig | None |
局部改进计划。 |
local_improvement_budget_iterations |
int | None |
局部改进预算。 |
local_improvement_trigger |
str | LocalImprovementTrigger |
触发局部改进的条件。 |
local_improvement_top_k |
int |
每代改进前 k 个个体。 |
exact_polish_solver |
str | OrchestratorSolver | None |
最后精确 polish 的求解器。 |
exact_polish_backend |
str | ExactBackendName | None |
最后精确 polish 的后端。 |
返回值:构造函数返回 EvolutionaryConfig 实例。
# 1. 设置小种群和代数上限,适合快速试跑。
# 2. stagnation_generations 防止长时间无改进。
evolutionary = EvolutionaryConfig(
population_size=8,
elite_size=2,
generation_limit=20,
stagnation_generations=5,
)
OrchestratorResult¶
run 的返回对象。
| 字段 | 类型 | 含义 |
|---|---|---|
global_state |
GlobalState |
编排过程中的全局状态。 |
decomposition_plan |
DecompositionPlan |
实际使用的问题分解计划。 |
merge_results |
list[MergeResult] |
阶段状态合并结果。 |
solver_traces |
list[SolverTrace] |
顶层求解阶段摘要。 |
phase_traces |
list[PhaseTrace] |
阶段状态交接记录。 |
heuristic_subphase_traces |
list[HeuristicSubphaseTrace] |
启发式子阶段记录。 |
evolutionary_generation_traces |
list[EvolutionaryGenerationTrace] |
进化式代际记录。 |
final_solution |
UnifiedSolution |
最终统一解对象。 |
conflict_events |
list[dict[str, object]] |
合并冲突事件。 |
diagnostics |
list[DiagnosticPayload] |
后端与求解诊断。 |
selected_preset_name |
str | None |
实际选中的预设名。 |
selected_preset_source |
str | None |
预设来源。 |
mutation_successes |
dict[str, int] |
变异成功统计。 |
mutation_trials |
dict[str, int] |
变异尝试统计。 |
# 1. final_solution 是业务读取求解状态和变量值的主入口。
solution = result.final_solution
print(solution.status.value)
print(solution.objective_values)
print(solution.variable_values)
# 2. trace 字段用于排查编排过程。
print(result.selected_preset_name)
print(len(result.solver_traces))
print(len(result.phase_traces))
SolverTrace¶
| 字段 | 类型 | 含义 |
|---|---|---|
phase_name |
str |
阶段名称。 |
solver_name |
str |
实际求解器名称。 |
status |
str |
阶段求解状态。 |
fallback_reason |
str | None |
回退原因;未回退时为 None。 |
score |
float |
阶段结束得分。 |
feasible |
bool |
阶段结束时是否可行。 |
metrics |
dict[str, int] |
阶段指标。 |
# 1. 遍历 solver trace,查看每个阶段实际跑了什么。
for trace in result.solver_traces:
print(trace.phase_name, trace.solver_name, trace.status, trace.feasible)
PhaseTrace¶
| 字段 | 类型 | 含义 |
|---|---|---|
phase_name |
str |
阶段名称。 |
solver_name |
str |
阶段使用的求解器。 |
budget_iterations |
int |
阶段预算。 |
version_before |
int |
合并前状态版本。 |
version_after |
int |
合并后状态版本。 |
merge_reason |
str |
状态合并原因。 |
changed_variable_ids |
tuple[int, ...] |
本阶段修改的变量 ID。 |
feasible_after_merge |
bool |
合并后是否可行。 |
# 1. 查看每个阶段是否真正改变了变量。
for trace in result.phase_traces:
print(trace.phase_name, trace.changed_variable_ids, trace.feasible_after_merge)
常用枚举¶
| 枚举 | 常见值 | 用途 |
|---|---|---|
OrchestratorSolver |
HEURISTIC、CP_SAT、MILP |
顶层求解器类型。 |
ExactBackendName |
CP_SAT_NATIVE、MATHOPT_MP、HIGHS_NATIVE |
精确后端约束。 |
ExecutionMode |
SEQUENTIAL |
顶层阶段执行模式。 |
HeuristicPhaseKind |
CONSTRUCT、REPAIR、INTENSIFY、DIVERSIFY、LNS、CUSTOM |
启发式子阶段语义。 |
HeuristicStrategy |
LOCAL_SEARCH、TABU、ANNEALING |
启发式搜索策略。 |
HeuristicTerminationMode |
ITERATIONS、TIME、UNIMPROVED_ITERATIONS、SOLUTION_LIMIT |
启发式终止模式。 |
Presets API¶
预设用于复用常见问题类型的求解策略。教程式说明见 策略配置。
本页标题只显示对象名或方法名,便于右侧目录扫描。签名、字段类型、参数含义和返回值放在正文中。
接口分组¶
| 分组 | 对象 |
|---|---|
| 问题特征 | ProblemFeatures、detect_problem_features |
| 预设对象 | BuiltInStrategyPreset、StrategyPreset |
| 推荐结果 | StrategyRecommendation |
| 工具函数 | recommend_preset、list_strategy_presets、get_strategy_preset、load_strategy_preset、describe_preset、describe_recommendation |
ProblemFeatures¶
ProblemFeatures 是对模型结构的摘要,用于驱动自动预设推荐。
| 字段 | 类型 | 含义 |
|---|---|---|
problem_type |
str |
问题类型,例如 scheduling、routing、knapsack、general。 |
estimated_search_space |
str |
粗粒度搜索空间大小:small、medium、large、huge。 |
num_variables |
int |
变量数量。 |
num_constraints |
int |
约束数量。 |
num_objectives |
int |
目标数量。 |
has_scheduling |
bool |
是否包含调度结构。 |
has_blackbox |
bool |
是否包含黑盒调用。 |
has_linear |
bool |
目标和约束是否线性友好。 |
has_sequence |
bool |
是否包含序列变量。 |
has_interval |
bool |
是否包含 interval 变量。 |
has_sequence_blackbox |
bool |
是否包含序列变量和黑盒调用的组合。 |
has_sequence_path_linear |
bool |
是否是序列路径线性场景。 |
from optagent.presets import detect_problem_features
# 1. 对已冻结的 program 做结构识别。
features = detect_problem_features(program)
# 2. 读取关键特征,判断系统会把它归入哪类问题。
print(features.problem_type)
print(features.has_linear)
print(features.num_variables)
detect_problem_features¶
签名:detect_problem_features(program: ExecutableProgram) -> ProblemFeatures
| 参数 | 类型 | 含义 |
|---|---|---|
program |
ExecutableProgram |
ModelBuilder.freeze() 生成的可执行模型。 |
返回值:ProblemFeatures,模型结构摘要。
# 1. 冻结模型后再做特征识别。
program = builder.freeze()
# 2. detect_problem_features 不求解模型,只分析模型结构。
features = detect_problem_features(program)
# 3. 根据特征决定是否需要手动覆盖预设。
if features.problem_type == "knapsack" and features.has_linear:
print("可以优先考虑 linear_* 预设")
BuiltInStrategyPreset¶
BuiltInStrategyPreset 是稳定的内置预设枚举,适合在业务代码中引用。
| 典型前缀 | 场景 |
|---|---|
SCHEDULING_* |
调度问题。 |
SEQUENCE_* |
排列、路径、顺序问题。 |
ROUTING_* |
路由或序列黑盒问题。 |
LINEAR_* |
线性、整数规划问题。 |
GENERAL_* |
通用启发式问题。 |
HYBRID_* |
混合求解路径。 |
from optagent import BuiltInStrategyPreset, Orchestrator
# 1. 用枚举值显式选择线性精确优先策略。
result = Orchestrator().run(
program,
preset=BuiltInStrategyPreset.LINEAR_EXACT_FIRST,
)
# 2. 返回结果中会记录实际选中的预设名和来源。
print(result.selected_preset_name)
print(result.selected_preset_source)
StrategyPreset¶
StrategyPreset 是已编译好的策略预设对象。
| 字段 | 类型 | 含义 |
|---|---|---|
name |
str |
预设名称。 |
description |
str |
预设说明。 |
orchestrator_config |
OrchestratorConfig |
顶层编排配置。 |
execution_plan |
ExecutionPlan |
编译后的执行计划。 |
family |
str |
预设族,例如 heuristic、exact、hybrid。 |
objective |
str |
策略目标倾向,例如 speed、balanced、quality。 |
tags |
tuple[str, ...] |
标签。 |
requirements |
tuple[str, ...] |
适用前提。 |
from optagent.presets import get_strategy_preset
# 1. 按名称取出一个与当前 program 匹配的预设对象。
preset = get_strategy_preset(program, "linear_exact_first")
# 2. 查看预设的目标倾向和阶段配置。
print(preset.family)
print(preset.objective)
print(len(preset.orchestrator_config.phases))
StrategyRecommendation¶
StrategyRecommendation 是自动推荐流程的解释对象,内部推荐路径会使用它。
| 字段 | 类型 | 含义 |
|---|---|---|
preset |
StrategyPreset |
推荐出的预设对象。 |
features |
ProblemFeatures |
用于推荐的问题特征。 |
recommendation_source |
str |
推荐来源,例如 rules 或 benchmark。 |
recommendation_reason |
str |
推荐原因。 |
matched_benchmark_cases |
tuple[str, ...] |
命中的 benchmark case。 |
benchmark_wins |
dict[str, int] |
benchmark 胜出统计。 |
from optagent.presets import describe_recommendation
# 1. recommendation 通常由自动推荐流程生成。
# 2. describe_recommendation 可把它转成适合日志记录的 dict。
payload = describe_recommendation(recommendation)
print(payload["recommendation_source"])
print(payload["preset"]["name"])
recommend_preset¶
签名:recommend_preset(program: ExecutableProgram) -> StrategyPreset
| 参数 | 类型 | 含义 |
|---|---|---|
program |
ExecutableProgram |
已冻结的可执行模型。 |
返回值:StrategyPreset,系统基于规则为当前模型推荐的预设。
from optagent.presets import recommend_preset
# 1. 让系统根据模型结构推荐一个预设。
preset = recommend_preset(program)
# 2. 可以先打印确认,再决定是否传给 Orchestrator。
print(preset.name)
result = Orchestrator().run(program, preset=preset)
list_strategy_presets¶
签名:list_strategy_presets(program: ExecutableProgram) -> tuple[StrategyPreset, ...]
| 参数 | 类型 | 含义 |
|---|---|---|
program |
ExecutableProgram |
已冻结的可执行模型。 |
返回值:tuple[StrategyPreset, ...],所有匹配当前模型特征的内置预设。
from optagent.presets import list_strategy_presets
# 1. 列出当前模型可用的预设,而不是所有内置预设。
presets = list_strategy_presets(program)
# 2. 展示名称、族和目标倾向,方便人工选择。
for preset in presets:
print(preset.name, preset.family, preset.objective)
get_strategy_preset¶
签名:get_strategy_preset(program: ExecutableProgram, preset_name: str | BuiltInStrategyPreset) -> StrategyPreset
| 参数 | 类型 | 含义 |
|---|---|---|
program |
ExecutableProgram |
已冻结的可执行模型。 |
preset_name |
str | BuiltInStrategyPreset |
预设名称或内置预设枚举。 |
返回值:StrategyPreset,名称匹配且适用于当前模型的预设。
from optagent import BuiltInStrategyPreset
from optagent.presets import get_strategy_preset
# 1. 枚举写法适合长期维护的业务代码。
preset = get_strategy_preset(program, BuiltInStrategyPreset.LINEAR_EXACT_FIRST)
# 2. 字符串写法适合配置文件或命令行参数。
same_preset = get_strategy_preset(program, "linear_exact_first")
load_strategy_preset¶
签名:load_strategy_preset(path: str | Path, *, program: ExecutableProgram | None = None, validate_match: bool = True) -> StrategyPreset
| 参数 | 类型 | 含义 |
|---|---|---|
path |
str | Path |
外部 preset manifest 文件路径。 |
program |
ExecutableProgram | None |
可选模型;传入后可校验预设是否匹配模型特征。 |
validate_match |
bool |
是否启用匹配校验。 |
返回值:StrategyPreset,从 manifest 编译出的预设对象。
from optagent.presets import load_strategy_preset
# 1. 从外部 manifest 加载自定义预设。
# 2. 传入 program 时会校验该预设是否适配当前模型。
preset = load_strategy_preset(
"custom_preset.json",
program=program,
validate_match=True,
)
# 3. 加载成功后可直接传给 Orchestrator。
result = Orchestrator().run(program, preset=preset)
describe_preset¶
签名:describe_preset(preset: StrategyPreset) -> dict[str, Any]
| 参数 | 类型 | 含义 |
|---|---|---|
preset |
StrategyPreset |
要描述的预设对象。 |
返回值:dict[str, Any],包含名称、描述、族、目标、标签、要求、执行概要和编排配置的字典。
from optagent.presets import describe_preset, get_strategy_preset
# 1. 取出预设对象。
preset = get_strategy_preset(program, "linear_exact_first")
# 2. 转成 dict,适合打印、日志或 JSON 序列化。
payload = describe_preset(preset)
print(payload["family"])
print(payload["strategy_outline"])
describe_recommendation¶
签名:describe_recommendation(recommendation: StrategyRecommendation) -> dict[str, Any]
| 参数 | 类型 | 含义 |
|---|---|---|
recommendation |
StrategyRecommendation |
要描述的推荐结果。 |
返回值:dict[str, Any],包含推荐预设、问题特征、推荐来源、推荐原因和 benchmark 反馈的字典。
精确求解器接口兼容性¶
本页说明 ModelBuilder 接口在 CP-SAT 和 MILP 两种精确解模式下的支持范围。完整接口签名和示例见 Builder API。
以完整表达式图为准
变量类型兼容不代表整个模型兼容。只要目标或约束中包含当前后端不支持的节点,精确解 lowering 仍会拒绝该模型。
标签¶
| 标签 | 含义 |
|---|---|
| HEURISTIC | 启发式路径可评估或搜索该语义。 |
| CP-SAT | 可进入当前 cp_sat_native 精确路径。 |
| MILP | 可进入当前 mathopt_mp / highs_native 精确路径。 |
| MODEL | 建模、登记或导出接口,不直接决定后端支持。 |
CP-SAT 模式¶
CP-SAT 当前适合整数、布尔、组合逻辑和调度结构。它不接收连续变量、集合变量、黑盒调用,也不支持所有序列操作。
CP-SAT 支持¶
| 建模接口 | 支持条件 |
|---|---|
| const | 参与 CP-SAT lowering 的常量必须是整数或布尔值。 |
| bool_var | 支持。 |
| int_var | 必须有有限 lb / ub。 |
| sequence_var | 仅支持绑定到 no_overlap 的调度序列。 |
| interval_var | 支持 interval 开始、结束、持续时间。 |
| abs | 支持整数表达式绝对值。 |
| sum | 支持整数表达式求和。 |
| min / max | 支持整数表达式聚合。 |
| and_ / or_ / not_ | 支持布尔表达式组合。 |
比较表达式 <= / >= / < / > / == / != |
支持整数和布尔上下文中的比较。 |
| all_different | 支持。 |
| interval_start / interval_end / interval_length | 支持 interval 派生标量。 |
| cumulative | 支持调度累积资源约束。 |
| no_overlap | 支持,并负责把 sequence_var 绑定到 interval 组。 |
| precedence | 支持前后工序约束。 |
| constraint / minimize / maximize | 登记接口可用;内部表达式必须属于 CP-SAT 支持范围。 |
| to_program_spec / freeze | 可用,但不改变后端支持范围。 |
CP-SAT 不支持¶
| 建模接口 | 原因 |
|---|---|
| float_var | 当前 CP-SAT 路径构建整数域模型,不接收连续变量。 |
| set_var、set_len、set_contains | 当前 canonical CP model 没有集合变量语义。 |
| sequence_contains | 当前 sequence lowering 只覆盖 no_overlap 排序语义。 |
| iif | 未实现条件表达式 lowering。 |
| element / table | 未接入 CP-SAT 原生 element/table 约束。 |
| external_call | 精确解后端不能调用任意 Python 黑盒函数并证明全局最优。 |
MILP 模式¶
MILP 当前是线性 MP lowering,适合布尔、整数、连续变量上的线性目标和线性约束。它不自动做调度、逻辑、集合、黑盒或非线性表达式改写。
MILP 支持¶
| 建模接口 | 支持条件 |
|---|---|
| const | 支持数值和布尔常量;不支持 list/dict 常量参与线性表达式。 |
| bool_var | 支持,降为 0/1 变量。 |
| int_var | 支持整数变量,可带上下界。 |
| float_var | 支持连续变量,可带上下界。 |
| sum | 支持线性求和。 |
| 加减与取负表达式 | 支持线性组合。 |
| 乘常量表达式 | 支持 变量或线性表达式 * 常量;不支持变量乘变量。 |
比较表达式 <= / >= / == |
支持线性比较。 |
严格比较 < / > |
仅当比较表达式可判定为整数时支持,会转为相差至少 1。 |
| constraint / minimize / maximize | 登记接口可用;内部表达式必须为线性 MP 支持范围。 |
| to_program_spec / freeze | 可用,但不改变后端支持范围。 |
MILP 不支持¶
| 建模接口 | 原因 |
|---|---|
| set_var、set_len、set_contains | 当前 canonical MP model 只包含标量变量和线性约束。 |
| sequence_var、sequence_contains | 当前 MILP 路径不自动生成排列或路径 formulation。 |
| interval_var、interval_start、interval_end、interval_length | 当前 MILP 路径不接收 interval 结构。 |
| abs | 未自动改写为辅助变量和线性约束。 |
| min / max | 未自动生成 epigraph / hypograph 线性 formulation。 |
| and_ / or_ / not_ / iif | 未自动做逻辑约束或 big-M 线性化。 |
!= 比较 |
当前线性 MP lowering 不支持 disjunction。 |
| all_different | 未自动构造 pairwise 或 assignment 线性化。 |
| element / table | 未自动线性化。 |
| cumulative / no_overlap / precedence | 调度结构当前不自动生成 MILP formulation。 |
| external_call | 精确解后端不能调用任意 Python 黑盒函数并证明全局最优。 |
选择建议¶
| 模型特征 | 建议 |
|---|---|
| 纯线性目标和线性约束,包含连续变量 | 优先尝试 MILP。 |
整数、布尔、调度 interval、no_overlap、cumulative |
优先尝试 CP-SAT。 |
| 黑盒目标、集合变量、路线评分函数 | 优先走启发式或预设路径。 |
| 同一模型同时包含线性和调度/黑盒结构 | 使用混合配置或预设,让不同阶段处理不同子结构。 |
示例¶
OptAgent 的可运行示例单独维护在公开仓库,公开文档负责说明每类问题要表达的建模语义、代码目录结构、入口脚本和运行注意事项:
https://github.com/Dongbox/optagent-examples
完整可运行代码、示例专用测试和小型公开数据文件以 optagent-examples 为准。MkDocs 页面不再内嵌完整文件,但必须保留足够的问题说明和代码骨架,方便读者判断应该从哪个示例开始。
代码目录结构¶
optagent-examples 的主结构如下:
optagent-examples/
examples/
linear/ # 0/1 选择、指派、选址、线性化路由
scheduling/ # job shop、flow shop、sequence/interval 调度
blackbox/ # Python 黑盒评分、TSP、启发式搜索
hybrid/ # 线性计划 + 调度变量 + 多阶段编排
presets/ # 内置预设、自动预设、外部 preset 文件
mps/ # MPS 文件导入、窗口模型重建、后端切换
steel/ # 钢卷排序、焊接兼容性、搜索归因
resource_flow/ # 资源流 CP/MILP 双建模与滚动窗口
mg/ # SQLite 驱动的 MG 迁移示例
tests/ # 示例回归测试
scripts/ # 可重复实验脚本
requirements-dev.txt
pyproject.toml
运行入口统一从仓库根目录执行,常见形式为:
PYTHONPATH=. python examples/linear/knapsack_mathopt.py
PYTHONPATH=. python examples/scheduling/job_shop_small.py
PYTHONPATH=. python examples/blackbox/tsp_blackbox_small.py
示例分类¶
- 快速开始:最小建模、求解和结果读取闭环。
- 线性问题:MILP / MathOpt / HiGHS、指派、选址、MPS 相关入口。
- 调度问题:
sequence_var、interval_var、no_overlap、precedence和 CP-SAT。 - 黑盒优化:
external_call、Python 评分函数、TSP 和序列启发式。 - 混合优化:同一 DAG 中组合线性数量、区间调度、启发式和精确阶段。
- 预设示例:内置策略预设、自动预设选择和外部 preset 文件。
- MPS 示例:从 MPS 数据重建模型、比较 heuristic / hybrid / exact 路径。
- 钢卷排序:直接焊接兼容性、DAG 路径模型和搜索归因。
- 资源流:资源网络流转、CP/MILP 两种 formulation 和滚动窗口。
- MG 迁移:APS 兼容 SQLite 数据流、黑盒规则评分和生产输出表。
仓库边界¶
核心实现、内部文档和核心测试保留在主仓库。公开 optagent-examples 仓库承载可运行样例,并应满足:
- 每个样例有独立 README
- README 写明依赖安装、运行命令和预期输出
- 测试数据只使用可公开的小型 SQLite、JSON、CSV、MPS 等文件
- 不包含私有业务数据、内部绝对路径或内部系统链接
运行注意事项¶
- 通用示例使用当前 OptAgent 支持的 Python 版本和公开/批准的 OptAgent 安装包。
- 精确求解示例可能需要
ortools或highspy。 - MG 示例需要 Python 3.9,并且需要额外安装管理员提供的 APS wheel;该 wheel 不随公开仓库分发。
- 迁移型比较脚本如果依赖外部系统或历史工程,只用于迁移验证,不作为默认公开运行路径。
示例
快速开始示例¶
快速开始示例用于确认 OptAgent 的三件事可以串起来:用 ModelBuilder 定义变量和约束、冻结为 program、交给 Orchestrator 求解。
https://github.com/Dongbox/optagent-examples
目录结构¶
optagent-examples/
examples/
linear/
knapsack_mathopt.py
scheduling/
job_shop_small.py
blackbox/
tsp_blackbox_small.py
hybrid/
hybrid_production_planning_small.py
requirements-dev.txt
quickstart 没有单独目录,建议按从简单到复杂的顺序运行上面四个入口。
最小代码形状¶
下面是线性背包示例的核心结构,完整文件见公开仓库的 examples/linear/knapsack_mathopt.py。
from optagent import ModelBuilder, Orchestrator
builder = ModelBuilder(metadata={"case": "quickstart_knapsack"})
picks = [builder.int_var(default=0, lb=0, ub=1, name=f"pick_{idx}") for idx in range(3)]
weights = [2, 3, 4]
values = [3, 4, 8]
builder.constraint(builder.sum(*(pick * weight for pick, weight in zip(picks, weights))) <= 5)
builder.maximize(builder.sum(*(pick * value for pick, value in zip(picks, values))))
result = Orchestrator().run(builder.freeze())
print(result.final_solution.status)
运行¶
git clone https://github.com/Dongbox/optagent-examples.git
cd optagent-examples
python -m pip install -r requirements-dev.txt
PYTHONPATH=. python examples/linear/knapsack_mathopt.py
运行前需要先按团队或发行渠道提供的方式安装 OptAgent 本体。需要精确后端时,再安装对应可选依赖,例如 ortools 或 highspy。
线性问题示例¶
线性问题示例展示可以降低到 CanonicalMpModel 的模型:0/1 背包、二分指派、设施选址、路由线性化,以及 MPS 数据重建后的 MILP 路径。
https://github.com/Dongbox/optagent-examples
问题说明¶
knapsack_mathopt.py:在容量约束下选择物品,最大化价值,展示二进制变量和线性目标。assignment_highs_native.py:把任务分配给工人,每个任务和工人受唯一性约束,展示 HiGHS native 后端。facility_location_small.py:决定开哪些仓库,并把客户分配给已开启设施,展示固定成本和服务成本组合。routing_linearized_small.py:把小型路由问题改写为线性 MILP,而不是通过黑盒路径评分求解。
目录结构¶
examples/linear/
README.md
knapsack_mathopt.py
assignment_highs_native.py
facility_location_small.py
routing_linearized_small.py
代码示例¶
线性示例的共同结构是创建标量变量、线性约束和线性目标,然后通过 OrchestratorConfig 指定 MILP 家族或具体后端。
from optagent import ExactBackendName, ModelBuilder, Orchestrator, OrchestratorConfig
builder = ModelBuilder(solve_config={"preferred_backend": "highs_native"})
x = builder.int_var(default=0, lb=0, ub=1, name="open_facility")
y = builder.int_var(default=0, lb=0, ub=10, name="served_demand")
builder.constraint(y <= x * 10, name="serve_only_if_open")
builder.minimize((x * 5) + y, name="total_cost")
result = Orchestrator().run(
builder.freeze(),
config=OrchestratorConfig(required_backend=ExactBackendName.HIGHS_NATIVE),
)
运行¶
PYTHONPATH=. python examples/linear/knapsack_mathopt.py
PYTHONPATH=. python examples/linear/assignment_highs_native.py
PYTHONPATH=. python examples/linear/facility_location_small.py
PYTHONPATH=. python examples/linear/routing_linearized_small.py
注意事项¶
- MathOpt 路径通常需要
ortools。 - HiGHS native 路径需要
highspy。 - 严格后端示例会在后端不可用时报错;普通示例可能由 registry 选择可用后端。
- MPS 相关的大窗口示例在 MPS 示例 中单独说明。
调度问题示例¶
调度问题示例展示 sequence_var、interval_var、no_overlap、precedence 等语义如何降低到 CP-SAT 路径。
https://github.com/Dongbox/optagent-examples
问题说明¶
flow_shop_cp_sat.py:多个作业按相同工艺路线经过机器,目标是压缩整体完工时间。job_shop_small.py:两个作业以不同机器顺序加工,机器容量通过no_overlap表达,工序链通过precedence表达。
目录结构¶
代码示例¶
from optagent import ExactBackendName, ModelBuilder, Orchestrator, OrchestratorConfig
builder = ModelBuilder(metadata={"case": "small_job_shop"})
machine = builder.sequence_var(size=2, default=[0, 1], name="machine")
op_a = builder.interval_var(start=0, length=2, lb_start=0, ub_start=8, lb_length=2, ub_length=2)
op_b = builder.interval_var(start=0, length=3, lb_start=0, ub_start=8, lb_length=3, ub_length=3)
builder.constraint(builder.no_overlap(machine, op_a, op_b), name="machine_capacity")
builder.constraint(builder.precedence(op_a, op_b), name="job_flow")
builder.minimize(builder.interval_end(op_b), name="makespan")
result = Orchestrator().run(
builder.freeze(),
config=OrchestratorConfig(required_backend=ExactBackendName.CP_SAT_NATIVE),
)
运行¶
PYTHONPATH=. python examples/scheduling/flow_shop_cp_sat.py
PYTHONPATH=. python examples/scheduling/job_shop_small.py
注意事项¶
- CP-SAT native 路径需要
ortools。 - 输出重点看 makespan、各机器上的工序顺序、每个 interval 的开始和结束时间。
- 如果只想看序列黑盒搜索,不需要 interval 精确约束,可以先看 黑盒优化。
调度建模 API 见 建模接口,预设对象和选择入口见 Presets API。
黑盒优化示例¶
黑盒优化示例用于表达“模型结构可由 OptAgent 管理,但目标或局部评分由 Python 函数计算”的场景。典型变量是 sequence_var,典型目标是 external_call(...)。
https://github.com/Dongbox/optagent-examples
问题说明¶
routing_heuristic.py:按路径顺序调用 Python 评分函数,展示基础启发式路径。tsp_blackbox_small.py:把 TSP 路径长度写成 Python 黑盒函数,通过 tabu 搜索优化。tsp_evolutionary_small.py:同一个 TSP 黑盒目标,切换到进化式启发式搜索。steel_transition_sequence.py:钢卷直焊兼容性示例的兼容入口,推荐直接看 钢卷排序。
目录结构¶
examples/blackbox/
README.md
routing_heuristic.py
tsp_blackbox_small.py
tsp_evolutionary_small.py
steel_transition_sequence.py
代码示例¶
from optagent import HeuristicStrategy, ModelBuilder, Orchestrator, OrchestratorConfig, OrchestratorSolver, PhaseConfig
def route_cost(order: list[int]) -> int:
tour = order + [order[0]]
return sum(distance[tour[i]][tour[i + 1]] for i in range(len(order)))
builder = ModelBuilder(metadata={"case": "tsp_blackbox"})
route = builder.sequence_var(size=4, default=[3, 2, 1, 0], name="route")
builder.minimize(builder.external_call(route_cost, route, name="route_cost"))
result = Orchestrator().run(
builder.freeze(),
config=OrchestratorConfig(
phases=[
PhaseConfig(
name="tabu_route_search",
solver=OrchestratorSolver.HEURISTIC,
strategy=HeuristicStrategy.TABU,
budget_iterations=60,
)
],
),
)
运行¶
PYTHONPATH=. python examples/blackbox/routing_heuristic.py
PYTHONPATH=. python examples/blackbox/tsp_blackbox_small.py
PYTHONPATH=. python examples/blackbox/tsp_evolutionary_small.py
注意事项¶
- 黑盒函数应尽量确定性、可重复,并只依赖输入变量和公开数据。
- 预算、随机种子和启发式策略会影响解质量;需要比较时应固定 seed。
- 黑盒目标不能直接降低到通用 MILP/CP-SAT 约束,通常走启发式或预设路径。
黑盒问题通常走启发式路径;精确解兼容范围见 精确求解器接口兼容性。
混合优化示例¶
混合优化示例展示同一个 DAG 里同时存在线性数量平衡、区间调度、启发式阶段和精确阶段时,如何用编排器组合求解。
https://github.com/Dongbox/optagent-examples
问题说明¶
hybrid_production_planning_small.py 是一个小型生产计划问题:
- 本地生产量由 interval 长度决定。
- 外包量和延期惩罚是线性变量。
- 两种产品共用一条生产线,需要
no_overlap。 - 编排器先用启发式阶段找可行/较好初值,再尝试 CP-SAT 精确细化。
目录结构¶
代码示例¶
from optagent import ExecutionMode, HeuristicStrategy, ModelBuilder, Orchestrator, OrchestratorConfig, OrchestratorSolver, PhaseConfig
builder = ModelBuilder(metadata={"case": "hybrid_plan"})
line = builder.sequence_var(size=2, default=[0, 1], name="line_sequence")
blend_a = builder.interval_var(start=0, length=2, lb_start=0, ub_start=8, lb_length=0, ub_length=4)
local_a = builder.int_var(default=2, lb=0, ub=4, name="local_a")
outsource_a = builder.int_var(default=2, lb=0, ub=4, name="outsource_a")
builder.constraint(builder.no_overlap(line, blend_a), name="single_line")
builder.constraint(local_a == builder.interval_length(blend_a), name="local_matches_time")
builder.constraint(local_a + outsource_a == 4, name="meet_demand")
builder.minimize(outsource_a * 2, name="plan_cost")
result = Orchestrator().run(
builder.freeze(),
config=OrchestratorConfig(
execution_mode=ExecutionMode.ALTERNATING,
phases=[
PhaseConfig(name="seed", solver=OrchestratorSolver.HEURISTIC, strategy=HeuristicStrategy.TABU),
PhaseConfig(name="refine", solver=OrchestratorSolver.CP_SAT),
],
),
)
运行¶
注意事项¶
- 输出中的
solver_traces用来检查每个阶段是否求解、回退或停滞。 - 混合模型并不表示所有约束都能同时由同一个 exact backend 完整接收;编排器会按阶段尝试。
- 对生产类问题,建议同时记录最终解和阶段 trace,便于解释为什么采用当前解。
混合求解的显式阶段配置见 求解接口 和 Orchestrator API。
预设示例¶
预设示例展示如何少写编排配置:用户可以选择内置 preset、传入外部 preset 文件,或者让 Orchestrator().run(program) 自动选择策略。
https://github.com/Dongbox/optagent-examples/tree/main/examples/presets
问题说明¶
scheduling_memetic_quality_preset.py:调度问题使用 memetic quality preset,强调 repair 和 local improvement。scheduling_evolutionary_repair_preset.py:小型 job shop 使用 evolutionary repair preset。scheduling_evolutionary_repair_large_preset.py:较大的两机 flow-shop 风格模型使用同类 preset。routing_blackbox_preset.py:黑盒路径目标使用显式 routing preset。routing_blackbox_auto_preset.py:不传 preset,由编排器自动选择并在结果 metadata 中暴露选择结果。hybrid_production_preset.py:混合生产计划使用 hybrid-capable preset。
目录结构¶
examples/presets/
README.md
scheduling_memetic_quality_preset.py
scheduling_evolutionary_repair_preset.py
scheduling_evolutionary_repair_large_preset.py
routing_blackbox_preset.py
routing_blackbox_auto_preset.py
hybrid_production_preset.py
代码示例¶
from optagent import BuiltInStrategyPreset, Orchestrator
program = build_program()
result = Orchestrator().run(
program,
preset=BuiltInStrategyPreset.SCHEDULING_EVOLUTIONARY_REPAIR,
)
自动选择 preset 时:
运行¶
PYTHONPATH=. python examples/presets/scheduling_evolutionary_repair_preset.py
PYTHONPATH=. python examples/presets/routing_blackbox_auto_preset.py
PYTHONPATH=. python examples/presets/hybrid_production_preset.py
注意事项¶
- preset 示例适合展示用户侧 API,不适合替代底层能力测试。
- 需要解释阶段行为时,仍应查看
result.solver_traces。 - 外部 preset 文件应使用公开、可审计的 JSON/YAML,不包含内部路径。
MPS 示例¶
MPS 示例展示如何保留公开的 window_*.mps 数据文件,同时在运行时把 MPS 子集重建为 OptAgent ModelBuilder 定义,再比较 heuristic、hybrid 和 exact 路径。
https://github.com/Dongbox/optagent-examples/tree/main/examples/mps
问题说明¶
该目录面向已有 MPS 数据的迁移场景。核心问题不是手写一个新模型,而是把窗口模型转换为 OptAgent 变量、目标和约束,让后续可以接入 preset、后端选择和统一结果输出。
目录结构¶
examples/mps/
README.md
mps_builder.py
solve_window.py
resource_flow_heuristic_preset.json
resource_flow_hybrid_preset.json
resource_flow_exact_preset.json
window_0.mps
window_1.mps
...
代码示例¶
from pathlib import Path
from optagent import Orchestrator
program = build_program_from_mps(Path("examples/mps/window_0.mps"))
result = Orchestrator().run(program, preset="resource_flow_heuristic")
print(result.final_solution.status)
完整转换逻辑在 mps_builder.py,命令入口在 solve_window.py。
运行¶
PYTHONPATH=. python examples/mps/solve_window.py --window 0 --summary-only
PYTHONPATH=. python examples/mps/solve_window.py --window 0
PYTHONPATH=. python examples/mps/solve_window.py --window 0 --mode hybrid
PYTHONPATH=. python examples/mps/solve_window.py --window 0 --mode exact --backend highs_native
注意事项¶
- 先用
--summary-only查看变量、约束和系数规模,再执行完整求解。 - 默认 heuristic preset 不调用 MP backend;
--backend主要影响hybrid和exact模式。 --backend highs_native需要highspy,--backend mathopt_mp通常需要ortools。- MPS 文件必须是可公开的小型或脱敏数据。
钢卷排序示例¶
钢卷排序示例用于展示序列优化问题:给定钢卷属性和直接焊接规则,选择排序以减少不兼容切换。该主题同时提供黑盒评分模型、DAG 路径模型、默认求解 profile 和搜索归因脚本。
https://github.com/Dongbox/optagent-examples/tree/main/examples/steel
问题说明¶
- 黑盒版本把焊接兼容性和切换次数封装为 Python 评分函数。
- DAG 版本把部分路径结构显式建模,便于精确或混合路径接入。
- 搜索归因脚本用于区分初始构造解、tabu/GA 搜索和扰动恢复分别贡献了多少质量。
目录结构¶
examples/steel/
README.md
steel_domain.py
blackbox_model.py
dag_path_model.py
solve_profiles.py
run_blackbox.py
run_dag.py
steel_blackbox_sequence.py
steel_dag_sequence.py
search_attribution.py
data/
steel_coils.json
scripts/
run_steel_experiments.py
run_steel_search_attribution.py
代码示例¶
from optagent import Orchestrator
from steel.blackbox_model import build_program
from steel.solve_profiles import build_blackbox_config
from steel.steel_domain import load_steel_instances
instance = load_steel_instances()["bundled_head40"]
program, sequence_node_id = build_program(instance)
result = Orchestrator().run(
program,
config=build_blackbox_config(mode="preset", budget_iterations=200, generation_limit=10, seed=0),
)
best_sequence = result.final_solution.variable_values[sequence_node_id]
运行¶
PYTHONPATH=. python examples/steel/steel_blackbox_sequence.py --instance toy --mode preset
PYTHONPATH=. python examples/steel/steel_dag_sequence.py --instance toy --mode preset
PYTHONPATH=. python examples/steel/steel_dag_sequence.py --instance bundled_head40 --mode exact
PYTHONPATH=. python scripts/run_steel_experiments.py --instance bundled_head40 --mode preset --seed 0 --seed 1
PYTHONPATH=. python scripts/run_steel_search_attribution.py --instance bundled --search-seed 11
注意事项¶
- 可用公开实例包括
toy、bundled_head40和bundled。 - 大实例的解质量和运行时间受 seed、预算和 profile 影响,评估时应记录命令参数。
- 兼容入口
examples/blackbox/steel_transition_sequence.py保留,但推荐使用examples/steel/下的入口。
资源流示例¶
资源流示例展示多阶段合同流、机器节点、物流边、库存和需求滚动之间的关系。该目录提供 CP-SAT 取向 formulation、MILP formulation、统一求解入口和滚动窗口 workflow。
https://github.com/Dongbox/optagent-examples/tree/main/examples/resource_flow
问题说明¶
同一份公开 case snapshot 可以用两种方式建模:
cpformulation:强调离散流量、整数约束和 CP-SAT 友好的重述。milpformulation:强调代数线性约束,便于走mathopt_mp或highs_native。
这类示例适合观察“业务工作流层”和“OptAgent 求解层”的边界:滚动窗口状态推进留在示例 workflow 中,单窗口模型交给 OptAgent。
目录结构¶
examples/resource_flow/
README.md
schema.py
case_loader.py
cp_builder.py
milp_builder.py
solve_case.py
rolling.py
original_bridge.py
compare_original.py
data/
zj_case_cp_sat_p3.json.gz
zj_case_mathopt_p3.json.gz
代码示例¶
from examples.resource_flow.case_loader import load_case
from examples.resource_flow.cp_builder import build_single_window_program
from optagent import Orchestrator
case = load_case(case_name="zj", formulation="cp", planning_period=3)
built = build_single_window_program(case.config, case.model_input, modeling_period=3, k=0)
result = Orchestrator().run(built.program)
运行¶
PYTHONPATH=. python examples/resource_flow/solve_case.py --formulation cp --summary-only
PYTHONPATH=. python examples/resource_flow/solve_case.py --formulation milp --summary-only
PYTHONPATH=. python examples/resource_flow/solve_case.py --formulation cp --mode hybrid
PYTHONPATH=. python examples/resource_flow/solve_case.py --formulation milp --mode exact --backend highs_native
注意事项¶
- 默认 bundled 路径只依赖公开仓库内的数据文件。
compare_original.py属于迁移比较脚本,需要额外的原始工程环境;普通公开运行不需要它。- 完整大实例 exact solve 时间和后端有关,建议先用
--summary-only评估规模。 - 滚动窗口当前是示例 workflow,不是核心
Orchestrator的内置功能。
MG 迁移示例¶
MG 示例是 SQLite 驱动的 APS 兼容迁移示例。它保留原 program/ 风格的数据流:preprocess 生成输入表,OptAgent 替代模型阶段,postprocess 写回兼容输出表。
https://github.com/Dongbox/optagent-examples/tree/main/examples/mg
问题说明¶
MG 当前采用 sequence-blackbox 建模:
- 从 SQLite 的
i_*和t_*表加载订单、工艺、规则成本和初始序列。 - 用
sequence_var(...)表达订单顺序。 - 用 Python 规则函数重算 MG 主要 rule cost,并通过
external_call(...)作为目标。 - 输出
o_mg_optagent_*表,便于与原 APS 输出和后处理流程对齐。
目录结构¶
examples/mg/
README.md
program/
main.py
data/
20260407000000.db
model/
model.py
rules.py
search.py
reports.py
scripts/
preprocess/
transformer.py
data.py
tables/
postprocess/
postprocess.py
tests/
test_examples_mg.py
代码示例¶
from pathlib import Path
from mg.program.model.reports import run_production_case
payload = run_production_case(
Path("examples/mg/program/data/20260407000000.db"),
dry_run=True,
)
print(payload["manifest"]["status"])
运行¶
MG 示例需要 APS 兼容运行环境。正式运行请使用 Python 3.9,并安装管理员提供的 APS wheel。APS wheel 不随 optagent-examples 公开分发;需要向管理员申请匹配版本的 wheel 文件和安装权限。
python3.9 -m venv .venv-mg
source .venv-mg/bin/activate
python -m pip install <admin-provided-aps-wheel.whl>
python -m pip install -r requirements-dev.txt
PYTHONPATH=. python examples/mg/program/main.py examples/mg/program/data/20260407000000.db
注意事项¶
- APS preprocess 失败时,MG 示例会停止后续模型和 postprocess 阶段。
- 生产入口只接收一个 SQLite 路径,preprocess 输入、模型输出、兼容输出和 postprocess 输出都写在同一个 SQLite 文件中。
program/model/reports.py提供 parity、structured edge、search replacement 和生产输出报告 API。- 当前模型以黑盒 Python scorer 为权威目标;规则尚未全部降低为结构化 DAG 约束。
- 全量生产切换需要验证上游
t_connectables生成、APS wheel 版本和全尺寸回归 case。