feat: add routed model training pipeline

README.md

@@ -37,12 +37,24 @@ tech-architecture/  技术架构报告与本地 Mermaid 资源
 - SQLite 骨架：`materials` 与 `experiments` 两张核心表，支持外键约束与实验记录往返。
 - 训练数据接口：从外部数据平台的 `material_records` 表只读加载记录，并分离元信息、特征和目标。
 - 目标拆分接口：按分类/回归任务组织 `is_cut_through`、`etching_depth` 等实验结果。
-- 训练预处理：数值特征采用中位数填补与标准化，类别特征采用众数填补与独热编码；未知类别在推理时安全忽略。
+- 训练预处理：数值特征采用中位数填补与标准化（全缺失列固定回退为 0），类别特征采用众数填补与独热编码；未知类别在推理时安全忽略。
 - 模型分派：连续目标默认路由至 GPR，布尔目标默认路由至随机森林分类器。
+- 训练流水线：按材料名称执行无泄漏 LOMO 验证，输出目标级指标并以 joblib 保存全量重训模型与预处理器。
 - 工程配置：`pyproject.toml` 统一依赖、pytest 与 Ruff 配置。
 
 后续将按架构报告逐步补齐 DoE 生成、CSV 交换、设备采集、建模管线和推理服务。
 
+## 训练一次完整模型
+
+当外部数据平台的 SQLite `material_records` 已填入完整实验结果后：
+
+```bash
+uv run python -m lmpm.scripts.train data/material_records.db data/models/lmpm.joblib
+```
+
+命令会同时生成模型文件 `lmpm.joblib` 和同名的 LOMO 验证指标报告
+`lmpm.metrics.json`。运行时数据与模型产物均应保留在 `data/`，不提交到 Git。
+
 ## 数据与产物
 
 本地数据库和导出的 CSV 统一放在 `data/`，这些运行时产物不会提交到 Git。数据库初始化后的表结构由 `src/lmpm/data/store.py` 维护。

pyproject.toml

@@ -8,6 +8,7 @@ version = "0.1.0"
 description = "Laser material parameter matching toolkit"
 requires-python = ">=3.10"
 dependencies = [
+  "joblib>=1.4,<2",
   "pandas>=2.2,<3",
   "pydantic>=2.7,<3",
   "pydantic-settings>=2.2,<3",

src/lmpm/scripts/__init__.py

@@ -0,0 +1 @@
+"""Command-line entry points for repeatable LMPM operations."""

src/lmpm/scripts/train.py

@@ -0,0 +1,36 @@
+"""Train every routed model from the external SQLite material dataset."""
+
+import argparse
+import json
+from pathlib import Path
+
+from lmpm.training.workflow import train_sqlite_database
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Train LMPM target models from a completed SQLite dataset."
+    )
+    parser.add_argument(
+        "database", type=Path, help="SQLite database with material_records"
+    )
+    parser.add_argument(
+        "artifact", type=Path, help="Destination .joblib model artifact"
+    )
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    bundle = train_sqlite_database(args.database, args.artifact)
+    report_path = args.artifact.with_suffix(".metrics.json")
+    report_path.write_text(
+        json.dumps(bundle.report(), ensure_ascii=False, indent=2) + "\n",
+        encoding="utf-8",
+    )
+    print(f"saved model artifact: {args.artifact}")
+    print(f"saved validation report: {report_path}")
+
+
+if __name__ == "__main__":
+    main()

src/lmpm/training/models.py

@@ -0,0 +1,247 @@
+"""Train, validate, persist, and serve the routed baseline estimators."""
+
+from dataclasses import asdict, dataclass
+from pathlib import Path
+from typing import Any
+
+import joblib
+import numpy as np
+from sklearn.ensemble import RandomForestClassifier
+from sklearn.gaussian_process import GaussianProcessRegressor
+from sklearn.gaussian_process.kernels import RBF, WhiteKernel
+from sklearn.metrics import (
+    accuracy_score,
+    f1_score,
+    mean_absolute_error,
+    r2_score,
+    roc_auc_score,
+    root_mean_squared_error,
+)
+from sklearn.model_selection import LeaveOneGroupOut
+
+from lmpm.training.dataset import TrainingDataset
+from lmpm.training.preprocessing import FeaturePreprocessor
+from lmpm.training.routing import TargetModelRoute, default_model_routes
+from lmpm.training.targets import split_targets
+
+Estimator = GaussianProcessRegressor | RandomForestClassifier
+
+
+@dataclass(frozen=True)
+class FoldEvaluation:
+    """Metrics for one material held out from model fitting."""
+
+    held_out_material: str
+    sample_count: int
+    metrics: dict[str, float | None]
+
+
+@dataclass(frozen=True)
+class TargetEvaluation:
+    """LOMO evaluation summary for one outcome field."""
+
+    route: TargetModelRoute
+    folds: tuple[FoldEvaluation, ...]
+    mean_metrics: dict[str, float | None]
+
+
+@dataclass(frozen=True)
+class TrainedTargetModel:
+    """A fitted preprocessor and estimator for one target field."""
+
+    route: TargetModelRoute
+    preprocessor: FeaturePreprocessor
+    estimator: Estimator
+
+
+@dataclass(frozen=True)
+class TrainingBundle:
+    """All final fitted target models and their validation evidence."""
+
+    models: dict[str, TrainedTargetModel]
+    evaluations: dict[str, TargetEvaluation]
+
+    def report(self) -> dict[str, Any]:
+        """Return a JSON-ready validation report without serializing estimators."""
+        return {
+            target_name: {
+                "task": evaluation.route.task,
+                "model_family": evaluation.route.model_family,
+                "mean_metrics": evaluation.mean_metrics,
+                "folds": [asdict(fold) for fold in evaluation.folds],
+            }
+            for target_name, evaluation in self.evaluations.items()
+        }
+
+
+def build_estimator(route: TargetModelRoute) -> Estimator:
+    """Build the deterministic baseline estimator compatible with one target."""
+    if route.model_family == "gaussian_process_regressor":
+        return GaussianProcessRegressor(
+            kernel=RBF(length_scale=1.0) + WhiteKernel(noise_level=1.0),
+            normalize_y=True,
+            optimizer=None,
+        )
+    return RandomForestClassifier(
+        n_estimators=300,
+        class_weight="balanced",
+        random_state=42,
+    )
+
+
+def _subset(dataset: TrainingDataset, indices: np.ndarray) -> TrainingDataset:
+    return TrainingDataset(records=tuple(dataset.records[index] for index in indices))
+
+
+def _target_values(dataset: TrainingDataset, route: TargetModelRoute) -> np.ndarray:
+    values: list[bool | float] = []
+    for record in dataset.records:
+        value = record.targets.get(route.target_name)
+        if value is None:
+            experiment_id = record.metadata.get("experiment_id", "<unknown>")
+            raise ValueError(f"{experiment_id}: missing target {route.target_name}")
+        if route.task == "classification":
+            if not isinstance(value, bool):
+                raise ValueError(f"{route.target_name} must be boolean")
+        elif isinstance(value, bool) or not isinstance(value, (int, float)):
+            raise ValueError(f"{route.target_name} must be numeric")
+        values.append(value)
+    return np.asarray(values)
+
+
+def _mean_metrics(folds: tuple[FoldEvaluation, ...]) -> dict[str, float | None]:
+    metric_names = {name for fold in folds for name in fold.metrics}
+    means: dict[str, float | None] = {}
+    for name in sorted(metric_names):
+        values = [
+            fold.metrics[name] for fold in folds if fold.metrics[name] is not None
+        ]
+        means[name] = float(np.mean(values)) if values else None
+    return means
+
+
+def _regression_metrics(
+    y_true: np.ndarray, y_predicted: np.ndarray
+) -> dict[str, float | None]:
+    return {
+        "mae": float(mean_absolute_error(y_true, y_predicted)),
+        "rmse": float(root_mean_squared_error(y_true, y_predicted)),
+        "r2": float(r2_score(y_true, y_predicted)) if len(y_true) > 1 else None,
+    }
+
+
+def _classification_metrics(
+    y_true: np.ndarray, y_predicted: np.ndarray
+) -> dict[str, float | None]:
+    return {
+        "accuracy": float(accuracy_score(y_true, y_predicted)),
+        "f1": float(f1_score(y_true, y_predicted, zero_division=0)),
+        "roc_auc": None,
+    }
+
+
+def _evaluate_fold(
+    route: TargetModelRoute,
+    estimator: Estimator,
+    test_features: np.ndarray,
+    test_targets: np.ndarray,
+) -> dict[str, float | None]:
+    predicted = estimator.predict(test_features)
+    if route.task == "regression":
+        return _regression_metrics(test_targets, predicted)
+
+    classifier = estimator
+    metrics = _classification_metrics(test_targets, predicted)
+    if len(np.unique(test_targets)) == 2 and True in classifier.classes_:
+        positive_index = list(classifier.classes_).index(True)
+        probabilities = classifier.predict_proba(test_features)[:, positive_index]
+        metrics["roc_auc"] = float(roc_auc_score(test_targets, probabilities))
+    return metrics
+
+
+def _evaluate_route(
+    dataset: TrainingDataset,
+    route: TargetModelRoute,
+    groups: np.ndarray,
+) -> TargetEvaluation:
+    target_values = _target_values(dataset, route)
+    splitter = LeaveOneGroupOut()
+    folds: list[FoldEvaluation] = []
+    splits = splitter.split(np.zeros(dataset.size), groups=groups)
+    for train_indices, test_indices in splits:
+        train_dataset = _subset(dataset, train_indices)
+        test_dataset = _subset(dataset, test_indices)
+        preprocessor = FeaturePreprocessor().fit(train_dataset)
+        train_features = preprocessor.transform(train_dataset).values
+        test_features = preprocessor.transform(test_dataset).values
+        estimator = build_estimator(route)
+        estimator.fit(train_features, target_values[train_indices])
+        folds.append(
+            FoldEvaluation(
+                held_out_material=str(groups[test_indices[0]]),
+                sample_count=len(test_indices),
+                metrics=_evaluate_fold(
+                    route,
+                    estimator,
+                    test_features,
+                    target_values[test_indices],
+                ),
+            )
+        )
+
+    frozen_folds = tuple(folds)
+    return TargetEvaluation(
+        route=route,
+        folds=frozen_folds,
+        mean_metrics=_mean_metrics(frozen_folds),
+    )
+
+
+def train_models(dataset: TrainingDataset) -> TrainingBundle:
+    """Run leakage-safe LOMO validation, then fit one final model per target.
+
+    Each validation fold fits its own feature preprocessor using only the
+    training materials. The returned estimators are finally refit on every
+    supplied record so they are ready for persisted deployment.
+    """
+    if dataset.size < 2:
+        raise ValueError("at least two completed experiment records are required")
+    split_targets(dataset)
+    groups = np.asarray(
+        [record.metadata.get("material_name", "") for record in dataset.records]
+    )
+    if not all(groups) or len(np.unique(groups)) < 2:
+        raise ValueError("LOMO validation requires records from at least two materials")
+
+    models: dict[str, TrainedTargetModel] = {}
+    evaluations: dict[str, TargetEvaluation] = {}
+    for route in default_model_routes():
+        evaluations[route.target_name] = _evaluate_route(dataset, route, groups)
+        preprocessor = FeaturePreprocessor().fit(dataset)
+        features = preprocessor.transform(dataset).values
+        estimator = build_estimator(route)
+        estimator.fit(features, _target_values(dataset, route))
+        models[route.target_name] = TrainedTargetModel(
+            route=route,
+            preprocessor=preprocessor,
+            estimator=estimator,
+        )
+    return TrainingBundle(models=models, evaluations=evaluations)
+
+
+def save_training_bundle(bundle: TrainingBundle, artifact_path: Path | str) -> Path:
+    """Persist models and preprocessing state as one versioned joblib artifact."""
+    path = Path(artifact_path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    joblib.dump(bundle, path)
+    return path
+
+
+def load_training_bundle(artifact_path: Path | str) -> TrainingBundle:
+    """Load an artifact created by :func:`save_training_bundle`."""
+    bundle = joblib.load(artifact_path)
+    if not isinstance(bundle, TrainingBundle):
+        raise ValueError(
+            f"artifact does not contain a training bundle: {artifact_path}"
+        )
+    return bundle

src/lmpm/training/preprocessing.py

@@ -71,7 +71,10 @@ class FeaturePreprocessor:
             steps=[
                 (
                     "impute",
-                    SimpleImputer(strategy=self.missing_value_strategy.numeric),
+                    SimpleImputer(
+                        strategy=self.missing_value_strategy.numeric,
+                        keep_empty_features=True,
+                    ),
                 ),
                 ("scale", StandardScaler()),
             ]
@@ -83,6 +86,7 @@ class FeaturePreprocessor:
                     SimpleImputer(
                         strategy=self.missing_value_strategy.categorical,
                         missing_values=None,
+                        keep_empty_features=True,
                     ),
                 ),
                 (

src/lmpm/training/workflow.py

@@ -0,0 +1,16 @@
+"""One-call workflow from a completed SQLite dataset to model artifacts."""
+
+from pathlib import Path
+
+from lmpm.training.dataset import load_sqlite_training_dataset
+from lmpm.training.models import TrainingBundle, save_training_bundle, train_models
+
+
+def train_sqlite_database(
+    database_path: Path | str, artifact_path: Path | str
+) -> TrainingBundle:
+    """Load completed experiments, validate them, train, and save one artifact."""
+    dataset = load_sqlite_training_dataset(database_path)
+    bundle = train_models(dataset)
+    save_training_bundle(bundle, artifact_path)
+    return bundle

tests/test_training_dataset.py

@@ -6,6 +6,7 @@ from lmpm.training.dataset import (
     TrainingDataError,
     load_sqlite_training_dataset,
 )
+from lmpm.training.workflow import train_sqlite_database
 
 MATERIAL_SCHEMA = """
 CREATE TABLE material_records (
@@ -150,3 +151,20 @@ def test_database_without_material_table_raises_expected_error(tmp_path):
 
     with pytest.raises(TrainingDataError, match="material_records"):
         load_sqlite_training_dataset(database_path)
+
+
+def test_sqlite_workflow_trains_and_saves_every_target_model(tmp_path):
+    database_path = create_database(tmp_path)
+    artifact_path = tmp_path / "lmpm.joblib"
+
+    bundle = train_sqlite_database(database_path, artifact_path)
+
+    assert artifact_path.exists()
+    assert set(bundle.models) == {
+        "is_cut_through",
+        "carbonized_edge_width",
+        "etching_depth",
+        "is_fire_smolder",
+        "pattern_clarity_score",
+        "presentation_balance_score",
+    }

tests/test_training_models.py

@@ -0,0 +1,98 @@
+import numpy as np
+import pytest
+from sklearn.ensemble import RandomForestClassifier
+from sklearn.gaussian_process import GaussianProcessRegressor
+
+from lmpm.training.dataset import (
+    CATEGORICAL_FEATURE_FIELDS,
+    NUMERIC_FEATURE_FIELDS,
+    TrainingDataset,
+    TrainingRecord,
+)
+from lmpm.training.models import (
+    build_estimator,
+    load_training_bundle,
+    save_training_bundle,
+    train_models,
+)
+from lmpm.training.routing import default_model_routes
+
+
+def make_record(number: int, material_name: str) -> TrainingRecord:
+    features = {
+        field: float(number + index + 1)
+        for index, field in enumerate(NUMERIC_FEATURE_FIELDS)
+    }
+    features.update(
+        {
+            CATEGORICAL_FEATURE_FIELDS[0]: "bidirectional",
+            CATEGORICAL_FEATURE_FIELDS[1]: "20x20",
+        }
+    )
+    return TrainingRecord(
+        metadata={
+            "experiment_id": f"LAS-2026-{number:04d}",
+            "material_name": material_name,
+        },
+        features=features,
+        targets={
+            "is_cut_through": number % 2 == 0,
+            "carbonized_edge_width": number / 10,
+            "etching_depth": number * 10.0,
+            "is_fire_smolder": False,
+            "pattern_clarity_score": float(number),
+            "presentation_balance_score": float(10 - number),
+        },
+    )
+
+
+def training_dataset() -> TrainingDataset:
+    return TrainingDataset(
+        records=(
+            make_record(1, "Acrylic"),
+            make_record(2, "Acrylic"),
+            make_record(3, "Basswood"),
+            make_record(4, "Basswood"),
+        )
+    )
+
+
+def test_model_factory_builds_the_routed_estimator_types():
+    routes = {route.target_name: route for route in default_model_routes()}
+
+    assert isinstance(
+        build_estimator(routes["etching_depth"]), GaussianProcessRegressor
+    )
+    assert isinstance(build_estimator(routes["is_cut_through"]), RandomForestClassifier)
+
+
+def test_training_runs_leakage_safe_lomo_and_persists_bundle(tmp_path):
+    bundle = train_models(training_dataset())
+
+    evaluation = bundle.evaluations["etching_depth"]
+    assert [fold.held_out_material for fold in evaluation.folds] == [
+        "Acrylic",
+        "Basswood",
+    ]
+    assert evaluation.mean_metrics["mae"] is not None
+    assert bundle.models["etching_depth"].route.model_family == (
+        "gaussian_process_regressor"
+    )
+
+    features = bundle.models["etching_depth"].preprocessor.transform(training_dataset())
+    predictions = bundle.models["etching_depth"].estimator.predict(features.values)
+    assert isinstance(predictions, np.ndarray)
+    assert predictions.shape == (4,)
+
+    artifact_path = save_training_bundle(bundle, tmp_path / "models.joblib")
+    restored = load_training_bundle(artifact_path)
+    assert restored.report() == bundle.report()
+
+
+def test_training_requires_two_distinct_materials():
+    dataset = TrainingDataset(
+        records=(make_record(1, "Acrylic"), make_record(2, "Acrylic"))
+    )
+
+    with pytest.raises(ValueError, match="at least two materials"):
+        train_models(dataset)

uv.lock

@@ -72,6 +72,7 @@ name = "lmpm"
 version = "0.1.0"
 source = { editable = "." }
 dependencies = [
+    { name = "joblib" },
     { name = "pandas" },
     { name = "pydantic" },
     { name = "pydantic-settings" },
@@ -87,6 +88,7 @@ dev = [
 
 [package.metadata]
 requires-dist = [
+    { name = "joblib", specifier = ">=1.4,<2" },
     { name = "pandas", specifier = ">=2.2,<3" },
     { name = "pydantic", specifier = ">=2.7,<3" },
     { name = "pydantic-settings", specifier = ">=2.2,<3" },