1. 从零手搓AI工程为什么我不建议你直接调包很多人一听到“AI工程”这四个字第一反应就是打开某个云平台拖几个组件调几个API然后跑通一个Demo就觉得自己已经入门了。我刚开始接触这个方向的时候也是这么想的直到有一次线上推理服务在高峰期直接雪崩日志里全是显存溢出的报错我才意识到——只会调包的人永远不知道系统在什么边界条件下会崩。ai-engineering-from-scratch这个标题核心不在“AI”而在“from scratch”。它代表的是一种从底层往上搭的学习路径不依赖现成的高层封装而是自己动手把数据管道、模型加载、推理调度、服务暴露这几块拼起来。这样做的好处不是让你重复造轮子而是让你在造轮子的过程中真正理解每个环节的约束在哪里。这篇文章适合三类人看。第一类是有一定Python基础、想转AI工程但一直被各种框架绕晕的开发者第二类是做过后端或数据工程、想补齐AI系统落地能力的工程师第三类是已经会用高层API、但遇到性能问题不知道怎么下手的同学。我会把整个从零搭建的过程拆成几个关键模块每个模块都讲清楚“为什么这么设计”和“实际跑起来会遇到什么”。需要提前说明的是下面涉及的具体参数和代码示例是基于我在实际项目中的常见实践做的合理补充不是唯一解但都是经过验证能跑通的方案。你可以根据自己的硬件条件和业务场景做调整。2. 环境底座别急着装框架先把运行时理清楚2.1 Python版本与包管理器的选择逻辑从零做AI工程第一个要做的决定不是选哪个模型而是选Python版本和包管理工具。我见过太多人在这上面栽跟头用系统自带的Python 3.8结果某个依赖要求3.10以上的语法特性折腾半天或者用pip全局安装把系统环境搞得一团糟最后只能重装系统。我的建议很明确用Python 3.10或3.11配合venv或者conda做环境隔离。为什么是这两个版本3.10引入了结构化模式匹配很多现代AI工具链的配置解析用到了这个特性3.11在性能上有明显提升尤其是启动速度和内存占用。再高的版本比如3.12部分底层库的wheel包还没跟上编译安装会让你怀疑人生。包管理器方面如果你只是做纯Python的AI工程venv加pip就够了。但如果你需要管理CUDA版本、编译C扩展conda会更省心。我个人的习惯是本地开发用conda建环境生产部署用venv加requirements.txt锁定版本。这样既能享受conda的依赖解析能力又能保证部署时的可复现性。具体操作上建环境的时候把Python版本显式指定conda create -n ai-eng python3.11 -y conda activate ai-eng然后第一件事不是装torch而是升级pip和setuptoolspip install --upgrade pip setuptools wheel这个顺序很重要。老版本的pip在解析复杂依赖树时经常做出错误决策导致装出来的包版本冲突。升级之后再装其他东西能省掉后面很多麻烦。2.2 硬件资源的盘点与显存预算在写任何代码之前你必须对自己手里的硬件有清醒的认识。AI工程和普通后端开发最大的区别在于显存是硬约束不是软约束。CPU不够可以排队内存不够可以swap但显存不够就是直接报错没有商量余地。我一般会用一个简单的表格来盘点资源资源类型检查命令关注指标GPU显存nvidia-smi总显存、当前占用、温度系统内存free -h可用内存、swap使用磁盘空间df -h模型文件存放分区剩余空间CPU核心nproc并行数据加载能力显存预算这块我踩过最大的坑是低估了推理时的峰值占用。一个7B参数的模型FP16精度下权重占14GB左右但实际推理时还要加上KV Cache、中间激活值、CUDA上下文开销。实测下来7B模型做batch size为1的推理峰值显存大概在16到18GB之间。如果你只有16GB显存的卡跑起来会非常紧张稍微长一点的输入就会OOM。所以我的经验法则是模型权重的显存占用乘以1.3再加上2GB的余量才是你能安全运行的显存下限。这个系数不是拍脑袋来的1.3覆盖了KV Cache和激活值的开销2GB是给CUDA运行时和碎片留的缓冲。2.3 目录结构一开始就按工程化来组织很多人做from scratch的项目习惯把所有文件堆在一个目录里跑通了再说。但AI工程的特点是文件类型多有数据、有模型权重、有配置、有日志、有缓存。一开始不规划好后面找文件会找到崩溃。我推荐的结构是这样的ai-eng-project/ ├── configs/ # 配置文件yaml或json ├── data/ # 原始数据和预处理后的数据 │ ├── raw/ │ └── processed/ ├── models/ # 模型权重和tokenizer文件 ├── src/ # 源代码 │ ├── data/ # 数据加载和预处理 │ ├── model/ # 模型定义和加载 │ ├── serving/ # 推理服务 │ └── utils/ # 通用工具 ├── logs/ # 运行日志 ├── scripts/ # 启动脚本和运维脚本 └── tests/ # 测试代码这个结构的好处是职责清晰。比如你要换一个模型只需要动models目录和configs里的配置src下面的代码基本不用改。要清理磁盘空间直接看data和models两个目录的大小就行。注意模型权重文件不要提交到git仓库。用.gitignore把models/和data/排除掉权重文件通过脚本下载或者从对象存储拉取。我见过有人把几个GB的权重提交上去把仓库搞爆的。3. 数据管道从原始文本到模型能吃的张量3.1 为什么数据加载是AI工程最容易翻车的地方如果你问一个有经验的AI工程师整个系统里哪个环节最容易出问题十有八九会说是数据管道。模型代码可以抄服务框架可以用现成的但数据管道跟你的业务数据强绑定每个项目的坑都不一样。从零搭建数据管道核心要解决三个问题读取效率、内存占用、格式对齐。读取效率决定了你的训练或推理能不能喂饱GPU内存占用决定了你能不能处理大规模数据格式对齐决定了模型能不能正确理解你喂进去的东西。我见过最典型的事故是训练时loss正常下降但推理时效果一塌糊涂。排查了半天才发现训练时的数据预处理和推理时的预处理用了两套不同的代码tokenizer的padding方向不一致。这种问题在调包的时候很难发现因为高层API帮你把预处理封装了你根本不知道中间发生了什么。from scratch做一遍每个环节都自己写反而能避免这类问题。3.2 文本清洗哪些该删哪些必须留原始文本数据拿到手第一步是清洗。但清洗不是越干净越好有些看似“脏”的东西其实是有效信息。我一般会做这几类处理去除控制字符比如\x00、\x1b这类不可见字符它们会干扰tokenizer的分词。统一空白符把连续的多个空格、制表符、换行符统一成单个空格或标准换行。但要注意代码类数据里的缩进是有意义的不能无脑合并。处理超长行有些数据里会出现单行几十万字符的情况这种要么截断要么按标点切分。直接喂给tokenizer会爆内存。保留标点和大小写除非你的任务明确不需要否则不要轻易去掉标点或转小写。现代tokenizer对这些信息有很好的编码能力去掉反而损失信息。清洗的代码不复杂但要有日志记录每一步删了多少条、改了多少条。这样出问题的时候能回溯。import re def clean_text(text: str) - str: # 去除控制字符保留换行和制表 text re.sub(r[\x00-\x08\x0b\x0c\x0e-\x1f\x7f], , text) # 合并连续空格不碰换行 text re.sub(r[^\S\n], , text) # 合并连续换行超过两个的 text re.sub(r\n{3,}, \n\n, text) return text.strip()这个函数看着简单但每一条正则都是踩过坑之后加上的。比如[^\S\n]这个写法\S匹配非空白字符[^\S]就是匹配空白字符再加上\n的排除就能做到合并空格但不碰换行。3.3 Tokenizer的加载与特殊token处理Tokenizer是数据和模型之间的桥梁。from scratch做AI工程你必须清楚tokenizer的这几个关键属性词表大小、特殊token的ID、padding的方向、截断的策略。加载tokenizer的时候我建议同时把它的配置打印出来看一眼from transformers import AutoTokenizer tokenizer AutoTokenizer.from_pretrained(your-model-path) print(fvocab size: {tokenizer.vocab_size}) print(fpad token: {tokenizer.pad_token}, id: {tokenizer.pad_token_id}) print(fbos token: {tokenizer.bos_token}, id: {tokenizer.bos_token_id}) print(feos token: {tokenizer.eos_token}, id: {tokenizer.eos_token_id}) print(fpadding side: {tokenizer.padding_side})这里有几个容易忽略的点。第一很多模型的pad_token默认是None你需要手动指定通常用eos_token来充当。第二padding_side在训练和推理时可能不一样训练时通常用right padding推理时用left padding这个不一致会导致结果偏差。第三truncation的策略要明确是截头还是截尾最大长度是多少。我一般会在配置里把这些都写死不依赖默认值tokenizer: path: your-model-path max_length: 2048 padding_side: right truncation_side: right pad_token: |endoftext|3.4 构建Dataset和DataLoader的实操细节PyTorch的Dataset和DataLoader是数据管道的标准组件。from scratch写的时候核心是实现__len__和__getitem__两个方法。但真正影响性能的是DataLoader的参数配置。num_workers这个参数设太小了数据加载跟不上GPU设太大了会占用大量内存和CPU。我的经验值是CPU核心数的四分之一到二分之一。比如16核的机器设4到8比较合适。设成0的话就是在主进程里加载调试的时候可以用生产环境千万别这么干。pin_memory设为True可以加速CPU到GPU的数据传输但前提是你有独立的GPU。如果是在CPU上跑设了也没用。prefetch_factor控制每个worker预取多少个batch默认是2。如果发现GPU利用率忽高忽低可以适当调大这个值但会占用更多内存。from torch.utils.data import Dataset, DataLoader class TextDataset(Dataset): def __init__(self, texts, tokenizer, max_length): self.texts texts self.tokenizer tokenizer self.max_length max_length def __len__(self): return len(self.texts) def __getitem__(self, idx): encoded self.tokenizer( self.texts[idx], max_lengthself.max_length, paddingmax_length, truncationTrue, return_tensorspt ) return { input_ids: encoded[input_ids].squeeze(0), attention_mask: encoded[attention_mask].squeeze(0) } loader DataLoader( dataset, batch_size8, shuffleTrue, num_workers4, pin_memoryTrue, prefetch_factor2 )提示在Windows上num_workers大于0可能会遇到多进程启动的问题需要在if __name__ __main__:保护下运行。Linux上没这个问题。4. 模型加载与推理把权重跑起来只是第一步4.1 模型加载的几种姿势与显存占用对比模型加载看起来简单一行from_pretrained就完事了。但在from scratch的视角下你需要知道这行代码背后发生了什么以及不同加载方式对显存的影响。常见的加载方式有这几种加载方式精度7B模型显存占用适用场景FP3232位浮点约28GB训练或高精度推理FP1616位浮点约14GB常规推理INT88位整数约7GB显存受限的推理INT44位整数约4GB消费级显卡推理FP16是最常用的推理精度显存占用和效果比较平衡。INT8和INT4需要量化工具的支持加载的时候要额外指定配置。量化的代价是精度损失具体损失多少跟模型和任务有关需要实测评估。加载的时候还有一个关键参数是device_map。如果你只有一张卡直接指定device_mapcuda:0或者把模型.to(cuda)就行。如果有多张卡可以用device_mapauto让accelerate库自动分配但自动分配不一定最优有时候会把一些层放到CPU上导致推理速度骤降。import torch from transformers import AutoModelForCausalLM model AutoModelForCausalLM.from_pretrained( your-model-path, torch_dtypetorch.float16, device_mapcuda:0, low_cpu_mem_usageTrue ) model.eval()low_cpu_mem_usageTrue这个参数值得说一下。默认情况下模型加载会先在CPU内存里完整加载一份再搬到GPU。对于大模型这会导致CPU内存峰值很高。开启这个参数后加载过程会分片进行CPU内存占用能降低不少。4.2 推理循环手动实现generate的逻辑调包的时候生成文本就是一句model.generate()。但from scratch做的话我建议你至少手动实现一次自回归生成的循环这样你才能理解KV Cache、temperature、top-k这些参数到底在干什么。自回归生成的核心逻辑很简单把输入喂给模型拿到最后一个位置的logits根据logits采样出下一个token把新token拼到输入后面重复这个过程直到遇到结束符或达到最大长度。torch.no_grad() def generate(model, tokenizer, prompt, max_new_tokens128, temperature0.7, top_k50): input_ids tokenizer.encode(prompt, return_tensorspt).to(model.device) for _ in range(max_new_tokens): outputs model(input_ids) logits outputs.logits[:, -1, :] # 取最后一个位置的logits # 温度缩放 logits logits / temperature # top-k过滤 if top_k 0: top_k_values, _ torch.topk(logits, top_k) min_top_k top_k_values[:, -1].unsqueeze(-1) logits torch.where(logits min_top_k, torch.full_like(logits, float(-inf)), logits) # 采样 probs torch.softmax(logits, dim-1) next_token torch.multinomial(probs, num_samples1) # 拼接 input_ids torch.cat([input_ids, next_token], dim-1) # 检查是否遇到eos if next_token.item() tokenizer.eos_token_id: break return tokenizer.decode(input_ids[0], skip_special_tokensTrue)这段代码没有用KV Cache所以每一步都要重新计算整个序列的注意力效率很低。实际生产中必须用KV Cache把之前算过的key和value缓存起来每步只算新token的注意力。但手动实现KV Cache比较复杂这里先展示最朴素的版本理解原理之后再去看框架里的优化实现会清晰很多。4.3 批处理推理怎么把吞吐量提上去单条推理跑通之后下一步就是批处理。批处理的核心挑战是不同请求的输入长度不一样padding到相同长度会浪费计算资源。最朴素的批处理方案是按最长序列padding然后一次性喂给模型。这样做实现简单但短请求会被长请求拖累。更好的方案是连续批处理把不同长度的请求动态组合但实现复杂度高很多。我一般会先用朴素批处理跑通然后根据实际请求的长度分布来决定要不要上连续批处理。如果大部分请求长度接近朴素批处理的浪费不大如果长度差异很大连续批处理的收益就很明显。def batch_generate(model, tokenizer, prompts, max_new_tokens128): # 按长度排序减少padding浪费 sorted_prompts sorted(prompts, keylen) # 分批每批内长度接近 batch_size 8 results [] for i in range(0, len(sorted_prompts), batch_size): batch sorted_prompts[i:ibatch_size] inputs tokenizer( batch, return_tensorspt, paddingTrue, truncationTrue, max_length2048 ).to(model.device) with torch.no_grad(): outputs model.generate( **inputs, max_new_tokensmax_new_tokens, do_sampleTrue, temperature0.7 ) decoded tokenizer.batch_decode(outputs, skip_special_tokensTrue) results.extend(decoded) return results按长度排序再分批这个技巧很实用能把padding的浪费降低不少。代价是输出顺序和输入顺序不一致需要额外记录索引来还原。5. 服务化把模型变成别人能调用的接口5.1 为什么不用Flask自带的开发服务器模型跑通之后下一步是暴露成HTTP接口。很多人第一反应是用Flask写个路由然后app.run()启动。但Flask自带的开发服务器是单线程的一次只能处理一个请求而且性能很差。生产环境必须用WSGI服务器比如gunicorn或者uwsgi。但AI推理服务和普通Web服务有个关键区别模型加载一次要占很多显存不能每个worker都加载一份。如果用gunicorn起多个worker每个worker都会加载模型显存直接爆掉。所以AI推理服务的部署架构通常是一个进程加载模型多个线程或协程处理请求。FastAPI配合uvicorn的异步模式比较适合这个场景。模型在启动时加载一次请求处理函数里直接调用模型用锁或者队列来保证并发安全。from fastapi import FastAPI from pydantic import BaseModel import torch import asyncio app FastAPI() # 全局模型启动时加载 model None tokenizer None lock asyncio.Lock() class GenerateRequest(BaseModel): prompt: str max_new_tokens: int 128 temperature: float 0.7 class GenerateResponse(BaseModel): text: str app.on_event(startup) async def load_model(): global model, tokenizer tokenizer AutoTokenizer.from_pretrained(your-model-path) model AutoModelForCausalLM.from_pretrained( your-model-path, torch_dtypetorch.float16, device_mapcuda:0 ) model.eval() app.post(/generate, response_modelGenerateResponse) async def generate(request: GenerateRequest): async with lock: # 推理是同步阻塞的放到线程池里跑 loop asyncio.get_event_loop() result await loop.run_in_executor( None, sync_generate, request.prompt, request.max_new_tokens, request.temperature ) return GenerateResponse(textresult)这里用asyncio.Lock保证同一时间只有一个请求在跑推理因为GPU推理本身是串行的多个请求同时跑反而会互相干扰。如果要做批处理可以维护一个请求队列攒够一批再一起推理。5.2 请求队列与动态批处理的实现思路动态批处理是提升吞吐量的关键。核心思路是请求进来之后不立即处理而是放到队列里等一小段时间或者攒够一定数量再合并成一个batch一起推理。这个“一小段时间”很关键。设太短了攒不到足够多的请求设太长了用户等待时间增加。我一般会设10到50毫秒具体看请求的QPS和延迟要求。实现上可以用一个后台任务不断从队列里取请求import asyncio from collections import deque request_queue deque() batch_event asyncio.Event() async def batch_worker(): while True: await batch_event.wait() batch_event.clear() # 等待一小段时间攒请求 await asyncio.sleep(0.02) # 取出当前队列里的所有请求 batch [] while request_queue and len(batch) 16: batch.append(request_queue.popleft()) if not batch: continue # 合并推理 prompts [req[prompt] for req in batch] results batch_generate(model, tokenizer, prompts) # 分发结果 for req, result in zip(batch, results): req[future].set_result(result)这个模式的好处是能把多个单条请求合并成一个batchGPU利用率大幅提升。代价是实现复杂度增加而且需要处理超时和错误传播。5.3 健康检查与优雅关闭服务上线之后健康检查和优雅关闭是两个必须做的功能。健康检查让负载均衡器知道你的服务还活着优雅关闭让正在处理的请求能跑完再退出。健康检查接口要区分“存活”和“就绪”两种状态。存活检查只判断进程还在不在就绪检查要判断模型是否加载完成、显存是否正常。app.get(/health/live) async def liveness(): return {status: alive} app.get(/health/ready) async def readiness(): if model is None: return JSONResponse(status_code503, content{status: model not loaded}) # 检查显存 free_mem torch.cuda.mem_get_info()[0] / 1024**3 if free_mem 1.0: return JSONResponse(status_code503, content{status: low memory}) return {status: ready}优雅关闭需要监听退出信号停止接受新请求等正在处理的请求完成后再退出。uvicorn支持通过lifespan事件来处理app.on_event(shutdown) async def shutdown(): # 等待队列清空 while request_queue: await asyncio.sleep(0.1) # 释放显存 torch.cuda.empty_cache()注意优雅关闭的超时时间要设置合理。如果某个请求卡住了不能无限等待。一般设30秒到60秒超时后强制退出。6. 性能调优从能跑到跑得快的几个关键杠杆6.1 显存优化KV Cache的量化与分页管理KV Cache是推理时显存占用的大头。对于长序列KV Cache的大小可能超过模型权重本身。优化KV Cache有几个方向量化KV Cache把KV Cache从FP16降到INT8显存占用直接减半。精度损失通常很小但需要模型和推理框架的支持。分页管理借鉴操作系统的虚拟内存思想把KV Cache分成固定大小的页按需分配。这样能减少碎片提升显存利用率。vLLM这个框架就是靠这个技术把吞吐量提升了好几倍。滑动窗口对于超长序列只保留最近N个token的KV Cache更早的丢弃。代价是模型看不到更早的上下文适合对长距离依赖要求不高的场景。这些优化在from scratch的项目里不一定都要自己实现但你要知道它们的存在和原理这样才能在选型的时候做出正确判断。6.2 计算优化算子融合与CUDA GraphPyTorch默认的推理模式是eager模式每个算子单独执行中间结果都要写回显存再读出来。算子融合能把多个算子合并成一个减少显存读写次数。CUDA Graph是另一个利器。它把整个推理过程录制成一个图后续执行时直接回放省掉了CPU到GPU的调度开销。对于小模型和小batch调度开销占比很高CUDA Graph能带来明显的延迟下降。# CUDA Graph的简单示例 static_input torch.randint(0, 1000, (1, 128), devicecuda) static_output model(static_input) # 录制 g torch.cuda.CUDAGraph() with torch.cuda.graph(g): static_output model(static_input) # 回放 static_input.copy_(new_input) g.replay() # static_output就是新输入的结果CUDA Graph的局限是输入的形状必须固定动态形状的输入用不了。所以它更适合输入长度固定的场景比如分类任务或者固定长度的生成任务。6.3 并发策略线程、进程还是协程AI推理服务的并发模型选择取决于你的瓶颈在哪里。如果瓶颈在GPU计算那么并发数设太高反而会导致显存竞争和上下文切换开销。如果瓶颈在数据预处理那么可以用多线程或协程来重叠IO和计算。我的经验是GPU推理用单线程或少量线程数据预处理用线程池。FastAPI的异步接口配合run_in_executor把推理放到线程池里是比较平衡的方案。如果要做多卡推理可以用多个进程每个进程绑定一张卡前面加一个负载均衡。这种架构的扩展性最好但部署复杂度也最高。并发模型适用场景优点缺点单线程低QPS、调试简单、无竞争吞吐量低多线程中等QPS、IO密集实现简单GIL限制、显存竞争多进程高QPS、多卡扩展性好部署复杂、显存翻倍协程高并发、IO密集资源占用低需要异步编程7. 踩坑实录那些让我熬夜排查的典型问题7.1 显存泄漏为什么服务跑久了就OOM显存泄漏是AI服务最常见的问题之一。表现是服务刚启动时正常跑几个小时或几天后突然OOM。原因通常有几个PyTorch的缓存分配器PyTorch会缓存已分配的显存不立即还给系统。这本身不是泄漏但如果你的请求形状变化很大缓存会碎片化导致明明有足够的总空闲显存但分配不出连续的大块。未释放的中间变量在推理循环里如果某些张量被意外地保持了引用它们就不会被回收。比如把loss或logits存到了全局列表里或者闭包捕获了不该捕获的变量。CUDA上下文累积每次创建新的CUDA流或事件都会占用一些显存。如果频繁创建不销毁也会累积。排查显存泄漏我一般用这几个手段# 打印当前显存分配情况 print(torch.cuda.memory_summary()) # 记录显存快照对比不同时间点的差异 torch.cuda.memory._record_memory_history() # ... 跑一段时间 ... torch.cuda.memory._dump_snapshot(snapshot.pickle)memory_summary能看出当前分配了多少、缓存了多少、碎片率是多少。如果缓存占比很高但分配不出来就是碎片问题可以通过设置PYTORCH_CUDA_ALLOC_CONFexpandable_segments:True来缓解。7.2 推理结果不稳定temperature和随机种子的坑有段时间我发现同一个请求每次返回的结果都不一样但temperature设的是0。排查后发现temperature为0时理论上应该做贪心解码但有些实现里如果同时设了top_k或top_p还是会引入随机性。正确的做法是temperature为0时直接取argmax不要走采样流程。或者显式设置随机种子import torch import random import numpy as np def set_seed(seed42): torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) random.seed(seed) np.random.seed(seed) torch.backends.cudnn.deterministic True torch.backends.cudnn.benchmark False但要注意设置cudnn.deterministicTrue会降低性能因为一些非确定性算法被禁用了。只在需要严格复现的场景下开启。7.3 服务启动慢模型加载的优化空间大模型加载慢是常态7B模型从磁盘加载到显存大概需要几十秒到几分钟。如果服务需要频繁重启这个时间就很痛苦。优化加载速度有几个方向使用更快的存储NVMe SSD比普通SATA SSD快好几倍模型文件放在NVMe上能明显缩短加载时间。模型格式转换把PyTorch的bin格式转成safetensors格式加载速度更快而且更安全。safetensors支持内存映射不需要把整个文件读进内存再解析。预热服务启动后先跑几个推理请求把CUDA内核编译缓存和显存分配都预热好这样第一批真实请求的延迟会低很多。# 预热 def warmup(model, tokenizer, num_iterations3): dummy_input tokenizer(warmup, return_tensorspt).to(model.device) for _ in range(num_iterations): with torch.no_grad(): model.generate(**dummy_input, max_new_tokens10) torch.cuda.synchronize()预热这个步骤在from scratch的项目里经常被忽略但对线上服务的首请求延迟影响很大。7.4 日志与监控出问题了怎么快速定位AI服务的日志不能只记“请求进来、请求出去”还要记录关键的性能指标推理耗时、显存占用、batch大小、输入输出长度。这些指标在排查问题时非常有用。我一般会在请求处理的前后打点import time import logging logger logging.getLogger(__name__) async def generate(request): start time.time() mem_before torch.cuda.memory_allocated() / 1024**3 result await do_generate(request) elapsed time.time() - start mem_after torch.cuda.memory_allocated() / 1024**3 logger.info( fgenerate done | input_len{len(request.prompt)} | foutput_len{len(result)} | elapsed{elapsed:.3f}s | fmem_delta{mem_after - mem_before:.3f}GB ) return result这些日志积累下来就能看出服务的性能趋势。比如显存delta持续为正说明有泄漏elapsed逐渐增大说明有资源竞争或碎片化。监控方面Prometheus加Grafana是标配。把推理延迟、QPS、显存占用、错误率这几个指标暴露出来设置告警阈值出问题能第一时间知道。8. 从Demo到生产还差哪些工程化的工作8.1 配置管理别把参数写死在代码里from scratch的项目最容易犯的错是把所有参数都硬编码在代码里。模型路径、最大长度、batch大小、端口号全写在代码里。换一个环境就要改代码非常容易出错。正确的做法是用配置文件代码只读配置不关心具体值。配置文件可以用YAML或JSON我偏好YAML因为支持注释可读性更好。model: path: /models/llama-7b dtype: float16 device: cuda:0 max_length: 2048 serving: host: 0.0.0.0 port: 8000 max_batch_size: 16 batch_timeout_ms: 20 generation: max_new_tokens: 256 temperature: 0.7 top_k: 50 top_p: 0.9代码里用argparse或pydantic来解析配置启动时校验必填项。这样部署到不同环境只需要换配置文件代码不用动。8.2 版本管理模型、代码、配置的对应关系AI项目和普通软件项目的一个区别是模型权重也是需要版本管理的。代码回滚了模型没回滚结果对不上模型更新了配置没更新参数不匹配。我的做法是给每次发布打一个组合版本号记录代码的git commit、模型的哈希值、配置文件的版本。部署的时候按这个组合版本来任何一个变了就是一个新版本。{ version: 1.2.0, code_commit: a1b2c3d, model_hash: sha256:..., config_version: v3, created_at: 2024-01-15T10:30:00Z }这个版本信息可以在健康检查接口里返回方便确认线上跑的是哪个版本。8.3 压力测试上线前必须做的功课服务开发完了别急着上线。先做压力测试搞清楚几个关键指标最大QPS、P99延迟、显存峰值、错误率拐点。压测工具可以用locust或wrk。我一般会从低并发开始逐步增加观察每个指标的变化。当延迟开始非线性增长或者错误率开始上升就找到了系统的容量上限。压测的时候要注意测试数据和真实数据的分布要接近。用一堆短请求压出来的QPS跟真实场景下长短混合的QPS可能差好几倍。# 用wrk做简单压测 wrk -t4 -c16 -d60s --latency -s post.lua http://localhost:8000/generatepost.lua里定义请求体模拟真实的输入长度分布。压测结果里的P99延迟比平均延迟更有参考价值因为线上超时通常是由长尾请求触发的。8.4 灰度发布与回滚预案AI服务的更新风险比普通服务高因为模型行为的变化很难完全预测。新模型可能在测试集上指标更好但在真实流量上表现更差。所以灰度发布是必须的。灰度发布的策略可以是按流量比例比如先切5%的流量到新版本观察一段时间没问题再逐步扩大。也可以是按用户分组先让内部用户用新版本再逐步放开。回滚预案要提前准备好。新版本出问题时能快速切回旧版本。这要求旧版本的模型和代码都还在没有被覆盖。所以部署的时候不要直接覆盖而是新版本部署到新的目录或容器通过路由切换来生效。9. 写在最后from scratch的意义在于知道边界在哪把整个流程走一遍之后你会发现AI工程的核心难点不在模型本身而在模型之外的工程约束显存怎么管、并发怎么控、数据怎么流、故障怎么查。这些东西在调包的时候是黑盒只有自己动手搭一遍才能建立起对系统边界的直觉。我现在看到一个新的推理框架第一反应不是看它的API怎么用而是看它的KV Cache怎么管理、批处理怎么调度、显存怎么分配。这些才是决定一个框架能不能用在生产环境的关键。而这种判断力就是从一次次from scratch的实践中积累出来的。如果你正在走这条路我的建议是先跑通一个最小的闭环哪怕性能很差、代码很丑先让它跑起来。然后一个模块一个模块地优化每次只改一个变量记录改之前和改之后的指标。这样积累下来的经验比看十篇教程都管用。
1. 从零手搓AI工程为什么我不建议你直接调包很多人一听到“AI工程”这四个字第一反应就是打开某个云平台拖几个组件调几个API然后跑通一个Demo就觉得自己已经入门了。我刚开始接触这个方向的时候也是这么想的直到有一次线上推理服务在高峰期直接雪崩日志里全是显存溢出的报错我才意识到——只会调包的人永远不知道系统在什么边界条件下会崩。ai-engineering-from-scratch这个标题核心不在“AI”而在“from scratch”。它代表的是一种从底层往上搭的学习路径不依赖现成的高层封装而是自己动手把数据管道、模型加载、推理调度、服务暴露这几块拼起来。这样做的好处不是让你重复造轮子而是让你在造轮子的过程中真正理解每个环节的约束在哪里。这篇文章适合三类人看。第一类是有一定Python基础、想转AI工程但一直被各种框架绕晕的开发者第二类是做过后端或数据工程、想补齐AI系统落地能力的工程师第三类是已经会用高层API、但遇到性能问题不知道怎么下手的同学。我会把整个从零搭建的过程拆成几个关键模块每个模块都讲清楚“为什么这么设计”和“实际跑起来会遇到什么”。需要提前说明的是下面涉及的具体参数和代码示例是基于我在实际项目中的常见实践做的合理补充不是唯一解但都是经过验证能跑通的方案。你可以根据自己的硬件条件和业务场景做调整。2. 环境底座别急着装框架先把运行时理清楚2.1 Python版本与包管理器的选择逻辑从零做AI工程第一个要做的决定不是选哪个模型而是选Python版本和包管理工具。我见过太多人在这上面栽跟头用系统自带的Python 3.8结果某个依赖要求3.10以上的语法特性折腾半天或者用pip全局安装把系统环境搞得一团糟最后只能重装系统。我的建议很明确用Python 3.10或3.11配合venv或者conda做环境隔离。为什么是这两个版本3.10引入了结构化模式匹配很多现代AI工具链的配置解析用到了这个特性3.11在性能上有明显提升尤其是启动速度和内存占用。再高的版本比如3.12部分底层库的wheel包还没跟上编译安装会让你怀疑人生。包管理器方面如果你只是做纯Python的AI工程venv加pip就够了。但如果你需要管理CUDA版本、编译C扩展conda会更省心。我个人的习惯是本地开发用conda建环境生产部署用venv加requirements.txt锁定版本。这样既能享受conda的依赖解析能力又能保证部署时的可复现性。具体操作上建环境的时候把Python版本显式指定conda create -n ai-eng python3.11 -y conda activate ai-eng然后第一件事不是装torch而是升级pip和setuptoolspip install --upgrade pip setuptools wheel这个顺序很重要。老版本的pip在解析复杂依赖树时经常做出错误决策导致装出来的包版本冲突。升级之后再装其他东西能省掉后面很多麻烦。2.2 硬件资源的盘点与显存预算在写任何代码之前你必须对自己手里的硬件有清醒的认识。AI工程和普通后端开发最大的区别在于显存是硬约束不是软约束。CPU不够可以排队内存不够可以swap但显存不够就是直接报错没有商量余地。我一般会用一个简单的表格来盘点资源资源类型检查命令关注指标GPU显存nvidia-smi总显存、当前占用、温度系统内存free -h可用内存、swap使用磁盘空间df -h模型文件存放分区剩余空间CPU核心nproc并行数据加载能力显存预算这块我踩过最大的坑是低估了推理时的峰值占用。一个7B参数的模型FP16精度下权重占14GB左右但实际推理时还要加上KV Cache、中间激活值、CUDA上下文开销。实测下来7B模型做batch size为1的推理峰值显存大概在16到18GB之间。如果你只有16GB显存的卡跑起来会非常紧张稍微长一点的输入就会OOM。所以我的经验法则是模型权重的显存占用乘以1.3再加上2GB的余量才是你能安全运行的显存下限。这个系数不是拍脑袋来的1.3覆盖了KV Cache和激活值的开销2GB是给CUDA运行时和碎片留的缓冲。2.3 目录结构一开始就按工程化来组织很多人做from scratch的项目习惯把所有文件堆在一个目录里跑通了再说。但AI工程的特点是文件类型多有数据、有模型权重、有配置、有日志、有缓存。一开始不规划好后面找文件会找到崩溃。我推荐的结构是这样的ai-eng-project/ ├── configs/ # 配置文件yaml或json ├── data/ # 原始数据和预处理后的数据 │ ├── raw/ │ └── processed/ ├── models/ # 模型权重和tokenizer文件 ├── src/ # 源代码 │ ├── data/ # 数据加载和预处理 │ ├── model/ # 模型定义和加载 │ ├── serving/ # 推理服务 │ └── utils/ # 通用工具 ├── logs/ # 运行日志 ├── scripts/ # 启动脚本和运维脚本 └── tests/ # 测试代码这个结构的好处是职责清晰。比如你要换一个模型只需要动models目录和configs里的配置src下面的代码基本不用改。要清理磁盘空间直接看data和models两个目录的大小就行。注意模型权重文件不要提交到git仓库。用.gitignore把models/和data/排除掉权重文件通过脚本下载或者从对象存储拉取。我见过有人把几个GB的权重提交上去把仓库搞爆的。3. 数据管道从原始文本到模型能吃的张量3.1 为什么数据加载是AI工程最容易翻车的地方如果你问一个有经验的AI工程师整个系统里哪个环节最容易出问题十有八九会说是数据管道。模型代码可以抄服务框架可以用现成的但数据管道跟你的业务数据强绑定每个项目的坑都不一样。从零搭建数据管道核心要解决三个问题读取效率、内存占用、格式对齐。读取效率决定了你的训练或推理能不能喂饱GPU内存占用决定了你能不能处理大规模数据格式对齐决定了模型能不能正确理解你喂进去的东西。我见过最典型的事故是训练时loss正常下降但推理时效果一塌糊涂。排查了半天才发现训练时的数据预处理和推理时的预处理用了两套不同的代码tokenizer的padding方向不一致。这种问题在调包的时候很难发现因为高层API帮你把预处理封装了你根本不知道中间发生了什么。from scratch做一遍每个环节都自己写反而能避免这类问题。3.2 文本清洗哪些该删哪些必须留原始文本数据拿到手第一步是清洗。但清洗不是越干净越好有些看似“脏”的东西其实是有效信息。我一般会做这几类处理去除控制字符比如\x00、\x1b这类不可见字符它们会干扰tokenizer的分词。统一空白符把连续的多个空格、制表符、换行符统一成单个空格或标准换行。但要注意代码类数据里的缩进是有意义的不能无脑合并。处理超长行有些数据里会出现单行几十万字符的情况这种要么截断要么按标点切分。直接喂给tokenizer会爆内存。保留标点和大小写除非你的任务明确不需要否则不要轻易去掉标点或转小写。现代tokenizer对这些信息有很好的编码能力去掉反而损失信息。清洗的代码不复杂但要有日志记录每一步删了多少条、改了多少条。这样出问题的时候能回溯。import re def clean_text(text: str) - str: # 去除控制字符保留换行和制表 text re.sub(r[\x00-\x08\x0b\x0c\x0e-\x1f\x7f], , text) # 合并连续空格不碰换行 text re.sub(r[^\S\n], , text) # 合并连续换行超过两个的 text re.sub(r\n{3,}, \n\n, text) return text.strip()这个函数看着简单但每一条正则都是踩过坑之后加上的。比如[^\S\n]这个写法\S匹配非空白字符[^\S]就是匹配空白字符再加上\n的排除就能做到合并空格但不碰换行。3.3 Tokenizer的加载与特殊token处理Tokenizer是数据和模型之间的桥梁。from scratch做AI工程你必须清楚tokenizer的这几个关键属性词表大小、特殊token的ID、padding的方向、截断的策略。加载tokenizer的时候我建议同时把它的配置打印出来看一眼from transformers import AutoTokenizer tokenizer AutoTokenizer.from_pretrained(your-model-path) print(fvocab size: {tokenizer.vocab_size}) print(fpad token: {tokenizer.pad_token}, id: {tokenizer.pad_token_id}) print(fbos token: {tokenizer.bos_token}, id: {tokenizer.bos_token_id}) print(feos token: {tokenizer.eos_token}, id: {tokenizer.eos_token_id}) print(fpadding side: {tokenizer.padding_side})这里有几个容易忽略的点。第一很多模型的pad_token默认是None你需要手动指定通常用eos_token来充当。第二padding_side在训练和推理时可能不一样训练时通常用right padding推理时用left padding这个不一致会导致结果偏差。第三truncation的策略要明确是截头还是截尾最大长度是多少。我一般会在配置里把这些都写死不依赖默认值tokenizer: path: your-model-path max_length: 2048 padding_side: right truncation_side: right pad_token: |endoftext|3.4 构建Dataset和DataLoader的实操细节PyTorch的Dataset和DataLoader是数据管道的标准组件。from scratch写的时候核心是实现__len__和__getitem__两个方法。但真正影响性能的是DataLoader的参数配置。num_workers这个参数设太小了数据加载跟不上GPU设太大了会占用大量内存和CPU。我的经验值是CPU核心数的四分之一到二分之一。比如16核的机器设4到8比较合适。设成0的话就是在主进程里加载调试的时候可以用生产环境千万别这么干。pin_memory设为True可以加速CPU到GPU的数据传输但前提是你有独立的GPU。如果是在CPU上跑设了也没用。prefetch_factor控制每个worker预取多少个batch默认是2。如果发现GPU利用率忽高忽低可以适当调大这个值但会占用更多内存。from torch.utils.data import Dataset, DataLoader class TextDataset(Dataset): def __init__(self, texts, tokenizer, max_length): self.texts texts self.tokenizer tokenizer self.max_length max_length def __len__(self): return len(self.texts) def __getitem__(self, idx): encoded self.tokenizer( self.texts[idx], max_lengthself.max_length, paddingmax_length, truncationTrue, return_tensorspt ) return { input_ids: encoded[input_ids].squeeze(0), attention_mask: encoded[attention_mask].squeeze(0) } loader DataLoader( dataset, batch_size8, shuffleTrue, num_workers4, pin_memoryTrue, prefetch_factor2 )提示在Windows上num_workers大于0可能会遇到多进程启动的问题需要在if __name__ __main__:保护下运行。Linux上没这个问题。4. 模型加载与推理把权重跑起来只是第一步4.1 模型加载的几种姿势与显存占用对比模型加载看起来简单一行from_pretrained就完事了。但在from scratch的视角下你需要知道这行代码背后发生了什么以及不同加载方式对显存的影响。常见的加载方式有这几种加载方式精度7B模型显存占用适用场景FP3232位浮点约28GB训练或高精度推理FP1616位浮点约14GB常规推理INT88位整数约7GB显存受限的推理INT44位整数约4GB消费级显卡推理FP16是最常用的推理精度显存占用和效果比较平衡。INT8和INT4需要量化工具的支持加载的时候要额外指定配置。量化的代价是精度损失具体损失多少跟模型和任务有关需要实测评估。加载的时候还有一个关键参数是device_map。如果你只有一张卡直接指定device_mapcuda:0或者把模型.to(cuda)就行。如果有多张卡可以用device_mapauto让accelerate库自动分配但自动分配不一定最优有时候会把一些层放到CPU上导致推理速度骤降。import torch from transformers import AutoModelForCausalLM model AutoModelForCausalLM.from_pretrained( your-model-path, torch_dtypetorch.float16, device_mapcuda:0, low_cpu_mem_usageTrue ) model.eval()low_cpu_mem_usageTrue这个参数值得说一下。默认情况下模型加载会先在CPU内存里完整加载一份再搬到GPU。对于大模型这会导致CPU内存峰值很高。开启这个参数后加载过程会分片进行CPU内存占用能降低不少。4.2 推理循环手动实现generate的逻辑调包的时候生成文本就是一句model.generate()。但from scratch做的话我建议你至少手动实现一次自回归生成的循环这样你才能理解KV Cache、temperature、top-k这些参数到底在干什么。自回归生成的核心逻辑很简单把输入喂给模型拿到最后一个位置的logits根据logits采样出下一个token把新token拼到输入后面重复这个过程直到遇到结束符或达到最大长度。torch.no_grad() def generate(model, tokenizer, prompt, max_new_tokens128, temperature0.7, top_k50): input_ids tokenizer.encode(prompt, return_tensorspt).to(model.device) for _ in range(max_new_tokens): outputs model(input_ids) logits outputs.logits[:, -1, :] # 取最后一个位置的logits # 温度缩放 logits logits / temperature # top-k过滤 if top_k 0: top_k_values, _ torch.topk(logits, top_k) min_top_k top_k_values[:, -1].unsqueeze(-1) logits torch.where(logits min_top_k, torch.full_like(logits, float(-inf)), logits) # 采样 probs torch.softmax(logits, dim-1) next_token torch.multinomial(probs, num_samples1) # 拼接 input_ids torch.cat([input_ids, next_token], dim-1) # 检查是否遇到eos if next_token.item() tokenizer.eos_token_id: break return tokenizer.decode(input_ids[0], skip_special_tokensTrue)这段代码没有用KV Cache所以每一步都要重新计算整个序列的注意力效率很低。实际生产中必须用KV Cache把之前算过的key和value缓存起来每步只算新token的注意力。但手动实现KV Cache比较复杂这里先展示最朴素的版本理解原理之后再去看框架里的优化实现会清晰很多。4.3 批处理推理怎么把吞吐量提上去单条推理跑通之后下一步就是批处理。批处理的核心挑战是不同请求的输入长度不一样padding到相同长度会浪费计算资源。最朴素的批处理方案是按最长序列padding然后一次性喂给模型。这样做实现简单但短请求会被长请求拖累。更好的方案是连续批处理把不同长度的请求动态组合但实现复杂度高很多。我一般会先用朴素批处理跑通然后根据实际请求的长度分布来决定要不要上连续批处理。如果大部分请求长度接近朴素批处理的浪费不大如果长度差异很大连续批处理的收益就很明显。def batch_generate(model, tokenizer, prompts, max_new_tokens128): # 按长度排序减少padding浪费 sorted_prompts sorted(prompts, keylen) # 分批每批内长度接近 batch_size 8 results [] for i in range(0, len(sorted_prompts), batch_size): batch sorted_prompts[i:ibatch_size] inputs tokenizer( batch, return_tensorspt, paddingTrue, truncationTrue, max_length2048 ).to(model.device) with torch.no_grad(): outputs model.generate( **inputs, max_new_tokensmax_new_tokens, do_sampleTrue, temperature0.7 ) decoded tokenizer.batch_decode(outputs, skip_special_tokensTrue) results.extend(decoded) return results按长度排序再分批这个技巧很实用能把padding的浪费降低不少。代价是输出顺序和输入顺序不一致需要额外记录索引来还原。5. 服务化把模型变成别人能调用的接口5.1 为什么不用Flask自带的开发服务器模型跑通之后下一步是暴露成HTTP接口。很多人第一反应是用Flask写个路由然后app.run()启动。但Flask自带的开发服务器是单线程的一次只能处理一个请求而且性能很差。生产环境必须用WSGI服务器比如gunicorn或者uwsgi。但AI推理服务和普通Web服务有个关键区别模型加载一次要占很多显存不能每个worker都加载一份。如果用gunicorn起多个worker每个worker都会加载模型显存直接爆掉。所以AI推理服务的部署架构通常是一个进程加载模型多个线程或协程处理请求。FastAPI配合uvicorn的异步模式比较适合这个场景。模型在启动时加载一次请求处理函数里直接调用模型用锁或者队列来保证并发安全。from fastapi import FastAPI from pydantic import BaseModel import torch import asyncio app FastAPI() # 全局模型启动时加载 model None tokenizer None lock asyncio.Lock() class GenerateRequest(BaseModel): prompt: str max_new_tokens: int 128 temperature: float 0.7 class GenerateResponse(BaseModel): text: str app.on_event(startup) async def load_model(): global model, tokenizer tokenizer AutoTokenizer.from_pretrained(your-model-path) model AutoModelForCausalLM.from_pretrained( your-model-path, torch_dtypetorch.float16, device_mapcuda:0 ) model.eval() app.post(/generate, response_modelGenerateResponse) async def generate(request: GenerateRequest): async with lock: # 推理是同步阻塞的放到线程池里跑 loop asyncio.get_event_loop() result await loop.run_in_executor( None, sync_generate, request.prompt, request.max_new_tokens, request.temperature ) return GenerateResponse(textresult)这里用asyncio.Lock保证同一时间只有一个请求在跑推理因为GPU推理本身是串行的多个请求同时跑反而会互相干扰。如果要做批处理可以维护一个请求队列攒够一批再一起推理。5.2 请求队列与动态批处理的实现思路动态批处理是提升吞吐量的关键。核心思路是请求进来之后不立即处理而是放到队列里等一小段时间或者攒够一定数量再合并成一个batch一起推理。这个“一小段时间”很关键。设太短了攒不到足够多的请求设太长了用户等待时间增加。我一般会设10到50毫秒具体看请求的QPS和延迟要求。实现上可以用一个后台任务不断从队列里取请求import asyncio from collections import deque request_queue deque() batch_event asyncio.Event() async def batch_worker(): while True: await batch_event.wait() batch_event.clear() # 等待一小段时间攒请求 await asyncio.sleep(0.02) # 取出当前队列里的所有请求 batch [] while request_queue and len(batch) 16: batch.append(request_queue.popleft()) if not batch: continue # 合并推理 prompts [req[prompt] for req in batch] results batch_generate(model, tokenizer, prompts) # 分发结果 for req, result in zip(batch, results): req[future].set_result(result)这个模式的好处是能把多个单条请求合并成一个batchGPU利用率大幅提升。代价是实现复杂度增加而且需要处理超时和错误传播。5.3 健康检查与优雅关闭服务上线之后健康检查和优雅关闭是两个必须做的功能。健康检查让负载均衡器知道你的服务还活着优雅关闭让正在处理的请求能跑完再退出。健康检查接口要区分“存活”和“就绪”两种状态。存活检查只判断进程还在不在就绪检查要判断模型是否加载完成、显存是否正常。app.get(/health/live) async def liveness(): return {status: alive} app.get(/health/ready) async def readiness(): if model is None: return JSONResponse(status_code503, content{status: model not loaded}) # 检查显存 free_mem torch.cuda.mem_get_info()[0] / 1024**3 if free_mem 1.0: return JSONResponse(status_code503, content{status: low memory}) return {status: ready}优雅关闭需要监听退出信号停止接受新请求等正在处理的请求完成后再退出。uvicorn支持通过lifespan事件来处理app.on_event(shutdown) async def shutdown(): # 等待队列清空 while request_queue: await asyncio.sleep(0.1) # 释放显存 torch.cuda.empty_cache()注意优雅关闭的超时时间要设置合理。如果某个请求卡住了不能无限等待。一般设30秒到60秒超时后强制退出。6. 性能调优从能跑到跑得快的几个关键杠杆6.1 显存优化KV Cache的量化与分页管理KV Cache是推理时显存占用的大头。对于长序列KV Cache的大小可能超过模型权重本身。优化KV Cache有几个方向量化KV Cache把KV Cache从FP16降到INT8显存占用直接减半。精度损失通常很小但需要模型和推理框架的支持。分页管理借鉴操作系统的虚拟内存思想把KV Cache分成固定大小的页按需分配。这样能减少碎片提升显存利用率。vLLM这个框架就是靠这个技术把吞吐量提升了好几倍。滑动窗口对于超长序列只保留最近N个token的KV Cache更早的丢弃。代价是模型看不到更早的上下文适合对长距离依赖要求不高的场景。这些优化在from scratch的项目里不一定都要自己实现但你要知道它们的存在和原理这样才能在选型的时候做出正确判断。6.2 计算优化算子融合与CUDA GraphPyTorch默认的推理模式是eager模式每个算子单独执行中间结果都要写回显存再读出来。算子融合能把多个算子合并成一个减少显存读写次数。CUDA Graph是另一个利器。它把整个推理过程录制成一个图后续执行时直接回放省掉了CPU到GPU的调度开销。对于小模型和小batch调度开销占比很高CUDA Graph能带来明显的延迟下降。# CUDA Graph的简单示例 static_input torch.randint(0, 1000, (1, 128), devicecuda) static_output model(static_input) # 录制 g torch.cuda.CUDAGraph() with torch.cuda.graph(g): static_output model(static_input) # 回放 static_input.copy_(new_input) g.replay() # static_output就是新输入的结果CUDA Graph的局限是输入的形状必须固定动态形状的输入用不了。所以它更适合输入长度固定的场景比如分类任务或者固定长度的生成任务。6.3 并发策略线程、进程还是协程AI推理服务的并发模型选择取决于你的瓶颈在哪里。如果瓶颈在GPU计算那么并发数设太高反而会导致显存竞争和上下文切换开销。如果瓶颈在数据预处理那么可以用多线程或协程来重叠IO和计算。我的经验是GPU推理用单线程或少量线程数据预处理用线程池。FastAPI的异步接口配合run_in_executor把推理放到线程池里是比较平衡的方案。如果要做多卡推理可以用多个进程每个进程绑定一张卡前面加一个负载均衡。这种架构的扩展性最好但部署复杂度也最高。并发模型适用场景优点缺点单线程低QPS、调试简单、无竞争吞吐量低多线程中等QPS、IO密集实现简单GIL限制、显存竞争多进程高QPS、多卡扩展性好部署复杂、显存翻倍协程高并发、IO密集资源占用低需要异步编程7. 踩坑实录那些让我熬夜排查的典型问题7.1 显存泄漏为什么服务跑久了就OOM显存泄漏是AI服务最常见的问题之一。表现是服务刚启动时正常跑几个小时或几天后突然OOM。原因通常有几个PyTorch的缓存分配器PyTorch会缓存已分配的显存不立即还给系统。这本身不是泄漏但如果你的请求形状变化很大缓存会碎片化导致明明有足够的总空闲显存但分配不出连续的大块。未释放的中间变量在推理循环里如果某些张量被意外地保持了引用它们就不会被回收。比如把loss或logits存到了全局列表里或者闭包捕获了不该捕获的变量。CUDA上下文累积每次创建新的CUDA流或事件都会占用一些显存。如果频繁创建不销毁也会累积。排查显存泄漏我一般用这几个手段# 打印当前显存分配情况 print(torch.cuda.memory_summary()) # 记录显存快照对比不同时间点的差异 torch.cuda.memory._record_memory_history() # ... 跑一段时间 ... torch.cuda.memory._dump_snapshot(snapshot.pickle)memory_summary能看出当前分配了多少、缓存了多少、碎片率是多少。如果缓存占比很高但分配不出来就是碎片问题可以通过设置PYTORCH_CUDA_ALLOC_CONFexpandable_segments:True来缓解。7.2 推理结果不稳定temperature和随机种子的坑有段时间我发现同一个请求每次返回的结果都不一样但temperature设的是0。排查后发现temperature为0时理论上应该做贪心解码但有些实现里如果同时设了top_k或top_p还是会引入随机性。正确的做法是temperature为0时直接取argmax不要走采样流程。或者显式设置随机种子import torch import random import numpy as np def set_seed(seed42): torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) random.seed(seed) np.random.seed(seed) torch.backends.cudnn.deterministic True torch.backends.cudnn.benchmark False但要注意设置cudnn.deterministicTrue会降低性能因为一些非确定性算法被禁用了。只在需要严格复现的场景下开启。7.3 服务启动慢模型加载的优化空间大模型加载慢是常态7B模型从磁盘加载到显存大概需要几十秒到几分钟。如果服务需要频繁重启这个时间就很痛苦。优化加载速度有几个方向使用更快的存储NVMe SSD比普通SATA SSD快好几倍模型文件放在NVMe上能明显缩短加载时间。模型格式转换把PyTorch的bin格式转成safetensors格式加载速度更快而且更安全。safetensors支持内存映射不需要把整个文件读进内存再解析。预热服务启动后先跑几个推理请求把CUDA内核编译缓存和显存分配都预热好这样第一批真实请求的延迟会低很多。# 预热 def warmup(model, tokenizer, num_iterations3): dummy_input tokenizer(warmup, return_tensorspt).to(model.device) for _ in range(num_iterations): with torch.no_grad(): model.generate(**dummy_input, max_new_tokens10) torch.cuda.synchronize()预热这个步骤在from scratch的项目里经常被忽略但对线上服务的首请求延迟影响很大。7.4 日志与监控出问题了怎么快速定位AI服务的日志不能只记“请求进来、请求出去”还要记录关键的性能指标推理耗时、显存占用、batch大小、输入输出长度。这些指标在排查问题时非常有用。我一般会在请求处理的前后打点import time import logging logger logging.getLogger(__name__) async def generate(request): start time.time() mem_before torch.cuda.memory_allocated() / 1024**3 result await do_generate(request) elapsed time.time() - start mem_after torch.cuda.memory_allocated() / 1024**3 logger.info( fgenerate done | input_len{len(request.prompt)} | foutput_len{len(result)} | elapsed{elapsed:.3f}s | fmem_delta{mem_after - mem_before:.3f}GB ) return result这些日志积累下来就能看出服务的性能趋势。比如显存delta持续为正说明有泄漏elapsed逐渐增大说明有资源竞争或碎片化。监控方面Prometheus加Grafana是标配。把推理延迟、QPS、显存占用、错误率这几个指标暴露出来设置告警阈值出问题能第一时间知道。8. 从Demo到生产还差哪些工程化的工作8.1 配置管理别把参数写死在代码里from scratch的项目最容易犯的错是把所有参数都硬编码在代码里。模型路径、最大长度、batch大小、端口号全写在代码里。换一个环境就要改代码非常容易出错。正确的做法是用配置文件代码只读配置不关心具体值。配置文件可以用YAML或JSON我偏好YAML因为支持注释可读性更好。model: path: /models/llama-7b dtype: float16 device: cuda:0 max_length: 2048 serving: host: 0.0.0.0 port: 8000 max_batch_size: 16 batch_timeout_ms: 20 generation: max_new_tokens: 256 temperature: 0.7 top_k: 50 top_p: 0.9代码里用argparse或pydantic来解析配置启动时校验必填项。这样部署到不同环境只需要换配置文件代码不用动。8.2 版本管理模型、代码、配置的对应关系AI项目和普通软件项目的一个区别是模型权重也是需要版本管理的。代码回滚了模型没回滚结果对不上模型更新了配置没更新参数不匹配。我的做法是给每次发布打一个组合版本号记录代码的git commit、模型的哈希值、配置文件的版本。部署的时候按这个组合版本来任何一个变了就是一个新版本。{ version: 1.2.0, code_commit: a1b2c3d, model_hash: sha256:..., config_version: v3, created_at: 2024-01-15T10:30:00Z }这个版本信息可以在健康检查接口里返回方便确认线上跑的是哪个版本。8.3 压力测试上线前必须做的功课服务开发完了别急着上线。先做压力测试搞清楚几个关键指标最大QPS、P99延迟、显存峰值、错误率拐点。压测工具可以用locust或wrk。我一般会从低并发开始逐步增加观察每个指标的变化。当延迟开始非线性增长或者错误率开始上升就找到了系统的容量上限。压测的时候要注意测试数据和真实数据的分布要接近。用一堆短请求压出来的QPS跟真实场景下长短混合的QPS可能差好几倍。# 用wrk做简单压测 wrk -t4 -c16 -d60s --latency -s post.lua http://localhost:8000/generatepost.lua里定义请求体模拟真实的输入长度分布。压测结果里的P99延迟比平均延迟更有参考价值因为线上超时通常是由长尾请求触发的。8.4 灰度发布与回滚预案AI服务的更新风险比普通服务高因为模型行为的变化很难完全预测。新模型可能在测试集上指标更好但在真实流量上表现更差。所以灰度发布是必须的。灰度发布的策略可以是按流量比例比如先切5%的流量到新版本观察一段时间没问题再逐步扩大。也可以是按用户分组先让内部用户用新版本再逐步放开。回滚预案要提前准备好。新版本出问题时能快速切回旧版本。这要求旧版本的模型和代码都还在没有被覆盖。所以部署的时候不要直接覆盖而是新版本部署到新的目录或容器通过路由切换来生效。9. 写在最后from scratch的意义在于知道边界在哪把整个流程走一遍之后你会发现AI工程的核心难点不在模型本身而在模型之外的工程约束显存怎么管、并发怎么控、数据怎么流、故障怎么查。这些东西在调包的时候是黑盒只有自己动手搭一遍才能建立起对系统边界的直觉。我现在看到一个新的推理框架第一反应不是看它的API怎么用而是看它的KV Cache怎么管理、批处理怎么调度、显存怎么分配。这些才是决定一个框架能不能用在生产环境的关键。而这种判断力就是从一次次from scratch的实践中积累出来的。如果你正在走这条路我的建议是先跑通一个最小的闭环哪怕性能很差、代码很丑先让它跑起来。然后一个模块一个模块地优化每次只改一个变量记录改之前和改之后的指标。这样积累下来的经验比看十篇教程都管用。
