1. 从零搭建AI工程能力为什么大多数人卡在第一步就放弃了“ai-engineering-from-scratch”这个标题我第一次看到的时候脑子里蹦出来的不是某个具体框架或者工具而是一个很现实的问题一个想入门AI工程的人到底该从哪里下手我见过太多人买了一堆课收藏了几百篇“AI学习路线图”结果三个月过去连一个能跑通的推理服务都没搭起来。问题不在于他们不够努力而在于“从零开始”这件事本身被过度神秘化了。AI工程和传统的软件开发有一个本质区别它的交付物不是确定性的。你写一个CRUD接口输入A一定得到B但你部署一个模型同样的输入可能因为显存碎片、批处理大小、量化精度甚至CUDA版本差异给出完全不同的延迟和吞吐。这就意味着AI工程师的核心能力不是“会调库”而是“能在一个不确定的系统里建立可观测、可复现、可迭代的工程闭环”。这篇文章适合谁如果你已经会写Python了解基本的命令行操作想从“跑通一个notebook”进阶到“部署一个能扛住真实请求的AI服务”那接下来的内容就是为你准备的。我不会给你一份“先学数学再学框架”的学院派路线图而是按照一个真实项目从裸机到上线的顺序把每个阶段最容易踩的坑和最关键的设计决策拆开来讲。整个过程我会用一个具体的场景贯穿从零搭建一个文本分类服务的推理端包含环境隔离、模型加载、批处理调度、性能压测和灰度发布。所有代码和配置都可以直接抄作业但更重要的是理解每一步为什么这么做。2. 环境隔离不是洁癖是AI工程的第一道防线2.1 为什么conda和venv在AI项目里都不够用很多人觉得环境隔离就是“建个虚拟环境”但AI项目的依赖地狱远比普通Web项目恐怖。PyTorch 2.0和2.1的CUDA运行时版本不兼容transformers 4.35和4.36对tokenizer的默认行为有细微差异更别提flash-attention这种需要编译的库对gcc版本还有要求。我试过在一个conda环境里同时装TensorFlow和PyTorch结果numpy版本冲突导致两边都跑不起来。正确的做法是按CUDA运行时版本划分环境而不是按项目划分。因为CUDA驱动是系统级的但CUDA Toolkit是环境级的。你可以用nvidia-smi看到驱动支持的CUDA最高版本然后所有环境都基于这个版本去装对应的PyTorch wheel。比如驱动显示CUDA 12.4那你就装cu124的PyTorch不要装cu121的否则虽然能跑但某些算子会走fallback路径性能直接打七折。具体操作上我推荐用uv替代pip和conda做包管理。uv的解析速度比pip快一个数量级而且它原生支持锁定CUDA版本。下面是我常用的环境初始化脚本# 安装uv curl -LsSf https://astral.sh/uv/install.sh | sh # 创建指定Python版本的环境 uv venv --python 3.11 .venv source .venv/bin/activate # 安装PyTorch明确指定CUDA版本 uv pip install torch2.3.0 --index-url https://download.pytorch.org/whl/cu124 # 安装其他依赖 uv pip install transformers4.41.0 fastapi0.111.0 uvicorn0.30.0注意不要用pip install torch不带index-url那样会从PyPI拉默认版本可能是CPU-only的。这个坑我踩过至少三次每次都是跑起来发现torch.cuda.is_available()返回False才反应过来。2.2 模型权重的版本管理别让“模型更新了”变成事故环境隔离只解决了代码依赖模型权重本身也需要版本管理。我见过一个团队开发环境用的是HuggingFace上某个模型的main分支生产环境用的是三个月前下载的本地副本结果一次“模型更新”导致线上分类准确率掉了15个百分点。问题出在tokenizer的added_tokens变了但没人注意到。我的做法是模型权重和tokenizer配置必须一起锁定commit hash。HuggingFace的每个模型仓库都有commit历史你可以在from_pretrained里指定revision参数from transformers import AutoModelForSequenceClassification, AutoTokenizer MODEL_REVISION a1b2c3d4e5f6 # 具体的commit hash model AutoModelForSequenceClassification.from_pretrained( bert-base-uncased, revisionMODEL_REVISION, torch_dtypetorch.float16, device_mapauto ) tokenizer AutoTokenizer.from_pretrained( bert-base-uncased, revisionMODEL_REVISION )这样即使上游仓库更新了你的服务也不会受影响。同时把MODEL_REVISION写进配置管理每次升级模型就是一次显式的版本变更需要走测试流程。2.3 显存预分配一个被低估的稳定性技巧PyTorch默认的显存分配策略是“按需分配”这在推理场景下会导致显存碎片化。特别是当你的服务处理变长输入时不同batch的序列长度不同显存分配器会不断申请和释放不同大小的块跑几个小时之后就会出现“明明还有显存但就是OOM”的情况。解决方案是在服务启动时设置PYTORCH_CUDA_ALLOC_CONF环境变量export PYTORCH_CUDA_ALLOC_CONFexpandable_segments:True,max_split_size_mb:128expandable_segments让分配器使用可扩展的内存段减少碎片max_split_size_mb限制单个内存块的最大分割尺寸避免大块被切得太碎。实测下来这个配置能让一个7B模型在24G显存上的稳定运行时间从4小时提升到72小时以上。3. 推理服务的核心不是模型是调度器3.1 为什么你的推理服务QPS上不去很多人第一次部署推理服务就是写一个FastAPI接口在/predict里直接调model.generate()。单请求测试没问题一上并发就崩。原因很简单GPU是串行设备同一时刻只能执行一个kernel。如果你不做批处理每个请求单独跑一次前向传播GPU利用率可能只有5%不到。真正的瓶颈不在模型计算而在调度。你需要一个动态批处理调度器把短时间内到达的多个请求合并成一个batch一次性送进GPU。这样GPU利用率能拉到80%以上QPS提升5到10倍。我实现过一个简化版的调度器核心逻辑如下import asyncio from collections import deque import torch class BatchScheduler: def __init__(self, model, tokenizer, max_batch_size32, max_wait_ms10): self.model model self.tokenizer tokenizer self.max_batch_size max_batch_size self.max_wait_ms max_wait_ms self.queue deque() self.lock asyncio.Lock() async def predict(self, text): future asyncio.Future() async with self.lock: self.queue.append((text, future)) if len(self.queue) self.max_batch_size: await self._flush() # 等待批处理完成 return await future async def _flush(self): if not self.queue: return batch list(self.queue) self.queue.clear() texts [item[0] for item in batch] futures [item[1] for item in batch] inputs self.tokenizer( texts, return_tensorspt, paddingTrue, truncationTrue ).to(cuda) with torch.no_grad(): outputs self.model(**inputs) probs torch.softmax(outputs.logits, dim-1) for future, prob in zip(futures, probs): future.set_result(prob.cpu().tolist())这个调度器的关键参数是max_wait_ms。如果设得太小比如1ms那基本每个请求都是单独处理批处理没意义设得太大比如100ms那延迟就上去了。我的经验值是10到20ms在延迟和吞吐之间取平衡。3.2 连续批处理从“等一批”到“流水线”上面的调度器有一个问题它必须等整个batch的所有请求都完成才能返回。如果batch里有一个请求的序列特别长其他短请求就得陪着等。这就是所谓的“队头阻塞”。更先进的方案是连续批处理continuous batching也叫迭代级调度。它的核心思想是不等整个batch完成而是每生成一个token就检查一次把已经完成的请求踢出去把新来的请求加进来。这样GPU永远不会空闲吞吐量能再提升2到3倍。vLLM和TensorRT-LLM都内置了连续批处理但如果你想自己实现核心逻辑是维护一个“运行中序列”的列表每次前向传播后检查哪些序列遇到了EOS token把它们标记为完成并释放显存然后从等待队列里补充新序列。提示自己实现连续批处理非常复杂涉及KV Cache的动态管理。生产环境建议直接用vLLM它的LLMEngine已经处理好了所有这些细节。但理解原理很重要否则你调参都不知道在调什么。3.3 批处理大小怎么定一个简单的计算公式批处理大小不是越大越好。它受两个因素约束显存和延迟。显存方面每个请求的KV Cache大小可以估算KV Cache大小 2 * num_layers * num_heads * head_dim * seq_len * batch_size * dtype_size以LLaMA-7B为例32层32个头head_dim128fp16精度2字节序列长度512batch_size82 * 32 * 32 * 128 * 512 * 8 * 2 2.1 GB这还只是KV Cache加上模型权重7B * 2字节 14GB和激活值24G显存基本就满了。所以batch_size8是一个比较安全的起点。延迟方面batch_size越大单个请求的等待时间越长。你可以用一个小实验来测固定输入长度逐步增大batch_size记录P99延迟。通常你会发现延迟在某个点之后急剧上升那个点就是你的最优batch_size。4. 性能压测别用“感觉”判断服务能不能扛4.1 压测工具选型wrk、locust还是自己写压测推理服务和压测普通HTTP服务不一样。普通服务你关心QPS和延迟推理服务你还得关心token吞吐量和显存占用曲线。wrk适合测纯HTTP吞吐但它不能模拟变长输入locust可以写Python脚本灵活但性能开销大自己写asyncio脚本最灵活但容易把客户端变成瓶颈。我的建议是用locust做功能验证用自己写的asyncio脚本做极限压测。locust的Web UI很直观适合观察不同并发下的延迟分布asyncio脚本可以精确控制请求速率和输入长度分布适合找极限。下面是一个简单的asyncio压测脚本import asyncio import aiohttp import time import random async def send_request(session, url, text): start time.perf_counter() async with session.post(url, json{text: text}) as resp: await resp.json() return time.perf_counter() - start async def main(): url http://localhost:8000/predict texts [这是一条测试文本 * random.randint(1, 10) for _ in range(1000)] async with aiohttp.ClientSession() as session: tasks [send_request(session, url, t) for t in texts] latencies await asyncio.gather(*tasks) latencies.sort() print(fP50: {latencies[len(latencies)//2]*1000:.2f}ms) print(fP95: {latencies[int(len(latencies)*0.95)]*1000:.2f}ms) print(fP99: {latencies[int(len(latencies)*0.99)]*1000:.2f}ms) asyncio.run(main())4.2 压测中必须监控的四个指标压测不是只看QPS。我每次压测都会同时盯四个指标指标含义健康范围异常时的排查方向GPU利用率SM占用率60%-90%低于60%说明批处理没生效持续100%说明计算瓶颈显存占用已分配显存稳定不增长持续增长说明有内存泄漏检查KV Cache释放P99延迟99分位响应时间小于业务容忍度突然飙升说明队头阻塞或显存碎片请求队列长度等待调度的请求数小于batch_size的2倍持续增长说明吞吐不足需要扩容或优化这四个指标要一起看。比如GPU利用率高但QPS低说明计算效率有问题可能是序列长度分布太分散GPU利用率低但队列长说明调度器有问题批处理没生效。4.3 一个真实的压测翻车案例我印象最深的一次翻车本地压测QPS跑到200信心满满上线结果生产环境QPS不到50就大量超时。排查了半天发现是网络带宽的问题。本地压测时客户端和服务端在同一台机器走loopback带宽无限生产环境客户端在另一台机器输入文本平均长度500字每个请求的body大概2KB200 QPS就是400KB/s看起来不大但服务端返回的logits向量是1000维float32每个响应8KB200 QPS就是1.6MB/s。加上TCP overhead千兆网卡直接跑满。解决方案有两个一是把响应改成只返回top-5的类别和概率响应大小降到200字节二是启用gzip压缩。两个一起上带宽降到原来的1/20。注意压测环境一定要尽量模拟生产环境的网络拓扑。如果做不到至少要在压测时监控网络带宽别让网络成为你没发现的瓶颈。5. 灰度发布模型更新不能“一刀切”5.1 为什么模型更新比代码更新更危险代码更新出问题你可以回滚因为代码是确定性的。模型更新出问题你可能连问题都发现不了。我见过一个案例新模型在离线评估集上准确率提升了2%上线后A/B测试发现点击率下降了5%。原因是新模型对某些边缘case的预测置信度分布变了导致下游的排序策略失效。这种问题在离线评估里根本看不出来。所以模型更新必须走灰度发布而且灰度的维度不能只是流量比例还要包括输入分布。具体来说我会把灰度分成三个阶段影子模式新模型和旧模型同时处理所有请求但只有旧模型的返回生效。新模型的输出只记录日志用于对比。小流量灰度1%的流量走新模型但按输入长度分层抽样确保短文本、中等文本、长文本都有覆盖。全量发布逐步扩大到100%同时监控核心业务指标。5.2 影子模式的实现细节影子模式的关键是不能影响主请求的延迟。如果你在主请求的线程里同步调用新模型那新模型的延迟会直接叠加到用户请求上。正确的做法是用异步队列import asyncio from concurrent.futures import ThreadPoolExecutor shadow_executor ThreadPoolExecutor(max_workers2) async def predict_with_shadow(text): # 主模型同步返回 main_result await main_model.predict(text) # 影子模型异步执行不阻塞主流程 loop asyncio.get_event_loop() loop.run_in_executor( shadow_executor, lambda: shadow_model.predict_sync(text) ) return main_result这里用ThreadPoolExecutor而不是ProcessPoolExecutor是因为模型已经在GPU上多进程会导致显存复制。但要注意影子模型的推理会占用GPU资源所以影子模式的流量不能太大通常控制在10%以下。5.3 灰度期间的指标对比方法灰度期间不能只看准确率。我通常会对比以下几组指标预测分布新旧模型的类别分布是否一致。如果新模型把大量样本预测到某个类别说明可能有偏。置信度分布新模型的平均置信度是否显著变化。置信度整体下降说明模型不确定需要警惕。延迟分布新模型的P99延迟是否在可接受范围内。输入长度分层指标按输入长度分桶看每个桶里的准确率变化。有时候整体指标没变但某个长度区间的指标恶化了。这些指标需要实时计算我一般用Prometheus Grafana做监控面板每5分钟刷新一次。如果某个指标超过阈值自动回滚。6. 从“能跑”到“好用”几个容易被忽略的工程细节6.1 日志里必须记录的东西推理服务的日志不是记“请求来了”“请求走了”就够的。我要求日志里必须包含请求ID、输入文本的hash不是原文避免隐私问题、输入长度、batch_size、排队时间、推理时间、显存占用、模型版本。这些字段在排查问题时缺一不可。特别是排队时间和推理时间要分开记。排队时间长说明调度器有问题推理时间长说明模型或硬件有问题。混在一起记你根本不知道瓶颈在哪。6.2 优雅关闭别让正在处理的请求变成孤儿服务重启时如果直接kill进程正在GPU上跑的请求就丢了。用户看到的是超时但你日志里可能连错误都没有。正确的做法是监听SIGTERM信号停止接受新请求等正在处理的请求完成后再退出import signal import asyncio shutdown_event asyncio.Event() def handle_sigterm(): shutdown_event.set() signal.signal(signal.SIGTERM, lambda s, f: handle_sigterm()) async def serve(): while not shutdown_event.is_set(): await asyncio.sleep(0.1) # 等待所有正在处理的请求完成 await scheduler.drain()drain()方法会等待队列里的请求全部处理完同时拒绝新请求。这个过程通常需要几秒钟取决于当前batch的大小。6.3 模型预热别让第一个请求等30秒模型加载到GPU后第一次前向传播会触发CUDA kernel编译和显存分配耗时可能是后续请求的10倍以上。如果你不做预热第一个真实请求就会超时。预热的方法很简单服务启动后用几条典型输入跑几次推理把CUDA graph和显存分配都初始化好。注意预热的输入要覆盖不同的序列长度因为不同长度会触发不同的kernel。def warmup(model, tokenizer, lengths[16, 64, 256, 512]): for length in lengths: dummy_input tokenizer( 预热文本 * (length // 4), return_tensorspt, truncationTrue, max_lengthlength ).to(cuda) with torch.no_grad(): model(**dummy_input) torch.cuda.synchronize()预热完成后第一个真实请求的延迟就能降到正常水平。6.4 错误处理模型推理失败时该返回什么模型推理可能因为各种原因失败输入太长超过模型最大长度、显存不足、CUDA错误。这些错误不能直接抛给用户也不能静默吞掉。我的做法是定义一套错误码错误码含义HTTP状态码用户提示1001输入超过最大长度400文本过长请缩短后重试1002显存不足503服务繁忙请稍后重试1003模型推理内部错误500服务异常请联系管理员1004请求超时504请求超时请重试对于1002显存不足除了返回错误还要触发告警。因为这说明你的容量规划有问题或者有内存泄漏。7. 我个人在实际操作中的体会从零搭建AI工程能力最难的不是学某个框架的API而是建立一套可观测、可复现、可回滚的工程习惯。我见过太多人能把模型跑出漂亮的指标但一上线就手忙脚乱因为他们从来没有认真对待过环境隔离、版本管理、压测和灰度这些“脏活累活”。如果你只能记住一件事那就是AI工程的核心是管理不确定性。模型本身是不确定的硬件是不确定的输入分布是不确定的。你的工程体系要做的就是把这些不确定性关进笼子里让每一次变更都有迹可循让每一个故障都能快速定位和回滚。最后分享一个我用了很久的检查清单每次上线新模型前都会过一遍环境依赖是否锁定到具体版本和commit hash模型权重和tokenizer是否指定了revision是否配置了显存预分配参数批处理调度器的max_wait_ms是否根据业务延迟要求调过压测是否覆盖了不同输入长度和并发级别是否开启了影子模式做新旧模型对比日志里是否记录了排队时间和推理时间优雅关闭和模型预热是否实现错误码和告警规则是否配置这个清单不长但每一条背后都是至少一次线上事故换来的。希望它能帮你少走一些弯路。