1. 从零搭建AI工程能力为什么大多数人卡在第一步聊到AI工程很多人第一反应是“调包”——装个transformers跑个demo觉得就算入门了。但真到了要自己搭一套能用的系统比如做一个检索增强生成的知识库问答、训练一个小规模的分类模型、或者把模型部署成API给前端调用立刻就卡住了。不是环境配不通就是显存爆了要么就是推理速度慢到没法用。这个项目标题“ai-engineering-from-scratch”之所以值得拿出来讲恰恰是因为它戳中了一个普遍痛点AI工程不是算法研究它是一门关于“把模型跑起来、跑稳、跑快”的手艺。我自己带过不少刚转方向的同学发现一个规律凡是能顺利从零把AI工程链路走通的人都不是靠背API文档而是靠理解每一层在干什么。比如为什么推理要用半精度为什么batch size不能随便调大为什么向量检索要分索引构建和查询两个阶段这些问题背后都有明确的工程逻辑搞懂了换任何框架都能快速上手。这篇文章面向的是有一定Python基础、想系统补齐AI工程实操能力的开发者。不管你是做后端想往AI方向靠还是做数据分析想深入模型落地或者纯粹想自己搭一套本地AI应用下面的内容都会围绕“从零构建”这个核心把环境、数据、训练、推理、部署这条链路上的关键决策点拆开讲清楚。我不会只给代码更会解释为什么这么选、不这么选会出什么问题。2. 环境与依赖别让版本冲突吃掉你的第一周2.1 为什么建议用conda而不是裸pip从零做AI工程第一个坑几乎永远是环境。Python生态里AI相关的包依赖关系极其复杂torch、cuda、numpy、transformers之间经常出现版本打架。裸pip装到最后大概率会遇到“numpy版本不兼容”或者“torch和cuda版本对不上”这类问题。我的习惯是用conda建独立环境再用pip装AI框架这样能把Python版本和底层C库的依赖交给conda管把纯Python包交给pip管两边各司其职。具体操作上先建一个指定Python版本的环境conda create -n ai-eng python3.10 -y conda activate ai-eng为什么选3.10而不是最新的3.12因为截至我写这篇内容时主流AI框架对3.10的支持最成熟很多预编译的wheel包只覆盖到3.10和3.11。选3.12会经常遇到“没有预编译包需要自己编译”的情况编译一次torch可能要半小时以上还容易失败。这是实打实的时间成本。2.2 CUDA版本的选择逻辑如果你有NVIDIA显卡CUDA版本决定了你能用哪个版本的torch。这里有个简单的判断方法先看显卡驱动支持的最高CUDA版本用nvidia-smi命令查看右上角的“CUDA Version”。然后去PyTorch官网看对应的安装命令。比如驱动显示支持12.1那你可以装cu121版本的torch。但这里有个经验不要盲目追最新CUDA。新版本CUDA对应的torch往往刚发布不久生态里的其他库比如某些向量检索库、量化工具可能还没跟上。我一般会选比最高支持版本低一档的CUDA比如驱动支持12.1我就装cu118的torch稳定性明显更好。安装命令示例pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118装完之后一定要验证import torch print(torch.__version__) print(torch.cuda.is_available()) print(torch.cuda.get_device_name(0))如果cuda.is_available()返回False别急着往下走先解决这个问题。常见原因有三个驱动版本太低、torch装成了CPU版本、conda环境里混入了系统Python的路径。逐个排查通常十分钟内能定位。2.3 那些容易被忽略的基础依赖除了torch还有几个包建议一开始就装好省得后面反复折腾sentencepiece很多tokenizer依赖它不装会在加载模型时报错。protobuf版本要控制好太新或太旧都会导致transformers加载失败。我一般锁定在3.20.x。accelerate做多卡推理或混合精度时必备即使单卡也建议装上方便后续扩展。datasets处理训练数据时比手写DataLoader省事得多。这些包看起来不起眼但缺一个就可能让你在某个深夜对着报错信息发呆。提前装好后面顺畅很多。3. 数据管道AI工程里最不性感但最要命的部分3.1 数据清洗不是可选项是必选项很多人从零做AI项目拿到数据就想直接喂给模型。结果训练loss不下降或者推理效果一塌糊涂。十有八九是数据没洗干净。AI工程里有个残酷的现实模型再强也救不了脏数据。我见过太多案例花三天调模型结构不如花三小时把数据里的乱码、重复、标注错误清理一遍。以文本数据为例至少要过这几道工序去重完全重复的样本直接删近似重复的用MinHash或SimHash检测。重复数据会让模型过拟合到特定样本上。长度过滤太短的样本比如少于10个字符通常没信息量太长的超过模型最大长度要截断或分段。异常字符清理HTML标签、控制字符、乱码统一处理掉。语言过滤如果做中文任务用langdetect把非中文样本筛掉。这些步骤写成一个预处理脚本跑一次可能只要几分钟但能省下后面几小时的调试时间。3.2 构建可复用的数据加载流程数据清洗完之后别急着写训练循环。先设计一个可复用的数据加载模块。我的习惯是用datasets库的Dataset和DataLoader组合把tokenization、padding、batch组装都封装进去。from datasets import Dataset from transformers import AutoTokenizer tokenizer AutoTokenizer.from_pretrained(bert-base-chinese) def tokenize_fn(examples): return tokenizer( examples[text], truncationTrue, max_length512, paddingmax_length ) dataset Dataset.from_dict({text: texts, label: labels}) dataset dataset.map(tokenize_fn, batchedTrue) dataset.set_format(typetorch, columns[input_ids, attention_mask, label])这里有个细节paddingmax_length会让所有样本补齐到512浪费显存。更好的做法是动态padding在DataLoader的collate_fn里按batch内最长样本补齐。这样能省30%以上的显存尤其是文本长度差异大的时候效果明显。3.3 数据版本管理别让自己后悔做AI工程数据会反复迭代。今天清洗一版明天补充标注后天发现标注有错要回滚。如果没有版本管理很快就会乱套。我的做法是每次数据处理都输出到一个带时间戳和哈希值的目录并在目录里放一个manifest文件记录处理参数。data/ processed_20240115_a3f2c1/ train.jsonl valid.jsonl manifest.jsonmanifest里记录原始数据路径、清洗规则、样本数量、处理时间。这样任何时候都能追溯某一版数据是怎么来的。这个习惯看起来麻烦但当你需要复现实验结果时会感谢自己当初多做了这一步。4. 模型训练与微调从“能跑”到“跑得好”的关键决策4.1 全量微调还是参数高效微调从零做AI工程迟早要面对微调。全量微调是把模型所有参数都更新效果好但显存需求大。参数高效微调比如LoRA只训练一小部分额外参数显存需求小效果在多数任务上接近全量微调。怎么选我的判断标准很简单场景推荐方案理由显存小于24GLoRA全量微调7B模型需要至少60G显存任务与预训练差异大全量微调LoRA表达能力有限差异大时效果打折需要快速迭代多版LoRA每个任务只存几十M的adapter切换方便数据量超过10万条全量微调数据充足时全量微调上限更高LoRA的配置里r秩和alpha缩放系数是两个关键参数。经验值是r8或16alpha16或32。r越大表达能力越强但参数量也越大。一般任务r8就够复杂任务可以试r16。4.2 学习率与warmup的实操设定学习率是训练里最敏感的参数。太大loss震荡太小收敛慢。对于微调任务我通常从2e-5开始试。如果是LoRA因为只训练少量参数学习率可以放大到1e-4到3e-4。warmup比例建议设成总步数的5%到10%。比如总共训练1000步warmup设50到100步。这样能让模型在初期稳定地进入训练状态避免一开始就大步更新导致崩溃。from transformers import TrainingArguments training_args TrainingArguments( output_dir./output, learning_rate2e-5, warmup_ratio0.1, num_train_epochs3, per_device_train_batch_size8, gradient_accumulation_steps4, fp16True, logging_steps50, save_strategyepoch, evaluation_strategyepoch )这里gradient_accumulation_steps4配合batch_size8等效batch size是32。为什么要用梯度累积而不是直接调大batch size因为显存不够。梯度累积是用时间换空间多跑几次前向传播再统一更新效果和大队列差不多但显存占用小很多。4.3 训练过程中的监控与早停训练不是跑完就行得盯着。我一般会监控三个指标训练loss、验证loss、验证集上的任务指标比如准确率或F1。如果训练loss持续下降但验证loss开始上升就是过拟合了该早停。早停的实现很简单用EarlyStoppingCallbackfrom transformers import EarlyStoppingCallback callbacks [EarlyStoppingCallback(early_stopping_patience2)]patience2表示验证指标连续两轮不提升就停。这个值别设太大否则浪费算力也别设太小否则可能停得太早。2到3是比较稳妥的选择。另外训练日志一定要存下来。我习惯把logging_dir指向一个固定目录然后用tensorboard看曲线。曲线比数字直观得多一眼就能看出是欠拟合还是过拟合。5. 推理优化让模型真正能用的最后一公里5.1 半精度推理与显存占用的关系模型训练完部署推理时第一个要做的就是开半精度。FP16推理相比FP32显存占用直接减半速度通常还能提升30%以上。对于7B模型FP32需要约28G显存FP16只要14G左右一张24G的卡就能跑起来。model AutoModelForCausalLM.from_pretrained( model_path, torch_dtypetorch.float16, device_mapauto )device_mapauto会让accelerate自动把模型层分配到可用设备上。如果显存不够它还会自动把部分层放到CPU上虽然速度慢但至少能跑。这个功能在多卡或显存紧张时特别实用。但半精度有个坑某些操作在FP16下会溢出导致输出NaN。如果遇到这种情况可以试试BF16如果显卡支持。BF16的动态范围比FP16大不容易溢出但精度略低。A100及以上的卡支持BF16消费级卡里30系和40系也支持。5.2 批处理与动态填充的配合推理时批处理能大幅提升吞吐量。但不同请求的输入长度不同如果统一padding到最大长度短请求会浪费大量计算。解决办法是按长度分桶把长度相近的请求放在同一个batch里这样padding浪费最少。实现上可以在请求队列里维护多个桶每个桶对应一个长度区间。新请求进来时放到对应桶里桶满了就触发一次推理。这样能把GPU利用率拉高不少。实测下来分桶批处理比单条推理吞吐量能提升5到10倍。5.3 量化用精度换速度的取舍如果显存实在不够或者想进一步提速可以考虑量化。8-bit量化能把显存再降一半4-bit量化降得更多。但量化会损失精度具体损失多少取决于任务和量化方法。from transformers import BitsAndBytesConfig bnb_config BitsAndBytesConfig( load_in_4bitTrue, bnb_4bit_compute_dtypetorch.float16, bnb_4bit_quant_typenf4 ) model AutoModelForCausalLM.from_pretrained( model_path, quantization_configbnb_config, device_mapauto )nf4是一种针对正态分布权重优化的4-bit量化格式效果比普通的int4好。实测在文本生成任务上4-bit量化的效果损失通常在可接受范围内但推理速度提升明显。不过要注意量化后的模型不能直接继续训练只能用于推理。6. 服务化与部署把模型变成别人能调用的接口6.1 用FastAPI搭一个最小可用的推理服务模型跑通之后下一步是把它变成API。FastAPI是我最常用的选择轻量、异步支持好、自动生成文档。一个最小的推理服务大概长这样from fastapi import FastAPI from pydantic import BaseModel import torch app FastAPI() class Request(BaseModel): text: str max_length: int 128 class Response(BaseModel): result: str app.post(/predict, response_modelResponse) async def predict(req: Request): inputs tokenizer(req.text, return_tensorspt).to(cuda) with torch.no_grad(): outputs model.generate(**inputs, max_lengthreq.max_length) result tokenizer.decode(outputs[0], skip_special_tokensTrue) return Response(resultresult)这个服务能跑但有几个问题没有并发控制、没有超时处理、模型加载在启动时完成所以第一次请求会慢。生产环境还需要加请求队列、限流、健康检查等。但作为起步这个结构已经够用了。6.2 模型加载策略启动时加载还是懒加载模型加载有两种策略启动时加载和首次请求时懒加载。启动时加载的好处是第一次请求就很快坏处是服务启动慢而且如果模型很大启动期间服务不可用。懒加载则相反启动快但第一次请求会卡很久。我的建议是启动时加载但加一个健康检查接口。在模型加载完成之前健康检查返回“not ready”负载均衡器就不会把流量导进来。这样既保证了请求速度又避免了服务不可用。model_ready False app.on_event(startup) async def load_model(): global model, tokenizer, model_ready model AutoModelForCausalLM.from_pretrained(...) tokenizer AutoTokenizer.from_pretrained(...) model_ready True app.get(/health) async def health(): if model_ready: return {status: ok} return {status: loading}, 5036.3 并发与显存管理的平衡推理服务最怕的是并发请求把显存打爆。一个请求进来如果模型正在处理第二个请求要么排队要么直接失败。我的做法是用一个信号量控制并发数同时限制最大batch size。import asyncio semaphore asyncio.Semaphore(4) # 最多4个并发 app.post(/predict) async def predict(req: Request): async with semaphore: # 推理逻辑 ...并发数设多少取决于显存和模型大小。7B模型FP16推理一张24G的卡大概能同时处理4到8个请求。设太大容易OOM设太小吞吐量上不去。建议从4开始试观察显存占用再调整。另外推理完的中间变量要及时释放。Python的垃圾回收不是实时的可以用torch.cuda.empty_cache()手动清理但别频繁调用否则反而影响性能。一般在一个batch处理完后调用一次就够了。7. 那些只有踩过才知道的工程细节7.1 随机种子不固定实验白做AI工程里有个反直觉的事实同样的代码跑两次结果可能不一样。原因是随机种子没固定。PyTorch、numpy、Python内置的random都有自己的种子要全部固定import random import numpy as np import torch def set_seed(seed42): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) torch.backends.cudnn.deterministic True torch.backends.cudnn.benchmark Falsecudnn.deterministicTrue会让卷积运算确定化但会牺牲一点速度。做实验对比时建议打开上线时可以关掉换性能。7.2 日志要记到文件别只看终端终端里的日志一关就没了。训练跑一晚上第二天想看loss曲线发现终端已经关了什么都查不到。我的习惯是所有日志同时输出到终端和文件并且按时间戳命名。import logging logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(ftrain_{timestamp}.log), logging.StreamHandler() ] )这样任何时候都能翻回去看。而且日志文件可以配合grep快速定位问题比在终端里翻滚动条高效得多。7.3 显存泄漏的排查思路推理服务跑久了显存越来越大最后OOM这是典型的显存泄漏。排查思路是先确认是不是缓存没释放再确认是不是有全局变量持有tensor引用。第一步在推理循环里加显存打印print(torch.cuda.memory_allocated() / 1024**3, GB) print(torch.cuda.memory_reserved() / 1024**3, GB)allocated是实际使用的显存reserved是PyTorch向系统申请的显存。如果reserved远大于allocated说明有缓存没释放可以用empty_cache()清理。如果allocated持续增长那就是有引用没释放检查是不是把tensor存到了全局列表里。第二步用gc.collect()强制垃圾回收看显存是否下降。如果下降说明是Python对象引用问题如果不降可能是CUDA层面的问题需要更深入排查。7.4 模型版本与代码版本的对应关系上线之后模型和代码会各自迭代。如果不记录对应关系回滚时就会懵这个模型是配哪版代码的我的做法是在模型保存目录里放一个version.json记录代码的git commit、训练参数、数据版本。{ code_commit: a3f2c1d, data_version: processed_20240115_a3f2c1, training_args: {...}, metrics: {...} }这样任何时候都能追溯。回滚时先看version.json找到对应的代码commitcheckout回去再加载对应的模型整个链路就清晰了。8. 从零到一之后下一步往哪走把上面这条链路走通你就算真正入了AI工程的门。但入门之后还有很多方向可以深入。比如推理加速可以用TensorRT或ONNX Runtime能把延迟再降一半服务化可以用Triton Inference Server支持多模型、多框架、动态批处理数据管道可以用Airflow或Prefect做编排实现自动化。但我的建议是别急着堆工具。先把一条最简单的链路跑稳理解每个环节的瓶颈在哪再针对性地引入工具。我见过太多人一上来就搭了一套复杂的MLOps平台结果模型本身效果都没调好工具反而成了负担。另外AI工程和传统后端工程有个很大的区别不确定性更高。同样的代码换个数据分布效果可能差很多同样的模型换个batch size速度可能差一倍。所以做AI工程实验记录和可复现性比什么都重要。每次改动只动一个变量记录结果逐步迭代。这个习惯坚持下来你会发现自己的效率比那些“凭感觉调参”的人高出一大截。最后分享一个我自己的小技巧建一个自己的“踩坑笔记”。每次遇到报错、性能问题、效果异常解决之后花两分钟记下来问题现象、排查过程、根因、解决方案。积累几个月这就是你最宝贵的工程经验库。下次再遇到类似问题翻笔记比搜网页快得多。AI工程这个领域很多问题没有标准答案靠的就是一次次踩坑积累出来的直觉。