1. 为什么我要从零手搓一套AI工程流水线第一次看到ai-engineering-from-scratch这个标题我脑子里蹦出来的不是某个具体框架而是一连串踩过的坑。过去两年团队里新来的同学问得最多的问题不是“Transformer 的注意力怎么算”而是“模型训完了怎么把它变成一个别人能用的服务”。这中间的鸿沟恰恰就是 AI 工程要填的。所谓 AI 工程说白了就是把实验室里的模型变成线上能扛住流量、能持续迭代、能被人维护的软件系统。它跟传统后端最大的区别在于模型本身是个概率黑盒输入输出不稳定资源消耗还特别大所以你不能用写 CRUD 的思路去写推理服务。这套从零开始的工程实践适合三类人一是算法出身但没碰过服务端部署的同学二是后端出身想转 AI 基础设施的工程师三是想搞清楚“模型上线到底经历了什么”的技术负责人。我打算把整条链路拆开从环境隔离、模型加载、推理优化、服务封装、监控告警到灰度发布每一步都给出可复现的操作和背后的取舍逻辑。读完你至少能自己搭一个能跑通的小型推理服务并且知道每个环节为什么这么设计而不是照抄一份 Dockerfile 就完事。2. 整体架构设计与技术选型思路2.1 从需求倒推架构先想清楚服务长什么样动手写代码之前我习惯先把需求翻译成几个硬指标。假设我们要部署一个文本分类模型业务方给的约束通常是单次请求延迟 P99 低于 200ms峰值 QPS 50模型文件 400MB 左右每天更新一次。这几个数字直接决定了架构形态。延迟要求决定了不能用太重的框架QPS 决定了并发模型模型大小决定了显存或内存的占用更新频率决定了要不要做热加载。我见过太多人一上来就上 Kubernetes结果一个内部工具类服务QPS 个位数运维成本比开发成本还高。所以选型的第一原则是按量级选工具别按潮流选工具。单机 QPS 50 以下一个 FastAPI 加 Gunicorn 多 worker 就够了QPS 上千再考虑专门的推理服务器如 Triton 或 TorchServe再往上才需要分布式和自动扩缩容。2.2 分层设计把变化的部分隔离出来我的习惯是把整个服务切成四层每层职责单一方便替换。最底层是模型层负责加载权重、执行前向计算往上是预处理层负责把原始输入转成张量再往上是服务层处理 HTTP 协议、参数校验、并发调度最外层是运维层管日志、指标、健康检查。这样分层的好处是哪天模型从 PyTorch 换成 ONNX只需要动模型层服务层完全不用改。这种隔离在真实项目里救过我很多次。有一次业务方临时要求把模型从 CPU 推理切到 GPU因为模型层和预处理层是分开的我只改了模型加载那几十行代码服务接口和调用方完全无感知。如果当初把加载逻辑和路由写在一起那次改动至少要多花两天。2.3 为什么不用现成的推理框架一步到位很多人会问既然有 Triton、TorchServe 这些成熟方案为什么还要从零写。我的答案是你得先知道轮子怎么造才有资格判断该不该用轮子。现成框架封装了大量细节出问题时你连日志都看不懂。从零写一遍你会被迫理解批处理怎么攒、显存怎么管、并发怎么控这些认知在你用框架时同样值钱。而且现成框架有它的适用边界。Triton 强在 GPU 多模型管理但如果你只是部署一个 CPU 上的小模型它的配置复杂度反而拖慢迭代。我的建议是先用从零的方式跑通一遍把每个环节的坑都踩过然后再根据实际量级决定要不要迁移到成熟框架。这个顺序不能反。3. 核心环节拆解与实操要点3.1 环境隔离别让依赖地狱毁掉你的周末AI 项目的依赖冲突比普通后端严重得多因为 PyTorch、CUDA、cuDNN 之间有严格的版本对应关系。我踩过最惨的一次是本地训练用的 torch 2.0服务器上装的是 1.13模型能加载但推理结果对不上排查了一整天才发现是某个算子行为变了。所以第一步永远是环境隔离。我的做法是用 conda 建独立环境并且把版本号全部锁死写进environment.yml。不要用pip install torch这种不指定版本的方式它今天装的和明天装的可能就不是一个东西。下面是我常用的环境定义模板name: ai-serving channels: - pytorch - conda-forge dependencies: - python3.10 - pytorch2.1.0 - numpy1.24.3 - pip: - fastapi0.104.1 - uvicorn0.24.0 - pydantic2.5.2注意CUDA 版本要和驱动匹配nvidia-smi右上角显示的 CUDA Version 是驱动支持的上限不是已安装版本。装 pytorch 时选 cuda 版本不要超过这个上限否则会报找不到设备的错。3.2 模型加载冷启动时间是可以优化的模型加载是服务启动最慢的一环400MB 的模型从磁盘读进来再初始化动辄十几秒。如果每次扩容都等这么久弹性伸缩就形同虚设。我试过几种优化手段效果最明显的是权重预加载到内存和延迟初始化结合。具体做法是服务启动时只加载模型结构权重文件用内存映射的方式打开第一次请求到来时才真正把权重读进显存。这样启动时间能从 15 秒压到 3 秒左右。代价是第一个请求会慢一些但可以通过预热请求来规避。下面是我常用的加载逻辑import torch import mmap class ModelLoader: def __init__(self, model_path): self.model_path model_path self.model None self._weight_map None def _map_weights(self): # 用内存映射打开权重文件避免一次性读入 with open(self.model_path, rb) as f: self._weight_map mmap.mmap(f.fileno(), 0, accessmmap.ACCESS_READ) def load(self): if self.model is None: self._map_weights() self.model torch.jit.load(self.model_path, map_locationcpu) self.model.eval() return self.model提示torch.jit.load加载的是 TorchScript 格式比直接加载 state_dict 再构建模型快很多。如果模型还没导出成 TorchScript建议先做这一步转换收益很大。3.3 预处理与后处理最容易被忽视的性能杀手很多人把注意力全放在模型推理上结果发现预处理占了总延迟的一半。我实测过一个文本分类任务分词加 padding 花了 80ms模型前向只要 30ms。问题出在每次请求都重新初始化 tokenizer而且用了 Python 循环逐条处理。优化思路有两个一是把 tokenizer 做成单例全局只初始化一次二是批量化处理把多个请求的文本攒在一起做 padding利用向量化操作。下面这个对比表是我实测的数据同一台机器单条文本长度 128优化手段预处理耗时总延迟 P99每次新建 tokenizer80ms210mstokenizer 单例45ms160ms单例 批量 padding12ms95ms批量 padding 的关键是动态 batch不能等太久也不能太小。我的经验值是攒够 8 条或者等待超过 10ms 就触发一次推理这样在延迟和吞吐之间比较平衡。3.4 并发模型同步还是异步这是个问题推理服务有个天然矛盾模型计算是 CPU/GPU 密集型的而 HTTP 服务是 IO 密集型的。如果你用纯异步框架模型计算会阻塞事件循环导致其他请求排队如果用纯同步多进程进程间通信又有开销。我的选择是同步 worker 加进程池用 Gunicorn 起多个 worker每个 worker 内部串行处理请求靠多进程实现并发。这个方案的好处是简单可靠每个 worker 独占一份模型没有锁竞争。坏处是显存占用翻倍因为每个进程都要加载一份模型。所以 worker 数量要算清楚如果模型占 2GB 显存显卡有 16GB那最多起 6 个 worker留 4GB 给运行时开销。计算公式是worker 数量 floor((显存总量 - 预留开销) / 单模型显存占用)我一般预留 20% 的显存做缓冲因为推理过程中会有临时张量分配算得太满容易 OOM。4. 完整实操流程与关键配置4.1 从零搭建一个可用的推理服务假设我们有一个已经训练好的文本分类模型导出成了 TorchScript 格式文件叫classifier.pt。现在要把它变成一个 HTTP 服务。我按顺序列出每一步的操作和意图。第一步建项目结构。我习惯这样组织让每个文件职责清晰ai-serving/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── model.py # 模型加载与推理 │ ├── preprocess.py # 预处理逻辑 │ └── config.py # 配置管理 ├── tests/ │ └── test_api.py ├── environment.yml └── gunicorn_conf.py第二步写配置管理。把所有可变参数抽到环境变量里不要硬编码。这样本地和线上可以用同一份代码只改环境变量。import os from pydantic_settings import BaseSettings class Settings(BaseSettings): model_path: str os.getenv(MODEL_PATH, ./classifier.pt) max_batch_size: int int(os.getenv(MAX_BATCH_SIZE, 8)) batch_timeout_ms: int int(os.getenv(BATCH_TIMEOUT_MS, 10)) device: str os.getenv(DEVICE, cpu) settings Settings()第三步实现模型推理类。这里要注意线程安全虽然每个 worker 是单进程但 FastAPI 内部可能用线程池所以推理时要加锁或者确保模型是只读的。import torch import threading from app.config import settings class Classifier: _instance None _lock threading.Lock() def __new__(cls): if cls._instance is None: with cls._lock: if cls._instance is None: cls._instance super().__new__(cls) cls._instance._load() return cls._instance def _load(self): self.model torch.jit.load(settings.model_path, map_locationsettings.device) self.model.eval() self.lock threading.Lock() def predict(self, batch_tensors): with self.lock: with torch.no_grad(): outputs self.model(batch_tensors) return torch.softmax(outputs, dim-1)第四步写 FastAPI 路由。这里的关键是参数校验和错误处理不要让异常直接抛给调用方。from fastapi import FastAPI, HTTPException from pydantic import BaseModel from app.model import Classifier from app.preprocess import tokenize_batch app FastAPI() classifier Classifier() class PredictRequest(BaseModel): texts: list[str] app.post(/predict) def predict(req: PredictRequest): if not req.texts: raise HTTPException(status_code400, detailtexts 不能为空) if len(req.texts) 32: raise HTTPException(status_code400, detail单次请求最多 32 条) try: tensors tokenize_batch(req.texts) probs classifier.predict(tensors) return {results: probs.tolist()} except Exception as e: raise HTTPException(status_code500, detailf推理失败: {str(e)}) app.get(/health) def health(): return {status: ok}第五步配置 Gunicorn。用 Uvicorn worker 配合 Gunicorn 做进程管理worker 数量按前面说的公式算。# gunicorn_conf.py import multiprocessing bind 0.0.0.0:8000 workers 4 worker_class uvicorn.workers.UvicornWorker timeout 60 graceful_timeout 30 keepalive 5启动命令是gunicorn app.main:app -c gunicorn_conf.py。实测下来4 个 worker 在 8 核 CPU 上能扛住 QPS 120 左右P99 延迟稳定在 110ms。4.2 批处理攒批的实现细节前面提到批处理能大幅降低延迟但攒批逻辑写不好会引入额外延迟。我的实现是用一个队列加一个后台线程请求进来先入队后台线程按条件触发推理。核心代码如下import queue import threading import time class BatchScheduler: def __init__(self, max_size, timeout_ms): self.queue queue.Queue() self.max_size max_size self.timeout timeout_ms / 1000.0 self.results {} self._start_worker() def _start_worker(self): t threading.Thread(targetself._worker, daemonTrue) t.start() def _worker(self): while True: batch [] deadline time.time() self.timeout while len(batch) self.max_size and time.time() deadline: try: item self.queue.get(timeoutdeadline - time.time()) batch.append(item) except queue.Empty: break if batch: self._process(batch) def _process(self, batch): texts [item[0] for item in batch] event_ids [item[1] for item in batch] tensors tokenize_batch(texts) probs classifier.predict(tensors) for eid, prob in zip(event_ids, probs): self.results[eid] prob注意攒批的 timeout 不要设太大10ms 是个比较安全的经验值。设成 50ms 虽然吞吐更高但 P99 延迟会明显恶化因为最坏情况下请求要等满 50ms 才开始推理。4.3 监控指标没有度量就没有优化服务上线只是开始你得知道它跑得好不好。我至少会埋三类指标延迟分布、吞吐量、资源占用。延迟不能只看平均值要看 P50、P95、P99因为平均值会被大量快请求拉低掩盖长尾问题。用 Prometheus 的 Python 客户端埋点很简单from prometheus_client import Histogram, Counter INFERENCE_LATENCY Histogram( inference_latency_seconds, 推理延迟, buckets[0.01, 0.05, 0.1, 0.2, 0.5, 1.0] ) REQUEST_COUNT Counter(request_total, 请求总数, [status]) app.post(/predict) def predict(req: PredictRequest): with INFERENCE_LATENCY.time(): # ... 推理逻辑 REQUEST_COUNT.labels(statussuccess).inc()这些指标接到 Grafana 上你就能一眼看出服务是否健康。我遇到过 P99 突然从 100ms 涨到 800ms 的情况查下来是某个 worker 的模型被换出到交换分区了因为内存不够。如果没有监控这种问题只能靠用户投诉才发现。5. 常见问题排查与避坑经验5.1 推理结果不稳定先查随机性来源模型上线后如果发现同样的输入两次结果不一样别急着怀疑模型坏了。先检查三个地方一是模型有没有设成 eval 模式训练模式下的 dropout 和 batch norm 会导致结果随机二是预处理里有没有随机增强没关掉三是浮点运算在不同硬件上本身就有微小差异。我踩过一次坑模型在本地测试结果稳定上线后偶发不一致。最后发现是预处理里用了random.shuffle做数据增强忘了在推理时关掉。这种问题靠单元测试很难发现得靠对比测试固定输入跑一百次看输出方差。5.2 显存泄漏定时重启不是长久之计GPU 服务跑久了显存慢慢涨最后 OOM这是很常见的问题。原因通常是某些张量没释放或者缓存没清理。临时方案是定时重启 worker但这是治标不治本。根治方法是排查代码里所有torch.no_grad()是否加全了以及有没有在循环里不断创建新张量而不释放。我一般会在推理函数里显式调用torch.cuda.empty_cache()但注意这个操作本身有开销不要每次请求都调可以每隔 N 次调一次。更彻底的做法是用torch.inference_mode()替代torch.no_grad()前者对内存管理更友好。5.3 冷启动慢导致扩容失败弹性伸缩时新实例启动太慢还没准备好就被健康检查判死这是很典型的坑。解决办法是加一个就绪探针在模型加载完成前不接收流量。FastAPI 里可以用 lifespan 事件from contextlib import asynccontextmanager asynccontextmanager async def lifespan(app: FastAPI): # 启动时预热模型 classifier.predict(tokenize_batch([预热文本])) yield # 关闭时清理 torch.cuda.empty_cache() app FastAPI(lifespanlifespan)预热请求很关键它能让模型完成第一次前向计算把该分配的显存都分配好。没有预热的话第一个真实请求会特别慢容易触发超时。5.4 常见问题速查表现象可能原因排查方向P99 延迟突然升高显存不足触发换页查 GPU 显存和系统 swap结果每次不同模型未设 eval 或预处理有随机检查 dropout 和增强逻辑服务启动失败依赖版本冲突核对 torch 和 CUDA 版本吞吐上不去worker 数太少或批处理没生效调整 worker 数和攒批参数内存持续增长张量未释放检查 no_grad 和缓存清理提示排查问题时先看日志再看指标日志能告诉你发生了什么指标能告诉你发生的频率。两者结合才能快速定位。6. 从单机到集群的扩展思路单机服务跑通之后下一步自然是考虑扩展。但我要泼一盆冷水大部分场景根本不需要集群。一个 8 核 16G 的机器配合合理的批处理能扛住相当可观的流量。真正需要集群的信号是单机资源利用率长期超过 70%或者有高可用要求不能接受单点故障。如果确实要扩展我的建议是先把无状态的部分和有状态的部分分开。推理服务本身是无状态的可以水平扩展模型文件是有状态的应该放在共享存储或对象存储上每个实例启动时拉取。这样扩容时新实例不依赖特定机器调度更灵活。负载均衡层用 Nginx 或云厂商的 LB 都行关键是健康检查要配好别把还没预热完的实例加进后端。我一般会设一个 30 秒的初始延迟给模型加载留足时间。至于自动扩缩容CPU 利用率是个不错的触发指标但要注意推理服务的 CPU 利用率波动大阈值设太低会导致频繁扩缩反而影响稳定性。这套从零搭建的流程我在不同项目里复用过很多次每次根据实际量级调整参数。核心思路始终没变先理解每个环节为什么存在再决定用什么工具实现。工具会过时但这份理解不会。