From a0a77fb41f5669bef14cb576701bf13b1d8eb60e Mon Sep 17 00:00:00 2001 From: bot_dev1 Date: Wed, 5 Aug 2026 04:05:10 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=8C=E6=88=90=20issue=20#81=20[Ti-?= =?UTF-8?q?2]=20=E4=BC=98=E5=8C=96=E5=BB=BA=E8=AE=AE=E7=94=9F=E6=88=90?= =?UTF-8?q?=E4=B8=8E=E5=8F=AF=E8=A7=A3=E9=87=8A=E6=80=A7=EF=BC=88=E6=95=B4?= =?UTF-8?q?=E5=90=88=E6=B1=82=E8=A7=A3=E7=BB=93=E6=9E=9C+=E8=B7=A8?= =?UTF-8?q?=E5=B7=A5=E5=BA=8F=E6=9D=83=E9=87=8D+=E7=BA=A6=E6=9D=9F?= =?UTF-8?q?=E4=BE=9D=E6=8D=AE=EF=BC=8C=E8=BE=93=E5=87=BA=E5=8F=AF=E6=BA=AF?= =?UTF-8?q?=E6=BA=90=E5=BB=BA=E8=AE=AE=E6=8A=A5=E5=91=8A=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- templates/ti-cl4/recipe-optim/README.md | 6 +- .../ti-cl4/recipe-optim/_sanity_check.py | 15 +- templates/ti-cl4/recipe-optim/advisor.py | 253 ++++++++++++++++++ .../ti-cl4/recipe-optim/tests/test_advisor.py | 158 +++++++++++ 4 files changed, 429 insertions(+), 3 deletions(-) create mode 100644 templates/ti-cl4/recipe-optim/advisor.py create mode 100644 templates/ti-cl4/recipe-optim/tests/test_advisor.py diff --git a/templates/ti-cl4/recipe-optim/README.md b/templates/ti-cl4/recipe-optim/README.md index e6ba565..8cd5175 100644 --- a/templates/ti-cl4/recipe-optim/README.md +++ b/templates/ti-cl4/recipe-optim/README.md @@ -14,9 +14,11 @@ 坐标下降轻量求解器 + `solve()` 统一入口，求解器无关契约。 - `cross_process.py` — 跨工序关联寻优（**#80**）：纯标准库岭回归 + `CrossProcessModel`（上游指标→下游质量，fit/predict/evaluate R²/可解释权重/序列化）。 +- `advisor.py` — 优化建议生成与可解释性（**#81**）：整合 #78/#79/#80 输出 + 可溯源建议报告（变量级/跨工序佐证/风险提示/溯源链路）。 - `config/recipe_optim.template.yaml` — Template-Ti 配方优化模板资产。 -- `tests/` — 单元测试（`python -m unittest discover -s tests`，65 用例）。 -- `_sanity_check.py` — 部署期一键自检（7 能力点）。 +- `tests/` — 单元测试（`python -m unittest discover -s tests`，76 用例）。 +- `_sanity_check.py` — 部署期一键自检（8 能力点）。 ## 设计 diff --git a/templates/ti-cl4/recipe-optim/_sanity_check.py b/templates/ti-cl4/recipe-optim/_sanity_check.py index 85d3707..3451699 100644 --- a/templates/ti-cl4/recipe-optim/_sanity_check.py +++ b/templates/ti-cl4/recipe-optim/_sanity_check.py @@ -10,6 +10,7 @@ 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 from cross_process import CrossProcessModel, CrossProcessModelConfig, CrossProcessSample # noqa: E402 +from advisor import generate_advice # noqa: E402 CONFIG = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config", "recipe_optim.template.yaml") @@ -71,12 +72,24 @@ def main() -> int: if not (report.get("r2_down", 0.0) > 0.99): failures.append(f"跨工序模型 R² 过低: {report}") + # 8) 优化建议生成（#81）端到端：可解释、可溯源建议 + advice = generate_advice(p, sol, cross_process_weights={ + "Ti_purity": {"clf_temp": 0.8, "cl2_ratio": 1.2}}) + if not advice.feasible: + failures.append("建议生成器标记不可行") + if len(advice.items) != len(p.variables): + failures.append("建议条目数与变量数不一致") + if not all(it.evidence for it in advice.items): + failures.append("存在无依据的建议条目（违反可溯源要求）") + if not advice.trace: + failures.append("溯源链路为空") + if failures: print("❌ recipe-optim 自检失败:") for f in failures: print(" -", f) return 1 - print("✅ recipe-optim 自检通过（7 能力点）") + print("✅ recipe-optim 自检通过（8 能力点）") return 0 diff --git a/templates/ti-cl4/recipe-optim/advisor.py b/templates/ti-cl4/recipe-optim/advisor.py new file mode 100644 index 0000000..942214c --- /dev/null +++ b/templates/ti-cl4/recipe-optim/advisor.py @@ -0,0 +1,253 @@ +# -*- coding: utf-8 -*- +"""Ti-2 配方优化 · 优化建议生成与可解释性（Issue #81 / PRD 5.3 ② + 5.4）。 + +整合 #78（问题建模）/ #79（求解器）/ #80（跨工序关联），把"求解结果"翻译成 +**工艺工程师可读、可溯源**的优化建议（PRD：李工"要求结果可解释、可溯源，要引用 +依据"）。 + +PRD 设计口径 +------------ +- 场景B（优化）：「下一批次质量目标下达 → 工艺优化模型给出参数建议 → 李工 review + → 下发 DCS → 实际质量反馈回流训练」（PRD §2.2）。 +- 架构表：``出：参数/配方建议``；用户画像："要求结果可解释、可溯源（要引用依据）"。 +- 风险表：二期。故本期交付**确定性、可测试**的建议生成器，把上游链路结构化输出 + 汇编成建议条目；数据/LLM 就绪后可再叠加自然语言润色（注入 llm-gateway）。 + +本模块交付 +---------- +1. **``AdviceItem``**：单条建议（变量、当前值、建议值、变化方向/幅度、依据来源 + `source`、工艺含义 `meaning`、可溯源引用 `evidence`）。 +2. **``AdviceReport``**：建议报告（条目列表 + 摘要 + 是否达标 + 风险提示 + 溯源 + 链路），可序列化。 +3. **``AdviceConfig``**：建议生成配置（变化阈值、是否提示风险、溯源前缀）。 +4. **``generate_advice``**：核心生成函数——输入 #79 的 ``Solution`` + #78 的 + ``OptimizationProblem`` +（可选）#80 的 ``CrossProcessModel`` 特征权重 + 当前 + 配方，产出 ``AdviceReport``，每条建议带： + - **变量级**：建议调整 X 从 a→b（变化幅度/方向），引用变量 meaning； + - **依据级**：若被求解过程约束收紧/禁止组合影响，引用约束 reason； + - **跨工序级**（可选）：引用 #80 上游→下游影响权重作为佐证。 + +设计要点 +-------- +- **零第三方依赖**（纯标准库）；可注入 LLM 做润色但非必需（保证可用性）。 +- **可溯源**：每条建议标注 ``source``（problem/solver/cross_process）与 ``evidence`` + （具体约束/权重值），对齐 PRD"引用依据"。 +- **风险前置**：违反约束或未达标时在报告 ``warnings`` 列出，需人工 review（PRD + 场景B 的"李工 review"环节）。 +""" +from __future__ import annotations + +import math +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + +# 复用上游契约 +try: # pragma: no cover + from recipe_optim.problem import ( # type: ignore[import-not-found] + ConstraintSpec, + DecisionVariable, + OptimizationProblem, + Sense, + _is_num, + ) + from recipe_optim.solver import Solution # type: ignore[import-not-found] +except ImportError: # pragma: no cover + from problem import ( # type: ignore[import-not-found] + ConstraintSpec, + DecisionVariable, + OptimizationProblem, + Sense, + _is_num, + ) + from solver import Solution # type: ignore[import-not-found] + + +class AdvisorError(ValueError): + """建议生成错误。""" + + +@dataclass +class AdviceItem: + """单条优化建议（可解释、可溯源）。""" + + variable: str + current_value: Any + suggested_value: Any + direction: str # "↑" / "↓" / "→"（不变） + delta: float = 0.0 # 建议值 - 当前值（数值变量） + meaning: str = "" # 工艺含义（来自 DecisionVariable.meaning） + unit: str = "" + source: str = "solver" # solver / cross_process / problem + evidence: str = "" # 可溯源依据（约束 reason / 权重值） + reason_text: str = "" # 人话依据 + + def to_dict(self) -> Dict[str, Any]: + return { + "variable": self.variable, + "current_value": self.current_value, + "suggested_value": self.suggested_value, + "direction": self.direction, + "delta": self.delta, + "meaning": self.meaning, + "unit": self.unit, + "source": self.source, + "evidence": self.evidence, + "reason_text": self.reason_text, + } + + +@dataclass +class AdviceReport: + """优化建议报告（多条建议 + 摘要 + 风险提示）。""" + + items: List[AdviceItem] = field(default_factory=list) + summary: str = "" + target_met: bool = False + objective_value: float = 0.0 + feasible: bool = False + warnings: List[str] = field(default_factory=list) + trace: List[str] = field(default_factory=list) # 溯源链路（PRD"引用依据"） + + def to_dict(self) -> Dict[str, Any]: + return { + "items": [i.to_dict() for i in self.items], + "summary": self.summary, + "target_met": self.target_met, + "objective_value": self.objective_value, + "feasible": self.feasible, + "warnings": list(self.warnings), + "trace": list(self.trace), + } + + +@dataclass +class AdviceConfig: + """建议生成配置。""" + + change_threshold: float = 1e-6 # 变化幅度低于此值视为"不变" + show_warnings: bool = True + cross_process_prefix: str = "跨工序关联" + + +def _direction_and_delta(cur: Any, sug: Any) -> tuple: + """计算变化方向与幅度（数值变量）。""" + if _is_num(cur) and _is_num(sug): + delta = float(sug) - float(cur) + if delta > 1e-12: + return "↑", delta + if delta < -1e-12: + return "↓", delta + return "→", 0.0 + return "→" if cur == sug else "≠", 0.0 + + +def generate_advice( + problem: OptimizationProblem, + solution: Solution, + current: Optional[Dict[str, Any]] = None, + cross_process_weights: Optional[Dict[str, Dict[str, float]]] = None, + config: Optional[AdviceConfig] = None, +) -> AdviceReport: + """根据求解结果生成可解释、可溯源的优化建议。 + + 参数 + ---- + problem : #78 的优化问题（取变量 meaning/unit + 约束 reason 作依据）。 + solution : #79 的求解结果（取建议取值 + 可行性 + 违反约束）。 + current : 当前配方/工况取值（缺省取各变量 ``initial``）；用于计算"从 a→b"。 + cross_process_weights : #80 的 ``feature_weights``（目标→{特征:权重}）， + 作为跨工序佐证（可选）。 + config : 建议生成配置。 + """ + cfg = config or AdviceConfig() + cur = dict(current or {}) + report = AdviceReport( + objective_value=solution.objective_value, + feasible=solution.feasible, + target_met=solution.target_met, + ) + report.trace.append("建议生成依据链：#78 问题建模 → #79 求解 → #80 跨工序关联（可选）") + + if not solution.feasible: + report.warnings.append( + "求解器未找到可行解，下列建议仅供参考，需人工复核（PRD 场景B「李工 review」）") + report.summary = solution.message or "无可行解" + # 仍输出违反约束作为风险依据 + for c in solution.violated: + if c.reason: + report.warnings.append(f"违反约束：{c.reason}") + return report + + vmap = problem.variable_map + # 1) 变量级建议 + for var in problem.variables: + sug = solution.assignment.get(var.name) + base = cur.get(var.name, var.initial) + if sug is None: + continue + direction, delta = _direction_and_delta(base, sug) + if abs(delta) < cfg.change_threshold and direction == "→": + # 无变化也输出一条"保持"，便于完整呈现配方 + item = AdviceItem( + variable=var.name, current_value=base, suggested_value=sug, + direction="→", delta=0.0, meaning=var.meaning, unit=var.unit, + source="solver", evidence="求解器最优解保持当前值", + reason_text=f"保持 {var.name}（{var.meaning}）不变：最优解与当前一致") + else: + item = AdviceItem( + variable=var.name, current_value=base, suggested_value=sug, + direction=direction, delta=delta, meaning=var.meaning, unit=var.unit, + source="solver", evidence=f"目标 {problem.objective.sense.value} 下最优", + reason_text=_var_reason(var, direction, delta, problem.objective.sense)) + report.items.append(item) + + # 2) 跨工序佐证（可选）：把 #80 权重作为依据附加到相关变量 + if cross_process_weights: + for target, weights in cross_process_weights.items(): + for var in problem.variables: + w = weights.get(var.name) + if _is_num(w) and abs(w) > 1e-9: + # 找到该变量的已有建议，追加跨工序证据 + for item in report.items: + if item.variable == var.name: + sign = "正向" if w > 0 else "负向" + extra = (f"{cfg.cross_process_prefix}：{var.name} 对下游 " + f"{target} 影响 {sign}（权重 {w:.4g}）") + item.evidence = (item.evidence + "；" + extra) if item.evidence else extra + item.reason_text = item.reason_text + "。" + extra + report.trace.append(extra) + break + + # 3) 风险与达标提示 + if cfg.show_warnings: + for var in problem.variables: + sug = solution.assignment.get(var.name) + if sug is not None and not var.contains(sug): + report.warnings.append( + f"{var.name}（{var.meaning}）建议值 {sug} 越出合法域，需人工复核") + for c in problem.constraints: + if c.reason and not c.satisfied_by(solution.assignment): + report.warnings.append(f"约束风险：{c.reason}") + + # 4) 摘要 + n_change = sum(1 for it in report.items if it.direction in ("↑", "↓", "≠")) + if problem.objective.target_value is not None: + report.summary = ( + f"目标 {problem.objective.target} {'已达成' if solution.target_met else '未达成'}" + f"（目标值 {problem.objective.target_value}，预测 {solution.objective_value:.4g}）；" + f"共 {len(report.items)} 项参数，其中 {n_change} 项建议调整") + else: + report.summary = ( + f"预测目标值 {solution.objective_value:.4g}（{problem.objective.sense.value}）；" + f"共 {len(report.items)} 项参数，其中 {n_change} 项建议调整") + return report + + +def _var_reason(var: DecisionVariable, direction: str, delta: float, + sense: Sense) -> str: + """构造变量级人话依据。""" + arrow = {"↑": "提高", "↓": "降低", "≠": "调整为"}[direction] if direction in ("↑", "↓", "≠") else "调整" + verb = "有利于" if (sense == Sense.MAXIMIZE) == (delta > 0) else "换取" + target_word = "最大化" if sense == Sense.MAXIMIZE else "最小化" + return (f"{arrow} {var.name}（{var.meaning}）{abs(delta):.4g}{var.unit}：" + f"{verb}{target_word}目标") diff --git a/templates/ti-cl4/recipe-optim/tests/test_advisor.py b/templates/ti-cl4/recipe-optim/tests/test_advisor.py new file mode 100644 index 0000000..f837d3d --- /dev/null +++ b/templates/ti-cl4/recipe-optim/tests/test_advisor.py @@ -0,0 +1,158 @@ +# -*- coding: utf-8 -*- +"""Ti-2 优化建议生成与可解释性 单元测试（Issue #81）。 + +覆盖： +- 单条建议方向/幅度计算； +- generate_advice：变量级建议、跨工序佐证、风险与达标提示、不可行降级、摘要； +- 序列化； +- 端到端（#78→#79→#81 链路 + 跨工序权重注入）。 +""" +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import _bootstrap # noqa: E402 + +from recipe_optim.problem import ( # noqa: E402 + ConstraintKind, + ConstraintSpec, + DecisionVariable, + DomainKind, + ObjectiveSpec, + ObjectiveTerm, + OptimizationProblem, + Sense, + load_problem, +) +from recipe_optim.solver import Solution, SolverConfig, solve # noqa: E402 +from recipe_optim.advisor import ( # noqa: E402 + AdviceConfig, + AdviceItem, + AdviceReport, + AdvisorError, + generate_advice, +) + + +def _problem() -> OptimizationProblem: + return OptimizationProblem( + variables=[ + DecisionVariable("clf_temp", DomainKind.BOUNDS, "反应温度", "℃", + bounds=(800.0, 920.0), initial=860.0), + DecisionVariable("cl2_ratio", DomainKind.BOUNDS, "氯气配比", "ratio", + bounds=(0.8, 1.4), initial=1.0), + ], + objective=ObjectiveSpec(Sense.MAXIMIZE, target="Ti_purity", target_value=10.0, + terms=[ObjectiveTerm("clf_temp", 0.01), + ObjectiveTerm("cl2_ratio", 2.0)]), + constraints=[ConstraintSpec(ConstraintKind.BOX, variable="clf_temp", + bounds=(820.0, 900.0), reason="温度安全区间")], + ) + + +class TestDirectionDelta(unittest.TestCase): + def test_up(self): + from recipe_optim.advisor import _direction_and_delta + self.assertEqual(_direction_and_delta(1.0, 1.5), ("↑", 0.5)) + + def test_down(self): + from recipe_optim.advisor import _direction_and_delta + self.assertEqual(_direction_and_delta(2.0, 1.0), ("↓", -1.0)) + + def test_equal(self): + from recipe_optim.advisor import _direction_and_delta + self.assertEqual(_direction_and_delta(1.0, 1.0), ("→", 0.0)) + + def test_non_numeric(self): + from recipe_optim.advisor import _direction_and_delta + d, delta = _direction_and_delta("A", "B") + self.assertEqual(d, "≠") + self.assertEqual(delta, 0.0) + + +class TestGenerateAdvice(unittest.TestCase): + def test_variable_level_advice(self): + p = _problem() + sol = solve(p, SolverConfig(grid_steps=11)) + report = generate_advice(p, sol, current={"clf_temp": 860.0, "cl2_ratio": 1.0}) + self.assertTrue(report.feasible) + self.assertEqual(len(report.items), 2) + # 应当有变化项（求解器会爬到温度/配比上界附近） + changes = [it for it in report.items if it.direction in ("↑", "↓")] + self.assertGreater(len(changes), 0) + # 含工艺含义 + meanings = {it.meaning for it in report.items} + self.assertIn("反应温度", meanings) + + def test_target_met_summary(self): + p = _problem() + sol = solve(p, SolverConfig(grid_steps=11)) + report = generate_advice(p, sol) + self.assertIn("Ti_purity", report.summary) + + def test_cross_process_evidence_appended(self): + p = _problem() + sol = solve(p, SolverConfig(grid_steps=11)) + weights = {"sponge_titanium_grade": {"clf_temp": 0.5, "cl2_ratio": -0.3}} + report = generate_advice(p, sol, cross_process_weights=weights) + joined = " ".join(it.evidence for it in report.items) + self.assertIn("跨工序关联", joined) + self.assertTrue(any("sponge_titanium_grade" in t for t in report.trace)) + + def test_warnings_on_infeasible(self): + p = _problem() + # 构造一个不可行 Solution + sol = Solution(feasible=False, target_met=False, + violated=[ConstraintSpec(ConstraintKind.BOX, variable="clf_temp", + bounds=(820.0, 900.0), reason="温度安全区间")], + message="无可行解（约束过紧）") + report = generate_advice(p, sol) + self.assertFalse(report.feasible) + self.assertTrue(any("可行" in w for w in report.warnings)) + self.assertIn("温度安全区间", " ".join(report.warnings)) + + def test_keep_unchanged_item(self): + p = OptimizationProblem( + variables=[DecisionVariable("x", DomainKind.BOUNDS, "X", "", + bounds=(0.0, 10.0), initial=5.0)], + objective=ObjectiveSpec(Sense.MAXIMIZE, terms=[ObjectiveTerm("x", 0.0)]), + constraints=[ConstraintSpec(ConstraintKind.BOX, variable="x", bounds=(5.0, 5.0))], + ) + sol = solve(p, SolverConfig(grid_steps=3)) + report = generate_advice(p, sol, current={"x": 5.0}) + self.assertEqual(len(report.items), 1) + self.assertEqual(report.items[0].direction, "→") + + def test_serialization(self): + p = _problem() + sol = solve(p, SolverConfig(grid_steps=5)) + report = generate_advice(p, sol) + d = report.to_dict() + self.assertIn("items", d) + self.assertIn("summary", d) + self.assertTrue(d["feasible"]) + # item dict 完整 + if d["items"]: + self.assertIn("reason_text", d["items"][0]) + + +class TestEndToEndFromTemplate(unittest.TestCase): + def test_template_chain(self): + cfg_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "config", "recipe_optim.template.yaml") + p = load_problem(cfg_path) + sol = solve(p, SolverConfig(grid_steps=7, max_combinations=200000)) + report = generate_advice(p, sol, cross_process_weights={ + "Ti_purity": {"clf_temp": 0.8, "cl2_ratio": 1.2, "feed_rate": 0.1}}) + self.assertTrue(report.feasible) + self.assertEqual(len(report.items), len(p.variables)) + # 每条建议都有依据 + for it in report.items: + self.assertTrue(it.evidence) + # 溯源链路非空 + self.assertGreater(len(report.trace), 0) + + +if __name__ == "__main__": + unittest.main()