feat: 完成 issue #81 [Ti-2] 优化建议生成与可解释性（整合求解结果+跨工序权重+约束依据，输出可溯源建议报告）

2026-08-05 04:05:10 +08:00
parent dcbd196fe9
commit a0a77fb41f
4 changed files with 429 additions and 3 deletions
@@ -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}目标")