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 |
启发式终止模式。 |