MLOps与AI系统运维完全教程
教程简介
MLOps（Machine Learning Operations）是将机器学习模型从实验阶段推向生产环境的工程化实践体系。本教程系统讲解MLOps的核心理念、工具链和最佳实践，帮助团队构建高效、可靠的AI系统运维体系。
第一章：MLOps成熟度模型
1.1 MLOps三个级别
Google提出的MLOps成熟度模型将AI运维分为三个级别：
Level 0 - 手动流程
- 手动训练模型、手动部署
- 无CI/CD、无监控
- 适合实验阶段
Level 1 - ML管道自动化
- 自动化训练管道
- 持续训练（CT）
- 特征存储
- 实验追踪
Level 2 - CI/CD/CT全自动化
- 完整的CI/CD/CT流水线
- 自动化测试与验证
- 模型监控与漂移检测
- 自动回滚机制
1.2 MLOps vs DevOps
# MLOps与DevOps的核心区别
comparison = {
"代码版本管理": "Git (两者相同)",
"数据版本管理": "DVC/Delta Lake (MLOps独有)",
"模型版本管理": "MLflow Model Registry (MLOps独有)",
"测试": "单元测试→模型质量测试 (MLOps扩展)",
"部署": "容器化部署→A/B测试+影子模式 (MLOps扩展)",
"监控": "系统指标→数据漂移+模型衰减 (MLOps扩展)",
}
第二章：实验追踪与版本管理
2.1 MLflow实验追踪
import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, f1_score
class ExperimentTracker:
"""实验追踪器"""
def __init__(self, experiment_name: str):
mlflow.set_experiment(experiment_name)
def track_training(self, X_train, y_train, X_test, y_test, params: dict):
"""追踪训练过程"""
with mlflow.start_run():
# 记录参数
mlflow.log_params(params)
# 训练模型
model = RandomForestClassifier(**params)
model.fit(X_train, y_train)
# 评估
y_pred = model.predict(X_test)
accuracy = accuracy_score(y_test, y_pred)
f1 = f1_score(y_test, y_pred, average='weighted')
# 记录指标
mlflow.log_metric("accuracy", accuracy)
mlflow.log_metric("f1_score", f1)
# 记录模型
mlflow.sklearn.log_model(
model,
"model",
registered_model_name="my_classifier"
)
# 记录数据版本
mlflow.log_artifact("data/train.csv")
return accuracy, f1
def compare_experiments(self, experiment_name: str):
"""对比实验结果"""
experiment = mlflow.get_experiment_by_name(experiment_name)
runs = mlflow.search_runs(experiment_ids=[experiment.experiment_id])
# 按F1分数排序
runs_sorted = runs.sort_values("metrics.f1_score", ascending=False)
return runs_sorted[["run_id", "metrics.accuracy", "metrics.f1_score", "params"]]
2.2 Weights & Biases追踪
import wandb
class WandBTracker:
"""W&B实验追踪器"""
def __init__(self, project: str, config: dict):
self.run = wandb.init(project=project, config=config)
def log_epoch(self, epoch: int, train_loss: float, val_loss: float, metrics: dict):
"""记录每个epoch的指标"""
wandb.log({
"epoch": epoch,
"train_loss": train_loss,
"val_loss": val_loss,
**metrics
})
def log_model(self, model_path: str, name: str):
"""记录模型"""
artifact = wandb.Artifact(name, type="model")
artifact.add_file(model_path)
self.run.log_artifact(artifact)
def log_confusion_matrix(self, y_true, y_pred, class_names):
"""记录混淆矩阵"""
wandb.log({
"confusion_matrix": wandb.plot.confusion_matrix(
y_true=y_true, preds=y_pred, class_names=class_names
)
})
2.3 DVC数据版本管理
# DVC初始化
dvc init
# 追踪数据文件
dvc add data/train.csv data/test.csv
# 数据版本管理
git add data/train.csv.dvc data/test.csv.dvc
git commit - "Add training data v1"
# 配置远程存储
dvc remote add -d myremote s3://my-bucket/dvc-store
# 推送数据到远程
dvc push
# 切换到历史版本
git checkout v1.0 -- data/train.csv.dvc
dvc checkout
# 数据管道
dvc repro
# dvc.yaml - 定义数据管道
stages:
prepare:
cmd: python src/prepare.py
deps:
- src/prepare.py
- data/raw.csv
outs:
- data/train.csv
- data/test.csv
train:
cmd: python src/train.py
deps:
- src/train.py
- data/train.csv
outs:
- models/model.pkl
metrics:
- metrics.json
第三章：特征工程与特征存储
3.1 Feast特征存储
from feast import FeatureStore, Entity, FeatureView, Field
from feast.types import Float32, Int64, String
from feast.infra.offline_stores.file_source import FileSource
from datetime import timedelta
# 定义实体
user = Entity(
name="user_id",
join_keys=["user_id"],
description="用户ID"
)
# 定义特征视图
user_features = FeatureView(
name="user_features",
entities=[user],
schema=[
Field(name="age", dtype=Int64),
Field(name="total_orders", dtype=Int64),
Field(name="avg_order_amount", dtype=Float32),
Field(name="membership_level", dtype=String),
],
source=FileSource(
path="data/user_features.parquet",
timestamp_field="event_timestamp"
),
ttl=timedelta(days=1),
)
# 注册特征
store = FeatureStore(repo_path=".")
store.apply([user, user_features])
# 在线检索
feature_vector = store.get_online_features(
features=[
"user_features:age",
"user_features:total_orders",
"user_features:avg_order_amount"
],
entity_rows=[{"user_id": "U001"}]
).to_dict()
# 离线训练数据集
training_df = store.get_historical_features(
entity_df=entity_df,
features=[
"user_features:age",
"user_features:total_orders",
"user_features:avg_order_amount"
]
).to_df()
第四章：CI/CD/CT流水线构建
4.1 模型训练自动化
# 使用 Prefect 构建训练管道
from prefect import task, flow
from prefect.tasks import task_input_hash
from datetime import timedelta
@task(cache_key_fn=task_input_hash, cache_expiration=timedelta(hours=1))
def load_data(data_path: str):
"""加载数据"""
import pandas as pd
return pd.read_csv(data_path)
@task
def preprocess(df):
"""数据预处理"""
from sklearn.preprocessing import StandardScaler
scaler = StandardScaler()
features = df.drop("target", axis=1)
scaled = scaler.fit_transform(features)
return scaled, df["target"], scaler
@task(retries=3, retry_delay_seconds=60)
def train_model(X_train, y_train, params: dict):
"""训练模型"""
from sklearn.ensemble import GradientBoostingClassifier
model = GradientBoostingClassifier(**params)
model.fit(X_train, y_train)
return model
@task
def evaluate_model(model, X_test, y_test) -> dict:
"""评估模型"""
from sklearn.metrics import accuracy_score, f1_score, classification_report
y_pred = model.predict(X_test)
return {
"accuracy": accuracy_score(y_test, y_pred),
"f1_score": f1_score(y_test, y_pred, average='weighted'),
"report": classification_report(y_test, y_pred)
}
@task
def validate_model(metrics: dict, threshold: float = 0.85) -> bool:
"""验证模型是否达标"""
return metrics["f1_score"] >= threshold
@task
def register_model(model, model_name: str):
"""注册模型"""
import mlflow
mlflow.sklearn.log_model(model, "model", registered_model_name=model_name)
@flow(name="training-pipeline")
def training_pipeline(data_path: str, params: dict):
"""完整训练管道"""
# 加载数据
df = load_data(data_path)
# 预处理
X, y, scaler = preprocess(df)
# 划分数据集
from sklearn.model_selection import train_test_split
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)
# 训练
model = train_model(X_train, y_train, params)
# 评估
metrics = evaluate_model(model, X_test, y_test)
print(f"评估结果: {metrics}")
# 验证
if validate_model(metrics):
register_model(model, "production_model")
print("模型已注册")
else:
print("模型未达标，跳过注册")
4.2 模型部署策略
class ModelDeployer:
"""模型部署器"""
def blue_green_deploy(self, model_name: str, model_version: str):
"""蓝绿部署"""
# 1. 在Green环境部署新模型
self._deploy_to_green(model_name, model_version)
# 2. 运行健康检查
if self._health_check("green"):
# 3. 切换流量
self._switch_traffic("green")
# 4. 停止Blue环境
self._stop_blue()
else:
# 回滚
self._stop_green()
raise Exception("Green环境健康检查失败")
def canary_deploy(self, model_name: str, model_version: str,
canary_ratio: float = 0.1):
"""金丝雀部署"""
# 1. 部署新版本到少量实例
self._deploy_canary(model_name, model_version)
# 2. 分流指定比例的流量
self._set_traffic_split(canary_ratio)
# 3. 监控新版本指标
metrics = self._monitor_canary(duration_minutes=30)
# 4. 判断是否全量发布
if self._evaluate_canary_metrics(metrics):
self._promote_canary_to_production()
else:
self._rollback_canary()
def shadow_deploy(self, model_name: str, model_version: str):
"""影子部署（用于验证）"""
# 1. 部署新模型但不服务真实流量
self._deploy_shadow(model_name, model_version)
# 2. 复制生产流量到影子模型
self._mirror_traffic("shadow")
# 3. 对比新旧模型的预测结果
comparison = self._compare_predictions(duration_hours=24)
return comparison
第五章：推理服务框架
5.1 vLLM高性能推理
from vllm import LLM, SamplingParams
class VLLMInferenceService:
"""vLLM推理服务"""
def __init__(self, model_name: str, **kwargs):
self.llm = LLM(
model=model_name,
tensor_parallel_size=kwargs.get("tp_size", 1),
gpu_memory_utilization=kwargs.get("gpu_util", 0.9),
max_num_batched_tokens=kwargs.get("max_batch_tokens", 8192),
max_num_seqs=kwargs.get("max_seqs", 256),
dtype="auto",
trust_remote_code=True
)
def generate(self, prompts: list, **kwargs) -> list:
"""批量生成"""
sampling_params = SamplingParams(
temperature=kwargs.get("temperature", 0.7),
top_p=kwargs.get("top_p", 0.9),
max_tokens=kwargs.get("max_tokens", 512),
stop=kwargs.get("stop", None)
)
outputs = self.llm.generate(prompts, sampling_params)
return [output.outputs[0].text for output in outputs]
def chat(self, messages: list, **kwargs) -> str:
"""对话接口"""
from vllm import ChatCompletionRequest
request = ChatCompletionRequest(
model=self.llm.model_config.model,
messages=messages,
temperature=kwargs.get("temperature", 0.7),
max_tokens=kwargs.get("max_tokens", 512)
)
response = self.llm.chat(request)
return response.choices[0].message.content
5.2 Triton Inference Server
# 模型仓库结构
# model_repository/
# └── my_model/
# ├── config.pbtxt
# ├── 1/
# │ └── model.py
# └── 2/
# └── model.py
# config.pbtxt 示例
"""
name: "my_llm_model"
platform: "python"
max_batch_size: 64
input [
{
name: "input_text"
data_type: TYPE_STRING
dims: [1]
}
]
output [
{
name: "output_text"
data_type: TYPE_STRING
dims: [1]
}
]
instance_group [
{
count: 2
kind: KIND_GPU
}
]
dynamic_batching {
preferred_batch_size: [8, 16, 32]
max_queue_delay_microseconds: 100000
}
"""
# Python模型实现 (model.py)
class TritonPythonModel:
def initialize(self, args):
"""初始化模型"""
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
model_path = "/models/my_llm_model/1/"
self.tokenizer = AutoTokenizer.from_pretrained(model_path)
self.model = AutoModelForCausalLM.from_pretrained(
model_path,
torch_dtype=torch.float16,
device_map="auto"
)
def execute(self, requests):
"""处理推理请求"""
responses = []
for request in requests:
input_text = pb_utils.get_input_tensor_by_name(request, "input_text").as_numpy()
# 推理
inputs = self.tokenizer(input_text[0].decode(), return_tensors="pt").to("cuda")
outputs = self.model.generate(**inputs, max_new_tokens=256)
result = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
# 构建响应
output_tensor = pb_utils.Tensor("output_text", np.array([result.encode()]))
responses.append(pb_utils.InferenceResponse([output_tensor]))
return responses
第六章：模型监控与数据漂移检测
6.1 数据漂移检测
import numpy as np
from scipy import stats
class DriftDetector:
"""数据漂移检测器"""
def __init__(self, reference_data: np.ndarray):
self.reference = reference_data
self.reference_stats = self._compute_stats(reference_data)
def _compute_stats(self, data: np.ndarray) -> dict:
"""计算数据统计特征"""
return {
"mean": np.mean(data, axis=0),
"std": np.std(data, axis=0),
"median": np.median(data, axis=0),
"q25": np.percentile(data, 25, axis=0),
"q75": np.percentile(data, 75, axis=0)
}
def ks_test(self, new_data: np.ndarray, threshold: float = 0.05) -> dict:
"""KS检验：检测分布变化"""
results = {}
for i in range(new_data.shape[1]):
statistic, p_value = stats.ks_2samp(
self.reference[:, i], new_data[:, i]
)
results[f"feature_{i}"] = {
"statistic": float(statistic),
"p_value": float(p_value),
"drift_detected": p_value < threshold
}
return results
def psi(self, new_data: np.ndarray, bins: int = 10) -> dict:
"""PSI（Population Stability Index）：量化分布偏移"""
results = {}
for i in range(new_data.shape[1]):
# 计算分箱边界
breakpoints = np.percentile(self.reference[:, i],
np.linspace(0, 100, bins + 1))
# 计算各箱的占比
ref_counts = np.histogram(self.reference[:, i], bins=breakpoints)[0]
new_counts = np.histogram(new_data[:, i], bins=breakpoints)[0]
ref_pct = ref_counts / len(self.reference) + 1e-10
new_pct = new_counts / len(new_data) + 1e-10
# 计算PSI
psi_value = np.sum((new_pct - ref_pct) * np.log(new_pct / ref_pct))
results[f"feature_{i}"] = {
"psi": float(psi_value),
"status": "stable" if psi_value < 0.1 else "moderate" if psi_value < 0.25 else "significant"
}
return results
def monitor_prediction_drift(self, reference_preds: list,
new_preds: list) -> dict:
"""监控预测结果漂移"""
ref_dist = np.bincount(reference_preds) / len(reference_preds)
new_dist = np.bincount(new_preds, minlength=len(ref_dist)) / len(new_preds)
# 卡方检验
chi2, p_value = stats.chisquare(new_dist * len(new_preds),
ref_dist * len(new_preds))
return {
"chi2_statistic": float(chi2),
"p_value": float(p_value),
"drift_detected": p_value < 0.05
}
6.2 模型性能监控
class ModelMonitor:
"""模型性能监控"""
def __init__(self, alert_threshold: dict):
self.threshold = alert_threshold
self.metrics_history = []
def log_prediction(self, input_data: dict, prediction: str,
ground_truth: str = None, latency_ms: float = 0):
"""记录预测日志"""
record = {
"timestamp": datetime.now().isoformat(),
"input": input_data,
"prediction": prediction,
"ground_truth": ground_truth,
"latency_ms": latency_ms
}
self.metrics_history.append(record)
def compute_metrics(self, window_minutes: int = 60) -> dict:
"""计算时间窗口内的指标"""
cutoff = datetime.now() - timedelta(minutes=window_minutes)
recent = [r for r in self.metrics_history
if datetime.fromisoformat(r["timestamp"]) > cutoff]
if not recent:
return {}
# 延迟统计
latencies = [r["latency_ms"] for r in recent if r["latency_ms"] > 0]
# 准确率（有ground truth的）
labeled = [r for r in recent if r["ground_truth"] is not None]
accuracy = None
if labeled:
correct = sum(1 for r in labeled if r["prediction"] == r["ground_truth"])
accuracy = correct / len(labeled)
return {
"total_predictions": len(recent),
"latency_p50": np.percentile(latencies, 50) if latencies else None,
"latency_p95": np.percentile(latencies, 95) if latencies else None,
"latency_p99": np.percentile(latencies, 99) if latencies else None,
"accuracy": accuracy,
"labeled_ratio": len(labeled) / len(recent) if recent else 0
}
def check_alerts(self, metrics: dict) -> list:
"""检查告警"""
alerts = []
if metrics.get("latency_p99") and metrics["latency_p99"] > self.threshold.get("max_latency_p99", 5000):
alerts.append({
"type": "high_latency",
"message": f"P99延迟过高: {metrics['latency_p99']:.0f}ms",
"severity": "warning"
})
if metrics.get("accuracy") and metrics["accuracy"] < self.threshold.get("min_accuracy", 0.8):
alerts.append({
"type": "low_accuracy",
"message": f"准确率过低: {metrics['accuracy']:.2%}",
"severity": "critical"
})
return alerts
第七章：LLMOps专项
7.1 LLM特有的运维挑战
class LLMOperations:
"""LLM运维管理"""
def __init__(self, llm_client):
self.llm = llm_client
self.guardrails = []
def add_guardrail(self, guardrail):
"""添加安全护栏"""
self.guardrails.append(guardrail)
async def safe_generate(self, prompt: str, **kwargs) -> dict:
"""带安全检查的生成"""
# 输入检查
for guard in self.guardrails:
check_result = await guard.check_input(prompt)
if not check_result["passed"]:
return {
"blocked": True,
"reason": check_result["reason"],
"guardrail": guard.name
}
# 生成
response = await self.llm.chat(
messages=[{"role": "user", "content": prompt}],
**kwargs
)
# 输出检查
for guard in self.guardrails:
check_result = await guard.check_output(response)
if not check_result["passed"]:
return {
"blocked": True,
"reason": check_result["reason"],
"guardrail": guard.name
}
return {"blocked": False, "response": response}
def cost_tracker(self, input_tokens: int, output_tokens: int,
model: str) -> float:
"""成本追踪"""
pricing = {
"gpt-4o": {"input": 2.5 / 1_000_000, "output": 10 / 1_000_000},
"gpt-4o-mini": {"input": 0.15 / 1_000_000, "output": 0.6 / 1_000_000},
"claude-3-5-sonnet": {"input": 3 / 1_000_000, "output": 15 / 1_000_000},
}
if model not in pricing:
return 0.0
cost = (input_tokens * pricing[model]["input"] +
output_tokens * pricing[model]["output"])
return cost
7.2 Prompt版本管理
import hashlib
import json
from datetime import datetime
class PromptRegistry:
"""Prompt版本注册中心"""
def __init__(self, storage_path: str = "prompts/"):
self.storage_path = storage_path
self.registry = {}
def register(self, name: str, template: str,
metadata: dict = None) -> str:
"""注册Prompt版本"""
# 计算版本哈希
version_hash = hashlib.sha256(template.encode()).hexdigest()[:8]
entry = {
"name": name,
"version": version_hash,
"template": template,
"metadata": metadata or {},
"created_at": datetime.now().isoformat(),
"is_active": False
}
if name not in self.registry:
self.registry[name] = []
self.registry[name].append(entry)
# 持久化
self._save(name, entry)
return version_hash
def activate(self, name: str, version: str):
"""激活指定版本"""
for entry in self.registry.get(name, []):
entry["is_active"] = entry["version"] == version
def get_active(self, name: str) -> str:
"""获取当前激活的Prompt"""
for entry in self.registry.get(name, []):
if entry["is_active"]:
return entry["template"]
return None
def compare_versions(self, name: str, v1: str, v2: str) -> dict:
"""对比两个版本"""
versions = {e["version"]: e for e in self.registry.get(name, [])}
return {
"v1": versions.get(v1),
"v2": versions.get(v2),
"diff": self._diff(versions[v1]["template"], versions[v2]["template"])
}
最佳实践总结
- 从小处开始：先实现基本的实验追踪，再逐步完善CI/CD
- 数据版本化：数据和代码同等重要，必须版本管理
- 自动化测试：模型测试包括准确性、延迟、公平性等维度
- 渐进式部署：蓝绿部署→金丝雀部署→影子部署
- 持续监控：不仅监控系统指标，还要监控数据漂移和模型衰减
- 成本意识：追踪GPU使用和API调用成本，优化资源利用
- 安全优先：LLM应用必须有Guardrails安全护栏
总结
MLOps是AI应用从实验到生产的关键桥梁。本教程涵盖了从实验追踪、特征管理、CI/CD流水线、推理服务到监控告警的完整技术栈。随着LLM应用的兴起，LLMOps作为MLOps的子领域也在快速发展。掌握这些工程化实践，将帮助团队高效、可靠地将AI能力交付给用户。
