从零搭建AI推理服务：后端老兵的工程化踩坑与重构实录

1. 从零搭建AI工程能力：一个后端老兵的踩坑与重构实录

"ai-engineering-from-scratch"这个标题，第一次看到的时候我愣了一下。不是因为它有多高深，恰恰相反——它太直白了，直白到像一句废话。但仔细想想，过去两年我面过不下三十个自称"做过AI项目"的候选人，能从头讲清楚一个推理服务怎么从裸机部署到扛住并发的人，一只手数得过来。大部分人所谓的"AI工程"，其实是在Jupyter Notebook里调通了API，然后写了个Flask包一层就上线了。真到了线上，显存泄漏、请求排队、模型热更新、GPU利用率忽高忽低这些问题一出来，基本就抓瞎。

所以这个标题背后真正想解决的问题，不是"怎么调用大模型API"，而是怎么像对待一个正经后端系统一样，从零构建一套可维护、可观测、可扩展的AI工程体系。它适合谁？适合那些已经会写Python、懂基本的服务端开发，但一碰到模型部署、推理优化、GPU资源管理就心里没底的中级工程师。也适合技术负责人，用来梳理团队在AI工程化上的能力缺口。

我自己是从传统后端转过来的，踩过的坑包括但不限于：把模型权重直接塞进Docker镜像导致镜像12个G、用同步阻塞的方式调推理接口把整个服务拖死、显存碎片化导致跑了两天突然OOM。这篇文章就是把这些经验拆开揉碎，从架构设计到代码落地，给你一套能直接抄的作业。

2. 整体架构设计：为什么我不建议一上来就上Kubernetes

2.1 从单机到集群的演进逻辑

很多团队一提到AI工程化，第一反应就是上K8s、上Triton、上Ray。我不否认这些是好东西，但如果你连单机上的推理服务都没跑稳，上这些只会让问题更难排查。我的建议是分三个阶段走：

阶段一：单机裸服务。一台带GPU的机器，一个FastAPI或Tornado服务，模型加载在进程启动时完成。这个阶段的目标是跑通"请求进来→预处理→推理→后处理→返回"的完整链路，并且能压测出单实例的QPS上限和P99延迟。

阶段二：多实例+负载均衡。当单实例扛不住时，先别急着上K8s。用Nginx做反向代理，起多个服务进程，每个进程绑定不同的GPU或共享同一块GPU的不同显存区域。这个阶段要解决的是进程间显存隔离和请求分发策略。

阶段三：容器编排。到了这一步，你才需要K8s来做自动扩缩容、滚动更新和故障转移。但注意，AI服务的扩缩容和普通Web服务完全不同——模型加载可能要几十秒甚至几分钟，所以HPA的指标不能只看CPU，得看请求队列长度或GPU利用率。

我见过太多团队跳过阶段一直接上阶段三，结果一个简单的显存泄漏问题排查了两周。因为K8s把日志、监控、网络都抽象了一层，你很难直接看到底层发生了什么。

2.2 推理框架选型：别被 benchmark 带偏

选推理框架的时候，网上到处都是"某框架比某框架快3倍"的benchmark。但实际选型时，吞吐量只是其中一个维度。我整理了一个更实用的对比表：

我的经验是：如果你做的是传统CV模型或小规模NLP模型，FastAPI+原生PyTorch足够，因为可控性最强，出问题好排查。如果做的是大语言模型推理，vLLM的PagedAttention机制确实能显著提升显存利用率和吞吐，但它的多模型管理能力弱，适合单模型高并发场景。Triton适合那种同时跑十几个不同模型的场景，比如推荐系统里召回、粗排、精排各一个模型。

注意：选型时一定要用你自己的真实请求做压测，不要信官方benchmark。因为官方benchmark通常用的是最优输入长度和batch size，而你的实际请求可能长短不一、分布极不均匀。

2.3 目录结构设计：为可维护性买单

一个AI工程项目最容易烂掉的地方就是目录结构。我见过把所有代码塞进一个main.py的，也见过把模型文件、配置文件、日志文件混在一起的。下面是我用了三年、迭代了五个版本后的目录结构：

ai-service/ ├── configs/ │ ├── base.yaml │ ├── dev.yaml │ └── prod.yaml ├── src/ │ ├── api/ │ │ ├── routes.py │ │ └── schemas.py │ ├── core/ │ │ ├── model_loader.py │ │ ├── inference.py │ │ └── preprocess.py │ ├── utils/ │ │ ├── logger.py │ │ └── metrics.py │ └── main.py ├── models/ │ └── .gitkeep ├── tests/ │ ├── test_api.py │ └── test_inference.py ├── Dockerfile ├── requirements.txt └── README.md

关键点在于：models/目录只放一个.gitkeep，模型权重通过环境变量或配置指定路径，绝对不要提交到Git。configs/用YAML做分层配置，base.yaml放通用配置，dev.yaml和prod.yaml覆盖差异项。src/core/里把模型加载、推理逻辑、预处理拆开，这样单元测试可以单独测预处理逻辑，不需要加载模型。

3. 核心细节解析：模型加载、显存管理与请求调度

3.1 模型加载的三种姿势与选择依据

模型加载看起来简单，其实有很多门道。我总结下来有三种方式：

方式一：进程启动时加载。在FastAPI的startup事件里加载模型，全局变量持有模型引用。优点是请求处理时没有加载开销，延迟稳定。缺点是启动慢，而且如果模型加载失败，整个服务起不来。

方式二：懒加载。第一次请求进来时才加载模型。优点是启动快，适合开发调试。缺点是第一个请求延迟极高，而且如果并发请求同时到达，可能触发多次加载。

方式三：预加载+健康检查。启动时异步加载模型，同时提供一个/health接口，只有模型加载完成后才返回200。负载均衡器根据健康检查结果决定是否转发流量。

生产环境我强烈推荐方式三。具体实现上，用asyncio.create_task在启动时触发加载，用一个全局的Event标记加载状态：

import asyncio from contextlib import asynccontextmanager from fastapi import FastAPI model_ready = asyncio.Event() model = None async def load_model(): global model # 模拟耗时加载 await asyncio.sleep(5) model = {"name": "my-model"} model_ready.set() @asynccontextmanager async def lifespan(app: FastAPI): asyncio.create_task(load_model()) yield app = FastAPI(lifespan=lifespan) @app.get("/health") async def health(): if model_ready.is_set(): return {"status": "ok"} return {"status": "loading"}, 503

这样K8s的readiness probe会一直失败直到模型加载完成，流量不会打到还没准备好的实例上。

3.2 显存管理的几个关键参数

显存是AI工程里最稀缺的资源。很多人只知道torch.cuda.memory_allocated()，但不知道还有几个关键参数需要关注：

• torch.cuda.memory_reserved()：PyTorch缓存分配器保留的显存，包括已分配和空闲的。
• torch.cuda.max_memory_allocated()：峰值显存占用，用来评估OOM风险。
• torch.cuda.memory_summary()：完整的显存报告，包括碎片情况。

我踩过最大的坑是：模型推理时显存够用，但跑了一段时间后突然OOM。原因是PyTorch的缓存分配器会产生显存碎片，虽然总空闲显存够，但没有连续的大块显存可用。解决办法有两个：一是设置PYTORCH_CUDA_ALLOC_CONF=expandable_segments:True，让分配器支持可扩展段；二是定期调用torch.cuda.empty_cache()，但注意这个操作会同步等待所有CUDA操作完成，频繁调用会严重影响性能。

实操心得：我通常会在服务里加一个后台任务，每处理完N个请求后检查一次显存碎片率（reserved - allocated/reserved），如果超过30%就触发一次empty_cache。N的值根据请求频率调整，一般设1000到5000之间。

3.3 请求调度的动态批处理实现

动态批处理是提升GPU利用率最有效的手段。原理很简单：把多个请求攒在一起，凑成一个batch送进模型，这样GPU的并行计算能力才能充分利用。但实现起来有几个坑：

坑一：攒批超时。如果只来了一个请求，你等不等？等多久？我的做法是设置一个最大等待时间，比如50ms。如果50ms内没有新请求，就单条推理。

坑二：batch内长度不一致。NLP模型通常要求输入长度一致，需要padding。padding太多会浪费计算资源。解决办法是按长度分桶，把长度相近的请求放在同一个batch里。

坑三：超长请求阻塞。如果一个batch里有一个超长请求，整个batch的延迟都会被拉高。我的做法是设置最大token数限制，超过限制的请求单独处理或直接拒绝。

下面是一个简化的动态批处理实现：

import asyncio from collections import deque class DynamicBatcher: def __init__(self, max_batch_size=8, max_wait_ms=50): self.max_batch_size = max_batch_size self.max_wait = max_wait_ms / 1000 self.queue = deque() self.lock = asyncio.Lock() async def add_request(self, request): future = asyncio.Future() async with self.lock: self.queue.append((request, future)) if len(self.queue) >= self.max_batch_size: await self._process_batch() return await future async def _process_batch(self): batch = [] futures = [] while self.queue and len(batch) < self.max_batch_size: req, fut = self.queue.popleft() batch.append(req) futures.append(fut) # 实际推理逻辑 results = await self._infer(batch) for fut, res in zip(futures, results): fut.set_result(res)

这个实现还有很多优化空间，比如用asyncio.wait_for做超时控制、用优先级队列处理不同优先级的请求。但核心思路就是：攒批、推理、分发结果。

4. 实操过程：从零搭建一个可用的推理服务

4.1 环境准备与依赖锁定

第一步永远是环境。我强烈建议用conda或venv创建独立环境，然后用pip-compile锁定依赖版本。AI项目的依赖冲突比普通后端项目严重得多，因为PyTorch、CUDA、cuDNN之间的版本兼容性非常严格。

# 创建环境 conda create -n ai-service python=3.10 conda activate ai-service # 安装PyTorch（根据CUDA版本选择） pip install torch==2.1.0 --index-url https://download.pytorch.org/whl/cu118 # 安装其他依赖 pip install fastapi uvicorn[standard] pydantic pyyaml # 锁定依赖 pip freeze > requirements.txt

注意：requirements.txt里不要直接写torch，要写清楚版本和CUDA版本。否则在不同机器上pip install可能装到CPU版本，导致推理速度慢几十倍。

4.2 配置文件设计与加载

配置文件用YAML，支持环境变量覆盖。这样本地开发、测试环境、生产环境可以用同一套代码，只改配置。

# configs/base.yaml model: name: "bert-base" path: "/models/bert-base" max_batch_size: 8 max_seq_length: 512 server: host: "0.0.0.0" port: 8000 workers: 1 logging: level: "INFO" format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"

加载配置的代码：

import os import yaml def load_config(env="dev"): with open("configs/base.yaml") as f: config = yaml.safe_load(f) env_file = f"configs/{env}.yaml" if os.path.exists(env_file): with open(env_file) as f: env_config = yaml.safe_load(f) config = deep_merge(config, env_config) # 环境变量覆盖 if os.getenv("MODEL_PATH"): config["model"]["path"] = os.getenv("MODEL_PATH") return config

4.3 推理核心逻辑实现

推理逻辑我拆成了三个函数：preprocess、infer、postprocess。这样每个函数都可以单独测试，也方便替换。

import torch from transformers import AutoTokenizer, AutoModel class InferenceEngine: def __init__(self, config): self.config = config self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") self.tokenizer = AutoTokenizer.from_pretrained(config["model"]["path"]) self.model = AutoModel.from_pretrained(config["model"]["path"]) self.model.to(self.device) self.model.eval() def preprocess(self, texts): encoded = self.tokenizer( texts, padding=True, truncation=True, max_length=self.config["model"]["max_seq_length"], return_tensors="pt" ) return {k: v.to(self.device) for k, v in encoded.items()} @torch.no_grad() def infer(self, inputs): outputs = self.model(**inputs) return outputs.last_hidden_state[:, 0, :].cpu().numpy() def postprocess(self, outputs): return outputs.tolist() def predict(self, texts): inputs = self.preprocess(texts) outputs = self.infer(inputs) return self.postprocess(outputs)

关键点：torch.no_grad()一定要加，否则PyTorch会构建计算图，显存占用翻倍。model.eval()也要加，否则Dropout和BatchNorm会处于训练模式，导致推理结果不稳定。

4.4 API层与中间件

API层用FastAPI，加上请求日志、耗时统计、异常处理三个中间件。

import time import logging from fastapi import FastAPI, Request from fastapi.responses import JSONResponse app = FastAPI() logger = logging.getLogger(__name__) @app.middleware("http") async def log_requests(request: Request, call_next): start = time.time() response = await call_next(request) duration = time.time() - start logger.info(f"{request.method} {request.url.path} - {response.status_code} - {duration:.3f}s") response.headers["X-Process-Time"] = str(duration) return response @app.exception_handler(Exception) async def global_exception_handler(request, exc): logger.error(f"Unhandled exception: {exc}", exc_info=True) return JSONResponse(status_code=500, content={"detail": "Internal server error"}) @app.post("/predict") async def predict(request: PredictRequest): results = engine.predict(request.texts) return {"results": results}

4.5 压测与性能调优

服务写完了，下一步是压测。我用的是locust，因为它支持分布式压测，而且可以自定义请求分布。

from locust import HttpUser, task, between class InferenceUser(HttpUser): wait_time = between(0.1, 0.5) @task def predict(self): self.client.post("/predict", json={ "texts": ["这是一个测试句子"] * 4 })

压测时重点关注三个指标：QPS、P99延迟、GPU利用率。如果GPU利用率低于50%，说明请求不够密集，需要增大batch size或增加并发。如果P99延迟远高于P50，说明有长尾请求，需要检查是否有超长输入或显存碎片。

我实测下来，一块A100 40G跑BERT-base，batch size=8时QPS能到120左右，P99延迟在80ms。如果batch size降到1，QPS只有30左右，GPU利用率不到20%。这就是动态批处理的价值。

5. 常见问题与排查技巧实录

5.1 显存OOM的排查路径

OOM是AI工程里最常见的错误。排查思路如下：

1. 先看torch.cuda.memory_summary()，确认是模型本身太大还是碎片问题。
2. 如果是模型太大，考虑量化（FP16或INT8）或模型并行。
3. 如果是碎片问题，设置PYTORCH_CUDA_ALLOC_CONF=expandable_segments:True。
4. 如果还不行，检查是否有未释放的中间变量，比如在循环里不断append tensor到list。

我踩过的坑：在预处理里把tokenizer的输出存到了一个全局list里做缓存，结果缓存越来越大，最后OOM。后来改成用LRU缓存，限制最大条目数。

5.2 推理结果不稳定的排查

有时候同一个输入，两次推理结果不一样。原因通常有三个：

• 模型没有设置eval()模式，Dropout还在起作用。
• 没有用torch.no_grad()，计算图影响了随机数生成器的状态。
• 输入没有做padding到固定长度，不同batch的padding位置不同。

排查方法：固定随机种子torch.manual_seed(42)，然后连续推理同一个输入10次，看结果是否一致。

5.3 服务启动慢的优化

模型加载慢是常态，但可以通过以下方式优化：

• 使用torch.jit.load加载TorchScript模型，比加载原生PyTorch模型快30%左右。
• 使用safetensors格式代替pytorch_model.bin，加载速度更快且更安全。
• 如果模型在远程存储上，先用wget或aws s3 cp下载到本地，再从本地加载。

5.4 常见问题速查表

6. 工程化扩展：监控、日志与持续迭代

6.1 监控指标设计

AI服务的监控和普通Web服务不同，除了QPS、延迟、错误率，还要关注：

• GPU利用率：低于30%说明资源浪费，高于90%说明可能成为瓶颈。
• 显存占用：持续增长说明有泄漏。
• 批处理大小分布：如果大部分请求都是batch size=1，说明攒批策略有问题。
• 预处理/推理/后处理耗时占比：如果预处理占了大头，说明tokenizer是瓶颈。

我用prometheus_client暴露指标，用Grafana做面板。关键代码：

from prometheus_client import Counter, Histogram, Gauge REQUEST_COUNT = Counter("inference_requests_total", "Total requests") REQUEST_LATENCY = Histogram("inference_latency_seconds", "Request latency") GPU_MEMORY = Gauge("gpu_memory_used_bytes", "GPU memory used") BATCH_SIZE = Histogram("inference_batch_size", "Batch size distribution")

6.2 日志规范

日志要结构化，方便后续用ELK或Loki做检索。我通常用JSON格式：

import json import logging class JsonFormatter(logging.Formatter): def format(self, record): log = { "time": self.formatTime(record), "level": record.levelname, "message": record.getMessage(), "module": record.module, } if hasattr(record, "extra"): log.update(record.extra) return json.dumps(log)

请求日志里要包含request_id、batch_size、preprocess_time、infer_time、postprocess_time，这样出问题时可以快速定位是哪个环节慢了。

6.3 模型热更新方案

模型热更新是个高级话题，但很有用。基本思路是：新模型加载到新的进程或新的GPU显存区域，然后通过负载均衡器切换流量。具体实现可以用蓝绿部署或金丝雀发布。

我自己的做法是：在服务里维护两个模型槽位，active和standby。新模型加载到standby，加载完成后原子切换active指针。切换过程中，正在处理的请求继续用旧模型，新请求用新模型。

class ModelManager: def __init__(self): self.active = None self.standby = None self.lock = threading.Lock() def load_new_model(self, path): new_model = load_model(path) with self.lock: self.standby = new_model def switch(self): with self.lock: self.active, self.standby = self.standby, self.active

这个方案的关键是切换要原子，而且旧模型不能立即释放，要等所有正在处理的请求完成后再释放。

6.4 持续迭代的节奏

AI工程不是一次性的，模型会更新、请求分布会变化、硬件会升级。我的建议是：

• 每周看一次监控面板，关注趋势而不是绝对值。
• 每月做一次压测，确认当前配置还能扛住峰值。
• 每季度review一次依赖版本，该升级就升级，但不要追最新版。

踩过最大的坑是：生产环境跑了一个两年没更新的PyTorch版本，结果新来的同事用新版本训练了一个模型，保存的权重在老版本上加载不了。所以版本管理一定要严格，requirements.txt要提交到Git，Docker镜像要打tag。

这个项目后续还可以扩展的方向包括：多GPU推理、模型量化、请求优先级队列、A/B测试框架。但那是另一个话题了，先把单机服务跑稳，比什么都重要。