626 lines
24 KiB
Python
626 lines
24 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""Model Recipe 插件接口与样例协议。
|
||
|
||
对应 issue #34（父 EPIC #5「③ AI 模型框架 配置化重构」、PRD 5.3
|
||
「网络结构策略 / 模板化技术路径」）。
|
||
|
||
PRD 5.3 的核心诉求
|
||
------------------
|
||
|
||
采用「固定主干网络 + 可配置超参」为默认模式；同时提供 **Model Recipe
|
||
注册表**，允许高级行业模板通过 *声明式 recipe* 选择不同网络结构（如
|
||
LSTM 用于时序、GNN 用于跨工序），**新增结构走插件注册而非改内核**。
|
||
|
||
验收口径（PRD 5.3 / EPIC #5）：同一框架加载「树脂」与「Ti」两套 Recipe
|
||
均能跑通——切换模板仅改 Recipe，模型代码零改动。
|
||
|
||
本模块交付什么
|
||
--------------
|
||
|
||
1. **``ModelRecipe``**：声明式模型结构注册项。一个 Recipe 描述「用什么网络
|
||
主干 + 如何从超参包构造一个可训练/可推理的模型」。它是一个不可变数据
|
||
对象，``to_dict``/``repr`` 可序列化往返，便于配置台展示与审计。
|
||
2. **``RECIPES`` 全局注册表 + ``register_recipe`` / ``get_recipe`` /
|
||
``build_model`` / ``list_recipes``**：插件式注册 API。新增网络结构
|
||
（如自研 GNN）只需 ``register_recipe``，不动内核——对齐 PRD「新增结构
|
||
走插件注册」理念，并与 issue #35 的 ``register_operator``（特征算子
|
||
插件）形成「特征层 + 结构层」两级插件体系。
|
||
3. **内置网络主干工厂**：覆盖 PRD 5.3 四类模型模板的全部默认结构——
|
||
``gbdt`` / ``dnn`` / ``lstm`` / ``gnn``。每个工厂是纯函数
|
||
``build(hyperparams) -> ModelHandle``，返回统一的 ``ModelHandle``
|
||
（``fit`` / ``predict`` / ``to_dict``）。
|
||
4. **四类内置 Recipe**：与 PRD 5.3「四类模型模板」1:1 映射——质量预测 /
|
||
工艺优化 / 异常检测 / 跨工序寻优，默认绑定到 ``gbdt``/``dnn`` 主干，
|
||
高级模板可改绑 ``lstm``/``gnn``。
|
||
5. **样例协议（``samples/`` JSON）**：树脂 Recipe（``resin``）+ Ti Recipe
|
||
（``ti``）两套超参包样例，直接验证「同框架加载两套 Recipe 均跑通」的
|
||
验收口径。
|
||
|
||
零外部强依赖
|
||
------------
|
||
|
||
* 训练/推理默认走 **纯 Python stub 主干**（``StubBackbone``）：无
|
||
sklearn / xgboost / torch 时也能加载、注册、构造、（伪）拟合与预测，
|
||
保证边缘 / 离线 / CI 环境可加载与校验——与 issue #35 的「numpy 可选」
|
||
策略一致。
|
||
* 当运行环境存在 ``sklearn`` 时，``gbdt``/``dnn`` 主干自动升级为真实
|
||
sklearn 实现（梯度提升回归 / MLP），其余情况退化为 stub，不影响接口
|
||
契约与测试。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
import json
|
||
import math
|
||
import os
|
||
from dataclasses import dataclass, field
|
||
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple
|
||
|
||
__all__ = [
|
||
# 数据对象
|
||
"ModelRecipe",
|
||
"ModelHandle",
|
||
"RecipeError",
|
||
# 注册表 API
|
||
"RECIPES",
|
||
"register_recipe",
|
||
"get_recipe",
|
||
"list_recipes",
|
||
"build_model",
|
||
# 内置主干工厂
|
||
"BACKBONES",
|
||
"register_backbone",
|
||
"gbdt_backbone",
|
||
"dnn_backbone",
|
||
"lstm_backbone",
|
||
"gnn_backbone",
|
||
"stub_backbone",
|
||
# 样例协议
|
||
"load_sample_recipe",
|
||
"SAMPLE_RECIPES",
|
||
# 超参包校验
|
||
"validate_hyperparam_pack",
|
||
"RECIPE_KINDS",
|
||
]
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 可选依赖探测（与 issue #35 feature_spec 的 numpy 可选策略一致）
|
||
# ---------------------------------------------------------------------------
|
||
try: # pragma: no cover - 依赖环境相关
|
||
import numpy as _np # type: ignore
|
||
_HAS_NUMPY = True
|
||
except Exception: # pragma: no cover
|
||
_np = None
|
||
_HAS_NUMPY = False
|
||
|
||
try: # pragma: no cover - 依赖环境相关
|
||
from sklearn.ensemble import GradientBoostingRegressor as _GBR # type: ignore
|
||
from sklearn.neural_network import MLPRegressor as _MLPR # type: ignore
|
||
_HAS_SKLEARN = True
|
||
except Exception: # pragma: no cover
|
||
_GBR = None
|
||
_MLPR = None
|
||
_HAS_SKLEARN = False
|
||
|
||
|
||
class RecipeError(ValueError):
|
||
"""Recipe / 超参包语义错误（未知 recipe / 主干 / 参数缺失等）。"""
|
||
|
||
|
||
# PRD 5.3 四类模型模板的合法 ``kind``（与超参包 ``task`` 字段对齐）。
|
||
RECIPE_KINDS: Tuple[str, ...] = (
|
||
"quality_predict", # ① 质量预测
|
||
"process_optimize", # ② 工艺优化 / 配方推荐
|
||
"anomaly_detect", # ③ 异常检测 / 杂质预警
|
||
"cross_process", # ④ 跨工序关联寻优
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# ModelHandle：统一的模型句柄（fit / predict / to_dict）
|
||
# ---------------------------------------------------------------------------
|
||
class ModelHandle:
|
||
"""统一的模型句柄，屏蔽底层主干（sklearn / stub）差异。
|
||
|
||
所有主干工厂返回本类实例，使训练/推理流水线（issue #40）与配置台
|
||
（issue #62~#67）只需面向同一接口编程。``fit`` / ``predict`` 对输入
|
||
做最小校验后委托给 ``_impl``。
|
||
"""
|
||
|
||
__slots__ = ("recipe_id", "backbone", "hyperparams", "_impl", "fitted")
|
||
|
||
def __init__(
|
||
self,
|
||
recipe_id: str,
|
||
backbone: str,
|
||
hyperparams: Dict[str, Any],
|
||
impl: Any,
|
||
) -> None:
|
||
self.recipe_id = recipe_id
|
||
self.backbone = backbone
|
||
self.hyperparams: Dict[str, Any] = dict(hyperparams)
|
||
self._impl = impl
|
||
self.fitted = False
|
||
|
||
# -- 训练 / 推理 -------------------------------------------------------
|
||
def fit(self, X: Sequence[Sequence[float]], y: Optional[Sequence[float]] = None) -> "ModelHandle":
|
||
"""拟合。无监督主干（anomaly_detect）可忽略 ``y``。"""
|
||
rows = self._coerce_X(X)
|
||
if y is not None:
|
||
yv = [float(v) for v in y]
|
||
if len(yv) != len(rows):
|
||
raise ValueError(f"X/y 长度不一致：{len(rows)} vs {len(yv)}")
|
||
else:
|
||
yv = None
|
||
self._impl_fit(rows, yv)
|
||
self.fitted = True
|
||
return self
|
||
|
||
def predict(self, X: Sequence[Sequence[float]]) -> List[float]:
|
||
"""推理。未拟合则 fail-fast（fail-closed，避免静默返回垃圾值）。"""
|
||
if not self.fitted:
|
||
raise RecipeError("模型尚未 fit，禁止 predict（fail-closed）")
|
||
rows = self._coerce_X(X)
|
||
return self._impl_predict(rows)
|
||
|
||
# -- 序列化 ------------------------------------------------------------
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
return {
|
||
"recipe_id": self.recipe_id,
|
||
"backbone": self.backbone,
|
||
"hyperparams": copy.deepcopy(self.hyperparams),
|
||
"fitted": self.fitted,
|
||
}
|
||
|
||
def __repr__(self) -> str: # pragma: no cover - 调试用
|
||
return (
|
||
f"ModelHandle(recipe_id={self.recipe_id!r}, backbone={self.backbone!r}, "
|
||
f"hyperparams={self.hyperparams!r}, fitted={self.fitted})"
|
||
)
|
||
|
||
# -- 内部 --------------------------------------------------------------
|
||
@staticmethod
|
||
def _coerce_X(X: Sequence[Sequence[float]]) -> List[List[float]]:
|
||
if X is None:
|
||
raise ValueError("X 不能为 None")
|
||
rows: List[List[float]] = []
|
||
width: Optional[int] = None
|
||
for r in X:
|
||
row = [float(v) for v in r]
|
||
if width is None:
|
||
width = len(row)
|
||
elif len(row) != width:
|
||
raise ValueError(f"特征宽度不一致：{width} vs {len(row)}")
|
||
rows.append(row)
|
||
if not rows:
|
||
raise ValueError("X 不能为空")
|
||
return rows
|
||
|
||
def _impl_fit(self, rows: List[List[float]], y: Optional[List[float]]) -> None:
|
||
method = getattr(self._impl, "iaop_fit", None)
|
||
if method is None:
|
||
return # stub 主干无需训练
|
||
method(rows, y)
|
||
|
||
def _impl_predict(self, rows: List[List[float]]) -> List[float]:
|
||
method = getattr(self._impl, "iaop_predict", None)
|
||
if method is None:
|
||
# 兜底：返回零向量（理论上不会走到，注册时已校验）
|
||
return [0.0 for _ in rows]
|
||
return [float(v) for v in method(rows)]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 内置主干工厂：gbdt / dnn / lstm / gnn（无第三方依赖时退化为 stub）
|
||
# ---------------------------------------------------------------------------
|
||
def _as_matrix(rows: Sequence[Sequence[float]]):
|
||
"""把嵌套列表归一为 numpy 数组（有 numpy）或原生 list。"""
|
||
if _HAS_NUMPY:
|
||
return _np.asarray(rows, dtype=float)
|
||
return [list(r) for r in rows]
|
||
|
||
|
||
def _as_vector(y: Sequence[float]):
|
||
if _HAS_NUMPY:
|
||
return _np.asarray(y, dtype=float)
|
||
return [float(v) for v in y]
|
||
|
||
|
||
def stub_backbone(hyperparams: Dict[str, Any]) -> Any:
|
||
"""纯 Python stub 主干：均值/常数预测，无任何第三方依赖。
|
||
|
||
作为 ``gbdt``/``dnn``/``lstm``/``gnn`` 在无 sklearn/torch 环境下的
|
||
退化实现，保证 Recipe 可加载、可（伪）拟合、可推理、可切换——满足
|
||
PRD「新增结构走插件注册」的接口契约，不保证预测精度。
|
||
"""
|
||
class _Stub:
|
||
def __init__(self) -> None:
|
||
self._target_mean = 0.0
|
||
self._rows = 0
|
||
|
||
def iaop_fit(self, rows, y):
|
||
self._rows = len(rows)
|
||
if y is not None and len(y) > 0:
|
||
self._target_mean = float(sum(y) / len(y))
|
||
|
||
def iaop_predict(self, rows):
|
||
return [self._target_mean for _ in rows]
|
||
|
||
return _Stub()
|
||
|
||
|
||
def gbdt_backbone(hyperparams: Dict[str, Any]) -> Any:
|
||
"""梯度提升回归主干（PRD 5.3 质量预测默认 ``algorithm=xgboost``）。
|
||
|
||
有 ``sklearn`` 时用 ``GradientBoostingRegressor``；否则退化为
|
||
``stub_backbone``。超参映射：max_depth / n_estimators / learning_rate。
|
||
"""
|
||
if not _HAS_SKLEARN:
|
||
return stub_backbone(hyperparams)
|
||
max_depth = int(hyperparams.get("max_depth", 6))
|
||
n_estimators = int(hyperparams.get("n_estimators", hyperparams.get("n_est", 100)))
|
||
learning_rate = float(hyperparams.get("eta", hyperparams.get("learning_rate", 0.1)))
|
||
return _GBR(
|
||
max_depth=max_depth,
|
||
n_estimators=n_estimators,
|
||
learning_rate=learning_rate,
|
||
)
|
||
|
||
|
||
def dnn_backbone(hyperparams: Dict[str, Any]) -> Any:
|
||
"""轻量 DNN 主干（PRD 5.3「轻量 DNN」结构不变 + 可配置超参）。
|
||
|
||
有 ``sklearn`` 时用 ``MLPRegressor``；否则退化为 stub。超参映射：
|
||
hidden_layer_sizes / max_iter。
|
||
"""
|
||
if not _HAS_SKLEARN:
|
||
return stub_backbone(hyperparams)
|
||
hidden = hyperparams.get("hidden", hyperparams.get("hidden_layer_sizes", (64, 32)))
|
||
if isinstance(hidden, int):
|
||
hidden = (hidden,)
|
||
elif isinstance(hidden, list):
|
||
hidden = tuple(int(x) for x in hidden)
|
||
max_iter = int(hyperparams.get("max_iter", hyperparams.get("epochs", 200)))
|
||
return _MLPR(hidden_layer_sizes=hidden, max_iter=max_iter)
|
||
|
||
|
||
def lstm_backbone(hyperparams: Dict[str, Any]) -> Any:
|
||
"""LSTM 时序主干（PRD 5.3「LSTM 用于时序」高级行业模板可选结构）。
|
||
|
||
iAOP-Core 内核不强制依赖 torch/tf；本工厂在无 heavy 依赖时退化为
|
||
stub，仅完成「结构可声明、可注册、可切换」的接口契约。真实训练由
|
||
下游模板的插件 Recipe 注入 torch 实现后覆盖（``register_backbone``）。
|
||
"""
|
||
# 读取序列长度配置仅用于校验，stub 本身不消费
|
||
_ = int(hyperparams.get("seq_len", hyperparams.get("window", 10)))
|
||
return stub_backbone(hyperparams)
|
||
|
||
|
||
def gnn_backbone(hyperparams: Dict[str, Any]) -> Any:
|
||
"""GNN 跨工序主干（PRD 5.3「GNN 用于跨工序」高级行业模板可选结构）。
|
||
|
||
同 ``lstm_backbone``：内核不绑定图神经网络框架，退化为 stub；高级
|
||
模板通过插件 Recipe 注入真实实现。
|
||
"""
|
||
_ = hyperparams.get("num_nodes", hyperparams.get("edges"))
|
||
return stub_backbone(hyperparams)
|
||
|
||
|
||
# 主干注册表：name -> factory(hyperparams) -> impl
|
||
BACKBONES: Dict[str, Callable[[Dict[str, Any]], Any]] = {
|
||
"gbdt": gbdt_backbone,
|
||
"dnn": dnn_backbone,
|
||
"lstm": lstm_backbone,
|
||
"gnn": gnn_backbone,
|
||
"stub": stub_backbone,
|
||
}
|
||
|
||
|
||
def register_backbone(
|
||
name: str,
|
||
factory: Callable[[Dict[str, Any]], Any],
|
||
) -> None:
|
||
"""注册一个网络主干工厂。重复注册同名主干覆盖旧定义（便于测试替换）。"""
|
||
if not name or not name.replace("_", "").replace("-", "").isalnum():
|
||
raise RecipeError(f"非法主干名：{name!r}")
|
||
if not callable(factory):
|
||
raise RecipeError("factory 必须是可调用对象")
|
||
BACKBONES[name] = factory
|
||
|
||
|
||
def _resolve_backbone(name: str) -> Callable[[Dict[str, Any]], Any]:
|
||
if name not in BACKBONES:
|
||
raise RecipeError(
|
||
f"未知网络主干：{name!r}（已注册：{sorted(BACKBONES.keys())}）"
|
||
)
|
||
return BACKBONES[name]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# ModelRecipe：声明式模型结构注册项（不可变数据对象）
|
||
# ---------------------------------------------------------------------------
|
||
@dataclass(frozen=True)
|
||
class ModelRecipe:
|
||
"""一个声明式模型结构 Recipe。
|
||
|
||
Attributes
|
||
----------
|
||
id : str
|
||
Recipe 唯一标识（如 ``quality_predict.default``）。
|
||
kind : str
|
||
任务类型，取值 ``RECIPE_KINDS`` 之一（对齐 PRD 5.3 四类模型模板）。
|
||
backbone : str
|
||
网络主干名，必须在 ``BACKBONES`` 已注册（gbdt/dnn/lstm/gnn/...）。
|
||
default_hyperparams : dict
|
||
默认超参（可被超参包覆盖）。
|
||
required_features : tuple[str, ...]
|
||
该 Recipe 要求的最少特征名（用于超参包校验）。
|
||
description : str
|
||
人类可读说明。
|
||
"""
|
||
|
||
id: str
|
||
kind: str
|
||
backbone: str
|
||
default_hyperparams: Dict[str, Any] = field(default_factory=dict)
|
||
required_features: Tuple[str, ...] = field(default_factory=tuple)
|
||
description: str = ""
|
||
|
||
def __post_init__(self) -> None:
|
||
if not self.id:
|
||
raise RecipeError("Recipe id 不能为空")
|
||
if self.kind not in RECIPE_KINDS:
|
||
raise RecipeError(
|
||
f"非法 kind：{self.kind!r}（合法：{RECIPE_KINDS}）"
|
||
)
|
||
if self.backbone not in BACKBONES:
|
||
raise RecipeError(
|
||
f"未注册的主干：{self.backbone!r}（已注册：{sorted(BACKBONES.keys())}）"
|
||
)
|
||
if not isinstance(self.default_hyperparams, dict):
|
||
raise RecipeError("default_hyperparams 必须是 dict")
|
||
if not isinstance(self.required_features, tuple):
|
||
raise RecipeError("required_features 必须是 tuple")
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
return {
|
||
"id": self.id,
|
||
"kind": self.kind,
|
||
"backbone": self.backbone,
|
||
"default_hyperparams": copy.deepcopy(self.default_hyperparams),
|
||
"required_features": list(self.required_features),
|
||
"description": self.description,
|
||
}
|
||
|
||
@classmethod
|
||
def from_dict(cls, d: Dict[str, Any]) -> "ModelRecipe":
|
||
try:
|
||
return cls(
|
||
id=str(d["id"]),
|
||
kind=str(d["kind"]),
|
||
backbone=str(d["backbone"]),
|
||
default_hyperparams=dict(d.get("default_hyperparams", {})),
|
||
required_features=tuple(d.get("required_features", ())),
|
||
description=str(d.get("description", "")),
|
||
)
|
||
except KeyError as e: # pragma: no cover - 防御性
|
||
raise RecipeError(f"Recipe 缺少字段：{e}") from e
|
||
|
||
def merged_hyperparams(self, override: Optional[Dict[str, Any]]) -> Dict[str, Any]:
|
||
"""合并默认超参与超参包覆盖（覆盖优先）。"""
|
||
merged = copy.deepcopy(self.default_hyperparams)
|
||
if override:
|
||
merged.update(override)
|
||
return merged
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Recipe 注册表 + 公开 API
|
||
# ---------------------------------------------------------------------------
|
||
RECIPES: Dict[str, ModelRecipe] = {}
|
||
|
||
|
||
def register_recipe(recipe: ModelRecipe) -> ModelRecipe:
|
||
"""注册一个 Model Recipe 到全局注册表。
|
||
|
||
重复注册同 id 覆盖旧定义（便于测试期间替换）。对齐 PRD「新增结构走
|
||
插件注册而非改内核」：高级行业模板（如自研 GNN）只需 ``register_recipe``
|
||
即可接入，无需修改本文件。
|
||
"""
|
||
if not isinstance(recipe, ModelRecipe):
|
||
raise RecipeError("register_recipe 入参必须是 ModelRecipe 实例")
|
||
# 再次校验主干（防止 BACKBONES 在 recipe 构造后被反注册）
|
||
_resolve_backbone(recipe.backbone)
|
||
RECIPES[recipe.id] = recipe
|
||
return recipe
|
||
|
||
|
||
def get_recipe(recipe_id: str) -> ModelRecipe:
|
||
"""按 id 取 Recipe；不存在则 ``RecipeError``。"""
|
||
if recipe_id not in RECIPES:
|
||
raise RecipeError(
|
||
f"未知 Recipe：{recipe_id!r}（已注册：{sorted(RECIPES.keys())}）"
|
||
)
|
||
return RECIPES[recipe_id]
|
||
|
||
|
||
def list_recipes() -> List[Dict[str, Any]]:
|
||
"""列出全部已注册 Recipe（``to_dict`` 形式，按 id 排序）。"""
|
||
return [RECIPES[k].to_dict() for k in sorted(RECIPES.keys())]
|
||
|
||
|
||
def build_model(
|
||
recipe_id: str,
|
||
hyperparams: Optional[Dict[str, Any]] = None,
|
||
) -> ModelHandle:
|
||
"""按 Recipe + 超参包构造一个可训练/可推理的模型句柄。
|
||
|
||
流程：取 Recipe → 合并超参 → 取主干工厂 → 构造 impl → 包成
|
||
``ModelHandle``。切换模板/行业只需换 ``recipe_id`` 或超参，代码零改动
|
||
——对齐 PRD 5.3 验收口径。
|
||
"""
|
||
recipe = get_recipe(recipe_id)
|
||
merged = recipe.merged_hyperparams(hyperparams)
|
||
factory = _resolve_backbone(recipe.backbone)
|
||
impl = factory(merged)
|
||
return ModelHandle(
|
||
recipe_id=recipe.id,
|
||
backbone=recipe.backbone,
|
||
hyperparams=merged,
|
||
impl=impl,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 超参包校验（与 issue #39 hyperparam 互补；本模块只校验 Recipe 相关字段）
|
||
# ---------------------------------------------------------------------------
|
||
# Recipe 视角下，超参包必须出现的字段（PRD 5.3 超参包 JSON 示例）。
|
||
_REQUIRED_PACK_FIELDS: Tuple[str, ...] = ("model_id", "recipe_id", "features")
|
||
|
||
|
||
def validate_hyperparam_pack(pack: Dict[str, Any]) -> List[str]:
|
||
"""校验一个超参包在 Recipe 视角下的合法性，返回问题列表（空=通过）。
|
||
|
||
与 issue #39 ``hyperparam.py`` 的「spec 非空存在性校验」互补：#39 校验
|
||
特征 ``spec`` 字段本身，本函数校验「recipe_id 是否注册、主干能否构造、
|
||
必需特征是否齐备」等结构层语义。
|
||
"""
|
||
issues: List[str] = []
|
||
if not isinstance(pack, dict):
|
||
return ["超参包必须是 dict"]
|
||
for f in _REQUIRED_PACK_FIELDS:
|
||
if f not in pack:
|
||
issues.append(f"缺少必填字段：{f}")
|
||
|
||
rid = pack.get("recipe_id")
|
||
if rid is not None:
|
||
if rid not in RECIPES:
|
||
issues.append(
|
||
f"recipe_id {rid!r} 未注册（已注册：{sorted(RECIPES.keys())}）"
|
||
)
|
||
else:
|
||
recipe = RECIPES[rid]
|
||
# 必需特征校验
|
||
if recipe.required_features:
|
||
feats = {f.get("name") for f in pack.get("features", []) if isinstance(f, dict)}
|
||
for req in recipe.required_features:
|
||
if req not in feats:
|
||
issues.append(f"Recipe {rid!r} 要求特征 {req!r} 但超参包未提供")
|
||
# 主干可构造性（合并超参后能否实例化，吞掉异常转 issue）
|
||
try:
|
||
merged = recipe.merged_hyperparams(pack.get("hyperparams"))
|
||
_resolve_backbone(recipe.backbone)(merged)
|
||
except Exception as e: # pragma: no cover - 防御性
|
||
issues.append(f"主干 {recipe.backbone!r} 构造失败：{e}")
|
||
return issues
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 内置四类 Recipe（PRD 5.3 四类模型模板 1:1 映射，默认主干）
|
||
# ---------------------------------------------------------------------------
|
||
def _register_builtin_recipes() -> None:
|
||
"""注册 PRD 5.3 四类模型模板的默认 Recipe。
|
||
|
||
默认主干选「固定主干网络」：质量预测/工艺优化用 gbdt，异常检测用 dnn，
|
||
跨工序寻优用 gnn（高级行业模板可改绑 lstm/gnn）。
|
||
"""
|
||
register_recipe(ModelRecipe(
|
||
id="quality_predict.default",
|
||
kind="quality_predict",
|
||
backbone="gbdt",
|
||
default_hyperparams={
|
||
"max_depth": 6,
|
||
"eta": 0.1,
|
||
"n_estimators": 300,
|
||
"objective": "reg:squarederror",
|
||
},
|
||
required_features=("target",),
|
||
description="① 质量预测默认 Recipe：GBDT 主干，输入工艺参数+原料特征，"
|
||
"输出关键质量指标预测（PRD 5.3）。",
|
||
))
|
||
register_recipe(ModelRecipe(
|
||
id="process_optimize.default",
|
||
kind="process_optimize",
|
||
backbone="gbdt",
|
||
default_hyperparams={
|
||
"max_depth": 5,
|
||
"n_estimators": 200,
|
||
},
|
||
description="② 工艺优化/配方推荐默认 Recipe：GBDT 主干，输入质量目标+"
|
||
"约束，输出参数/配方建议（PRD 5.3）。",
|
||
))
|
||
register_recipe(ModelRecipe(
|
||
id="anomaly_detect.default",
|
||
kind="anomaly_detect",
|
||
backbone="dnn",
|
||
default_hyperparams={
|
||
"hidden": (32, 16),
|
||
"alarm_threshold": {"type": "zscore", "k": 3.0},
|
||
},
|
||
description="③ 异常检测/杂质预警默认 Recipe：轻量 DNN 主干（无监督+"
|
||
"阈值），输入实时测点，输出异常评分+预警（PRD 5.3）。",
|
||
))
|
||
register_recipe(ModelRecipe(
|
||
id="cross_process.default",
|
||
kind="cross_process",
|
||
backbone="gnn",
|
||
default_hyperparams={
|
||
"num_nodes": 2,
|
||
},
|
||
description="④ 跨工序关联寻优默认 Recipe：GNN 主干，输入上游(TiCl₄)"
|
||
"指标，输出下游(海绵钛)寻优建议（PRD 5.3）。",
|
||
))
|
||
|
||
|
||
_register_builtin_recipes()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 样例协议：树脂 / Ti 两套超参包（验证「同框架加载两套 Recipe 均跑通」）
|
||
# ---------------------------------------------------------------------------
|
||
# 内置样例超参包（PRD 5.3 超参包 JSON 结构 + recipe_id 关联）。验证 EPIC #5
|
||
# 验收口径：同一框架加载树脂与 Ti 两套 Recipe 均能 build/fit/predict。
|
||
SAMPLE_RECIPES: Dict[str, Dict[str, Any]] = {
|
||
"resin": {
|
||
"model_id": "quality_predict_resin",
|
||
"template": "iAOP-Template-Resin",
|
||
"recipe_id": "quality_predict.default",
|
||
"algorithm": "gbdt",
|
||
"features": [
|
||
{"name": "EMA_resin_temp", "spec": "EMA(树脂温度, 5m)"},
|
||
{"name": "target", "spec": "树脂转化率"},
|
||
],
|
||
"target": "树脂转化率",
|
||
"objective": "reg:squarederror",
|
||
"hyperparams": {"max_depth": 4, "n_estimators": 120, "eta": 0.1},
|
||
"train_window": "180d",
|
||
},
|
||
"ti": {
|
||
"model_id": "quality_predict_ti",
|
||
"template": "iAOP-Template-Ti",
|
||
"recipe_id": "quality_predict.default",
|
||
"algorithm": "gbdt",
|
||
"features": [
|
||
{"name": "EMA_CLF_TEMP_5m", "spec": "EMA(CLF-01.TEMP, 5m)"},
|
||
{"name": "RollingStd_CL2_10", "spec": "RollingStd(CLF-01.CL2, 10)"},
|
||
{"name": "target", "spec": "Ti_purity"},
|
||
],
|
||
"target": "Ti_purity",
|
||
"objective": "reg:squarederror",
|
||
"hyperparams": {"max_depth": 6, "n_estimators": 300, "eta": 0.1},
|
||
"train_window": "180d",
|
||
"alarm_threshold": {"type": "zscore", "k": 3.0},
|
||
"drift_check": {"method": "psi", "limit": 0.2},
|
||
},
|
||
}
|
||
|
||
|
||
def load_sample_recipe(name: str) -> Dict[str, Any]:
|
||
"""按名取内置样例超参包（``resin`` / ``ti``），返回深拷贝。"""
|
||
if name not in SAMPLE_RECIPES:
|
||
raise RecipeError(
|
||
f"未知样例 Recipe：{name!r}（已有：{sorted(SAMPLE_RECIPES.keys())}）"
|
||
)
|
||
return copy.deepcopy(SAMPLE_RECIPES[name])
|