1. 从零手搓AI工程为什么我不建议你直接调包很多人一听到“AI工程”这四个字第一反应就是打开某个云平台拖几个组件调几个API然后跑通一个Demo就觉得自己已经入门了。我刚开始接触这个方向的时候也是这么想的直到有一次线上推理服务在高峰期直接雪崩日志里全是显存溢出的报错我才意识到——只会调包的人永远不知道系统在什么情况下会崩更不知道怎么救。ai-engineering-from-scratch这个项目标题核心讲的其实就是一件事把AI工程当成一门真正的工程学科来对待从最底层的张量运算、模型加载、推理调度、显存管理一路搭到上层的服务编排和监控告警。它不是教你如何用某个框架跑一个MNIST手写数字识别而是教你理解当你把模型丢进生产环境之后到底会发生什么。这篇文章适合三类人看。第一类是有一定Python基础、想往AI工程方向转的开发者你不需要先成为算法专家但你需要知道一个模型从磁盘加载到显存里到底经历了什么。第二类是已经在做AI应用但总是被性能问题折磨的工程师你可能已经会调API了但你不清楚为什么你的推理延迟忽高忽低。第三类是对系统底层感兴趣的技术爱好者你想知道那些看起来神秘的“AI基础设施”到底是怎么拼出来的。我写这篇东西的出发点很简单市面上讲AI工程的文章要么停留在“pip install然后调API”的层面要么直接跳到分布式训练和CUDA优化这种深水区中间那一大段“从单机到服务化”的工程实践几乎是空白的。而这段空白恰恰是绝大多数团队真正踩坑的地方。2. 整体架构设计从单文件脚本到可维护的工程结构2.1 为什么不能把所有代码塞进一个main.py我见过太多AI项目的代码结构是这样的一个main.py里面包含了模型定义、数据加载、训练循环、推理逻辑、甚至Flask接口总共两千多行。这种代码在Demo阶段没问题但一旦你要改一个推理参数你得在两千行里翻半天而且改完之后你根本不知道会不会影响到训练部分的逻辑。从零搭建AI工程的第一步不是写模型而是定目录结构。我的习惯是按照职责边界来划分而不是按照文件类型来划分。什么意思呢很多人喜欢建一个utils/文件夹把所有工具函数塞进去结果这个文件夹最后变成了一个什么都往里扔的垃圾桶。更好的做法是按领域分project/ ├── configs/ # 配置文件按环境区分 │ ├── base.yaml │ ├── dev.yaml │ └── prod.yaml ├── core/ # 核心业务逻辑 │ ├── model/ # 模型定义与加载 │ ├── data/ # 数据处理管道 │ └── inference/ # 推理引擎封装 ├── serving/ # 服务层 │ ├── api/ # 接口定义 │ └── middleware/ # 中间件限流、鉴权、日志 ├── ops/ # 运维相关 │ ├── monitor/ # 监控指标采集 │ └── deploy/ # 部署脚本 └── tests/ # 测试用例这个结构看起来简单但它解决了一个核心问题当线上出问题的时候你能在30秒内定位到相关代码在哪个目录。比如推理延迟飙升你直接去core/inference/和serving/下面找不用在两千行的main.py里大海捞针。2.2 配置管理别把超参数硬编码在代码里我踩过最大的坑之一就是把模型路径、batch size、超时时间这些参数直接写死在代码里。结果换一个环境部署要改十几个文件改漏一个就出问题。后来我强制自己遵守一条规则任何可能随环境变化的参数必须走配置文件。配置管理我用的是YAML加环境变量覆盖的方案。基础配置放在base.yaml里不同环境用不同的YAML覆盖敏感信息比如数据库密码走环境变量。加载的时候用一个简单的递归合并逻辑import yaml import os def load_config(envdev): with open(configs/base.yaml) as f: base yaml.safe_load(f) env_path fconfigs/{env}.yaml if os.path.exists(env_path): with open(env_path) as f: override yaml.safe_load(f) base deep_merge(base, override) # 环境变量覆盖前缀统一加 APP_ for key, value in os.environ.items(): if key.startswith(APP_): path key[4:].lower().split(__) set_nested(base, path, value) return base这个方案的好处是本地开发用dev.yaml线上用prod.yamlCI/CD流水线里通过环境变量注入密钥代码本身完全不用改。deep_merge和set_nested这两个函数实现起来不到二十行但省下来的排查时间是以小时计的。注意配置文件里千万不要放密钥。我见过有人在YAML里写数据库密码然后提交到代码仓库这是安全事故级别的错误。密钥一律走环境变量或者密钥管理服务。2.3 依赖管理为什么我最终选择了锁定版本AI项目的依赖管理是个噩梦。PyTorch、CUDA、cuDNN、transformers、numpy这几个东西的版本兼容性组合能让你调一整天。我的经验是生产环境必须锁定所有依赖的精确版本包括间接依赖。requirements.txt里不要写torch2.0这种模糊版本要写torch2.1.0。然后用pip-compile或者poetry lock生成完整的依赖树锁定文件。这样做的好处是今天能跑通的代码三个月后重新部署还能跑通。我吃过这个亏一个项目上线两个月后要扩容重新装环境的时候发现transformers自动升级到了新版本API变了服务直接起不来。3. 核心模块拆解推理引擎到底该怎么封装3.1 模型加载懒加载与预热策略模型加载看起来简单torch.load()一行就完事了但在工程上这里面的门道很多。首先模型加载是阻塞的一个7B参数的模型从磁盘加载到显存可能需要几十秒。如果你的服务在启动时才加载模型那这段时间内所有请求都会超时。我的做法是分两步启动时异步加载加载完成后进行预热推理。预热的意思是用几条假数据跑一遍完整的推理流程让CUDA的kernel完成编译和缓存。不做预热的话第一条真实请求的延迟可能是后续请求的十倍以上。class ModelLoader: def __init__(self, model_path, devicecuda): self.model_path model_path self.device device self.model None self._ready False def load(self): # 加载到CPU再移到GPU避免显存碎片 self.model torch.load(self.model_path, map_locationcpu) self.model.to(self.device) self.model.eval() self._warmup() self._ready True def _warmup(self): dummy torch.zeros(1, 3, 224, 224).to(self.device) with torch.no_grad(): for _ in range(3): self.model(dummy) property def ready(self): return self._ready这里有个细节map_locationcpu然后再.to(device)比直接map_locationcuda要稳。原因是直接加载到GPU时如果模型很大加载过程中可能会产生显存碎片导致后续推理时显存不足。先加载到CPU内存再整块搬到GPU显存分配更干净。3.2 批处理与动态合并吞吐量和延迟的平衡推理服务的核心矛盾是吞吐量和延迟的权衡。单条推理延迟低但GPU利用率上不去大批量推理吞吐高但单条延迟会变大。工程上的解法是动态批处理维护一个请求队列在极短的时间窗口内比如10毫秒收集请求凑成一个批次一起推理。import asyncio from collections import deque class DynamicBatcher: def __init__(self, max_batch_size32, max_wait_ms10): 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, input_data): future asyncio.Future() async with self.lock: self.queue.append((input_data, future)) if len(self.queue) self.max_batch_size: await self._flush() return await future async def _flush(self): batch list(self.queue) self.queue.clear() inputs [item[0] for item in batch] futures [item[1] for item in batch] # 实际推理 results self._infer(inputs) for future, result in zip(futures, results): future.set_result(result)这个逻辑的关键参数是max_wait_ms。设得太小批次凑不满GPU利用率低设得太大用户等得久。我的经验值是10到20毫秒具体要看你的延迟预算。如果端到端延迟要求是100毫秒以内那批处理等待时间不要超过20毫秒。实操心得动态批处理在流量低谷时效果不明显但在流量高峰时能把GPU利用率从30%拉到80%以上。上线前一定要做压力测试观察不同并发下的P99延迟。3.3 显存管理OOM不是靠加卡解决的显存溢出OOM是AI工程里最常见的线上故障。很多人第一反应是加卡或者换大显存的卡但这治标不治本。显存问题的根源通常是碎片化和峰值占用。碎片化的产生是因为PyTorch的缓存分配器在反复申请和释放不同大小的显存块。解决方案是设置PYTORCH_CUDA_ALLOC_CONF环境变量export PYTORCH_CUDA_ALLOC_CONFmax_split_size_mb:128这个参数的意思是当分配器需要拆分一个空闲块时如果拆分后剩余部分小于128MB就不拆了直接整块分配。这样可以减少碎片代价是稍微浪费一点显存。对于大多数推理场景这个取舍是值得的。峰值占用的问题更隐蔽。比如你在推理过程中临时创建了一个大tensor做中间计算这个tensor在计算完成后应该被释放但如果它被某个变量引用着垃圾回收就不会触发。我的习惯是在推理函数里用with torch.no_grad():包裹所有计算并且在返回结果前手动del掉中间变量。4. 服务化与可观测性让系统自己告诉你哪里出了问题4.1 接口设计同步还是异步推理服务的接口设计第一个要决定的是同步还是异步。同步接口实现简单客户端发请求服务端算完返回结果。异步接口需要客户端轮询或者用WebSocket接收结果实现复杂但适合长耗时任务。我的判断标准是如果单次推理时间超过5秒考虑异步否则同步就够了。大多数模型推理在几百毫秒到几秒之间同步接口完全能扛住。异步接口的复杂度在于任务状态管理、结果存储、超时清理这些都会增加出错的可能性。接口的输入输出格式我强烈建议用JSON Schema做校验。不要相信客户端传来的数据一定要在入口处做类型和范围检查。我见过因为客户端传了一个空字符串导致整个服务崩溃的案例就是因为没有做输入校验。from pydantic import BaseModel, Field, validator class InferenceRequest(BaseModel): text: str Field(..., min_length1, max_length512) temperature: float Field(0.7, ge0.0, le2.0) validator(text) def text_not_blank(cls, v): if not v.strip(): raise ValueError(text cannot be blank) return v.strip()Pydantic做校验的好处是校验逻辑和模型定义在一起改起来不容易漏。而且校验失败时返回的错误信息很清晰客户端能直接知道哪个字段有问题。4.2 监控指标延迟、吞吐、错误率之外还要看什么基础的监控指标大家都知道QPS、P50/P95/P99延迟、错误率。但AI服务有几个特有的指标必须监控GPU利用率如果GPU利用率长期低于50%说明批处理没做好或者模型太小。如果长期高于90%说明快到瓶颈了要考虑扩容。显存使用率显存使用率要留至少20%的余量。如果长期在90%以上一次流量高峰就可能OOM。队列等待时间请求在队列里等待被批处理的时间。这个指标突然升高说明推理速度跟不上请求速度了。模型加载时间如果是多模型服务每个模型的加载时间都要监控。加载时间异常升高可能是磁盘IO问题。我用的是Prometheus加Grafana的方案。在代码里埋点很简单from prometheus_client import Histogram, Gauge, Counter INFERENCE_LATENCY Histogram( inference_latency_seconds, Inference latency in seconds, buckets[0.01, 0.05, 0.1, 0.5, 1.0, 5.0] ) GPU_MEMORY_USAGE Gauge( gpu_memory_usage_bytes, GPU memory usage in bytes ) REQUEST_COUNT Counter( inference_requests_total, Total inference requests, [status] )埋点的时候注意Histogram的buckets要根据你的实际延迟分布来设。如果buckets设得不合理P99延迟就算不准。比如你的延迟大部分在100毫秒左右buckets里却只有0.01和5.0两个边界那算出来的P99就没有参考价值。4.3 日志规范出问题时能快速定位日志不是越多越好关键是在正确的地方打正确的日志。我的日志规范是请求入口打一条INFO日志包含请求ID、输入摘要、时间戳推理开始和结束各打一条DEBUG日志包含批次大小、耗时异常发生时打ERROR日志包含完整堆栈和请求ID定期打一条INFO日志汇报当前队列长度和GPU状态请求ID是串联所有日志的关键。客户端生成一个UUID放在请求头里服务端所有日志都带上这个ID。出问题时用请求ID一搜整个请求的生命周期一目了然。注意日志里不要打完整的用户输入尤其是涉及隐私的场景。打摘要或者哈希值就够了。我见过因为日志里记录了用户完整对话内容导致合规问题的案例。5. 常见问题与排查技巧实录5.1 推理延迟忽高忽低这是最常见的线上问题。延迟波动通常有三个原因批处理等待时间不稳定、GPU被其他进程抢占、内存交换。排查步骤先看队列等待时间的监控曲线如果队列等待时间波动大说明批处理逻辑有问题。再看GPU利用率如果利用率曲线有周期性掉底说明有其他进程在抢GPU。最后看系统内存使用率如果内存快满了操作系统会把部分内存交换到磁盘导致推理时从磁盘读数据延迟飙升。解决方案批处理等待时间用固定窗口而不是动态窗口GPU独占不要和其他服务混部系统内存预留至少30%的余量。5.2 模型输出结果不一致同一个输入两次推理结果不一样这通常是因为随机种子没有固定。PyTorch的随机性来源有三个权重初始化、Dropout、以及某些算子的非确定性实现。推理阶段要确保model.eval()被调用这会关闭Dropout。然后设置随机种子import torch import random import numpy as np def set_seed(seed42): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) torch.backends.cudnn.deterministic True torch.backends.cudnn.benchmark False注意cudnn.deterministic True会降低性能因为GPU上某些算子的确定性实现比非确定性实现慢。如果对性能要求高且能接受微小差异可以不开这个选项。5.3 服务启动后第一次请求特别慢这是CUDA的惰性初始化导致的。第一次执行某个算子时CUDA需要编译kernel并加载到GPU这个过程可能耗时几百毫秒到几秒。解决方案就是前面提到的预热启动时用假数据跑几遍完整流程。预热的输入数据要覆盖各种形状。比如你的模型支持变长输入那预热时就要用最短、最长、以及中间长度的输入各跑一遍。只预热一种形状的话其他形状的第一次请求还是会慢。5.4 常见问题速查表问题现象可能原因排查方法解决方案推理延迟突然飙升GPU被抢占或显存不足查看GPU利用率和显存监控独占GPU设置显存上限服务启动后第一次请求慢CUDA惰性初始化对比第一次和后续请求延迟启动时预热推理输出结果不一致随机种子未固定同一输入多次推理对比设置随机种子调用eval()显存缓慢增长不释放内存泄漏监控显存随时间变化检查中间变量引用手动del批处理吞吐上不去批处理窗口太小查看平均批次大小增大批处理等待时间请求超时但GPU空闲队列阻塞或死锁查看队列长度和线程状态检查异步逻辑加超时机制5.5 一个真实的排查案例有一次线上服务在凌晨三点突然开始报超时但GPU利用率只有10%。我远程连上去看发现请求队列长度在持续增长但推理线程好像卡住了。加了日志之后发现推理线程在等待一个锁而这个锁被一个已经崩溃的请求持有。原因是这样的我在推理函数里加了一个全局锁来保护模型状态但某个请求在持有锁的时候抛了异常异常被上层捕获了但锁没有被释放。后续所有请求都在等这个永远不会释放的锁。修复方案很简单用with lock:代替手动acquire()和release()这样即使抛异常锁也会自动释放。这个坑让我明白了一个道理任何手动资源管理的地方都是潜在的故障点。6. 从单机到多实例扩展时要注意什么6.1 什么时候需要多实例单实例的推理服务瓶颈通常在GPU。当你发现GPU利用率长期在90%以上且延迟开始上升时就需要考虑多实例了。但多实例不是简单地把服务复制几份有几个问题必须提前解决。第一个问题是模型加载时间。每个实例都要加载一份模型如果模型很大启动时间会很长。解决方案是用共享内存或者内存映射文件多个实例共享同一份模型权重。PyTorch的torch.load支持mmapTrue参数可以把模型文件映射到内存多个进程共享物理内存。第二个问题是请求分发。需要一个负载均衡器把请求分发到各个实例。简单的轮询策略在推理场景下不一定最优因为不同实例的负载可能不一样。更好的做法是根据队列长度来分发队列短的实例优先。6.2 优雅关闭别让正在处理的请求丢失服务更新时如果直接杀掉进程正在处理的请求会失败。优雅关闭的流程是收到关闭信号后停止接受新请求等待正在处理的请求完成然后释放资源退出。import signal import asyncio class GracefulShutdown: def __init__(self, server): self.server server self.shutting_down False def setup(self): signal.signal(signal.SIGTERM, self._handle) signal.signal(signal.SIGINT, self._handle) def _handle(self, signum, frame): if self.shutting_down: return self.shutting_down True # 停止接受新请求 self.server.stop_accepting() # 等待正在处理的请求完成最多等30秒 asyncio.create_task(self._wait_and_exit()) async def _wait_and_exit(self): await self.server.wait_for_pending(timeout30) self.server.cleanup() exit(0)这个逻辑的关键是wait_for_pending的超时时间。设得太短请求处理不完设得太长发布流程会被拖慢。我的经验值是30秒覆盖绝大多数推理请求的处理时间。6.3 版本管理与灰度发布模型更新是AI服务特有的问题。代码更新可以回滚但模型更新如果出了问题回滚需要重新加载旧模型耗时更长。所以模型更新一定要做灰度发布。我的做法是新模型先加载到一个单独的实例上把1%的流量导过去观察延迟和输出质量。如果指标正常逐步增加流量比例直到100%。如果发现问题把流量切回旧实例新实例下线。这个过程中请求路由要支持按比例分流。在网关层根据请求ID的哈希值来决定走新实例还是旧实例保证同一个用户的请求始终走同一个版本避免体验不一致。7. 一些踩坑之后的个人体会做AI工程这几年我最大的体会是模型本身的质量决定了上限但工程实现决定了下限。一个90分的模型如果工程做得差线上表现可能只有60分一个70分的模型如果工程做得好线上表现能稳定在70分。另一个体会是不要过早优化。我见过有人在日请求量只有几百的时候就开始搞分布式推理、模型量化、算子融合结果复杂度上去了稳定性下来了收益却几乎为零。先把单机服务做稳把监控做全等真正遇到瓶颈了再针对性优化。还有一个很实用的建议每次线上出问题之后除了修复问题本身一定要问一句“为什么监控没有提前发现”。如果监控能提前5分钟告警你就能在用户感知到之前把问题解决掉。监控的价值不在于事后排查而在于事前预警。最后分享一个我常用的压测方法用locust或者wrk模拟真实流量但不要只测恒定流量要测阶梯式增长和突发尖峰。恒定流量下表现良好的服务在突发尖峰下可能直接崩掉。我一般会测三个场景逐步加压到系统极限、瞬间打满流量、以及长时间高负载运行。这三个场景能覆盖绝大多数线上故障模式。