尧图网站建设 
1. 从零搭建AI工程能力一个后端老兵的踩坑与重构实录ai-engineering-from-scratch这个标题第一次看到的时候我愣了一下。不是因为它有多高深恰恰相反——它太直白了直白到像一句废话。但仔细想想过去两年我面过不下三十个自称做过AI项目的候选人能从头讲清楚一个推理服务怎么从裸机部署到扛住并发的人一只手数得过来。大部分人所谓的AI工程其实是在Jupyter Notebook里调通了API然后写了个Flask包一层就上线了。真到了线上显存泄漏、请求排队、模型热更新、GPU利用率忽高忽低这些问题一出来基本就抓瞎。所以这个标题背后真正想解决的问题不是怎么调用大模型API而是怎么像对待一个正经后端系统一样从零构建一套可维护、可观测、可扩展的AI工程体系。它适合谁适合那些已经会写Python、懂基本的服务端开发但一碰到模型部署、推理优化、GPU资源管理就心里没底的中级工程师。也适合技术负责人用来梳理团队在AI工程化上的能力缺口。我自己是从传统后端转过来的踩过的坑包括但不限于把模型权重直接塞进Docker镜像导致镜像12个G、用同步阻塞的方式调推理接口把整个服务拖死、显存碎片化导致跑了两天突然OOM。这篇文章就是把这些经验拆开揉碎从架构设计到代码落地给你一套能直接抄的作业。2. 整体架构设计为什么我不建议一上来就上Kubernetes2.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。但实际选型时吞吐量只是其中一个维度。我整理了一个更实用的对比表维度FastAPI原生PyTorchTorchServeTriton Inference ServervLLM上手难度低中中高中动态批处理需自己实现支持支持原生支持多模型管理手动支持强弱显存优化无一般较好极好适合场景原型验证中小规模多模型混合大模型推理我的经验是如果你做的是传统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(lifespanlifespan) 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_CONFexpandable_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_size8, max_wait_ms50): 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 python3.10 conda activate ai-service # 安装PyTorch根据CUDA版本选择 pip install torch2.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(envdev): with open(configs/base.yaml) as f: config yaml.safe_load(f) env_file fconfigs/{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 config4.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, paddingTrue, truncationTrue, max_lengthself.config[model][max_seq_length], return_tensorspt ) 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(fUnhandled exception: {exc}, exc_infoTrue) return JSONResponse(status_code500, 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-basebatch size8时QPS能到120左右P99延迟在80ms。如果batch size降到1QPS只有30左右GPU利用率不到20%。这就是动态批处理的价值。5. 常见问题与排查技巧实录5.1 显存OOM的排查路径OOM是AI工程里最常见的错误。排查思路如下先看torch.cuda.memory_summary()确认是模型本身太大还是碎片问题。如果是模型太大考虑量化FP16或INT8或模型并行。如果是碎片问题设置PYTORCH_CUDA_ALLOC_CONFexpandable_segments:True。如果还不行检查是否有未释放的中间变量比如在循环里不断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 常见问题速查表问题现象可能原因排查方法解决方案OOM显存碎片memory_summaryexpandable_segments结果不稳定未设eval检查model.trainingmodel.eval()延迟高无批处理看GPU利用率动态批处理启动慢模型加载计时加载过程TorchScript/safetensorsQPS低同步阻塞看请求队列异步多进程6. 工程化扩展监控、日志与持续迭代6.1 监控指标设计AI服务的监控和普通Web服务不同除了QPS、延迟、错误率还要关注GPU利用率低于30%说明资源浪费高于90%说明可能成为瓶颈。显存占用持续增长说明有泄漏。批处理大小分布如果大部分请求都是batch size1说明攒批策略有问题。预处理/推理/后处理耗时占比如果预处理占了大头说明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要提交到GitDocker镜像要打tag。这个项目后续还可以扩展的方向包括多GPU推理、模型量化、请求优先级队列、A/B测试框架。但那是另一个话题了先把单机服务跑稳比什么都重要。 从零搭建AI推理服务：后端老兵的工程化踩坑与重构实录
1. 从零搭建AI工程能力一个后端老兵的踩坑与重构实录ai-engineering-from-scratch这个标题第一次看到的时候我愣了一下。不是因为它有多高深恰恰相反——它太直白了直白到像一句废话。但仔细想想过去两年我面过不下三十个自称做过AI项目的候选人能从头讲清楚一个推理服务怎么从裸机部署到扛住并发的人一只手数得过来。大部分人所谓的AI工程其实是在Jupyter Notebook里调通了API然后写了个Flask包一层就上线了。真到了线上显存泄漏、请求排队、模型热更新、GPU利用率忽高忽低这些问题一出来基本就抓瞎。所以这个标题背后真正想解决的问题不是怎么调用大模型API而是怎么像对待一个正经后端系统一样从零构建一套可维护、可观测、可扩展的AI工程体系。它适合谁适合那些已经会写Python、懂基本的服务端开发但一碰到模型部署、推理优化、GPU资源管理就心里没底的中级工程师。也适合技术负责人用来梳理团队在AI工程化上的能力缺口。我自己是从传统后端转过来的踩过的坑包括但不限于把模型权重直接塞进Docker镜像导致镜像12个G、用同步阻塞的方式调推理接口把整个服务拖死、显存碎片化导致跑了两天突然OOM。这篇文章就是把这些经验拆开揉碎从架构设计到代码落地给你一套能直接抄的作业。2. 整体架构设计为什么我不建议一上来就上Kubernetes2.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。但实际选型时吞吐量只是其中一个维度。我整理了一个更实用的对比表维度FastAPI原生PyTorchTorchServeTriton Inference ServervLLM上手难度低中中高中动态批处理需自己实现支持支持原生支持多模型管理手动支持强弱显存优化无一般较好极好适合场景原型验证中小规模多模型混合大模型推理我的经验是如果你做的是传统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(lifespanlifespan) 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_CONFexpandable_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_size8, max_wait_ms50): 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 python3.10 conda activate ai-service # 安装PyTorch根据CUDA版本选择 pip install torch2.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(envdev): with open(configs/base.yaml) as f: config yaml.safe_load(f) env_file fconfigs/{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 config4.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, paddingTrue, truncationTrue, max_lengthself.config[model][max_seq_length], return_tensorspt ) 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(fUnhandled exception: {exc}, exc_infoTrue) return JSONResponse(status_code500, 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-basebatch size8时QPS能到120左右P99延迟在80ms。如果batch size降到1QPS只有30左右GPU利用率不到20%。这就是动态批处理的价值。5. 常见问题与排查技巧实录5.1 显存OOM的排查路径OOM是AI工程里最常见的错误。排查思路如下先看torch.cuda.memory_summary()确认是模型本身太大还是碎片问题。如果是模型太大考虑量化FP16或INT8或模型并行。如果是碎片问题设置PYTORCH_CUDA_ALLOC_CONFexpandable_segments:True。如果还不行检查是否有未释放的中间变量比如在循环里不断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 常见问题速查表问题现象可能原因排查方法解决方案OOM显存碎片memory_summaryexpandable_segments结果不稳定未设eval检查model.trainingmodel.eval()延迟高无批处理看GPU利用率动态批处理启动慢模型加载计时加载过程TorchScript/safetensorsQPS低同步阻塞看请求队列异步多进程6. 工程化扩展监控、日志与持续迭代6.1 监控指标设计AI服务的监控和普通Web服务不同除了QPS、延迟、错误率还要关注GPU利用率低于30%说明资源浪费高于90%说明可能成为瓶颈。显存占用持续增长说明有泄漏。批处理大小分布如果大部分请求都是batch size1说明攒批策略有问题。预处理/推理/后处理耗时占比如果预处理占了大头说明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要提交到GitDocker镜像要打tag。这个项目后续还可以扩展的方向包括多GPU推理、模型量化、请求优先级队列、A/B测试框架。但那是另一个话题了先把单机服务跑稳比什么都重要。