1. 从零搭建AI工程能力为什么大多数人卡在第一步就放弃了如果你最近在技术社区里频繁看到ai-engineering-from-scratch这个说法不用怀疑它不是什么新框架的名字也不是某个大厂开源的库。它描述的是一种学习路径——从最底层开始把AI工程化所需的每一项能力亲手搭起来而不是上来就调API、跑现成模型。我之所以想聊这个话题是因为过去大半年里我带过几个刚入行的朋友做AI相关的项目发现一个非常普遍的现象他们能熟练地调用各种模型接口能跑通LangChain的示例代码但一旦遇到需要自己处理数据管道、自己设计推理服务、自己优化延迟的场景就完全不知道从哪里下手。这个问题的根源不在于他们不够聪明而在于学习路径被捷径绑架了。现在的AI工具链太成熟了成熟到你只需要几行代码就能得到一个看起来能用的结果。但这种能用和工程化之间的距离比大多数人想象的要大得多。ai-engineering-from-scratch这个方向本质上是在补那段被跳过的基本功。这篇文章适合谁看如果你是刚接触AI工程的学生或者转行者它能帮你建立一条清晰的学习路线避免在工具链的海洋里迷失方向。如果你已经有一定经验但总觉得自己的知识体系是拼凑起来的它也能帮你把散落的点串成线。我不会给你一个21天速成的计划因为那不存在。我会做的是把从零构建AI工程能力这件事拆开告诉你每个阶段该做什么、为什么这么做、以及最容易踩的坑在哪里。2. 先搞清楚AI工程到底在工程什么2.1 模型不是全部甚至不是最重要的部分很多人对AI工程的想象是这样的训练一个模型然后部署上线。但真实情况是在一个完整的AI系统里模型本身可能只占20%的工作量。剩下的80%是什么是数据采集和清洗、是特征处理、是推理服务的搭建、是监控和告警、是版本管理和回滚机制。我见过太多项目模型指标在实验室里很漂亮一上线就崩原因往往跟模型无关而是数据管道在真实流量下出了问题。举个具体的例子。假设你要做一个文本分类服务。在notebook里你加载一个预训练模型喂几条测试数据准确率90%皆大欢喜。但到了生产环境你需要考虑的事情包括输入文本的长度分布是什么样的有没有异常字符并发请求量有多大单个请求的延迟要求是多少模型文件有多大加载需要多久需不需要做批处理这些问题没有一个能靠调API解决。所以ai-engineering-from-scratch的第一课不是学某个框架而是建立系统思维。你要把自己当成一个造房子的人模型只是其中一块砖你还需要地基、框架、水电、装修。2.2 从零开始不等于从轮子开始造这里有一个常见的误解需要澄清。From scratch不意味着你要自己实现矩阵乘法、自己写反向传播、自己造一个PyTorch出来。那是研究者的工作不是工程师的。AI工程领域的从零指的是从工程化的角度把每一个环节都理解透彻知道它是怎么工作的、为什么这么设计、什么时候会出问题。你可以用现成的库但你要知道这个库在背后做了什么。比如你用HuggingFace的transformers加载模型一行代码就搞定了。但你应该知道这行代码背后发生了这些事情从磁盘或网络加载模型权重文件、初始化模型结构、把权重填进去、把模型移到指定设备上、设置推理模式。每一步都可能出问题——权重文件损坏、内存不够、设备不可用、版本不匹配。如果你不理解这些出了问题就只能靠重启和祈祷。2.3 一条被验证过的学习路线基于我带人的经验从零构建AI工程能力可以分成四个阶段每个阶段有明确的目标和产出物。阶段核心目标关键产出预计投入基础夯实理解数据流动的全链路一个完整的数据处理管道4-6周模型服务化把模型变成可调用的服务一个带监控的推理API3-4周性能与可靠性让服务在真实流量下稳定运行压测报告和优化方案4-6周迭代与运维建立持续改进的闭环自动化评估和部署流程持续进行这个路线不是线性的你可能会在某个阶段卡很久也可能会跳着学。但大方向是这样的先让数据能顺畅地流起来再让模型能稳定地提供服务然后让服务能扛住压力最后让整个系统能持续进化。3. 数据管道最不起眼但最容易翻车的地方3.1 为什么数据管道值得你花最多时间我在实际项目里观察到一个规律一个AI项目延期十有八九是因为数据问题。模型选型可以很快决定服务框架可以快速搭建但数据管道往往是一团乱麻。原因很简单数据是脏的、乱的、不完整的、格式不统一的。你在网上下的数据集是别人清洗过的但真实业务里的数据什么妖魔鬼怪都有。从零构建数据管道你需要处理的事情包括数据采集从哪里来、数据清洗去掉什么、数据转换变成什么格式、数据存储放在哪里、数据版本管理怎么追踪变化。每一件事都有坑而且这些坑往往在项目后期才会暴露出来。3.2 一个最小可用的数据管道长什么样我不打算给你一个复杂的架构图那没有意义。我们从最简单的开始一个能跑通的数据管道核心就是三个环节读取、处理、写入。import json from pathlib import Path from typing import Iterator def read_raw_data(path: str) - Iterator[dict]: 从JSONL文件逐行读取原始数据 with open(path, r, encodingutf-8) as f: for line in f: line line.strip() if not line: continue try: yield json.loads(line) except json.JSONDecodeError: # 记录坏数据但不中断整个流程 continue def clean_record(record: dict) - dict | None: 清洗单条记录返回None表示丢弃 text record.get(text, ).strip() if len(text) 10: return None # 去掉控制字符 text .join(ch for ch in text if ch.isprintable() or ch in \n\t) return {text: text, label: record.get(label, unknown)} def write_processed(records: Iterator[dict], output_path: str): 写入处理后的数据 Path(output_path).parent.mkdir(parentsTrue, exist_okTrue) with open(output_path, w, encodingutf-8) as f: for record in records: f.write(json.dumps(record, ensure_asciiFalse) \n)这段代码看起来很简单但里面有几个关键的设计决策值得说。第一逐行读取而不是一次性加载这是为了处理大文件时不会爆内存。第二遇到坏数据时跳过而不是崩溃这是生产环境的基本要求。第三清洗逻辑独立成函数方便测试和复用。3.3 数据版本管理一个被严重低估的环节你可能会问数据还需要版本管理当然需要。想象一下这个场景你用一份数据训练了模型A效果不错。两周后你更新了数据重新训练了模型B结果效果变差了。你想回退到模型A但发现数据已经变了没法复现。这时候你就知道数据版本管理有多重要了。最土但最有效的办法是每次数据处理的结果都带一个时间戳或者哈希值存到不同的目录里。然后在训练脚本里记录用了哪个版本的数据。不需要上什么高级工具一个简单的命名规范就能解决80%的问题。注意千万不要覆盖原始数据。原始数据是你的最后一道防线任何时候都不要直接修改它。所有的清洗和转换都应该是生成新的文件。3.4 数据质量的检查清单在把数据喂给模型之前我通常会跑一遍检查清单。这个清单帮我避免了很多低级错误。样本数量够不够太少会导致过拟合太多可能训练时间过长。类别分布均不均衡严重不均衡需要做采样或加权。文本长度分布有没有异常长或异常短的样本重复率有没有大量重复样本重复样本会让模型记住而不是学习。标签一致性同一个样本有没有被标成不同类别特殊字符有没有乱码、控制字符、HTML标签残留这些检查不需要复杂的工具用pandas加几行代码就能搞定。但就是这几行代码能帮你省下几天甚至几周的调试时间。4. 把模型变成服务从notebook到API的距离4.1 notebook里的模型和生产环境的模型是两回事在notebook里模型是一个对象你调用它的predict方法它返回结果。在生产环境里模型是一个服务它需要处理并发请求、需要控制延迟、需要在出错时优雅降级。这两者之间的差距就是AI工程的核心战场。我见过很多团队模型在notebook里跑得好好的一做成API就各种问题。最常见的问题包括内存泄漏每次请求都加载模型、延迟波动大没有做批处理、并发上不去用了同步框架、错误处理缺失一个坏请求搞崩整个服务。4.2 一个最小可用的推理服务我们用FastAPI来搭一个最简单的推理服务。选择FastAPI的理由很直接异步支持好、性能不错、代码量少、自带文档。from fastapi import FastAPI, HTTPException from pydantic import BaseModel from contextlib import asynccontextmanager import torch from transformers import AutoTokenizer, AutoModelForSequenceClassification # 全局变量在启动时加载 model None tokenizer None asynccontextmanager async def lifespan(app: FastAPI): 服务启动时加载模型关闭时释放 global model, tokenizer model_name your-model-path tokenizer AutoTokenizer.from_pretrained(model_name) model AutoModelForSequenceClassification.from_pretrained(model_name) model.eval() if torch.cuda.is_available(): model model.cuda() yield # 清理 del model del tokenizer app FastAPI(lifespanlifespan) class PredictRequest(BaseModel): text: str class PredictResponse(BaseModel): label: str score: float app.post(/predict, response_modelPredictResponse) async def predict(request: PredictRequest): if not request.text.strip(): raise HTTPException(status_code400, detailtext cannot be empty) try: inputs tokenizer( request.text, return_tensorspt, truncationTrue, max_length512, paddingTrue ) if torch.cuda.is_available(): inputs {k: v.cuda() for k, v in inputs.items()} with torch.no_grad(): outputs model(**inputs) probs torch.softmax(outputs.logits, dim-1) score, pred torch.max(probs, dim-1) return PredictResponse( labelstr(pred.item()), scorefloat(score.item()) ) except Exception as e: raise HTTPException(status_code500, detailstr(e))这段代码有几个关键点。第一模型在lifespan里加载只加载一次所有请求共享。这是最基本的优化但很多人会忘记。第二用了torch.no_grad()推理时不需要计算梯度能省不少内存和计算。第三加了输入校验空文本直接返回400。第四异常处理避免一个坏请求导致服务崩溃。4.3 批处理提升吞吐量的第一把钥匙上面的服务能跑但吞吐量很低。每个请求单独做一次前向传播GPU利用率可能只有10%。解决办法是批处理把多个请求攒在一起一次性送给模型。批处理的实现有两种思路。一种是客户端攒批就是客户端自己把多个请求合并成一个。另一种是服务端攒批服务端维护一个队列每隔几毫秒或者攒够一定数量就做一次推理。服务端攒批对客户端透明更实用。import asyncio from collections import deque class BatchProcessor: def __init__(self, model, tokenizer, max_batch_size32, max_wait_ms10): self.model model self.tokenizer tokenizer self.max_batch_size max_batch_size self.max_wait_ms max_wait_ms self.queue deque() self.lock asyncio.Lock() async def predict(self, text: str) - dict: future asyncio.get_event_loop().create_future() async with self.lock: self.queue.append((text, future)) if len(self.queue) self.max_batch_size: await self._process_batch() # 等待结果 return await future async def _process_batch(self): if not self.queue: return batch list(self.queue) self.queue.clear() texts [item[0] for item in batch] futures [item[1] for item in batch] try: inputs self.tokenizer( texts, return_tensorspt, truncationTrue, max_length512, paddingTrue ) if torch.cuda.is_available(): inputs {k: v.cuda() for k, v in inputs.items()} with torch.no_grad(): outputs self.model(**inputs) probs torch.softmax(outputs.logits, dim-1) scores, preds torch.max(probs, dim-1) for i, future in enumerate(futures): future.set_result({ label: str(preds[i].item()), score: float(scores[i].item()) }) except Exception as e: for future in futures: future.set_exception(e)这个批处理器还需要一个定时器来触发否则请求不够一批时会一直等。实际实现中可以用asyncio的定时任务每隔max_wait_ms检查一次队列。批处理能把吞吐量提升5到10倍代价是稍微增加了一点延迟。这个权衡在大多数场景下是值得的。4.4 模型服务的监控指标服务上线之后你需要知道它运行得怎么样。最少要监控这几个指标请求量QPS、延迟分布P50、P95、P99、错误率、GPU利用率、内存占用。这些指标不需要复杂的系统用Prometheus加Grafana就能搞定。如果不想搭这套至少要在日志里记录每个请求的处理时间和结果状态。提示延迟的P99比平均值重要得多。平均值会被大量快速请求拉低掩盖掉那些慢请求。而用户体验往往由最慢的那1%决定。5. 性能优化当服务开始扛不住的时候5.1 先定位瓶颈再动手优化性能优化最忌讳的就是凭感觉。你觉得是模型推理慢结果发现是数据预处理占了80%的时间。所以第一步永远是测量不是优化。我常用的方法是在代码的关键路径上打时间戳记录每个阶段的耗时。比如一个请求进来记录tokenization花了多久、模型推理花了多久、后处理花了多久。跑几百个请求看时间分布。这样你就能清楚地知道瓶颈在哪里。常见的瓶颈和对应的优化方向瓶颈位置典型表现优化方向数据预处理CPU占用高GPU空闲并行化、缓存、用更快的tokenizer模型推理GPU利用率高延迟大量化、蒸馏、换更小的模型后处理延迟波动大优化算法、异步处理网络传输延迟与请求大小相关压缩、批处理、连接复用内存频繁GC延迟抖动减少对象创建、用对象池5.2 模型量化用一点精度换大量速度量化是把模型的权重从浮点数变成低精度整数比如从FP32变成INT8。这样模型文件变小了推理速度也快了代价是精度可能会下降一点点。对于大多数应用场景这个代价是可以接受的。用PyTorch做动态量化很简单import torch.quantization # 动态量化适用于LSTM和Linear层 quantized_model torch.quantization.quantize_dynamic( model, {torch.nn.Linear}, dtypetorch.qint8 )动态量化不需要校准数据直接转换就行。实测下来模型大小能减少到原来的四分之一推理速度能提升2到3倍精度损失通常在1%以内。如果你的场景对精度极其敏感可以先在小批量数据上对比量化前后的输出确认差异在可接受范围内再上线。5.3 缓存最简单也最有效的优化如果你的服务有大量重复请求缓存是最划算的优化。比如一个分类服务热门文本可能被反复请求。用一个简单的LRU缓存就能挡住大部分流量。from functools import lru_cache import hashlib def get_cache_key(text: str) - str: return hashlib.md5(text.encode()).hexdigest() # 在服务层做缓存 cache {} async def predict_with_cache(text: str): key get_cache_key(text) if key in cache: return cache[key] result await batch_processor.predict(text) # 限制缓存大小 if len(cache) 10000: cache.clear() cache[key] result return result缓存的关键是选好key。对于文本任务直接用文本的哈希做key就行。对于更复杂的输入需要设计合适的key生成策略。缓存也要注意失效策略如果模型更新了缓存必须清空。5.4 一个真实的优化案例我之前优化过一个文本分类服务初始版本QPS只有20P99延迟800毫秒。优化过程分了三步。第一步加了批处理QPS提升到80P99延迟降到300毫秒。第二步做了动态量化QPS提升到150P99延迟降到180毫秒。第三步加了缓存热门请求直接命中整体QPS提升到400以上P99延迟稳定在100毫秒以内。每一步优化之前我都先测量了瓶颈在哪里。第一步的瓶颈是GPU利用率低所以做批处理。第二步的瓶颈是模型本身的计算量所以做量化。第三步的瓶颈是重复计算所以做缓存。如果顺序反了比如先做缓存效果就不会这么明显因为缓存只能挡住重复请求挡不住首次请求的慢。6. 持续迭代让系统自己进化6.1 离线评估和在线评估的鸿沟模型上线不是终点而是起点。你需要持续监控它的表现发现问题然后迭代。这里有一个经典的陷阱离线指标好不代表线上效果好。离线评估用的是历史数据线上面对的是真实流量两者的分布可能不一样。我遇到过好几次这样的情况新模型在离线测试集上准确率提升了3%上线后线上指标反而下降了。原因通常是训练数据和线上数据的分布有偏移或者离线测试集不能代表真实场景。解决办法是建立在线评估机制用真实流量做A/B测试让数据说话。6.2 一个简单的A/B测试框架A/B测试的核心思想是把流量分成两组一组用旧模型一组用新模型对比关键指标。实现起来不需要很复杂一个分流逻辑加一个指标收集就够了。import random class ABTestRouter: def __init__(self, model_a, model_b, split_ratio0.1): self.model_a model_a self.model_b model_b self.split_ratio split_ratio def route(self, request_id: str): # 用请求ID做哈希保证同一请求始终走同一组 hash_val int(hashlib.md5(request_id.encode()).hexdigest(), 16) if (hash_val % 100) (self.split_ratio * 100): return self.model_b, B return self.model_a, A分流的关键是保证同一个用户或同一个请求始终走同一组否则体验会不一致。用请求ID或者用户ID做哈希是最简单的办法。指标收集方面至少要有准确率、延迟、错误率这三个维度按组分别统计。6.3 模型更新的安全流程模型更新不能直接覆盖要有回滚机制。我推荐的流程是这样的新模型先在小流量上验证确认没问题后逐步扩大流量同时旧模型保持可用。一旦新模型出问题立即切回旧模型。这个流程需要版本管理来支撑。每个模型版本都要有唯一的标识服务能根据配置加载指定版本。配置的变更要能快速生效最好不用重启服务。注意模型文件要保留至少两个版本。我见过有人更新模型时直接覆盖了旧文件结果新模型有问题想回滚发现旧模型已经没了只能连夜重新训练。6.4 数据回流让模型越用越好线上产生的数据是宝贵的资产。用户的实际输入、模型的输出、用户的反馈这些数据可以用来持续改进模型。建立一个数据回流管道把线上数据收集起来定期标注加入训练集重新训练模型。这个闭环一旦建立起来模型的效果会随着时间自然提升。数据回流要注意隐私和合规问题。敏感信息要脱敏用户数据的使用要符合规范。这些不是技术问题但比技术问题更重要。7. 我踩过的那些坑和总结的经验7.1 不要过早优化但也不要忽视明显的瓶颈我刚入行的时候总想把每个环节都优化到极致。结果花了两周做模型量化上线后发现瓶颈根本不在模型而在数据预处理。后来我学乖了先上线一个能跑的版本测量真实瓶颈再针对性优化。但反过来如果某个瓶颈非常明显比如每次请求都重新加载模型那就不要犹豫立刻修掉。7.2 日志和监控不是可选项我吃过最大的亏是一个服务上线后没有加监控结果半夜挂了没人知道第二天用户投诉才发现。从那以后我要求自己做的每个服务都必须有基本的监控和告警。不需要很复杂至少要有请求量、错误率、延迟这三个指标超过阈值就发通知。7.3 测试要覆盖边界情况模型服务的测试不能只测正常输入。空字符串、超长文本、特殊字符、并发请求、模型加载失败这些边界情况都要测。我习惯在服务上线前跑一轮破坏性测试故意发一些奇怪的请求看服务会不会崩。这个习惯帮我避免了好几次线上事故。7.4 文档是写给未来的自己的我现在写代码会强制自己写两类文档。一类是接口文档说明每个接口的输入输出和错误码。另一类是运维文档说明服务怎么启动、怎么配置、出问题了怎么排查。写的时候觉得麻烦但每次回头查的时候都觉得值。特别是当你半夜被叫起来处理故障时一份清晰的运维文档能救命。7.5 保持学习但不要追新AI工程领域的新工具和新框架层出不穷今天流行这个明天流行那个。我的策略是保持关注但不轻易切换。除非新工具能解决我当前面临的具体问题否则我不会花时间去学。把精力放在基本功上比追新框架的回报率高得多。数据管道、服务化、性能优化、监控运维这些核心能力不会因为框架的更替而贬值。从零构建AI工程能力是一条长路没有捷径。但每一步都算数每一个坑都会让你变得更强。我到现在也不敢说自己完全掌握了但至少我知道遇到问题时该往哪个方向找答案。这种知道怎么找答案的能力可能比任何具体的知识点都重要。