feat: 完成 issue #79 [Ti-2] 配方优化求解器集成(网格枚举+坐标下降轻量求解器+solve统一入口,求解器无关契约)

This commit is contained in:
2026-08-05 04:00:57 +08:00
parent 2aa70d078b
commit c42846e18a
4 changed files with 524 additions and 3 deletions
+4 -2
View File
@@ -10,9 +10,11 @@
- `problem.py` — 优化问题建模(**#78**):决策变量 / 目标 / 约束的声明式规格 +
校验 + 可行性判定 + 零依赖 YAML 子集加载。
- `solver.py` — 求解器集成(**#79**):`SolverConfig` + `Solution` + 网格枚举/
坐标下降轻量求解器 + `solve()` 统一入口,求解器无关契约。
- `config/recipe_optim.template.yaml` — Template-Ti 配方优化模板资产。
- `tests/` — 单元测试(`python -m unittest discover -s tests`)。
- `_sanity_check.py` — 部署期一键自检(5 能力点)。
- `tests/` — 单元测试(`python -m unittest discover -s tests`,46 用例)。
- `_sanity_check.py` — 部署期一键自检(6 能力点)。
## 设计
@@ -8,6 +8,7 @@ import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from problem import ConstraintKind, OptimizationProblem, load_problem # noqa: E402
from solver import SolverConfig, solve # noqa: E402
CONFIG = os.path.join(os.path.dirname(os.path.abspath(__file__)),
"config", "recipe_optim.template.yaml")
@@ -48,12 +49,19 @@ def main() -> int:
if [v.name for v in rt.variables] != [v.name for v in p.variables]:
failures.append("序列化往返丢失变量")
# 6) 求解器(#79)端到端:加载模板后能求出可行解
sol = solve(p, SolverConfig(grid_steps=7, max_combinations=200000))
if not sol.feasible:
failures.append(f"求解器未求出可行解: {sol.message}")
if sol.strategy != "grid":
failures.append(f"求解策略非 grid: {sol.strategy}")
if failures:
print("❌ recipe-optim 自检失败:")
for f in failures:
print(" -", f)
return 1
print("✅ recipe-optim 自检通过(5 能力点)")
print("✅ recipe-optim 自检通过(6 能力点)")
return 0
+304
View File
@@ -0,0 +1,304 @@
# -*- 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)
@@ -0,0 +1,207 @@
# -*- coding: utf-8 -*-
"""Ti-2 配方优化求解器集成 单元测试(Issue #79)。
覆盖:
- 求解器配置(grid_steps/max_combinations 合法性);
- 候选取值生成(bounds 等分/integer 去重/choices 枚举);
- 网格求解(全局最优、可行性、target 达成、无可行解降级);
- 坐标下降(规模超限降级、收敛);
- solve 统一入口(自动选策略、静态校验失败、空问题);
- Solution 序列化;
- 加载 #78 模板后端到端求解。
"""
import os
import sys
import unittest
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import _bootstrap # noqa: E402 挂载 recipe_optim 包
from recipe_optim.problem import ( # noqa: E402
ConstraintKind,
ConstraintSpec,
DecisionVariable,
DomainKind,
ObjectiveSpec,
ObjectiveTerm,
OptimizationProblem,
Sense,
load_problem,
)
from recipe_optim.solver import ( # noqa: E402
SolverConfig,
SolverError,
Solution,
candidate_values,
solve,
)
CONFIG_PATH = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"config", "recipe_optim.template.yaml",
)
def _toy_problem() -> OptimizationProblem:
"""minimize -(x+y),x∈[0,4] 5 点,y∈[0,4] 5 点 → 网格 25 组合。"""
return OptimizationProblem(
variables=[
DecisionVariable("x", DomainKind.BOUNDS, bounds=(0.0, 4.0)),
DecisionVariable("y", DomainKind.BOUNDS, bounds=(0.0, 4.0)),
],
objective=ObjectiveSpec(Sense.MAXIMIZE,
terms=[ObjectiveTerm("x", 1.0), ObjectiveTerm("y", 1.0)]),
)
class TestSolverConfig(unittest.TestCase):
def test_defaults(self):
c = SolverConfig()
self.assertGreaterEqual(c.grid_steps, 2)
self.assertGreaterEqual(c.max_combinations, 1)
def test_invalid_grid_steps(self):
with self.assertRaises(SolverError):
SolverConfig(grid_steps=1)
def test_invalid_max_combinations(self):
with self.assertRaises(SolverError):
SolverConfig(max_combinations=0)
class TestCandidateValues(unittest.TestCase):
def test_bounds_grid(self):
v = DecisionVariable("t", DomainKind.BOUNDS, bounds=(0.0, 4.0))
vals = candidate_values(v, SolverConfig(grid_steps=5))
self.assertEqual(vals[0], 0.0)
self.assertEqual(vals[-1], 4.0)
self.assertEqual(len(vals), 5)
def test_integer_dedupe(self):
v = DecisionVariable("n", DomainKind.BOUNDS, bounds=(0.0, 4.0), integer=True)
vals = candidate_values(v, SolverConfig(grid_steps=5))
self.assertEqual(vals, [0, 1, 2, 3, 4])
def test_choices_enum(self):
v = DecisionVariable("c", DomainKind.CHOICES, choices=["A", "B", "C"])
self.assertEqual(candidate_values(v, SolverConfig()), ["A", "B", "C"])
self.assertEqual(candidate_values(v, SolverConfig(enumerate_choices=False)), ["A"])
class TestSolveGrid(unittest.TestCase):
def test_global_optimum_maximize(self):
p = _toy_problem()
sol = solve(p, SolverConfig(grid_steps=5))
self.assertTrue(sol.feasible)
# 最优 x=y=4 → obj=8
self.assertAlmostEqual(sol.objective_value, 8.0)
self.assertEqual(sol.assignment["x"], 4.0)
self.assertEqual(sol.assignment["y"], 4.0)
self.assertEqual(sol.strategy, "grid")
def test_minimize(self):
p = OptimizationProblem(
variables=[DecisionVariable("x", DomainKind.BOUNDS, bounds=(0.0, 4.0))],
objective=ObjectiveSpec(Sense.MINIMIZE, terms=[ObjectiveTerm("x", 1.0)]),
)
sol = solve(p, SolverConfig(grid_steps=5))
self.assertTrue(sol.feasible)
self.assertEqual(sol.assignment["x"], 0.0)
self.assertAlmostEqual(sol.objective_value, 0.0)
def test_target_met(self):
p = OptimizationProblem(
variables=[DecisionVariable("x", DomainKind.BOUNDS, bounds=(0.0, 10.0))],
objective=ObjectiveSpec(Sense.MAXIMIZE, target_value=8.0,
terms=[ObjectiveTerm("x", 1.0)]),
)
sol = solve(p, SolverConfig(grid_steps=11))
self.assertTrue(sol.feasible)
self.assertTrue(sol.target_met) # x=10 >= 8
def test_target_not_met(self):
p = OptimizationProblem(
variables=[DecisionVariable("x", DomainKind.BOUNDS, bounds=(0.0, 5.0))],
objective=ObjectiveSpec(Sense.MAXIMIZE, target_value=99.0,
terms=[ObjectiveTerm("x", 1.0)]),
)
sol = solve(p, SolverConfig(grid_steps=6))
self.assertTrue(sol.feasible)
self.assertFalse(sol.target_met)
def test_no_feasible_solution(self):
# box 收紧到与域不交 → 无可行
p = OptimizationProblem(
variables=[DecisionVariable("x", DomainKind.BOUNDS, bounds=(0.0, 1.0))],
constraints=[ConstraintSpec(ConstraintKind.BOX, variable="x", bounds=(5.0, 6.0))],
objective=ObjectiveSpec(Sense.MAXIMIZE, terms=[ObjectiveTerm("x", 1.0)]),
)
sol = solve(p, SolverConfig(grid_steps=3))
self.assertFalse(sol.feasible)
self.assertIn("无可行解", sol.message)
class TestCoordinateDescent(unittest.TestCase):
def test_falls_back_when_grid_too_large(self):
# 三个变量 × grid_steps=5 = 125;设 max_combinations=10 → 降级
p = OptimizationProblem(
variables=[
DecisionVariable(f"v{i}", DomainKind.BOUNDS, bounds=(0.0, 4.0))
for i in range(3)
],
objective=ObjectiveSpec(Sense.MAXIMIZE,
terms=[ObjectiveTerm(f"v{i}", 1.0) for i in range(3)]),
)
sol = solve(p, SolverConfig(grid_steps=5, max_combinations=10))
self.assertEqual(sol.strategy, "coordinate_descent")
# 坐标下降应能爬到各维上界附近(贪心可收敛到此线性目标的全局最优)
self.assertTrue(sol.feasible)
self.assertAlmostEqual(sol.objective_value, 12.0, places=6)
class TestSolveEntry(unittest.TestCase):
def test_validate_failure_returns_infeasible(self):
p = OptimizationProblem(
variables=[DecisionVariable("x", DomainKind.BOUNDS, bounds=(0.0, 1.0))],
objective=ObjectiveSpec(Sense.MAXIMIZE,
terms=[ObjectiveTerm("ghost", 1.0)]),
)
sol = solve(p, SolverConfig())
self.assertFalse(sol.feasible)
self.assertEqual(sol.strategy, "validate")
self.assertIn("校验失败", sol.message)
def test_empty_problem_trivially_feasible(self):
p = OptimizationProblem()
sol = solve(p, SolverConfig())
self.assertTrue(sol.feasible)
self.assertEqual(sol.strategy, "empty")
def test_solution_to_dict(self):
p = _toy_problem()
sol = solve(p, SolverConfig(grid_steps=3))
d = sol.to_dict()
self.assertIn("assignment", d)
self.assertIn("objective_value", d)
self.assertIn("evaluated", d)
self.assertTrue(d["feasible"])
class TestEndToEndFromTemplate(unittest.TestCase):
def test_solve_loaded_problem(self):
p = load_problem(CONFIG_PATH)
sol = solve(p, SolverConfig(grid_steps=7, max_combinations=200000))
# 模板含 4 变量;7^3 * 3 = 1029 组合 < 上限 → 网格
self.assertEqual(sol.strategy, "grid")
self.assertTrue(sol.feasible)
# 取值应满足 box 收紧(clf_temp∈[820,900])与 forbidden(非 C@910)
self.assertGreaterEqual(sol.assignment["clf_temp"], 820 - 1e-6)
self.assertLessEqual(sol.assignment["clf_temp"], 900 + 1e-6)
self.assertFalse(sol.assignment["catalyst"] == "C"
and sol.assignment.get("clf_temp") == 910)
# evaluated>0
self.assertGreater(sol.evaluated, 0)
if __name__ == "__main__":
unittest.main()