diff --git a/templates/ti-cl4/recipe-optim/README.md b/templates/ti-cl4/recipe-optim/README.md index d065ae5..0cf7756 100644 --- a/templates/ti-cl4/recipe-optim/README.md +++ b/templates/ti-cl4/recipe-optim/README.md @@ -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 能力点）。 ## 设计 diff --git a/templates/ti-cl4/recipe-optim/_sanity_check.py b/templates/ti-cl4/recipe-optim/_sanity_check.py index eac776c..45078c5 100644 --- a/templates/ti-cl4/recipe-optim/_sanity_check.py +++ b/templates/ti-cl4/recipe-optim/_sanity_check.py @@ -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 diff --git a/templates/ti-cl4/recipe-optim/solver.py b/templates/ti-cl4/recipe-optim/solver.py new file mode 100644 index 0000000..09bbc82 --- /dev/null +++ b/templates/ti-cl4/recipe-optim/solver.py @@ -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) diff --git a/templates/ti-cl4/recipe-optim/tests/test_solver.py b/templates/ti-cl4/recipe-optim/tests/test_solver.py new file mode 100644 index 0000000..ab73846 --- /dev/null +++ b/templates/ti-cl4/recipe-optim/tests/test_solver.py @@ -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()