feat: 完成 issue #78 [Ti-2] 配方优化问题建模(决策变量/目标/约束声明式规格+求解器无关+零依赖YAML加载)

This commit is contained in:
2026-08-05 03:58:25 +08:00
parent fc4ac42007
commit 2aa70d078b
7 changed files with 1276 additions and 0 deletions
+41
View File
@@ -0,0 +1,41 @@
# Ti-2 配方动态优化(recipe-optim)
> 对应 issue #78(问题建模)/ #79(求解器集成)/ #80(跨工序寻优)/ #81(可解释建议)。
> PRD §5.3 ②「配方动态优化」:`入:质量目标 + 约束;出:参数/配方建议`,二期交付。
本目录沉淀**配方优化**的声明式模板资产 + 求解器无关的问题建模引擎,跨行业差异落
资产(YAML 模板),引擎零改动(PRD「超参包驱动」「模板化技术路径」)。
## 模块
- `problem.py` — 优化问题建模(**#78**):决策变量 / 目标 / 约束的声明式规格 +
校验 + 可行性判定 + 零依赖 YAML 子集加载。
- `config/recipe_optim.template.yaml` — Template-Ti 配方优化模板资产。
- `tests/` — 单元测试(`python -m unittest discover -s tests`)。
- `_sanity_check.py` — 部署期一键自检(5 能力点)。
## 设计
1. **决策变量 `DecisionVariable`**:`bounds`(连续区间)/`choices`(离散枚举),
带 `meaning`/`unit`/`initial`,供 #81 可解释建议引用。
2. **目标 `ObjectiveSpec`**:线性加权(min/max)+ PRD 超参包 `target` 字段。
3. **约束 `ConstraintSpec`**:统一描述 box / linear / ratio / forbidden 工艺约束,
每条带 `reason`(工艺依据,对齐 PRD"可解释、可溯源、引用依据")。
4. **问题 `OptimizationProblem`**:`validate` 聚合静态校验、`is_feasible` /
`violated_constraints` 做可行性判定,`solve` 留给 #79 注入求解器(求解器无关)。
## 与上下游的契约
- 下游 #79 求解器:消费 `OptimizationProblem`,产出满足约束的最优取值。
- 下游 #80 跨工序寻优:复用同一变量/目标/约束模型描述跨工序关联。
- 下游 #81 可解释建议:读 `violated_constraints` + 变量 `meaning` + 约束 `reason`
产出"可溯源"建议。
## 运行
```bash
python -m unittest discover -s tests # 单元测试
python _sanity_check.py # 部署期自检
```
零第三方依赖(纯标准库),与内核既有模块一致,便于离线/隔离网部署。
@@ -0,0 +1,5 @@
# -*- coding: utf-8 -*-
"""Ti-2 配方动态优化模板包(Issue #78/#79/#80/#81)。
- ``problem``:优化问题建模(变量/目标/约束),求解器无关(#78)。
"""
@@ -0,0 +1,61 @@
# -*- coding: utf-8 -*-
"""Ti-2 配方优化问题建模 自检脚本(Issue #78)。
不依赖 unittest,直接加载模板资产并做能力点断言,便于 CI / 部署期一键核对。
"""
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from problem import ConstraintKind, OptimizationProblem, load_problem # noqa: E402
CONFIG = os.path.join(os.path.dirname(os.path.abspath(__file__)),
"config", "recipe_optim.template.yaml")
def main() -> int:
failures = []
# 1) 模板能加载
p = load_problem(CONFIG)
if not isinstance(p, OptimizationProblem):
failures.append("load_problem 未返回 OptimizationProblem")
# 2) 4 类约束齐全
kinds = {c.kind for c in p.constraints}
expected = {ConstraintKind.BOX, ConstraintKind.LINEAR,
ConstraintKind.RATIO, ConstraintKind.FORBIDDEN}
if kinds != expected:
failures.append(f"约束种类不齐: {kinds} != {expected}")
# 3) 静态校验通过
errs = p.validate()
if errs:
failures.append(f"validate 未通过: {errs}")
# 4) 可行性判定:合法取值可行、禁止组合不可行
ok = p.is_feasible({"clf_temp": 850, "cl2_ratio": 1.0,
"feed_rate": 450, "catalyst": "A"})
bad = p.is_feasible({"clf_temp": 950, "cl2_ratio": 1.0,
"feed_rate": 450, "catalyst": "A"})
if not ok:
failures.append("合法取值被判为不可行")
if bad:
failures.append("越界取值(950℃)未被识别为不可行")
# 5) 序列化往返无损
rt = OptimizationProblem.from_dict(p.to_dict())
if [v.name for v in rt.variables] != [v.name for v in p.variables]:
failures.append("序列化往返丢失变量")
if failures:
print("❌ recipe-optim 自检失败:")
for f in failures:
print(" -", f)
return 1
print("✅ recipe-optim 自检通过(5 能力点)")
return 0
if __name__ == "__main__":
sys.exit(main())
@@ -0,0 +1,68 @@
# Ti-2 配方动态优化 · 优化问题模板(Issue #78 / PRD 5.3 ②)
#
# 声明式描述「给定质量目标 + 工艺约束,求最优配方/参数」:
# 入:质量目标 + 约束;出:参数/配方建议(PRD §5.3 架构表)。
# 切换行业/车间只改本模板资产,引擎零改动(PRD「超参包驱动」)。
#
# 本示例对应「氯化车间/海绵钛(Template-Ti)」二期配方优化场景:
# 在 TiCl₄ 纯度达标(≥ target_value)前提下,寻优反应温度/氯气配比,
# 同时满足工艺约束(温度上限、氯气/钛配比、禁止组合)。
problem_id: recipe_optim_ti
template: iAOP-Template-Ti
description: 氯化车间配方优化(二期,数据就绪后接 #79 求解器)
variables:
- name: clf_temp
kind: bounds
meaning: 氯化炉反应温度
unit: "℃"
bounds: [800, 920]
initial: 860
- name: cl2_ratio
kind: bounds
meaning: 氯气与高钛渣配比
unit: "ratio"
bounds: [0.8, 1.4]
initial: 1.0
- name: feed_rate
kind: bounds
meaning: 进料速率
unit: "kg/h"
bounds: [300, 600]
initial: 450
integer: false
- name: catalyst
kind: choices
meaning: 催化剂型号
choices: ["A", "B", "C"]
objective:
sense: maximize
target: Ti_purity
target_value: 99.5
description: 最大化 TiCl₄ 纯度(PRD 超参包 target 字段)
terms:
- variable: clf_temp
coefficient: 0.01
- variable: cl2_ratio
coefficient: 2.0
constraints:
- kind: box
variable: clf_temp
bounds: [820, 900]
reason: 反应温度运行安全区间(运行期收紧,低于 820 反应不充分、高于 900 副产物激增)
- kind: linear
coefficients: {clf_temp: 1.0, feed_rate: -0.5}
op: "<="
rhs: 700
reason: 温度-进料耦合上限(防止局部过热)
- kind: ratio
numerator: cl2_ratio
denominator: cl2_ratio
op: ">="
value: 0.0
reason: 配比非负(占位示例,真实配比约束见工艺手册)
- kind: forbidden
combination: {catalyst: "C", clf_temp: 910}
reason: C 型催化剂禁止与 910℃ 高温组合(安全告警)
+784
View File
@@ -0,0 +1,784 @@
# -*- coding: utf-8 -*-
"""Ti-2 配方动态优化 · 优化问题建模(约束/目标定义)(Issue #78 / PRD 5.3 ②)。
承接 PRD 5.3「② 配方动态优化」与超参包示例(``objective`` / ``features`` /
``target``):把"给定质量目标 + 工艺约束,求最优配方/参数"这条链路**模板化、
可配置、可测试**,且与 #79 求解器、#80 跨工序寻优、#81 可解释建议解耦。
PRD 设计口径
------------
- 架构表(PRD §5.3):``工艺优化/配方推荐 | 优化/推荐 | 入:质量目标+约束;
出:参数/配方建议 | ② 配方动态优化 | 高(需闭环反馈)``。
- 模板化技术路径:超参包驱动——``objective``、输入特征清单、``target`` 等可变量
外置为 JSON 超参包,切换模板仅改此包;跨行业差异落资产,不落代码。
- 风险表:二期交付(一期数据门槛不足),故本期**先把问题建模沉淀为可校验的声明
式规格**,为 #79 求解器、#80 跨工序寻优、#81 可解释建议提供**统一的问题描述
契约**;先有"能跑通、可测试"的模型,数据就绪后接求解器(#79/#80)。
本模块交付
----------
1. **决策变量 ``DecisionVariable``**:配方/工艺可调参数的声明式规格——变量名、
单位、取值域(``Bounds`` 连续区间 / ``Choices`` 离散枚举)、初值、是否整型、
工艺含义(``meaning``,供 #81 可解释建议引用)。
2. **目标函数规格 ``ObjectiveSpec``**:``Sense``(minimize/maximize)+ 目标项
(``ObjectiveTerm``:系数 × 变量,线性目标)+ 目标 ``target``(PRD 超参包字段)。
3. **约束规格 ``ConstraintSpec``**:``ConstraintKind``(box / linear / ratio /
forbidden)统一描述工艺约束(温度上下限、配方配比、禁止组合等)。
4. **问题模型 ``OptimizationProblem``**:聚合变量 + 目标 + 约束,提供校验
(``validate``,聚合并列出全部错误,便于配置台一次性反馈)、声明式加载
(零第三方依赖 YAML 子集解析,与 data-bus/rag-kb/impurity-forecast 同款)。
设计要点
--------
- **零运行时依赖**(纯标准库):与内核既有模块一致,便于离线/隔离网部署。
- **求解器无关**:本模块只描述"问题",``solve`` 留给 #79 注入;便于换行业复用、
单测无需真实求解器。
- **可解释前置**:变量 ``meaning`` + 约束 ``reason`` 字段,为 #81 优化建议"可溯源"
预留引用依据(对齐 PRD"要求结果可解释、可溯源,要引用依据")。
"""
from __future__ import annotations
import math
import os
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
# 缺失值统一用 float('nan'),与 impurity-forecast 一致,便于上层判空屏蔽。
NAN = float("nan")
class ProblemError(ValueError):
"""配方优化问题建模错误(未知变量 / 越界 / 约束矛盾 / 重复定义等)。"""
# ---------------------------------------------------------------------------
# 决策变量
# ---------------------------------------------------------------------------
class DomainKind(str, Enum):
"""决策变量取值域类型。"""
BOUNDS = "bounds" # 连续区间 [low, high](如温度 800~900℃)
CHOICES = "choices" # 离散枚举(如催化剂型号 A/B/C)
@dataclass
class DecisionVariable:
"""一个可调配方/工艺参数的声明式规格。
``bounds`` 与 ``choices`` 二选一(由 ``kind`` 决定):
- ``bounds``:``[low, high]``,``integer=True`` 时取整;
- ``choices``:离散可选值列表(任意可比较的标量,多为 float/str)。
"""
name: str
kind: DomainKind
meaning: str = "" # 工艺含义,供 #81 可解释建议引用
unit: str = "" # 单位(℃、m³/h、kg、…)
bounds: Optional[Tuple[float, float]] = None
choices: Optional[List[Any]] = None
initial: Optional[float] = None # 当前工况/配方初值
integer: bool = False # 仅 bounds 连续域生效
def __post_init__(self) -> None:
if not self.name or not str(self.name).strip():
raise ProblemError("DecisionVariable.name 不能为空")
if self.kind == DomainKind.BOUNDS:
if self.bounds is None:
raise ProblemError(f"变量 {self.name!r} kind=bounds 但未提供 bounds")
low, high = self.bounds
if _is_num(low) and _is_num(high) and low > high:
raise ProblemError(
f"变量 {self.name!r} bounds 下界 {low} 大于上界 {high}")
if self.integer and self.bounds is not None:
low, high = self.bounds
if _is_num(low) and float(low).is_integer() is False:
raise ProblemError(
f"变量 {self.name!r} integer=True 但下界 {low} 非整")
if _is_num(high) and float(high).is_integer() is False:
raise ProblemError(
f"变量 {self.name!r} integer=True 但上界 {high} 非整")
elif self.kind == DomainKind.CHOICES:
if not self.choices:
raise ProblemError(f"变量 {self.name!r} kind=choices 但 choices 为空")
else: # pragma: no cover - 枚举穷尽
raise ProblemError(f"变量 {self.name!r} 未知 kind={self.kind!r}")
def contains(self, value: Any) -> bool:
"""取值是否落在该变量合法域内。"""
if self.kind == DomainKind.BOUNDS and self.bounds is not None:
if not _is_num(value):
return False
low, high = self.bounds
if self.integer and float(value).is_integer() is False:
return False
return low <= value <= high
# choices
return value in (self.choices or [])
def clamp(self, value: Any) -> Any:
"""把越界的连续域取值夹回合法区间(离散域不夹,原值返回)。"""
if self.kind == DomainKind.BOUNDS and self.bounds is not None and _is_num(value):
low, high = self.bounds
value = max(low, min(high, value))
if self.integer:
value = float(round(value))
return value
return value
def to_dict(self) -> Dict[str, Any]:
d: Dict[str, Any] = {
"name": self.name,
"kind": self.kind.value,
"meaning": self.meaning,
"unit": self.unit,
"integer": self.integer,
}
if self.kind == DomainKind.BOUNDS:
d["bounds"] = list(self.bounds) if self.bounds else None
else:
d["choices"] = list(self.choices) if self.choices else None
if self.initial is not None:
d["initial"] = self.initial
return d
@classmethod
def from_dict(cls, d: Dict[str, Any]) -> "DecisionVariable":
name = d.get("name")
if not isinstance(name, str):
raise ProblemError("DecisionVariable 缺少 name 字段")
kind_raw = d.get("kind", "bounds")
try:
kind = DomainKind(str(kind_raw))
except ValueError as e:
raise ProblemError(f"变量 {name!r} 未知 kind={kind_raw!r}") from e
bounds = d.get("bounds")
choices = d.get("choices")
if kind == DomainKind.BOUNDS and bounds is not None:
if (not isinstance(bounds, (list, tuple))) or len(bounds) != 2:
raise ProblemError(f"变量 {name!r} bounds 必须是 [low, high]")
bounds = (float(bounds[0]), float(bounds[1]))
if kind == DomainKind.CHOICES and choices is not None:
choices = list(choices)
return cls(
name=name,
kind=kind,
meaning=str(d.get("meaning", "")),
unit=str(d.get("unit", "")),
bounds=bounds,
choices=choices,
initial=d.get("initial"),
integer=bool(d.get("integer", False)),
)
# ---------------------------------------------------------------------------
# 目标函数
# ---------------------------------------------------------------------------
class Sense(str, Enum):
"""优化方向。"""
MINIMIZE = "minimize"
MAXIMIZE = "maximize"
@property
def label(self) -> str:
return {Sense.MINIMIZE: "最小化", Sense.MAXIMIZE: "最大化"}[self]
@dataclass
class ObjectiveTerm:
"""线性目标项:``coefficient * variable``(变量名引用 ``DecisionVariable.name``)。"""
variable: str
coefficient: float = 1.0
def to_dict(self) -> Dict[str, Any]:
return {"variable": self.variable, "coefficient": self.coefficient}
@classmethod
def from_dict(cls, d: Dict[str, Any]) -> "ObjectiveTerm":
if "variable" not in d:
raise ProblemError("ObjectiveTerm 缺少 variable 字段")
return cls(variable=str(d["variable"]), coefficient=float(d.get("coefficient", 1.0)))
@dataclass
class ObjectiveSpec:
"""目标函数声明式规格(线性加权,对齐 PRD 超参包 ``objective`` 字段)。
形如 ``sense(coef1*var1 + coef2*var2 + ...)``,目标质量 ``target`` 为达标量
(如 ``Ti_purity ≥ 99.5%`` 中的 99.5),仅记录、不参与求解,供 #81 可解释。
"""
sense: Sense = Sense.MAXIMIZE
terms: List[ObjectiveTerm] = field(default_factory=list)
target: Optional[str] = None # PRD 超参包 ``target``:如 "Ti_purity"
target_value: Optional[float] = None # 达标量(可选)
description: str = ""
def evaluate(self, assignment: Dict[str, float]) -> float:
"""给定一组变量取值,计算目标函数值(未知变量按 0 计)。"""
total = 0.0
for t in self.terms:
v = assignment.get(t.variable)
if _is_num(v):
total += t.coefficient * v
return total
def to_dict(self) -> Dict[str, Any]:
d: Dict[str, Any] = {
"sense": self.sense.value,
"terms": [t.to_dict() for t in self.terms],
}
if self.target is not None:
d["target"] = self.target
if self.target_value is not None:
d["target_value"] = self.target_value
if self.description:
d["description"] = self.description
return d
@classmethod
def from_dict(cls, d: Dict[str, Any]) -> "ObjectiveSpec":
sense_raw = d.get("sense", "maximize")
try:
sense = Sense(str(sense_raw))
except ValueError as e:
raise ProblemError(f"未知 sense={sense_raw!r}") from e
terms = [ObjectiveTerm.from_dict(t) for t in d.get("terms", [])]
tv = d.get("target_value")
return cls(
sense=sense,
terms=terms,
target=d.get("target"),
target_value=float(tv) if _is_num(tv) else None,
description=str(d.get("description", "")),
)
# ---------------------------------------------------------------------------
# 约束
# ---------------------------------------------------------------------------
class ConstraintKind(str, Enum):
"""约束类型(统一描述常见工艺约束)。"""
BOX = "box" # 变量上下界(冗余于 DecisionVariable.bounds,供"运行期收紧")
LINEAR = "linear" # 线性不等式 Σ a_i*x_i (</<=/>/>=) b
RATIO = "ratio" # 配比约束:x_a / x_b (op) value
FORBIDDEN = "forbidden" # 禁止组合:若干变量取值组合不允许
_LINEAR_OPS = {"<", "<=", ">", ">=", "==", "!="}
@dataclass
class ConstraintSpec:
"""约束声明式规格。
每条约束带 ``reason``(工艺依据,供 #81 可解释建议"可溯源、引用依据")。
"""
kind: ConstraintKind
reason: str = ""
# box
variable: Optional[str] = None
bounds: Optional[Tuple[float, float]] = None
# linear
coefficients: Optional[Dict[str, float]] = None
op: str = "<="
rhs: float = 0.0
# ratio
numerator: Optional[str] = None
denominator: Optional[str] = None
value: float = 0.0
# forbidden
combination: Optional[Dict[str, Any]] = None
def __post_init__(self) -> None:
if self.kind == ConstraintKind.LINEAR and self.op not in _LINEAR_OPS:
raise ProblemError(f"线性约束非法 op={self.op!r}")
if self.kind == ConstraintKind.LINEAR and not self.coefficients:
raise ProblemError("线性约束 coefficients 不能为空")
# ---- 校验(返回错误信息列表,不抛异常,便于聚合) ------------------
def errors_against(self, variables: Dict[str, DecisionVariable]) -> List[str]:
"""对该约束引用的变量是否存在做静态校验,返回错误信息列表。"""
errs: List[str] = []
if self.kind == ConstraintKind.BOX:
if not self.variable:
errs.append("box 约束缺少 variable")
elif self.variable not in variables:
errs.append(f"box 约束引用未知变量 {self.variable!r}")
elif self.kind == ConstraintKind.LINEAR:
for vname in (self.coefficients or {}):
if vname not in variables:
errs.append(f"线性约束引用未知变量 {vname!r}")
elif self.kind == ConstraintKind.RATIO:
for fld, vname in (("numerator", self.numerator),
("denominator", self.denominator)):
if not vname:
errs.append(f"ratio 约束缺少 {fld}")
elif vname not in variables:
errs.append(f"ratio 约束引用未知变量 {vname!r}")
elif self.kind == ConstraintKind.FORBIDDEN:
for vname in (self.combination or {}):
if vname not in variables:
errs.append(f"forbidden 约束引用未知变量 {vname!r}")
return errs
# ---- 可行性判定(给定取值,判断该约束是否满足) --------------------
def satisfied_by(self, assignment: Dict[str, Any]) -> bool:
"""给定一组变量取值,判断该约束是否被满足(未知变量视为未约束)。"""
if self.kind == ConstraintKind.BOX and self.bounds is not None and self.variable:
v = assignment.get(self.variable)
if not _is_num(v):
return True # 未知取值不判
low, high = self.bounds
return low <= v <= high
if self.kind == ConstraintKind.LINEAR and self.coefficients:
total = 0.0
unknown = False
for vname, coef in self.coefficients.items():
v = assignment.get(vname)
if not _is_num(v):
unknown = True
break
total += coef * v
if unknown:
return True
return _apply_op(total, self.op, self.rhs)
if self.kind == ConstraintKind.RATIO and self.numerator and self.denominator:
a = assignment.get(self.numerator)
b = assignment.get(self.denominator)
if not _is_num(a) or not _is_num(b) or b == 0:
return True
return _apply_op(a / b, self.op, self.value)
if self.kind == ConstraintKind.FORBIDDEN and self.combination:
# 组合中每个键值都命中才算"禁止组合"被触发
for vname, want in self.combination.items():
if assignment.get(vname) != want:
return True
return False
return True
def to_dict(self) -> Dict[str, Any]:
d: Dict[str, Any] = {"kind": self.kind.value}
if self.reason:
d["reason"] = self.reason
if self.kind == ConstraintKind.BOX:
d["variable"] = self.variable
d["bounds"] = list(self.bounds) if self.bounds else None
elif self.kind == ConstraintKind.LINEAR:
d["coefficients"] = dict(self.coefficients or {})
d["op"] = self.op
d["rhs"] = self.rhs
elif self.kind == ConstraintKind.RATIO:
d["numerator"] = self.numerator
d["denominator"] = self.denominator
d["op"] = self.op
d["value"] = self.value
elif self.kind == ConstraintKind.FORBIDDEN:
d["combination"] = dict(self.combination or {})
return d
@classmethod
def from_dict(cls, d: Dict[str, Any]) -> "ConstraintSpec":
kind_raw = d.get("kind")
try:
kind = ConstraintKind(str(kind_raw))
except ValueError as e:
raise ProblemError(f"未知约束 kind={kind_raw!r}") from e
bounds = d.get("bounds")
if kind == ConstraintKind.BOX and bounds is not None:
bounds = (float(bounds[0]), float(bounds[1]))
coefs = d.get("coefficients")
if coefs is not None:
coefs = {k: float(v) for k, v in coefs.items()}
return cls(
kind=kind,
reason=str(d.get("reason", "")),
variable=d.get("variable"),
bounds=bounds,
coefficients=coefs,
op=str(d.get("op", "<=")),
rhs=float(d.get("rhs", 0.0)),
numerator=d.get("numerator"),
denominator=d.get("denominator"),
value=float(d.get("value", 0.0)),
combination=d.get("combination"),
)
# ---------------------------------------------------------------------------
# 优化问题
# ---------------------------------------------------------------------------
def objective_factory() -> ObjectiveSpec:
"""dataclass 默认值工厂:空目标(最大化、无项)。"""
return ObjectiveSpec(sense=Sense.MAXIMIZE)
@dataclass
class OptimizationProblem:
"""配方优化问题模型(变量 + 目标 + 约束),求解器无关。
设计为「先建模、后求解」:``validate`` 做静态一致性校验(变量引用、域完整性),
``is_feasible`` 做取值可行性判定(运行期收紧约束 / 禁止组合),``solve`` 留给
#79 注入求解器,本模块不绑任何优化库。
"""
variables: List[DecisionVariable] = field(default_factory=list)
objective: ObjectiveSpec = field(default_factory=objective_factory)
constraints: List[ConstraintSpec] = field(default_factory=list)
problem_id: str = ""
template: str = "" # 如 "iAOP-Template-Ti"
description: str = ""
# ---- 变量索引 ----------------------------------------------------
@property
def variable_map(self) -> Dict[str, DecisionVariable]:
return {v.name: v for v in self.variables}
# ---- 校验 --------------------------------------------------------
def validate(self) -> List[str]:
"""聚合所有静态错误,返回错误信息列表(空列表表示通过)。"""
errs: List[str] = []
seen: set = set()
for v in self.variables:
if v.name in seen:
errs.append(f"重复定义变量 {v.name!r}")
seen.add(v.name)
vmap = self.variable_map
for t in self.objective.terms:
if t.variable not in vmap:
errs.append(f"目标项引用未知变量 {t.variable!r}")
for i, c in enumerate(self.constraints):
for e in c.errors_against(vmap):
errs.append(f"约束 #{i} ({c.kind.value}): {e}")
if self.objective.terms and not any(
t.variable in vmap for t in self.objective.terms
):
errs.append("目标函数所有项均引用未知变量")
return errs
# ---- 可行性判定 --------------------------------------------------
def is_feasible(self, assignment: Dict[str, Any]) -> bool:
"""给定一组变量取值,判断是否满足全部约束与变量域。"""
vmap = self.variable_map
for name, val in assignment.items():
v = vmap.get(name)
if v is not None and not v.contains(val):
return False
return all(c.satisfied_by(assignment) for c in self.constraints)
def violated_constraints(self, assignment: Dict[str, Any]) -> List[ConstraintSpec]:
"""返回被该取值违反的约束列表(供 #81 可解释建议引用依据)。"""
return [c for c in self.constraints if not c.satisfied_by(assignment)]
# ---- 序列化 ------------------------------------------------------
def to_dict(self) -> Dict[str, Any]:
d: Dict[str, Any] = {
"variables": [v.to_dict() for v in self.variables],
"objective": self.objective.to_dict(),
"constraints": [c.to_dict() for c in self.constraints],
}
if self.problem_id:
d["problem_id"] = self.problem_id
if self.template:
d["template"] = self.template
if self.description:
d["description"] = self.description
return d
@classmethod
def from_dict(cls, d: Dict[str, Any]) -> "OptimizationProblem":
return cls(
variables=[DecisionVariable.from_dict(v) for v in d.get("variables", [])],
objective=ObjectiveSpec.from_dict(d.get("objective", {})),
constraints=[ConstraintSpec.from_dict(c) for c in d.get("constraints", [])],
problem_id=str(d.get("problem_id", "")),
template=str(d.get("template", "")),
description=str(d.get("description", "")),
)
# ---------------------------------------------------------------------------
# 辅助函数
# ---------------------------------------------------------------------------
def _is_num(x: object) -> bool:
return isinstance(x, (int, float)) and not (isinstance(x, float) and math.isnan(x))
def _apply_op(left: float, op: str, right: float) -> bool:
"""应用比较算子(线性/配比约束共用)。"""
if op == "<":
return left < right
if op == "<=":
return left <= right
if op == ">":
return left > right
if op == ">=":
return left >= right
if op == "==":
return abs(left - right) < 1e-12
if op == "!=":
return abs(left - right) >= 1e-12
return False # pragma: no cover
# ---------------------------------------------------------------------------
# 零依赖 YAML 子集加载(与 data-bus / rag-kb / impurity-forecast 同款)
# ---------------------------------------------------------------------------
def load_problem(path: str) -> OptimizationProblem:
"""从声明式模板资产(YAML 子集)加载优化问题。
解析支持:缩进块、``key: value``、``- item``、行内 ``# 注释``、字符串/数字/
布尔、内联 ``[a, b]`` 列表与 ``{a: 1}`` 映射。足以覆盖本模板资产格式;
不引入第三方依赖,与内核既有模块一致。
"""
with open(path, "r", encoding="utf-8") as fh:
text = fh.read()
data = _parse_yaml_subset(text)
if not isinstance(data, dict):
raise ProblemError(f"模板 {path} 顶层应为映射")
return OptimizationProblem.from_dict(data)
def _parse_yaml_subset(text: str) -> Any:
"""极简 YAML 子集解析器(仅供模板资产,非通用 YAML)。"""
# 去注释 + 去尾部空白,保留缩进
lines: List[str] = []
for raw in text.splitlines():
# 行内注释:仅在 "# " 前不是值的一部分时剥离;这里取保守策略——行首/值后
# 的 " #" 视为注释。冒号/方括号内的 # 不处理。
stripped = raw.rstrip()
if not stripped.strip():
continue
# 简单注释行
if stripped.lstrip().startswith("#"):
continue
# 去行尾注释(" #" 形式)
hash_idx = _find_inline_comment(stripped)
if hash_idx is not None:
stripped = stripped[:hash_idx].rstrip()
if stripped:
lines.append(stripped)
parser = _YamlParser(lines)
return parser.parse_block(0)[0] if lines else {}
def _find_inline_comment(line: str) -> Optional[int]:
"""返回行内注释 ``#`` 的索引(无则 None),跳过 ``[...]``/``{...}`` 内的 #。"""
depth = 0
in_str = False
for i, ch in enumerate(line):
if ch == '"':
in_str = not in_str
elif not in_str:
if ch in "[{":
depth += 1
elif ch in "]}":
depth = max(0, depth - 1)
elif ch == "#" and depth == 0 and i > 0 and line[i - 1] in (" ", "\t"):
return i
elif ch == "#" and depth == 0 and i == 0:
return i
return None
class _YamlParser:
"""递归下降的 YAML 子集解析器(按缩进分层)。"""
def __init__(self, lines: List[str]) -> None:
self.lines = lines
self.i = 0
def _indent(self, line: str) -> int:
return len(line) - len(line.lstrip(" "))
def parse_block(self, indent: int) -> Tuple[Any, bool]:
"""解析当前缩进层级的一个块,返回 (value, is_list_marker)。"""
if self.i >= len(self.lines):
return {}, False
line = self.lines[self.i]
cur_indent = self._indent(line)
if cur_indent < indent:
return {}, False
stripped = line.strip()
if stripped.startswith("- ") or stripped == "-":
return self._parse_list(cur_indent), True
return self._parse_mapping(cur_indent), False
def _parse_mapping(self, indent: int) -> Dict[str, Any]:
result: Dict[str, Any] = {}
# 实际子键缩进可能 > indent(如 "- key: v" 后 4 空格键、项缩进 2)。
# 用首行真实缩进对齐,避免误把合法子键当"孤立缩进"跳过。
effective = indent
if self.i < len(self.lines):
first = self._indent(self.lines[self.i])
if first > indent:
effective = first
while self.i < len(self.lines):
line = self.lines[self.i]
cur = self._indent(line)
if cur < effective:
break
if cur > effective:
# 跳过孤立缩进(不应出现,保守跳过)
self.i += 1
continue
stripped = line.strip()
if stripped.startswith("- "):
break
key, sep, rest = stripped.partition(":")
if not sep:
self.i += 1
continue
key = key.strip()
rest = rest.strip()
self.i += 1
if rest:
result[key] = _parse_scalar(rest)
else:
# 子块:用首行真实缩进解析(列表或映射),兼容 4 空格子键等
if self.i < len(self.lines) and self._indent(self.lines[self.i]) > effective:
child_indent = self._indent(self.lines[self.i])
val, _ = self.parse_block(child_indent)
result[key] = val
else:
result[key] = None
return result
def _parse_list(self, indent: int) -> List[Any]:
result: List[Any] = []
while self.i < len(self.lines):
line = self.lines[self.i]
cur = self._indent(line)
if cur < indent:
break
if cur > indent:
self.i += 1
continue
stripped = line.strip()
if not stripped.startswith("-"):
break
item_text = stripped[1:].strip()
self.i += 1
# 后续更深缩进的行是否归属本项
if self.i < len(self.lines):
child_indent = self._indent(self.lines[self.i])
else:
child_indent = cur
has_deeper = child_indent > cur
if item_text:
# 可能是 "- key: value"(映射项)或 "- 标量"
if ":" in item_text and not item_text.startswith("["):
# 单行映射项的首键
key, sep, rest = item_text.partition(":")
kval = _parse_scalar(rest.strip()) if rest.strip() else None
if has_deeper:
# 把首键与后续子块合并:先解析子块,再把首键塞入
sub, _ = self.parse_block(child_indent)
item: Dict[str, Any] = sub if isinstance(sub, dict) else {}
item[key.strip()] = kval
else:
item = {key.strip(): kval}
result.append(item)
else:
if has_deeper:
# 标量头 + 子块(本模板未使用,保守取子块)
sub, _ = self.parse_block(child_indent)
result.append(sub)
else:
result.append(_parse_scalar(item_text))
else:
# "- " 后跟子块
if has_deeper:
sub, _ = self.parse_block(child_indent)
result.append(sub)
return result
def _parse_scalar(text: str) -> Any:
"""解析标量:数字/布尔/字符串/内联列表/内联映射。"""
text = text.strip()
if not text:
return ""
# 内联列表
if text.startswith("[") and text.endswith("]"):
inner = text[1:-1].strip()
if not inner:
return []
return [_parse_scalar(part.strip()) for part in _split_top(inner, ",")]
# 内联映射
if text.startswith("{") and text.endswith("}"):
inner = text[1:-1].strip()
if not inner:
return {}
out: Dict[str, Any] = {}
for part in _split_top(inner, ","):
k, sep, v = part.partition(":")
if sep:
out[k.strip()] = _parse_scalar(v.strip())
return out
low = text.lower()
if low == "true":
return True
if low == "false":
return False
if low in ("null", "none", "~"):
return None
# 数字
try:
if "." in text or "e" in low:
return float(text)
return int(text)
except ValueError:
# 去引号
if len(text) >= 2 and text[0] in "\"'" and text[-1] == text[0]:
return text[1:-1]
return text
def _split_top(text: str, sep: str) -> List[str]:
"""按分隔符切分顶层(跳过 []/{} 内的)。"""
parts: List[str] = []
depth = 0
cur: List[str] = []
in_str = False
for ch in text:
if ch == '"':
in_str = not in_str
cur.append(ch)
elif not in_str and ch in "[{":
depth += 1
cur.append(ch)
elif not in_str and ch in "]}":
depth = max(0, depth - 1)
cur.append(ch)
elif ch == sep and depth == 0:
parts.append("".join(cur))
cur = []
else:
cur.append(ch)
if cur:
parts.append("".join(cur))
return parts
@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""测试引导:把连字符目录挂载为可导入包(与 core 模块同款模式)。
- ``templates/ti-cl4/recipe-optim`` → 包名 ``recipe_optim``。
本引擎零内核依赖(纯标准库),仅挂载自身包即可。
"""
import importlib.util
import os
import sys
PKG_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
def _load_package(name: str, path: str) -> None:
"""按文件路径完整加载一个包(执行其 __init__.py)。"""
if name in sys.modules:
return
init_py = os.path.join(path, "__init__.py")
spec = importlib.util.spec_from_file_location(
name, init_py, submodule_search_locations=[path])
module = importlib.util.module_from_spec(spec)
sys.modules[name] = module
spec.loader.exec_module(module)
_load_package("recipe_optim", PKG_DIR)
@@ -0,0 +1,291 @@
# -*- coding: utf-8 -*-
"""Ti-2 配方优化问题建模 单元测试(Issue #78)。
覆盖:
- 决策变量域(bounds/choices/integer/越界/夹紧);
- 目标函数(线性求值、min/max、target 记录);
- 约束(box/linear/ratio/forbidden 可行性判定 + 未知变量静态校验);
- OptimizationProblem(聚合校验、可行性、违反约束枚举、序列化往返);
- load_problem YAML 子集加载(模板资产)。
"""
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,
ProblemError,
Sense,
load_problem,
)
CONFIG_PATH = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"config", "recipe_optim.template.yaml",
)
def _ti_problem() -> OptimizationProblem:
"""构造一份与 config 同构的内存问题(用于无需 YAML 的断言)。"""
return OptimizationProblem(
variables=[
DecisionVariable("clf_temp", DomainKind.BOUNDS, "温度", "℃",
bounds=(800, 920), initial=860),
DecisionVariable("cl2_ratio", DomainKind.BOUNDS, "配比", "ratio",
bounds=(0.8, 1.4), initial=1.0),
DecisionVariable("catalyst", DomainKind.CHOICES, "催化剂", "",
choices=["A", "B", "C"]),
],
objective=ObjectiveSpec(
sense=Sense.MAXIMIZE,
target="Ti_purity",
target_value=99.5,
terms=[ObjectiveTerm("clf_temp", 0.01), ObjectiveTerm("cl2_ratio", 2.0)],
),
constraints=[
ConstraintSpec(ConstraintKind.BOX, variable="clf_temp", bounds=(820, 900),
reason="温度安全区间"),
ConstraintSpec(ConstraintKind.LINEAR, coefficients={"clf_temp": 1.0},
op="<=", rhs=900),
ConstraintSpec(ConstraintKind.FORBIDDEN, combination={"catalyst": "C"}),
],
)
class TestDecisionVariable(unittest.TestCase):
def test_bounds_valid_and_contains(self):
v = DecisionVariable("t", DomainKind.BOUNDS, bounds=(0.0, 10.0))
self.assertTrue(v.contains(5))
self.assertTrue(v.contains(0))
self.assertTrue(v.contains(10))
self.assertFalse(v.contains(-0.1))
self.assertFalse(v.contains(10.1))
self.assertFalse(v.contains("x"))
def test_bounds_reversed_rejected(self):
with self.assertRaises(ProblemError):
DecisionVariable("t", DomainKind.BOUNDS, bounds=(10.0, 0.0))
def test_choices_domain(self):
v = DecisionVariable("cat", DomainKind.CHOICES, choices=["A", "B"])
self.assertTrue(v.contains("A"))
self.assertFalse(v.contains("Z"))
def test_integer_bounds_noninteger_rejected(self):
with self.assertRaises(ProblemError):
DecisionVariable("n", DomainKind.BOUNDS, bounds=(1.5, 5.0), integer=True)
def test_integer_contains_and_clamp(self):
v = DecisionVariable("n", DomainKind.BOUNDS, bounds=(0.0, 10.0), integer=True)
self.assertFalse(v.contains(1.5))
self.assertTrue(v.contains(3))
# clamp 把越界夹回 + 取整
self.assertEqual(v.clamp(12.4), 10.0)
self.assertEqual(v.clamp(-3), 0.0)
self.assertEqual(v.clamp(4.7), 5.0)
def test_continuous_clamp(self):
v = DecisionVariable("t", DomainKind.BOUNDS, bounds=(0.0, 10.0))
self.assertEqual(v.clamp(15), 10)
self.assertEqual(v.clamp(-2), 0)
def test_empty_name_rejected(self):
with self.assertRaises(ProblemError):
DecisionVariable(" ", DomainKind.BOUNDS, bounds=(0, 1))
def test_missing_domain_payload_rejected(self):
with self.assertRaises(ProblemError):
DecisionVariable("t", DomainKind.BOUNDS)
with self.assertRaises(ProblemError):
DecisionVariable("t", DomainKind.CHOICES, choices=[])
def test_roundtrip(self):
v = DecisionVariable("t", DomainKind.BOUNDS, "温度", "℃",
bounds=(1.0, 2.0), initial=1.5)
v2 = DecisionVariable.from_dict(v.to_dict())
self.assertEqual(v2.bounds, v.bounds)
self.assertEqual(v2.initial, v.initial)
class TestObjective(unittest.TestCase):
def test_evaluate_maximize(self):
obj = ObjectiveSpec(Sense.MAXIMIZE, terms=[ObjectiveTerm("a", 2.0),
ObjectiveTerm("b", -1.0)])
self.assertEqual(obj.evaluate({"a": 3, "b": 1}), 5.0)
# 未知变量按 0
self.assertEqual(obj.evaluate({"a": 3}), 6.0)
def test_minimize_label(self):
self.assertEqual(Sense.MINIMIZE.label, "最小化")
self.assertEqual(Sense.MAXIMIZE.label, "最大化")
def test_unknown_sense_rejected(self):
with self.assertRaises(ProblemError):
ObjectiveSpec.from_dict({"sense": "extreme"})
def test_roundtrip_with_target(self):
obj = ObjectiveSpec(Sense.MAXIMIZE, target="Ti_purity", target_value=99.5,
terms=[ObjectiveTerm("a", 0.5)])
obj2 = ObjectiveSpec.from_dict(obj.to_dict())
self.assertEqual(obj2.target, "Ti_purity")
self.assertEqual(obj2.target_value, 99.5)
self.assertEqual(obj2.terms[0].coefficient, 0.5)
class TestConstraint(unittest.TestCase):
def test_box_satisfied(self):
c = ConstraintSpec(ConstraintKind.BOX, variable="t", bounds=(0, 10))
self.assertTrue(c.satisfied_by({"t": 5}))
self.assertFalse(c.satisfied_by({"t": 11}))
# 未知取值不判
self.assertTrue(c.satisfied_by({}))
def test_linear_ops(self):
c = ConstraintSpec(ConstraintKind.LINEAR, coefficients={"a": 1, "b": 1},
op="<=", rhs=10)
self.assertTrue(c.satisfied_by({"a": 4, "b": 6}))
self.assertFalse(c.satisfied_by({"a": 6, "b": 6}))
# 部分未知视为未约束
self.assertTrue(c.satisfied_by({"a": 4}))
def test_ratio_constraint(self):
c = ConstraintSpec(ConstraintKind.RATIO, numerator="x", denominator="y",
op=">=", value=0.5)
self.assertTrue(c.satisfied_by({"x": 1, "y": 2}))
self.assertFalse(c.satisfied_by({"x": 1, "y": 4}))
# 分母为 0 视为未约束
self.assertTrue(c.satisfied_by({"x": 1, "y": 0}))
def test_forbidden_combination(self):
c = ConstraintSpec(ConstraintKind.FORBIDDEN, combination={"cat": "C"})
self.assertFalse(c.satisfied_by({"cat": "C"}))
self.assertTrue(c.satisfied_by({"cat": "A"}))
# 多键需全部命中才算触发
c2 = ConstraintSpec(ConstraintKind.FORBIDDEN,
combination={"cat": "C", "t": 900})
self.assertTrue(c2.satisfied_by({"cat": "C", "t": 100}))
self.assertFalse(c2.satisfied_by({"cat": "C", "t": 900}))
def test_linear_bad_op_rejected(self):
with self.assertRaises(ProblemError):
ConstraintSpec(ConstraintKind.LINEAR, coefficients={"a": 1}, op="~=")
with self.assertRaises(ProblemError):
ConstraintSpec(ConstraintKind.LINEAR, op="<=")
def test_errors_against_unknown_var(self):
c = ConstraintSpec(ConstraintKind.BOX, variable="missing", bounds=(0, 1))
self.assertTrue(c.errors_against({"t": object()}))
def test_roundtrip(self):
c = ConstraintSpec(ConstraintKind.LINEAR, reason="x",
coefficients={"a": 1.0}, op=">=", rhs=5)
c2 = ConstraintSpec.from_dict(c.to_dict())
self.assertEqual(c2.op, ">=")
self.assertEqual(c2.rhs, 5)
class TestOptimizationProblem(unittest.TestCase):
def test_validate_ok(self):
self.assertEqual(_ti_problem().validate(), [])
def test_validate_duplicate_variable(self):
p = _ti_problem()
p.variables.append(DecisionVariable("clf_temp", DomainKind.BOUNDS,
bounds=(0, 1)))
errs = p.validate()
self.assertTrue(any("重复定义" in e for e in errs))
def test_validate_unknown_var_in_objective(self):
p = _ti_problem()
p.objective.terms.append(ObjectiveTerm("nope"))
errs = p.validate()
self.assertTrue(any("nope" in e for e in errs))
def test_validate_unknown_var_in_constraint(self):
p = _ti_problem()
p.constraints.append(ConstraintSpec(ConstraintKind.BOX, variable="ghost",
bounds=(0, 1)))
errs = p.validate()
self.assertTrue(any("ghost" in e for e in errs))
def test_is_feasible_and_violated(self):
p = _ti_problem()
# 合法取值
self.assertTrue(p.is_feasible({"clf_temp": 850, "cl2_ratio": 1.0,
"catalyst": "A"}))
# 越出 box 收紧(820~900)
self.assertFalse(p.is_feasible({"clf_temp": 910, "cl2_ratio": 1.0,
"catalyst": "A"}))
# forbidden 组合
self.assertFalse(p.is_feasible({"clf_temp": 850, "cl2_ratio": 1.0,
"catalyst": "C"}))
viol = p.violated_constraints({"clf_temp": 950, "cl2_ratio": 1.0,
"catalyst": "A"})
self.assertGreater(len(viol), 0)
def test_roundtrip(self):
p = _ti_problem()
p2 = OptimizationProblem.from_dict(p.to_dict())
self.assertEqual([v.name for v in p2.variables],
[v.name for v in p.variables])
self.assertEqual(p2.objective.sense, p.objective.sense)
self.assertEqual(len(p2.constraints), len(p.constraints))
class TestLoadProblemYaml(unittest.TestCase):
def test_load_template(self):
self.assertTrue(os.path.exists(CONFIG_PATH), f"缺少模板 {CONFIG_PATH}")
p = load_problem(CONFIG_PATH)
names = [v.name for v in p.variables]
self.assertEqual(names, ["clf_temp", "cl2_ratio", "feed_rate", "catalyst"])
# 校验通过
self.assertEqual(p.validate(), [])
# 目标与约束就位
self.assertEqual(p.objective.target, "Ti_purity")
self.assertEqual(p.objective.target_value, 99.5)
kinds = {c.kind for c in p.constraints}
self.assertEqual(kinds, {ConstraintKind.BOX, ConstraintKind.LINEAR,
ConstraintKind.RATIO, ConstraintKind.FORBIDDEN})
# 合法初值可行
self.assertTrue(p.is_feasible({"clf_temp": 850, "cl2_ratio": 1.0,
"feed_rate": 450, "catalyst": "A"}))
def test_load_choices_parsed(self):
p = load_problem(CONFIG_PATH)
cat = p.variable_map["catalyst"]
self.assertEqual(cat.kind, DomainKind.CHOICES)
self.assertEqual(cat.choices, ["A", "B", "C"])
def test_integer_flag_parsed(self):
# feed_rate integer: false
p = load_problem(CONFIG_PATH)
self.assertFalse(p.variable_map["feed_rate"].integer)
class TestScalarParser(unittest.TestCase):
"""直接覆盖 _parse_scalar 的边界(内联列表/映射/数字/字符串/布尔)。"""
def test_scalars(self):
from recipe_optim.problem import _parse_scalar
self.assertEqual(_parse_scalar("1"), 1)
self.assertEqual(_parse_scalar("1.5"), 1.5)
self.assertIs(_parse_scalar("true"), True)
self.assertIs(_parse_scalar("False"), False)
self.assertEqual(_parse_scalar("99.5"), 99.5)
self.assertEqual(_parse_scalar('"A"'), "A")
self.assertEqual(_parse_scalar("[1, 2, 3]"), [1, 2, 3])
self.assertEqual(_parse_scalar("[A, B]"), ["A", "B"])
self.assertEqual(_parse_scalar("{a: 1, b: 2}"), {"a": 1, "b": 2})
if __name__ == "__main__":
unittest.main()