1. 从零构建AI工程能力:为什么“会用模型”和“会做工程”是两回事
很多人第一次接触AI项目时,都会经历一个相似的阶段:在笔记本里跑通一个模型,准确率看起来还不错,于是觉得“AI不过如此”。可一旦要把这个模型放到真实业务里,问题就全冒出来了——推理延迟高得离谱、显存动不动就爆、并发一上来服务直接挂掉、模型更新后线上效果莫名其妙变差。这时候你才会意识到,训练一个模型和交付一个AI系统,中间隔着一整条工程链路。
“ai-engineering-from-scratch”这个标题,说的正是这条链路。它不是教你调包跑demo,而是从零开始,把AI工程里那些真正决定成败的环节一个个搭起来。关键词里的“from scratch”很关键,它意味着我们要理解每一层的原理,而不是停留在调用API的层面。适合读这篇内容的人,是那些已经会写Python、懂一点机器学习基础,但一到工程落地就抓瞎的开发者;也适合带团队的技术负责人,用来梳理AI项目的完整工程视角。
我自己踩过的第一个大坑,就是以为模型精度是核心指标。后来才发现,在真实场景里,吞吐量、延迟、成本、可观测性这些工程指标,往往比精度更能决定一个项目能不能活下去。一个精度95%但响应要3秒的模型,和一个精度90%但响应只要200毫秒的模型,在大多数线上业务里,后者才是能用的那个。这就是AI工程要解决的问题:把实验室里的“能跑”,变成生产环境里的“能扛”。
所以这篇内容我会按照一条真实的工程链路来展开:从环境与依赖管理,到数据处理流水线,再到模型推理服务的构建,最后是监控与迭代。每一块我都会讲清楚“为什么这么做”,而不只是“怎么做”。因为AI工程里最贵的不是代码,是那些没人告诉你、只能自己踩出来的经验。
2. 环境与依赖:AI项目里最容易被低估的“地基工程”
2.1 为什么AI项目的环境问题比普通后端更棘手
普通Web项目的依赖相对稳定,一个requirements.txt基本能搞定。但AI项目不一样,它的依赖链条又长又脆:Python版本、CUDA驱动、深度学习框架、算子库、编译工具链,任何一环版本对不上,就是一堆看不懂的报错。我见过太多团队,光是让新同事把开发环境跑起来,就要花掉一整天。
这里有个反直觉的结论:AI工程的第一步不是写模型代码,而是把环境固化下来。因为AI项目的复现成本极高,同样的代码在不同机器上跑出不同结果,是家常便饭。所以从项目第一天起,就要把环境当成代码来管理。
我的做法是用容器化把环境彻底锁死。不是简单写个Dockerfile就完事,而是要把基础镜像、依赖版本、甚至随机种子都固定下来。下面是一个我常用的基础镜像结构思路:
FROM nvidia/cuda:12.1.0-cudnn8-runtime-ubuntu22.04 # 固定Python版本,避免系统自带版本干扰 RUN apt-get update && apt-get install -y python3.10 python3.10-venv python3-pip # 用requirements锁死所有依赖版本 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 设置随机种子环境变量,保证可复现 ENV PYTHONHASHSEED=0 ENV CUBLAS_WORKSPACE_CONFIG=:4096:8这里有个细节值得说:CUBLAS_WORKSPACE_CONFIG这个环境变量,是很多人不知道的。如果你用PyTorch做确定性训练,不设置它,即使固定了所有随机种子,结果还是会有微小差异。这个坑我在一个需要严格复现的实验里卡了整整两天。
2.2 依赖管理:别让“版本漂移”毁掉你的项目
依赖管理上,我强烈建议区分“开发环境”和“生产环境”。开发环境可以宽松一点,方便试新东西;但生产环境必须严格锁定,每一个包的版本都要写死。我习惯用两层文件:requirements.in写顶层依赖,requirements.txt用pip-compile生成完整的锁定版本。
| 管理方式 | 适用场景 | 风险 |
|---|---|---|
| 直接pip install | 临时试验 | 版本漂移,无法复现 |
| requirements.txt手写 | 小项目 | 间接依赖未锁定 |
| pip-compile锁定 | 生产项目 | 需要额外维护 |
| Poetry/Pipenv | 中大型项目 | 学习成本略高 |
实测下来,对于AI项目,pip-compile这套组合最省心。它会把所有间接依赖的版本都算出来,生成一个完全确定的依赖树。这样无论谁在哪台机器上安装,装出来的环境都是一模一样的。
提示:AI项目里要特别注意那些带C扩展的包,比如numpy、scipy、tokenizers。它们的wheel包和系统架构、Python版本强相关,跨平台时最容易出问题。建议在CI里针对目标平台单独构建。
2.3 硬件资源的抽象:让代码不绑死在特定机器上
AI工程还有一个特殊之处:它和硬件绑得太紧。GPU型号、显存大小、甚至驱动版本,都会影响代码能不能跑。如果代码里到处写死cuda:0,那换台机器就废了。
我的经验是,在代码最上层做一个设备抽象层。用一个统一的接口来获取设备,而不是到处写torch.device('cuda')。这样以后要支持多卡、要切换到其他加速硬件,改动量会小很多。这个抽象层不需要多复杂,几十行代码就够,但它带来的灵活性是巨大的。
import torch def get_device(): if torch.cuda.is_available(): return torch.device("cuda") return torch.device("cpu") # 所有模型和数据的设备都从这里取 device = get_device() model = model.to(device)看起来简单,但这一步能帮你避开后面无数的硬编码麻烦。AI工程里,可移植性是一个经常被忽略但极其重要的指标。
3. 数据流水线:AI系统里最脏最累但最不能省的活
3.1 数据质量决定了模型效果的上限
业内有一句话:数据和特征决定了机器学习的上限,而模型和算法只是在逼近这个上限。这句话在工程实践里体现得淋漓尽致。我做过一个项目,模型结构换了三四版,效果提升都不明显;后来花了两周时间清洗数据、修正标注错误,指标直接涨了8个点。
数据流水线的核心任务,是把原始、杂乱、格式各异的数据,变成模型能稳定消费的格式。这个过程包括采集、清洗、转换、增强、分片、缓存。每一步都有坑,而且这些坑往往在项目后期才暴露出来。
举个真实的例子。我们有个文本分类任务,训练数据是从多个来源汇总的。一开始没注意,直接混在一起训练。结果模型上线后,对某个特定来源的输入表现特别差。排查后发现,那个来源的数据在训练集里占比不到2%,模型根本没学好。这就是数据分布问题,如果不做数据流水线的统计分析,你根本发现不了。
3.2 构建可复用的数据加载层
数据加载这块,最常见的错误是把所有逻辑塞进一个巨大的脚本里。正确的做法是分层:底层是数据源适配,中间是转换逻辑,上层是给训练用的Dataset接口。这样每一层都可以单独测试和替换。
我习惯用这样的结构来组织:
- 数据源层:负责从各种存储读取原始数据,屏蔽存储差异
- 清洗层:处理缺失值、异常值、格式统一
- 转换层:做tokenize、归一化、特征提取
- 缓存层:把处理好的数据缓存起来,避免重复计算
- Dataset层:对接训练框架的标准接口
其中缓存层是最容易被忽略但收益最大的。AI训练往往要跑很多轮,如果每轮都重新做一遍数据预处理,浪费的时间非常可观。把预处理结果缓存成二进制格式,加载速度能提升一个数量级。
import hashlib import pickle from pathlib import Path def cached_process(raw_data, process_fn, cache_dir=".cache"): # 用数据内容哈希做缓存键,数据变了缓存自动失效 key = hashlib.md5(str(raw_data).encode()).hexdigest() cache_path = Path(cache_dir) / f"{key}.pkl" if cache_path.exists(): with open(cache_path, "rb") as f: return pickle.load(f) result = process_fn(raw_data) cache_path.parent.mkdir(exist_ok=True) with open(cache_path, "wb") as f: pickle.dump(result, f) return result这个缓存模式我在多个项目里用过,简单但极其有效。关键是缓存键要用数据内容的哈希,而不是文件名或时间戳,这样数据一变缓存就自动失效,不会用到过期数据。
3.3 数据版本管理:别让“这份数据是哪来的”成为悬案
AI项目里另一个高频问题是:模型效果变差了,但不知道是哪次数据更新导致的。没有数据版本管理,你连回滚都做不到。
数据版本管理不需要多复杂的工具,核心是记录三件事:数据来源、处理脚本版本、处理参数。每次生成一份训练数据,就把这三样东西记下来,存成一个元数据文件。这样任何时候都能追溯。
| 记录项 | 作用 | 示例 |
|---|---|---|
| 数据来源 | 定位原始数据 | s3://bucket/raw/2024-01 |
| 脚本版本 | 定位处理逻辑 | git commit abc123 |
| 处理参数 | 复现处理过程 | max_len=512, lower=True |
| 数据指纹 | 校验数据一致性 | md5: xxxx |
| 生成时间 | 排查时间线 | 2024-01-15 10:30 |
这套东西看起来笨,但真出问题时能救命。我经历过一次线上事故,模型突然对某类输入全部误判。靠数据版本记录,半小时内就定位到是当天上午更新的一批数据里混入了错误标注,回滚后立刻恢复。
注意:数据版本管理不要追求一步到位上大工具。先用文件记录的方式跑起来,等团队和流程成熟了再考虑专门的平台。过早引入重工具,反而会拖慢迭代。
4. 模型推理服务:把实验室模型变成能扛并发的线上服务
4.1 推理服务的核心指标不是精度,是延迟和吞吐
模型训练完,下一步就是把它变成服务。这一步是AI工程里技术含量最高的部分之一。很多算法工程师写的推理代码,在单条数据上跑得好好的,一上并发就崩。原因很简单:训练和推理的优化目标完全不同。
训练追求的是收敛和精度,可以慢慢跑;推理追求的是在有限资源下,尽可能快地处理尽可能多的请求。这两个目标经常是冲突的。比如训练时可以用大batch提升效率,但推理时batch太大会导致单条延迟飙升。
所以推理服务设计的第一件事,是明确你的延迟预算和吞吐目标。这两个指标决定了你后面所有的技术选型。下面是我总结的一个决策参考:
| 场景 | 延迟要求 | 推荐方案 |
|---|---|---|
| 实时交互 | <200ms | 小模型+量化+单条推理 |
| 准实时 | <1s | 中等模型+动态batch |
| 离线批处理 | 无硬要求 | 大模型+大批量+异步队列 |
这个表不是绝对的,但能帮你快速定位方向。我见过太多团队,明明做的是离线任务,却非要追求实时推理的架构,结果复杂度上去了,收益却没多少。
4.2 动态批处理:吞吐和延迟的平衡术
动态批处理是推理服务里最实用的优化手段之一。它的思路是:不固定batch大小,而是攒一小段时间的请求,凑成一批一起推理。这样既能利用GPU的并行能力,又不会让单个请求等太久。
实现上,核心是一个带超时机制的队列。请求进来先入队,后台线程每隔几毫秒取一批出来推理。超时时间设多少,取决于你的延迟预算。一般设成延迟预算的十分之一到五分之一比较合适。
import time import threading from queue import Queue class DynamicBatcher: def __init__(self, max_batch=32, max_wait_ms=10): self.max_batch = max_batch self.max_wait = max_wait_ms / 1000 self.queue = Queue() def infer(self, item): # 请求入队,等待结果 future = Future() self.queue.put((item, future)) return future.get() def _worker(self): while True: batch = [] deadline = time.time() + self.max_wait # 攒批:要么攒够max_batch,要么等到超时 while len(batch) < self.max_batch and time.time() < deadline: try: batch.append(self.queue.get(timeout=0.001)) except Exception: break if batch: self._run_batch(batch)这段代码是简化版,但核心逻辑都在。实际用的时候还要考虑异常处理、优雅退出、批内请求的排序等。动态批处理调参是个细活,max_batch和max_wait要结合你的硬件和流量特征反复试。我的经验是,先用保守参数上线,再根据监控数据慢慢调。
4.3 模型量化与加速:用可接受的精度损失换性能
如果延迟还是压不下来,下一步就是模型加速。量化是最常用的手段,把FP32的权重转成INT8,模型体积和计算量都能大幅下降。实测下来,INT8量化通常能带来2到4倍的推理加速,精度损失在1个点以内。
但量化不是无脑转就行。有些层对量化特别敏感,比如LayerNorm、Softmax,强行量化会导致精度崩掉。所以实践中常用的是混合量化:敏感层保持FP32,其余层用INT8。这个需要针对具体模型做实验,没有通用答案。
除了量化,还有算子融合、图优化、KV Cache等手段。这些优化叠加起来,效果很可观。但我要提醒一句:优化要有优先级,先解决瓶颈。用profiler找出真正的耗时点,再针对性优化。盲目优化不耗时的地方,纯属浪费时间。
提示:推理优化做完后,一定要做精度回归测试。我见过量化后精度掉了5个点却没发现的案例,因为测试集和线上数据分布不一致。回归测试要用真实线上数据抽样,而不是只用训练时的验证集。
5. 监控与迭代:AI系统上线只是开始,不是结束
5.1 为什么AI系统的监控比普通服务更复杂
普通后端服务的监控相对直接:CPU、内存、QPS、错误率。AI系统除了这些,还要监控模型层面的指标:输入分布、输出分布、置信度、特征漂移。因为模型的效果会随着线上数据的变化而衰减,这种衰减不会体现在CPU或内存上,但会实实在在影响业务。
我经历过一次典型的模型衰减。一个推荐模型上线三个月后,点击率慢慢下滑。基础设施监控一切正常,但业务指标就是不行。后来分析发现,用户的兴趣分布已经变了,而模型还是用三个月前的数据训练的。这就是数据漂移,是AI系统特有的问题。
所以AI系统的监控要分三层:
- 基础设施层:CPU、内存、GPU利用率、延迟、错误率
- 模型服务层:请求量、批大小分布、推理耗时、超时率
- 模型效果层:输入特征分布、输出分布、置信度分布、业务指标
第三层是最容易被忽略的,但恰恰是最重要的。没有它,你根本不知道模型什么时候开始“变质”。
5.2 特征分布监控:捕捉数据漂移的早期信号
特征分布监控的核心思路是:把线上推理时的输入特征统计出来,和训练时的特征分布做对比。如果差异超过阈值,就报警。这样能在业务指标恶化之前就发现问题。
具体做法上,我会对每个关键特征记录训练时的均值、方差、分位数,然后线上定期采样计算同样的统计量,做对比。对于类别特征,则对比各类别的占比。
| 监控项 | 训练时基准 | 线上实测 | 告警阈值 |
|---|---|---|---|
| 特征A均值 | 0.52 | 0.71 | 偏差>20% |
| 特征B方差 | 1.3 | 2.8 | 偏差>50% |
| 类别C占比 | 15% | 4% | 偏差>50% |
这套监控不需要多复杂,用简单的统计脚本就能做。关键是定期跑、有基准、能报警。我一般设成每小时跑一次,采样最近一小时的线上请求。
5.3 模型迭代的闭环:从监控到再训练
监控发现问题后,下一步就是迭代。AI工程的迭代闭环,理想状态是:监控发现漂移 → 触发数据收集 → 标注新数据 → 再训练 → 评估 → 灰度上线 → 继续监控。
这个闭环里,最耗时的是数据标注和评估。所以工程上要尽量把能自动化的部分自动化。比如数据收集可以自动落盘,评估可以用固定的评估集自动跑,灰度上线可以用流量切分自动控制。
但我要强调一点:再训练不是越频繁越好。频繁再训练会带来两个问题:一是成本高,二是模型行为不稳定,用户体感会变差。我的经验是,根据业务的数据变化速度来定再训练周期。变化快的场景(比如热点推荐)可能每天都要更新,变化慢的场景(比如通用分类)一个月一次就够了。
# 一个简化的再训练触发逻辑 def should_retrain(drift_score, days_since_last_train, min_interval_days=7): # 漂移严重且距上次训练超过最小间隔,才触发 if drift_score > 0.3 and days_since_last_train >= min_interval_days: return True return False这个逻辑很朴素,但能避免无意义的频繁训练。实际用的时候,drift_score的计算方式要根据业务调整,没有标准答案。
6. 一些踩过坑才明白的工程经验
6.1 别过早追求“完美架构”
我刚开始做AI工程时,总想着一步到位搭一个完美的架构:微服务、消息队列、特征平台、模型仓库全上。结果项目进度被拖得一塌糊涂,很多组件根本用不上。后来才明白,AI工程的架构应该跟着业务规模走,而不是跟着技术潮流走。
小项目就用单体服务加脚本,能跑通就行。等流量上来了、团队扩大了,再逐步拆分。过早引入复杂架构,只会增加维护成本,降低迭代速度。我现在的原则是:能用简单方案解决的,绝不上复杂方案。
6.2 日志和可观测性要从第一天就做
这个教训是用血换来的。早期项目为了赶进度,日志随便打,结果线上出问题时完全不知道发生了什么。模型输出异常,但没有任何输入输出的记录,排查无从下手。
现在我要求所有AI服务必须记录:请求ID、输入摘要、输出摘要、推理耗时、模型版本。这些信息在排查问题时价值极高。记录方式可以用结构化日志,方便后续检索和分析。
注意:记录输入输出时要注意数据隐私。敏感字段要脱敏,或者只记录哈希值。这个在项目初期就要考虑,后期补很麻烦。
6.3 模型版本管理要和生产发布解耦
模型更新和代码发布是两件事,但很多团队把它们绑在一起。结果每次模型更新都要走一遍完整的代码发布流程,慢且容易出错。正确的做法是把模型文件独立管理,服务启动时从模型仓库拉取指定版本,更新模型只需要切换版本号,不需要重新部署服务。
这样做的另一个好处是回滚快。模型出问题,改个版本号重启就行,不用回滚代码。我在一个项目里靠这个机制,把模型回滚时间从半小时缩短到了两分钟。
6.4 压测要模拟真实流量分布
上线前的压测,很多人用均匀分布的请求去压,结果上线后真实流量一来就崩。因为真实流量往往是不均匀的,某些类型的请求特别多,或者请求大小差异很大。
所以压测数据要从真实流量里采样,保持和线上一致的分布。如果还没有线上流量,就尽量模拟真实的分布特征。压测不只是测极限QPS,更要测在真实分布下的表现。
7. 写在最后:AI工程是一门“妥协的艺术”
做了这些年AI工程,我最大的体会是:AI工程不是追求技术最优,而是在各种约束下找平衡。延迟和吞吐要平衡,精度和成本要平衡,迭代速度和稳定性要平衡。没有完美的方案,只有适合当前场景的方案。
从零构建AI工程能力,最重要的不是掌握多少工具,而是建立起一套工程思维:知道每个环节为什么存在,知道问题可能出在哪里,知道怎么用最小的代价解决问题。这套思维,比任何具体的框架和工具都更持久。
如果你正准备开始自己的AI工程项目,我的建议是:先跑通最小闭环,再逐步优化。不要一上来就追求大而全,那样很容易在半路就耗尽精力。先把数据、模型、服务、监控这条主线打通,哪怕每一环都很粗糙,也比一个精致但跑不起来的架构强得多。