254 lines
11 KiB
Python
254 lines
11 KiB
Python
# -*- 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}目标")
|