305 lines
12 KiB
Python
305 lines
12 KiB
Python
|
# -*- coding: utf-8 -*-
|
|||
|
|
"""Ti-2 配方动态优化 · 求解器集成（Issue #79 / PRD 5.3 ②）。
|
|||
|
|
|
|||
|
|
承接 #78 的 ``OptimizationProblem``：把"问题模型"喂给**求解器**，产出满足全部
|
|||
|
|
约束、逼近目标最优的**配方/参数取值**，并给出可解释的求解报告。
|
|||
|
|
|
|||
|
|
PRD 设计口径
|
|||
|
|
------------
|
|||
|
|
- 架构表（PRD §5.3）：``出：参数/配方建议``；``高（需闭环反馈）``。
|
|||
|
|
- 模板化技术路径：默认「固定主干 + 可配置超参」；新增结构走插件注册而非改内核。
|
|||
|
|
- 风险表：二期交付（数据门槛高）。故本期求解器采用**纯标准库、零第三方依赖**的
|
|||
|
|
轻量策略（坐标下降 + 网格采样），数据就绪/精度不足时可注入更强的外部求解器
|
|||
|
|
（PuLP/scipy/optuna，走 #78 预留的 ``solve`` 扩展点），**内核不绑优化库**。
|
|||
|
|
|
|||
|
|
本模块交付
|
|||
|
|
----------
|
|||
|
|
1. **``SolverConfig``**：求解策略声明式配置（网格粒度、迭代轮数、随机种子、
|
|||
|
|
是否枚举离散选择），对齐 PRD「超参包驱动」。
|
|||
|
|
2. **``Solution``**：求解结果（取值 ``assignment``、目标值、是否可行、是否达成
|
|||
|
|
``target_value``、迭代轨迹、违反约束枚举），为 #81 可解释建议提供结构化输入。
|
|||
|
|
3. **``GridSolver``**：确定性网格 + 坐标下降求解器（纯标准库）：
|
|||
|
|
- 连续域变量按 ``grid_steps`` 等分离散化；
|
|||
|
|
- 离散域变量枚举 ``choices``；
|
|||
|
|
- 笛卡尔积里筛可行解、按目标 ``sense`` 选最优（全局最优保证）；
|
|||
|
|
- 规模过大时退化为坐标下降（贪心）保可用性（``max_combinations`` 阈值）。
|
|||
|
|
4. **``solve(problem, config=None)``**：统一入口，便于 #80/#81 调用。
|
|||
|
|
|
|||
|
|
设计要点
|
|||
|
|
--------
|
|||
|
|
- **确定性可复现**：``random_seed`` 固定，同输入同输出（对齐 PRD"结论可复现"）。
|
|||
|
|
- **可行优先**：无任何可行解时返回 ``feasible=False`` 的 Solution，不抛异常，
|
|||
|
|
便于上层降级（对齐 PRD"可用性 ≥ 99.8%"）。
|
|||
|
|
- **求解器无关契约**：``solve`` 是薄入口，可被外部更强求解器替换；本模块的
|
|||
|
|
``Solution`` 结构即外部求解器需返回的契约。
|
|||
|
|
"""
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import itertools
|
|||
|
|
import math
|
|||
|
|
import random
|
|||
|
|
from dataclasses import dataclass, field
|
|||
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|||
|
|
|
|||
|
|
# 复用 #78 的问题模型。作为包成员导入用 ``recipe_optim.problem``；当本文件被直接
|
|||
|
|
# 执行（与 problem.py 同目录）时回落到裸名 ``problem``。
|
|||
|
|
try: # pragma: no cover - 分支取决于导入方式
|
|||
|
|
from recipe_optim.problem import ( # type: ignore[import-not-found]
|
|||
|
|
ConstraintSpec,
|
|||
|
|
DecisionVariable,
|
|||
|
|
DomainKind,
|
|||
|
|
ObjectiveSpec,
|
|||
|
|
OptimizationProblem,
|
|||
|
|
Sense,
|
|||
|
|
_is_num,
|
|||
|
|
)
|
|||
|
|
except ImportError: # pragma: no cover
|
|||
|
|
from problem import ( # type: ignore[import-not-found,no-redef]
|
|||
|
|
ConstraintSpec,
|
|||
|
|
DecisionVariable,
|
|||
|
|
DomainKind,
|
|||
|
|
ObjectiveSpec,
|
|||
|
|
OptimizationProblem,
|
|||
|
|
Sense,
|
|||
|
|
_is_num,
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class SolverError(ValueError):
|
|||
|
|
"""求解器配置或执行错误（网格粒度非法、变量规模溢出等）。"""
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass
|
|||
|
|
class SolverConfig:
|
|||
|
|
"""求解策略声明式配置（对齐 PRD 超参包驱动）。"""
|
|||
|
|
|
|||
|
|
grid_steps: int = 11 # 连续域每个变量等分点数（含端点）
|
|||
|
|
max_combinations: int = 200000 # 笛卡尔积规模上限，超过则退化为坐标下降
|
|||
|
|
random_seed: int = 20260805 # 固定随机种子，保证确定性可复现
|
|||
|
|
enumerate_choices: bool = True # 是否完整枚举离散 choices（False 时取首个）
|
|||
|
|
|
|||
|
|
def __post_init__(self) -> None:
|
|||
|
|
if self.grid_steps < 2:
|
|||
|
|
raise SolverError("grid_steps 必须 ≥ 2（至少含两端点）")
|
|||
|
|
if self.max_combinations < 1:
|
|||
|
|
raise SolverError("max_combinations 必须 ≥ 1")
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass
|
|||
|
|
class Solution:
|
|||
|
|
"""求解结果（#81 可解释建议的结构化输入）。"""
|
|||
|
|
|
|||
|
|
assignment: Dict[str, Any] = field(default_factory=dict)
|
|||
|
|
objective_value: float = 0.0
|
|||
|
|
feasible: bool = False
|
|||
|
|
target_met: bool = False
|
|||
|
|
violated: List[ConstraintSpec] = field(default_factory=list)
|
|||
|
|
iterations: int = 0
|
|||
|
|
evaluated: int = 0
|
|||
|
|
strategy: str = "" # "grid" / "coordinate_descent"
|
|||
|
|
message: str = ""
|
|||
|
|
|
|||
|
|
def to_dict(self) -> Dict[str, Any]:
|
|||
|
|
return {
|
|||
|
|
"assignment": dict(self.assignment),
|
|||
|
|
"objective_value": self.objective_value,
|
|||
|
|
"feasible": self.feasible,
|
|||
|
|
"target_met": self.target_met,
|
|||
|
|
"violated": [c.to_dict() for c in self.violated],
|
|||
|
|
"iterations": self.iterations,
|
|||
|
|
"evaluated": self.evaluated,
|
|||
|
|
"strategy": self.strategy,
|
|||
|
|
"message": self.message,
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
# 变量取值候选生成
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
|
|||
|
|
def candidate_values(var: DecisionVariable, config: SolverConfig) -> List[Any]:
|
|||
|
|
"""为单个变量生成求解候选取值集合。"""
|
|||
|
|
if var.kind == DomainKind.CHOICES:
|
|||
|
|
return list(var.choices) if config.enumerate_choices else [var.choices[0]]
|
|||
|
|
# bounds 连续域：等分离散化
|
|||
|
|
if var.bounds is None:
|
|||
|
|
return []
|
|||
|
|
low, high = var.bounds
|
|||
|
|
step = (high - low) / (config.grid_steps - 1)
|
|||
|
|
vals = [low + i * step for i in range(config.grid_steps)]
|
|||
|
|
if var.integer:
|
|||
|
|
vals = [float(round(v)) for v in vals]
|
|||
|
|
# 去重保序
|
|||
|
|
seen: set = set()
|
|||
|
|
uniq: List[Any] = []
|
|||
|
|
for v in vals:
|
|||
|
|
iv = int(v)
|
|||
|
|
if iv not in seen:
|
|||
|
|
seen.add(iv)
|
|||
|
|
uniq.append(iv)
|
|||
|
|
return uniq
|
|||
|
|
return vals
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _grid_size(problem: OptimizationProblem, config: SolverConfig) -> int:
|
|||
|
|
total = 1
|
|||
|
|
for v in problem.variables:
|
|||
|
|
total *= len(candidate_values(v, config))
|
|||
|
|
return total
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
# 求解器
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _better(new: float, best: float, sense: Sense) -> bool:
|
|||
|
|
"""判断 new 是否比 best 更优。"""
|
|||
|
|
if sense == Sense.MAXIMIZE:
|
|||
|
|
return new > best
|
|||
|
|
return new < best
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _initial_objective(sense: Sense) -> float:
|
|||
|
|
return -math.inf if sense == Sense.MAXIMIZE else math.inf
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _solve_grid(problem: OptimizationProblem, config: SolverConfig) -> Solution:
|
|||
|
|
"""完整网格枚举：笛卡尔积里筛可行、选最优（全局最优保证）。"""
|
|||
|
|
rng = random.Random(config.random_seed)
|
|||
|
|
per_var = [candidate_values(v, config) for v in problem.variables]
|
|||
|
|
names = [v.name for v in problem.variables]
|
|||
|
|
sense = problem.objective.sense
|
|||
|
|
best_obj = _initial_objective(sense)
|
|||
|
|
best_assign: Optional[Dict[str, Any]] = None
|
|||
|
|
evaluated = 0
|
|||
|
|
iterations = 0
|
|||
|
|
# 为控制内存，逐组合判定，不一次性 materialize
|
|||
|
|
for combo in itertools.product(*per_var):
|
|||
|
|
evaluated += 1
|
|||
|
|
iterations += 1
|
|||
|
|
assignment = dict(zip(names, combo))
|
|||
|
|
if not problem.is_feasible(assignment):
|
|||
|
|
continue
|
|||
|
|
obj = problem.objective.evaluate(assignment)
|
|||
|
|
if best_assign is None or _better(obj, best_obj, sense):
|
|||
|
|
best_obj = obj
|
|||
|
|
best_assign = assignment
|
|||
|
|
feasible = best_assign is not None
|
|||
|
|
return _build_solution(problem, config, best_assign or {}, best_obj,
|
|||
|
|
feasible, iterations, evaluated, "grid",
|
|||
|
|
"网格枚举完成" if feasible else "无可行解（约束过紧或域为空）")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _solve_coordinate_descent(
|
|||
|
|
problem: OptimizationProblem, config: SolverConfig
|
|||
|
|
) -> Solution:
|
|||
|
|
"""坐标下降：固定其余变量、逐维选当前最优取值（贪心，规模过大时降级用）。
|
|||
|
|
|
|||
|
|
从初值（``initial`` 缺省取域中点）出发，反复扫描各变量、在候选值里取使目标
|
|||
|
|
最优且保持可行者；迭代至收敛或达 ``max_rounds``。非全局最优，但保可用性。
|
|||
|
|
"""
|
|||
|
|
sense = problem.objective.sense
|
|||
|
|
names = [v.name for v in problem.variables]
|
|||
|
|
per_var = {v.name: candidate_values(v, config) for v in problem.variables}
|
|||
|
|
# 初值
|
|||
|
|
assignment: Dict[str, Any] = {}
|
|||
|
|
for v in problem.variables:
|
|||
|
|
if v.initial is not None and v.contains(v.initial):
|
|||
|
|
assignment[v.name] = v.initial
|
|||
|
|
elif v.kind == DomainKind.CHOICES and v.choices:
|
|||
|
|
assignment[v.name] = v.choices[0]
|
|||
|
|
elif v.bounds is not None:
|
|||
|
|
assignment[v.name] = (v.bounds[0] + v.bounds[1]) / 2.0
|
|||
|
|
else: # pragma: no cover - 防御
|
|||
|
|
assignment[v.name] = None
|
|||
|
|
max_rounds = max(3, len(names))
|
|||
|
|
evaluated = 0
|
|||
|
|
iterations = 0
|
|||
|
|
for _round in range(max_rounds):
|
|||
|
|
improved = False
|
|||
|
|
for name in names:
|
|||
|
|
cur_best = assignment[name]
|
|||
|
|
cur_assign = dict(assignment)
|
|||
|
|
cur_obj = problem.objective.evaluate(cur_assign) if problem.is_feasible(cur_assign) else None
|
|||
|
|
best_val = cur_best
|
|||
|
|
best_obj = cur_obj if cur_obj is not None else _initial_objective(sense)
|
|||
|
|
for cand in per_var[name]:
|
|||
|
|
evaluated += 1
|
|||
|
|
trial = dict(assignment)
|
|||
|
|
trial[name] = cand
|
|||
|
|
if not problem.is_feasible(trial):
|
|||
|
|
continue
|
|||
|
|
obj = problem.objective.evaluate(trial)
|
|||
|
|
if cur_obj is None or _better(obj, best_obj, sense):
|
|||
|
|
best_obj = obj
|
|||
|
|
best_val = cand
|
|||
|
|
if best_val != cur_best:
|
|||
|
|
assignment[name] = best_val
|
|||
|
|
improved = True
|
|||
|
|
iterations += 1
|
|||
|
|
if not improved:
|
|||
|
|
break
|
|||
|
|
feasible = problem.is_feasible(assignment)
|
|||
|
|
final_obj = problem.objective.evaluate(assignment) if feasible else 0.0
|
|||
|
|
return _build_solution(problem, config, assignment, final_obj, feasible,
|
|||
|
|
iterations, evaluated, "coordinate_descent",
|
|||
|
|
"坐标下降完成" if feasible else "坐标下降未找到可行解")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _build_solution(
|
|||
|
|
problem: OptimizationProblem,
|
|||
|
|
config: SolverConfig,
|
|||
|
|
assignment: Dict[str, Any],
|
|||
|
|
obj: float,
|
|||
|
|
feasible: bool,
|
|||
|
|
iterations: int,
|
|||
|
|
evaluated: int,
|
|||
|
|
strategy: str,
|
|||
|
|
message: str,
|
|||
|
|
) -> Solution:
|
|||
|
|
violated = problem.violated_constraints(assignment) if assignment else []
|
|||
|
|
target_met = False
|
|||
|
|
if feasible and problem.objective.target_value is not None:
|
|||
|
|
if problem.objective.sense == Sense.MAXIMIZE:
|
|||
|
|
target_met = obj >= problem.objective.target_value
|
|||
|
|
else:
|
|||
|
|
target_met = obj <= problem.objective.target_value
|
|||
|
|
elif feasible and problem.objective.target_value is None:
|
|||
|
|
target_met = True # 未设达标量则视为达成
|
|||
|
|
return Solution(
|
|||
|
|
assignment=assignment,
|
|||
|
|
objective_value=obj,
|
|||
|
|
feasible=feasible,
|
|||
|
|
target_met=target_met,
|
|||
|
|
violated=violated,
|
|||
|
|
iterations=iterations,
|
|||
|
|
evaluated=evaluated,
|
|||
|
|
strategy=strategy,
|
|||
|
|
message=message,
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def solve(problem: OptimizationProblem,
|
|||
|
|
config: Optional[SolverConfig] = None) -> Solution:
|
|||
|
|
"""统一求解入口。
|
|||
|
|
|
|||
|
|
自动按规模选择策略：网格规模 ≤ ``max_combinations`` 用全局网格枚举，
|
|||
|
|
否则退化为坐标下降（保可用性）。先做静态校验，校验失败直接返回不可行解。
|
|||
|
|
"""
|
|||
|
|
cfg = config or SolverConfig()
|
|||
|
|
# 静态校验
|
|||
|
|
errs = problem.validate()
|
|||
|
|
if errs:
|
|||
|
|
return Solution(feasible=False, strategy="validate",
|
|||
|
|
message="问题校验失败: " + "; ".join(errs))
|
|||
|
|
# 空问题：无可调变量
|
|||
|
|
if not problem.variables:
|
|||
|
|
return Solution(feasible=True, target_met=True, strategy="empty",
|
|||
|
|
message="无决策变量，视为平凡可行")
|
|||
|
|
size = _grid_size(problem, cfg)
|
|||
|
|
if size <= cfg.max_combinations:
|
|||
|
|
return _solve_grid(problem, cfg)
|
|||
|
|
return _solve_coordinate_descent(problem, cfg)
|