1. 从零手搓AI工程：为什么我不建议你直接调包

很多人第一次接触AI工程，脑子里想的都是“找个开源模型，pip install一下，跑通demo就完事”。我刚开始也是这么想的，直到我在实际项目里被现实反复摩擦——模型推理慢得像蜗牛、显存动不动就爆、部署到生产环境后延迟高得离谱。这时候你才会意识到，调包只能让你跑起来，但从零理解AI工程才能让你跑得稳、跑得快、跑得省。

“ai-engineering-from-scratch”这个方向，核心不是让你重新发明Transformer，而是让你亲手把AI系统的每一层都摸一遍：数据怎么流进来、模型怎么加载、推理怎么调度、显存怎么管理、服务怎么暴露、性能怎么压榨。这些东西，文档里不会写，教程里只会给你一个“hello world”，但生产环境里每一个都是要命的坑。

这篇文章适合谁？如果你已经会写Python，用过PyTorch或TensorFlow，但一遇到“模型部署”“推理优化”“显存爆炸”就头大，那这篇就是给你写的。我会从最底层的张量操作开始，一步步搭出一个能跑、能扛、能扩展的AI推理服务，中间所有踩过的坑、绕过的弯、总结出来的经验，全部摊开讲。不堆砌术语，不复制官方文档，只讲一个从业者真正会用到的东西。

2. 先搞清楚AI工程到底在工程什么

2.1 训练和推理是两套完全不同的逻辑

很多人把AI工程等同于“训练模型”，这是个巨大的误解。训练是离线的一次性投入，推理是在线的持续输出。训练的时候你可以等，可以跑一晚上，可以容忍失败重来；推理的时候用户等不了，延迟超过200毫秒就开始骂娘，显存超了直接服务崩溃。

从工程角度看，训练关注的是吞吐量和收敛性，推理关注的是延迟和并发能力。这两个目标经常是矛盾的。比如训练时你会用大batch size来提升GPU利用率，但推理时batch size太大会导致单个请求的延迟飙升。所以AI工程的第一课，就是把训练思维和推理思维彻底分开。

我在实际项目里见过太多团队，拿训练脚本改吧改吧就上线推理服务，结果QPS一上来就雪崩。原因很简单：训练脚本里没有请求队列、没有动态batching、没有显存池化，每个请求都重新加载一遍模型，不崩才怪。

2.2 从零搭建的四个核心模块

一个完整的AI推理系统，拆开来看就是四块：

• 模型加载与初始化：怎么把权重从磁盘搬到显存，怎么处理多卡、多模型共存，怎么做到热更新不中断服务。
• 请求调度与批处理：怎么把零散的请求攒成batch，怎么在延迟和吞吐之间找平衡，怎么处理超时和优先级。
• 显存管理与优化：怎么复用显存块，怎么避免碎片化，怎么在有限显存下跑更大的模型。
• 服务暴露与监控：怎么用HTTP/gRPC暴露接口，怎么做健康检查、指标采集、日志追踪。

这四块每一块都有坑，而且坑坑相扣。比如显存管理没做好，动态batching就会因为OOM频繁失败；请求调度没设计好，监控指标就会失真，你根本不知道瓶颈在哪。

2.3 为什么“从零”比“调包”更值得投入

调包最大的问题是黑盒。你不知道里面发生了什么，出了问题只能靠猜。而从零搭建的过程，逼着你去理解每一个环节的输入输出、资源消耗、失败模式。这种理解一旦建立，你再去看任何推理框架的文档，都能一眼看出它的设计取舍和适用边界。

举个例子：vLLM的PagedAttention为什么能大幅提升吞吐？如果你自己实现过简单的KV Cache管理，就会立刻明白——它本质上是用操作系统的虚拟内存分页思想来解决显存碎片问题。没有这个背景，你只能死记硬背“vLLM快”，但不知道它为什么快、什么时候不快。

3. 模型加载：从磁盘到显存的第一公里

3.1 权重文件的格式选择与加载策略

模型权重常见的格式有PyTorch的.pt/.pth、SafeTensors、GGUF、ONNX等。从零搭建的话，我建议先用SafeTensors，原因有三：加载速度快、内存映射友好、安全性好（不会执行任意代码）。

加载策略上，最朴素的做法是torch.load()然后model.load_state_dict()。但这里有个隐藏的坑：默认加载是在CPU上完成的，然后再搬到GPU。对于大模型，这意味着你需要两倍的内存——一份在CPU，一份在GPU。正确的做法是边加载边搬运：

import torch from safetensors.torch import load_file def load_model_weights(model, weight_path, device): state_dict = load_file(weight_path, device=device) model.load_state_dict(state_dict) model.to(device) return model

load_file的device参数会直接把权重加载到目标设备，省掉中间拷贝。实测下来，7B参数的模型用这种方式加载，峰值内存能降低40%左右。

3.2 多卡加载与模型并行

单卡装不下的时候，就要考虑模型并行。最简单的做法是按层切分：前几层放GPU0，后几层放GPU1。但这样会带来一个问题——跨卡通信成为瓶颈。每经过一次切分点，就要做一次all-reduce或者点对点传输。

我的经验是：能流水线并行就不要张量并行。流水线并行按层切，通信量小，实现简单；张量并行要把矩阵运算拆开，通信频繁，对带宽要求极高。除非你的模型大到单层都放不下，否则优先选流水线并行。

具体实现时，可以用torch.distributed的pipeline接口，也可以手动控制每层的device。手动控制更灵活，但容易出错。我一般会写一个LayerDispatcher类，根据每层的参数量和当前显存占用动态分配设备。

3.3 热更新：不重启服务换模型

生产环境不可能每次换模型都重启服务。热更新的核心思路是双缓冲：新模型在后台加载，加载完成后原子性地切换推理入口的指针。

class ModelManager: def __init__(self): self.current_model = None self.lock = threading.Lock() def hot_swap(self, new_model): with self.lock: old_model = self.current_model self.current_model = new_model # 延迟释放旧模型，等待正在处理的请求完成 threading.Timer(30.0, lambda: del old_model).start()

这里的关键是延迟释放。如果立刻删掉旧模型，正在处理的请求会直接崩溃。给一个30秒的缓冲期，让存量请求跑完，再回收显存。这个时间要根据你的P99延迟来定，一般设成最大延迟的两倍比较安全。

4. 请求调度：让每一块GPU都吃饱

4.1 动态批处理的实现逻辑

动态批处理的核心思想是：不等所有请求到齐，而是攒够一批就走。具体来说，维护一个请求队列，当队列长度达到阈值或者等待时间超过上限时，就把当前队列里的请求打包成一个batch送进模型。

class DynamicBatcher: def __init__(self, max_batch_size=8, max_wait_ms=50): self.queue = [] self.max_batch_size = max_batch_size self.max_wait_ms = max_wait_ms def add_request(self, request): self.queue.append(request) if len(self.queue) >= self.max_batch_size: return self._flush() return None def _flush(self): batch = self.queue[:self.max_batch_size] self.queue = self.queue[self.max_batch_size:] return batch

max_batch_size和max_wait_ms是两个需要调优的参数。batch size越大，GPU利用率越高，但单个请求的延迟也越大。我的经验值是：在线服务max_batch_size设8到16，max_wait_ms设20到50毫秒。离线批处理可以设得更大，但在线服务必须优先保证延迟。

4.2 优先级队列与超时处理

不是所有请求都同等重要。付费用户的请求应该优先处理，超时请求应该直接丢弃而不是继续排队。实现上可以用heapq维护一个优先级队列，优先级由用户等级和等待时间共同决定。

import heapq import time class PriorityQueue: def __init__(self): self.heap = [] self.counter = 0 def push(self, request, priority): # 优先级越高，值越小 heapq.heappush(self.heap, (priority, self.counter, request)) self.counter += 1 def pop(self): if not self.heap: return None priority, _, request = heapq.heappop(self.heap) if time.time() - request.arrival_time > request.timeout: return None # 超时丢弃 return request

超时处理有个容易忽略的点：超时时间应该从请求到达开始算，而不是从进入队列开始算。很多实现搞错了这个，导致请求在队列里等了很久，刚进模型就超时了，白白浪费计算资源。

4.3 背压机制：防止服务被压垮

当请求量超过服务处理能力时，必须有背压机制。最简单的做法是限制队列长度，队列满了直接返回503。更优雅的做法是自适应限流：根据当前的平均延迟动态调整接受速率。

class AdaptiveLimiter: def __init__(self, target_latency_ms=100): self.target_latency = target_latency_ms self.current_rate = 100 # 初始QPS def should_accept(self, current_latency_ms): if current_latency_ms > self.target_latency * 1.5: self.current_rate *= 0.9 # 降低速率 elif current_latency_ms < self.target_latency * 0.8: self.current_rate *= 1.1 # 提高速率 return random.random() < self.current_rate / 1000

这个算法的核心是让延迟稳定在目标值附近。延迟高了就降速，延迟低了就提速。实测下来，比固定QPS限流稳定得多，尤其是在请求模式波动大的场景下。

5. 显存管理：AI工程最硬的骨头

5.1 显存碎片化是怎么产生的

显存碎片化是推理服务最大的隐形杀手。每次分配和释放显存块，都会在显存池里留下空洞。时间一长，明明总空闲显存够用，但就是找不到一块连续的空间来放新模型或新batch。

碎片化的根源在于变长分配。不同请求的输入长度不同，需要的KV Cache大小也不同。如果每次都按需分配，碎片化几乎不可避免。解决办法是预分配+池化：提前申请一大块显存，自己管理分配和回收。

class MemoryPool: def __init__(self, total_size, block_size=1024): self.total_size = total_size self.block_size = block_size self.free_blocks = list(range(total_size // block_size)) self.used_blocks = {} def allocate(self, num_blocks): if len(self.free_blocks) < num_blocks: raise OutOfMemoryError() blocks = self.free_blocks[:num_blocks] self.free_blocks = self.free_blocks[num_blocks:] return blocks def free(self, blocks): self.free_blocks.extend(blocks) self.free_blocks.sort()

这个简化版的池化分配器，核心思想就是固定块大小、按块分配。虽然会有内部碎片（最后一个块用不满），但外部碎片被彻底消除了。实际生产中，块大小一般设成2MB到16MB，根据模型大小和请求长度分布来定。

5.2 KV Cache的按需分配与回收

KV Cache是显存消耗的大头。对于自回归生成模型，每个请求的KV Cache大小是2 * num_layers * num_heads * head_dim * seq_len * dtype_size。以7B模型为例，FP16精度下，每个token的KV Cache大约是1MB。生成1000个token，就是1GB。

按需分配的策略是：请求开始时分配初始块，生成过程中按需扩展。扩展时如果当前块用完了，就再申请一块，用链表把块串起来。这就是PagedAttention的核心思想。

class KVCacheManager: def __init__(self, pool, block_size=16): self.pool = pool self.block_size = block_size self.sequences = {} def append_token(self, seq_id, kv_pair): if seq_id not in self.sequences: self.sequences[seq_id] = { 'blocks': [self.pool.allocate(1)], 'length': 0 } seq = self.sequences[seq_id] if seq['length'] % self.block_size == 0: seq['blocks'].append(self.pool.allocate(1)) seq['length'] += 1 # 将kv_pair写入对应块

回收时，把序列的所有块归还给池子。这里有个优化点：块可以复用。如果两个请求的前缀相同（比如相同的system prompt），它们的KV Cache可以共享。这就是Prefix Caching，能大幅降低显存占用。

5.3 量化：用精度换显存的取舍

量化是显存优化的终极手段。FP16转INT8，显存直接减半；转INT4，再减半。但量化会带来精度损失，需要仔细评估。

我的经验是：权重可以量化到INT4，激活值最好保持FP16。权重对精度不敏感，INT4量化后困惑度上升通常不到5%；但激活值对精度敏感，量化后容易导致输出乱码。如果非要量化激活值，至少保持INT8，并且要做校准。

def quantize_weights(model, bits=4): for name, param in model.named_parameters(): if 'weight' in name and param.dim() > 1: scale = param.abs().max() / (2 ** (bits - 1) - 1) quantized = torch.round(param / scale).clamp(-2**(bits-1), 2**(bits-1)-1) param.data = quantized * scale return model

这个简单的对称量化实现，实际用的时候还要考虑per-channel和per-group的粒度。粒度越细，精度损失越小，但计算开销越大。一般权重用per-channel，激活值用per-tensor就够了。

6. 服务暴露：从本地脚本到生产接口

6.1 HTTP vs gRPC：怎么选

HTTP简单通用，gRPC高效紧凑。对于AI推理服务，我的建议是：对外用HTTP，对内用gRPC。对外HTTP方便调试和集成，对内gRPC减少序列化开销和网络延迟。

如果用HTTP，推荐FastAPI，异步支持好，自动生成文档。如果用gRPC，推荐官方grpcio，配合protobuf定义接口。不管用哪个，都要注意请求体大小限制和超时设置。AI请求的输入可能很大（比如长文本），默认的1MB限制很容易超。

from fastapi import FastAPI, Request import uvicorn app = FastAPI() @app.post("/generate") async def generate(request: Request): data = await request.json() prompt = data.get("prompt", "") max_tokens = data.get("max_tokens", 128) result = await inference_engine.generate(prompt, max_tokens) return {"result": result} if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000, limit_concurrency=100)

limit_concurrency是关键参数，控制同时处理的请求数。设得太高，显存扛不住；设得太低，GPU利用率上不去。一般设成max_batch_size * 2比较合适。

6.2 健康检查与优雅关闭

健康检查不能只返回200 OK，要真正检查模型是否可用。最简单的做法是跑一次微型推理，比如生成一个token，看是否成功。

@app.get("/health") async def health(): try: result = await inference_engine.generate("test", max_tokens=1) return {"status": "healthy", "model": "loaded"} except Exception as e: return {"status": "unhealthy", "error": str(e)}, 503

优雅关闭也很重要。收到SIGTERM信号后，停止接受新请求，等待存量请求完成，再释放显存退出。这个过程一般给30秒到60秒。

6.3 监控指标：延迟、吞吐、显存

没有监控的AI服务就是盲人摸象。必须采集的指标包括：

• 请求延迟：P50、P95、P99，按模型和接口分组。
• 吞吐量：QPS、tokens/s，按batch size分组。
• 显存使用：已分配、已保留、峰值，按设备分组。
• 队列长度：等待处理的请求数，按优先级分组。

这些指标用Prometheus采集，Grafana展示。关键是设置合理的告警阈值：P99延迟超过500毫秒告警，显存使用超过90%告警，队列长度持续增长告警。

7. 性能调优：从能跑到跑得快的最后一公里

7.1 算子融合与图优化

PyTorch的eager模式方便调试，但性能差。生产环境应该用torch.compile或者导出ONNX再优化。torch.compile能把多个小算子融合成一个大算子，减少kernel launch开销。

@torch.compile(mode="max-autotune") def forward(self, input_ids): return self.model(input_ids)

max-autotune模式会自动搜索最优的算子实现，但编译时间较长。第一次编译可能要几分钟，之后就走缓存了。如果模型结构固定，这个投入是值得的，实测能提升20%到40%的吞吐。

7.2 连续批处理与迭代级调度

传统的静态批处理要等整个batch跑完才能处理下一批。连续批处理（Continuous Batching）则是每生成一个token就检查一次，有请求完成就立刻移出，有新请求就立刻加入。这样GPU永远不会空闲。

实现上，需要把推理循环拆成迭代级：

def continuous_batching_loop(engine, max_iterations=1000): active_sequences = [] for _ in range(max_iterations): # 移除已完成的序列 active_sequences = [s for s in active_sequences if not s.done] # 加入新请求 while len(active_sequences) < engine.max_batch_size: new_req = engine.get_next_request() if new_req is None: break active_sequences.append(new_req) if not active_sequences: break # 执行一步推理 engine.step(active_sequences)

这个循环的核心是每一步都重新组batch。相比静态批处理，GPU利用率能从60%提升到90%以上。

7.3 实测数据与调优记录

我在一台单卡A100上做过对比测试，模型是Llama-2-7B，输入长度128，输出长度128：

从数据可以看出，连续批处理是吞吐提升的最大功臣，量化则是显存优化的关键。两者结合，能在单卡上把7B模型的QPS从3提升到19，提升超过5倍。

调优过程中最大的坑是显存碎片。连续批处理频繁分配释放KV Cache，如果不做池化，跑几个小时就会OOM。上了内存池之后，连续跑24小时显存占用依然稳定。

8. 一些踩坑之后的真心话

从零搭建AI推理服务这件事，我前后折腾了大半年，踩过的坑比写过的代码还多。最大的体会是：不要追求一步到位。先跑通最简单的版本，然后一个模块一个模块地优化。每次只改一个变量，做好基准测试，确认有效再继续。

另一个体会是：监控比优化更重要。没有监控，你根本不知道瓶颈在哪，优化就是瞎猜。我见过太多团队花大力气优化模型推理，结果发现瓶颈在网络传输或者请求序列化上。

最后说一个容易被忽略的点：测试数据要真实。用随机生成的输入做压测，和用真实用户输入做压测，结果可能差好几倍。真实输入的分布往往更不均匀，长尾请求对系统稳定性的考验更大。所以压测一定要用生产环境的采样数据，或者至少模拟出相似的分布。

这个方向还有很多可以深挖的东西，比如多模型共存时的显存调度、异构硬件下的负载均衡、边缘设备上的轻量化部署。每一个都是独立的课题，每一个都有足够的坑等着你去踩。但只要你把上面这些基础打牢了，后面遇到什么问题都不会慌，因为你知道系统里每一块是怎么运转的。