# -*- 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])