1. 从零搭建AI工程能力为什么大多数人卡在第一步“ai-engineering-from-scratch”这个标题我第一次看到的时候心里其实是有点抵触的。因为市面上打着“从零开始”旗号的内容太多了点进去一看要么是调个API就敢叫自己AI工程师要么是甩一堆数学公式把人劝退。但仔细想想这个标题背后其实藏着一个非常真实的需求一个具备基本编程能力的人想真正理解AI系统是怎么从一行行代码变成能跑起来、能扛住流量、能持续迭代的工程系统到底该走哪条路。我自己在这个方向上摸索了挺长时间踩过的坑不算少。最开始我以为AI工程就是学框架PyTorch、TensorFlow、JAX轮着来哪个火学哪个。后来发现框架只是工具真正难的是把模型从notebook里拽出来让它在一个真实的环境里稳定运行。这个过程涉及的东西远比想象中杂数据管道怎么设计、模型怎么版本化、推理服务怎么部署、延迟怎么压、成本怎么控、监控怎么做。每一项单独拎出来都不算特别难但要把它们串成一条完整的链路中间任何一个环节掉链子整个系统就转不起来。所以这篇内容我想聊的不是某个具体的框架怎么用也不是某个模型怎么训。我想聊的是AI工程能力的搭建路径——从一个空目录开始到一套能跑通、能扩展、能维护的AI系统中间需要经历哪些阶段每个阶段的核心任务是什么以及哪些地方最容易翻车。适合的读者是那些已经会写代码、但对AI工程还没有系统认知的人也适合那些已经在做AI相关开发、但总觉得自己的知识是碎片化的人。接下来的内容我会按照实际搭建的顺序来展开从环境准备到数据层、模型层、服务层、监控层最后聊一下持续迭代的问题。每个部分我都会尽量说清楚“为什么这么做”而不只是“怎么做”因为工具会变但底层的工程逻辑变化没那么快。2. 环境与工具链的选型别一上来就追求“全家桶”2.1 开发环境的最小可用集合很多人一开始就想着把环境搭得特别完善Docker、Kubernetes、MLflow、Airflow全装上结果光配置环境就花了一周真正写代码的时间反而没多少。我的建议是起步阶段只装真正需要的东西。一个最小可用的AI工程开发环境其实就这几样Python环境管理用conda或者uv都行关键是能隔离不同项目的依赖。我个人的习惯是用uv速度快依赖解析也干净。如果你团队里用conda的人多那就跟着用conda别在这种事情上标新立异。代码编辑器VS Code加上Python和Jupyter插件基本够用。如果你习惯PyCharm也没问题但VS Code在远程开发场景下确实更方便。版本控制Git是必须的但更重要的是怎么组织仓库结构。我见过太多项目把所有代码堆在一个目录里到后面连自己都找不到东西在哪。一个简单的实验跟踪工具最开始用不着MLflow这种重型的哪怕就是在文件里记一下每次实验的参数和结果都行。关键是养成记录的习惯而不是工具本身。提示不要在项目初期就引入Kubernetes。我见过不止一个团队模型还没训出来K8s集群倒是搭得挺漂亮。除非你的部署环境已经确定是K8s否则用Docker Compose甚至直接裸机跑都够用。2.2 仓库结构的设计逻辑仓库结构这件事看起来是小事实际上影响很大。一个清晰的目录结构能让协作效率提升不少也能让后来的维护者少骂几句。我一般会按照功能分层的方式来组织project/ ├── data/ # 数据处理相关 │ ├── raw/ # 原始数据只读 │ ├── processed/ # 处理后的数据 │ └── scripts/ # 数据处理脚本 ├── models/ # 模型定义和训练 │ ├── architectures/ # 模型结构 │ ├── training/ # 训练脚本 │ └── evaluation/ # 评估脚本 ├── serving/ # 推理服务 │ ├── api/ # API接口 │ └── workers/ # 后台任务 ├── configs/ # 配置文件 ├── tests/ # 测试 └── notebooks/ # 探索性分析这个结构的关键在于职责分离。data目录只负责数据models目录只负责模型serving目录只负责服务。每个目录内部的代码不应该跨层调用比如serving里的代码不应该直接去读data/raw里的文件而应该通过一个明确的数据接口来获取。为什么要这么设计因为AI系统的迭代速度很快今天用的模型明天可能就换了今天的数据源后天可能就变了。如果各层之间耦合太紧改一个地方就要动全身。职责分离之后替换某一层的实现就不会影响到其他层。2.3 配置管理别把参数写死在代码里这是新手最容易犯的错误之一。学习率、batch size、模型路径、数据库连接串全都硬编码在代码里。等到需要改一个参数的时候得去代码里翻半天。正确的做法是把所有可变的参数抽到配置文件里。YAML、JSON、TOML都行选一个团队里大家都熟悉的格式。我一般用YAML因为可读性好支持注释。# configs/training/base.yaml model: name: resnet50 num_classes: 10 pretrained: true training: batch_size: 32 learning_rate: 0.001 epochs: 50 optimizer: adam data: train_path: data/processed/train val_path: data/processed/val num_workers: 4然后在代码里用一个配置加载器来读取import yaml def load_config(path): with open(path, r) as f: config yaml.safe_load(f) return config config load_config(configs/training/base.yaml)这样做的好处是不同环境可以用不同的配置文件比如base.yaml、dev.yaml、prod.yaml通过环境变量或者命令行参数来切换。代码本身不需要任何修改。注意配置文件也要纳入版本控制但敏感信息比如API密钥、数据库密码不要直接写在配置文件里用环境变量或者密钥管理服务来注入。3. 数据层AI工程里最容易被低估的部分3.1 数据管道的设计原则很多人觉得数据层没什么技术含量不就是读文件、洗数据、喂给模型吗但实际做起来数据层往往是整个系统里最复杂、最容易出问题的部分。数据管道的核心设计原则就一条可复现。同样的输入经过同样的处理流程应该得到同样的输出。听起来很简单但做起来不容易。随机种子没固定、文件读取顺序不确定、并行处理时结果合并顺序不一致这些都会导致每次跑出来的数据不一样。我一般会遵循这几个实践原始数据只读data/raw目录下的文件永远不修改所有处理结果写到data/processed。处理脚本幂等同一个脚本跑多次结果应该是一样的。如果做不到至少要有明确的版本标记。数据版本化每次处理后的数据打一个版本号记录在元数据文件里。这样出了问题可以追溯到具体是哪个版本的数据。import hashlib import json from datetime import datetime def generate_data_version(raw_path, processed_path, config): 生成数据版本标识 version_info { timestamp: datetime.now().isoformat(), raw_path: raw_path, processed_path: processed_path, config_hash: hashlib.md5( json.dumps(config, sort_keysTrue).encode() ).hexdigest()[:8] } return version_info3.2 数据质量检查的实操方法数据质量检查这件事说起来重要做起来容易忘。我的经验是把检查逻辑写成代码而不是靠人眼去看。人眼只能看几条样本代码可以检查全量数据。最基本的检查包括检查项检查内容处理方式缺失值各字段的空值比例超过阈值告警数值范围数值字段的min/max/均值超出预期范围告警类别分布分类字段的取值分布分布偏移超过阈值告警重复样本完全重复或近似重复的记录去重或标记格式一致性日期、ID等字段的格式格式不符则拒绝这些检查不需要很复杂的工具用pandas加上一些自定义函数就能搞定。关键是要在数据进入训练流程之前执行而不是等到模型效果不好再回头查数据。import pandas as pd def check_data_quality(df, config): 基础数据质量检查 report {} # 缺失值检查 missing_ratio df.isnull().mean() report[missing] missing_ratio[missing_ratio config[missing_threshold]].to_dict() # 数值范围检查 for col in config[numeric_columns]: if col in df.columns: stats df[col].describe() report[f{col}_stats] { min: float(stats[min]), max: float(stats[max]), mean: float(stats[mean]) } # 类别分布检查 for col in config[categorical_columns]: if col in df.columns: report[f{col}_distribution] df[col].value_counts(normalizeTrue).to_dict() return report3.3 数据加载的性能优化数据加载往往是训练流程的瓶颈。GPU在等数据的时候利用率可能只有30%甚至更低。解决这个问题有几个方向第一用更高效的数据格式。CSV读取慢可以转成Parquet或者WebDataset格式。Parquet是列式存储读取特定列的时候比CSV快很多。WebDataset适合大规模数据集支持流式读取。第二预取和缓存。PyTorch的DataLoader本身支持多进程加载和预取把num_workers设大一点prefetch_factor也调一下通常能明显提升吞吐。第三数据预处理离线化。能在离线阶段做的处理比如resize、归一化、tokenization就不要放到训练循环里做。训练循环里只做最轻量的操作。from torch.utils.data import DataLoader dataloader DataLoader( dataset, batch_size32, shuffleTrue, num_workers8, # 根据CPU核心数调整 pin_memoryTrue, # 加速GPU传输 prefetch_factor4, # 每个worker预取batch数 persistent_workersTrue # 避免每个epoch重新创建worker )提示num_workers不是越大越好。设太大反而会因为进程切换开销导致性能下降。一般从CPU核心数的一半开始试逐步调整。4. 模型层从实验代码到可维护的训练系统4.1 训练代码的组织方式在notebook里写训练代码做实验没问题但要把实验代码变成可维护的训练系统需要做一些结构调整。核心思路是把训练过程拆成可配置的组件模型、优化器、损失函数、数据加载器、训练循环每个组件都有明确的接口。这样替换其中一个组件的时候不需要改动其他部分。class Trainer: def __init__(self, model, optimizer, loss_fn, train_loader, val_loader, config): self.model model self.optimizer optimizer self.loss_fn loss_fn self.train_loader train_loader self.val_loader val_loader self.config config self.device torch.device(config[device]) self.model.to(self.device) def train_epoch(self): self.model.train() total_loss 0 for batch in self.train_loader: inputs, targets batch inputs inputs.to(self.device) targets targets.to(self.device) self.optimizer.zero_grad() outputs self.model(inputs) loss self.loss_fn(outputs, targets) loss.backward() self.optimizer.step() total_loss loss.item() return total_loss / len(self.train_loader) def validate(self): self.model.eval() total_loss 0 with torch.no_grad(): for batch in self.val_loader: inputs, targets batch inputs inputs.to(self.device) targets targets.to(self.device) outputs self.model(inputs) loss self.loss_fn(outputs, targets) total_loss loss.item() return total_loss / len(self.val_loader) def fit(self): for epoch in range(self.config[epochs]): train_loss self.train_epoch() val_loss self.validate() print(fEpoch {epoch1}: train_loss{train_loss:.4f}, val_loss{val_loss:.4f}) # 保存checkpoint、记录日志等这种写法看起来比notebook里直接写循环要麻烦一点但好处是可测试、可复用、可扩展。比如你想换一个优化器只需要在初始化的时候传入不同的optimizer对象Trainer本身的代码不用动。4.2 实验跟踪别靠记忆和Excel实验跟踪这件事刚开始做的时候觉得没必要等到实验多了之后就会发现没有系统化的跟踪根本记不住哪个配置对应哪个结果。最轻量的做法是用一个简单的日志文件import json from datetime import datetime def log_experiment(config, metrics, notes): record { timestamp: datetime.now().isoformat(), config: config, metrics: metrics, notes: notes } with open(experiments.jsonl, a) as f: f.write(json.dumps(record) \n)每次实验结束后调用一下把配置和结果记下来。时间长了这个文件就是你的实验历史。需要对比的时候用pandas读进来分析就行。如果团队规模大一点可以考虑用MLflow或者Weights Biases。但核心原则是一样的每次实验都要有记录记录要包含配置和结果。4.3 模型版本管理模型版本管理不只是给模型文件打个标签那么简单。一个完整的模型版本应该包含模型权重文件具体的参数模型结构定义代码层面的结构训练配置超参数、数据版本评估结果在验证集和测试集上的表现依赖环境Python版本、库版本我一般会在保存模型的时候同时保存一个元数据文件def save_model_with_metadata(model, path, metadata): 保存模型和元数据 # 保存权重 torch.save(model.state_dict(), f{path}/model.pt) # 保存元数据 with open(f{path}/metadata.json, w) as f: json.dump(metadata, f, indent2) # 保存模型结构定义如果有的话 # 或者记录模型类的导入路径元数据里至少要有训练时间、数据版本、关键超参数、评估指标。这样加载模型的时候能清楚地知道这个模型是怎么来的。注意不要只保存state_dict而不保存模型结构。如果模型结构变了光有权重文件是加载不出来的。要么保存完整的模型对象要么确保模型结构的代码也在版本控制里。5. 服务层把模型变成能用的接口5.1 推理服务的两种基本形态模型训练出来之后要能用起来就需要一个服务层。推理服务基本上分两种形态在线服务和离线批处理。在线服务适合实时性要求高的场景比如用户上传一张图片系统立刻返回分类结果。这种服务通常用HTTP或者gRPC接口暴露要求低延迟、高并发。离线批处理适合数据量大、实时性要求不高的场景比如每天凌晨跑一遍全量数据的预测。这种服务更关注吞吐量对单次请求的延迟不敏感。两种形态的技术选型不一样维度在线服务离线批处理延迟要求毫秒级分钟级可接受吞吐要求中等高框架选择FastAPI、TritonSpark、Ray部署方式容器化、自动扩缩定时任务、队列错误处理快速失败、降级重试、断点续跑5.2 用FastAPI搭建推理接口FastAPI是目前Python生态里做推理服务比较顺手的选择。性能不错异步支持好自动生成API文档。一个基本的推理服务大概长这样from fastapi import FastAPI, HTTPException from pydantic import BaseModel import torch import numpy as np app FastAPI() # 全局加载模型 model None class PredictRequest(BaseModel): features: list[float] class PredictResponse(BaseModel): prediction: float confidence: float app.on_event(startup) def load_model(): global model model torch.load(models/production/model.pt, map_locationcpu) model.eval() app.post(/predict, response_modelPredictResponse) async def predict(request: PredictRequest): try: features np.array(request.features, dtypenp.float32) tensor torch.from_numpy(features).unsqueeze(0) with torch.no_grad(): output model(tensor) prob torch.softmax(output, dim1) confidence, prediction torch.max(prob, dim1) return PredictResponse( predictionprediction.item(), confidenceconfidence.item() ) except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.get(/health) async def health(): return {status: ok, model_loaded: model is not None}这个服务有几个关键点模型在启动时加载而不是每次请求都加载。加载模型是耗时操作放在请求处理里会严重拖慢响应。model.eval()必须调用否则dropout和batch norm的行为会不对。torch.no_grad()关闭梯度计算减少内存占用和计算量。健康检查接口方便负载均衡器和监控系统判断服务状态。5.3 性能优化的几个实用手段推理服务的性能优化效果最明显的几个手段批处理Batching把多个请求合并成一个batch一起推理能显著提升GPU利用率。但要注意延迟和吞吐的权衡——等batch凑满会增加延迟不等又浪费算力。常见的做法是设置一个最大等待时间超时或者凑满就发。模型量化把FP32的模型转成FP16或者INT8推理速度能提升不少精度损失通常在可接受范围内。PyTorch支持动态量化和静态量化具体用哪种取决于模型结构。ONNX Runtime把PyTorch模型导出成ONNX格式用ONNX Runtime推理在很多场景下比原生PyTorch快。特别是CPU推理场景ONNX Runtime的优化做得比较好。# 导出ONNX模型 dummy_input torch.randn(1, 3, 224, 224) torch.onnx.export( model, dummy_input, model.onnx, input_names[input], output_names[output], dynamic_axes{input: {0: batch_size}, output: {0: batch_size}} )提示量化之前一定要在验证集上评估精度损失。有些模型对量化很敏感INT8之后精度掉得厉害那就得不偿失了。6. 监控与迭代上线只是开始6.1 推理服务的监控指标服务上线之后如果没有监控就等于在黑暗中开车。最基本的监控指标包括延迟P50、P95、P99分位数。只看平均值会被长尾请求掩盖问题。吞吐每秒处理的请求数。错误率HTTP 5xx错误的比例。资源利用率CPU、GPU、内存的使用情况。模型指标预测结果的分布变化。这些指标可以用Prometheus采集Grafana展示。如果不想搭这么重的监控系统至少也要把日志记好定期检查。import time from prometheus_client import Histogram, Counter REQUEST_LATENCY Histogram( inference_latency_seconds, Inference latency, buckets[0.01, 0.05, 0.1, 0.5, 1.0, 5.0] ) REQUEST_COUNT Counter( inference_requests_total, Total inference requests, [status] ) app.post(/predict) async def predict(request: PredictRequest): start time.time() try: # ... 推理逻辑 ... REQUEST_COUNT.labels(statussuccess).inc() return result except Exception as e: REQUEST_COUNT.labels(statuserror).inc() raise finally: REQUEST_LATENCY.observe(time.time() - start)6.2 数据漂移与模型退化模型上线之后效果不是一成不变的。数据分布会变用户行为会变模型的效果会慢慢下降。这就是数据漂移和概念漂移。数据漂移指的是输入数据的分布变了。比如一个推荐模型训练的时候用户主要是年轻人上线半年后中老年用户比例大幅增加输入分布就变了。概念漂移指的是输入和输出之间的关系变了。比如一个信用评分模型经济环境变化之后同样的收入水平对应的违约风险不一样了。检测漂移的方法有很多最简单的是监控预测结果的分布。如果预测结果的分布和训练时相比发生了明显变化那就可能有问题。更严谨的做法是定期用新数据评估模型看指标是否下降。def detect_drift(reference_data, current_data, threshold0.05): 简单的分布漂移检测 from scipy.stats import ks_2samp drift_scores {} for col in reference_data.columns: stat, p_value ks_2samp(reference_data[col], current_data[col]) drift_scores[col] { statistic: stat, p_value: p_value, drift_detected: p_value threshold } return drift_scores6.3 持续迭代的工程化流程AI系统的迭代和传统软件不一样。传统软件的迭代是改代码、测试、上线。AI系统的迭代还多了一个维度数据和模型。一个完整的迭代流程大概是这样收集新数据从线上服务收集新的输入和反馈。数据标注如果需要对新数据进行标注。重新训练用新数据加上历史数据重新训练模型。离线评估在验证集和测试集上评估新模型。影子模式新模型和旧模型同时运行对比效果。灰度发布逐步把流量切到新模型。全量上线确认没问题后全量切换。回滚准备保留旧模型出问题能快速回滚。这个流程里影子模式和灰度发布是最容易被忽略的。很多人训练完新模型评估指标好就直接全量上线结果出了线上问题才发现。影子模式可以让新模型在不影响用户的情况下跑一段时间积累真实场景的表现数据。注意回滚不是失败而是工程成熟度的体现。没有回滚方案的上线是在赌运气。7. 一些踩过的坑和实际体会7.1 环境不一致导致的“在我机器上能跑”这个问题太常见了。开发环境是Python 3.9生产环境是3.8某个库的版本不兼容服务直接起不来。或者开发机有GPU生产环境只有CPU代码里写死了.cuda()上线就报错。解决办法就是容器化。Dockerfile把环境定义清楚开发、测试、生产用同一个镜像。代码里不要写死设备用配置来控制。FROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [uvicorn, serving.api.main:app, --host, 0.0.0.0, --port, 8000]# 不要这样写 model.cuda() # 这样写 device torch.device(config.get(device, cpu)) model.to(device)7.2 日志太多和日志太少都不行日志太少出了问题不知道从哪里查。日志太多关键信息被淹没磁盘也扛不住。我的经验是分层记录DEBUG详细的中间结果只在开发环境开。INFO关键流程节点比如请求开始、请求结束、模型加载完成。WARNING不影响功能但需要注意的情况比如重试、降级。ERROR影响功能的错误需要人工介入。生产环境一般只开INFO及以上。但要注意ERROR日志里不要包含敏感信息比如用户数据、密钥等。7.3 模型文件的管理模型文件通常很大不适合直接放在Git仓库里。但模型又是核心资产不能随便丢。常见的做法是用对象存储比如S3兼容的存储来保存模型文件Git仓库里只保存模型的元数据和下载脚本。每次训练完把模型上传到对象存储打上版本标签然后在代码里通过版本号来引用。import boto3 def upload_model(local_path, bucket, key): s3 boto3.client(s3) s3.upload_file(local_path, bucket, key) return fs3://{bucket}/{key} def download_model(bucket, key, local_path): s3 boto3.client(s3) s3.download_file(bucket, key, local_path) return local_path这样既避免了Git仓库膨胀又能保证模型文件的可追溯性。7.4 关于“从零开始”的一点个人看法“ai-engineering-from-scratch”这个说法容易让人误以为要从最底层开始造轮子。但实际上AI工程的核心能力不是造轮子而是把现有的轮子组装成一辆能跑的车。你不需要自己实现一个Transformer但你需要知道Transformer的输入输出是什么推理的时候显存占用大概多少batch size能开到多大。你不需要自己写一个Web框架但你需要知道怎么把模型包装成一个HTTP接口怎么处理并发请求怎么做超时控制。所以我的建议是先跑通一个最小的端到端流程然后再逐步深入每个环节。不要一开始就追求完美先让系统能跑起来哪怕很粗糙。跑起来之后你自然会发现哪里是瓶颈哪里需要优化。这种问题驱动式的学习比从头到尾啃文档效率高得多。我在最开始做AI工程的时候花了很多时间在“准备”上——研究各种工具、对比各种方案、搭建各种基础设施。后来发现真正让我进步最快的是把一个模型从训练到上线完整走一遍。走一遍之后所有零散的知识点就串起来了哪些重要哪些不重要也清楚了。所以如果你正在这个方向上摸索我的建议就一句话别想太多先跑起来。