1. 从零搭建AI工程能力一个项目标题背后的完整学习路径第一次看到ai-engineering-from-scratch这个标题我的直觉是这又是一个从入门到放弃的教程合集。但仔细琢磨了一下这个标题其实精准地戳中了一个被大多数人忽略的痛点——市面上讲AI的内容要么是纯理论推导要么是调包侠式的API调用真正教你从零把AI工程这件事搭起来的内容少得可怜。我自己在这个方向上摸索了挺长时间踩过的坑不算少。这篇文章想做的事情很简单把这个标题背后的核心领域、技术栈、实操路径、常见陷阱全部拆开揉碎讲清楚。不管你是刚转行想入局AI工程的开发者还是已经在做相关项目但总觉得地基不牢的工程师都能从里面找到可以直接抄作业的东西。先说清楚这个标题到底在讲什么。AI工程不是AI研究也不是机器学习算法推导。它指的是把AI模型能力变成可用的产品和服务这中间涉及的一整套工程实践。包括数据处理、模型训练与微调、推理服务部署、性能优化、监控运维等等。而from scratch强调的是从最底层开始不依赖高度封装的平台自己动手把每一层搭起来。为什么这件事值得做因为当你只用过现成的API和平台时遇到问题你根本不知道从哪里排查。推理延迟高了是模型的问题还是服务框架的问题显存爆了是batch size设大了还是内存泄漏这些问题的答案只有真正从零搭过一遍的人才能快速定位。2. 整体设计思路为什么选择从零搭建而不是直接调API2.1 从零搭建的核心价值在哪里很多人会问现在各大平台都有现成的模型API一键调用就能出结果为什么还要费劲从零搭这个问题我在不同的场合被问过很多次我的回答一直是调API和从零搭建解决的是完全不同层次的问题。调API适合快速验证想法、做原型、赶项目进度。但如果你要做的产品对延迟、成本、数据隐私有要求或者你需要对模型行为做深度定制那API方案很快就会碰到天花板。举个很实际的例子你做一个面向企业的文档问答系统客户要求数据绝对不能出内网这时候你怎么办只能自己部署模型。而自己部署模型就涉及模型选型、量化、推理框架选择、并发处理、显存管理这一整套工程问题。从零搭建的另一个价值在于它强迫你理解每一层的原理和边界。当你自己动手实现过一个简单的推理服务你就知道一个请求从进入到返回结果中间经过了哪些步骤每个步骤的瓶颈可能在哪里。这种理解是调API永远给不了你的。2.2 技术栈选型的底层逻辑从零搭建AI工程能力技术栈的选择非常关键。我的建议是分层来看层级推荐技术选择理由编程语言Python 少量C/RustPython生态最全性能瓶颈部分用底层语言补深度学习框架PyTorch动态图友好社区活跃调试方便推理框架ONNX Runtime / TensorRT / vLLM不同场景选不同后面详细说服务框架FastAPI / Triton Inference Server轻量选FastAPI生产级选Triton容器化Docker Docker Compose环境隔离部署标准化监控Prometheus Grafana指标采集和可视化的事实标准这个选型不是拍脑袋定的。Python作为主力语言是因为AI领域的库和工具几乎都优先支持Python。PyTorch而不是TensorFlow是因为调试体验好太多你可以在任意位置打断点看张量的值这对从零搭建的人来说太重要了。推理框架的选择要看具体场景如果你只是想把一个PyTorch模型快速部署起来ONNX Runtime够用了如果要追求极致性能TensorRT是更好的选择如果是大语言模型vLLM在吞吐量上有明显优势。2.3 学习路径的编排原则从零搭建AI工程能力最怕的是东一榔头西一棒子。我建议按照数据→模型→服务→运维这条主线来编排学习路径。先搞定数据处理因为这是所有AI工程的基础然后理解模型训练和微调的基本流程接着把模型变成可调用的服务最后补上监控和运维的能力。每个阶段都要有可运行的产出物。比如数据处理阶段产出应该是一个可复用的数据清洗和预处理脚本模型阶段产出是一个能跑通的训练或微调脚本服务阶段产出是一个能接收请求并返回结果的API运维阶段产出是一套基本的监控面板和告警规则。这种每步都有产出的方式能让你始终保持正反馈不至于学到一半就放弃了。3. 核心细节解析从零搭建必须掌握的四个关键环节3.1 环境搭建别小看这一步坑最多环境搭建听起来是最没技术含量的环节但实际上它是新手翻车最多的地方。我见过太多人卡在CUDA版本不匹配、PyTorch装不上、Docker权限报错这些问题上折腾一两天都进不了正题。我的建议是用Docker来做环境隔离不要在宿主机上直接装一堆东西。具体做法是先装好NVIDIA驱动和Docker然后拉一个PyTorch的官方镜像作为基础镜像在容器里面做开发。这样即使你把环境搞乱了删掉容器重新拉一个就行不会影响宿主机。FROM pytorch/pytorch:2.1.0-cuda12.1-cudnn8-devel RUN pip install --no-cache-dir \ fastapi0.104.1 \ uvicorn0.24.0 \ onnxruntime-gpu1.16.3 \ transformers4.35.2 \ prometheus-client0.19.0 WORKDIR /workspace COPY . /workspace这个Dockerfile是我常用的基础模板包含了推理服务需要的大部分依赖。注意PyTorch镜像的tag要和你宿主机的CUDA驱动版本匹配不匹配的话容器里跑不起来GPU。查看宿主机CUDA版本用nvidia-smi命令右上角那个CUDA Version就是你要对齐的版本。注意Docker容器里能看到的CUDA版本和宿主机驱动版本是两回事。容器里的CUDA toolkit版本由镜像决定但实际能用的CUDA版本上限由宿主机驱动决定。如果镜像里的CUDA版本高于驱动支持的版本GPU就用不了。3.2 数据处理AI工程的地基数据处理这块很多人觉得就是把数据读进来、洗干净、喂给模型就完事了。但实际做起来数据处理往往占整个项目70%以上的工作量。我总结下来数据处理要关注三个核心问题格式统一、质量过滤、高效加载。格式统一的意思是不管你原始数据是CSV、JSON、还是数据库里的记录最终都要转成模型能吃的格式。对于文本任务通常是tokenize之后的input_ids和attention_mask对于图像任务是归一化之后的张量。这一步的关键是写一个可复用的预处理管道而不是每次手动处理。质量过滤是很多人忽略的一步。原始数据里往往有大量噪声重复样本、过短或过长的文本、编码错误的字符、标注错误的样本。如果不做过滤模型学到的就是垃圾。我的经验是至少要做去重、长度过滤、字符集检查这三件事。高效加载这块PyTorch的DataLoader是标准方案但有几个参数需要特别注意。num_workers设成CPU核心数的一半到三分之二比较合适设太大了反而会因为进程切换开销导致变慢。pin_memoryTrue在GPU训练时能加速数据传输。prefetch_factor控制每个worker预取的batch数量默认是2如果IO是瓶颈可以适当调大。from torch.utils.data import DataLoader, Dataset from transformers import AutoTokenizer class TextDataset(Dataset): def __init__(self, texts, labels, tokenizer, max_len512): self.texts texts self.labels labels self.tokenizer tokenizer self.max_len max_len def __len__(self): return len(self.texts) def __getitem__(self, idx): encoding self.tokenizer( self.texts[idx], max_lengthself.max_len, paddingmax_length, truncationTrue, return_tensorspt ) return { input_ids: encoding[input_ids].squeeze(), attention_mask: encoding[attention_mask].squeeze(), label: torch.tensor(self.labels[idx]) } tokenizer AutoTokenizer.from_pretrained(bert-base-chinese) dataset TextDataset(texts, labels, tokenizer) loader DataLoader( dataset, batch_size32, shuffleTrue, num_workers4, pin_memoryTrue, prefetch_factor4 )这段代码看起来简单但里面有几个细节值得说。paddingmax_length会统一padding到最大长度这样batch内不需要动态padding实现简单但浪费计算资源。如果追求效率可以用paddingTrue配合DataCollator做动态padding每个batch按实际最长样本padding。truncationTrue是必须的不然超长文本会直接报错。3.3 模型训练与微调从能跑到跑得好模型训练这块从零搭建的话我建议先从微调预训练模型开始而不是从头训练。从头训练一个可用的模型需要的算力和数据量不是个人能承受的。微调则相对友好一张消费级显卡就能做很多有意思的事情。微调的核心是搞清楚三件事冻结哪些层、学习率设多少、训练多久。冻结层的选择取决于你的数据量和任务相似度。数据量小、任务和预训练任务相似就多冻结一些层只训练最后的分类头数据量大、任务差异大就解冻更多层甚至全部解冻。学习率是微调中最关键的参数。我的经验值是全量微调用1e-5到5e-5只训练分类头用1e-3到1e-4。用AdamW优化器配合线性预热和余弦退火的学习率调度效果通常比固定学习率好。from transformers import AutoModelForSequenceClassification, AdamW, get_cosine_schedule_with_warmup import torch model AutoModelForSequenceClassification.from_pretrained( bert-base-chinese, num_labelsnum_classes ) # 分层设置学习率底层小顶层大 optimizer_grouped_parameters [ {params: model.bert.embeddings.parameters(), lr: 1e-5}, {params: model.bert.encoder.layer[:6].parameters(), lr: 2e-5}, {params: model.bert.encoder.layer[6:].parameters(), lr: 5e-5}, {params: model.classifier.parameters(), lr: 1e-4}, ] optimizer AdamW(optimizer_grouped_parameters, weight_decay0.01) total_steps len(loader) * num_epochs scheduler get_cosine_schedule_with_warmup( optimizer, num_warmup_stepsint(0.1 * total_steps), num_training_stepstotal_steps )分层学习率是我强烈推荐的一个技巧。底层学的是通用语言特征不需要大改顶层学的是任务相关特征需要大改。给不同层设置不同的学习率能让模型在保留通用能力的同时快速适应新任务。训练过程中要盯紧两个指标训练loss和验证集指标。如果训练loss一直降但验证集指标不涨说明过拟合了要加正则化或者早停。如果训练loss都不降说明学习率太小或者模型结构有问题。我一般会每训练一个epoch就在验证集上评估一次保存验证集指标最好的那个checkpoint。3.4 推理服务部署把模型变成产品模型训练好了下一步是把它变成别人能调用的服务。这一步的核心是接口设计、并发处理、性能优化。接口设计用FastAPI是最省事的方案。定义一个POST接口接收输入文本返回模型预测结果。注意要做好输入校验和错误处理不然线上出问题很难排查。from fastapi import FastAPI, HTTPException from pydantic import BaseModel import torch app FastAPI() class PredictRequest(BaseModel): text: str max_length: int 512 class PredictResponse(BaseModel): label: str confidence: float app.post(/predict, response_modelPredictResponse) async def predict(request: PredictRequest): if not request.text.strip(): raise HTTPException(status_code400, detail文本不能为空) inputs tokenizer( request.text, max_lengthrequest.max_length, truncationTrue, return_tensorspt ).to(device) with torch.no_grad(): outputs model(**inputs) probs torch.softmax(outputs.logits, dim-1) confidence, pred torch.max(probs, dim-1) return PredictResponse( labelid2label[pred.item()], confidenceconfidence.item() )并发处理是推理服务的难点。FastAPI默认是单线程处理请求的如果模型推理是同步阻塞的并发请求会排队。解决方案有两种一是用async配合线程池把推理放到线程池里执行二是用专门的推理服务器比如Triton它内置了动态批处理和并发执行的能力。性能优化这块最有效的手段是量化。把FP32的模型转成FP16或者INT8推理速度能提升2到4倍精度损失通常在可接受范围内。ONNX Runtime和TensorRT都支持量化具体选哪个看你的部署环境。# 导出ONNX模型 torch.onnx.export( model, dummy_input, model.onnx, input_names[input_ids, attention_mask], output_names[logits], dynamic_axes{ input_ids: {0: batch, 1: sequence}, attention_mask: {0: batch, 1: sequence}, logits: {0: batch} }, opset_version14 ) # 量化 from onnxruntime.quantization import quantize_dynamic, QuantType quantize_dynamic( model.onnx, model_quantized.onnx, weight_typeQuantType.QUInt8 )动态量化对Transformer类模型效果很好模型体积能缩小到原来的四分之一推理速度也有明显提升。但要注意量化后的模型精度需要重新评估如果精度下降太多就要考虑只量化部分层或者用量化感知训练来恢复精度。4. 实操过程从零到一搭建一个完整的AI服务4.1 项目结构规划动手之前先把项目结构定好后面才不会乱。我常用的结构是这样的ai-service/ ├── data/ │ ├── raw/ # 原始数据 │ └── processed/ # 处理后的数据 ├── src/ │ ├── data/ │ │ ├── dataset.py # Dataset定义 │ │ └── preprocess.py # 数据预处理 │ ├── model/ │ │ ├── train.py # 训练脚本 │ │ └── evaluate.py # 评估脚本 │ ├── service/ │ │ ├── app.py # FastAPI应用 │ │ └── schemas.py # 请求响应模型 │ └── utils/ │ ├── logger.py # 日志配置 │ └── metrics.py # 监控指标 ├── configs/ │ └── config.yaml # 配置文件 ├── tests/ │ └── test_service.py # 接口测试 ├── Dockerfile ├── docker-compose.yaml └── requirements.txt这个结构的好处是职责清晰。数据相关的代码在data目录模型相关的在model目录服务相关的在service目录。配置抽到configs里不同环境用不同的配置文件。测试单独放tests目录方便持续集成。4.2 数据准备与预处理实操假设我们要做一个中文文本分类服务第一步是准备数据。数据来源可能是CSV文件、数据库、或者爬取的网页。不管来源是什么统一转成text,label两列的格式。import pandas as pd from sklearn.model_selection import train_test_split def load_and_clean_data(path): df pd.read_csv(path) # 去重 df df.drop_duplicates(subset[text]) # 过滤过短文本 df df[df[text].str.len() 10] # 过滤过长文本 df df[df[text].str.len() 2000] # 去除空白字符 df[text] df[text].str.strip() # 过滤空文本 df df[df[text] ! ] return df df load_and_clean_data(data/raw/dataset.csv) train_df, val_df train_test_split(df, test_size0.2, random_state42, stratifydf[label])这里有几个细节值得注意。stratifydf[label]保证训练集和验证集的类别分布一致避免某个类别在验证集里完全没有样本。random_state42固定随机种子保证每次划分结果一样方便复现。长度过滤的阈值要根据实际数据分布来定可以先画个长度分布直方图看看。4.3 模型训练完整流程训练脚本要包含这几个部分加载数据、初始化模型、设置优化器和调度器、训练循环、验证、保存最佳模型。import torch from torch.utils.data import DataLoader from transformers import AutoTokenizer, AutoModelForSequenceClassification from transformers import AdamW, get_cosine_schedule_with_warmup from tqdm import tqdm def train(config): device torch.device(cuda if torch.cuda.is_available() else cpu) tokenizer AutoTokenizer.from_pretrained(config[model_name]) model AutoModelForSequenceClassification.from_pretrained( config[model_name], num_labelsconfig[num_labels] ).to(device) train_dataset TextDataset(train_texts, train_labels, tokenizer) val_dataset TextDataset(val_texts, val_labels, tokenizer) train_loader DataLoader(train_dataset, batch_sizeconfig[batch_size], shuffleTrue, num_workers4) val_loader DataLoader(val_dataset, batch_sizeconfig[batch_size], shuffleFalse, num_workers4) optimizer AdamW(model.parameters(), lrconfig[lr], weight_decay0.01) total_steps len(train_loader) * config[epochs] scheduler get_cosine_schedule_with_warmup( optimizer, num_warmup_stepsint(0.1 * total_steps), num_training_stepstotal_steps ) best_val_acc 0 for epoch in range(config[epochs]): model.train() total_loss 0 for batch in tqdm(train_loader, descfEpoch {epoch1}): input_ids batch[input_ids].to(device) attention_mask batch[attention_mask].to(device) labels batch[label].to(device) outputs model(input_idsinput_ids, attention_maskattention_mask, labelslabels) loss outputs.loss loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0) optimizer.step() scheduler.step() optimizer.zero_grad() total_loss loss.item() val_acc evaluate(model, val_loader, device) print(fEpoch {epoch1}: train_loss{total_loss/len(train_loader):.4f}, val_acc{val_acc:.4f}) if val_acc best_val_acc: best_val_acc val_acc torch.save(model.state_dict(), best_model.pt) print(f保存最佳模型验证准确率: {val_acc:.4f})clip_grad_norm_是防止梯度爆炸的常用手段max_norm设1.0是经验值。梯度裁剪在Transformer类模型训练中几乎是标配不加的话偶尔会遇到loss突然变成NaN的情况。4.4 服务部署与接口测试模型训练好之后写一个FastAPI应用把它包起来。启动命令是uvicorn app:app --host 0.0.0.0 --port 8000 --workers 1。注意workers不要设太大因为每个worker都会加载一份模型到显存设多了显存直接爆掉。# 启动服务 uvicorn src.service.app:app --host 0.0.0.0 --port 8000 --workers 1 # 测试接口 curl -X POST http://localhost:8000/predict \ -H Content-Type: application/json \ -d {text: 这个产品质量很好推荐购买}接口测试要覆盖正常输入、空输入、超长输入、特殊字符输入这几种情况。空输入应该返回400错误超长输入应该被截断而不是报错特殊字符输入不应该导致服务崩溃。5. 常见问题与排查技巧实录5.1 环境与依赖问题速查问题现象可能原因解决方法CUDA out of memorybatch size太大或显存泄漏减小batch size检查是否有未释放的中间变量torch not compiled with CUDAPyTorch版本和CUDA不匹配重新安装对应CUDA版本的PyTorchDocker容器内无法访问GPU未安装nvidia-container-toolkit安装toolkit并重启Docker服务模型加载报KeyError模型结构和checkpoint不匹配检查模型配置和保存时的结构是否一致推理速度远低于预期未使用GPU或未做量化确认模型在GPU上考虑ONNX/TensorRT加速5.2 训练过程中的典型问题训练loss不下降是最常见的问题。排查顺序是先确认数据有没有问题标签是否正确、输入是否正常再确认模型有没有正常初始化可以试着用随机数据过拟合一个小数据集最后检查学习率和优化器设置。验证集指标波动大通常是batch size太小或者学习率太大导致的。可以尝试增大batch size或者降低学习率或者用梯度累积来模拟更大的batch size。显存不够用的时候除了减小batch size还可以用梯度累积、混合精度训练、梯度检查点这几个技巧。混合精度训练用torch.cuda.amp能省一半显存速度还更快。from torch.cuda.amp import autocast, GradScaler scaler GradScaler() for batch in train_loader: with autocast(): outputs model(**batch) loss outputs.loss scaler.scale(loss).backward() scaler.step(optimizer) scaler.update() optimizer.zero_grad()混合精度训练的核心是autocast上下文管理器和GradScaler。autocast自动把部分运算转成FP16GradScaler防止FP16下梯度下溢。这套组合用下来训练速度通常能提升30%到50%。5.3 服务部署中的坑服务部署最常遇到的问题是内存泄漏。表现是服务跑一段时间后内存占用越来越高最后OOM被杀掉。原因通常是全局变量累积、缓存没清理、或者PyTorch的CUDA缓存没释放。排查方法是加内存监控看内存增长的趋势和触发条件。另一个常见问题是并发请求下响应时间飙升。这是因为FastAPI默认是单线程处理请求的模型推理又是CPU/GPU密集型的请求只能排队。解决方案是用run_in_executor把推理放到线程池或者用Triton这类专门的推理服务器。实操心得部署推理服务时一定要加一个健康检查接口返回模型是否加载成功、GPU是否可用、当前显存占用等信息。这个接口在排查线上问题时能省很多时间。5.4 性能优化的经验总结性能优化要分清楚瓶颈在哪里。如果是GPU利用率低说明是IO瓶颈或者CPU预处理瓶颈可以增大num_workers、用更快的存储、或者把预处理放到GPU上做。如果是GPU利用率高但吞吐量还是上不去说明是计算瓶颈可以考虑量化、剪枝、或者换更小的模型。批处理是提升吞吐量最有效的手段。单条推理和批量推理的GPU利用率差距可能有好几倍。但batch size也不是越大越好太大了延迟会变高。生产环境通常要在吞吐量和延迟之间做权衡我的经验是batch size设在8到32之间比较合适。6. 从零搭建之后下一步可以往哪里走把上面这套流程走通之后你手里就有了一个能跑的AI服务。但这只是起点后面还有很多可以深入的方向。模型层面可以尝试更高效的微调方法比如LoRA、QLoRA用很少的参数就能达到接近全量微调的效果。也可以尝试模型蒸馏把大模型的能力迁移到小模型上推理成本能降一个数量级。服务层面可以引入Triton Inference Server来做动态批处理和模型集成吞吐量能再上一个台阶。也可以加上模型版本管理支持灰度发布和快速回滚。运维层面可以搭建完整的监控体系采集QPS、延迟、错误率、GPU利用率这些指标配上告警规则。这样线上出问题能第一时间发现而不是等用户反馈才知道。我自己在这个方向上摸索下来的体会是从零搭建的价值不在于你搭出来的东西有多完美而在于你亲手摸过每一层之后对整个系统有了肌肉记忆般的理解。这种理解是看多少篇教程都换不来的。踩过的坑越多后面做技术决策的时候就越有底气。