core/model-framework/README.md

@@ -0,0 +1,107 @@
+# iAOP-Core · model_framework（③ AI 模型框架 · 配置化重构）
+
+本目录承载 EPIC #5「③ AI 模型框架 配置化重构」（PRD 5.3）的内核模块。PRD 5.3
+要求把"写死在树脂场景"的 4 类模型重构为**可配置模板**，由超参包驱动，核心诉求：
+
+> **固定主干网络 + 可配置超参**为默认；同时提供 **Model Recipe 注册表**，允许
+> 高级行业模板通过声明式 recipe 选择不同网络结构（LSTM 用于时序、GNN 用于跨工序），
+> **新增结构走插件注册而非改内核**。
+
+## 模块一览
+
+| 文件 | Issue | 职责 |
+|------|-------|------|
+| `model_recipe.py` | **#34** | Model Recipe 插件接口与样例协议（**结构层**插件注册表） |
+| `feature_spec.py` | #35（并行 PR） | FeatureSpec 声明式特征定义引擎（**特征层**插件注册表） |
+
+> 注：`feature_spec.py`（#35）与 `model_recipe.py`（#34）是 EPIC #5 下两个并行
+> 子任务，各自独立但理念呼应——#35 在**特征算子**层提供 `register_operator`
+> 插件注册，#34 在**模型结构**层提供 `register_recipe`/`register_backbone`
+> 插件注册，共同构成「特征层 + 结构层」两级插件体系。两个 PR 合入后本 README
+> 会整合两段说明。
+
+---
+
+## `model_recipe.py` — Model Recipe 插件接口（issue #34）
+
+对应 PRD 5.3「网络结构策略 / 模板化技术路径」与 EPIC #5 验收口径：**同一框架
+加载树脂与 Ti 两套 Recipe 均能跑通——切换模板仅改 Recipe，模型代码零改动**。
+
+### 设计要点
+
+* **声明式 Recipe**：`ModelRecipe` 是不可变数据对象（dataclass frozen），描述
+  「用什么网络主干 + 默认超参 + 必需特征」。`to_dict`/`from_dict` 可序列化往返，
+  便于配置台（#62~#67）展示与审计。
+* **两级插件注册**：
+  * `register_backbone(name, factory)` —— 注册网络主干工厂（结构层扩展点）；
+  * `register_recipe(ModelRecipe(...))` —— 注册声明式 Recipe（业务层扩展点）。
+  * 高级行业模板（如自研 GNN）只需 `register_backbone` + `register_recipe` 即可
+    接入内核，**零改码**（对齐 PRD「新增结构走插件注册」）。
+* **统一模型句柄**：所有主干工厂返回 `ModelHandle`（`fit`/`predict`/`to_dict`），
+  使训练/推理流水线（#40）与配置台只面向同一接口编程。
+* **零外部强依赖**：无 sklearn/torch 时，`gbdt`/`dnn`/`lstm`/`gnn` 主干自动
+  退化为纯 Python `stub`（均值预测），保证边缘/离线/CI 环境可加载、注册、构造、
+  （伪）拟合与推理。有 sklearn 时 `gbdt`/`dnn` 自动升级为真实实现。
+* **fail-closed**：未 `fit` 的模型 `predict` 直接抛错，绝不静默返回垃圾值。
+
+### 内置主干（覆盖 PRD 5.3 四类模型模板全部默认结构）
+
+| 主干 | 工厂 | 说明 | 有第三方依赖时 |
+|------|------|------|----------------|
+| `gbdt` | `gbdt_backbone` | 梯度提升回归（质量预测默认） | sklearn `GradientBoostingRegressor` |
+| `dnn` | `dnn_backbone` | 轻量 DNN（异常检测默认） | sklearn `MLPRegressor` |
+| `lstm` | `lstm_backbone` | LSTM 时序（高级模板可选） | 退化为 stub，插件注入真实实现 |
+| `gnn` | `gnn_backbone` | GNN 跨工序（高级模板可选） | 退化为 stub，插件注入真实实现 |
+| `stub` | `stub_backbone` | 纯 Python 均值预测（兜底） | — |
+
+### 内置四类 Recipe（与 PRD 5.3 四类模型模板 1:1 映射）
+
+```
+quality_predict.default    ① 质量预测      backbone=gbdt
+process_optimize.default   ② 工艺优化      backbone=gbdt
+anomaly_detect.default     ③ 异常检测      backbone=dnn
+cross_process.default      ④ 跨工序寻优    backbone=gnn
+```
+
+### 使用
+
+```python
+from model_framework.model_recipe import (
+    build_model, register_recipe, register_backbone, ModelRecipe,
+    validate_hyperparam_pack, load_sample_recipe,
+)
+
+# 1) 用内置 Recipe 构造模型并训练
+m = build_model("quality_predict.default", {"max_depth": 6, "n_estimators": 300})
+m.fit(X_train, y_train)
+pred = m.predict(X_test)
+
+# 2) 插件扩展：自研 GNN 主干，零改码接入
+register_backbone("my-gnn", lambda hp: MyGnnImpl(**hp))
+register_recipe(ModelRecipe(
+    id="cross_process.my_gnn", kind="cross_process", backbone="my-gnn",
+    default_hyperparams={"hidden": 128}, description="自研 GNN",
+))
+m2 = build_model("cross_process.my_gnn")
+
+# 3) 超参包校验（Recipe 视角，与 #39 的 spec 校验互补）
+pack = load_sample_recipe("ti")
+issues = validate_hyperparam_pack(pack)   # [] = 通过
+```
+
+### 样例协议（EPIC #5 验收口径）
+
+`SAMPLE_RECIPES` 内置 `resin`（树脂）与 `ti`（氯化车间/海绵钛）两套超参包样例，
+两者共用同一个 `recipe_id=quality_predict.default`，仅超参不同——直接验证
+「同一框架加载两套 Recipe 均能 build/fit/predict，切换模板仅改超参包，模型代码
+零改动」。
+
+## 测试
+
+```bash
+cd core/model-framework/tests
+python -m unittest test_model_recipe            # 30 个用例
+
+cd core/model-framework
+python _sanity_check.py                          # 端到端 sanity
+```

core/model-framework/__init__.py

@@ -0,0 +1,15 @@
+# -*- coding: utf-8 -*-
+"""iAOP-Core · model_framework 包入口。
+
+目录名 ``model-framework`` 含连字符，无法直接以 ``import model-framework``
+加载；本包通过 ``tests/_bootstrap.py`` 以包名 ``model_framework`` 挂载到
+``sys.modules`` 后再导入（与 ``core/data-bus`` 一致）。
+
+当前模块：
+    * ``model_recipe``  —— Model Recipe 插件接口与样例协议（issue #34）
+
+后续 issue（#35 FeatureSpec / #39 超参包校验）合入后，本 ``__init__`` 再
+re-export 其公开符号。
+"""
+
+__version__ = "0.1.0"

core/model-framework/_sanity_check.py

@@ -0,0 +1,77 @@
+# -*- coding: utf-8 -*-
+"""Sanity check: model_framework 包导入 + Model Recipe 插件接口可用性。
+
+直接 ``python _sanity_check.py`` 运行（在 ``core/model-framework`` 目录下）。
+验证 issue #34 的核心交付：内置 Recipe 注册、build_model 跨主干构造、
+插件扩展（register_backbone/register_recipe）、树脂+Ti 两套样例同框架跑通
+（EPIC #5 / PRD 5.3 验收口径）。
+"""
+import importlib.util
+import os
+import sys
+
+d = os.path.dirname(os.path.abspath(__file__))
+
+# 以包身份加载 model-framework（目录含连字符）
+spec = importlib.util.spec_from_file_location(
+    "model_framework", os.path.join(d, "__init__.py"),
+    submodule_search_locations=[d],
+)
+pkg = importlib.util.module_from_spec(spec)
+sys.modules["model_framework"] = pkg
+spec.loader.exec_module(pkg)
+
+from model_framework.model_recipe import (  # noqa: E402
+    BACKBONES,
+    RECIPE_KINDS,
+    ModelRecipe,
+    build_model,
+    get_recipe,
+    list_recipes,
+    load_sample_recipe,
+    register_backbone,
+    register_recipe,
+    validate_hyperparam_pack,
+)
+
+# 1) 内置四类 Recipe 与四类 kind 全覆盖
+ids = [r["id"] for r in list_recipes()]
+assert "quality_predict.default" in ids
+assert "process_optimize.default" in ids
+assert "anomaly_detect.default" in ids
+assert "cross_process.default" in ids
+kinds = {get_recipe(i).kind for i in ids if i.endswith(".default")}
+assert kinds == set(RECIPE_KINDS), f"kind 不全: {kinds}"
+print("内置 Recipe:", ids)
+
+# 2) build_model 跨主干构造 + fit/predict
+m = build_model("quality_predict.default", {"max_depth": 5})
+m.fit([[1.0, 2.0], [3.0, 4.0]], [1.0, 2.0])
+pred = m.predict([[2.0, 3.0]])
+assert len(pred) == 1
+print("gbdt fit/predict ok; pred=", pred[0])
+
+# 3) 插件扩展：自研主干 + 自定义 Recipe 零改码接入
+register_backbone("toy", lambda hp: type("T", (), {
+    "iaop_fit": lambda self, r, y: None,
+    "iaop_predict": lambda self, r: [7.0 for _ in r],
+})())
+register_recipe(ModelRecipe(
+    id="quality_predict.toy", kind="quality_predict", backbone="toy",
+    description="sanity: 自研主干插件接入",
+))
+m2 = build_model("quality_predict.toy")
+m2.fit([[1.0]], [1.0])
+assert m2.predict([[1.0], [2.0]]) == [7.0, 7.0]
+print("插件主干已注册:", "toy" in BACKBONES)
+
+# 4) EPIC #5 验收口径：树脂 + Ti 两套样例同框架均跑通
+for name in ("resin", "ti"):
+    pack = load_sample_recipe(name)
+    assert validate_hyperparam_pack(pack) == [], f"{name} 校验失败"
+    mm = build_model(pack["recipe_id"], pack.get("hyperparams"))
+    mm.fit([[1.0, 2.0], [3.0, 4.0], [5.0, 6.0]], [1.0, 2.0, 3.0])
+    assert len(mm.predict([[1.0, 2.0]])) == 1
+    print(f"样例 {name} recipe_id={pack['recipe_id']} 跑通")
+
+print("OK — issue #34 Model Recipe 插件接口 sanity check 通过")

core/model-framework/model_recipe.py

@@ -0,0 +1,625 @@
+# -*- 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])

core/model-framework/tests/_bootstrap.py

@@ -0,0 +1,17 @@
+# -*- coding: utf-8 -*-
+"""测试引导：把 `core/model-framework` 以包名 `model_framework` 挂载到 sys.modules。
+
+目录名 `model-framework` 含连字符，无法直接以包名 import；挂载后模块内
+``from model_framework.model_recipe import ...`` 在 unittest 发现机制下可
+正常解析（与 `core/data-bus/tests/_bootstrap.py` 一致）。
+"""
+import os
+import sys
+import types
+
+MODEL_FRAMEWORK_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, MODEL_FRAMEWORK_DIR)
+if "model_framework" not in sys.modules:
+    pkg = types.ModuleType("model_framework")
+    pkg.__path__ = [MODEL_FRAMEWORK_DIR]
+    sys.modules["model_framework"] = pkg

core/model-framework/tests/test_model_recipe.py

@@ -0,0 +1,303 @@
+# -*- coding: utf-8 -*-
+"""issue #34 Model Recipe 插件接口与样例协议 单元测试。
+
+覆盖：
+* 内置四类 Recipe 已注册、字段合法；
+* build_model 跨主干（gbdt/dnn/lstm/gnn/stub）可构造、fit/predict 契约；
+* 插件注册（register_recipe / register_backbone）零改码扩展；
+* ModelRecipe 不可变 + to_dict/from_dict 往返；
+* 超参包校验（recipe_id / 必需特征 / 主干可构造性）；
+* 样例协议：树脂 + Ti 两套 Recipe 同框架均跑通（EPIC #5 验收口径）。
+"""
+import os
+import sys
+
+# 引导：挂载 model_framework 包（目录含连字符）
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import _bootstrap  # noqa: F401,E402
+
+import unittest
+
+from model_framework.model_recipe import (  # noqa: E402
+    BACKBONES,
+    RECIPE_KINDS,
+    ModelRecipe,
+    RecipeError,
+    build_model,
+    get_recipe,
+    list_recipes,
+    load_sample_recipe,
+    register_backbone,
+    register_recipe,
+    validate_hyperparam_pack,
+)
+
+
+class TestBuiltinRecipes(unittest.TestCase):
+    """内置四类 Recipe 注册与字段合法性。"""
+
+    def test_four_builtin_recipes_registered(self):
+        ids = {r["id"] for r in list_recipes()}
+        for rid in (
+            "quality_predict.default",
+            "process_optimize.default",
+            "anomaly_detect.default",
+            "cross_process.default",
+        ):
+            self.assertIn(rid, ids, f"缺少内置 Recipe {rid}")
+
+    def test_each_builtin_kind_covered(self):
+        kinds = {get_recipe(rid).kind for rid in (
+            "quality_predict.default",
+            "process_optimize.default",
+            "anomaly_detect.default",
+            "cross_process.default",
+        )}
+        self.assertEqual(kinds, set(RECIPE_KINDS))
+
+    def test_backbone_registered(self):
+        for name in ("gbdt", "dnn", "lstm", "gnn", "stub"):
+            self.assertIn(name, BACKBONES, f"缺少内置主干 {name}")
+
+
+class TestModelRecipeDataclass(unittest.TestCase):
+    """ModelRecipe 不可变 + 序列化往返 + 校验。"""
+
+    def test_immutable(self):
+        r = get_recipe("quality_predict.default")
+        with self.assertRaises(Exception):
+            r.id = "x"  # type: ignore[misc]
+
+    def test_to_from_dict_roundtrip(self):
+        r = get_recipe("quality_predict.default")
+        d = r.to_dict()
+        r2 = ModelRecipe.from_dict(d)
+        self.assertEqual(r2.to_dict(), d)
+        self.assertEqual(r2.id, r.id)
+        self.assertEqual(r2.backbone, r.backbone)
+
+    def test_invalid_kind_rejected(self):
+        with self.assertRaises(RecipeError):
+            ModelRecipe(id="x.bad", kind="bogus", backbone="gbdt")
+
+    def test_unregistered_backbone_rejected(self):
+        with self.assertRaises(RecipeError):
+            ModelRecipe(id="x.nobackbone", kind="quality_predict", backbone="no-such")
+
+    def test_merged_hyperparams_override_wins(self):
+        r = get_recipe("quality_predict.default")
+        base = r.default_hyperparams
+        merged = r.merged_hyperparams({"max_depth": 99})
+        self.assertEqual(merged["max_depth"], 99)
+        # 默认值未被污染
+        self.assertEqual(base["max_depth"], 6)
+        self.assertIn("eta", merged)
+
+
+class TestBuildModel(unittest.TestCase):
+    """build_model 跨主干构造 + fit/predict 契约。"""
+
+    def test_build_each_backbone(self):
+        for rid, bb in (
+            ("quality_predict.default", "gbdt"),
+            ("process_optimize.default", "gbdt"),
+            ("anomaly_detect.default", "dnn"),
+            ("cross_process.default", "gnn"),
+        ):
+            m = build_model(rid)
+            self.assertEqual(m.backbone, bb)
+            self.assertFalse(m.fitted)
+
+    def test_fit_then_predict_returns_correct_length(self):
+        m = build_model("quality_predict.default")
+        X = [[1.0, 2.0], [3.0, 4.0], [5.0, 6.0]]
+        y = [1.0, 2.0, 3.0]
+        m.fit(X, y)
+        self.assertTrue(m.fitted)
+        pred = m.predict([[2.0, 3.0], [4.0, 5.0]])
+        self.assertEqual(len(pred), 2)
+        for v in pred:
+            self.assertIsInstance(v, float)
+
+    def test_predict_before_fit_fails_closed(self):
+        m = build_model("anomaly_detect.default")
+        with self.assertRaises(RecipeError):
+            m.predict([[1.0, 2.0]])
+
+    def test_unsupervised_fit_without_y(self):
+        # anomaly_detect 主干应允许无 y 拟合
+        m = build_model("anomaly_detect.default")
+        m.fit([[1.0, 2.0], [3.0, 4.0]])
+        self.assertTrue(m.fitted)
+        out = m.predict([[1.0, 2.0]])
+        self.assertEqual(len(out), 1)
+
+    def test_X_width_mismatch_rejected(self):
+        m = build_model("quality_predict.default")
+        with self.assertRaises(ValueError):
+            m.fit([[1.0, 2.0], [3.0]], [1.0, 2.0])
+
+    def test_Xy_length_mismatch_rejected(self):
+        m = build_model("quality_predict.default")
+        with self.assertRaises(ValueError):
+            m.fit([[1.0, 2.0], [3.0, 4.0]], [1.0])
+
+    def test_empty_X_rejected(self):
+        m = build_model("quality_predict.default")
+        with self.assertRaises(ValueError):
+            m.fit([], [])
+
+    def test_handle_to_dict(self):
+        m = build_model("quality_predict.default", {"max_depth": 7})
+        d = m.to_dict()
+        self.assertEqual(d["recipe_id"], "quality_predict.default")
+        self.assertEqual(d["backbone"], "gbdt")
+        self.assertEqual(d["hyperparams"]["max_depth"], 7)
+        self.assertFalse(d["fitted"])
+
+    def test_unknown_recipe_raises(self):
+        with self.assertRaises(RecipeError):
+            build_model("no.such.recipe")
+
+
+class TestPluginRegistration(unittest.TestCase):
+    """register_recipe / register_backbone 零改码扩展（PRD「新增结构走插件注册」）。"""
+
+    def test_register_custom_backbone_and_recipe(self):
+        seen = {}
+
+        def my_bb(hp):
+            class _Impl:
+                def iaop_fit(self, rows, y):
+                    seen["fit_called"] = True
+
+                def iaop_predict(self, rows):
+                    return [42.0 for _ in rows]
+            return _Impl()
+
+        register_backbone("my-gnn", my_bb)
+        self.assertIn("my-gnn", BACKBONES)
+
+        register_recipe(ModelRecipe(
+            id="cross_process.custom_gnn",
+            kind="cross_process",
+            backbone="my-gnn",
+            description="自研 GNN 主干，验证插件扩展",
+        ))
+        m = build_model("cross_process.custom_gnn")
+        m.fit([[1.0, 2.0]], [1.0])
+        self.assertTrue(seen.get("fit_called"))
+        self.assertEqual(m.predict([[9.0, 9.0]]), [42.0])
+
+    def test_register_recipe_overwrites(self):
+        # 用独立的临时 recipe 验证"重复注册同 id 覆盖"，不污染内置表
+        register_recipe(ModelRecipe(
+            id="quality_predict.temp",
+            kind="quality_predict",
+            backbone="gbdt",
+            description="第一版",
+        ))
+        self.assertEqual(get_recipe("quality_predict.temp").description, "第一版")
+        register_recipe(ModelRecipe(
+            id="quality_predict.temp",
+            kind="quality_predict",
+            backbone="stub",
+            description="第二版覆盖",
+        ))
+        self.assertEqual(get_recipe("quality_predict.temp").backbone, "stub")
+        self.assertEqual(get_recipe("quality_predict.temp").description, "第二版覆盖")
+
+    def test_register_invalid_backbone_name_rejected(self):
+        with self.assertRaises(RecipeError):
+            register_backbone("bad name!", lambda hp: None)
+
+    def test_register_non_callable_factory_rejected(self):
+        with self.assertRaises(RecipeError):
+            register_backbone("oops", "not callable")  # type: ignore[arg-type]
+
+    def test_register_non_recipe_rejected(self):
+        with self.assertRaises(RecipeError):
+            register_recipe("not a recipe")  # type: ignore[arg-type]
+
+
+class TestHyperparamPackValidation(unittest.TestCase):
+    """超参包校验（Recipe 视角）。"""
+
+    def test_valid_pack_no_issues(self):
+        pack = load_sample_recipe("ti")
+        self.assertEqual(validate_hyperparam_pack(pack), [])
+
+    def test_missing_required_field(self):
+        issues = validate_hyperparam_pack({"recipe_id": "quality_predict.default"})
+        msgs = " ".join(issues)
+        self.assertIn("model_id", msgs)
+        self.assertIn("features", msgs)
+
+    def test_unknown_recipe_id(self):
+        issues = validate_hyperparam_pack({
+            "model_id": "x", "recipe_id": "no.such", "features": [],
+        })
+        self.assertTrue(any("未注册" in i for i in issues))
+
+    def test_missing_required_feature(self):
+        # quality_predict.default 要求 'target' 特征
+        issues = validate_hyperparam_pack({
+            "model_id": "x",
+            "recipe_id": "quality_predict.default",
+            "features": [{"name": "only_a"}],
+        })
+        self.assertTrue(any("target" in i for i in issues))
+
+
+class TestSampleRecipesAcceptance(unittest.TestCase):
+    """EPIC #5 / PRD 5.3 验收口径：同框架加载树脂与 Ti 两套 Recipe 均跑通。"""
+
+    def test_both_samples_build_fit_predict(self):
+        for name in ("resin", "ti"):
+            pack = load_sample_recipe(name)
+            self.assertEqual(validate_hyperparam_pack(pack), [],
+                             f"样例 {name} 校验未通过")
+            m = build_model(pack["recipe_id"], pack.get("hyperparams"))
+            # 构造与目标维度无关的训练样本（2 特征列）
+            X = [[float(i), float(i + 1)] for i in range(6)]
+            y = [float(i) for i in range(6)]
+            m.fit(X, y)
+            self.assertTrue(m.fitted)
+            pred = m.predict([[1.0, 2.0]])
+            self.assertEqual(len(pred), 1)
+
+    def test_samples_share_same_framework(self):
+        # 关键：两套样例用同一个 recipe_id（quality_predict.default），
+        # 仅超参不同——证明「切换模板仅改超参包，模型代码零改动」
+        r1 = load_sample_recipe("resin")
+        r2 = load_sample_recipe("ti")
+        self.assertEqual(r1["recipe_id"], r2["recipe_id"])
+        # 但超参不同（max_depth 4 vs 6）
+        self.assertNotEqual(
+            r1["hyperparams"]["max_depth"],
+            r2["hyperparams"]["max_depth"],
+        )
+        # 各自 build 得到不同超参的句柄
+        m1 = build_model(r1["recipe_id"], r1["hyperparams"])
+        m2 = build_model(r2["recipe_id"], r2["hyperparams"])
+        self.assertEqual(m1.hyperparams["max_depth"], 4)
+        self.assertEqual(m2.hyperparams["max_depth"], 6)
+
+    def test_load_unknown_sample_raises(self):
+        with self.assertRaises(RecipeError):
+            load_sample_recipe("bogus")
+
+
+class TestBackboneFallback(unittest.TestCase):
+    """主干在无第三方依赖时退化为 stub，接口契约不变。"""
+
+    def test_lstm_gnn_fallback_to_stub_contract(self):
+        # 无论是否有 torch，lstm/gnn 主干都应能构造并 fit/predict
+        for rid in ("cross_process.default",):
+            m = build_model(rid)
+            m.fit([[1.0, 2.0]], [1.0])
+            self.assertEqual(len(m.predict([[1.0, 2.0]])), 1)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)