1. 从零搭建AI工程体系为什么我劝你别急着调库ai-engineering-from-scratch这个标题第一次看到的时候我愣了一下。市面上讲AI的教程铺天盖地但绝大多数都是教你import torch然后跑个预训练模型或者调个API做个聊天机器人。真正从工程角度、从零开始把一套AI系统搭起来的内容少得可怜。我自己在这个坑里摸爬滚打了几年带过几个从零起步的团队也见过太多人一上来就装环境、拉框架、跑demo结果模型训练崩了不知道去哪看日志推理延迟高了不知道瓶颈在哪数据管道堵了不知道从哪查起。这些问题的根源都一样跳过了工程底座直接去够算法天花板。所以这篇内容我想聊的是AI工程从零开始这件事到底意味着什么。它不是让你手写一个Transformer的每一行代码而是让你理解一套AI系统从数据进到结果出中间到底经过了哪些环节、每个环节的工程约束是什么、哪些地方最容易出问题。适合谁看刚转行做AI工程的同学、带小团队从零搭建AI能力的负责人、以及那些模型能跑但系统不稳的开发者。如果你已经在大厂做MLOps好几年了这篇可能对你偏基础但里面的一些排查思路和参数取舍逻辑或许还是能给你一点参考。我先把核心观点摆出来AI工程的核心不是算法是系统工程。算法决定了天花板工程决定了你能不能摸到那个天花板。而from scratch的价值在于你亲手搭过一遍之后再去看那些封装好的框架和平台你会知道它们到底帮你做了什么、在哪些地方做了妥协、什么时候该绕开它们。2. 整体设计思路先画数据流再选技术栈2.1 为什么我坚持数据流优先而不是模型优先大多数人做AI项目的顺序是这样的先想用什么模型然后找数据然后训练然后部署。这个顺序在学术研究里没问题但在工程里是灾难。因为模型选型是相对容易替换的而数据流一旦设计错了后面改起来伤筋动骨。我习惯的做法是反过来的先把数据从产生到消费的完整链路画出来。数据从哪来是用户上传、数据库同步、还是第三方接口拉取数据进来之后要经过哪些清洗、转换、标注步骤训练时怎么读推理时怎么读这两条路径是共用还是分开每个环节的吞吐量和延迟要求是多少这些问题想清楚了技术栈的选择就变成了填空题。比如你的数据量在百万级以下、更新频率是每天一次那用文件系统加Pandas完全够用没必要上Spark。但如果数据是实时流式的、每秒几千条那消息队列和流处理框架就是刚需。我见过一个团队一上来就搭了Kafka加Flink的实时管道结果他们的数据源其实是一个每天凌晨导出的CSV文件。整套流处理架构跑了三个月运维成本高得离谱最后换回了一个定时脚本加数据库稳定得很。2.2 技术栈选型的三个约束条件选型的时候我一般看三个东西团队能力、数据规模、迭代速度。团队能力是最容易被忽略的。如果你团队里没人写过Kubernetes的YAML那硬上K8s做模型部署就是给自己找罪受。用Docker Compose加一台配置好点的服务器能撑很久。数据规模决定了你需不需要分布式很多项目的数据量其实单机完全扛得住上分布式纯粹是心理安慰。迭代速度决定了你的架构要有多灵活如果模型每周都要换一版那训练管道的自动化程度就比推理性能更重要。这三个条件之间还会互相拉扯。比如迭代速度快但团队能力一般那就应该选托管服务而不是自建用钱换时间和稳定性。数据规模大但迭代速度慢那可以花时间做深度优化把成本压下来。2.3 一个典型的从零架构长什么样我拿一个最常见的场景举例做一个文本分类服务数据是业务方定期给的一批标注数据模型需要每周更新一次线上QPS大概几十。这个场景下我的架构是这样的数据层用对象存储放原始数据和标注结果中间用SQLite或者PostgreSQL做元数据管理记录每条数据的版本和状态。训练层用一个Python脚本串起数据加载、预处理、训练、评估输出模型文件和评估报告。推理层用FastAPI包一个HTTP服务加载模型文件提供预测接口。部署层用Docker Compose把推理服务和数据库跑起来前面挂一个Nginx做反向代理和限流。整套东西没有用任何AI平台全是通用工程组件。为什么因为在这个规模下通用组件的稳定性和可维护性远高于专用平台。而且团队里任何人只要会Python和Docker就能接手没有学习成本。3. 核心细节解析数据管道、训练循环、推理服务3.1 数据管道最脏最累但最重要的一环数据管道是AI工程里最没有技术含量、但最容易出问题的部分。我统计过自己处理过的线上问题大概六成跟数据有关模型本身的问题不到两成。数据管道的核心任务是三件事采集、清洗、版本化。采集的关键是幂等性。同一批数据重复采集不能产生重复记录这需要在采集端做去重通常用数据内容的哈希值作为唯一标识。清洗的关键是可追溯每一步清洗操作都要记录日志出了问题能回滚到上一步。版本化的关键是原子性一批数据要么全部更新成功要么全部保持原样不能出现半新半旧的状态。我一般会用这样的目录结构来管理数据data/ raw/ # 原始数据只读 2024-01-15/ batch_001.jsonl processed/ # 清洗后的数据 2024-01-15/ batch_001_clean.jsonl metadata.db # 记录每个批次的处理状态和校验和每个批次处理完之后往metadata.db里写一条记录包含批次ID、原始文件路径、处理后文件路径、记录条数、校验和、处理时间。训练脚本只读processed目录并且会校验metadata里的状态确保不会用到处理失败的数据。这里有个坑很多人清洗数据的时候直接覆盖原文件结果发现清洗逻辑有问题想回退原始数据已经没了。raw目录必须只读任何清洗操作都输出到新目录这是铁律。3.2 训练循环别小看日志和检查点训练循环看起来简单无非是读数据、前向传播、算损失、反向传播、更新参数。但工程上要考虑的东西很多。首先是检查点策略。我一般会同时保存两种检查点按步数保存的定期检查点和验证集指标最好的最佳检查点。定期检查点用于恢复训练最佳检查点用于最终部署。保存的时候要把优化器状态、学习率调度器状态、当前的epoch和step都存下来不然恢复训练的时候会有微妙的不一致。其次是日志粒度。训练日志要记录每个step的损失、学习率、梯度范数每个epoch的验证指标以及显存占用和吞吐量。梯度范数特别重要它能提前告诉你训练是不是要崩了。如果梯度范数突然涨到几百那大概率是数据里有异常样本或者学习率太大了。# 一个简化的训练循环骨架重点看日志和检查点部分 for epoch in range(num_epochs): for step, batch in enumerate(train_loader): outputs model(batch) loss criterion(outputs, batch.labels) optimizer.zero_grad() loss.backward() # 记录梯度范数这是训练稳定性的重要指标 grad_norm torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0) optimizer.step() scheduler.step() if step % log_interval 0: logger.info(fepoch{epoch} step{step} loss{loss.item():.4f} flr{scheduler.get_last_lr()[0]:.2e} grad_norm{grad_norm:.2f}) if step % checkpoint_interval 0: save_checkpoint(model, optimizer, scheduler, epoch, step, pathfcheckpoints/step_{step}.pt)最后是随机种子的管理。做实验的时候随机种子不固定结果就没法复现。我一般会在训练脚本开头固定Python、NumPy、框架的随机种子并且把种子值写进日志。但要注意数据加载的多进程worker如果各自有随机性还需要额外处理。3.3 推理服务延迟、吞吐、稳定性的三角博弈推理服务和训练完全是两个世界。训练可以慢慢跑推理必须快、必须稳。延迟和吞吐是一对矛盾。批处理能提高吞吐但会增加单条请求的延迟。我的经验是如果QPS在几十到几百这个量级不做批处理每条请求单独推理延迟最低。如果QPS上千那就需要做动态批处理把短时间内到达的请求攒成一批一起推理。稳定性方面最重要的是优雅降级。模型加载失败怎么办输入格式不对怎么办推理超时怎么办这些都要有明确的处理逻辑。我的做法是服务启动时加载模型如果加载失败就拒绝启动并报警而不是启动一个不能用的服务。推理时对输入做严格校验格式不对直接返回错误码不要试图猜用户意图。推理超时设置一个上限超了就返回超时错误避免请求堆积拖垮整个服务。# FastAPI推理服务的关键部分 app.post(/predict) async def predict(request: PredictRequest): # 输入校验 if not request.text or len(request.text) max_length: raise HTTPException(status_code400, detailinvalid input) try: # 推理超时控制 result await asyncio.wait_for( run_inference(request.text), timeoutinference_timeout ) return {label: result.label, score: result.score} except asyncio.TimeoutError: raise HTTPException(status_code504, detailinference timeout)推理服务的日志要记录每条请求的输入长度、推理耗时、返回结果。这些日志是排查性能问题的唯一依据。我一般会把耗时超过阈值的请求单独打一条warning日志方便后续分析。4. 实操过程从空目录到能用的服务4.1 环境准备用Docker把依赖锁死从零开始的第一步不是写代码是搭环境。我强烈建议用Docker不是为了高级是为了可复现。Python的依赖地狱大家都懂今天能跑的代码明天换个机器就报错这种事太常见了。我的做法是写一个Dockerfile把Python版本、系统依赖、Python包全部锁死。基础镜像用官方的python:3.10-slim系统依赖只装必要的比如编译工具和OpenCV需要的库Python包用requirements.txt固定版本号。FROM python:3.10-slim RUN apt-get update apt-get install -y --no-install-recommends \ build-essential \ rm -rf /var/lib/apt/lists/* WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [uvicorn, app.main:app, --host, 0.0.0.0, --port, 8000]requirements.txt里每个包都要写死版本号比如torch2.1.0而不是torch2.0。为什么因为意味着你每次构建镜像可能装到不同的版本今天能跑明天就崩。写死版本号虽然不够灵活但稳定压倒一切。4.2 数据准备小步快跑先跑通再优化数据准备阶段最容易犯的错是一步到位。想着一口气把清洗、去重、标注、增强全做完结果管道太复杂一个环节出错整个流程卡住。我的建议是先跑通最小闭环。拿一小批数据比如1000条只做最基本的清洗去空、去重、截断直接喂给模型训练。跑通之后再逐步往管道里加环节每加一个环节就验证一次。这样出问题的时候你明确知道是哪个环节引入的。数据格式我一般用JSONL每行一个样本包含文本和标签。为什么不用CSV因为文本里可能有逗号、换行CSV处理起来麻烦。JSONL天然支持嵌套结构而且可以流式读取不用一次性加载到内存。{id: 001, text: 这个产品质量很好, label: positive} {id: 002, text: 发货太慢了, label: negative}4.3 训练与评估把实验当代码管理训练脚本我一般会做成命令行工具用argparse或者click传参数。学习率、批大小、训练轮数这些超参数全部通过命令行传入不写死在代码里。这样跑不同实验的时候只需要改命令行参数不用改代码。python train.py \ --data_dir data/processed/2024-01-15 \ --output_dir experiments/exp_001 \ --learning_rate 2e-5 \ --batch_size 32 \ --num_epochs 3 \ --seed 42每次实验的输出目录里除了模型文件还要有完整的命令行参数记录、训练日志、验证集评估报告、以及一份环境信息Python版本、包版本、GPU型号。这些东西看起来琐碎但当你三个月后想复现某个实验的时候它们就是救命稻草。评估不能只看准确率。分类任务我至少会看准确率、精确率、召回率、F1以及混淆矩阵。如果类别不平衡准确率会骗人。比如99%的样本是正类模型全预测正类也有99%准确率但召回率是100%、精确率是99%看起来很好实际上负类一个都没识别出来。4.4 部署上线先能跑再谈优化部署的第一步是让服务能跑起来。用Docker Compose把推理服务和依赖的数据库、缓存拉起来本地测试通过之后再考虑上服务器。version: 3.8 services: api: build: . ports: - 8000:8000 volumes: - ./models:/app/models environment: - MODEL_PATH/app/models/latest.pt restart: unless-stopped上线之后要做的第一件事是压测。用locust或者wrk模拟并发请求看服务的QPS、P99延迟、错误率。压测的目的是找到服务的容量上限以及在上限附近的表现。如果P99延迟在可接受范围内那就可以按这个容量来配置限流。压测的时候要注意第一次请求通常会慢很多因为模型需要加载到显存、各种缓存需要预热。所以压测前要先发一批预热请求等延迟稳定了再开始统计。5. 常见问题与排查技巧实录5.1 训练不收敛先查数据再查模型训练loss不下降或者震荡大多数人第一反应是调学习率、换优化器。但我的经验是先查数据。最常见的几个数据问题标签错了正负样本标反、数据里有大量重复样本模型记住了、文本长度分布极端大部分是短文本少数超长文本把batch撑爆。这些问题不解决调参调到天荒地老也没用。排查方法很简单随机抽100条训练数据人工看一遍标签对不对。统计一下文本长度的分布看看有没有异常值。检查一下每个类别的样本数看看是不是严重不平衡。如果数据没问题再查模型。梯度范数是不是爆炸了如果是加梯度裁剪。loss是不是NaN了如果是检查有没有除零或者log(0)。学习率是不是太大了用学习率扫描跑几个小实验看看。5.2 推理延迟高从这三个地方找原因推理延迟高我一般按这个顺序排查输入预处理、模型前向、输出后处理。输入预处理经常是隐藏的瓶颈。比如文本分词如果用Python的循环逐条处理几百条请求下来光分词就耗掉大半时间。解决办法是用批量分词或者用更快的分词库。模型前向的瓶颈通常是batch size太小或者模型太大。如果GPU利用率很低说明batch size太小GPU在等数据。如果GPU利用率很高但延迟还是高那就是模型本身计算量大需要考虑量化或者蒸馏。输出后处理的瓶颈通常是排序或者格式化。如果返回结果需要按分数排序数据量大的时候排序本身就很耗时。解决办法是只返回top-k结果不要全量排序。排查位置常见原因解决办法输入预处理逐条处理、分词慢批量处理、换分词库模型前向batch太小、模型太大增大batch、量化、蒸馏输出后处理全量排序、格式化慢只返回top-k、简化格式5.3 服务内存泄漏九成是缓存没管好推理服务跑着跑着内存越来越高最后OOM被杀这是很常见的问题。九成的情况是缓存没管好。比如用functools.lru_cache缓存了分词结果但maxsize设得太大或者没设缓存无限增长。或者用全局字典缓存了模型输出但从来不清理。解决办法是给所有缓存设上限并且定期清理。还有一种情况是PyTorch的显存碎片。频繁创建和销毁tensor会导致显存碎片化虽然总显存够用但没有连续的大块显存。解决办法是尽量复用tensor或者定期调用torch.cuda.empty_cache()。我遇到过一次内存泄漏排查了两天才发现是一个日志handler持有了请求对象的引用导致每个请求的对象都无法被GC回收。所以日志里不要直接记录请求对象记录请求的ID或者关键字段就行。5.4 模型更新后效果变差先别急着回滚模型更新后线上效果变差第一反应是回滚。但回滚之前先确认几件事新模型的离线评估指标是不是真的比旧模型好如果离线指标好但线上差那大概率是训练和推理的数据分布不一致。最常见的不一致训练时用了某些特征但推理时这些特征拿不到或者拿到的值不一样。比如训练数据里有个用户历史点击率特征但线上推理时这个特征的计算逻辑和训练时不同。这种问题离线评估发现不了只有上线才知道。排查方法是把线上推理的输入数据采样一批下来用新模型离线跑一遍对比线上返回的结果。如果离线结果和线上结果不一致那就是推理管道有问题。如果一致但效果就是差那就是模型过拟合了训练数据需要重新审视训练数据的代表性。5.5 常见问题速查表现象可能原因排查动作训练loss不降数据标签错、学习率不当人工检查数据、学习率扫描训练loss震荡batch太小、学习率太大增大batch、降低学习率推理延迟高预处理慢、batch小批量预处理、增大batch服务内存涨缓存无上限、显存碎片设缓存上限、复用tensor更新后效果差数据分布不一致采样线上数据离线对比服务启动失败模型文件缺失、依赖冲突检查模型路径、锁定依赖版本6. 一些踩坑之后的个人体会做AI工程这几年我最大的体会是简单方案的上限比想象中高复杂方案的下限比想象中低。很多团队一上来就追求先进架构结果被运维复杂度拖垮。反而是那些用最朴素方案搭起来的系统稳定跑了很久。另一个体会是日志和监控的投入永远不亏。训练时多打一行日志排查问题时可能省下一天。推理时多记一个指标容量规划时就有据可依。这些东西在项目顺利的时候看起来多余但出问题的时候就是救命稻草。最后说一个具体的技巧给每个实验和每次部署打标签。实验标签记录数据版本、代码版本、超参数部署标签记录模型版本、配置版本、部署时间。这样当线上出问题的时候你能快速定位到是哪个版本引入的。我一般用Git的commit hash作为代码版本用数据目录的日期作为数据版本简单但够用。这套从零搭建的思路后续还可以往几个方向扩展。比如加上自动化的模型评估和对比每次新模型训练完自动和线上模型PK赢了才允许部署。或者加上A/B测试框架新模型先跑小流量验证效果后再全量。这些都是在基础闭环跑通之后自然而然可以叠加的能力。但前提是基础闭环得先跑通。