别再手动调参了！用Made-With-ML打造生产级ML系统的完整实战指南
在机器学习的实际应用中，我们经常会遇到这样的困境：明明在Jupyter Notebook里效果很好的模型，一到生产环境就问题百出——推理速度慢、内存占用高、部署流程混乱、版本管理缺失。这些问题折磨着无数ML工程师，也让许多项目难产于“从实验到生产”的最后一公里。
今天要介绍的这个GitHub项目——Made-With-ML，或许能为你提供一个完整的解决方案。这个项目由资深ML工程师Goku Mohandas创建，收到了超过2.3万颗星标，被广泛应用于ML教学和工业实践中。它不是简单的代码仓库，而是一套完整的生产级ML系统构建方法论，涵盖了从数据处理、模型训练到服务部署的全流程。
本文将带你深入了解这个项目，手把手教你如何利用Made-With-ML的思路和代码，构建属于自己的生产级ML系统。无论你是刚入门的新手，还是有一定经验的开发者，都能从中获得实用的知识和经验。
为什么值得关注 / 为什么值得关注
在深入了解具体内容之前，我们先来探讨一个根本问题：Made-With-ML与其他ML学习资源有什么不同？为什么它值得你投入时间去学习？
首先，让我们回顾一下学习机器学习的常见路径。大多数教程和课程都会教你如何训练一个模型——给定数据集，使用某种算法，输出一个准确率。这种学习方式固然重要，但它忽略了一个关键事实：在真实的工业场景中，模型训练只是整个系统的很小一部分。
根据一些行业调研数据，一个完整的ML系统，模型训练代码可能只占整个项目的5%到10%。剩余的时间都花在了数据收集与清洗、特征工程、模型验证、服务部署、监控维护等工作上。这就是所谓的ML工程化问题，也是许多团队在推进ML项目时面临的最大挑战。
Made-With-ML的出现正是为了解决这个痛点。它不是教你如何设计一个新的神经网络架构，也不是教你某个特定算法的数学原理，而是教你如何将这些“学术层面”的模型转化为可靠、可维护、可扩展的生产系统。这个项目涵盖的内容非常全面，包括代码结构设计、实验追踪与版本控制、模型序列化与服务化、容器化部署、API开发、持续集成与持续部署（CI/CD）等等。
项目的另一个亮点是它的实用性。所有的代码和概念都基于真实的生产环境经验，而非纸上谈兵。项目的作者曾在Google Brain等顶级ML团队工作过，他将实际工作中遇到的挑战和解决方案融入到这个项目中。你学到的不是理想化的玩具项目代码，而是经过生产环境验证的最佳实践。
此外，Made-With-ML的代码质量非常高，注释详尽，结构清晰。对于想要提升代码组织能力和工程素养的开发者来说，阅读这个项目的源码本身就是一种很好的学习方式。它遵循Python社区的最佳实践，使用类型提示（type hints）进行类型检查，使用现代的依赖管理工具，采用清晰的模块划分，这些都是值得学习的点。
项目的社区支持也相当活跃。它不仅是一个代码仓库，更是一个持续更新的学习资源库。作者会定期根据社区反馈和行业发展更新内容，确保你学到的是最新的技术和方法。
环境搭建 / Getting Started
了解了项目的基本情况后，让我们开始动手实践。第一步是搭建开发环境，这是所有后续工作的基础。
在开始之前，请确保你的电脑上已经安装了Python（建议使用Python 3.8或更高版本）和Git。如果还没有安装，可以从Python官网和Git官网下载安装包。安装完成后，我们就可以开始克隆项目并进行环境配置了。
首先，打开终端（命令行界面），执行以下命令克隆项目仓库：
git clone https://github.com/GokuMohandas/Made-With-ML.git
cd Made-With-ML
克隆完成后，你会看到一个名为Made-With-ML的文件夹。里面的目录结构如下：
Made-With-ML/
├── .github/
│ └── workflows/ # CI/CD配置文件
├── notebooks/ # Jupyter笔记本
├── project/ # 核心项目代码
│ ├── __init__.py
│ ├── config.py # 配置文件
│ ├── predict.py # 预测逻辑
│ ├── train.py # 训练逻辑
│ └── utils.py # 工具函数
├── tests/ # 测试代码
├── Dockerfile # Docker配置
├── docker-compose.yml # Docker编排文件
├── Makefile # 便捷命令
├── requirements.txt # 依赖列表
└── README.md # 项目说明
接下来，创建并激活一个虚拟环境。虚拟环境可以帮助你隔离项目依赖，避免不同项目之间的包版本冲突。在Python中，我们可以使用venv模块来创建虚拟环境：
# 创建虚拟环境
python -m venv venv
# 在Linux或macOS上激活虚拟环境
source venv/bin/activate
# 在Windows上激活虚拟环境
# venv\Scripts\activate
激活虚拟环境后，你会看到终端提示符前面多了一个(venv)的标识，表示你现在处于虚拟环境中。现在安装项目依赖：
pip install -r requirements.txt
requirements.txt文件中包含了项目所需的所有Python包，包括PyTorch、FastAPI、 scikit-learn等核心库。安装过程可能需要几分钟时间，取决于你的网络速度和电脑性能。
安装完成后，为了验证环境配置是否正确，我们可以运行项目提供的一个简单测试：
python -c "import torch; import fastapi; print('环境配置成功！')"
如果终端输出了”环境配置成功！”，说明所有依赖都已正确安装。
项目还提供了Docker支持，如果你更习惯使用Docker来管理环境，可以直接使用项目提供的Dockerfile构建镜像：
docker build -t made-with-ml .
docker run -it --rm -p 8000:8000 made-with-ml
使用Docker的好处是环境配置更加标准化，不受宿主机环境影响，而且可以确保开发环境和生产环境的一致性。
Makefile文件中还预定义了一些常用命令，比如：
make install # 安装依赖
make train # 运行训练
make test # 运行测试
make predict # 运行预测
make docker # 构建Docker镜像
这些命令封装了常用的操作步骤，让你不需要记住复杂的命令参数。这是一个很好的实践，在实际项目中我们也建议使用Makefile来管理常用命令。
核心功能详解 / Core Features with Detailed Explanations
Made-With-ML项目的核心设计理念是将ML系统划分为几个相对独立的模块，每个模块负责特定的功能。这种模块化设计不仅提高了代码的可维护性，也使得系统更加灵活，便于后续扩展和优化。
配置管理系统
在大型项目中，配置管理是一个容易被忽视但却至关重要的环节。硬编码的配置不仅难以修改，还容易导致代码混乱。Made-With-ML采用集中化的配置管理，所有可配置的参数都集中在一个地方。
项目的config.py文件定义了训练和预测所需的各种配置参数：
# 项目中的配置管理示例
class Config:
"""训练配置类"""
# 数据配置
data_dir = "data"
batch_size = 32
num_workers = 4
# 模型配置
model_name = "bert-base-uncased"
hidden_size = 768
num_layers = 12
# 训练配置
learning_rate = 2e-5
epochs = 10
warmup_steps = 100
# 其他配置
seed = 42
device = "cuda"
这种配置类的设计使得配置参数一目了然，也便于在不同环境（开发、测试、生产）之间切换。你可以通过继承和重写来创建不同的配置类，而不需要修改核心代码。
数据处理流水线
数据处理是ML系统中最耗时也是最关键的环节之一。Made-With-ML提供了一套完整的数据处理流水线，包括数据加载、预处理、批处理等功能。
# 数据集类的典型实现
class TextDataset(Dataset):
"""文本数据集类"""
def __init__(self, texts, labels, tokenizer, max_length):
self.texts = texts
self.labels = labels
self.tokenizer = tokenizer
self.max_length = max_length
def __len__(self):
return len(self.texts)
def __getitem__(self, idx):
text = self.texts[idx]
label = self.labels[idx]
# 对文本进行tokenize
encoding = self.tokenizer(
text,
max_length=self.max_length,
padding="max_length",
truncation=True,
return_tensors="pt"
)
return {
"input_ids": encoding["input_ids"].squeeze(),
"attention_mask": encoding["attention_mask"].squeeze(),
"label": torch.tensor(label, dtype=torch.long)
}
这个数据集类遵循了PyTorch Dataset的标准接口，可以与PyTorch的DataLoader无缝配合。通过这种设计，数据处理的各个步骤被清晰地分离，便于调试和优化。
数据加载器负责批量读取数据并支持多进程加载：
# 创建数据加载器
def create_data_loaders(train_dataset, val_dataset, test_dataset, batch_size, num_workers):
"""创建训练、验证和测试数据加载器"""
train_loader = DataLoader(
train_dataset,
batch_size=batch_size,
shuffle=True,
num_workers=num_workers,
pin_memory=True # 加速CPU到GPU的数据传输
)
val_loader = DataLoader(
val_dataset,
batch_size=batch_size,
shuffle=False,
num_workers=num_workers
)
test_loader = DataLoader(
test_dataset,
batch_size=batch_size,
shuffle=False,
num_workers=num_workers
)
return train_loader, val_loader, test_loader
模型定义
模型定义模块负责封装神经网络的具体结构。良好的模型设计应该兼顾性能和可维护性：
import torch.nn as nn
class TextClassifier(nn.Module):
"""文本分类模型"""
def __init__(self, num_classes, vocab_size, embedding_dim, hidden_dim, num_layers, dropout):
super(TextClassifier, self).__init__()
# 词嵌入层
self.embedding = nn.Embedding(vocab_size, embedding_dim, padding_idx=0)
# 循环层（可以是LSTM或GRU）
self.lstm = nn.LSTM(
embedding_dim,
hidden_dim,
num_layers=num_layers,
batch_first=True,
dropout=dropout if num_layers > 1 else 0
)
# 全连接层
self.fc = nn.Linear(hidden_dim, num_classes)
# Dropout层
self.dropout = nn.Dropout(dropout)
def forward(self, input_ids, attention_mask):
# 词嵌入
embedded = self.embedding(input_ids)
# 处理可变长度序列（使用attention_mask）
lengths = attention_mask.sum(dim=1).cpu()
packed = nn.utils.rnn.pack_padded_sequence(
embedded, lengths, batch_first=True, enforce_sorted=False
)
# LSTM前向传播
packed_output, (hidden, cell) = self.lstm(packed)
# 取最后一层的隐藏状态
last_hidden = hidden[-1, :, :]
# Dropout和全连接
output = self.dropout(last_hidden)
logits = self.fc(output)
return logits
这个模型使用了LSTM进行序列建模，适用于文本分类任务。代码中展示了几个重要的设计模式：使用pack_padded_sequence处理可变长度序列、使用dropout防止过拟合、使用最后一层的隐藏状态作为句子表示。
训练循环
训练模块是ML系统的核心，它负责执行模型的学习过程。Made-With-ML的训练代码遵循了最佳实践，包括梯度清零、学习率调度、早停机制等：
def train_epoch(model, data_loader, optimizer, criterion, device, scheduler=None):
"""训练一个epoch"""
model.train()
total_loss = 0
correct = 0
total = 0
for batch in data_loader:
# 将数据移到设备上
input_ids = batch["input_ids"].to(device)
attention_mask = batch["attention_mask"].to(device)
labels = batch["label"].to(device)
# 梯度清零
optimizer.zero_grad()
# 前向传播
outputs = model(input_ids, attention_mask)
loss = criterion(outputs, labels)
# 反向传播
loss.backward()
# 梯度裁剪，防止梯度爆炸
torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
# 更新参数
optimizer.step()
# 如果有学习率调度器，更新学习率
if scheduler is not None:
scheduler.step()
# 统计损失和准确率
total_loss += loss.item()
_, predicted = outputs.max(1)
total += labels.size(0)
correct += predicted.eq(labels).sum().item()
avg_loss = total_loss / len(data_loader)
accuracy = correct / total
return avg_loss, accuracy
def evaluate(model, data_loader, criterion, device):
"""评估模型"""
model.eval()
total_loss = 0
correct = 0
total = 0
with torch.no_grad():
for batch in data_loader:
input_ids = batch["input_ids"].to(device)
attention_mask = batch["attention_mask"].to(device)
labels = batch["label"].to(device)
outputs = model(input_ids, attention_mask)
loss = criterion(outputs, labels)
total_loss += loss.item()
_, predicted = outputs.max(1)
total += labels.size(0)
correct += predicted.eq(labels).sum().item()
avg_loss = total_loss / len(data_loader)
accuracy = correct / total
return avg_loss, accuracy
训练函数中包含了几个关键的优化技巧。梯度裁剪（gradient clipping）可以防止RNN类模型训练时的梯度爆炸问题；学习率调度可以根据训练进度动态调整学习率；dropout只在训练时生效，评估时会自动关闭。
预测与服务
模型训练完成后，需要能够对外提供预测服务。Made-With-ML使用FastAPI构建RESTful API，这使得模型可以轻松地与前端应用或其他服务集成：
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import torch
app = FastAPI(title="ML Prediction API")
# 加载模型（单例模式）
model = None
class PredictionRequest(BaseModel):
"""预测请求模型"""
text: str
max_length: int = 128
class PredictionResponse(BaseModel):
"""预测响应模型"""
label: str
confidence: float
probabilities: dict
@app.on_event("startup")
async def load_model():
"""应用启动时加载模型"""
global model
model = TextClassifier(num_classes=10)
model.load_state_dict(torch.load("models/best_model.pt"))
model.eval()
@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
"""执行预测"""
if model is None:
raise HTTPException(status_code=503, detail="模型未加载")
# Tokenize输入
encoding = tokenizer(
request.text,
max_length=request.max_length,
padding="max_length",
truncation=True,
return_tensors="pt"
)
# 推理
with torch.no_grad():
outputs = model(
encoding["input_ids"],
encoding["attention_mask"]
)
probs = torch.softmax(outputs, dim=1)
pred_class = probs.argmax().item()
confidence = probs[0][pred_class].item()
return PredictionResponse(
label=label_map[pred_class],
confidence=confidence,
probabilities={label_map[i]: prob.item() for i, prob in enumerate(probs[0])}
)
FastAPI会自动生成API文档，你可以通过访问/docs来查看和测试API。这种设计使得模型部署变得非常简单，只需要几行代码就能将训练好的模型变成可用的Web服务。
实战教程 / Step-by-Step Practical Tutorial
理论部分已经讲得足够多了，现在让我们通过一个完整的实战项目来巩固所学知识。我们将构建一个文本分类系统，实现从数据准备到服务部署的完整流程。
第一步：准备数据集
为了使教程具有实际可操作性，我们使用一个真实的数据集。这里以新闻分类任务为例，数据集可以从公开的数据源获取：
# data_loader.py
import os
import json
import pandas as pd
from pathlib import Path
def load_data(data_path):
"""加载并解析数据集"""
data_path = Path(data_path)
texts = []
labels = []
# 遍历数据目录
for label_dir in data_path.iterdir():
if label_dir.is_dir():
label_name = label_dir.name
# 遍历该类别下的所有文件
for file_path in label_dir.glob("*.txt"):
with open(file_path, "r", encoding="utf-8") as f:
text = f.read().strip()
if text:
texts.append(text)
labels.append(label_name)
return texts, labels
def create_label_mapping(labels):
"""创建标签到索引的映射"""
unique_labels = sorted(set(labels))
label_to_idx = {label: idx for idx, label in enumerate(unique_labels)}
idx_to_label = {idx: label for label, idx in label_to_idx.items()}
return label_to_idx, idx_to_label
def split_data(texts, labels, train_ratio=0.8, val_ratio=0.1, seed=42):
"""划分训练集、验证集和测试集"""
import numpy as np
np.random.seed(seed)
indices = np.arange(len(texts))
np.random.shuffle(indices)
n_train = int(len(texts) * train_ratio)
n_val = int(len(texts) * val_ratio)
train_indices = indices[:n_train]
val_indices = indices[n_train:n_train + n_val]
test_indices = indices[n_train + n_val:]
train_texts = [texts[i] for i in train_indices]
train_labels = [labels[i] for i in train_indices]
val_texts = [texts[i] for i in val_indices]
val_labels = [labels[i] for i in val_indices]
test_texts = [texts[i] for i in test_indices]
test_labels = [labels[i] for i in test_indices]
return (train_texts, train_labels), (val_texts, val_labels), (test_texts, test_labels)
# 使用示例
if __name__ == "__main__":
texts, labels = load_data("data/news")
label_to_idx, idx_to_label = create_label_mapping(labels)
print(f"总共加载了 {len(texts)} 条数据")
print(f"类别数量: {len(label_to_idx)}")
print(f"类别映射: {label_to_idx}")
# 保存标签映射
with open("data/label_mapping.json", "w") as f:
json.dump({"label_to_idx": label_to_idx, "idx_to_label": idx_to_label}, f, indent=2)
第二步：实现完整训练流程
现在我们将各个模块组合起来，实现一个完整的训练脚本：
# train.py
import os
import json
import argparse
import numpy as np
import pandas as pd
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
from transformers import AutoTokenizer, AutoModel, get_linear_schedule_with_warmup
from sklearn.metrics import classification_report, confusion_matrix
# ============================================
# 配置参数
# ============================================
class Config:
# 数据配置
data_dir = "data"
max_length = 256
batch_size = 16
# 模型配置
model_name = "bert-base-chinese" # 使用中文预训练模型
num_classes = 10
learning_rate = 2e-5
weight_decay = 0.01
# 训练配置
epochs = 5
warmup_ratio = 0.1
grad_clip = 1.0
# 其他配置
seed = 42
device = "cuda" if torch.cuda.is_available() else "cpu"
save_dir = "models"
# ============================================
# 数据集类
# ============================================
class NewsDataset(Dataset):
"""新闻分类数据集"""
def __init__(self, texts, labels, tokenizer, label_to_idx, max_length):
self.texts = texts
self.labels = [label_to_idx[l] for l in labels]
self.tokenizer = tokenizer
self.max_length = max_length
def __len__(self):
return len(self.texts)
def __getitem__(self, idx):
text = self.texts[idx]
label = self.labels[idx]
encoding = self.tokenizer(
text,
max_length=self.max_length,
padding="max_length",
truncation=True,
return_tensors="pt"
)
return {
"input_ids": encoding["input_ids"].squeeze(0),
"attention_mask": encoding["attention_mask"].squeeze(0),
"label": torch.tensor(label, dtype=torch.long)
}
# ============================================
# 模型定义
# ============================================
class BertClassifier(nn.Module):
"""基于BERT的分类模型"""
def __init__(self, model_name, num_classes, dropout=0.3):
super(BertClassifier, self).__init__()
self.bert = AutoModel.from_pretrained(model_name)
self.dropout = nn.Dropout(dropout)
self.classifier = nn.Linear(self.bert.config.hidden_size, num_classes)
def forward(self, input_ids, attention_mask):
outputs = self.bert(input_ids=input_ids, attention_mask=attention_mask)
pooled_output = outputs.pooler_output # 使用[CLS] token的表示
pooled_output = self.dropout(pooled_output)
logits = self.classifier(pooled_output)
return logits
# ============================================
# 训练函数
# ============================================
def train_epoch(model, data_loader, optimizer, scheduler, criterion, device, grad_clip=None):
"""训练一个epoch"""
model.train()
total_loss = 0
predictions = []
actual_labels = []
for batch_idx, batch in enumerate(data_loader):
input_ids = batch["input_ids"].to(device)
attention_mask = batch["attention_mask"].to(device)
labels = batch["label"].to(device)
# 前向传播
optimizer.zero_grad()
outputs = model(input_ids, attention_mask)
loss = criterion(outputs, labels)
# 反向传播
loss.backward()
# 梯度裁剪
if grad_clip:
nn.utils.clip_grad_norm_(model.parameters(), grad_clip)
optimizer.step()
scheduler.step()
total_loss += loss.item()
predictions.extend(outputs.argmax(dim=1).cpu().numpy())
actual_labels.extend(labels.cpu().numpy())
# 每100个batch打印一次进度
if (batch_idx + 1) % 100 == 0:
print(f" Batch {batch_idx + 1}/{len(data_loader)}, Loss: {loss.item():.4f}")
avg_loss = total_loss / len(data_loader)
accuracy = np.mean(np.array(predictions) == np.array(actual_labels))
return avg_loss, accuracy
def evaluate(model, data_loader, criterion, device):
"""评估模型"""
model.eval()
total_loss = 0
predictions = []
actual_labels = []
with torch.no_grad():
for batch in data_loader:
input_ids = batch["input_ids"].to(device)
attention_mask = batch["attention_mask"].to(device)
labels = batch["label"].to(device)
outputs = model(input_ids, attention_mask)
loss = criterion(outputs, labels)
total_loss += loss.item()
predictions.extend(outputs.argmax(dim=1).cpu().numpy())
actual_labels.extend(labels.cpu().numpy())
avg_loss = total_loss / len(data_loader)
accuracy = np.mean(np.array(predictions) == np.array(actual_labels))
return avg_loss, accuracy, predictions, actual_labels
# ============================================
# 主训练流程
# ============================================
def main(args):
# 设置随机种子
torch.manual_seed(Config.seed)
np.random.seed(Config.seed)
if torch.cuda.is_available():
torch.cuda.manual_seed_all(Config.seed)
device = torch.device(Config.device)
print(f"使用设备: {device}")
# 加载tokenizer
print("加载tokenizer...")
tokenizer = AutoTokenizer.from_pretrained(Config.model_name)
# 加载标签映射
with open(os.path.join(Config.data_dir, "label_mapping.json"), "r") as f:
mapping = json.load(f)
label_to_idx = mapping["label_to_idx"]
idx_to_label = {int(k): v for k, v in mapping["idx_to_label"].items()}
# 加载数据
print("加载数据...")
# 这里假设数据已经按目录组织好了
# train_texts, train_labels = ...
# val_texts, val_labels = ...
# test_texts, test_labels = ...
# 创建数据集和数据加载器
train_dataset = NewsDataset(train_texts, train_labels, tokenizer, label_to_idx, Config.max_length)
val_dataset = NewsDataset(val_texts, val_labels, tokenizer, label_to_idx, Config.max_length)
test_dataset = NewsDataset(test_texts, test_labels, tokenizer, label_to_idx, Config.max_length)
train_loader = DataLoader(train_dataset, batch_size=Config.batch_size, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=Config.batch_size)
test_loader = DataLoader(test_dataset, batch_size=Config.batch_size)
# 创建模型
print("创建模型...")
model = BertClassifier(Config.model_name, Config.num_classes)
model = model.to(device)
# 设置优化器
optimizer = torch.optim.AdamW(
model.parameters(),
lr=Config.learning_rate,
weight_decay=Config.weight_decay
)
# 学习率调度器
total_steps = len(train_loader) * Config.epochs
warmup_steps = int(total_steps * Config.warmup_ratio)
scheduler = get_linear_schedule_with_warmup(
optimizer, num_warmup_steps=warmup_steps, num_training_steps=total_steps
)
# 损失函数
criterion = nn.CrossEntropyLoss()
# 训练循环
best_val_acc = 0
for epoch in range(Config.epochs):
print(f"\nEpoch {epoch + 1}/{Config.epochs}")
print("-" * 40)
# 训练
train_loss, train_acc = train_epoch(
model, train_loader, optimizer, scheduler, criterion, device, Config.grad_clip
)
print(f"训练损失: {train_loss:.4f}, 训练准确率: {train_acc:.4f}")
# 验证
val_loss, val_acc, _, _ = evaluate(model, val_loader, criterion, device)
print(f"验证损失: {val_loss:.4f}, 验证准确率: {val_acc:.4f}")
# 保存最佳模型
if val_acc > best_val_acc:
best_val_acc = val_acc
os.makedirs(Config.save_dir, exist_ok=True)
torch.save(model.state_dict(), os.path.join(Config.save_dir, "best_model.pt"))
print(f"保存最佳模型，准确率: {val_acc:.4f}")
# 测试集评估
print("\n" + "=" * 40)
print("在测试集上评估最终模型:")
model.load_state_dict(torch.load(os.path.join(Config.save_dir, "best_model.pt")))
test_loss, test_acc, predictions, actuals = evaluate(model, test_loader, criterion, device)
print(f"测试损失: {test_loss:.4f}, 测试准确率: {test_acc:.4f}")
# 打印分类报告
print("\n分类报告:")
target_names = [idx_to_label[i] for i in sorted(idx_to_label.keys())]
print(classification_report(actuals, predictions, target_names=target_names))
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--data_dir", type=str, default="data")
parser.add_argument("--epochs", type=int, default=5)
parser.add_argument("--batch_size", type=int, default=16)
args = parser.parse_args()
main(args)
这个训练脚本涵盖了完整的训练流程，包括数据加载、模型构建、训练循环、验证和模型保存。代码中使用了Hugging Face的Transformers库，它提供了大量预训练模型和便捷的工具函数。
第三步：构建预测服务
模型训练完成后，下一步是将其部署为可用的预测服务。我们使用FastAPI来构建这个服务：
# api.py
import os
import json
from typing import List, Optional
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, Field
import torch
from transformers import AutoTokenizer, AutoModel
import numpy as np
# ============================================
# 数据模型定义
# ============================================
class TextInput(BaseModel):
"""文本输入模型"""
text: str = Field(..., min_length=1, max_length=5000, description="待分类的文本")
model_version: Optional[str] = Field(None, description="模型版本")
class PredictionResult(BaseModel):
"""预测结果模型"""
label: str = Field(..., description="预测的类别")
confidence: float = Field(..., ge=0, le=1, description="预测置信度")
probabilities: dict = Field(..., description="各类别的概率分布")
class BatchPredictionInput(BaseModel):
"""批量预测输入模型"""
texts: List[str] = Field(..., min_items=1, max_items=100)
class BatchPredictionResult(BaseModel):
"""批量预测结果模型"""
results: List[PredictionResult]
# ============================================
# 模型加载
# ============================================
class ModelService:
"""模型服务类"""
def __init__(self, model_path: str, model_name: str, label_mapping_path: str):
self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# 加载模型
print(f"加载模型从 {model_path}...")
self.model = AutoModel.from_pretrained(model_name)
self.model.classifier = torch.nn.Linear(
self.model.config.hidden_size,
len(open(label_mapping_path).read().splitlines())
)
self.model.load_state_dict(torch.load(model_path, map_location=self.device))
self.model.to(self.device)
self.model.eval()
# 加载tokenizer
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
# 加载标签映射
with open(label_mapping_path, "r") as f:
mapping = json.load(f)
self.idx_to_label = {int(k): v for k, v in mapping["idx_to_label"].items()}
def predict(self, text: str) -> PredictionResult:
"""单条文本预测"""
# Tokenize
encoding = self.tokenizer(
text,
max_length=256,
padding="max_length",
truncation=True,
return_tensors="pt"
)
input_ids = encoding["input_ids"].to(self.device)
attention_mask = encoding["attention_mask"].to(self.device)
# 推理
with torch.no_grad():
outputs = self.model(input_ids, attention_mask)
probs = torch.softmax(outputs.logits, dim=1)
pred_idx = probs.argmax(dim=1).item()
confidence = probs[0][pred_idx].item()
# 构建结果
probabilities = {self.idx_to_label[i]: probs[0][i].item() for i in range(len(probs[0]))}
return PredictionResult(
label=self.idx_to_label[pred_idx],
confidence=confidence,
probabilities=probabilities
)
def batch_predict(self, texts: List[str]) -> BatchPredictionResult:
"""批量预测"""
results = []
for text in texts:
result = self.predict(text)
results.append(result)
return BatchPredictionResult(results=results)
# ============================================
# FastAPI应用
# ============================================
app = FastAPI(
title="新闻分类预测API",
description="基于BERT的新闻分类服务",
version="1.0.0"
)
# 全局模型服务实例
model_service = None
@app.on_event("startup")
async def startup_event():
"""应用启动时初始化模型"""
global model_service
model_service = ModelService(
model_path="models/best_model.pt",
model_name="bert-base-chinese",
label_mapping_path="data/label_mapping.json"
)
print("模型加载完成!")
@app.post("/predict", response_model=PredictionResult)
async def predict(input_data: TextInput):
"""单条文本分类接口"""
if model_service is None:
raise HTTPException(status_code=503, detail="服务未就绪")
try:
result = model_service.predict(input_data.text)
return result
except Exception as e:
raise HTTPException(status_code=500, detail=f"预测失败: {str(e)}")
@app.post("/batch_predict", response_model=BatchPredictionResult)
async def batch_predict(input_data: BatchPredictionInput):
"""批量文本分类接口"""
if model_service is None:
raise HTTPException(status_code=503, detail="服务未就绪")
try:
result = model_service.batch_predict(input_data.texts)
return result
except Exception as e:
raise HTTPException(status_code=500, detail=f"批量预测失败: {str(e)}")
@app.get("/health")
async def health_check():
"""健康检查接口"""
return {"status": "healthy", "model_loaded": model_service is not None}
@app.get("/labels")
async def get_labels():
"""获取所有标签"""
if model_service is None:
raise HTTPException(status_code=503, detail="服务未就绪")
return {"labels": list(model_service.idx_to_label.values())}
# ============================================
# 启动服务
# ============================================
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
运行这个API服务的命令很简单：
uvicorn api:app --host 0.0.0.0 --port 8000 --reload
服务启动后，你可以访问http://localhost:8000/docs来查看自动生成的API文档，并进行交互式测试。
第四步：Docker容器化部署
为了使服务可以在任何环境中运行，我们将API服务容器化。使用Docker可以实现“构建一次，到处运行”的目标：
# Dockerfile
FROM python:3.9-slim
# 设置工作目录
WORKDIR /app
# 安装系统依赖
RUN apt-get update && apt-get install -y \
build-essential \
&& rm -rf /var/lib/apt/lists/*
# 复制依赖文件
COPY requirements.txt .
# 安装Python依赖
RUN pip install --no-cache-dir -r requirements.txt
# 复制项目代码
COPY . .
# 下载预训练模型（如果需要）
# RUN python -c "from transformers import AutoModel; AutoModel.from_pretrained('bert-base-chinese')"
# 暴露端口
EXPOSE 8000
# 设置环境变量
ENV PYTHONUNBUFFERED=1
ENV PORT=8000
# 运行服务
CMD ["uvicorn", "api:app", "--host", "0.0.0.0", "--port", "8000"]
Docker Compose文件可以方便地管理多容器应用：
# docker-compose.yml
version: '3.8'
services:
api:
build: .
container_name: news-classifier-api
ports:
- "8000:8000"
environment:
- MODEL_PATH=/app/models/best_model.pt
- DEVICE=cuda # 或 cpu
volumes:
- ./models:/app/models # 挂载模型目录
restart: unless-stopped
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 30s
timeout: 10s
retries: 3
start_period: 40s
# 可以添加其他服务，如Redis缓存、监控等
# redis:
# image: redis:alpine
# ports:
# - "6379:6379"
部署时只需要执行：
docker-compose up -d
docker-compose ps # 查看容器状态
docker-compose logs -f # 查看日志
典型应用场景 / Common Use Cases and Scenarios
通过前面的学习，你已经掌握了使用Made-With-ML构建生产级ML系统的核心技能。现在让我们来看看这个框架在现实世界中的典型应用场景。
情感分析系统
情感分析是最常见的NLP应用之一。电商平台可以使用它来分析用户评价，自动识别好评和差评；社交媒体平台可以用它来监测舆情，及时发现负面信息的传播；客服系统可以用它来优先处理情绪激动的用户投诉。
使用Made-With-ML的思路构建情感分析系统时，关键是要收集高质量的标注数据。由于情感分析的主观性较强，数据标注的质量直接影响模型效果。建议使用多标注者投票的方式，或者雇佣专业的标注团队。
此外，情感分析需要处理各种口语化表达、网络用语和表情符号。在数据预处理阶段，你需要考虑这些问题，并设计相应的处理策略。
内容推荐系统
推荐系统是ML在工业界最成功的应用之一。一个完整的内容推荐系统通常包括召回（retrieval）和排序（ranking）两个阶段。召回阶段从海量内容中快速筛选出候选集；排序阶段对候选集进行精细排序，输出最终推荐结果。
Made-With-ML的模块化设计非常适合构建推荐系统。你可以为召回模型和排序模型分别创建独立的训练流程，共享数据处理和评估逻辑。这种设计既保证了代码的复用性，又便于针对不同阶段进行优化。
垃圾邮件检测
垃圾邮件检测是一个经典的二分类问题。虽然看似简单，但实际应用中需要处理很多边界情况，比如新型垃圾邮件样式、域名伪装、图像垃圾邮件等。
对于这种实时性要求较高的场景，建议使用轻量级模型（如DistilBERT），并在模型层面做一些优化，比如量化（quantization）和知识蒸馏（knowledge distillation）。FastAPI的异步特性也非常适合这种IO密集型的预测任务。
文本分类的企业应用
在企业内部，文本分类有着广泛的应用场景。人力资源部门可以用它来自动筛选简历；法务部门可以用它来分类合同条款；客服部门可以用它来自动分配工单。
企业应用通常对准确率要求较高，因为错误分类可能导致严重后果。建议在这种场景中设置人工复核机制，对于置信度较低的预测结果，交由人工处理。同时，需要建立完善的日志系统，记录所有预测结果，便于后续分析和改进。
技巧与最佳实践 / Tips and Best Practices
在学习和使用Made-With-ML的过程中，我总结了一些实用的技巧和最佳实践，希望能帮助你少走弯路。
代码组织建议
良好的代码组织是项目可维护性的基础。建议遵循以下原则：
首先，保持模块的单一职责。每个Python文件应该只负责一个功能模块，比如数据加载、模型定义、训练逻辑等。这样做的好处是便于测试和复用，也使得代码更容易理解。
其次，使用相对导入。当项目结构比较复杂时，使用相对导入可以让模块之间的引用更加清晰：
# 在project包内部使用相对导入
from .config import Config
from .utils import set_seed
第三，编写详细的文档字符串。不仅是模块和函数级别的文档，还要为配置参数、返回值等编写清晰的说明。这对于团队协作和后续维护非常重要：
def train_epoch(model, data_loader, optimizer, scheduler, criterion, device, grad_clip=None):
"""
训练一个epoch。
参数:
model: 要训练的模型
data_loader: 数据加载器
optimizer: 优化器
scheduler: 学习率调度器
criterion: 损失函数
device: 计算设备
grad_clip: 梯度裁剪阈值，如果为None则不进行裁剪
返回:
tuple: (平均损失, 准确率)
"""
调试技巧
调试ML代码比调试普通代码更加困难，因为问题可能出在数据、模型结构、训练过程等多个环节。以下是一些实用的调试技巧。
在开始训练之前，先用一小部分数据（几十条）验证整个流程是否能正常运行。这样可以快速排除代码语法错误和基本的逻辑问题：
# 快速验证脚本
def quick_validation():
"""快速验证训练流程"""
# 只用10条数据
small_dataset = train_dataset[:10]
small_loader = DataLoader(small_dataset, batch_size=2)
# 只训练1个batch
for batch in small_loader:
outputs = model(batch["input_ids"], batch["attention_mask"])
loss = criterion(outputs, batch["label"])
loss.backward()
optimizer.step()
break
print("快速验证通过!")
使用日志记录中间结果，而不仅仅是最终的损失值。在关键步骤添加print语句或使用logging模块，可以帮助你追踪数据流动和梯度传播：
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 在训练过程中记录
logger.info(f"Epoch {epoch}, Step {step}, Loss: {loss.item()}, LR: {scheduler.get_last_lr()}")
性能优化建议
当模型训练速度成为瓶颈时，可以考虑以下优化策略。
数据加载优化是最容易取得突破的地方。使用num_workers参数可以并行加载数据，pin_memory可以加速CPU到GPU的数据传输：
train_loader = DataLoader(
train_dataset,
batch_size=32,
num_workers=4, # 使用4个进程加载数据
pin_memory=True, # 加速数据传输
prefetch_factor=2 # 预取因子
)
混合精度训练可以显著减少显存占用和加速训练。PyTorch 1.6以上版本原生支持自动混合精度：
from torch.cuda.amp import autocast, GradScaler
scaler = GradScaler()
for batch in data_loader:
optimizer.zero_grad()
with autocast():
outputs = model(input_ids, attention_mask)
loss = criterion(outputs, labels)
scaler.scale(loss).backward()
scaler.step(optimizer)
scaler.update()
对于推理阶段，可以考虑使用ONNX Runtime或者模型量化来提升推理速度。这些技术在保持模型精度的同时，可以获得数倍的性能提升。
版本控制与实验管理
建议使用Git来管理代码版本，并编写清晰的commit message。对于ML项目，可以考虑使用DVC（Data Version Control）来管理数据和模型版本：
# 初始化DVC
dvc init
# 添加数据目录
dvc add data/
# 添加模型目录
dvc add models/
实验追踪可以使用MLflow、Weights & Biases或TensorBoard。这些工具可以自动记录实验参数、指标变化和模型版本，方便你比较不同实验的效果：
# 使用MLflow记录实验
import mlflow
mlflow.set_experiment("news-classification")
with mlflow.start_run():
mlflow.log_param("learning_rate", Config.learning_rate)
mlflow.log_param("batch_size", Config.batch_size)
# 训练过程
for epoch in range(Config.epochs):
train_loss, train_acc = train_epoch(...)
mlflow.log_metric("train_loss", train_loss, step=epoch)
mlflow.log_metric("train_acc", train_acc, step=epoch)
总结 / Conclusion
到这里，我们已经完整地学习了Made-With-ML项目的核心理念和实战技能。回顾一下，本文涵盖了以下主要内容：
首先，我们了解了Made-With-ML项目的背景和价值。这个项目不仅仅是一些示例代码，而是一套完整的生产级ML系统构建方法论，填补了“从学习到应用”之间的鸿沟。
其次，我们详细学习了项目的核心功能模块，包括配置管理系统、数据处理流水线、模型定义、训练循环和预测服务。每个模块都遵循了软件工程的最佳实践，体现了模块化、可复用、易维护的设计理念。
第三，我们通过一个完整的实战项目——新闻分类系统——实践了从数据准备、模型训练到服务部署的全流程。这个项目涵盖了实际工作中会遇到的大部分场景，可以作为你自己项目的起点。
最后，我们讨论了Made-With-ML的典型应用场景和最佳实践。这些经验来自于真实项目的积累，可以帮助你避免常见的陷阱，提升项目的成功率。
如果你想进一步深入学习，我推荐关注以下相关资源：
Hugging Face的官方文档和教程提供了丰富的预训练模型和实战案例，是学习NLP和Transformers的绝佳资源。PyTorch的官方教程涵盖了从基础到高级的各个方面，非常适合系统学习深度学习框架。MLOps社区汇集了大量关于机器学习工程化的讨论和实践经验，是了解行业最佳实践的好去处。
Made-With-ML的GitHub仓库本身也在不断更新，建议定期查看以获取最新内容。如果你对项目有任何问题或建议，可以在GitHub上提issue或pull request，参与到这个开源项目的建设中来。
学习是一个持续的过程。希望本文能成为你ML工程化之旅的一个良好起点。在实际应用中，你会遇到各种意想不到的挑战，但只要掌握了正确的方法论和工具，就一定能够找到解决方案。祝你学习愉快，项目顺利！
相关项目链接
Made-With-ML官方资源
- GitHub仓库：https://github.com/GokuMohandas/Made-With-ML
- 项目文档：https://madewithml.com/
- 作者Twitter：@GokuMohandas
学习资源推荐
- PyTorch官方教程：https://pytorch.org/tutorials/
- Hugging Face课程：https://huggingface.co/course/chapter1/1
- fastapi：https://fastapi.tiangolo.com/zh/tutorial/
- MLflow：https://mlflow.org/docs/latest/index.html
工具与框架
- Docker官方文档：https://docs.docker.com/
- Weights & Biases：https://wandb.ai/
- DVC数据版本控制：https://dvc.org/
- ONNX模型转换：https://onnxruntime.ai/
这些资源涵盖了ML工程化的各个方面，可以帮助你构建更完善的生产级ML系统。建议从实际项目出发，边做边学，逐步积累经验。
评论区