1. 从零手搓AI工程为什么我不建议你直接调包很多人一听到“AI工程”这四个字第一反应就是打开某个云平台调一个现成的大模型接口写几行胶水代码然后告诉自己“我做过AI项目了”。我刚开始也是这么想的直到有一次线上服务在高峰期直接被打挂排查了半天才发现问题根本不在模型本身而在于我对整个推理链路的资源调度、批处理策略、显存管理完全没有概念。那一刻我才意识到只会调接口和真正理解AI工程之间隔着一整条工程化的鸿沟。ai-engineering-from-scratch这个标题核心讲的其实就是一件事把AI工程当成一门需要从底层搭建的手艺来学而不是当成一个黑盒API来用。它适合那些已经会写Python、懂一点机器学习基础但一遇到“模型怎么部署”“推理怎么加速”“显存怎么省”“服务怎么扛住并发”就发懵的开发者。说白了就是帮你把AI从“能跑通demo”推进到“能上线扛流量”这个阶段。我自己走过这条路踩过的坑足够写一本小册子。所以这篇内容我不打算给你灌一堆概念而是按照一个真实项目从零搭建的顺序把每个环节为什么这么做、怎么做、做完之后怎么验证一层一层拆开讲。你跟着走一遍至少能建立起一套完整的AI工程心智模型知道每个决策背后的取舍在哪里。2. 先把地基打牢环境与依赖的工程化思维2.1 为什么虚拟环境不是可选项而是必选项我见过太多人直接在系统Python里pip install一堆包然后过两个月回来发现项目跑不起来了。AI工程涉及的东西特别杂深度学习框架、CUDA驱动、推理加速库、Web框架、数据处理库这些库之间的版本依赖关系极其脆弱。一个典型的例子是PyTorch的某个版本可能要求特定版本的CUDA而某个推理加速库又要求另一个版本的CUDA如果你没有隔离环境最后就是一团乱麻。我的做法是每个AI项目都用独立的虚拟环境而且把环境配置写成可复现的脚本。具体来说用conda或者venv创建隔离环境然后用requirements.txt或者environment.yml锁定所有依赖的精确版本。这里有个细节很多人忽略不要只写torch要写torch2.1.0这种精确版本因为AI库的次版本更新经常引入不兼容的改动。# 创建隔离环境 conda create -n ai-eng python3.10 -y conda activate ai-eng # 安装核心依赖锁定版本 pip install torch2.1.0 torchvision0.16.0 --index-url https://download.pytorch.org/whl/cu118 pip install transformers4.36.0 pip install fastapi0.104.0 uvicorn0.24.0 pip install numpy1.24.0 pandas2.0.0提示如果你的机器没有GPU把cu118换成cpu即可但要注意CPU推理的性能会差一个数量级后面讲推理优化时会专门说怎么应对。2.2 目录结构决定了你三个月后还能不能看懂自己的代码AI项目最容易变成一锅粥因为数据、模型、配置、日志、脚本全混在一起。我建议在项目一开始就定好目录结构哪怕前期看起来有点“过度设计”但三个月后你会感谢自己。我的习惯是这样分的ai-engineering-from-scratch/ ├── configs/ # 所有配置文件YAML格式 ├── data/ # 原始数据和预处理后的数据 ├── models/ # 模型权重和架构定义 ├── src/ # 核心源码 │ ├── data/ # 数据加载和预处理 │ ├── model/ # 模型定义和训练逻辑 │ ├── inference/ # 推理服务和优化 │ └── utils/ # 通用工具函数 ├── scripts/ # 训练、评估、部署脚本 ├── tests/ # 单元测试和集成测试 ├── logs/ # 运行日志 └── requirements.txt这个结构的关键在于关注点分离配置和代码分离数据和模型分离训练和推理分离。我踩过的一个坑是早期把模型路径硬编码在代码里后来换了个环境路径全失效改了几十个文件。从那以后所有路径、超参数、模型名称全部走配置文件代码里只读配置不写死任何环境相关的东西。2.3 配置管理别让超参数散落在代码各处AI工程和普通后端开发最大的区别之一就是超参数特别多。学习率、批大小、序列长度、温度系数、top-k、top-p……这些参数如果散落在代码各个角落调参的时候你会疯掉。我的做法是用YAML统一管理然后用一个配置加载类把它读进来。# configs/inference.yaml model: name: my-model path: ./models/weights device: cuda dtype: float16 inference: batch_size: 8 max_length: 512 temperature: 0.7 top_p: 0.9 server: host: 0.0.0.0 port: 8000 workers: 4# src/utils/config.py import yaml from dataclasses import dataclass dataclass class ModelConfig: name: str path: str device: str dtype: str dataclass class InferenceConfig: batch_size: int max_length: int temperature: float top_p: float def load_config(path: str): with open(path, r) as f: raw yaml.safe_load(f) return { model: ModelConfig(**raw[model]), inference: InferenceConfig(**raw[inference]), server: raw[server] }这样做的好处是换环境只需要改YAML不用动代码调参的时候所有参数一目了然而且配置可以纳入版本管理每次实验的配置都能追溯。我实测下来这套做法至少帮我省掉了30%的调试时间。3. 模型推理的工程化从能跑到跑得快3.1 推理性能的三个核心指标延迟、吞吐、显存很多人优化推理的时候只盯着“快不快”但“快”其实要拆成三个维度来看。延迟是单个请求从发出到收到结果的时间吞吐是单位时间内能处理的请求数量显存是模型加载和推理过程中占用的GPU内存。这三个指标往往是互相制约的你想降低延迟可能会减小批大小但吞吐就下来了你想提高吞吐增大批大小显存又可能爆掉。我一般会先明确业务场景的优先级。如果是实时对话类应用延迟优先批大小设小一点用流式输出让用户感觉快如果是离线批量处理吞吐优先批大小拉满延迟高一点无所谓。这个决策必须在写代码之前就想清楚否则后面优化方向会乱。指标定义优化手段代价延迟单请求响应时间减小批大小、模型量化、算子融合吞吐下降吞吐单位时间处理请求数增大批大小、请求排队、多卡并行延迟上升显存GPU内存占用量化、梯度检查点、模型切分精度损失或复杂度上升3.2 批处理最容易被低估的推理加速手段批处理是推理优化里性价比最高的手段没有之一。原理很简单GPU的并行计算能力很强但如果你一次只送一个请求进去大部分计算单元是闲置的。把多个请求打包成一个批次一起送进去GPU的利用率能提升好几倍。但批处理有个工程上的难点请求是异步到达的你不能为了凑批让先到的请求等太久。我的做法是设置一个最大等待时间和最大批大小哪个先到就触发一次推理。比如最大等待50毫秒最大批大小16这样既不会让用户等太久又能尽量凑批。import asyncio from collections import deque class BatchScheduler: def __init__(self, max_batch_size16, 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): async with self.lock: self.queue.append(request) if len(self.queue) self.max_batch_size: return await self._process_batch() await asyncio.sleep(self.max_wait) async with self.lock: if self.queue: return await self._process_batch() async def _process_batch(self): batch list(self.queue) self.queue.clear() # 这里调用实际的推理函数 results await self._infer(batch) return results注意批处理不是越大越好。批大小增大到一定程度后显存会成为瓶颈而且延迟会线性增长。我一般会从8开始试逐步翻倍观察吞吐和延迟的变化曲线找到拐点。3.3 模型量化用精度换速度和显存量化是我最常用的显存优化手段。简单说就是把模型权重从32位浮点数float32降到16位float16甚至8位整数int8。float16通常能省一半显存速度也有提升而且精度损失很小大多数场景下感知不到。int8更激进显存能降到四分之一但精度损失需要评估。import torch # float16量化最简单几乎无精度损失 model model.half().to(cuda) # 动态int8量化适合CPU推理 model_int8 torch.quantization.quantize_dynamic( model, {torch.nn.Linear}, dtypetorch.qint8 )我实测下来float16在大多数生成式模型上显存占用直接减半推理速度提升20%到40%而生成质量肉眼几乎看不出差别。int8则要看具体模型有些模型量化后会出现明显的重复生成或逻辑混乱需要仔细评估。3.4 算子融合与推理引擎选型如果你用的是PyTorch原生推理很多算子是一个一个执行的中间会有很多不必要的内存读写。算子融合就是把多个连续算子合并成一个减少内存访问开销。手动做算子融合很麻烦但你可以用现成的推理引擎比如ONNX Runtime或者TensorRT它们会自动做图优化和算子融合。我的经验是如果你的模型结构比较标准Transformer类直接用ONNX Runtime导出推理性能提升很明显而且部署简单。如果追求极致性能可以用TensorRT但转换过程比较折腾而且对模型结构有要求。选型的时候要权衡是愿意花两天时间折腾TensorRT换来30%的性能提升还是用ONNX Runtime半小时搞定换来15%的提升。大多数业务场景下后者更划算。4. 服务化部署让模型真正能被调用4.1 为什么我选FastAPI而不是FlaskAI服务有个特点请求处理时间长尤其是生成式模型并发量不一定高但单个请求耗时长。Flask是同步框架一个请求处理不完后面的请求就得等着。FastAPI是异步框架可以在等待推理结果的时候处理其他请求吞吐量明显更高。另外FastAPI自带请求校验和文档生成你定义好请求和响应的数据结构它自动帮你做参数校验和API文档。这在多人协作的时候特别有用前端同学直接看文档就能对接不用你反复解释参数格式。from fastapi import FastAPI from pydantic import BaseModel import asyncio app FastAPI() class GenerateRequest(BaseModel): prompt: str max_length: int 512 temperature: float 0.7 class GenerateResponse(BaseModel): text: str latency_ms: float app.post(/generate, response_modelGenerateResponse) async def generate(req: GenerateRequest): start asyncio.get_event_loop().time() # 这里调用推理逻辑 result await inference_engine.generate( req.prompt, req.max_length, req.temperature ) latency (asyncio.get_event_loop().time() - start) * 1000 return GenerateResponse(textresult, latency_mslatency)4.2 健康检查与优雅退出上线前必须补的课我见过太多AI服务上线后运维同学不知道怎么判断服务是否正常只能靠“能不能调通”来猜。正确的做法是提供两个端点/health用于存活检查/ready用于就绪检查。存活检查只判断进程是否还在就绪检查要判断模型是否加载完成、GPU是否可用。app.get(/health) async def health(): return {status: alive} app.get(/ready) async def ready(): if not model_loaded: return {status: not ready, reason: model loading} if not torch.cuda.is_available(): return {status: not ready, reason: cuda unavailable} return {status: ready}优雅退出也很重要。服务收到停止信号时不能直接杀进程要先把正在处理的请求处理完拒绝新请求然后释放GPU资源。否则正在推理的请求会直接失败用户体验很差。import signal import sys shutdown_flag False def handle_shutdown(signum, frame): global shutdown_flag shutdown_flag True print(收到停止信号等待当前请求处理完成...) signal.signal(signal.SIGTERM, handle_shutdown) signal.signal(signal.SIGINT, handle_shutdown)4.3 日志与监控出问题时你能看到什么AI服务的日志不能只记“请求进来了”“请求出去了”要记录关键指标请求ID、输入长度、输出长度、推理耗时、显存占用、批大小。这些数据在排查性能问题时极其重要。比如你发现延迟突然升高一看日志发现是某个请求的输入长度特别长导致批处理时其他请求被拖慢这就找到了根因。我习惯用结构化日志每条日志是一个JSON方便后续用日志系统检索和分析。import logging import json import time logger logging.getLogger(ai-service) def log_inference(request_id, input_len, output_len, latency_ms, batch_size, gpu_mem_mb): logger.info(json.dumps({ request_id: request_id, input_len: input_len, output_len: output_len, latency_ms: round(latency_ms, 2), batch_size: batch_size, gpu_mem_mb: gpu_mem_mb, timestamp: time.time() }))监控方面至少要盯住GPU利用率、显存占用、请求队列长度、P99延迟这几个指标。GPU利用率长期低于30%说明批处理没做好或者请求量不够显存占用持续接近上限说明有内存泄漏或者批大小设太大了请求队列长度持续增长说明吞吐跟不上请求速度需要扩容或优化。5. 踩坑实录那些让我熬夜的典型问题5.1 显存泄漏为什么服务跑几小时就崩这是我遇到过最隐蔽的坑。服务刚启动时显存占用正常跑几个小时后显存慢慢涨上去最后OOM崩溃。排查了很久才发现问题出在推理时没有用torch.no_grad()导致PyTorch一直在构建计算图中间变量不被释放。# 错误做法没有no_grad计算图会一直累积 def infer(model, input_ids): output model(input_ids) return output # 正确做法推理时禁用梯度计算 torch.no_grad() def infer(model, input_ids): output model(input_ids) return output除了no_grad还要注意及时把不再使用的张量从GPU上移走用del删除引用必要时调用torch.cuda.empty_cache()。但empty_cache()不能滥用它会导致显存碎片化一般只在显存紧张时用。5.2 批处理导致的延迟毛刺批处理虽然能提升吞吐但会引入延迟毛刺。比如你设了最大等待50毫秒正常情况下请求10毫秒就返回了但偶尔会遇到一个请求等了50毫秒才凑批。用户感知就是“有时候快有时候慢”。我的解决办法是动态调整等待时间请求量大的时候缩短等待时间请求量小的时候延长等待时间。class AdaptiveBatchScheduler: def __init__(self): self.max_wait 0.05 # 初始50ms self.request_rate 0.0 def update_wait_time(self): # 请求率高时缩短等待降低延迟 if self.request_rate 100: # 每秒100个请求 self.max_wait 0.01 elif self.request_rate 10: self.max_wait 0.03 else: self.max_wait 0.055.3 模型加载慢导致的启动超时大模型加载动辄几分钟如果部署平台有启动超时限制比如60秒服务还没加载完就被杀掉了。解决办法是把模型加载放到后台线程先让服务起来/ready端点返回未就绪等模型加载完再返回就绪。这样存活检查能通过就绪检查会等待部署平台不会误杀。import threading model_loaded False def load_model_background(): global model_loaded, model model load_model_from_disk() model_loaded True app.on_event(startup) async def startup_event(): thread threading.Thread(targetload_model_background) thread.start()5.4 输入长度不一致导致的批处理效率低下批处理要求同一个批次里的所有输入长度一致但实际请求的输入长度千差万别。如果简单地把短输入补齐到最长输入的长度短输入就会浪费大量计算。我的做法是按长度分桶把长度相近的请求放在同一个批次里减少补齐带来的浪费。def bucket_requests(requests, bucket_size64): buckets {} for req in requests: bucket_key len(req.input_ids) // bucket_size if bucket_key not in buckets: buckets[bucket_key] [] buckets[bucket_key].append(req) return buckets这个优化在输入长度差异大的场景下效果特别明显我实测吞吐能提升30%以上。6. 从单机到多卡扩展时的新问题6.1 数据并行与模型并行的选择单卡跑不动的时候就要考虑多卡了。多卡有两种基本策略数据并行是把模型复制到每张卡上每张卡处理不同的数据最后汇总梯度模型并行是把模型切开放到不同的卡上每张卡负责一部分计算。数据并行实现简单适合模型能放进单卡但吞吐不够的场景模型并行适合模型太大单卡放不下的场景但实现复杂通信开销大。我的建议是优先考虑数据并行因为大多数推理场景下模型是能放进单卡的瓶颈在吞吐不在模型大小。如果模型实在太大再考虑模型并行或者量化。6.2 多卡推理的负载均衡多卡推理时请求怎么分配到不同的卡上是个问题。最简单的做法是轮询但不同请求的计算量不一样轮询会导致某些卡忙死某些卡闲死。更好的做法是根据每张卡的当前队列长度来分配队列短的优先。class GPUAllocator: def __init__(self, num_gpus): self.queues {i: 0 for i in range(num_gpus)} def allocate(self): # 选择队列最短的GPU gpu_id min(self.queues, keyself.queues.get) self.queues[gpu_id] 1 return gpu_id def release(self, gpu_id): self.queues[gpu_id] - 16.3 多卡环境下的显存管理多卡环境下每张卡的显存是独立的但模型加载、中间变量、通信缓冲区都会占用显存。我踩过的一个坑是数据并行时每张卡都加载了一份完整的模型显存占用是单卡的N倍。如果显存不够要么减少批大小要么用模型并行把模型切开。另外要注意多卡通信本身也会占用显存和带宽。如果卡间通信成为瓶颈增加卡数反而会降低性能。我一般会先用单卡跑满确认单卡性能到极限了再上多卡而不是一上来就堆卡。7. 测试与验证怎么证明你的AI服务是可靠的7.1 单元测试从数据预处理到推理输出AI工程的测试比普通后端难因为输出是不确定的。但不确定不代表不能测。我的做法是分层测试数据预处理层测输入输出格式和边界情况模型推理层测输出形状、取值范围、确定性固定随机种子后输出是否一致服务层测接口的请求响应格式、错误处理、超时行为。import pytest import torch def test_model_output_shape(): model load_test_model() input_ids torch.randint(0, 1000, (1, 32)) with torch.no_grad(): output model(input_ids) assert output.shape[0] 1 assert output.shape[1] 32 def test_deterministic_output(): torch.manual_seed(42) model load_test_model() input_ids torch.randint(0, 1000, (1, 32)) with torch.no_grad(): out1 model(input_ids) out2 model(input_ids) assert torch.allclose(out1, out2)7.2 压力测试找到服务的真实上限上线前一定要做压力测试否则你根本不知道服务能扛多少并发。我一般用Locust或者wrk来做逐步增加并发数观察延迟和错误率的变化。关键是要找到那个拐点并发数增加到某个值之后延迟开始急剧上升错误率开始出现这个值就是服务的实际上限。压力测试的时候要注意测试环境和生产环境的硬件配置要一致否则测出来的数据没有参考价值。另外要测长时间稳定性跑个几小时看看有没有显存泄漏或者性能衰减。7.3 灰度发布与回滚策略AI服务更新模型或者推理逻辑后不能直接全量替换要先灰度。我的做法是保留旧版本服务新版本先接10%的流量观察延迟、错误率、输出质量确认没问题再逐步扩大比例。如果发现问题立即切回旧版本。灰度发布的关键是要有对比机制能同时看到新旧版本的指标。我一般会在日志里标记请求走的是哪个版本然后用监控面板对比两个版本的P99延迟和错误率。8. 一些让我少走弯路的经验之谈做AI工程这几年我最大的体会是模型本身只是冰山一角水面下的工程化才是决定项目成败的关键。一个准确率90%的模型如果推理延迟高、服务不稳定、显存经常爆那它还不如一个准确率80%但稳定可靠的模型。另一个体会是不要过早优化。我见过有人一上来就搞TensorRT、搞模型并行、搞分布式推理结果连基本的批处理都没做好。正确的顺序是先让服务能跑通再做批处理提升吞吐再做量化降低显存最后才考虑多卡和推理引擎。每一步优化都要有数据支撑知道优化前后的指标变化而不是凭感觉。还有一点日志和监控要尽早做。我早期做项目的时候觉得日志麻烦等出了问题才发现没有日志根本没法排查。现在我的习惯是服务还没上线日志和监控先搭好这样任何问题都有迹可循。最后分享一个我常用的排查思路当服务出现性能问题时先看GPU利用率再看显存占用再看请求队列长度最后看单请求的输入输出特征。这个顺序能帮你快速定位问题是在计算、内存、调度还是数据层面。大多数时候问题都出在调度和数据层面而不是模型本身。