☰
从零搭建AI工程能力:手写推理服务与性能优化实战
2026/10/3 9:35:48 网站建设 项目流程

1. 从零搭建AI工程能力:为什么我劝你别一上来就调包

这两年“AI工程”这个词被说得太多了,多到有点变味。招聘JD上写着“AI工程师”,进去一看是调API;课程广告里写着“从零到一”,点开是教你pip install三个库然后跑个demo。我自己带过几茬新人,也帮朋友看过不少简历,发现一个特别普遍的现象:很多人能说出Transformer的公式,却说不清一次推理请求从输入到输出中间到底经过了哪些环节、每个环节的瓶颈在哪、显存是怎么被吃掉的。

ai-engineering-from-scratch这个标题,我理解的核心不是“从零学AI理论”,而是从零把AI工程这套东西真正搭起来——不依赖现成的高级封装,自己动手把数据、模型、推理、服务、监控这条链路走通一遍。它解决的不是“模型怎么训”的问题,而是“模型训出来之后,怎么让它稳定、高效、可观测地跑在真实业务里”的问题。适合谁看?适合已经会写Python、懂一点深度学习基础,但一到工程落地就发怵的人;也适合那些天天调包、心里却没底、想补上底层认知的从业者。

我下面要聊的,就是我自己走过一遍、也踩过不少坑的这套路径。不讲虚的,讲每一步为什么这么做、参数怎么定、哪里最容易翻车。

2. 整体设计思路:为什么我坚持“手写一遍”而不是直接上框架

2.1 先想清楚:AI工程到底在工程什么

很多人把AI工程等同于“训练模型”,这是最大的误解。训练只是其中一环,而且往往是迭代最慢的一环。真正的AI工程,我习惯把它拆成五块:数据处理、模型加载与推理、服务封装、性能优化、可观测性。这五块里,训练相关的只占一小部分,剩下全是传统后端工程师熟悉的活儿,只不过对象换成了张量和显存。

为什么强调“from scratch”?因为高级框架把太多东西藏起来了。你调一个pipeline(),它帮你做了分词、张量搬运、batch、后处理,看起来很爽,但一旦线上出现延迟抖动、显存OOM、输出乱码,你根本不知道从哪查。手写一遍的目的,是让每个环节都暴露在你眼前,你知道每一步发生了什么,才有能力去优化和排查。

我自己的做法是:先用最原始的方式把链路跑通,哪怕性能很差,然后再逐步引入优化。这个顺序不能反。先上框架再回头补底层,学习曲线反而更陡,因为你连“正常应该是什么样”都没有参照。

2.2 技术选型的几个关键取舍

选型这块我踩过坑,说几个我现在的默认选择以及理由。

推理后端:早期我用Flask直接套模型,简单是简单,但并发一上来就崩。后来换成FastAPI,主要是因为它原生支持异步,配合uvicorn能扛住更高的并发。再往后如果追求极致吞吐,会考虑专门的推理服务器,但对大多数中小规模场景,FastAPI够用了,别过度设计。

模型格式:训练出来的权重(比如PyTorch的.pt)直接拿去线上推理,加载慢、占用大。我一般会转成中间格式,做一次图优化和算子融合。这一步能带来明显的启动速度和显存收益,具体数字后面讲。

批处理策略:这是AI工程和普通后端最大的区别。普通接口一次处理一个请求,AI推理如果也这样,GPU利用率会低得可怜。所以必须做动态批处理——把短时间内到达的多个请求攒成一个batch一起送进模型。但攒多久是个学问,攒太久延迟高,攒太短吞吐上不去。

下面这张表是我在不同业务场景下的经验取值,供参考:

场景类型单请求延迟要求建议最大batch攒批等待窗口备注
实时对话< 500ms8~1610~20ms延迟敏感,窗口要小
内容审核< 2s32~6450~100ms吞吐优先
离线批处理无硬要求128+不适用直接攒满再跑
搜索排序< 200ms16~325~15ms极敏感,慎用攒批

提示:攒批窗口不是拍脑袋定的,要结合你的P99延迟目标和GPU单次推理耗时反推。公式后面会给。

2.3 目录结构:一开始就规划好,别等乱了再改

我见过太多项目,代码全堆在一个main.py里,跑到后面自己都找不到东西。from scratch搭的时候,就该把结构定下来。我常用的结构是这样的:

ai-service/ ├── configs/ # 配置文件,模型路径、超参、阈值都放这 ├── core/ # 核心逻辑:模型加载、推理封装 │ ├── model.py │ ├── tokenizer.py │ └── postprocess.py ├── api/ # 接口层 │ ├── routes.py │ └── schemas.py # 请求/响应数据结构校验 ├── utils/ # 工具:日志、计时、显存监控 ├── tests/ # 测试,尤其是边界输入 └── app.py # 入口

这么分的好处是,模型相关的改动只动core/,接口改动只动api/,互不干扰。等你要换模型或者加新接口时,就知道这个结构有多省心了。

3. 核心细节拆解:每个环节的坑和参数怎么定

3.1 模型加载:别小看这一步,它能决定你服务能不能起来

模型加载看着简单,其实坑不少。第一个问题是加载时机。如果你在每次请求里加载模型,那服务基本没法用。正确做法是服务启动时加载一次,常驻内存。但这里有个细节:如果用多进程部署(比如gunicorn起多个worker),每个worker都会加载一份模型,显存直接翻倍。我早期就吃过这个亏,4个worker把一张卡的显存吃爆了。

解决办法有两个:一是用单进程多线程(配合异步),二是用共享内存或者专门的模型服务把模型独立出来。小规模场景我推荐前者,简单可控。

第二个问题是精度选择。训练常用FP32,但推理用FP16甚至INT8往往就够了,而且显存占用和速度都有明显改善。我实测过一个中等规模的模型,FP32转FP16后显存占用降了约40%,推理速度提升约30%,而输出质量在绝大多数任务上肉眼几乎看不出差异。当然,涉及数值敏感的任务(比如某些回归预测)要谨慎,得做对比验证。

# 模型加载的核心逻辑示意 import torch def load_model(model_path, device="cuda", use_half=True): # 先加载到CPU,避免直接占显存 model = torch.load(model_path, map_location="cpu") model.eval() # 必须!否则dropout等层行为不对 if use_half: model = model.half() model = model.to(device) # 关闭梯度计算,省显存 for param in model.parameters(): param.requires_grad = False return model

注意:model.eval()这行千万别漏。我有次排查一个“输出每次都不一样”的诡异问题,查了半天发现是忘了切eval模式,dropout还在起作用。

3.2 分词与预处理:最容易被忽视的性能杀手

大家注意力都在模型上,但分词和预处理往往是隐藏的瓶颈。尤其是Python里的纯Python分词实现,单条处理可能就要几毫秒到几十毫秒,batch一大,这部分耗时甚至超过模型推理本身。

我的优化思路是:能向量化就向量化,能缓存就缓存。比如分词结果如果输入重复率高,可以加一层LRU缓存。再比如padding,不要每条都pad到最大长度,而是按batch内实际最大长度pad,能省不少计算。

还有一个细节是截断策略。超长输入直接截断会丢信息,我一般用“头尾保留+中间截断”的方式,因为很多任务里开头和结尾的信息密度更高。截断长度要根据你的模型最大长度和业务实际分布来定,别直接取模型上限,那样浪费算力。

3.3 推理执行:batch、显存、延迟三者的平衡

推理执行是核心。这里最关键的概念是吞吐和延迟的权衡。增大batch能提升吞吐(单位时间处理的请求数),但会增加单请求延迟(因为要等攒批,且大batch单次计算更久)。

我一般会做一个简单的压测来确定最优batch。方法是这样:固定一批测试请求,分别用batch=1、2、4、8、16……跑,记录吞吐和P99延迟,画成曲线,找那个“吞吐已经接近饱和但延迟还没爆”的点。这个点通常就是你的最优工作点。

显存方面,要留足余量。我习惯让峰值显存占用不超过总显存的80%,剩下20%留给碎片和突发。显存不够时的表现往往是OOM直接崩,而不是优雅降级,所以宁可保守。

# 动态批处理的简化逻辑 import asyncio from collections import deque class BatchScheduler: def __init__(self, max_batch=16, wait_ms=20): self.max_batch = max_batch self.wait_ms = wait_ms self.queue = deque() async def add_request(self, item): self.queue.append(item) # 攒批逻辑:要么攒满,要么等够时间 if len(self.queue) >= self.max_batch: return await self._flush() await asyncio.sleep(self.wait_ms / 1000) return await self._flush() async def _flush(self): batch = list(self.queue) self.queue.clear() # 送进模型推理 return await self._infer(batch)

这段逻辑是简化版,真实场景还要处理超时、异常、请求取消等情况,但核心思想就是这个。

3.4 后处理与输出:别让最后一步毁了体验

后处理包括解码、格式化、业务规则过滤等。这里最常见的坑是解码参数。比如生成类任务,temperature、top_p、top_k这几个参数直接决定输出风格。温度太高输出发散,太低又死板重复。我一般从temperature=0.7、top_p=0.9起步,再根据业务调。

还有一个容易忽略的点是输出长度控制。不设上限,模型可能一直生成下去,既浪费算力又可能产生无意义内容。设上限又要考虑业务需求,太短截断,太长浪费。我的经验是设一个合理上限,同时配合“遇到结束符就停”的逻辑。

4. 实操过程:从空目录到能跑的服务

4.1 环境准备与依赖管理

环境这块,我强烈建议用虚拟环境,别在系统Python里乱装。conda或者venv都行,我个人偏好venv,轻量。依赖管理用requirements.txt或者pyproject.toml,把版本号锁死。AI这块的库版本兼容性特别差,今天能跑的代码明天可能就因为某个库升级跑不了了。

python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate pip install torch fastapi uvicorn transformers pip freeze > requirements.txt

提示:torch的安装要对应你的CUDA版本,装错了要么用不了GPU,要么直接报错。装之前先nvidia-smi看驱动支持的CUDA版本。

4.2 把模型跑起来:最小可用版本

先别管性能,先让它能跑。写一个最简单的推理函数,输入一条文本,输出结果。这一步的目的是验证模型加载、分词、推理、解码这条链路是通的。

def simple_infer(text): inputs = tokenizer(text, return_tensors="pt").to(device) with torch.no_grad(): outputs = model(**inputs) result = postprocess(outputs) return result

跑通之后,你会对整条链路的耗时有个直观感受。我建议在这里加个计时,把分词、推理、后处理各自的耗时打出来,这样你就知道瓶颈在哪了。

4.3 封装成HTTP服务

用FastAPI封装,定义好请求和响应的数据结构。这里要注意输入校验,别让非法输入把服务搞崩。

from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class InferRequest(BaseModel): text: str max_length: int = 128 class InferResponse(BaseModel): result: str latency_ms: float @app.post("/infer", response_model=InferResponse) async def infer(req: InferRequest): import time start = time.time() result = await scheduler.add_request(req.text) latency = (time.time() - start) * 1000 return InferResponse(result=result, latency_ms=latency)

启动命令:uvicorn app:app --host 0.0.0.0 --port 8000 --workers 1。注意worker数,前面说过,多worker会多份模型。

4.4 压测与调优:用数据说话

服务起来后,用压测工具打一波。我用的是locust或者简单的ab。重点看三个指标:QPS、P50延迟、P99延迟。然后调整batch大小和攒批窗口,重复压测,找到最优组合。

我做过一组对比,同一个模型,batch=1时QPS约15,P99约80ms;batch=16时QPS约120,P99约220ms。吞吐提升了8倍,延迟增加了不到3倍。对于吞吐优先的场景,这个交换非常划算。但如果你的业务要求P99必须低于150ms,那batch就不能开太大。

4.5 加监控:出问题时你能第一时间知道

监控至少要有:请求量、延迟分布、错误率、显存占用、GPU利用率。我用prometheus加grafana,轻量好用。显存和GPU利用率可以用pynvml采集。

from pynvml import nvmlInit, nvmlDeviceGetHandleByIndex, nvmlDeviceGetMemoryInfo nvmlInit() handle = nvmlDeviceGetHandleByIndex(0) def get_gpu_mem(): info = nvmlDeviceGetMemoryInfo(handle) return info.used / info.total

这个指标特别重要,显存一旦接近上限,服务随时可能崩,提前告警能救命。

5. 常见问题与排查技巧实录

5.1 服务启动就OOM怎么办

这是最常见的问题。排查顺序:先看是不是多worker导致的多份模型,是的话减worker;再看模型精度,能不能转FP16;然后看是不是加载时没释放中间变量。我遇到过一次,是加载模型时同时保留了原始权重和转换后的权重,两份都在显存里,白白浪费一倍。

5.2 延迟忽高忽低怎么查

延迟抖动通常有几个来源:攒批窗口不稳定、GC(垃圾回收)停顿、显存碎片、其他进程抢占GPU。排查时先看是不是周期性的,周期性的话多半是GC或者定时任务。我一般会在关键路径上打点,把每个阶段耗时都记下来,抖动时一看就知道是哪段。

5.3 输出结果不稳定/每次不一样

如果模型是生成类任务,本身就有随机性,这是正常的。但如果确定性任务输出不稳定,先检查model.eval()有没有加,再检查有没有随机种子没固定。还有一次我是因为输入预处理里有个字典遍历顺序不确定,导致结果飘,这种隐蔽的坑最难查。

5.4 常见问题速查表

现象可能原因排查方向解决思路
启动OOM多worker/精度过高看worker数、显存占用减worker、转FP16
延迟抖动GC/攒批不稳打点看各阶段耗时调GC参数、固定窗口
输出不稳定未eval/随机种子检查模型模式加eval、固定种子
吞吐上不去batch太小/GPU没吃满看GPU利用率增大batch、优化预处理
请求超时攒批窗口过长看窗口配置缩短窗口、加超时

5.5 几个我踩过的坑

第一个坑是日志打太多。有次我在推理循环里打了详细日志,结果IO成了瓶颈,QPS直接掉一半。后来改成采样打日志,只在异常时打全量。

第二个坑是异常没兜住。模型推理偶尔会因为输入异常抛错,如果没try-except,整个服务可能挂掉。我现在的做法是每个请求都包一层异常处理,出错返回兜底结果,同时记录日志。

第三个坑是版本没锁。有次线上更新,某个依赖库自动升级了,接口行为变了,排查了一整天才发现。从那以后所有依赖版本全部锁死。

6. 性能优化的进阶思路

6.1 算子融合与图优化

模型里的很多小算子可以合并成一个大算子,减少kernel启动开销和内存搬运。这一步通常通过模型转换工具完成,转换后能带来可观的加速。我实测过一个模型,转换后推理速度提升约25%,显存降低约15%。

6.2 量化:用精度换速度

INT8量化能把模型大小压到原来的四分之一,速度提升也很明显。但量化有精度损失,必须做充分的对比验证。我的做法是准备一批代表性测试数据,量化前后跑一遍,看关键指标差异是否在可接受范围内。

6.3 缓存:能不算就不算

如果业务里重复输入多,加一层结果缓存能省大量算力。缓存key用输入的哈希,注意要处理缓存失效和内存上限。我用的是LRU策略,简单有效。

6.4 异步与流水线

把预处理、推理、后处理拆成流水线,用异步并行起来,能进一步提升吞吐。这个复杂度较高,建议在单阶段优化到位后再考虑。

7. 我个人的一些体会

这套东西我从头走了一遍,最大的感受是:AI工程的难点不在AI,在工程。模型本身有现成的,但怎么让它稳定高效地跑起来,全是工程活。很多人卡住不是因为不懂模型,是因为不懂服务、不懂性能、不懂排查。

另外就是,别追求一步到位。先跑通,再优化,这个顺序不能乱。我见过太多人一上来就想搞动态批处理、量化、流水线,结果基础链路都没通,debug都无从下手。

最后分享一个小技巧:每次改动只动一个变量,然后压测对比。同时改好几个地方,出了问题你根本不知道是哪个改动导致的。这个习惯帮我省了无数时间。

这套路径后续还能往几个方向扩展:一是多模型管理,二是A/B测试框架,三是自动扩缩容。等基础打牢了,这些都不难加。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询