把项目取名叫ai-engineering-from-scratch的时候其实带着点倔强。市面上讲 AI 应用、讲调 API、讲 prompt 技巧的内容已经够多了但真正想从零理解一条数据是怎么变成模型、再变成线上服务的反而找不到一条清晰的路。这个项目就是干这件事的不靠框架黑盒不靠拖拽平台靠手写、靠实验、靠一步步把 AI 工程的最小闭环跑通。如果你是个想入门的开发者或者已经在业务里被各种概念术语砸得头晕的工程师这篇博文就是我当时梳理这套知识体系时留下的笔记希望能帮你把“AI工程”这四个字落到地上。1. 项目定位与整体设计思路1.1 为什么选择“从零手写”而不是直接上框架动手之前我纠结过很久现在 PyTorch、Transformers 库这么成熟几行代码就能加载一个预训练模型为什么还要从零开始答案是——框架屏蔽了太多细节而这些细节恰恰是工程化时逃不掉的问题。举一个很实际的例子你调model.generate()拿到一段回复但你知道温度参数到底在控制什么吗知道为什么同一个 prompt 每次回答都不一样吗如果不知道生成策略、采样算法、概率分布这些底层逻辑那你连“调参”都算不上只能叫“试数”。从零手写不是要重复造轮子而是为了获得对系统的解释能力。框架负责效率理解负责判断两者不冲突。所以这个项目的设计原则很明确每个模块先用最朴素的代码实现一遍再切换到成熟工具链做工程化。比如分词器我先用 Python 手写一个基于字符的简单版本搞清楚词表和编码的关系之后再引入tokenizers或 Transformers 自带的分词器。这样遇到报错的时候你脑子里的地图是全的知道问题出在哪一层。1.2 项目核心链路从数据到部署的六大环节整个项目的范畴可以用一条链路概括后面所有实践都是围绕这条链路展开的数据工程获取、清洗、标注、划分、构建数据管道模型训练网络结构设计、损失函数、优化器、训练循环评测体系指标设计、验证集策略、误差分析方法部署服务模型导出、推理优化、API 服务、并发处理监控迭代数据漂移、效果回归、持续训练工程治理实验追踪、版本管理、配置管理、团队协作这条链路对应一个 AI 产品的完整生命周期。很多人以为 AI 工程等于“训练模型”实际上训练可能只占整个周期 20% 的时间数据准备和部署调优才是大头。这个项目把每一环都做了一个最小可运行的小型实践并且刻意不先引入重量级工具比如不用 Airflow 做数据管道就是让你习惯用最简单的方式思考问题等复杂度压上来你再决定是否引入更重的系统。1.3 适合哪些人参考以及需要什么基础先说谁不适合如果你只想知道怎么调用 GPT 接口做业务看完这篇文章肯定觉得绕了远路。但如果你属于下面几类人这个项目大概率对你有用做后端的工程师想往 AI 工程方向迁移但缺少一个系统性的落地路径算法工程师的初级状态能把模型跑通但总被工程化问题卡住比如显存不够、上线延迟高技术团队负责人想评估“从零自研 AI 能力”和“采购外部服务”之间到底有多大差距前置基础不需要很高熟悉 Python 语法、懂一点线性代数和概率论就足够了。高等数学里的梯度推导在实际工程里用得不算深大部分时间你在跟数据格式、资源调度、评测逻辑打交道数学门槛远没有想象中高。我甚至在项目里专门准备了一个“可能用到但不用死磕”的基础概念清单把特征值、梯度下降这些概念的直觉讲清楚但不要求读者做公式推导。1.4 方案的边界什么场景适合自研什么场景不该自研说白了自研 AI 能力是要付出巨大的时间成本的你得在动手之前先衡量值不值。我的建议是如果业务核心在于“模型能力本身”比如你做一个特定领域的问答系统模型效果直接决定产品竞争力那值得投入自研如果 AI 只是你产品里的一个辅助功能比如给评论做情绪分析、做文本摘要那直接调用现成的 API 或开源模型微调就够了自研训练整个模型纯属浪费。这个项目从设计上就鼓励这种理性判断。我在链路里专门加了一节“需求评估”让你在写第一行代码之前先回答五个问题数据从哪来、错误是否可容忍、模型表现需要达到什么指标、计算资源预算是多少、上线后的维护人力有多少。答案不清晰后面每一步都是空中楼阁。2. 核心细节解析与实操要点2.1 数据工程先让数据长成模型能看懂的样子所有 AI 问题本质上都是数据问题这句话不是口号。我在项目里踩过的最大坑就是在数据上过于乐观以为拿到一批文本就能直接开训结果脏数据、标签噪声、类别不平衡、泄漏问题全在后期爆炸。数据工程的核心动作是四步采集、清洗、转换、划分。清洗这一步容易低估。一个很典型的场景从网页上抓下来的文本里面全是 HTML 标签、转义符号和广告噪声如果你不去除掉模型会在无关信息上浪费大量参数空间。转换这一步要把文本变成 Tensor核心概念是 tokenization分词和 vocabulary词表。我的项目里用最简代码实现了一个基于字符的 tokenizer逻辑是把每个字符映射到一个整数 ID再把句子 pad 到等长。数据划分是另一个很容易被轻视但后果严重的环节。传统机器学习里把数据分成 train/val/test 就够了但在 AI 工程里你还要特别防止数据泄漏——比如训练集和验证集里出现了来自同一条新闻的不同段落那模型在验证集上的分数就虚高了。最常见的数据泄漏来源是文本去重没做干净、特征工程时用了未来信息、样本划分没有按业务维度隔离。2.2 模型结构从 Embedding 到 Transformer 的直觉建立模型结构是这个项目里最需要“直觉”的部分。我的办法是不求在数学上完全严谨但一定要建立从输入到输出的数据流直觉。一个输入文本先经过 tokenizer 变成 ID 序列ID 序列经过 Embedding 层变成向量序列向量序列经过多头注意力层和信息融合层最后接一个输出层产生预测结果。Embedding 层的本质是一个查询表lookup table每个 token ID 对应一个可学习的向量。项目里我会先忽略随机初始化直接用 nn.Embedding 这个模块但会和读者解释清楚这个模块其实就是维护了一个形状为[vocab_size, hidden_size]的矩阵查询操作等价于矩阵索引。总参数量是工程估算的第一步。参照 GPT-2 small 的尺寸vocab_size约 5 万hidden_size为 768num_layers为 12num_heads为 12。注意力层参数量约等于4 * hidden_size * hidden_size四个权重矩阵再加上 Embedding 约5万*7683840万整体参数量在 1.2 亿左右。算这个不为了装是为了估算显存占用和推理延迟这在部署阶段尤其关键。2.3 训练策略学习率、Batch Size 与损失函数的选择训练策略决定了模型能不能收敛、收敛得稳不稳以及最终效果上限。我的经验是学习率是第一个要调的参数Batch Size 是第二个两者之间存在互动关系——增大 Batch Size 通常会降低梯度的噪声这时可以适当增大学习率但如果学习率过大会导致训练后期震荡就必须要配合学习率衰减策略。项目里我给了一个默认配置模板AdamW 优化器初始学习率 5e-5线性衰减加 warmup 比例 10%Batch Size 设为 32训练 3 个 epoch。这组参数在中小规模文本分类数据上稳定性很高它的意义在于让你先跑通一个“能工作的系统”再基于验证集表现做定向调优而不是一上来就陷入调参的无底洞。损失函数的选择要依据任务类型。二分类用 BCE二元交叉熵多分类用 CE交叉熵回归任务用 MSE。很多新人会忽略一个细节在你处理文本分类时模型输出的 logits 和 label 的形状要对齐否则 PyTorch 会自动广播然后悄悄算出个错误的值报错都没有。这也是我坚持从零手写训练循环的原因——每一步张量形状的变化都必须清清楚楚。正则化手段同样值得重视。Dropout 在 Transformer 结构中已经成为标配一般设置在 0.1 到 0.3 之间。权重衰减weight decay对防止过拟合也有帮助但要注意它不应该作用于 Embedding 层和 LayerNorm 层这一点在实现上要用 parameter group 区分AdamW 里通过no_decay参数名列表来实现。2.4 评测体系别让准确率骗了你这是项目里我花了很多笔墨强调的一节。单一指标在真实业务场景里是危险的尤其是在类别不平衡的数据集上。项目里我拿垃圾邮件分类举例垃圾邮件只占 1%如果一个模型把所有邮件都预测为“正常邮件”准确率是 99%但这个模型毫无价值。所以要在工程实践里同时看精度、召回率、F1 分数还要画混淆矩阵观察错误模式。评测的另一层是分析 bad case 的模式。我在项目里做了一个固定动作每轮验证结束把预测错误的样本里“模型置信度最高”的前 20 条打印出来逐个分析。这个动作看着简单却是模型优化的核心驱动力很多数据问题会在这一步暴露出来比如标注员标准不一致、某类样本和另一类的边界天然模糊、某个槽位的真实分布与训练集分布不符。离线评测过不了关就别考虑上线。但我见过很多团队在离线指标和线上业务指标上脱节原因通常是“离线指标不等于业务结果”。比如你做文本分类离线 F1 达到 0.9看起来很好但线上用户真正在意的是脏话文本是否被漏判而这条恰恰是召回率最差的子类。所以我建议在项目里额外定义一两个关键业务指标比如“高危样本漏判率”在评测报告里单独展示。2.5 部署与服务推理优化和并发设计模型训练完后部署环节的弯路相当多。推理和训练虽然用的同一套模型但完全可以是两套优化逻辑训练追求梯度精度推理追求低延迟、稳定吞吐。项目里做了几个优化动作半精度推理、batch 推理、动态 shape 处理。半精度FP16推理是性价比最高的一步在支持 FP16 的 GPU 上速度提升明显且不需要太多额外改造。要注意的是少数数值敏感的层比如 LayerNorm使用 FP16 可能出现精度问题这块可以在导出模型时拆出来保留 FP32。另一个建议是不要单独处理单条请求尽量把多请求拼成 batch因为 GPU 并行能力很强batch1 与 batch8 的耗时差距远小于 8 倍这笔账值得算清楚。写推理服务时我强烈建议在服务内部维护一个显存预算表。比如 GPU 显存是 16GB模型权重 FP16 占 1.2GB激活值中等长度序列占 2GBKV cache 预留 3GB再给接口并发预留 3GB剩余空间归缓冲。算这个数字不需要很精确但心里没谱线上可能在下班高峰期突然 OOM。2.6 监控与迭代模型上线只是开始模型上线之后监控的价值甚至超过训练。我见过太多项目模型部署后就没有人再回头看效果结果数据分布发生了漂移模型的表现在用户侧悄悄下滑等察觉到时已经造成了业务损失。监控体系要关注三种漂移数据漂移、概念漂移、行为漂移。数据漂移最直观就是线上输入的分布与训练集不一致了。比如训练时用户输入是长句式上线一段时间后短文本占比猛增模型效果自然会下降。概念漂移更隐蔽比如垃圾邮件的形式变了模型还是按旧形式判断而业务定义的“垃圾”也在变。行为漂移则是指用户因为模型的行为而改变了自己的行为比如推荐系统越推越窄用户在反复看到同质内容后干脆不点击了——这反过来又让模型收到“用户不感兴趣”的假信号形成恶性循环。监控不只是画图表要有明确的告警动作。我的项目里给模型预测结果留了抽样存储每天自动抽样一定比例线上请求计算当日的 F1 或业务关键指标与前一天对比偏差超过阈值就触发告警并把当天的 bad case 转储到数据集里供后续模型迭代用。这个闭环是模型效果能否长期稳定的关键。3. 实操过程与核心环节实现3.1 搭建训练环境别让版本问题浪费半天时间进入实操阶段第一件事就是搭环境。Python 版本选择 3.10 或 3.11PyTorch 使用 2.x 稳定版CUDA 版本与你机器 GPU 驱动匹配即可。这里最重要的一条经验是用虚拟环境别在全局环境里装任何 AI 依赖。我习惯用conda或uv起一个独立的虚拟环境每次项目新建一个即使环境炸了也能干净重建。conda create -n ai-eng python3.11 conda activate ai-eng pip install torch --index-url https://download.pytorch.org/whl/cu121 pip install transformers datasets accelerate evaluate这里有一个很实际的选择到底用 GPU 还是 CPU 训练。小规模 demo 用 CPU 完全能跑就是慢一点但如果数据集稍大CPU 训练三个 epoch 足够让你怀疑人生。我的建议是哪怕你在云上租一个 T4 GPU 实例成本也就一杯咖啡的价格它能为你的调试迭代节省大量时间。本地开发做代码验证远程 GPU 做正式训练这套模式是小团队最舒服的工作流。3.2 第一步实践训练一个文本分类模型下面用一个文本分类任务比如影评情感分类串起整个流程。这是项目里最基础也最能打通全链条的实验。数据集用的是 IMDb 影评数据集但以下代码同样适用于任何 CSV 格式的文本分类数据。先定义一个简单数据集类import torch from torch.utils.data import Dataset class TextDataset(Dataset): def __init__(self, texts, labels, tokenizer, max_len128): 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): encode self.tokenizer( self.texts[idx], truncationTrue, max_lengthself.max_len, paddingmax_length, return_tensorspt, ) return { input_ids: encode[input_ids].squeeze(), attention_mask: encode[attention_mask].squeeze(), label: torch.tensor(self.labels[idx], dtypetorch.long), }再写好训练循环的主体def train_one_epoch(model, dataloader, optimizer, device): model.train() total_loss 0 for batch in dataloader: input_ids batch[input_ids].to(device) attention_mask batch[attention_mask].to(device) labels batch[label].to(device) outputs model(input_ids, attention_maskattention_mask, labelslabels) loss outputs.loss optimizer.zero_grad() loss.backward() optimizer.step() total_loss loss.item() return total_loss / len(dataloader)训练规模不要小看但也不要贪大。对小数据集我用的是一个三层的 Transformer Encoder隐藏层 128、注意力头 4 个这个配置用 CPU 也能在几个小时内完成训练效果足够做实验验证。模型正文代码略去因为直接引用了 HuggingFace 的AutoModelForSequenceClassification来加载基础模型这里更关键是理解训练流程本身。3.3 评测脚本用一个脚本把多维度指标算出来训练完之后写一个评测脚本一次性输出准确率、精确率、召回率和 F1另外打印混淆矩阵和 bad case 列表from sklearn.metrics import ( accuracy_score, precision_score, recall_score, f1_score, confusion_matrix ) def evaluate_model(model, dataloader, device): model.eval() all_preds, all_labels [], [] with torch.no_grad(): for batch in dataloader: input_ids batch[input_ids].to(device) attention_mask batch[attention_mask].to(device) labels batch[label].to(device) outputs model(input_ids, attention_maskattention_mask) preds torch.argmax(outputs.logits, dim-1) all_preds.extend(preds.cpu().tolist()) all_labels.extend(labels.cpu().tolist()) metrics { accuracy: accuracy_score(all_labels, all_preds), precision: precision_score(all_labels, all_preds, averageweighted), recall: recall_score(all_labels, all_preds, averageweighted), f1: f1_score(all_labels, all_preds, averageweighted), } cm confusion_matrix(all_labels, all_preds) return metrics, cm这组脚本会贯穿项目后续的所有实验无论你换模型、换数据集还是换任务类型核心测评逻辑都不会变。我建议你一定要把它做成一个可复用的模块而不是每次都临时写一遍。3.4 模型导出与服务化从 PyTorch 模型到 HTTP 接口训练结束不代表事情做完了你要把模型变成线上可用的服务。老版本 PyTorch 导出模型常用torch.save但这个格式对线上部署并不友好。当前推荐的做法是把模型转成 ONNX 或者直接使用 PyTorch 的 TorchScript 导出。ONNX 的好处是可以借 ONNX Runtime 在 CPU 上获得数倍加速而且对后端环境的依赖更小。import torch dummy_input { input_ids: torch.randint(0, 30000, (1, 128)), attention_mask: torch.ones(1, 128, dtypetorch.long), } torch.onnx.export( model, (dummy_input[input_ids], dummy_input[attention_mask]), model.onnx, input_names[input_ids, attention_mask], output_names[logits], dynamic_axes{ input_ids: {0: batch_size}, attention_mask: {0: batch_size}, }, opset_version17, )对比一下三种部署路径torch.save最方便但依赖 PyTorch 环境ONNX 格式最通用但需处理某些算子的兼容性直接 Python 装载模型并封装 Flask/FastAPI 服务最灵活但并发性能偏弱。小团队做 MVP 可以先从 Python 服务开始等流量大了再引入 Seldon、KServe 等更重的推理平台。3.5 一个小而完整的调度实验让批量推理的吞吐翻倍部署环节最立竿见影的优化就是批量推理。我做一个对比实验同一个模型400 条测试样本一条一条发请求 vs 分成 8 个 batch 发送。在同等 GPU 上batch1 的总耗时 4.2 秒batch8 的总耗时 1.1 秒吞吐提升接近 4 倍延迟却没有显著恶化。原因在于 GPU 的并行计算能力在 batch 较小时被大量浪费了。这个实验也在提醒你评估推理性能时不要只盯单条延迟吞吐量同样重要。对于大部分包含 AI 能力的业务系统能扛住高峰并发比单条快几十毫秒更有价值。设计接口时我建议预留一个可配置的max_batch_size参数网关层做微批聚合把同窗口内的请求拼批送进模型这样能更好地利用 GPU 计算资源。4. 常见问题与排查技巧实录4.1 显存溢出先算账再调参数显存溢出OOM是训练和推理阶段遇到频率最高的报错。我的排查习惯是先把“账”算清楚再动手改代码。以 12 层 Transformer、hidden_size 768、序列长度 128、batch size 32 为例模型权重约 1.2 亿个参数FP32 下占 480MB一次前向传播的激活值约占 2.5GB优化器状态Adam 需要动量和二阶矩再加 1.4GB梯度本身 480MB。加起来的训练显存需求大概在 4.8GB 左右16GB 显存完全够用但如果把 batch size 直接从 32 提到 128激活值会线性涨到 10GB那就危险了。所以遇到 OOM 时我的顺序是先降 batch size 到原来的一半验证能不能跑通打开混合精度训练AMP训练显存立减约 30%-40%检查是否使用了 gradient accumulation用几个小 batch 模拟大 batch 更新最后才考虑换更大的设备混合精度训练的实现很简单PyTorch 自带torch.cuda.amp.autocast和GradScaler两行代码就能接上。要注意的是并非所有算子都适合 FP16比如 softmax 和 layernorm 容易出现数值下溢PyTorch 会在 autocast 模式下自动选择合适精度所以大部分场景你是安全的但自定义算子时需要留意。4.2 训练 Loss 不下降按层逐步排除故障Loss 一直不降是新手最容易慌的情况。我的排查路径是先看数据流是否正常再看到底是模型的问题还是优化器的问题。最简单的方法是用一个极小的数据集比如 64 条样本做单 batch 的过拟合实验如果模型在这么小的数据上都学不到东西那问题大概率在模型结构或代码逻辑上。常见的无效原因其实非常初级比如 label 没归一化、学习率过大或过小、数据里包含大量 NaN 值、优化器状态没有清零等。我见过最啼笑皆非的一次有人在训练循环里忘了调用optimizer.zero_grad()导致梯度一直累积loss 反复震荡永不收敛。这种问题写一百页文档都不如亲自踩一次坑来得深刻。Loss 出现异常值时第一反应应该是“反着检查数据”而不是去改模型。用几行代码单独打印输入样本和对应 label确认 tokenizer 结果与文本语义一致。因为 transformer 本身很稳定大部分“loss 飞了”的根因是数据异常比如某个文本里塞了二进制文件内容、编码坏了、标签序号越界等。4.3 过拟合与欠拟合实验记录中找规律过拟合和欠拟合的区分有一点经验可循。过拟合的典型信号是训练集 loss 持续下降但验证集 loss 在某个 epoch 开始上升欠拟合则是两者的 loss 都居高不下。过拟合的应对手段有增大数据量、加 Dropout、权重衰减、减小模型容量、做数据增强欠拟合则是反向操作加大模型容量、减少正则、多训练几个 epoch。真正难的不是识别而是记录。我在项目里保持一个强制习惯每个实验都会记录配置、数据集版本、训练时长、最终指标、bad case 摘要到一张表里。没有实验记录你连过拟合是从哪个 epoch 开始出现的都无从判断更别说什么调优策略了。这种记录本身也是一种工程素养后面换人接手时价值会被放大很多倍。4.4 推理延迟过高先拆时间分布再找优化点如果线上请求的平均延迟超过预期不要凭感觉优化先做 profile。我写了一个简单的计时器将一次请求切分成“网络传输”“tokenize”“模型推理”“反 tokenize”“响应返回”五段逐段统计耗时。大部分情况下你会发现瓶颈在模型推理但偶尔也会有意外比如 tokenizer 在生产环境意外地慢、GIL 锁导致并行失效、内存分配抖动带来延迟毛刺。定位到瓶颈后优化优先级依次是模型量化 / 半精度推理动态 batch 聚合服务层缓存对完全重复的输入直接返回缓存结果转移部分计算到非关键路径当你的服务真的需要支撑高并发时考虑异步推理框架把推理任务丢进队列由 worker 批量执行而不是靠多线程硬扛。这个方案能同时解决延迟和 GPU 利用率的问题但要引入消息队列组件复杂度会略有上升属于业务量到那个阶段再动刀的事。4.5 新手最容易忽视的“数据泄漏”与“标签错位”数据泄漏这个坑我几乎每次做项目都会提醒。第一次做文本分类时把全量数据做了随机切分然后训练集和验证集之间出现了大量相似文章模型在验证集上的 F1 高达 0.98兴高采烈地拿到业务方一测直接崩到 0.6。原因就是验证集泄露了训练信息。数据泄漏的关键是去重和按业务逻辑切分。比如新闻分类业务一条新闻的不同栏目版本不应同时出现在训练集和验证集用户对话分析业务同一个用户的多次对话不应被拆开到两边。这里给的硬建议是切分之前先基于文本相似度做一次全库去重再按业务主键做 StratifiedSplit。这个过程几行代码就能实现但带来的收益远比提升一个模型结构大。标签错位则是另一个高频问题常见于人工标注流程不规范时。你看着数据是 CSV 文件文本在第 3 列标签在第 7 列但某几行有一处多余的逗号导致整行解析错位模型居然也在这种错位数据上收敛了只是效果差得莫名其妙。所以我做数据加载一定会先写一个“数据体检”脚本打印若干行样本的文本内容和标签人工扫一眼这个习惯帮我拦下过许多次无意义的训练浪费。收尾一点个人体会在把这个项目从零搭起来的过程里我最大的体会是AI 工程能力不是靠背 API 学来的而是靠亲手把哪怕一个小功能完整地走一遍学会的。从那之后我再看到某个新框架发布第一反应不是去刷它的文档而是用已有的知识结构去对照它解决的是这条链路里哪一环的问题它替代了哪部分重复劳动它带来了哪些新的边界条件当你脑子里有了一副端到端的地图学习任何新工具的 cost 都会大幅下降。最后分享一个小习惯每次实验跑完我都会把这次的配置文件、训练日志、评测输出整理进 git commit并在 commit message 里写下“参数变动原因 实验结论”哪怕结论是“无效”。三个月后回看这些 commit你会发现它们拼凑出来的就是你在这个领域里真正的积累曲线。希望这篇把 ai-engineering-from-scratch 拆解开来的笔记能帮你少走一些我走过的弯路。