1. 从零搭建AI工程能力:为什么我劝你别一上来就调包
这两年“AI工程”这个词被说得太多了,多到有点变味。打开任何一个技术社区,满屏都是“三行代码调用大模型”“十分钟搭建RAG应用”“零基础转行AI工程师”。我不否认这些内容有它的价值,但如果你真的想在这个方向上走得远一点,迟早会发现一个尴尬的事实:会调API的人一抓一大把,能把一个AI系统稳定跑在生产环境里的人,少得可怜。
ai-engineering-from-scratch这个标题,我第一次看到的时候就觉得它戳中了痛点。它讲的不是“怎么用现成工具拼一个demo”,而是“从底层开始,把AI工程这件事真正搞明白”。这两者的差别,就像“会开车”和“会修车”的差别——平时看不出什么,一旦上了高速抛锚,差距就出来了。
我写这篇东西,是想把自己从零搭建AI工程能力这条路上踩过的坑、绕过的弯、以及那些“早知道就好了”的经验,系统地梳理一遍。适合谁看?如果你是刚入行一两年、能写Python但没真正做过AI系统的开发者,这篇内容能帮你少走至少半年的弯路。如果你已经做过一些AI项目,但总觉得“知其然不知其所以然”,那这篇也能帮你把知识体系里的窟窿补上。如果你是完全零基础的小白,也别急着关掉,我会尽量用生活化的类比把复杂的东西讲清楚,你至少能搞清楚这个领域到底在干什么、需要学什么。
核心关键词就一个:ai-engineering-from-scratch。我会围绕它,把从环境搭建、数据处理、模型训练、推理部署到监控运维的完整链路拆开来讲,每个环节都告诉你“为什么这么做”和“不这么做会怎样”。
2. 整体设计思路:为什么“从零”比“调包”更值得投入
2.1 先搞清楚“AI工程”到底在工程什么
很多人对AI工程的理解停留在“训练模型”这一步。实际上,训练模型只是整个链路里的一环,而且往往不是最耗时的那一环。一个完整的AI工程系统,至少包含以下几个部分:
- 数据管道:数据的采集、清洗、标注、版本管理、特征工程
- 训练管道:模型选型、超参调优、分布式训练、实验追踪
- 推理服务:模型导出、量化压缩、服务封装、负载均衡
- 监控运维:性能监控、数据漂移检测、模型退化告警、A/B测试
- 迭代闭环:bad case收集、数据回流、模型再训练、灰度发布
你去看那些“三行代码调用大模型”的教程,它们只覆盖了“推理服务”里最表层的一点点。真正让一个AI系统在生产环境里活下来的,是剩下那百分之八十的脏活累活。
我见过太多团队,demo跑得飞起,一上生产就崩。为什么?因为demo阶段的数据是干净的、请求量是小的、模型是不需要更新的。一旦这些假设被打破,没有底层工程能力支撑的系统就会像纸糊的房子一样塌掉。
2.2 为什么选择“从零”而不是“从框架”
这里说的“从零”,不是让你用纯Python手写矩阵乘法(虽然手写一遍确实有帮助),而是说你要理解每一层在做什么,而不是把它当黑盒。
举个例子。你用PyTorch训练一个模型,一行loss.backward()就完成了反向传播。这行代码背后发生了什么?计算图的构建、链式法则的应用、梯度的累加和清零。如果你不理解这些,当loss不收敛的时候,你只能靠猜。但如果你理解,你就能系统地排查:是梯度消失了?是学习率太大了?是数据分布有问题?还是计算图被意外截断了?
再举个例子。你把模型部署成API服务,用户请求过来,你调用model.predict()返回结果。看起来很简单。但如果QPS上来了,延迟飙升,你怎么优化?是模型太大需要量化?是批处理没做好?是GPU利用率不够?还是Python的GIL在拖后腿?这些问题,调包侠是回答不了的。
“从零”的价值不在于你不用框架,而在于你理解框架帮你做了什么,从而在出问题的时候知道去哪里找原因。
2.3 技术选型的几个关键决策
在搭建AI工程能力的过程中,有几个选型决策会深刻影响你后续的开发效率。我把自己做过的选择和一些思考整理成表格,供你参考。
| 环节 | 常见选项 | 我的选择 | 选择理由 |
|---|---|---|---|
| 深度学习框架 | PyTorch / TensorFlow / JAX | PyTorch | 动态图调试友好,社区生态活跃,学术界主流 |
| 实验追踪 | MLflow / Weights & Biases / TensorBoard | MLflow | 开源可自托管,与训练代码解耦,不绑定云服务 |
| 数据版本管理 | DVC / Git LFS / 自建 | DVC | 与Git工作流无缝集成,支持大文件远程存储 |
| 模型服务 | TorchServe / Triton / FastAPI自建 | FastAPI自建 | 灵活可控,适合理解底层原理,后期可迁移到Triton |
| 容器化 | Docker / Podman | Docker | 生态成熟,文档丰富,CI/CD集成方便 |
| 编排 | Kubernetes / Docker Compose | Docker Compose起步 | 学习曲线平缓,单机足够,后期平滑迁移K8s |
这些选择没有绝对的对错,关键是你要知道每个选项背后的权衡。比如我选FastAPI自建推理服务而不是TorchServe,不是因为TorchServe不好,而是因为我想先理解一个推理服务需要处理哪些问题——请求解析、批处理、超时控制、错误处理、日志记录。理解了这些之后,再用TorchServe或者Triton,我就知道它们帮我解决了什么,而不是把它们当魔法盒子。
提示:选型的时候不要只看“哪个最流行”,要看“哪个最能帮你理解问题”。学习阶段,可控性比性能更重要。
3. 核心细节解析:从环境搭建到第一个训练管道
3.1 开发环境:别在这上面浪费时间
我见过太多人卡在环境配置上,装CUDA装了两天,最后放弃了。说实话,环境配置确实烦,但它是必须过的坎。我的建议是:用Docker,别在宿主机上折腾。
为什么?因为AI工程涉及的东西太多了——Python版本、CUDA版本、cuDNN版本、各种框架的版本兼容性。你在宿主机上装,装崩了很难恢复。用Docker,每个项目一个镜像,互不干扰,崩了删掉重来就行。
一个典型的PyTorch开发环境的Dockerfile大概长这样:
FROM nvidia/cuda:12.1.0-cudnn8-devel-ubuntu22.04 RUN apt-get update && apt-get install -y \ python3.10 \ python3-pip \ git \ vim \ && rm -rf /var/lib/apt/lists/* RUN pip3 install --no-cache-dir \ torch==2.1.0 \ torchvision==0.16.0 \ numpy \ pandas \ scikit-learn \ matplotlib \ jupyterlab WORKDIR /workspace这个镜像基于NVIDIA官方的CUDA镜像,装了Python 3.10和PyTorch 2.1.0。版本号要写死,不要用latest,否则今天能跑的代码明天可能就跑不了了。
构建和运行:
docker build -t ai-dev:latest . docker run --gpus all -it -v $(pwd):/workspace -p 8888:8888 ai-dev:latest--gpus all把GPU透传给容器,-v把当前目录挂载进去,-p把Jupyter的端口映射出来。这样你在容器里改代码,宿主机上也能看到,容器删了代码还在。
注意:CUDA版本、PyTorch版本、显卡驱动版本三者之间有兼容性矩阵,装之前一定要去官网查一下。我踩过的坑是驱动太老,装最新版CUDA跑不起来,折腾了半天才发现是驱动的问题。
3.2 数据管道:脏活里的脏活
数据管道是AI工程里最不起眼但最重要的部分。我敢说,一个AI项目百分之七十的时间花在数据处理上,剩下百分之三十里又有百分之七十花在因为数据处理没做好而导致的bug上。
数据管道要解决几个核心问题:
数据版本管理。你今天用这份数据训练了一个模型,效果不错。下周数据更新了,模型效果掉了。你想回滚到上周的数据重新训练,发现数据已经被覆盖了。这就是没有数据版本管理的后果。DVC可以解决这个问题,它的工作方式和Git很像,但专门针对大文件。
# 初始化DVC dvc init # 添加数据文件到DVC管理 dvc add data/training_set.csv # 这会生成一个.dvc文件,把它提交到Git git add data/training_set.csv.dvc data/.gitignore git commit -m "add training data v1"DVC会把实际的数据文件存到远程存储(比如S3、MinIO或者本地目录),Git里只保留一个指针文件。这样你的Git仓库不会因为数据文件而膨胀,同时又能精确追踪每个版本的数据。
数据清洗和验证。原始数据永远是脏的——缺失值、异常值、重复值、格式不一致。你需要一套可复用的清洗流程,而不是每次手动处理。我习惯用pandas写一个清洗脚本,把每一步都记录下来。
import pandas as pd import numpy as np def clean_data(df): # 记录原始行数 original_rows = len(df) # 去除完全重复的行 df = df.drop_duplicates() # 处理缺失值:数值列用中位数填充,类别列用众数填充 for col in df.columns: if df[col].dtype in ['float64', 'int64']: df[col] = df[col].fillna(df[col].median()) else: df[col] = df[col].fillna(df[col].mode()[0]) # 处理异常值:用IQR方法 for col in df.select_dtypes(include=[np.number]).columns: Q1 = df[col].quantile(0.25) Q3 = df[col].quantile(0.75) IQR = Q3 - Q1 lower = Q1 - 1.5 * IQR upper = Q3 + 1.5 * IQR df[col] = df[col].clip(lower, upper) print(f"清洗完成:{original_rows}行 -> {len(df)}行") return df这个脚本看起来简单,但每一步都有讲究。比如用中位数而不是均值填充缺失值,是因为中位数对异常值更鲁棒。用IQR方法处理异常值而不是直接删除,是因为删除可能会丢失重要信息。
特征工程。这是最需要领域知识的部分。同样的数据,不同的人做出来的特征可能天差地别。我的经验是:先做最简单的特征,跑通整个管道,然后再逐步迭代。不要一上来就搞复杂的特征交叉,那样你连baseline都没有,根本不知道复杂特征有没有用。
3.3 训练管道:从单机到分布式
训练管道的第一步是写一个能跑通的训练脚本。这个脚本不需要多复杂,但必须结构清晰,方便后续扩展。
import torch import torch.nn as nn from torch.utils.data import DataLoader, Dataset import mlflow class MyDataset(Dataset): def __init__(self, features, labels): self.features = torch.FloatTensor(features) self.labels = torch.LongTensor(labels) def __len__(self): return len(self.labels) def __getitem__(self, idx): return self.features[idx], self.labels[idx] def train_epoch(model, dataloader, optimizer, criterion, device): model.train() total_loss = 0 for batch_features, batch_labels in dataloader: batch_features = batch_features.to(device) batch_labels = batch_labels.to(device) optimizer.zero_grad() outputs = model(batch_features) loss = criterion(outputs, batch_labels) loss.backward() optimizer.step() total_loss += loss.item() return total_loss / len(dataloader) def evaluate(model, dataloader, criterion, device): model.eval() total_loss = 0 correct = 0 total = 0 with torch.no_grad(): for batch_features, batch_labels in dataloader: batch_features = batch_features.to(device) batch_labels = batch_labels.to(device) outputs = model(batch_features) loss = criterion(outputs, batch_labels) total_loss += loss.item() _, predicted = torch.max(outputs, 1) total += batch_labels.size(0) correct += (predicted == batch_labels).sum().item() accuracy = correct / total return total_loss / len(dataloader), accuracy def train(config): device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') # 加载数据 train_dataset = MyDataset(config['train_features'], config['train_labels']) val_dataset = MyDataset(config['val_features'], config['val_labels']) train_loader = DataLoader(train_dataset, batch_size=config['batch_size'], shuffle=True) val_loader = DataLoader(val_dataset, batch_size=config['batch_size'], shuffle=False) # 初始化模型 model = config['model_fn']().to(device) optimizer = torch.optim.Adam(model.parameters(), lr=config['learning_rate']) criterion = nn.CrossEntropyLoss() # 实验追踪 mlflow.set_experiment(config['experiment_name']) with mlflow.start_run(): mlflow.log_params({ 'batch_size': config['batch_size'], 'learning_rate': config['learning_rate'], 'epochs': config['epochs'] }) best_val_acc = 0 for epoch in range(config['epochs']): train_loss = train_epoch(model, train_loader, optimizer, criterion, device) val_loss, val_acc = evaluate(model, val_loader, criterion, device) mlflow.log_metrics({ 'train_loss': train_loss, 'val_loss': val_loss, 'val_accuracy': val_acc }, step=epoch) print(f"Epoch {epoch+1}/{config['epochs']} | " f"Train Loss: {train_loss:.4f} | " f"Val Loss: {val_loss:.4f} | " f"Val Acc: {val_acc:.4f}") if val_acc > best_val_acc: best_val_acc = val_acc torch.save(model.state_dict(), 'best_model.pt') mlflow.log_artifact('best_model.pt') return model这个训练脚本有几个关键设计:
实验追踪集成。用MLflow记录每次实验的超参数和指标。这样你跑了二十次实验之后,还能清楚地知道哪次用了什么参数、效果如何。没有实验追踪,你的实验记录就是一堆乱七八糟的文件名。
模型保存策略。只保存验证集上效果最好的模型,而不是最后一个epoch的模型。这是防止过拟合的基本操作,但我见过很多人忘了做。
配置与代码分离。超参数通过config字典传入,而不是硬编码在函数里。这样你可以用不同的配置跑多次实验,而不需要改代码。
当单机单卡跑不动的时候,就需要上分布式训练。PyTorch提供了DDP(DistributedDataParallel)来实现多卡训练。核心改动是把模型用DDP包装一下,然后用DistributedSampler来分配数据。
import torch.distributed as dist from torch.nn.parallel import DistributedDataParallel as DDP from torch.utils.data.distributed import DistributedSampler def setup(rank, world_size): dist.init_process_group("nccl", rank=rank, world_size=world_size) def cleanup(): dist.destroy_process_group() def train_ddp(rank, world_size, config): setup(rank, world_size) model = config['model_fn']().to(rank) model = DDP(model, device_ids=[rank]) train_dataset = MyDataset(config['train_features'], config['train_labels']) sampler = DistributedSampler(train_dataset, num_replicas=world_size, rank=rank) train_loader = DataLoader(train_dataset, batch_size=config['batch_size'], sampler=sampler) # ... 训练循环 cleanup()分布式训练的水很深,涉及通信后端选择、梯度同步、混合精度训练等。我的建议是:先把单卡跑通,确保代码逻辑没问题,再上分布式。否则你连bug出在逻辑上还是分布式实现上都分不清。
4. 实操过程:从训练到部署的完整链路
4.1 模型导出与优化
训练好的模型不能直接扔给推理服务用。PyTorch的模型保存的是state_dict,推理的时候需要重新构建模型结构再加载权重。这个过程容易出错,而且Python的模型定义代码在生产环境里可能不可用。
解决方案是把模型导出成与框架无关的格式。ONNX是最常用的选择。
import torch.onnx # 加载训练好的模型 model = MyModel() model.load_state_dict(torch.load('best_model.pt')) model.eval() # 构造一个示例输入 dummy_input = torch.randn(1, input_dim) # 导出为ONNX torch.onnx.export( model, dummy_input, "model.onnx", export_params=True, opset_version=13, input_names=['input'], output_names=['output'], dynamic_axes={ 'input': {0: 'batch_size'}, 'output': {0: 'batch_size'} } )dynamic_axes参数很关键。如果不设置,导出的ONNX模型会固定batch size为1,推理的时候只能一次处理一条数据,效率极低。设置之后,模型可以接受任意batch size的输入。
导出ONNX之后,还可以进一步优化。ONNX Runtime提供了图优化功能,可以合并算子、消除冗余计算。
import onnxruntime as ort from onnxruntime.transformers import optimizer # 优化ONNX模型 optimized_model = optimizer.optimize_model( "model.onnx", model_type='bert', # 根据实际模型类型选择 num_heads=12, hidden_size=768 ) optimized_model.save_model_to_file("model_optimized.onnx")如果推理延迟还是不够低,可以考虑量化。把FP32的权重转成INT8,模型大小缩小四倍,推理速度提升两到三倍,精度损失通常在可接受范围内。
from onnxruntime.quantization import quantize_dynamic, QuantType quantize_dynamic( "model_optimized.onnx", "model_quantized.onnx", weight_type=QuantType.QUInt8 )注意:量化不是万能的。对于某些对精度敏感的模型,量化后的效果可能下降明显。一定要在验证集上对比量化前后的指标,确认精度损失在可接受范围内再上线。
4.2 推理服务封装
推理服务的核心要求是:低延迟、高吞吐、稳定可靠。用FastAPI封装一个推理服务,代码大概是这样:
from fastapi import FastAPI, HTTPException from pydantic import BaseModel import onnxruntime as ort import numpy as np import time import logging app = FastAPI() # 全局加载模型,避免每次请求都重新加载 session = ort.InferenceSession("model_quantized.onnx") class PredictRequest(BaseModel): features: list class PredictResponse(BaseModel): prediction: int confidence: float latency_ms: float @app.post("/predict", response_model=PredictResponse) async def predict(request: PredictRequest): start_time = time.time() try: input_data = np.array([request.features], dtype=np.float32) input_name = session.get_inputs()[0].name outputs = session.run(None, {input_name: input_data}) logits = outputs[0][0] prediction = int(np.argmax(logits)) confidence = float(np.max(logits)) latency = (time.time() - start_time) * 1000 logging.info(f"Prediction: {prediction}, Latency: {latency:.2f}ms") return PredictResponse( prediction=prediction, confidence=confidence, latency_ms=latency ) except Exception as e: logging.error(f"Prediction failed: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/health") async def health(): return {"status": "healthy"}这个服务有几个关键点:
模型全局加载。在应用启动时加载模型,而不是每次请求都加载。模型加载是IO密集型操作,每次请求都加载的话延迟会高得离谱。
健康检查接口。/health接口用于负载均衡器或者Kubernetes的健康检查。没有这个接口,编排系统不知道你的服务是否正常。
延迟记录。每次请求都记录延迟,方便后续监控和优化。延迟数据是发现性能问题的第一手资料。
错误处理。推理过程中可能出各种问题——输入格式不对、模型加载失败、内存不足。用try-except包起来,返回有意义的错误信息,而不是让服务直接崩溃。
启动服务:
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4--workers 4启动四个工作进程。因为Python有GIL,单进程只能利用一个CPU核心。多进程可以充分利用多核CPU,提升吞吐量。
4.3 容器化部署
把推理服务打包成Docker镜像,确保在任何环境里都能一致运行。
FROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY model_quantized.onnx . COPY main.py . EXPOSE 8000 CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]构建和运行:
docker build -t inference-service:latest . docker run -d -p 8000:8000 --name inference inference-service:latest如果需要GPU推理,基础镜像换成NVIDIA的CUDA镜像,运行时加上--gpus all。
用Docker Compose编排多个服务(推理服务、监控、日志收集):
version: '3.8' services: inference: build: . ports: - "8000:8000" deploy: resources: reservations: devices: - driver: nvidia count: 1 capabilities: [gpu] healthcheck: test: ["CMD", "curl", "-f", "http://localhost:8000/health"] interval: 30s timeout: 10s retries: 3 prometheus: image: prom/prometheus:latest ports: - "9090:9090" volumes: - ./prometheus.yml:/etc/prometheus/prometheus.yml grafana: image: grafana/grafana:latest ports: - "3000:3000" depends_on: - prometheus这个编排文件把推理服务、Prometheus监控、Grafana可视化面板组合在一起。推理服务的指标暴露给Prometheus,Grafana从Prometheus拉数据展示。
4.4 监控与告警
服务上线只是开始,真正的挑战在于让它稳定运行。监控是发现问题的眼睛。
需要监控的指标分几类:
系统指标:CPU使用率、内存使用率、GPU使用率、GPU显存占用、磁盘IO、网络IO。这些指标反映的是基础设施的健康状况。
服务指标:QPS、延迟分布(P50、P95、P99)、错误率、超时率。这些指标反映的是服务的性能表现。
模型指标:预测分布、置信度分布、特征分布。这些指标反映的是模型的行为是否正常。
业务指标:转化率、点击率、用户满意度。这些指标反映的是模型对业务的实际影响。
在FastAPI里暴露Prometheus格式的指标:
from prometheus_client import Counter, Histogram, generate_latest from fastapi import Response REQUEST_COUNT = Counter('inference_requests_total', 'Total inference requests') REQUEST_LATENCY = Histogram('inference_latency_seconds', 'Inference latency') PREDICTION_DISTRIBUTION = Counter('prediction_distribution', 'Prediction distribution', ['prediction']) @app.post("/predict") async def predict(request: PredictRequest): REQUEST_COUNT.inc() with REQUEST_LATENCY.time(): # ... 推理逻辑 PREDICTION_DISTRIBUTION.labels(prediction=str(prediction)).inc() return response @app.get("/metrics") async def metrics(): return Response(content=generate_latest(), media_type="text/plain")数据漂移检测是模型监控里最容易被忽视但最重要的部分。模型上线后,输入数据的分布可能会慢慢变化,导致模型效果下降。比如一个电商推荐模型,训练数据是夏季的商品,到了冬季用户行为变了,模型效果就会掉。
检测数据漂移的基本方法是:定期计算线上数据的统计量(均值、方差、分位数),和训练数据的统计量对比。如果差异超过阈值,就触发告警。
import numpy as np from scipy import stats def detect_drift(reference_data, current_data, threshold=0.05): """用KS检验检测数据漂移""" drift_detected = False drift_features = [] for i in range(reference_data.shape[1]): statistic, p_value = stats.ks_2samp( reference_data[:, i], current_data[:, i] ) if p_value < threshold: drift_detected = True drift_features.append(i) return drift_detected, drift_featuresKS检验的原理是比较两个分布的累积分布函数。如果p值小于阈值,说明两个分布有显著差异。这个方法简单有效,适合作为第一道防线。
5. 常见问题与排查技巧实录
5.1 训练阶段的典型问题
Loss不收敛或者变成NaN。这是最常见的问题,原因可能有很多。我的排查顺序是:
- 检查学习率。学习率太大是最常见的原因。试试降低10倍。
- 检查数据。有没有NaN值?有没有标签错误?有没有特征尺度差异过大?
- 检查模型。有没有梯度消失或爆炸?加梯度裁剪试试。
- 检查损失函数。分类问题用交叉熵,回归问题用MSE,别用错了。
梯度裁剪的代码:
torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)这行代码放在loss.backward()之后、optimizer.step()之前,可以防止梯度爆炸。
过拟合。训练集效果好,验证集效果差。解决方法:
- 增加数据量(最有效)
- 数据增强
- 正则化(L1、L2、Dropout)
- 早停(验证集loss不再下降就停止训练)
- 减小模型复杂度
训练速度慢。可能的原因:
- 数据加载是瓶颈。用
num_workers参数增加数据加载进程数。 - GPU利用率低。用
nvidia-smi查看GPU利用率,如果低于50%,说明数据加载或预处理拖了后腿。 - 没有用混合精度训练。
torch.cuda.amp可以自动把部分计算转成FP16,速度提升明显。
from torch.cuda.amp import autocast, GradScaler scaler = GradScaler() for batch_features, batch_labels in dataloader: optimizer.zero_grad() with autocast(): outputs = model(batch_features) loss = criterion(outputs, batch_labels) scaler.scale(loss).backward() scaler.step(optimizer) scaler.update()5.2 部署阶段的典型问题
推理延迟高。排查思路:
| 可能原因 | 排查方法 | 解决方案 |
|---|---|---|
| 模型太大 | 查看模型文件大小 | 量化、剪枝、蒸馏 |
| 没有批处理 | 查看GPU利用率 | 实现动态批处理 |
| 预处理慢 | 打时间戳测量各阶段耗时 | 优化预处理代码,用C++扩展 |
| 网络传输慢 | 测量请求响应时间 | 压缩输入数据,就近部署 |
| 进程数不够 | 查看CPU利用率 | 增加worker数量 |
内存泄漏。服务跑一段时间后内存持续增长,最终OOM。常见原因:
- 全局变量累积。比如把每次请求的数据都append到一个全局list里。
- 没有释放中间变量。PyTorch的tensor如果一直持有引用,不会被GC回收。
- 日志文件没有轮转。日志写满了磁盘。
排查方法:用tracemalloc或者memory_profiler定位内存增长的位置。
import tracemalloc tracemalloc.start() # ... 运行一段时间 snapshot = tracemalloc.take_snapshot() top_stats = snapshot.statistics('lineno') for stat in top_stats[:10]: print(stat)版本兼容性问题。训练用的PyTorch版本和推理用的ONNX Runtime版本不兼容,导致模型加载失败。解决方案:固定所有依赖的版本号,用requirements.txt或者poetry.lock锁定。
5.3 监控阶段的典型问题
告警风暴。一个服务出问题,触发几百条告警,把值班的人淹没了。解决方案:告警聚合和抑制。同一类型的告警在短时间内只发一条,关联告警合并成一条。
数据漂移误报。数据本身波动就大,KS检验频繁触发告警。解决方案:调整阈值,或者用更鲁棒的方法(比如PSI,Population Stability Index)。另外,漂移检测的频率不要太高,每天一次就够了,没必要每分钟都跑。
模型退化发现太晚。模型效果已经掉了两周,才发现。解决方案:除了监控预测分布,还要监控业务指标。业务指标的变化往往比模型指标更早反映问题。
5.4 独家避坑技巧
技巧一:永远保留一个baseline。不管你做什么改进,都要有一个最简单的baseline作为参照。没有baseline,你无法判断改进是否有效。baseline可以是一个规则模型,也可以是一个简单的逻辑回归。
技巧二:先跑通再优化。不要一上来就追求最优方案。先用最简单的方法把整个链路跑通,然后再逐个环节优化。我见过太多人卡在某个环节的优化上,结果整个项目都没跑起来。
技巧三:日志要打够,但别太多。关键节点打日志——数据加载完成、训练开始、每个epoch结束、模型保存、推理请求、错误发生。但不要在循环里打日志,那样日志文件会爆炸。
技巧四:配置文件用YAML,别用Python。YAML可读性好,非技术人员也能改。Python配置文件虽然灵活,但容易引入bug,而且不好做版本对比。
技巧五:模型文件不要提交到Git。用DVC或者Git LFS管理。模型文件动辄几百MB,提交到Git会让仓库变得巨大无比,clone一次要半天。
技巧六:推理服务要有超时控制。没有超时控制的服务,一个慢请求就能拖垮整个服务。FastAPI可以用asyncio.wait_for实现超时。
import asyncio @app.post("/predict") async def predict(request: PredictRequest): try: result = await asyncio.wait_for( run_inference(request), timeout=5.0 ) return result except asyncio.TimeoutError: raise HTTPException(status_code=504, detail="Inference timeout")技巧七:定期做故障演练。故意杀掉推理服务的进程,看看监控能不能发现、告警能不能触发、服务能不能自动恢复。没做过故障演练的系统,真出问题的时候一定手忙脚乱。
6. 从零到一之后:持续迭代的工程习惯
把整个链路跑通只是第一步。真正让AI工程能力持续提升的,是日常的工程习惯。
实验记录要规范。每次实验都要记录:日期、目的、改动内容、超参数、结果、结论。用MLflow或者Weights & Biases自动记录,但也要写文字说明。三个月后你回头看,只有数字没有文字说明的实验记录,你根本看不懂当时在干什么。
代码review不能省。AI项目的代码也是代码,也需要review。特别是数据处理和特征工程部分,一个微妙的bug可能导致模型效果差很多,而且很难发现。让同事帮你看看代码,往往能发现你忽略的问题。
文档要写,但别写废话。文档的目的是让新人能快速上手,让未来的你能回忆起当时的决策。重点写:架构设计、关键决策的理由、踩过的坑、运维手册。不要写“这个函数的作用是计算损失”这种废话。
技术债要还。AI项目特别容易积累技术债——临时的数据处理脚本、硬编码的超参数、没有测试的代码。这些债不还,项目会越来越难维护。每个迭代留出一定比例的时间还技术债,比如20%。
保持学习,但别追新。AI领域每天都有新东西出来,但你不需要每个都学。先把基础打牢——数据结构、算法、操作系统、网络、数据库。这些基础知识十年不过时。新框架、新工具,用到的时候再学。
我自己在这条路上走了几年,最大的体会是:AI工程的核心不是AI,是工程。模型可以调包,但工程能力调不了包。数据管道、训练管道、推理服务、监控运维,这些东西需要你一行一行代码写出来,一个一个坑踩过来。
ai-engineering-from-scratch这个方向,值得投入。不是因为AI热,而是因为这套工程能力是通用的——你今天用它做图像分类,明天用它做推荐系统,后天用它做任何其他AI应用,底层的东西是不变的。
最后分享一个我最近在用的技巧:给每个项目建一个LESSONS.md文件,每次踩坑之后就往里面记一条。不用写得多正式,一句话就行。比如“ONNX导出时忘了设dynamic_axes,导致batch size固定为1”。积累几个月,这个文件就是你最宝贵的财富。下次做新项目的时候,先翻一遍这个文件,能省掉很多重复踩坑的时间。