从零搭建AI工程体系：模型推理服务架构设计与性能优化实战

1. 从零搭建AI工程体系，我为什么劝你别急着调包

"ai-engineering-from-scratch"这个标题，第一次看到的时候我愣了一下。市面上讲AI的教程铺天盖地，但绝大多数都在教你import torch然后跑一个预训练模型，或者调个API把结果打印出来。真正从零开始、把AI工程当作一门系统工程来拆解的内容，少得可怜。

我自己在这个方向上踩了差不多两年的坑。最开始我也是那种"能跑通就行"的心态，模型加载出来、推理结果对了，就觉得万事大吉。直到有一次线上服务在高峰期直接雪崩，排查了整整一个通宵才发现，问题根本不在模型本身，而是整个工程链路上有七八个环节都在裸奔——没有批处理、没有显存管理、没有降级策略、没有输入校验。那次之后我才真正意识到，AI工程和AI算法是两码事。

这个项目标题的核心价值在于"from scratch"这四个字。它不是让你从零训练一个大模型，那不现实也没必要；它说的是从零构建一套能支撑AI应用稳定运行的工程体系。这里面涉及的东西非常具体：数据管道怎么设计、推理服务怎么部署、显存怎么管理、并发怎么处理、监控怎么做、故障怎么降级。适合谁来参考？我认为是三类人：一是刚入行做AI应用开发、只会调包但不懂工程的新人；二是从传统后端转过来做AI服务、对模型推理特性不熟悉的工程师；三是自己做一些小项目、想把东西做得更靠谱的独立开发者。

接下来的内容，我会按照我自己实际搭建过的一套流程来展开，从整体设计思路到具体实现细节，再到踩过的坑和排查技巧，尽量把每个环节的"为什么"讲清楚。你不需要全部照搬，但至少能知道一个完整的AI工程体系应该长什么样。

2. 整体架构设计与技术选型思路

2.1 为什么不能"模型加载完就完事"

很多人对AI工程的认知停留在"把模型跑起来"这个层面。我刚开始也是这样，写个Flask接口，加载模型，收到请求就推理，返回结果。这套东西在本地测试的时候跑得挺好，一上线就出问题。

根本原因在于，模型推理和普通的CRUD接口有本质区别。普通接口处理一个请求可能就几毫秒，消耗的资源可以忽略不计；但一次模型推理可能占用几百MB甚至几个GB的显存，耗时从几十毫秒到几秒不等。这意味着你不能用传统的"来一个请求处理一个"的模式，否则并发稍微一上来，显存直接爆掉，整个服务挂死。

所以整体架构设计的第一个核心原则就是：把推理过程当作一种稀缺资源来管理。这跟数据库连接池的思路类似，但比连接池复杂得多，因为显存不像数据库连接那样可以简单地创建和销毁。

我最终采用的架构大致分为四层：

• 接入层：负责请求接收、参数校验、限流和鉴权。这一层不碰模型，只做前置处理。
• 调度层：核心中的核心。负责请求排队、批处理组装、优先级调度、超时控制。
• 推理层：实际执行模型推理的Worker池。每个Worker持有模型实例，从调度层拉取任务。
• 监控层：贯穿所有层，采集延迟、吞吐、显存占用、错误率等指标。

这四层看起来简单，但每一层都有大量细节需要处理。比如调度层的批处理策略，如果设计不好，要么吞吐上不去，要么延迟飙升。后面我会详细展开。

2.2 技术选型：为什么选这些而不是那些

在技术选型上，我走过不少弯路。最开始用的是Flask + Gunicorn，后来换成FastAPI + Uvicorn，再后来引入了消息队列做调度。每一步切换都有具体的触发原因。

Web框架选FastAPI而不是Flask，原因很直接：FastAPI原生支持异步，而AI推理服务经常需要处理IO等待（比如等待GPU计算完成、等待下游服务响应）。用Flask的话，每个请求占一个线程，并发量一上来线程池就满了。FastAPI的异步模型配合Uvicorn的worker机制，能更高效地利用CPU资源。另外FastAPI自带Pydantic做请求校验，省去了手写校验逻辑的麻烦。

调度层用Redis而不是内存队列，这个决策花了我不少时间纠结。内存队列（比如Python的queue.Queue）速度更快，但问题是无法跨进程共享。当你有多个Worker进程时，内存队列就没法用了。Redis的List或者Stream结构可以做跨进程的任务分发，而且支持持久化，服务重启后任务不丢。虽然引入Redis增加了一个依赖，但从可扩展性和可靠性角度看，这个代价是值得的。

推理层用进程池而不是线程池，这是被Python的GIL逼的。模型推理是计算密集型任务，多线程在Python里根本跑不出并行效果。用多进程的话，每个进程独立持有模型实例，虽然显存占用会翻倍，但能真正实现并行推理。当然，进程数不能无限加，要根据显存大小来算。

下面这张表是我在实际项目中对比过的几种方案，供你参考：

我最终选择的是"多进程+批处理"的方案，动态批处理因为实现复杂度太高，在小规模场景下收益不明显，暂时没有引入。

2.3 目录结构设计：让代码可维护

一个容易被忽视但非常重要的点是目录结构。我见过太多AI项目的代码全部堆在一个main.py里，几百行混在一起，改一个地方要翻半天。从零搭建工程体系，目录结构必须一开始就设计好。

我目前使用的结构大致如下：

project/ ├── config/ # 配置文件 │ ├── model.yaml # 模型相关配置 │ ├── server.yaml # 服务相关配置 │ └── logging.yaml # 日志配置 ├── src/ │ ├── api/ # 接入层 │ │ ├── routes.py │ │ ├── middleware.py │ │ └── schemas.py │ ├── scheduler/ # 调度层 │ │ ├── queue.py │ │ ├── batcher.py │ │ └── dispatcher.py │ ├── inference/ # 推理层 │ │ ├── worker.py │ │ ├── model_loader.py │ │ └── postprocess.py │ ├── monitor/ # 监控层 │ │ ├── metrics.py │ │ └── health.py │ └── utils/ # 工具函数 │ ├── logger.py │ └── timer.py ├── tests/ # 测试 ├── scripts/ # 运维脚本 ├── Dockerfile └── requirements.txt

这个结构的好处是职责清晰。接入层只关心请求怎么进来，调度层只关心任务怎么分配，推理层只关心模型怎么跑。每层之间的接口通过明确的函数签名约定，改一层不会影响其他层。

注意：目录结构不是越细越好。我见过有人把每个函数都拆成一个文件，结果找代码比写代码还费时间。一般来说，一个模块内的文件数控制在5到8个比较合适。

3. 核心模块拆解与关键实现细节

3.1 模型加载：别小看这一步

模型加载看起来简单，model = load_model(path)就完事了。但在工程环境里，这一步有很多讲究。

首先是加载时机。我试过两种方案：一种是服务启动时一次性加载所有模型；另一种是懒加载，第一次请求某个模型时才加载。前者启动慢但请求响应快，后者启动快但首次请求延迟高。最终我选择了启动时加载，因为生产环境更看重请求延迟的稳定性，启动慢几十秒可以接受。

其次是显存预分配。PyTorch默认是动态分配显存的，这会导致显存碎片化，长时间运行后可能出现"明明还有显存但分配不出来"的情况。解决办法是设置PYTORCH_CUDA_ALLOC_CONF环境变量，启用显存池化：

export PYTORCH_CUDA_ALLOC_CONF=max_split_size_mb:128

这个参数的含义是，当需要分配一块显存时，如果现有的空闲块大于128MB且足够使用，就直接复用，而不是向系统申请新的。这样可以有效减少碎片。

还有一个容易踩的坑是模型加载的线程安全问题。如果你用多进程方案，每个进程独立加载模型，互不干扰，没问题。但如果你用多线程，多个线程共享同一个模型实例，就必须确保模型在推理时是线程安全的。PyTorch的模型在推理模式下（model.eval()）通常是线程安全的，但如果你用了自定义的Layer或者有状态的操作，就需要加锁。

我的做法是在model_loader.py里封装一个ModelManager类，负责模型的加载、缓存和释放：

import torch import logging from pathlib import Path logger = logging.getLogger(__name__) class ModelManager: def __init__(self, model_dir: str, device: str = "cuda"): self.model_dir = Path(model_dir) self.device = device self._models = {} def load(self, model_name: str): if model_name in self._models: return self._models[model_name] model_path = self.model_dir / model_name if not model_path.exists(): raise FileNotFoundError(f"Model not found: {model_path}") logger.info(f"Loading model: {model_name}") model = torch.load(model_path, map_location=self.device) model.eval() # 预热：跑一次空推理，触发CUDA初始化 with torch.no_grad(): dummy_input = torch.zeros(1, 3, 224, 224).to(self.device) _ = model(dummy_input) self._models[model_name] = model logger.info(f"Model loaded: {model_name}") return model def unload(self, model_name: str): if model_name in self._models: del self._models[model_name] torch.cuda.empty_cache()

这里有个细节值得说：预热。模型加载后第一次推理会特别慢，因为CUDA需要做一系列初始化工作。如果不预热，第一个真实请求的延迟会高得离谱。预热就是用一个假数据跑一次推理，把初始化成本提前消化掉。

3.2 请求调度：批处理是吞吐量的关键

批处理是AI工程里提升吞吐量最有效的手段，没有之一。原理很简单：GPU擅长并行计算，一次处理16个样本和一次处理1个样本，耗时可能只差20%。如果不做批处理，GPU的计算能力就被严重浪费了。

但批处理不是简单地把请求攒在一起就行。核心问题是：攒多久？攒多少个？

攒太久，延迟高；攒太少，吞吐上不去。我最初设的是"攒够8个或者等50毫秒"，实际跑下来发现两个问题：一是低峰期请求少，每个请求都要等满50毫秒，延迟白白增加；二是高峰期请求多，8个一批处理不过来，队列越积越长。

后来改成了动态批处理策略，核心逻辑是：

• 维护一个待处理队列。
• 调度线程每隔一个很短的时间窗口（比如5毫秒）检查一次队列。
• 如果队列长度超过阈值（比如16），立即组批。
• 如果队列长度不足，但最早的那个请求已经等了超过最大等待时间（比如30毫秒），也立即组批。
• 批大小不超过模型能处理的上限。

用伪代码表示大概是：

import time from collections import deque class DynamicBatcher: def __init__(self, max_batch_size=32, max_wait_ms=30, check_interval_ms=5): self.queue = deque() self.max_batch_size = max_batch_size self.max_wait = max_wait_ms / 1000.0 self.check_interval = check_interval_ms / 1000.0 def add(self, request): self.queue.append((time.time(), request)) def next_batch(self): if not self.queue: return [] now = time.time() batch = [] while self.queue and len(batch) < self.max_batch_size: enqueue_time, request = self.queue[0] waited = now - enqueue_time if len(batch) == 0 or waited >= self.max_wait or len(self.queue) >= self.max_batch_size: self.queue.popleft() batch.append(request) else: break return batch

这个策略的好处是自适应：高峰期批大、延迟低；低峰期批小、不空等。实测下来，相比固定批处理，吞吐量提升了大约40%，而P99延迟只增加了不到10毫秒。

实操心得：批处理的大小上限不是拍脑袋定的，要根据模型的实际显存占用和计算特性来测。我的做法是从batch_size=1开始，逐步翻倍，观察显存占用和单批耗时，找到"显存快满但还没满"的那个点，然后取它的80%作为上限，留出余量。

3.3 显存管理：最容易被忽视的雷区

显存管理是AI工程里最容易出问题的地方，没有之一。我踩过的坑包括但不限于：显存泄漏导致服务跑几天就挂、显存碎片导致大batch分配失败、多进程显存争抢导致OOM。

显存泄漏是最隐蔽的问题。PyTorch的自动求导机制会保留计算图，如果你在推理时忘了加torch.no_grad()，每次推理都会积累计算图，显存越用越多。这个问题的排查方法是定期打印显存占用：

import torch def print_gpu_memory(prefix=""): if torch.cuda.is_available(): allocated = torch.cuda.memory_allocated() / 1024**2 reserved = torch.cuda.memory_reserved() / 1024**2 print(f"{prefix} Allocated: {allocated:.1f}MB, Reserved: {reserved:.1f}MB")

如果发现allocated持续增长不下降，基本可以确定是泄漏。解决办法就是确保所有推理代码都在with torch.no_grad():块里。

显存碎片是另一个头疼的问题。表现是torch.cuda.memory_allocated()显示还有很多空闲，但分配新张量时却报OOM。这是因为空闲显存不连续，无法满足大块分配请求。解决办法前面提过，设置PYTORCH_CUDA_ALLOC_CONF。另外，尽量避免频繁创建和销毁大张量，能复用就复用。

多进程显存争抢的解决办法是给每个进程设定显存上限。PyTorch提供了torch.cuda.set_per_process_memory_fraction()：

import torch # 假设有4个进程，每个进程最多用1/4的显存 torch.cuda.set_per_process_memory_fraction(0.25, device=0)

这个设置要在进程启动后、加载模型前调用。注意，这个比例是相对于GPU总显存的，不是相对于当前可用显存。设置得太小会导致模型加载失败，太大则起不到隔离作用。我的经验是留出20%的余量，比如4个进程的话，每个设0.2而不是0.25。

3.4 输入校验与预处理：把问题挡在推理之前

很多线上事故的根源不是模型本身，而是输入数据有问题。我遇到过的情况包括：图片格式不对导致解码失败、文本长度超限导致显存爆掉、请求参数缺失导致KeyError。

这些问题的共同点是：它们本可以在进入推理层之前就被拦截。所以接入层的输入校验非常重要。

我的做法是在Pydantic的Schema里定义严格的校验规则：

from pydantic import BaseModel, Field, validator from typing import List, Optional class InferenceRequest(BaseModel): model_name: str = Field(..., min_length=1, max_length=64) inputs: List[str] = Field(..., min_items=1, max_items=32) max_length: Optional[int] = Field(default=128, ge=1, le=512) @validator("inputs") def check_input_length(cls, v): for item in v: if len(item) > 10000: raise ValueError("Single input too long") return v @validator("model_name") def check_model_name(cls, v): allowed = {"text-classifier", "sentiment", "ner"} if v not in allowed: raise ValueError(f"Unknown model: {v}") return v

Pydantic会在请求进入路由函数之前自动执行这些校验，不通过的直接返回422错误，根本不会走到推理层。这样既保护了推理层，也给调用方提供了清晰的错误信息。

预处理环节还有一个容易忽视的点是数据类型的统一。比如图片输入，有的客户端传base64，有的传URL，有的传二进制。如果不做统一，推理层就要处理各种格式，代码会变得很乱。我的做法是在接入层就把所有输入统一转换成模型需要的格式（比如numpy数组或torch张量），推理层只处理统一格式。

4. 完整实操流程：从零到跑通

4.1 环境准备与依赖安装

环境准备这一步，我建议用conda而不是pip来管理Python环境，因为AI相关的包依赖关系比较复杂，conda在处理CUDA版本匹配上更省心。

# 创建环境 conda create -n ai-eng python=3.10 -y conda activate ai-eng # 安装PyTorch（根据你的CUDA版本选择） pip install torch==2.1.0 torchvision==0.16.0 --index-url https://download.pytorch.org/whl/cu118 # 安装服务相关依赖 pip install fastapi==0.104.0 uvicorn==0.24.0 redis==5.0.0 pydantic==2.4.0 pip install prometheus-client==0.18.0

这里有个细节：PyTorch的版本要和CUDA版本匹配。cu118表示CUDA 11.8，如果你的是CUDA 12.1，就要换成cu121。版本不匹配的话，运行时会报CUDA error: no kernel image is available。

安装完成后，用一段简单的代码验证环境：

import torch print(f"PyTorch version: {torch.__version__}") print(f"CUDA available: {torch.cuda.is_available()}") print(f"CUDA version: {torch.version.cuda}") print(f"GPU count: {torch.cuda.device_count()}") print(f"GPU name: {torch.cuda.get_device_name(0)}")

如果CUDA available是False，说明PyTorch没装对，需要重新检查版本。

4.2 配置文件设计

配置文件我选择用YAML而不是JSON，因为YAML支持注释，可读性更好。配置分三个文件：模型配置、服务配置、日志配置。

config/model.yaml：

models: text-classifier: path: "models/text_classifier.pt" max_batch_size: 32 max_seq_length: 256 warmup: true sentiment: path: "models/sentiment.pt" max_batch_size: 16 max_seq_length: 128 warmup: true device: "cuda" memory_fraction: 0.4

config/server.yaml：

server: host: "0.0.0.0" port: 8000 workers: 2 scheduler: max_wait_ms: 30 check_interval_ms: 5 queue_max_size: 1000 redis: host: "localhost" port: 6379 db: 0

config/logging.yaml：

version: 1 formatters: standard: format: "%(asctime)s [%(levelname)s] %(name)s: %(message)s" handlers: console: class: logging.StreamHandler formatter: standard level: INFO file: class: logging.handlers.RotatingFileHandler formatter: standard filename: "logs/app.log" maxBytes: 10485760 backupCount: 5 level: DEBUG root: level: DEBUG handlers: [console, file]

配置文件的好处是，改参数不用改代码，重启服务就行。而且不同环境（开发、测试、生产）可以用不同的配置文件，通过环境变量指定。

4.3 推理Worker的实现

推理Worker是整个系统的核心，它从调度层拉取任务，执行推理，返回结果。我用多进程的方式实现，每个Worker是一个独立的进程。

import torch import logging import time from multiprocessing import Process, Queue logger = logging.getLogger(__name__) class InferenceWorker(Process): def __init__(self, worker_id, model_manager, task_queue, result_queue): super().__init__() self.worker_id = worker_id self.model_manager = model_manager self.task_queue = task_queue self.result_queue = result_queue self.running = True def run(self): logger.info(f"Worker {self.worker_id} started") while self.running: try: task = self.task_queue.get(timeout=1.0) except Exception: continue if task is None: break task_id, model_name, inputs = task try: start = time.time() model = self.model_manager.load(model_name) with torch.no_grad(): # 预处理 batch = self._preprocess(inputs, model_name) # 推理 outputs = model(batch) # 后处理 results = self._postprocess(outputs) elapsed = (time.time() - start) * 1000 logger.debug(f"Worker {self.worker_id} task {task_id} done in {elapsed:.1f}ms") self.result_queue.put((task_id, results, None)) except Exception as e: logger.error(f"Worker {self.worker_id} task {task_id} failed: {e}") self.result_queue.put((task_id, None, str(e))) def _preprocess(self, inputs, model_name): # 根据模型类型做不同的预处理 # 这里以文本为例 return inputs def _postprocess(self, outputs): # 把模型输出转换成可序列化的格式 return outputs.tolist() def stop(self): self.running = False

这里有几个关键点：

超时控制。task_queue.get(timeout=1.0)里的超时是为了让Worker能定期检查self.running标志，否则进程会一直阻塞在get上，无法优雅退出。

异常隔离。每个任务都用try-except包起来，一个任务失败不会影响Worker继续处理下一个任务。错误信息通过result_queue返回给调度层。

结果序列化。模型输出通常是torch张量，不能直接跨进程传递，需要转成list或numpy数组。outputs.tolist()是最简单的方式，但如果输出很大，序列化开销会比较高。另一种方式是共享内存，但实现复杂度高，小规模场景下没必要。

4.4 调度层的实现

调度层负责接收请求、组批、分发给Worker、收集结果。我用Redis做任务队列，因为需要跨进程通信。

import redis import json import time import uuid import logging logger = logging.getLogger(__name__) class Scheduler: def __init__(self, redis_client, max_batch_size=32, max_wait_ms=30): self.redis = redis_client self.max_batch_size = max_batch_size self.max_wait = max_wait_ms / 1000.0 self.pending = {} # task_id -> (enqueue_time, request) def submit(self, model_name, inputs): task_id = str(uuid.uuid4()) self.pending[task_id] = (time.time(), { "model_name": model_name, "inputs": inputs }) return task_id def dispatch_loop(self, task_queue): while True: if not self.pending: time.sleep(0.005) continue now = time.time() batch = [] batch_ids = [] for task_id, (enqueue_time, request) in list(self.pending.items()): waited = now - enqueue_time if len(batch) >= self.max_batch_size: break if len(batch) == 0 or waited >= self.max_wait: batch.append(request) batch_ids.append(task_id) del self.pending[task_id] if batch: # 按模型分组 grouped = {} for task_id, req in zip(batch_ids, batch): model_name = req["model_name"] if model_name not in grouped: grouped[model_name] = [] grouped[model_name].append((task_id, req["inputs"])) for model_name, tasks in grouped.items(): task_queue.put((model_name, tasks)) time.sleep(0.005)

这个调度循环每5毫秒跑一次，检查待处理队列，决定是否组批。组批后按模型分组，因为不同模型的批处理不能混在一起。

注意：self.pending是内存中的字典，如果调度进程挂了，未处理的任务就丢了。生产环境需要考虑持久化，比如把pending也写到Redis里。我这里为了简化没有做，你可以根据实际需求决定。

4.5 监控与健康检查

监控是生产环境的眼睛。没有监控，出了问题你都不知道从哪里查起。我用Prometheus的Python客户端来采集指标。

from prometheus_client import Counter, Histogram, Gauge, generate_latest from fastapi import Response # 定义指标 REQUEST_COUNT = Counter( "inference_requests_total", "Total inference requests", ["model_name", "status"] ) REQUEST_LATENCY = Histogram( "inference_latency_seconds", "Inference latency", ["model_name"], buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0] ) QUEUE_SIZE = Gauge( "inference_queue_size", "Current queue size" ) GPU_MEMORY = Gauge( "gpu_memory_used_mb", "GPU memory used in MB" ) @app.get("/metrics") async def metrics(): # 更新GPU显存指标 if torch.cuda.is_available(): GPU_MEMORY.set(torch.cuda.memory_allocated() / 1024**2) return Response(generate_latest(), media_type="text/plain")

健康检查接口也很重要，用于负载均衡器判断服务是否可用：

@app.get("/health") async def health(): checks = { "redis": check_redis(), "model": check_model_loaded(), "gpu": check_gpu_available() } if all(checks.values()): return {"status": "healthy", "checks": checks} else: return JSONResponse( status_code=503, content={"status": "unhealthy", "checks": checks} )

健康检查不能只返回一个"ok"，要检查所有关键依赖。我见过服务本身活着但Redis挂了，结果所有请求都超时的情况。如果健康检查能发现Redis不可用，负载均衡器就会把流量切走，避免用户请求堆积。

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

5.1 服务启动就OOM怎么办

这是最常见的问题。服务一启动就报CUDA out of memory，通常有几个原因：

模型太大，单卡放不下。解决办法是用torch.cuda.set_per_process_memory_fraction()限制每个进程的显存，或者用模型并行把模型拆到多张卡上。如果是推理场景，还可以考虑量化，把FP32转成FP16或INT8，显存占用能降到1/2到1/4。

多进程同时加载模型。如果你开了4个Worker，每个都加载一份模型，显存占用就是4倍。解决办法是减少Worker数量，或者用共享内存的方式让多个进程共享同一份模型权重。PyTorch的torch.multiprocessing支持通过share_memory()共享张量，但实现起来比较麻烦。

显存碎片。前面提过，设置PYTORCH_CUDA_ALLOC_CONF=max_split_size_mb:128可以缓解。

排查步骤我一般是这样：

1. 先用nvidia-smi看GPU上还有什么进程在占显存，确认没有残留进程。
2. 用torch.cuda.memory_summary()打印详细的显存分配情况。
3. 逐步减少batch_size和Worker数量，找到能跑起来的最小配置。
4. 如果还是不行，考虑换更小的模型或者用量化版本。

5.2 延迟忽高忽低怎么排查

延迟不稳定比延迟高更让人头疼，因为很难复现。我遇到过几种典型情况：

批处理等待导致的延迟波动。低峰期请求少，每个请求都要等满max_wait时间才被处理，延迟自然高。解决办法是动态调整max_wait，低峰期调小，高峰期调大。或者干脆在低峰期跳过批处理，直接单条推理。

GC导致的停顿。Python的垃圾回收在回收大量对象时会暂停整个进程。如果推理过程中创建了大量临时对象，GC触发时延迟就会飙升。解决办法是尽量复用对象，避免在热路径上创建新对象。另外可以调整GC阈值：

import gc gc.set_threshold(10000, 50, 50)

显存交换。如果显存不够，PyTorch会把部分数据换到内存里，需要时再换回来，这个过程非常慢。表现是延迟突然从几十毫秒跳到几秒。解决办法就是确保显存充足，不要超配。

下游依赖慢。如果推理服务依赖了其他服务（比如特征存储、模型仓库），下游慢会导致整体延迟高。解决办法是加超时和降级，下游超时就返回默认结果，不要一直等。

排查延迟问题，我的经验是先看监控，再看日志，最后上profiler。监控能告诉你延迟是什么时候开始高的、影响范围有多大；日志能告诉你具体是哪个环节慢；profiler能告诉你代码里哪一行最耗时。

5.3 常见问题速查表

5.4 几个救命的排查工具

py-spy：不侵入进程的情况下查看Python堆栈，服务卡死时特别有用。

pip install py-spy py-spy dump --pid <PID>

nvidia-smi：查看GPU使用情况，配合watch -n 1 nvidia-smi可以实时监控。

torch.cuda.memory_summary()：打印详细的显存分配情况，包括每个张量的大小和位置。

Redis CLI：查看队列长度和积压情况。

redis-cli LLEN task_queue redis-cli INFO memory

这些工具我基本每次排查问题都会用到，建议提前装好。

6. 一些踩坑后的经验之谈

做AI工程这两年，最大的体会是：模型只是冰山一角，水面下的工程体系才是决定成败的关键。我见过太多团队花大力气调模型，把准确率从90%提到92%，结果线上服务因为一个显存泄漏三天两头挂，用户体验反而更差。

另一个体会是不要过度设计。我一开始想搞一套完美的微服务架构，每个模块独立部署，结果运维复杂度爆炸，光排查一个请求经过的链路就要半天。后来回归到单体多进程的架构，简单直接，反而更稳定。架构的复杂度应该和团队规模、业务量匹配，小团队小业务，单体就够了。

还有一点是监控要先行。不要等出了问题才加监控，而是在服务上线前就把关键指标采集好。我现在的习惯是，任何一个新服务上线，第一件事就是确认监控面板能看到请求量、延迟、错误率、资源占用这四个核心指标。没有监控的服务，就像闭着眼睛开车。

最后说一个具体的技巧：给每个请求打上trace_id。从接入层生成一个唯一ID，透传到调度层、推理层，所有日志都带上这个ID。这样排查问题时，你可以用trace_id把所有相关日志串起来，快速定位是哪个环节出了问题。这个习惯帮我省了无数排查时间。

提示：trace_id的生成可以用uuid.uuid4().hex[:16]，足够短且碰撞概率极低。日志格式里加上%(trace_id)s，通过contextvars传递。

这套体系我目前跑了大概半年，日均处理几十万次推理请求，P99延迟稳定在200毫秒以内，可用性99.9%以上。当然还有很多可以优化的地方，比如引入动态批处理、支持模型热更新、做多机部署等，但那是下一步的事了。先把基础打牢，比什么都重要。