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 PYTHONHASHSEED0 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_okTrue) 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_len512, lowerTrue数据指纹校验数据一致性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_batch32, max_wait_ms10): 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(timeout0.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.520.71偏差20%特征B方差1.32.8偏差50%类别C占比15%4%偏差50%这套监控不需要多复杂用简单的统计脚本就能做。关键是定期跑、有基准、能报警。我一般设成每小时跑一次采样最近一小时的线上请求。5.3 模型迭代的闭环从监控到再训练监控发现问题后下一步就是迭代。AI工程的迭代闭环理想状态是监控发现漂移 → 触发数据收集 → 标注新数据 → 再训练 → 评估 → 灰度上线 → 继续监控。这个闭环里最耗时的是数据标注和评估。所以工程上要尽量把能自动化的部分自动化。比如数据收集可以自动落盘评估可以用固定的评估集自动跑灰度上线可以用流量切分自动控制。但我要强调一点再训练不是越频繁越好。频繁再训练会带来两个问题一是成本高二是模型行为不稳定用户体感会变差。我的经验是根据业务的数据变化速度来定再训练周期。变化快的场景比如热点推荐可能每天都要更新变化慢的场景比如通用分类一个月一次就够了。# 一个简化的再训练触发逻辑 def should_retrain(drift_score, days_since_last_train, min_interval_days7): # 漂移严重且距上次训练超过最小间隔才触发 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工程项目我的建议是先跑通最小闭环再逐步优化。不要一上来就追求大而全那样很容易在半路就耗尽精力。先把数据、模型、服务、监控这条主线打通哪怕每一环都很粗糙也比一个精致但跑不起来的架构强得多。