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)
|