1. 从零手搓AI工程为什么我不建议你直接调包很多人一听到“AI工程”这四个字第一反应就是打开某个云平台拖几个组件调几个API然后跑通一个Demo就觉得自己已经入门了。我刚开始接触这个领域的时候也是这么想的直到有一次线上推理服务在高峰期直接雪崩日志里全是显存溢出的报错我才意识到只会调包的人连问题出在哪一层都定位不了。ai-engineering-from-scratch这个标题核心不在“AI”而在“from scratch”。它代表的是一种从底层往上搭的学习路径不依赖现成的高层封装而是自己动手把数据管道、模型加载、推理调度、显存管理这些环节一层层拼起来。这样做的好处不是让你重复造轮子而是让你在轮子爆胎的时候知道该拧哪颗螺丝。这篇文章适合三类人第一类是有一定Python基础但没真正碰过模型部署的开发者第二类是做过Demo但一上生产就翻车的工程师第三类是想理解AI系统全貌、不想被框架黑盒困住的技术负责人。我会围绕从零搭建一个可用的AI工程骨架把每个环节的选型理由、踩坑记录和实操细节都摊开讲。需要提前说明的是这里说的“从零”不是让你从写CUDA核函数开始那属于另一个维度的硬核。我定义的“从零”是不依赖高度集成的推理服务平台用相对底层的库把一条完整的推理链路搭起来包括环境隔离、模型加载、批处理调度、显存监控和简单的服务暴露。这条链路跑通之后你再去看那些高层框架会发现它们帮你做的事一目了然。2. 环境隔离与依赖管理别让版本冲突毁掉你的第一个模型2.1 为什么虚拟环境不是可选项而是必选项我见过太多人在自己的主力开发机上直接pip install一堆包结果跑通了一个模型之后另一个模型死活加载不了报的错还特别隐晦比如某个C库的符号找不到。这种问题的根源往往不是代码写错了而是不同模型依赖的底层库版本互相打架。AI工程涉及的东西比普通Web开发要杂得多Python包、系统级的数学库、GPU驱动、编译器工具链任何一层版本对不上都可能让整个链路崩掉。所以从零开始的第一步不是写代码而是把环境隔离做干净。我的习惯是每个项目一个独立的虚拟环境并且把环境创建的命令写进项目的Makefile或者README里确保换一台机器也能复现。具体操作上venv和conda都可以但如果涉及GPU相关的库我更倾向于用conda来管理因为它在处理非Python依赖时更省心。conda create -n ai-scratch python3.10 -y conda activate ai-scratch选Python 3.10而不是更新的版本是因为很多底层推理库对3.11和3.12的支持还不完善踩过一次编译失败的坑之后我就固定用3.10作为基线。2.2 依赖锁定requirements.txt不够得用锁文件很多人习惯手写一个requirements.txt里面只写包名不写版本或者写个大概的版本范围。这在个人项目里没问题但一旦你要复现实验结果或者部署到另一台机器这种模糊的依赖声明就是灾难。我的做法是用pip-tools生成锁文件。先在一个requirements.in里写顶层依赖然后编译出带完整版本号和哈希值的requirements.txt。pip install pip-tools pip-compile requirements.in -o requirements.txt pip-sync requirements.txtpip-sync的好处是它会把你环境里多余的包删掉保证环境和锁文件完全一致。这一点在排查“为什么我的机器上能跑服务器上跑不了”这类问题时特别有用。注意如果你用的是GPU版本的推理库锁文件里要确保CUDA相关的包版本和宿主机的驱动版本匹配。我遇到过锁文件里写的是cu121但服务器驱动只支持到cu118的情况结果就是import的时候直接报错。2.3 一个容易被忽略的细节系统级依赖也要记录Python层面的依赖锁住了不代表整个环境就锁住了。有些库依赖系统级的共享库比如libGL、libsndfile这类。这些库不在requirements.txt里但缺了就是跑不起来。我的习惯是在项目根目录放一个setup.sh把系统级依赖的安装命令也写进去并且加上注释说明每个包是给哪个模块用的。这样新人拿到项目跑一遍脚本就能把环境准备好不用一个个去猜报错信息。#!/bin/bash # 系统级依赖按需安装 apt-get update apt-get install -y libgl1-mesa-glx libsndfile1 ffmpeg这个脚本不需要多复杂关键是让环境准备这件事从“口口相传”变成“可执行文档”。3. 模型加载的三种姿势与显存占用的真实账本3.1 从checkpoint到推理引擎中间到底发生了什么很多人以为模型加载就是一行model.load_state_dict()但实际上从磁盘上的权重文件到GPU里可以执行推理的引擎中间经过了好几个步骤反序列化、张量分配、设备搬运、图优化。每一步都有显存开销而且这些开销在加载完成后不一定立刻释放。我拿一个中等规模的视觉模型做过实测权重文件本身是几百MB但加载到GPU之后显存占用会膨胀到原来的两到三倍。膨胀的部分主要来自框架的运行时上下文、中间激活值的缓存池、以及为了加速推理而做的图优化副本。理解这个账本很重要因为它决定了你在一张卡上到底能同时跑几个模型实例。如果你按权重文件大小来估算显存大概率会在运行一段时间后遇到OOM。3.2 三种加载方式的取舍全量加载、分片加载、按需加载从零搭建的时候我建议先把三种加载方式都手动实现一遍这样你对显存和延迟的权衡会有肌肉记忆。加载方式显存占用首次推理延迟适用场景全量加载高低单模型独占GPU追求低延迟分片加载中中大模型多卡拆分或显存受限按需加载低高多模型共享GPU请求稀疏全量加载最简单就是把所有权重一次性搬到GPU。分片加载是把模型按层切分不同层放在不同设备上推理时数据在设备间流动。按需加载则是只把当前计算需要的层留在GPU上其余放在内存里用的时候再搬。我实测下来按需加载在请求稀疏的场景下能省下大量显存但首次推理延迟会明显上升因为要等权重从内存搬到GPU。如果你的服务QPS很低比如每分钟才几个请求按需加载是划算的但如果是持续高并发全量加载更稳。3.3 显存监控别等OOM了才去看日志从零搭建的一个核心优势就是你可以把显存监控嵌到自己的代码里而不是依赖框架的黑盒日志。我的做法是在推理循环里定期打印显存使用情况并且记录峰值。import torch def log_gpu_memory(tag): allocated torch.cuda.memory_allocated() / 1024**2 reserved torch.cuda.memory_reserved() / 1024**2 print(f[{tag}] allocated{allocated:.1f}MB reserved{reserved:.1f}MB)allocated是实际被张量占用的显存reserved是框架向GPU申请的总显存。这两个值的差值就是缓存池的大小。如果你发现reserved远大于allocated说明缓存池里有大量碎片这时候可以考虑调整内存分配策略或者手动触发一次缓存整理。提示torch.cuda.empty_cache()能释放未使用的缓存显存但频繁调用会拖慢推理速度因为它会强制同步。我一般只在模型切换或者长时间空闲之后调用一次。4. 批处理调度吞吐量和延迟之间的那道窄门4.1 为什么单条推理是性能杀手刚搭好推理链路的时候最自然的做法是来一个请求就推理一次。这种模式在Demo阶段没问题但一旦请求量上来GPU利用率会低得可怜。原因很简单每次推理都有固定的启动开销包括内核启动、内存搬运、同步等待。单条推理时这些开销占比极高GPU大部分时间在等而不是在算。批处理的核心思想是把多个请求攒在一起一次性送进模型让GPU的计算单元被充分利用。但批处理不是简单地把请求堆起来就行它涉及一个关键的权衡批越大吞吐越高但单条请求的等待时间也越长。4.2 动态批处理的最小实现从零实现一个动态批处理调度器不需要多复杂。核心逻辑就是一个队列加一个定时器请求进来先入队调度器每隔一个很短的时间窗口比如10毫秒把队列里的请求打包成一个批次送进模型推理然后把结果分发给对应的请求方。import time import threading from queue import Queue class BatchScheduler: def __init__(self, max_batch_size8, window_ms10): self.queue Queue() self.max_batch_size max_batch_size self.window window_ms / 1000.0 def submit(self, request): self.queue.put(request) def run(self, infer_fn): while True: batch [] deadline time.time() self.window while len(batch) self.max_batch_size and time.time() deadline: try: req self.queue.get(timeout0.001) batch.append(req) except Exception: continue if batch: results infer_fn([r.data for r in batch]) for req, res in zip(batch, results): req.callback(res)这段代码的关键参数是max_batch_size和window_ms。max_batch_size决定了显存占用的上限window_ms决定了延迟的下限。我一般会先用一个保守的值跑起来然后根据实际的QPS和延迟指标慢慢调。4.3 批处理踩过的三个坑第一个坑是批次内样本长度不一致。如果你的模型对输入长度敏感比如文本模型那么把长短不一的请求放在一个批次里要么得padding到最长浪费算力要么得做更复杂的分桶策略。我的做法是在入队之前先按长度分桶同一个桶里的请求才允许打包。第二个坑是超时请求的处理。如果一个请求在队列里等了太久用户可能已经断开了连接这时候再推理就是浪费。我的做法是给每个请求加一个deadline调度器在打包之前先检查一遍过期的直接丢弃并回调超时错误。第三个坑是批处理和高优先级请求的冲突。有些请求需要低延迟有些可以容忍高延迟。如果混在一个队列里低延迟请求会被高延迟请求拖累。解决办法是维护多个优先级队列调度器按优先级依次取。5. 服务暴露与并发模型从脚本到可用接口的距离5.1 为什么我不推荐一上来就用重型Web框架从零搭建的时候很多人会直接上Django或者Spring Boot这种重型框架结果发现大部分功能用不上反而增加了调试的复杂度。我的建议是先用Python标准库里的http.server或者轻量的FastAPI把接口跑通理解请求从进入到返回的完整路径然后再根据需求决定要不要换框架。FastAPI的好处是它自带异步支持和请求校验而且和Python的类型系统结合得很好。但即使是用FastAPI我也建议先把同步版本的推理接口写出来确认模型加载和批处理逻辑没问题再改成异步。5.2 并发模型的选择线程、进程还是协程AI推理服务的并发模型和普通Web服务不太一样因为推理本身是计算密集型的而且通常要占用GPU。Python的GIL意味着多线程并不能真正并行执行计算任务所以如果你的推理是在Python层面做大量计算多线程帮助不大。我的实测经验是如果推理框架在底层释放了GIL比如PyTorch在CUDA操作时会释放那么用多线程做请求调度是可行的因为计算部分在GPU上并行Python线程只负责调度和IO。但如果推理涉及大量Python层面的预处理那么多进程更合适每个进程独立加载模型各自处理请求。协程适合IO密集型的场景比如请求排队、结果分发这些环节。我通常的组合是主进程用协程处理网络IO推理任务提交给一个线程池或者进程池批处理调度器在池子里运行。5.3 一个最小可用的推理服务骨架下面这个骨架把前面几节的内容串起来环境隔离好的Python进程加载模型启动批处理调度器暴露一个HTTP接口。from fastapi import FastAPI from pydantic import BaseModel import uvicorn app FastAPI() scheduler BatchScheduler(max_batch_size8, window_ms10) class InferRequest(BaseModel): data: list class InferResponse(BaseModel): result: list app.post(/infer, response_modelInferResponse) async def infer(req: InferRequest): # 这里简化处理实际应该走异步回调 result scheduler.submit_and_wait(req.data) return InferResponse(resultresult) if __name__ __main__: # 启动调度器线程 threading.Thread(targetscheduler.run, args(infer_fn,), daemonTrue).start() uvicorn.run(app, host0.0.0.0, port8000)这个骨架离生产还有距离比如缺少健康检查、指标暴露、优雅关闭这些。但它足够让你理解一条推理请求从HTTP进入到结果返回的完整链路。把这条链路跑通之后再往上加东西就是增量改进而不是从零猜。6. 从能跑到好用监控、日志与故障复盘的实战心得6.1 监控指标不是越多越好先盯住这四个从零搭建的服务最容易犯的错是上来就接一堆监控系统结果指标太多反而不知道看哪个。我的经验是先盯住四个核心指标请求延迟的P99、GPU利用率、显存峰值、批处理队列长度。P99延迟反映的是最慢的那批请求的体验比平均延迟更有参考价值。GPU利用率告诉你计算资源有没有被浪费。显存峰值决定了你的服务能扛多大的并发。队列长度则是预警信号如果队列持续增长说明推理速度跟不上请求速度迟早要雪崩。这四个指标我一般直接打印到日志里用最简单的文本格式方便用grep和awk做快速分析。等确实需要可视化了再接时序数据库。6.2 日志要记什么请求ID、批次大小、耗时拆解日志的价值在于故障复盘。我见过很多服务的日志只记一个“请求成功”或者“请求失败”出了问题根本没法定位。我的做法是给每个请求分配一个唯一ID日志里记录这个ID、批次大小、以及耗时的拆解排队时间、预处理时间、推理时间、后处理时间。import uuid import time def handle_request(data): req_id str(uuid.uuid4())[:8] t0 time.time() # 入队 t1 time.time() # 等待批处理 t2 time.time() # 推理 t3 time.time() print(freq{req_id} queue{t1-t0:.3f}s wait{t2-t1:.3f}s infer{t3-t2:.3f}s)有了这个拆解你就能判断延迟到底出在哪个环节。如果排队时间很长说明批处理窗口或者并发度需要调整如果推理时间很长说明模型或者硬件是瓶颈。6.3 一次真实的故障复盘显存碎片导致的间歇性OOM最后分享一个我踩过的坑。有一次服务跑得好好的突然开始间歇性OOM但重启之后又能跑一段时间。查了显存监控发现reserved一直在缓慢增长但allocated是稳定的。这说明有内存碎片在累积。根因是不同批次的输入长度差异很大框架为每个批次分配了不同大小的缓存块长时间运行后缓存池里全是碎片没有连续的大块可用。解决办法是在批处理调度器里加了长度分桶让同一批次的输入长度尽量接近减少碎片产生。另外定期在低峰期重启一次服务也能缓解这个问题。这个坑让我深刻体会到从零搭建的意义不在于省掉框架而在于当框架出问题时你有能力往下钻一层看到真正的原因。只会调包的人遇到这种问题只能重启了事而理解底层的人可以定位到碎片这个层面做出针对性的优化。6.4 后续可以继续深挖的方向这条链路跑通之后还有几个方向可以继续往下钻。一个是量化把模型权重从浮点转成整数显存占用能降一半以上但精度会有损失需要做校准。另一个是算子融合把多个小算子合并成一个大算子减少内核启动开销。还有一个是请求调度策略比如根据请求的优先级和预估耗时做更智能的排队。这些方向每一个都够写一篇独立的文章但前提是你已经把基础链路搭起来了。没有这个基础直接上量化或者算子融合出了问题你连是量化误差还是调度bug都分不清。所以我的建议是先把ai-engineering-from-scratch这条最小链路跑通跑稳再去碰那些高级优化。