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

进阶框架实现

实际项目中，我们更多使用 TensorFlow 或 PyTorch。以下是两个框架的对比实现思路。

TensorFlow 实现

import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers

class TensorFlowModel:
    """TensorFlow 版本的模型实现"""
    def __init__(self, input_dim: int, hidden_units: List[int] = [64, 32]):
        """初始化 TensorFlow 模型
        
        Args:
            input_dim: 输入维度
            hidden_units: 隐藏层单元数列表
        """
        self.model = self._build_model(input_dim, hidden_units)

    def _build_model(self, input_dim: int, hidden_units: List[int]) -> keras.Model:
        """构建模型架构"""
        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

    def train(self, X_train, y_train, X_val, y_val, epochs=100, batch_size=32):
        """训练模型"""
        history = self.model.fit(
            X_train, y_train,
            validation_data=(X_val, y_val),
            epochs=epochs,
            batch_size=batch_size,
            verbose=1)
        return history

    def predict(self, X):
        """预测"""
        return self.model.predict(X)

PyTorch 实现

import torch
import torch.nn as nn
import torch.optim as optim

class PyTorchModel(nn.Module):
    """PyTorch 版本的模型实现"""
    def __init__(self, input_dim: int, hidden_units: List[int] = [64, 32]):
        """初始化 PyTorch 模型
        
        Args:
            input_dim: 输入维度
            hidden_units: 隐藏层单元数列表
        """
        super(PyTorchModel, self).__init__()
        layers_list = []
        prev_units = input_dim
        for units in hidden_units:
            layers_list.append(nn.Linear(prev_units, units))
            layers_list.append(nn.ReLU())
            layers_list.append(nn.BatchNorm1d(units))
            layers_list.append(nn.Dropout(0.2))
            prev_units = units
        layers_list.append(nn.Linear(prev_units, 1))
        self.network = nn.Sequential(*layers_list)

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """前向传播"""
        return self.network(x)

    def train_model(self, train_loader, val_loader, epochs=100, lr=0.001):
        """训练模型"""
        criterion = nn.MSELoss()
        optimizer = optim.Adam(self.parameters(), lr=lr)
        train_losses = []
        val_losses = []
        
        for epoch in range(epochs):
            # 训练阶段
            self.train()
            train_loss = 0.0
            for X_batch, y_batch in train_loader:
                optimizer.zero_grad()
                outputs = self(X_batch)
                loss = criterion(outputs, y_batch)
                loss.backward()
                optimizer.step()
                train_loss += loss.item()

            # 验证阶段
            self.eval()
            val_loss = 0.0
            with torch.no_grad():
                for X_batch, y_batch in val_loader:
                    outputs = self(X_batch)
                    loss = criterion(outputs, y_batch)
                    val_loss += loss.item()
            
            train_losses.append(train_loss / len(train_loader))
            val_losses.append(val_loss / len(val_loader))
            
            if (epoch + 1) % 10 == 0:
                print(f"Epoch {epoch+1}/{epochs}, "
                      f"Train Loss: {train_losses[-1]:.4f}, "
                      f"Val Loss: {val_losses[-1]:.4f}")
        return train_losses, val_losses

数据处理与评估

数据处理流程

规范的数据处理是模型成功的前提。以下是一个通用的预处理类，涵盖缺失值填充、类别编码和标准化。

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:
        """完整的数据处理流程
        
        Args:
            data: 原始数据
            target_col: 目标列名
            categorical_cols: 类别列名列表
            test_size: 测试集比例
        Returns:
            处理后的训练集和测试集
        """
        # 1. 分离特征和目标
        X = data.drop(columns=[target_col])
        y = data[target_col]

        # 2. 处理缺失值
        X = pd.DataFrame(
            self.imputer.fit_transform(X.select_dtypes(include=[np.number])),
            columns=X.select_dtypes(include=[np.number]).columns
        )

        # 3. 编码类别特征
        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

        # 4. 标准化
        X_scaled = self.scaler.fit_transform(X)

        # 5. 划分数据集
        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

# 使用示例
if __name__ == "__main__":
    data = pd.DataFrame({
        'feature1': np.random.randn(1000),
        'feature2': np.random.randn(1000),
        'feature3': np.random.choice(['A', 'B', 'C'], 1000),
        'target': np.random.randn(1000)
    })
    processor = DataProcessor()
    X_train, X_test, y_train, y_test = processor.process(
        data, target_col='target', categorical_cols=['feature3'])
    print(f"训练集形状：{X_train.shape}")
    print(f"测试集形状：{X_test.shape}")

模型评估方法

选择合适的评估指标至关重要。分类问题看准确率、F1 分数，回归问题看 RMSE、R²。

from sklearn.metrics import (
    accuracy_score, precision_score, recall_score,
    f1_score, roc_auc_score, confusion_matrix,
    mean_squared_error, mean_absolute_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': mean_absolute_error(y_true, y_pred),
            'r2': r2_score(y_true, y_pred)
        }

    @staticmethod
    def plot_confusion_matrix(y_true, y_pred, labels=None):
        """绘制混淆矩阵"""
        cm = confusion_matrix(y_true, y_pred)
        plt.figure(figsize=(8, 6))
        sns.heatmap(cm, annot=True, fmt='d', cmap='Blues',
                    xticklabels=labels, yticklabels=labels)
        plt.title('混淆矩阵')
        plt.xlabel('预测值')
        plt.ylabel('真实值')
        plt.show()

    @staticmethod
    def plot_learning_curve(train_losses, val_losses):
        """绘制学习曲线"""
        plt.figure(figsize=(10, 6))
        plt.plot(train_losses, label='训练损失')
        plt.plot(val_losses, label='验证损失')
        plt.xlabel('Epoch')
        plt.ylabel('Loss')
        plt.title('学习曲线')
        plt.legend()
        plt.grid(True)
        plt.show()

# 使用示例
if __name__ == "__main__":
    # 分类评估示例
    y_true_cls = [0, 1, 0, 1, 0, 1, 0, 0, 1, 1]
    y_pred_cls = [0, 1, 0, 0, 0, 1, 1, 0, 1, 1]
    cls_metrics = ModelEvaluator.evaluate_classification(y_true_cls, y_pred_cls)
    print("分类指标:", cls_metrics)

    # 回归评估示例
    y_true_reg = np.array([1.0, 2.0, 3.0, 4.0, 5.0])
    y_pred_reg = np.array([1.1, 1.9, 3.2, 3.8, 5.1])
    reg_metrics = ModelEvaluator.evaluate_regression(y_true_reg, y_pred_reg)
    print("回归指标:", reg_metrics)

最佳实践与常见问题

实施步骤建议

1. 环境准备：建议使用虚拟环境隔离依赖。

   conda create -n ai_env python=3.9
   conda activate ai_env
   pip install numpy pandas scikit-learn tensorflow torch jupyter
   

2. 项目结构：保持目录清晰，便于维护。

   project/
   ├── data/          # 数据目录
   ├── notebooks/     # 探索性分析
   ├── src/           # 源代码
   │   ├── data/      # 数据处理
   │   ├── models/    # 模型定义
   │   └── utils/     # 工具函数
   ├── tests/         # 测试代码
   ├── configs/       # 配置文件
   └── requirements.txt # 依赖列表
   

3. 代码规范：遵循 PEP8，添加类型注解和文档字符串，编写单元测试。

常见问题解答

Q1：如何选择合适的模型？

• 小样本：传统机器学习（不易过拟合）
• 中等样本：集成学习（性能稳定）
• 大样本：深度学习（潜力更大）

Q2：如何处理数据不平衡？ 可以使用 SMOTE 进行过采样，或使用 RandomUnderSampler 进行欠采样，也可以调整类别权重。

from imblearn.over_sampling import SMOTE
smote = SMOTE(random_state=42)
X_resampled, y_resampled = smote.fit_resample(X, y)

Q3：如何避免常见错误？

• 防止数据泄露（先划分再拟合）
• 正确选择评估方法
• 合理设置超参数
• 确保代码可复现

总结

本文通过 Python 代码示例，展示了从数据处理、模型构建到评估的完整流程。重点介绍了如何规范代码结构、处理数据不平衡问题以及避免过拟合等常见陷阱。在实际工作中，结合 MLflow 等工具管理实验，能显著提升团队协作效率和模型迭代质量。