feat(#36): 质量预测模型模板化(固定主干+配方加载,PRD 5.3 ①质量预测)

对应 issue #36(父 EPIC #5「③ AI 模型框架 配置化重构」、PRD 5.3
「网络结构策略 / 模板化技术路径」)。

落地 PRD 5.3「固定主干网络 + 可配置超参」默认模式:
- core/model-framework/quality_forecast.py:QualityForecastModel 固定主干
  (默认 gbdt 梯度提升回归,PRD 5.3 监督回归默认结构)+ Recipe 配方加载器
  (load_recipe/build_from_recipe 声明式 JSON 超参包)。切换行业/工况只改
  配方,模型代码零改动——对齐 PRD 验收口径「切换模板仅改超参包」。
- 主干注册表 BACKBONES + register_backbone:gbdt/dnn/stub 三类内置主干,
  有 sklearn 升级真实 GBDT/MLP,无依赖退化确定性 stub(零外部强依赖,
  CI 可加载校验);新增结构走插件注册而非改内核(PRD 5.3 理念,风格对齐 #34)。
- Accuracy 验收口径:PRD 5.3/第6章里程碑「质量预测准确率≥90%」,
  evaluate 直接给出 accuracy/MAE/RMSE 与是否达标。
- 样例协议 samples/quality-forecast/:Ti(海绵钛氯化车间)+ 树脂 两套超参包,
  验证「同框架加载两套配方均跑通」。
- 接口风格对齐 #34 ModelHandle/ModelRecipe,自包含不依赖未合并的 model_recipe;
  待 PR #102(#34) 合入后主干可平滑注册为具名 backbone、配方映射为 ModelRecipe。
- 24 个单元测试全通过 + sanity check 通过。
This commit is contained in:
2026-08-04 23:33:58 +08:00
parent 8dc379c479
commit e41e2a5a3f
8 changed files with 980 additions and 0 deletions
+48
View File
@@ -0,0 +1,48 @@
# -*- coding: utf-8 -*-
"""iAOP-Core · 模型框架层(AI Model Framework)。
对应 PRD 5.3「③ AI 模型框架」与 EPIC #5(内核平台化改造)。
当前已交付(自包含,不依赖未合并分支):
- ``quality_forecast``:质量预测模型模板化(固定主干 + 配方加载),
issue #36。同一主干代码不变,切换行业/工况只改配方(声明式 JSON
超参包)——对齐 PRD 5.3「固定主干 + 可配置超参」默认模式。
规划(待相关 PR 合入后无缝对接,业务侧零改动):
- ``model_recipe``:Model Recipe 插件接口(issue #34,PR #102 待审核)。
``quality_forecast`` 的主干届时可注册为 ``register_backbone`` 的一个
具名主干,配方可映射为一条 ``ModelRecipe``。
"""
from model_framework.quality_forecast import (
Accuracy,
BACKBONES,
ModelHandle,
QualityForecastError,
QualityForecastModel,
Recipe,
build_from_recipe,
dnn_backbone,
gbdt_backbone,
list_sample_recipes,
load_recipe,
register_backbone,
sample_recipe_path,
stub_backbone,
)
__all__ = [
"Recipe",
"Accuracy",
"QualityForecastError",
"QualityForecastModel",
"ModelHandle",
"BACKBONES",
"register_backbone",
"gbdt_backbone",
"dnn_backbone",
"stub_backbone",
"load_recipe",
"build_from_recipe",
"list_sample_recipes",
"sample_recipe_path",
]
+71
View File
@@ -0,0 +1,71 @@
# -*- coding: utf-8 -*-
"""质量预测模型模板化 sanity 检查(无构建环境下的离线基本验证)。
验证 PRD 5.3 验收口径「同框架加载 Ti / 树脂两套配方均跑通」:
1. 两套样例配方均可被 ``build_from_recipe`` 加载;
2. 加载后模型可 fit / predict / evaluate 走通完整链路;
3. 切换模板仅改配方,模型代码(``type(m1) == type(m2)``)零改动;
4. 两套配方的 backbone / 特征列确实不同(确属两套模板,非同一份复制)。
用法:python _sanity_check.py
"""
import os
import sys
HERE = os.path.dirname(os.path.abspath(__file__))
if HERE not in sys.path:
sys.path.insert(0, HERE)
from quality_forecast import ( # noqa: E402
build_from_recipe,
list_sample_recipes,
sample_recipe_path,
)
def main() -> int:
failures = []
names = list_sample_recipes()
required = ("recipe.ti.json", "recipe.resin.json")
for r in required:
if r not in names:
failures.append(f"缺少样例配方:{r}")
models = {}
for r in required:
try:
m = build_from_recipe(sample_recipe_path(r))
# 用配方声明的特征数构造演示数据
n_feat = len(m.recipe_meta["feature_columns"]) or 2
X = [[float(i + j) for j in range(n_feat)] for i in range(10)]
y = [float(i % 4) + 1.0 for i in range(10)]
m.fit(X, y)
preds = m.predict(X)
assert len(preds) == len(y), "预测长度异常"
m.evaluate(X, y)
models[r] = m
except Exception as exc: # pragma: no cover - 诊断输出
failures.append(f"{r} 加载/训练/评估失败:{exc!r}")
# 切换模板仅改配方,模型代码零改动
if len(models) == 2:
ms = list(models.values())
if type(ms[0]) is not type(ms[1]):
failures.append("两套配方使用了不同的模型类,违反「模型代码零改动」")
if (models["recipe.ti.json"].recipe_meta["feature_columns"]
== models["recipe.resin.json"].recipe_meta["feature_columns"]):
failures.append("Ti/树脂配方特征列完全相同,疑似复制")
if failures:
print("FAIL")
for f in failures:
print(" -", f)
return 1
print(f"OK: {len(required)} 套配方均加载/训练/评估通过,"
f"模型代码零改动(type 一致),PRD 5.3 验收口径达成")
return 0
if __name__ == "__main__":
sys.exit(main())
+534
View File
@@ -0,0 +1,534 @@
# -*- coding: utf-8 -*-
"""质量预测模型模板化(固定主干 + 配方加载)。
对应 issue #36(父 EPIC #5「③ AI 模型框架 配置化重构」、PRD 5.3
「网络结构策略 / 模板化技术路径」)。
PRD 5.3 的核心诉求
------------------
质量预测属于 PRD 5.3「四类模型模板」之一(① 质量预测),采用
「**固定主干网络 + 可配置超参**」为默认模式:同一主干代码不变,
切换行业/工况只改 *配方(recipe)* —— 一个声明式 JSON 超参包。
本模块交付什么
--------------
1. **``QualityForecastModel``**:固定主干的质量预测模型。默认主干是
``gbdt``(梯度提升回归,PRD 5.3 推荐的监督回归默认结构);当运行
环境存在 ``sklearn`` 时自动升级为真实实现,否则退化为确定性 stub,
保证边缘 / 离线 / CI 环境可加载与校验——与 issue #34 / #35 的
「numpy/sklearn 可选」策略一致。
2. **``Recipe`` 配方加载器**:声明式 JSON 超参包(``load_recipe`` /
``build_from_recipe``)。配方描述「主干类型 + 超参 + 特征列 + 目标列
+ 验收口径」,业务侧只 ``build_from_recipe(path)`` 一行即可拿到一个
可训练/可推理的模型——切换模板仅改配方,模型代码零改动。
3. **``Accuracy`` 验收口径**:PRD 5.3 / 第 6 章里程碑要求「关键质量指标
预测准确率 ≥ 90%」。``evaluate`` 直接给出准确率 / MAE / RMSE,便于
配置台与 UAT 直接读取。
4. **样例配方(``samples/`` JSON)**:Ti(海绵钛氯化车间)+ 树脂两套
质量预测超参包样例,验证「同框架加载两套配方均跑通」的验收口径。
与 issue #34 ``model_recipe`` 的关系
------------------------------------
接口风格对齐 #34 的 ``ModelHandle`` / ``ModelRecipe``(``fit`` / ``predict``
/ ``to_dict``、不可变声明式数据对象)。本模块**自包含、不依赖 #34 未合并
的 ``model_recipe``**,待 #34(PR #102)合入后,质量预测主干可平滑注册为
``register_backbone("gbdt", ...)`` 的一个具名主干,配方可映射为一条
``ModelRecipe``——届时本模块零业务侧改动。
零外部强依赖
------------
* 主干默认走纯 Python stub(``StubBackbone``):无 sklearn 时也能加载、
构造、(伪)拟合与预测,保证 CI 可加载与校验;
* 存在 ``sklearn`` 时,``gbdt`` 主干自动升级为真实
``GradientBoostingRegressor`` 实现,其余情况退化为 stub,不影响接口
契约与测试。
"""
from __future__ import annotations
import json
import math
import os
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Sequence, Tuple
__all__ = [
# 数据对象
"Recipe",
"Accuracy",
"QualityForecastError",
# 模型
"QualityForecastModel",
"ModelHandle",
# 主干工厂
"BACKBONES",
"register_backbone",
"gbdt_backbone",
"dnn_backbone",
"stub_backbone",
# 配方 API
"load_recipe",
"build_from_recipe",
"list_sample_recipes",
"sample_recipe_path",
]
class QualityForecastError(Exception):
"""质量预测模板化层的统一异常(配方非法 / 主干未注册 / 校验失败)。"""
# ---------------------------------------------------------------------------
# 配方(Recipe):声明式超参包,不可变数据对象
# ---------------------------------------------------------------------------
#: PRD 5.3 允许的固定主干类型(默认 gbdt,PRD 5.3 推荐监督回归默认结构)
ALLOWED_BACKBONES = ("gbdt", "dnn", "stub")
#: PRD 5.3 / 第 6 章里程碑:质量预测准确率验收线 ≥ 90%
DEFAULT_ACCURACY_FLOOR = 0.90
@dataclass(frozen=True)
class Recipe:
"""质量预测配方(声明式超参包)。
一个 Recipe 描述「用什么固定主干 + 如何从超参构造一个可训练/可推理的
质量预测模型 + 用哪些特征/目标列 + 验收口径」。它是不可变数据对象,
``to_dict`` / ``from_dict`` 可序列化往返,便于配置台展示与审计。
切换行业/工况只改 Recipe,模型代码(``QualityForecastModel``)零改动
——对齐 PRD 5.3「固定主干 + 可配置超参」默认模式。
"""
name: str
backbone: str = "gbdt"
hyperparams: Dict[str, Any] = field(default_factory=dict)
feature_columns: Tuple[str, ...] = field(default_factory=tuple)
target_column: str = "quality_index"
accuracy_floor: float = DEFAULT_ACCURACY_FLOOR
industry: str = ""
notes: str = ""
def __post_init__(self) -> None:
if not self.name:
raise QualityForecastError("Recipe 缺少 name")
if self.backbone not in ALLOWED_BACKBONES:
raise QualityForecastError(
f"非法主干类型 {self.backbone!r},允许:{ALLOWED_BACKBONES}")
if self.accuracy_floor < 0 or self.accuracy_floor > 1:
raise QualityForecastError(
f"accuracy_floor 越界:{self.accuracy_floor}(应在 [0,1])")
def to_dict(self) -> Dict[str, Any]:
return {
"name": self.name,
"backbone": self.backbone,
"hyperparams": dict(self.hyperparams),
"feature_columns": list(self.feature_columns),
"target_column": self.target_column,
"accuracy_floor": self.accuracy_floor,
"industry": self.industry,
"notes": self.notes,
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "Recipe":
try:
return cls(
name=data["name"],
backbone=data.get("backbone", "gbdt"),
hyperparams=dict(data.get("hyperparams", {})),
feature_columns=tuple(data.get("feature_columns", [])),
target_column=data.get("target_column", "quality_index"),
accuracy_floor=float(data.get(
"accuracy_floor", DEFAULT_ACCURACY_FLOOR)),
industry=data.get("industry", ""),
notes=data.get("notes", ""),
)
except KeyError as exc: # pragma: no cover - 防御性
raise QualityForecastError(f"配方缺少必填字段:{exc}") from exc
def load_recipe(path: str) -> Recipe:
"""从 JSON 文件加载一个质量预测配方。
配方 JSON 结构见 ``Recipe.to_dict``;样例见 ``samples/``。
"""
with open(path, "r", encoding="utf-8") as fh:
data = json.load(fh)
if not isinstance(data, dict):
raise QualityForecastError(f"配方根必须是对象:{path}")
return Recipe.from_dict(data)
# ---------------------------------------------------------------------------
# 主干工厂:固定主干网络(gbdt / dnn / stub)
# ---------------------------------------------------------------------------
class ModelHandle:
"""统一模型句柄:fit / predict / to_dict,与硬件和具体库无关。
业务代码只持有 ``ModelHandle``,不感知底层是 sklearn 还是 stub。
"""
def __init__(self, backbone: str, params: Dict[str, Any],
fitted: bool = False, meta: Optional[Dict[str, Any]] = None):
self.backbone = backbone
self.params = dict(params)
self._fitted = fitted
self.meta: Dict[str, Any] = dict(meta or {})
@property
def fitted(self) -> bool:
return self._fitted
def fit(self, X: Sequence[Sequence[float]], y: Sequence[float]) -> "ModelHandle":
"""拟合主干。stub 主干记录均值/极差用于确定性预测。"""
X = list(X)
y = list(y)
if not X or not y:
raise QualityForecastError("训练数据为空")
if len(X) != len(y):
raise QualityForecastError(
f"X/y 样本数不一致:{len(X)} != {len(y)}")
self._fit_impl(X, y)
self._fitted = True
return self
# 子类/工厂填充
def _fit_impl(self, X: Sequence[Sequence[float]], y: Sequence[float]) -> None:
raise NotImplementedError
def predict(self, X: Sequence[Sequence[float]]) -> List[float]:
if not self._fitted:
raise QualityForecastError("模型未拟合,无法预测")
return [self._predict_one(list(row)) for row in X]
def _predict_one(self, row: Sequence[float]) -> float:
raise NotImplementedError
def to_dict(self) -> Dict[str, Any]:
return {
"backbone": self.backbone,
"params": dict(self.params),
"fitted": self._fitted,
"meta": dict(self.meta),
}
class _StubBackbone(ModelHandle):
"""确定性 stub 主干:无 sklearn 时的保底实现。
拟合阶段记录训练目标的均值与极差;预测返回一个由输入求和驱动的
确定性值(落在训练目标范围内),保证可复现、可校验、可对比,便于
CI 与配置台预览。
"""
def __init__(self, params: Dict[str, Any]):
super().__init__(backbone="stub", params=params)
self._y_mean: float = 0.0
self._y_amp: float = 1.0
def _fit_impl(self, X, y) -> None:
self._y_mean = sum(y) / len(y)
self._y_amp = (max(y) - min(y)) or 1.0
self.meta.update({"y_mean": self._y_mean, "y_amp": self._y_amp})
def _predict_one(self, row) -> float:
# 确定性:输入和的 tanh 压缩到 [y_mean-amp/2, y_mean+amp/2]
s = sum(float(v) for v in row) if row else 0.0
# 归一化到 [-1,1] 附近,再映射回目标域
norm = math.tanh(s / (self._y_amp or 1.0))
return self._y_mean + 0.5 * self._y_amp * norm
class _SklearnGbdtBackbone(ModelHandle):
"""真实 GBDT 主干(sklearn GradientBoostingRegressor)。
仅当运行环境存在 sklearn 时启用;与 stub 接口完全一致。
"""
def __init__(self, params: Dict[str, Any]):
super().__init__(backbone="gbdt", params=params)
# 延迟 import,避免无 sklearn 环境加载失败
from sklearn.ensemble import GradientBoostingRegressor # type: ignore
self._Clz = GradientBoostingRegressor
self._model: Any = None
def _fit_impl(self, X, y) -> None:
kw = {
"n_estimators": int(self.params.get("n_estimators", 100)),
"max_depth": int(self.params.get("max_depth", 3)),
"learning_rate": float(self.params.get("learning_rate", 0.1)),
"random_state": int(self.params.get("random_state", 42)),
}
self._model = self._Clz(**kw)
self._model.fit(list(X), list(y))
self.meta.update(kw)
def _predict_one(self, row) -> float:
return float(self._model.predict([list(row)])[0])
class _SklearnDnnBackbone(ModelHandle):
"""真实轻量 DNN 主干(sklearn MLPRegressor)。
PRD 5.3 备选结构;仅当运行环境存在 sklearn 时启用。
"""
def __init__(self, params: Dict[str, Any]):
super().__init__(backbone="dnn", params=params)
from sklearn.neural_network import MLPRegressor # type: ignore
self._Clz = MLPRegressor
self._model: Any = None
def _fit_impl(self, X, y) -> None:
kw = {
"hidden_layer_sizes": tuple(
self.params.get("hidden_layer_sizes", (32, 16))),
"max_iter": int(self.params.get("max_iter", 500)),
"random_state": int(self.params.get("random_state", 42)),
}
self._model = self._Clz(**kw)
self._model.fit(list(X), list(y))
self.meta.update({"hidden_layer_sizes": list(kw["hidden_layer_sizes"]),
"max_iter": kw["max_iter"]})
def _predict_one(self, row) -> float:
return float(self._model.predict([list(row)])[0])
def _has_sklearn() -> bool:
try:
import sklearn # noqa: F401
return True
except Exception:
return False
def stub_backbone(hyperparams: Dict[str, Any]) -> ModelHandle:
"""stub 主干工厂(恒可用)。"""
return _StubBackbone(hyperparams)
def gbdt_backbone(hyperparams: Dict[str, Any]) -> ModelHandle:
"""gbdt 主干工厂:有 sklearn 用真实 GBDT,否则退化为 stub。
PRD 5.3 推荐的监督回归默认结构(梯度提升回归)。
"""
if _has_sklearn():
return _SklearnGbdtBackbone(hyperparams)
# 无 sklearn:退化 stub 但保留声明主干名,便于审计
h = _StubBackbone(hyperparams)
h.meta["degraded_from"] = "gbdt"
return h
def dnn_backbone(hyperparams: Dict[str, Any]) -> ModelHandle:
"""dnn 主干工厂:有 sklearn 用真实 MLP,否则退化为 stub。"""
if _has_sklearn():
return _SklearnDnnBackbone(hyperparams)
h = _StubBackbone(hyperparams)
h.meta["degraded_from"] = "dnn"
return h
#: 主干注册表:新增结构走 ``register_backbone`` 注册,不动内核
#: (对齐 PRD 5.3「新增结构走插件注册」理念,风格对齐 #34)。
BACKBONES: Dict[str, Any] = {
"gbdt": gbdt_backbone,
"dnn": dnn_backbone,
"stub": stub_backbone,
}
def register_backbone(name: str, factory: Any) -> None:
"""注册一个新主干工厂 ``factory(hyperparams) -> ModelHandle``。
允许高级行业模板声明非默认主干(如自研网络),不动内核——对齐 PRD
「新增结构走插件注册而非改内核」。
"""
if not callable(factory):
raise QualityForecastError("主干工厂必须是可调用对象")
BACKBONES[name] = factory
def _build_backbone(backbone: str, hyperparams: Dict[str, Any]) -> ModelHandle:
factory = BACKBONES.get(backbone)
if factory is None:
raise QualityForecastError(
f"未注册的主干类型:{backbone!r},已注册:{list(BACKBONES)}")
return factory(hyperparams)
# ---------------------------------------------------------------------------
# 质量预测模型:固定主干 + 配方加载
# ---------------------------------------------------------------------------
class QualityForecastModel:
"""质量预测模型(固定主干 + 配方加载)。
业务侧两种等价入口:
1. 直接构造(显式主干)::
m = QualityForecastModel(backbone="gbdt", hyperparams={...})
2. 配方加载(推荐,切换模板仅改配方)::
m = build_from_recipe("templates/.../quality-forecast/recipe.ti.json")
"""
def __init__(self, backbone: str = "gbdt",
hyperparams: Optional[Dict[str, Any]] = None,
feature_columns: Optional[Sequence[str]] = None,
target_column: str = "quality_index",
accuracy_floor: float = DEFAULT_ACCURACY_FLOOR):
self.recipe_meta: Dict[str, Any] = {
"backbone": backbone,
"hyperparams": dict(hyperparams or {}),
"feature_columns": list(feature_columns or []),
"target_column": target_column,
"accuracy_floor": accuracy_floor,
}
self._handle: ModelHandle = _build_backbone(backbone, hyperparams or {})
@classmethod
def from_recipe(cls, recipe: Recipe) -> "QualityForecastModel":
"""从一个 ``Recipe`` 构造模型(推荐入口)。"""
m = cls(
backbone=recipe.backbone,
hyperparams=recipe.hyperparams,
feature_columns=recipe.feature_columns,
target_column=recipe.target_column,
accuracy_floor=recipe.accuracy_floor,
)
m.recipe_meta["recipe_name"] = recipe.name
m.recipe_meta["industry"] = recipe.industry
return m
# ---- 训练 / 推理 ----
def fit(self, X: Sequence[Sequence[float]], y: Sequence[float]) -> "QualityForecastModel":
self._handle.fit(X, y)
return self
def predict(self, X: Sequence[Sequence[float]]) -> List[float]:
return self._handle.predict(X)
@property
def fitted(self) -> bool:
return self._handle.fitted
# ---- 验收口径 ----
def evaluate(self, X: Sequence[Sequence[float]],
y: Sequence[float]) -> "Accuracy":
"""评估并返回准确率/MAE/RMSE 与是否达标。
准确率口径(PRD 5.3 / 里程碑):相对误差在容忍带
``tolerance``(默认 10%)内计为命中。``accuracy >= accuracy_floor``
即视为达标(默认 90%)。
"""
preds = self.predict(X)
return Accuracy.compute(
y_true=list(y), y_pred=preds,
accuracy_floor=self.recipe_meta["accuracy_floor"])
def to_dict(self) -> Dict[str, Any]:
return {
"recipe_meta": dict(self.recipe_meta),
"handle": self._handle.to_dict(),
}
# ---------------------------------------------------------------------------
# 验收:Accuracy
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class Accuracy:
"""质量预测验收结果(PRD 5.3 准确率口径)。"""
accuracy: float
mae: float
rmse: float
tolerance: float
accuracy_floor: float
passed: bool
def to_dict(self) -> Dict[str, Any]:
return {
"accuracy": self.accuracy,
"mae": self.mae,
"rmse": self.rmse,
"tolerance": self.tolerance,
"accuracy_floor": self.accuracy_floor,
"passed": self.passed,
}
@classmethod
def compute(cls, y_true: Sequence[float], y_pred: Sequence[float],
tolerance: float = 0.10,
accuracy_floor: float = DEFAULT_ACCURACY_FLOOR) -> "Accuracy":
if len(y_true) != len(y_pred):
raise QualityForecastError(
f"y_true/y_pred 长度不一致:{len(y_true)} != {len(y_pred)}")
if not y_true:
raise QualityForecastError("评估数据为空")
n = len(y_true)
hits = 0
abs_err_sum = 0.0
sq_err_sum = 0.0
for yt, yp in zip(y_true, y_pred):
denom = abs(yt) if abs(yt) > 1e-9 else 1.0
rel = abs(yp - yt) / denom
if rel <= tolerance:
hits += 1
abs_err_sum += abs(yp - yt)
sq_err_sum += (yp - yt) ** 2
accuracy = hits / n
mae = abs_err_sum / n
rmse = math.sqrt(sq_err_sum / n)
return cls(
accuracy=accuracy, mae=mae, rmse=rmse,
tolerance=tolerance, accuracy_floor=accuracy_floor,
passed=accuracy >= accuracy_floor,
)
# ---------------------------------------------------------------------------
# 配方构建入口 + 样例协议
# ---------------------------------------------------------------------------
def build_from_recipe(path: str) -> QualityForecastModel:
"""从 JSON 配方文件加载并构造一个质量预测模型(推荐入口)。
切换模板仅改配方文件,业务代码零改动——对齐 PRD 5.3 验收口径。
"""
return QualityForecastModel.from_recipe(load_recipe(path))
def _samples_dir() -> str:
return os.path.join(os.path.dirname(os.path.abspath(__file__)),
"samples", "quality-forecast")
def list_sample_recipes() -> List[str]:
"""列出内置样例配方(树脂 + Ti 两套,验证同框架加载多套配方)。"""
d = _samples_dir()
if not os.path.isdir(d):
return []
return sorted(f for f in os.listdir(d) if f.endswith(".json"))
def sample_recipe_path(name: str) -> str:
"""返回样例配方的完整路径。"""
if not name.endswith(".json"):
name = name + ".json"
return os.path.join(_samples_dir(), name)
@@ -0,0 +1,21 @@
{
"name": "resin-quality",
"backbone": "gbdt",
"industry": "吸附树脂(已终验化工新材料AI平台 baseline)",
"hyperparams": {
"n_estimators": 100,
"max_depth": 3,
"learning_rate": 0.1,
"random_state": 7
},
"feature_columns": [
"reactor_temp",
"reactor_pressure",
"flow_rate",
"ph_value",
"conversion_rate"
],
"target_column": "resin_purity_index",
"accuracy_floor": 0.90,
"notes": "PRD 5.3 ① 质量预测:树脂纯度/合格率预测,复用已交付化工AI平台 baseline 超参。"
}
@@ -0,0 +1,22 @@
{
"name": "ti-cl4-quality",
"backbone": "gbdt",
"industry": "海绵钛氯化车间(Template-Ti 一期)",
"hyperparams": {
"n_estimators": 120,
"max_depth": 4,
"learning_rate": 0.08,
"random_state": 42
},
"feature_columns": [
"furnace_temp",
"furnace_pressure",
"cl2_flow",
"ti_feed_rate",
"impurity_fe",
"impurity_v"
],
"target_column": "ti_product_grade_index",
"accuracy_floor": 0.90,
"notes": "PRD 5.3 ① 质量预测:氯化车间一次合格率预测,验收准确率≥90%(PRD 第6章里程碑)。一期数据门槛:≥6个月标注(LIMS对接后补标)。"
}
+1
View File
@@ -0,0 +1 @@
# -*- coding: utf-8 -*-
+28
View File
@@ -0,0 +1,28 @@
# -*- coding: utf-8 -*-
"""测试引导:把连字符目录 ``core/model-framework`` 加载为可导入包
``model_framework``,使测试可 ``from model_framework import ...``。
与仓库内各 core 模块的测试引导同款模式(importlib 完整加载包,执行
``__init__.py``,保持顶层导出可用)。
"""
import importlib.util
import os
import sys
PKG_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
REPO_ROOT = os.path.dirname(os.path.dirname(
os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
def _load_package(name: str, path: str) -> None:
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("model_framework", PKG_DIR)
@@ -0,0 +1,255 @@
# -*- coding: utf-8 -*-
"""``quality_forecast`` 单元测试(issue #36)。
覆盖:
- 配方(Recipe)不可变性 / 序列化往返 / 非法主干与越界校验;
- 主干工厂注册表 + 自定义主干注册(PRD 5.3「新增结构走插件注册」);
- stub / gbdt / dnn 三类主干的 fit/predict/evaluate 契约;
- 固定主干 + 配方加载:同框架加载 Ti / 树脂两套配方均跑通(PRD 5.3
验收口径);
- Accuracy 验收口径(PRD 5.3 / 里程碑:准确率 ≥ 90%);
- 零外部强依赖:无 sklearn 时 stub 退化仍可加载与校验。
"""
import json
import os
import sys
import unittest
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import _bootstrap # noqa: E402 注册 model_framework 包
from model_framework import ( # noqa: E402
Accuracy,
BACKBONES,
ModelHandle,
QualityForecastError,
QualityForecastModel,
Recipe,
build_from_recipe,
dnn_backbone,
gbdt_backbone,
list_sample_recipes,
load_recipe,
register_backbone,
sample_recipe_path,
stub_backbone,
)
def _linear_dataset(n=40, noise=0.0):
"""构造一个 y ≈ 2*x0 + x1 的可学习数据集(带可选噪声)。"""
X, y = [], []
for i in range(n):
x0 = float(i % 7) + 1.0
x1 = float(i % 5) * 0.5 + 0.5
yv = 2.0 * x0 + x1 + noise * (i % 3 - 1)
X.append([x0, x1])
y.append(yv)
return X, y
class TestRecipe(unittest.TestCase):
"""配方数据对象与校验。"""
def test_defaults_and_immutability(self):
r = Recipe(name="t")
self.assertEqual(r.backbone, "gbdt")
self.assertEqual(r.target_column, "quality_index")
self.assertAlmostEqual(r.accuracy_floor, 0.90)
with self.assertRaises(Exception):
r.name = "other" # frozen
def test_roundtrip(self):
r = Recipe(name="t", backbone="dnn",
hyperparams={"max_iter": 50},
feature_columns=("a", "b"),
target_column="y",
accuracy_floor=0.8, industry="树脂", notes="n")
d = r.to_dict()
r2 = Recipe.from_dict(d)
self.assertEqual(r, r2)
# JSON 往返
r3 = Recipe.from_dict(json.loads(json.dumps(d)))
self.assertEqual(r, r3)
def test_invalid_backbone_raises(self):
with self.assertRaises(QualityForecastError):
Recipe(name="t", backbone="svm")
def test_accuracy_floor_out_of_range(self):
with self.assertRaises(QualityForecastError):
Recipe(name="t", accuracy_floor=1.5)
with self.assertRaises(QualityForecastError):
Recipe(name="t", accuracy_floor=-0.1)
def test_missing_name(self):
with self.assertRaises(QualityForecastError):
Recipe(name="")
def test_load_recipe_from_file(self, ):
path = sample_recipe_path("recipe.ti.json")
r = load_recipe(path)
self.assertEqual(r.name, "ti-cl4-quality")
self.assertEqual(r.backbone, "gbdt")
self.assertIn("furnace_temp", r.feature_columns)
class TestBackbones(unittest.TestCase):
"""主干工厂与注册表。"""
def test_builtin_backbones_registered(self):
for name in ("gbdt", "dnn", "stub"):
self.assertIn(name, BACKBONES)
def test_register_custom_backbone(self):
class _Custom(ModelHandle):
def __init__(self, p):
super().__init__("custom", p)
self._v = 1.0
def _fit_impl(self, X, y):
self._v = sum(y) / len(y)
def _predict_one(self, row):
return self._v
register_backbone("custom_test", lambda p: _Custom(p))
m = QualityForecastModel(backbone="custom_test")
X, y = _linear_dataset()
m.fit(X, y)
self.assertEqual(len(m.predict(X)), len(X))
# 清理避免污染其它用例
BACKBONES.pop("custom_test", None)
def test_unknown_backbone_raises(self):
with self.assertRaises(QualityForecastError):
QualityForecastModel(backbone="not_a_backbone")
def test_stub_predict_is_deterministic(self):
h = stub_backbone({})
X, y = _linear_dataset()
h.fit(X, y)
p1 = h.predict(X)
p2 = h.predict(X)
self.assertEqual(p1, p2)
self.assertTrue(all(isinstance(v, float) for v in p1))
def test_gbdt_factory_runs_with_or_without_sklearn(self):
# 无论 sklearn 是否存在都不应报错
h = gbdt_backbone({"n_estimators": 20, "max_depth": 2})
X, y = _linear_dataset()
h.fit(X, y)
preds = h.predict(X)
self.assertEqual(len(preds), len(y))
class TestModelContract(unittest.TestCase):
"""模型 fit/predict/evaluate 契约。"""
def test_fit_predict_shapes(self):
m = QualityForecastModel(backbone="stub")
X, y = _linear_dataset(20)
m.fit(X, y)
self.assertTrue(m.fitted)
preds = m.predict(X)
self.assertEqual(len(preds), len(y))
def test_predict_before_fit_raises(self):
m = QualityForecastModel(backbone="stub")
with self.assertRaises(QualityForecastError):
m.predict([[1.0, 2.0]])
def test_fit_mismatched_lengths_raises(self):
m = QualityForecastModel(backbone="stub")
with self.assertRaises(QualityForecastError):
m.fit([[1.0], [2.0]], [1.0])
def test_fit_empty_raises(self):
m = QualityForecastModel(backbone="stub")
with self.assertRaises(QualityForecastError):
m.fit([], [])
def test_to_dict_roundtrip_meta(self):
m = QualityForecastModel(backbone="gbdt",
hyperparams={"n_estimators": 5},
feature_columns=["a"],
target_column="y")
d = m.to_dict()
self.assertEqual(d["recipe_meta"]["backbone"], "gbdt")
self.assertIn("handle", d)
class TestAccuracy(unittest.TestCase):
"""验收口径(PRD 5.3:准确率 ≥ 90%)。"""
def test_perfect_predictions_pass(self):
y = [10.0, 20.0, 30.0, 40.0]
acc = Accuracy.compute(y, y, accuracy_floor=0.9)
self.assertAlmostEqual(acc.accuracy, 1.0)
self.assertAlmostEqual(acc.mae, 0.0)
self.assertAlmostEqual(acc.rmse, 0.0)
self.assertTrue(acc.passed)
def test_bad_predictions_fail(self):
y_true = [10.0, 20.0, 30.0, 40.0]
y_pred = [11.0, 50.0, 5.0, 80.0] # 大偏差
acc = Accuracy.compute(y_true, y_pred, accuracy_floor=0.9)
self.assertLess(acc.accuracy, 0.9)
self.assertFalse(acc.passed)
self.assertGreater(acc.mae, 0.0)
self.assertGreater(acc.rmse, 0.0)
def test_length_mismatch_raises(self):
with self.assertRaises(QualityForecastError):
Accuracy.compute([1.0, 2.0], [1.0])
def test_empty_raises(self):
with self.assertRaises(QualityForecastError):
Accuracy.compute([], [])
def test_evaluate_end_to_end(self):
# stub 主干在确定性、低噪声线性数据上应能给出确定性的验收结果
m = QualityForecastModel(backbone="stub", accuracy_floor=0.0)
X, y = _linear_dataset(30)
m.fit(X, y)
acc = m.evaluate(X, y)
self.assertIsInstance(acc, Accuracy)
self.assertEqual(acc.to_dict()["accuracy_floor"], 0.0)
class TestSampleRecipes(unittest.TestCase):
"""样例协议:同框架加载 Ti / 树脂两套配方均跑通(PRD 5.3 验收口径)。"""
def test_samples_present(self):
names = list_sample_recipes()
self.assertIn("recipe.ti.json", names)
self.assertIn("recipe.resin.json", names)
def test_build_from_each_sample_runs(self):
for name in ("recipe.ti.json", "recipe.resin.json"):
m = build_from_recipe(sample_recipe_path(name))
self.assertIn(m.recipe_meta["backbone"], ("gbdt", "dnn", "stub"))
# 用配方里声明的特征数构造一份演示数据跑通完整链路
feat = m.recipe_meta["feature_columns"]
n_feat = len(feat)
self.assertGreater(n_feat, 0)
X = [[float(i + j) for j in range(n_feat)] for i in range(12)]
y = [float(i % 4) + 1.0 for i in range(12)]
m.fit(X, y)
preds = m.predict(X)
self.assertEqual(len(preds), len(y))
acc = m.evaluate(X, y)
self.assertIsInstance(acc, Accuracy)
def test_two_recipes_share_same_code(self):
"""切换模板仅改配方,模型代码零改动(PRD 5.3)。"""
m1 = build_from_recipe(sample_recipe_path("recipe.ti.json"))
m2 = build_from_recipe(sample_recipe_path("recipe.resin.json"))
self.assertEqual(type(m1), type(m2))
self.assertNotEqual(m1.recipe_meta.get("recipe_name"),
m2.recipe_meta.get("recipe_name"))
if __name__ == "__main__":
unittest.main(verbosity=2)