Python AI 基础设施演进：从 Jupyter 到 Script 到 MLOps

Python AI 基础设施演进：从 Jupyter 到 Script 到 MLOps

一、Jupyter 的便利与工程化陷阱

Python AI 开发的传统起点是 Jupyter Notebook。它提供了交互式开发体验，允许数据科学家和算法工程师快速验证想法、可视化数据和迭代模型。然而，当项目从实验阶段走向生产环境时，Jupyter 的局限性变得尤为突出。

Jupyter 的核心问题：

# Jupyter 典型问题示例：隐藏状态依赖
# 在 Notebook 中，以下代码可能正常运行，但在脚本中会失败

# 单元格 1（已执行）
import pandas as pd
data = pd.read_csv('data.csv')

# 单元格 2（已执行，依赖单元格 1 的变量）
filtered_data = data[data['value'] > 100]

# 单元格 3（依赖单元格 2 的结果）
model = train_model(filtered_data)

# 如果按脚本执行，必须保证所有依赖按顺序导入，且变量作用域清晰

生产环境中的经验教训表明，过度依赖 Jupyter 进行 AI 开发会导致技术债务累积。团队需要建立从实验到生产的规范化流程，而这就引出了下一个演进阶段：Script 化。

二、Script 化：从实验到工程的桥梁

Script 化是将 Jupyter Notebook 中的实验代码重构为可维护、可测试、可复用的 Python 脚本的过程。这一步是 AI 工程化的关键转折点，也是许多团队容易忽视的环节。

Script 化的核心原则：

# Script 化示例：将 Jupyter 代码重构为可维护的模块

# config.yaml
# model:
# name: "bert-base-chinese"
# max_length: 512
# batch_size: 32
# data:
# train_path: "data/train.csv"
# valid_path: "data/valid.csv"
# training:
# learning_rate: 2e-5
# epochs: 10

import argparse
import logging
from pathlib import Path
from typing import Tuple, Optional
import pandas as pd
import torch
from transformers import BertTokenizer, BertForSequenceClassification
import yaml

# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s – %(name)s – %(levelname)s – %(message)s'
)
logger = logging.getLogger(__name__)

class Config:
"""配置管理类"""
def __init__(self, config_path: str):
with open(config_path, 'r') as f:
self._config = yaml.safe_load(f)

def get(self, key: str, default=None):
"""支持点号分隔的配置访问"""
keys = key.split('.')
value = self._config
for k in keys:
value = value.get(k, default)
if value is None:
return default
return value

class DataProcessor:
"""数据处理模块"""
def __init__(self, config: Config):
self.config = config
self.tokenizer = BertTokenizer.from_pretrained(config.get('model.name'))

def load_data(self, data_path: str) -> pd.DataFrame:
"""加载数据，包含错误处理"""
try:
df = pd.read_csv(data_path)
logger.info(f"成功加载数据：{data_path}，共 {len(df)} 条记录")
return df
except FileNotFoundError:
logger.error(f"文件不存在：{data_path}")
raise
except pd.errors.EmptyDataError:
logger.error(f"文件为空：{data_path}")
raise

def tokenize(self, texts: list) -> dict:
"""批量tokenize处理"""
return self.tokenizer(
texts,
padding=True,
truncation=True,
max_length=self.config.get('model.max_length'),
return_tensors='pt'
)

class ModelTrainer:
"""模型训练模块"""
def __init__(self, config: Config):
self.config = config
self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
logger.info(f"使用设备：{self.device}")

def train(self, train_data: pd.DataFrame, valid_data: pd.DataFrame) -> None:
"""训练模型，包含完整的错误处理和日志"""
try:
model = BertForSequenceClassification.from_pretrained(
self.config.get('model.name'),
num_labels=2
)
model.to(self.device)

# 训练逻辑…
logger.info("开始训练…")

except torch.cuda.OutOfMemoryError:
logger.error("GPU 内存不足，请减少 batch_size")
raise
except Exception as e:
logger.error(f"训练过程出现异常：{str(e)}")
raise

def main():
"""主函数：解析参数、加载配置、执行训练"""
parser = argparse.ArgumentParser(description="BERT 文本分类训练脚本")
parser.add_argument('–config', type=str, required=True, help='配置文件路径')
args = parser.parse_args()

# 加载配置
config = Config(args.config)

# 初始化模块
processor = DataProcessor(config)
trainer = ModelTrainer(config)

# 执行训练流程
train_data = processor.load_data(config.get('data.train_path'))
valid_data = processor.load_data(config.get('data.valid_path'))

trainer.train(train_data, valid_data)

if __name__ == "__main__":
main()

Script 化不仅仅是代码格式的变更，更是开发思维的转换：从"快速验证"到"可靠交付"。然而，当项目规模扩大、团队协作增多时，单纯的 Script 化仍显不足。这就需要一个更系统化的解决方案：MLOps。

三、MLOps：AI 工程化的系统化实践

MLOps（Machine Learning Operations）是将 DevOps 原则应用于机器学习系统的实践集合。它涵盖了从数据准备、模型训练、模型部署到模型监控的完整生命周期管理。MLOps 的目标是实现机器学习系统的可复现、可扩展、可维护和自动化。

MLOps 的核心组成部分：

# MLOps 实践示例：使用 MLflow 进行实验跟踪和模型管理

import mlflow
import mlflow.pytorch
from mlflow.models import infer_signature
import torch
import torch.nn as nn
from torch.utils.data import DataLoader
from sklearn.metrics import accuracy_score, f1_score
import numpy as np

class MLOpsTrainer:
"""集成 MLOps 功能的模型训练器"""

def __init__(self, config: dict):
self.config = config
# 设置 MLflow 跟踪服务器
mlflow.set_tracking_uri(config.get('mlflow_tracking_uri', 'http://localhost:5000'))
mlflow.set_experiment(config.get('experiment_name', 'bert_classification'))

def train_with_tracking(self, model, train_loader, valid_loader):
"""训练模型并记录到 MLflow"""

# 开始 MLflow 运行
with mlflow.start_run(run_name=self.config.get('run_name')) as run:
run_id = run.info.run_id
logger.info(f"MLflow Run ID: {run_id}")

# 记录超参数
mlflow.log_params({
'learning_rate': self.config.get('learning_rate'),
'batch_size': self.config.get('batch_size'),
'epochs': self.config.get('epochs'),
'model_name': self.config.get('model_name')
})

# 训练循环
best_f1 = 0
for epoch in range(self.config.get('epochs')):
# 训练阶段
model.train()
train_loss = self._train_epoch(model, train_loader, epoch)

# 验证阶段
model.eval()
val_loss, val_acc, val_f1 = self._validate(model, valid_loader)

# 记录指标
mlflow.log_metrics({
'train_loss': train_loss,
'val_loss': val_loss,
'val_accuracy': val_acc,
'val_f1': val_f1
}, step=epoch)

logger.info(f"Epoch {epoch+1}/{self.config.get('epochs')} – "
f"Train Loss: {train_loss:.4f}, Val Acc: {val_acc:.4f}, Val F1: {val_f1:.4f}")

# 保存最佳模型
if val_f1 > best_f1:
best_f1 = val_f1
self._save_checkpoint(model, epoch, val_f1)

# 记录最佳模型到 MLflow
signature = infer_signature(
self._get_sample_input(valid_loader),
model(valid_loader.dataset[0][0].unsqueeze(0))
)

mlflow.pytorch.log_model(
model,
"model",
signature=signature,
input_example=self._get_sample_input(valid_loader)
)

mlflow.log_metric('best_f1', best_f1, step=epoch)

# 记录最终模型性能
mlflow.set_tag('final_status', 'completed')
mlflow.set_tag('model_type', 'bert_classification')

# 记录依赖环境
mlflow.log_artifact('requirements.txt')

logger.info(f"训练完成，最佳 F1 Score: {best_f1:.4f}")

def _train_epoch(self, model, train_loader, epoch):
"""训练一个 epoch"""
total_loss = 0
for batch in train_loader:
# 训练逻辑…
pass
return total_loss / len(train_loader)

def _validate(self, model, valid_loader):
"""验证模型"""
all_preds = []
all_labels = []
total_loss = 0

with torch.no_grad():
for batch in valid_loader:
# 验证逻辑…
pass

acc = accuracy_score(all_labels, all_preds)
f1 = f1_score(all_labels, all_preds, average='weighted')

return total_loss / len(valid_loader), acc, f1

def _save_checkpoint(self, model, epoch, f1):
"""保存模型检查点"""
checkpoint_path = f"checkpoints/model_epoch_{epoch}_f1_{f1:.4f}.pt"
torch.save({
'epoch': epoch,
'model_state_dict': model.state_dict(),
'f1_score': f1
}, checkpoint_path)

# 记录检查点到 MLflow
mlflow.log_artifact(checkpoint_path)

def _get_sample_input(self, dataloader):
"""获取样本输入用于 MLflow 签名推断"""
sample_batch = next(iter(dataloader))
return sample_batch[0][:1] # 返回第一个样本

# 模型部署与服务化示例（使用 FastAPI）
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uvicorn

app = FastAPI(title="BERT 文本分类 API")

class PredictionRequest(BaseModel):
text: str

class PredictionResponse(BaseModel):
label: str
confidence: float
inference_time_ms: float

@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
"""预测接口"""
import time
start_time = time.time()

try:
# 加载模型（实际应用中应使用全局模型实例）
model = mlflow.pytorch.load_model("models:/bert_classification/Production")

# 预处理
inputs = tokenizer(request.text, return_tensors='pt', truncation=True, max_length=512)

# 推理
with torch.no_grad():
outputs = model(**inputs)
probabilities = torch.softmax(outputs.logits, dim=1)
predicted_class = torch.argmax(probabilities, dim=1).item()
confidence = probabilities[0][predicted_class].item()

inference_time = (time.time() – start_time) * 1000

return {
'label': 'positive' if predicted_class == 1 else 'negative',
'confidence': confidence,
'inference_time_ms': inference_time
}

except Exception as e:
logger.error(f"预测失败：{str(e)}")
raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health():
"""健康检查接口"""
return {"status": "healthy"}

# 模型监控示例（检测数据漂移）
class ModelMonitor:
"""模型性能监控"""

def __init__(self, reference_data: np.ndarray):
self.reference_data = reference_data
self.reference_mean = np.mean(reference_data, axis=0)
self.reference_std = np.std(reference_data, axis=0)

def detect_data_drift(self, current_data: np.ndarray, threshold: float = 0.05):
"""使用 KS 检验检测数据漂移"""
from scipy.stats import ks_2samp

drift_detected = False
drift_features = []

for i in range(current_data.shape[1]):
statistic, p_value = ks_2samp(self.reference_data[:, i], current_data[:, i])

if p_value < threshold:
drift_detected = True
drift_features.append(i)

if drift_detected:
logger.warning(f"检测到数据漂移，特征索引：{drift_features}")

return drift_detected, drift_features

四、2026 年 AI 基础设施趋势与最佳实践

进入 2026 年，Python AI 基础设施继续快速演进。以下几个趋势值得关注：

1. 大模型训练的基础设施优化

随着 LLM（大语言模型）的普及，模型训练对计算资源、存储和网络的需求达到了前所未有的高度。分布式训练框架（如 DeepSpeed、Megatron-LM、FSDP）成为标配。计算存储分离架构使得训练数据可以无限扩展，而计算资源可以弹性调度。

checkpoint 管理也变得复杂。大模型的一个 checkpoint 可能达到数百 GB，需要高效的存储和快速恢复机制。增量 checkpoint、异构存储（内存+SSD+对象存储）成为必要手段。

2. 推理优化的工程化

生产环境中的模型推理需要极致的性能优化。量化（INT8、INT4）、剪枝、蒸馏等技术从研究走向工程化工具链。ONNX Runtime、TensorRT、OpenVINO 等推理引擎提供了跨平台的高性能推理能力。

模型服务化也面临新的挑战。高并发场景下的动态 Batching、KV Cache 管理、Prefix Caching 等技术直接影响成本和用户体验。开源项目如 vLLM、TGI（Text Generation Inference）提供了生产级的 LLM 推理方案。

3. 多模态数据的统一处理

现实世界的 AI 应用往往涉及文本、图像、音频、视频等多种模态。传统上，不同模态需要独立的预处理流程。2026 年的趋势是建立统一的多模态数据处理管道，支持不同模态之间的对齐、融合和转换。

工具如 PyTorch Multimodal、Hugging Face Transformers（支持 CLIP、BLIP 等多模态模型）降低了多模态开发的门槛。但工程化挑战依然存在：不同模态的数据量差异巨大（视频 vs 文本），存储和带宽成本需要仔细权衡。

4. 边缘设备上的 AI 部署

云计算并非唯一选择。在物联网、自动驾驶、移动设备等场景，AI 模型需要部署在资源受限的边缘设备上。TensorFlow Lite、PyTorch Mobile、Core ML 等框架支持模型在移动设备上的高效运行。

边缘 AI 的基础设施建设包括：模型压缩工具链、设备端推理引擎、边缘-云端协同推理框架、模型更新的 OTA（Over-the-Air）机制等。这些技术的成熟度直接影响 AI 应用的普及速度。

5. AI 安全与合规基础设施

随着 AI 监管的加强（如欧盟 AI Act），AI 系统的可解释性、公平性、隐私保护成为合规要求。基础设施需要支持：

• 数据血缘跟踪：记录数据从采集到模型训练再到推理的完整链路。
• 模型卡片（Model Card）自动生成：记录模型的性能指标、训练数据分布、已知限制等信息。
• 对抗攻击检测：监控输入数据是否包含对抗样本，保护模型安全。
• 隐私计算集成：支持联邦学习、差分隐私、同态加密等隐私保护技术。

# 2026 年 AI 基础设施最佳实践示例：大模型分布式训练

import torch
import torch.distributed as dist
from torch.nn.parallel import DistributedDataParallel as DDP
from deepspeed import initialize as ds_initialize
import wandb # 实验跟踪

class LargeModelTrainer:
"""支持大模型分布式训练的训练器"""

def __init__(self, config):
# 初始化分布式环境
dist.init_process_group(backend='nccl')
self.local_rank = int(os.environ['LOCAL_RANK'])
torch.cuda.set_device(self.local_rank)

# 初始化 WandB 实验跟踪
if self.local_rank == 0:
wandb.init(
project=config['project_name'],
name=config['experiment_name'],
config=config
)

def setup_model(self, model):
"""设置模型，使用 DeepSpeed 进行分布式训练优化"""

# DeepSpeed 配置（支持 ZeRO 优化）
ds_config = {
"train_batch_size": 32,
"gradient_accumulation_steps": 4,
"optimizer": {
"type": "AdamW",
"params": {
"lr": 1e-5,
"betas": [0.9, 0.95],
"eps": 1e-8,
"weight_decay": 0.01
}
},
"fp16": {
"enabled": True
},
"zero_optimization": {
"stage": 3, # ZeRO-3: 优化器状态、梯度、参数均分片
"offload_optimizer": {
"device": "cpu",
"pin_memory": True
},
"offload_param": {
"device": "cpu",
"pin_memory": True
}
},
"activation_checkpointing": {
"partition_activations": True,
"cpu_checkpointing": True
}
}

# 初始化 DeepSpeed
model_engine, optimizer, _, _ = ds_initialize(
model=model,
config=ds_config
)

return model_engine, optimizer

def train(self, model_engine, train_loader, valid_loader):
"""训练循环"""
for epoch in range(self.config['epochs']):
model_engine.train()

for step, batch in enumerate(train_loader):
# 前向传播
outputs = model_engine(batch)
loss = outputs.loss

# 反向传播（DeepSpeed 自动处理梯度累积和同步）
model_engine.backward(loss)
model_engine.step()

# 记录指标
if self.local_rank == 0 and step % self.config['log_interval'] == 0:
wandb.log({
'train_loss': loss.item(),
'epoch': epoch,
'step': step
})

logger.info(f"Epoch {epoch}, Step {step}, Loss: {loss.item():.4f}")

# 验证
if self.local_rank == 0:
val_loss = self.validate(model_engine, valid_loader)
wandb.log({'val_loss': val_loss, 'epoch': epoch})

# 保存模型检查点
self.save_checkpoint(model_engine, epoch, val_loss)

def validate(self, model_engine, valid_loader):
"""验证"""
model_engine.eval()
total_loss = 0

with torch.no_grad():
for batch in valid_loader:
outputs = model_engine(batch)
total_loss += outputs.loss.item()

return total_loss / len(valid_loader)

def save_checkpoint(self, model_engine, epoch, val_loss):
"""保存检查点（DeepSpeed 自动处理分布式存储）"""
save_path = f"checkpoints/epoch_{epoch}_loss_{val_loss:.4f}"
model_engine.save_checkpoint(save_path)

# 推理优化示例：使用 vLLM 进行高性能 LLM 推理

from vllm import LLM, SamplingParams

class OptimizedLLMInference:
"""使用 vLLM 进行优化推理"""

def __init__(self, model_path: str):
# 初始化 vLLM（支持 PagedAttention、Continuous Batching）
self.llm = LLM(
model=model_path,
tensor_parallel_size=2, # 使用 2 张 GPU 进行张量并行
max_num_batched_tokens=8192,
gpu_memory_utilization=0.95
)

self.sampling_params = SamplingParams(
temperature=0.7,
top_p=0.95,
max_tokens=2048
)

def batch_infer(self, prompts: list) -> list:
"""批量推理（自动连续批处理）"""
outputs = self.llm.generate(prompts, self.sampling_params)

results = []
for output in outputs:
generated_text = output.outputs[0].text
results.append(generated_text)

return results

def stream_infer(self, prompt: str):
"""流式推理"""
from vllm import RequestOutput

# vLLM 支持生成过程中的流式输出
for output in self.llm.generate([prompt], self.sampling_params, stream=True):
yield output.outputs[0].text

五、总结

Python AI 基础设施从 Jupyter 到 Script 到 MLOps 的演进，反映了 AI 工程化从实验到生产的成熟过程。2026 年的趋势表明，AI 基础设施正在向更大规模、更高性能、更安全合规的方向发展。

关键要点：

AI 工程化不是一蹴而就的过程，需要在项目实践中不断迭代和优化。从一开始就建立规范的开发流程，可以避免后期大量的重构工作。

参考资料

本文基于作者在 Python AI 工程化领域的实践经验，结合 2026 年技术趋势分析。如有不同观点，欢迎讨论。