AI 工具链实战：MLflow 实验跟踪指南

进阶框架实现

实际项目中通常使用 TensorFlow 或 PyTorch。以下展示两种框架的对比结构，注意 BatchNormalization 和 Dropout 的配置差异。

# TensorFlow 示例
class TensorFlowModel:
    def __init__(self, input_dim: int, hidden_units: List[int] = [64, 32]):
        self.model = self._build_model(input_dim, hidden_units)

    def _build_model(self, input_dim: int, hidden_units: List[int]):
        from tensorflow import keras
        from tensorflow.keras import layers
        inputs = keras.Input(shape=(input_dim,))
        x = inputs
        for units in hidden_units:
            x = layers.Dense(units, activation='relu')(x)
            x = layers.BatchNormalization()(x)
            x = layers.Dropout(0.2)(x)
        outputs = layers.Dense(1)(x)
        model = keras.Model(inputs=inputs, outputs=outputs)
        model.compile(optimizer=keras.optimizers.Adam(learning_rate=0.001), loss='mse', metrics=['mae'])
        return model

# PyTorch 示例
class PyTorchModel(torch.nn.Module):
    def __init__(self, input_dim: int, hidden_units: List[int] = [64, 32]):
        super(PyTorchModel, self).__init__()
        layers_list = []
        prev_units = input_dim
        for units in hidden_units:
            layers_list.append(torch.nn.Linear(prev_units, units))
            layers_list.append(torch.nn.ReLU())
            layers_list.append(torch.nn.BatchNorm1d(units))
            layers_list.append(torch.nn.Dropout(0.2))
            prev_units = units
        layers_list.append(torch.nn.Linear(prev_units, 1))
        self.network = torch.nn.Sequential(*layers_list)

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        return self.network(x)

数据处理与评估

完整处理流程

数据质量直接决定模型上限。标准化的预处理流程包括缺失值填充、类别编码及特征缩放。

import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler, LabelEncoder
from sklearn.impute import SimpleImputer
from typing import List, Tuple

class DataProcessor:
    def __init__(self):
        self.scaler = StandardScaler()
        self.label_encoders = {}
        self.imputer = SimpleImputer(strategy='mean')

    def process(self, data: pd.DataFrame, target_col: str, categorical_cols: List[str] = None, test_size: float = 0.2) -> Tuple:
        X = data.drop(columns=[target_col])
        y = data[target_col]
        # 数值型缺失值处理
        numeric_cols = X.select_dtypes(include=[np.number]).columns
        X[numeric_cols] = self.imputer.fit_transform(X[numeric_cols])
        # 类别特征编码
        if categorical_cols:
            for col in categorical_cols:
                if col in X.columns:
                    le = LabelEncoder()
                    X[col] = le.fit_transform(X[col].astype(str))
                    self.label_encoders[col] = le
        # 标准化
        X_scaled = self.scaler.fit_transform(X)
        # 划分数据集
        X_train, X_test, y_train, y_test = train_test_split(
            X_scaled, y, test_size=test_size, random_state=42)
        return X_train, X_test, y_train, y_test

模型评估体系

选择合适的评估指标至关重要。分类任务关注准确率与 F1 分数，回归任务则侧重 MSE 与 R²。

from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score, roc_auc_score, mean_squared_error, r2_score
import matplotlib.pyplot as plt
import seaborn as sns

class ModelEvaluator:
    @staticmethod
    def evaluate_classification(y_true, y_pred, y_prob=None):
        metrics = {
            'accuracy': accuracy_score(y_true, y_pred),
            'precision': precision_score(y_true, y_pred, average='weighted'),
            'recall': recall_score(y_true, y_pred, average='weighted'),
            'f1': f1_score(y_true, y_pred, average='weighted')
        }
        if y_prob is not None:
            metrics['roc_auc'] = roc_auc_score(y_true, y_prob, multi_class='ovr')
        return metrics

    @staticmethod
    def evaluate_regression(y_true, y_pred):
        return {
            'mse': mean_squared_error(y_true, y_pred),
            'rmse': np.sqrt(mean_squared_error(y_true, y_pred)),
            'mae': np.mean(np.abs(y_true - y_pred)),
            'r2': r2_score(y_true, y_pred)
        }

实战案例与最佳实践

房价预测案例

通过 Pipeline 整合预处理器与模型，可以有效避免数据泄露并简化部署流程。

from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.ensemble import GradientBoostingRegressor
from sklearn.metrics import mean_absolute_error

class HousePricePredictor:
    def __init__(self):
        self.model = None
        self.preprocessor = None

    def prepare_data(self, data: pd.DataFrame, target_col: str):
        X = data.drop(columns=[target_col])
        y = data[target_col]
        numeric_features = X.select_dtypes(include=[np.number]).columns.tolist()
        categorical_features = X.select_dtypes(exclude=[np.number]).columns.tolist()
        self.preprocessor = ColumnTransformer(
            transformers=[
                ('num', StandardScaler(), numeric_features),
                ('cat', OneHotEncoder(handle_unknown='ignore'), categorical_features)
            ])
        return train_test_split(X, y, test_size=0.2, random_state=42)

    def train(self, X_train, y_train):
        self.model = Pipeline([
            ('preprocessor', self.preprocessor),
            ('regressor', GradientBoostingRegressor(n_estimators=200, learning_rate=0.1, max_depth=5, random_state=42))
        ])
        self.model.fit(X_train, y_train)
        return self

    def evaluate(self, X_test, y_test):
        y_pred = self.model.predict(X_test)
        metrics = {
            'RMSE': np.sqrt(mean_squared_error(y_test, y_pred)),
            'MAE': mean_absolute_error(y_test, y_pred),
            'R2': r2_score(y_test, y_pred)
        }
        return metrics, y_pred

实施建议

1. 环境管理：使用虚拟环境隔离依赖，记录 requirements.txt。
2. 项目结构：遵循标准目录规范（data, src, tests, configs），便于维护。
3. 代码规范：添加类型注解与文档字符串，遵循 PEP8 规范。
4. 实验管理：利用版本控制记录参数变化，保存模型检查点。

常见问题

• 模型选择：小样本用传统机器学习，中等样本用集成学习，大样本考虑深度学习。
• 数据不平衡：可使用 SMOTE 过采样、欠采样或调整类别权重解决。
• 性能提升：重点投入在数据增强、特征工程及超参数调优上。
• 常见错误：警惕数据泄露，确保评估方法正确，保证代码可复现。

未来 AI 发展将聚焦于 AutoML、大模型微调及多模态融合。对于开发者而言，夯实 Python 基础，积累项目实战经验，并保持对新技术的敏感度，是职业成长的关键路径。