# -*- 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}目标")