1. 从零手搓AI工程为什么我不建议你直接调包很多人一听到“AI工程”这四个字第一反应就是打开某个云平台拖几个预训练模型出来拼一个API调用链然后对外宣称自己做了个AI应用。我早期也这么干过结果在一次内部技术评审上被问得哑口无言——模型为什么在这个场景下会输出这种结果推理延迟的瓶颈到底在GPU还是IO如果要把模型量化到INT8精度损失怎么评估我一个都答不上来。这就是“ai-engineering-from-scratch”这个项目标题真正戳中的痛点。它不是让你从零训练一个GPT而是让你从零理解AI工程这条链路上的每一个环节数据怎么进、模型怎么跑、推理怎么加速、服务怎么部署、效果怎么评估。只有亲手把这条链路搭一遍你才有资格说“我懂AI工程”。这篇文章适合三类人第一类是有一定Python基础但没碰过模型部署的后端或全栈工程师第二类是做算法研究但缺乏工程落地经验的同学第三类是想转行AI工程但被各种框架文档绕晕的开发者。我会按照一个真实项目的推进顺序把数据准备、模型加载、推理优化、服务封装、性能压测这几个核心环节拆开讲每个环节都告诉你“为什么这么做”以及“我踩过什么坑”。需要提前说明的是我不会推荐任何特定的云服务或商业平台所有内容都基于开源工具和本地环境你可以直接在自己的开发机上复现。整个项目的目标很明确用一张消费级显卡甚至CPU把一个中等规模的模型跑起来并且让它具备可用的推理性能。2. 数据管线的搭建别让脏数据毁掉你的推理服务2.1 为什么数据清洗要在推理之前做很多人觉得推理服务嘛输入就是用户传过来的文本或图片直接喂给模型就行了。这个想法在实际生产环境里会死得很惨。我接手过一个文本分类服务上线第一天就崩了原因是有用户传了一段包含大量特殊Unicode字符的文本tokenizer直接抛异常整个服务进程挂掉。所以数据管线的第一原则是永远不要信任输入。在推理之前必须有一层预处理逻辑把输入规范化。对于文本任务至少要做这几件事去除控制字符、统一换行符、限制最大长度、处理空输入。对于图像任务要检查通道数、分辨率范围、像素值范围。我在项目里用了一个很轻量的方案写一个preprocess.py里面定义一组纯函数每个函数只做一件事。比如normalize_text()负责Unicode规范化truncate_tokens()负责按tokenizer的最大长度截断。这样做的好处是每个环节都可以单独测试出问题的时候能快速定位是哪一步挂了。import unicodedata def normalize_text(text: str) - str: # 统一Unicode形式去除控制字符 text unicodedata.normalize(NFKC, text) text .join(ch for ch in text if unicodedata.category(ch)[0] ! C) return text.strip() def truncate_tokens(tokens: list, max_len: int) - list: if len(tokens) max_len: return tokens # 保留头尾中间截断避免丢失关键信息 half max_len // 2 return tokens[:half] tokens[-(max_len - half):]2.2 批处理与流式处理的取舍数据管线的第二个决策点是用批处理还是流式处理这个选择直接决定了你的服务架构。批处理适合离线场景比如每天跑一次全量数据的推理把结果存到数据库。流式处理适合在线服务用户请求来了就处理。但很多人忽略了一点在线服务也可以做微批处理。比如把100毫秒内到达的请求攒成一个batch一起送进模型这样GPU利用率能提升好几倍。我在项目里实现了一个简单的微批处理器核心逻辑是用一个队列加一个定时器。队列满或者超时就触发一次推理。这个方案在QPS不高的时候效果不明显但当QPS超过50之后吞吐量能提升3到5倍。注意微批处理会引入额外延迟最大延迟等于你的超时时间。如果你的业务对延迟极其敏感比如实时对话超时时间要设得很短比如20毫秒。2.3 数据版本管理一个容易被忽视的坑数据管线还有一个隐藏的坑版本管理。你今天用了一套预处理逻辑明天改了一个参数如果没有记录出了问题根本回滚不了。我的做法是在预处理函数的入口处计算一个哈希值把原始输入的哈希和预处理后的哈希都记下来写入日志。这样一旦发现模型输出异常可以快速定位是输入变了还是模型变了。这个做法看起来很简单但在实际排查问题时能省下大量时间。我曾经遇到过一个case模型对某类输入的准确率突然掉了10个点查了半天以为是模型退化最后发现是预处理里一个正则表达式被同事改了把某些关键字符过滤掉了。如果有输入哈希记录这个问题五分钟就能定位。3. 模型加载与推理引擎的选择逻辑3.1 PyTorch原生推理够不够用刚接触AI工程的人通常会直接用PyTorch的model.generate()或者model.forward()来做推理。这在开发阶段没问题但在生产环境里PyTorch原生推理有几个硬伤Python GIL限制并发、没有针对推理做图优化、内存管理不够精细。我做过一个对比测试同一个BERT-base模型用PyTorch原生推理和用ONNX Runtime推理在CPU上后者快了将近2倍在GPU上差距小一些但也有30%左右的提升。原因在于ONNX Runtime做了算子融合、常量折叠、内存复用这些优化而PyTorch的动态图机制在推理时会有额外开销。所以我的建议是开发阶段用PyTorch部署阶段导出成ONNX或TorchScript。导出过程本身也是一次对模型的体检能发现很多隐藏问题比如不支持的算子、动态shape导致的导出失败等。3.2 ONNX导出的具体步骤与常见报错导出ONNX的代码看起来很简单但实际操作中会遇到各种报错。我整理了一个标准的导出流程import torch import torch.onnx def export_to_onnx(model, dummy_input, output_path, input_names, output_names, dynamic_axesNone): model.eval() with torch.no_grad(): torch.onnx.export( model, dummy_input, output_path, input_namesinput_names, output_namesoutput_names, dynamic_axesdynamic_axes, opset_version14, do_constant_foldingTrue )这里有几个关键参数需要解释。opset_version建议用14或更高因为低版本对Transformer类模型的支持不好。dynamic_axes用来声明哪些维度是动态的比如batch size和sequence length如果不声明导出的模型只能接受固定shape的输入。do_constant_folding会把能提前计算的常量算好减小模型体积。常见的报错有这么几类一是算子不支持比如某些自定义的Attention实现解决办法是换成标准算子或者自己写ONNX自定义算子二是shape不匹配通常是dummy_input的维度跟模型预期不一致三是数据类型问题PyTorch默认float32但有些操作需要int64。3.3 量化用精度换速度的边界在哪里量化是推理优化的一个大杀器。简单说就是把float32的权重和激活值用int8表示模型体积缩小4倍推理速度提升2到4倍。但量化会带来精度损失关键是要找到那个可接受的边界。我通常用动态量化做第一轮尝试因为它不需要校准数据直接对权重做量化激活值在推理时动态量化。对于Transformer类模型动态量化通常能把精度损失控制在1%以内速度提升1.5到2倍。import torch.quantization def dynamic_quantize(model): quantized_model torch.quantization.quantize_dynamic( model, {torch.nn.Linear}, dtypetorch.qint8 ) return quantized_model如果动态量化不够快可以上静态量化但需要准备校准数据集。校准集不用很大几百条代表性样本就够了但一定要覆盖各种输入分布。我见过有人用训练集的前100条做校准结果模型对长文本的推理精度暴跌因为前100条都是短文本。提示量化后的模型一定要做完整的评估不能只看整体准确率。要分场景、分输入长度、分数据来源分别看指标否则很容易在某个细分场景上翻车。4. 服务封装从脚本到可用API的距离4.1 为什么FastAPI比Flask更适合推理服务把模型跑起来只是第一步要让它成为一个服务还需要一层Web框架。Flask和FastAPI我都用过最终推荐FastAPI原因有三个原生异步支持、自动生成API文档、基于Pydantic的请求校验。推理服务的一个典型场景是多个请求同时到达每个请求都要等模型推理完成。如果用Flask的同步模式请求会排队后面的请求要等前面的处理完。FastAPI的异步模式可以让请求在等待IO时释放事件循环虽然模型推理本身是CPU/GPU密集型的但预处理和后处理可以异步化整体吞吐量能提升不少。from fastapi import FastAPI from pydantic import BaseModel, Field import numpy as np app FastAPI() class InferenceRequest(BaseModel): text: str Field(..., min_length1, max_length512) top_k: int Field(default5, ge1, le20) class InferenceResponse(BaseModel): labels: list scores: list latency_ms: float app.post(/predict, response_modelInferenceResponse) async def predict(req: InferenceRequest): # 预处理、推理、后处理 ...Pydantic的校验功能特别实用。比如上面的max_length512如果用户传了超过512个字符的文本FastAPI会自动返回422错误根本不会进入推理逻辑。这比在业务代码里手动判断要优雅得多。4.2 模型加载的时机与内存管理模型应该在哪里加载这个问题看似简单但选错了会导致严重的性能问题。我见过有人在每个请求的处理函数里加载模型结果QPS低得可怜因为加载模型本身就要几百毫秒到几秒。正确的做法是在服务启动时加载一次放在全局变量或应用状态里。FastAPI提供了lifespan机制可以在服务启动和关闭时执行特定逻辑。from contextlib import asynccontextmanager ml_models {} asynccontextmanager async def lifespan(app: FastAPI): # 启动时加载模型 ml_models[classifier] load_model(model.onnx) yield # 关闭时释放 ml_models.clear() app FastAPI(lifespanlifespan)内存管理还有一个容易忽略的点推理过程中的中间张量。如果不在推理结束后及时释放显存会逐渐被占满。PyTorch的torch.no_grad()上下文管理器能减少一部分内存占用但更彻底的做法是在每次推理后调用torch.cuda.empty_cache()。不过这个操作本身有开销不要每次推理都调可以每隔N次或者显存使用超过阈值时再调。4.3 错误处理与降级策略生产环境的推理服务必须考虑错误处理。模型可能因为各种原因失败输入格式不对、显存不足、推理超时。每一种失败都要有对应的处理策略。我的做法是定义一组异常类型每个类型对应不同的HTTP状态码和降级方案。比如输入格式错误返回400显存不足返回503并触发降级到CPU推理推理超时返回504并记录日志。降级策略特别重要。我负责过一个服务GPU偶尔会因为驱动问题不可用如果没有降级方案整个服务就挂了。后来加了一个CPU推理的备用路径虽然慢很多但至少能保证服务可用。实现方式很简单捕获GPU推理的异常切换到CPU模型重新推理。5. 性能压测你的服务到底能扛多少QPS5.1 压测工具的选择与脚本编写服务搭好了接下来要知道它能扛多少请求。压测工具我用过JMeter、Locust和wrk最终推荐Locust原因是它用Python写测试脚本跟我们的技术栈一致而且支持分布式压测。一个基本的Locust脚本长这样from locust import HttpUser, task, between import random class InferenceUser(HttpUser): wait_time between(0.01, 0.1) task def predict(self): text 这是一段测试文本 * random.randint(1, 10) self.client.post(/predict, json{ text: text, top_k: 5 })这个脚本模拟用户不断发送推理请求Locust会统计响应时间、QPS、错误率等指标。关键是要设置合理的wait_time太小会导致压测机本身成为瓶颈太大则压不出真实性能。5.2 关键指标解读P99延迟比平均延迟更重要压测报告里有一堆指标但真正需要关注的是P99延迟和P999延迟而不是平均延迟。原因很简单平均延迟会被大量快速请求拉低掩盖掉那些慢请求。而在生产环境里用户体验是由最慢的那1%请求决定的。我遇到过一个典型案例平均延迟50毫秒看起来很好但P99延迟到了2秒。排查后发现是某些长文本请求触发了不同的推理路径走了CPU fallback。如果只看平均延迟这个问题根本发现不了。除了延迟还要看吞吐量和资源利用率。GPU利用率如果长期低于30%说明批处理大小可以调大如果接近100%但QPS上不去说明瓶颈在GPU计算需要考虑模型优化或加卡。5.3 压测中发现的典型瓶颈与调优压测最大的价值是暴露瓶颈。我总结了几类常见瓶颈和对应的调优手段瓶颈类型表现调优手段CPU预处理CPU利用率高GPU利用率低用多进程或C扩展加速预处理GPU计算GPU利用率接近100%量化模型、减小batch、换更小模型内存拷贝推理延迟波动大使用pin memory、减少Host-Device传输网络IO响应时间随并发数线性增长启用HTTP/2、压缩响应体锁竞争多线程下QPS不增反降减少全局锁、用无锁数据结构我在项目里遇到的最大的坑是内存拷贝。当时模型在GPU上推理只要10毫秒但整体延迟有50毫秒查了半天发现是输入数据从CPU拷贝到GPU花了30多毫秒。后来用了pin_memoryTrue和异步拷贝把这部分开销降到了5毫秒以内。6. 从能跑到好用那些文档里不会写的经验6.1 日志与监控出问题时你能看到什么服务上线之后最怕的就是出问题却不知道从哪里查。我的经验是日志要记全但不要记太多。每个请求记录请求ID、输入长度、推理耗时、输出摘要这就够了。不要把完整的输入输出都打进日志一是隐私问题二是日志量太大会拖慢服务。监控方面至少要盯这几个指标QPS、P50/P99延迟、错误率、GPU利用率、显存使用率。这些指标可以用Prometheus采集用Grafana展示。我还会加一个自定义指标推理队列长度。如果队列长度持续增长说明服务处理不过来需要扩容或优化。6.2 模型热更新不重启服务换模型业务需求变化时模型需要更新。如果每次更新都重启服务会造成服务中断。热更新的思路是新模型加载到内存后原子性地切换推理入口的指针。实现方式有很多种我用的是双缓冲方案维护两个模型槽位当前使用的槽位和备用槽位。更新时把新模型加载到备用槽位加载完成后切换指针等所有进行中的请求结束后释放旧模型。这个方案的关键是要处理好并发请求确保切换过程中不会有请求用到已经被释放的模型。6.3 我踩过的最大的坑版本不一致最后分享一个我踩过的最大的坑。有一次本地测试一切正常部署到服务器上推理结果完全不对。查了一整天最后发现是ONNX Runtime的版本不一致本地是1.15服务器是1.12两个版本对某些算子的实现有差异。从那以后我养成了一个习惯把所有依赖的版本号写死在requirements.txt里并且在CI流程里加一步环境一致性检查。具体做法是在服务启动时打印所有关键库的版本号跟预期版本对比不一致就告警。这个习惯帮我避免了好几次类似的问题。另外模型文件本身也要做版本管理。我见过有人直接覆盖模型文件结果新模型有问题想回滚都回滚不了。正确的做法是模型文件带版本号比如model_v1.2.3.onnx服务配置里指定用哪个版本回滚只需要改配置。整个项目做下来我最大的体会是AI工程的门槛不在模型本身而在模型之外的那些工程细节。数据怎么流、内存怎么管、服务怎么部署、性能怎么调这些才是决定一个AI应用能不能上生产的关键。把这条链路亲手走一遍比看十篇论文都有用。