1. 从零搭建AI工程能力为什么我劝你别一上来就调包这两年AI应用开发的门槛肉眼可见地降低了随便拉个框架、调个API就能跑出一个能对话的Demo。但我带过不少新人也面试过不少号称“做过大模型应用”的候选人发现一个很普遍的现象模型能跑起来但一问到“为什么这么设计”“数据怎么流转”“线上出问题怎么排查”基本就卡壳了。这就是典型的“会调包但不懂工程”。ai-engineering-from-scratch这个项目标题核心讲的其实就是一件事抛开那些封装好的高级框架从最底层的工程视角把AI应用从数据到推理再到服务的整条链路自己搭一遍。它解决的不是“模型怎么训”的问题而是“模型训出来之后怎么变成一个真正能用的系统”的问题。适合谁看我觉得有三类人最该认真过一遍一是刚入行做AI应用、只会调API的开发者二是从传统后端转过来、想补齐AI工程链路的工程师三是想搞清楚“大模型应用到底难在哪”的技术负责人。我自己最早做AI项目的时候也是直接拿现成框架往上堆结果遇到显存溢出、推理延迟抖动、并发上不去这些问题完全不知道从哪下手。后来逼着自己把整条链路拆开重写了一遍才真正理解每个环节的取舍。这篇就按我实际踩过的路把从零搭建AI工程能力的核心思路、关键细节和实操过程完整讲一遍尽量说人话能抄作业的地方直接给方案。2. 整体设计思路为什么选择“从零手搭”而不是“框架堆叠”2.1 先想清楚AI工程到底在工程什么很多人把AI工程等同于“模型训练”这是个很大的误区。训练只是其中一环真正占工作量大头的是数据管道、推理服务、资源调度、监控运维这几块。我习惯把AI工程拆成四层来看数据层原始数据怎么清洗、怎么切分、怎么版本化训练和推理的数据格式怎么对齐。模型层模型怎么加载、怎么做量化、怎么做批处理显存怎么管。服务层请求怎么进来、怎么排队、怎么并发、超时怎么处理。运维层日志怎么打、指标怎么采、出问题怎么定位、怎么灰度上线。ai-engineering-from-scratch的价值就在于它逼着你把这四层都自己实现一遍。用框架的时候这四层被封装成了几个配置项你根本不知道里面发生了什么自己搭的时候每一层的边界和代价你都会清清楚楚。2.2 方案选型为什么不用现成的高层框架我一开始也纠结明明有那么多成熟的推理框架和服务框架为什么还要自己写。后来想明白了学习阶段和 production 阶段的选型逻辑是完全不同的。学习阶段的目标是“理解每一层的代价”production 阶段的目标是“稳定和效率”。这个项目定位是前者所以选型原则是能用标准库就不用第三方比如HTTP服务直接用Python内置的http.server起步理解请求生命周期后再换生产级方案。能显式就不隐式批处理、显存分配这些宁可自己写循环也不用框架的自动机制目的是看清每一步的开销。先跑通再优化第一版一定是“能跑就行”哪怕性能很差先建立正确的数据流再逐层替换优化。提示这里说的“不用框架”是指学习阶段的理解性搭建不是让你在生产环境重复造轮子。生产环境该用vLLM、TGI还是自己写取决于你的并发量和成本这是另一个话题。2.3 整体架构一条最小可用的AI工程链路我给自己定的最小链路是这样的数据加载 → 预处理 → 模型推理 → 后处理 → HTTP服务 → 日志监控。每一环都自己实现环与环之间用明确的数据结构衔接。这样做的好处是任何一环出问题我都能快速定位到具体是哪一层的锅而不是在框架的黑盒里瞎猜。举个实际例子我最早遇到“推理结果偶尔乱码”的问题用框架的时候完全不知道从哪查。自己搭之后我在预处理和后处理各打了一个日志一眼就看出是分词器的padding策略在batch内不一致导致的。这种问题只有你把链路拆开才看得见。3. 核心细节解析数据管道与推理引擎的实操要点3.1 数据管道别小看清洗和切分数据管道是AI工程里最不起眼但最容易埋雷的地方。我见过太多项目模型效果不好最后查出来是训练数据和推理数据的预处理不一致。自己搭管道的时候我强制要求训练和推理共用同一份预处理代码哪怕性能差一点也要保证一致性。具体做法是写一个Preprocessor类训练时和推理时都调它。核心方法就两个clean()和tokenize()。clean()负责去重、去噪、统一编码tokenize()负责分词、截断、padding。这里有个坑padding的长度一定要在batch内动态计算不能写死。我一开始写死成512结果短文本浪费算力长文本被截断效果一塌糊涂。class Preprocessor: def __init__(self, tokenizer, max_len512): self.tokenizer tokenizer self.max_len max_len def clean(self, texts): # 去重、去空、统一编码 seen set() result [] for t in texts: t t.strip() if t and t not in seen: seen.add(t) result.append(t) return result def tokenize(self, texts): # 动态paddingbatch内最长为准 encoded self.tokenizer( texts, paddingTrue, truncationTrue, max_lengthself.max_len, return_tensorspt ) return encoded数据版本化也是必须的。我用最简单的方案每次预处理完把结果和对应的配置tokenizer版本、max_len、清洗规则一起存成一个带时间戳的目录并在文件名里带上配置的hash。这样任何时候都能复现某次训练用的数据。3.2 推理引擎批处理与显存管理的取舍推理引擎是性能的关键。自己搭的时候我重点解决两个问题批处理和显存管理。批处理的核心是“攒批”。请求进来先不急着推理放进一个队列等攒够一定数量或者等够一定时间再一起送进模型。这里有个经典权衡batch越大吞吐越高但延迟也越高。我的经验值是在线服务batch size控制在8到16之间比较稳离线任务可以拉到64甚至更高。import time from collections import deque class BatchScheduler: def __init__(self, max_batch16, max_wait0.05): self.max_batch max_batch self.max_wait max_wait self.queue deque() def add(self, request): self.queue.append((time.time(), request)) def get_batch(self): if not self.queue: return [] batch [] start time.time() while self.queue and len(batch) self.max_batch: ts, req self.queue.popleft() batch.append(req) if time.time() - start self.max_wait: break return batch显存管理这块我踩过最大的坑是没有及时释放中间张量。PyTorch的自动回收有时候不及时尤其是循环里反复创建大张量的时候。我的做法是显式调用torch.cuda.empty_cache()并且在推理时用torch.no_grad()包住避免不必要的计算图构建。实测下来这两个操作能让显存占用降低20%到30%。注意empty_cache()本身有开销不要在每个请求里都调建议在batch之间或者显存告警时调。3.3 服务层从单线程到并发的演进服务层我经历了三个阶段单线程 → 多线程 → 异步。每个阶段解决的问题不同代价也不同。单线程最简单一个请求处理完再处理下一个适合调试。但线上肯定不行一个慢请求就把后面全堵死了。多线程能解决并发但Python的GIL让CPU密集型的预处理成了瓶颈。最后我换成了异步方案用asyncio配合线程池把IO密集的请求接收和CPU密集的推理分开。import asyncio from concurrent.futures import ThreadPoolExecutor executor ThreadPoolExecutor(max_workers4) async def handle_request(request): loop asyncio.get_event_loop() # 预处理和推理放到线程池避免阻塞事件循环 result await loop.run_in_executor(executor, inference, request) return result这里的关键是把阻塞操作全部扔进线程池事件循环只负责调度。我实测下来4个worker线程配合异步QPS比纯多线程方案高了将近一倍。4. 实操过程从零跑通一条完整链路4.1 环境准备与依赖安装环境这块我建议用conda建独立环境避免和系统Python打架。核心依赖就几个torch、transformers、fastapi服务层用比内置的http.server好用太多、uvicorn。版本上torch和transformers的版本要匹配我一般锁死一组验证过的版本避免自动升级出问题。conda create -n ai-eng python3.10 conda activate ai-eng pip install torch2.1.0 transformers4.36.0 fastapi0.104.0 uvicorn0.24.0提示如果你的机器有GPU装torch的时候记得选对应CUDA版本的wheel不然会默认装CPU版推理慢到怀疑人生。4.2 模型加载与量化让显存扛得住模型加载我推荐用transformers的from_pretrained但要注意几个参数。torch_dtype设成float16能省一半显存device_map设成auto能自动分配多卡。如果显存还是不够就上量化8bit量化基本不掉效果4bit量化要测一下你的任务能不能接受。from transformers import AutoModelForCausalLM, AutoTokenizer import torch model_name your-model-path tokenizer AutoTokenizer.from_pretrained(model_name) model AutoModelForCausalLM.from_pretrained( model_name, torch_dtypetorch.float16, device_mapauto, load_in_8bitTrue # 显存不够时开启 ) model.eval()加载完之后我习惯先跑一个warmup用几条假数据过一遍模型把CUDA的kernel编译缓存预热好。这样第一个真实请求的延迟不会特别难看。4.3 完整推理流程从请求到响应完整的推理流程我拆成六步接收请求 → 预处理 → 攒批 → 推理 → 后处理 → 返回响应。每一步都有明确的输入输出方便打日志和排查。def inference(texts): # 1. 预处理 inputs preprocessor.tokenize(texts) inputs {k: v.to(model.device) for k, v in inputs.items()} # 2. 推理 with torch.no_grad(): outputs model.generate( **inputs, max_new_tokens128, do_sampleTrue, temperature0.7, top_p0.9 ) # 3. 后处理 results tokenizer.batch_decode(outputs, skip_special_tokensTrue) return results参数这块我重点说三个max_new_tokens控制生成长度别设太大不然显存和延迟都扛不住temperature控制随机性0.7是比较稳的默认值top_p做核采样0.9能过滤掉长尾的低质量token。这三个参数我调了很久最后发现没有万能值必须按任务测。创意类任务temperature可以到0.9事实类任务最好降到0.3以下。4.4 服务封装与压测服务层用FastAPI封装主要是因为它自带异步支持和自动文档省事。核心就一个POST接口接收文本列表返回生成结果。from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class Request(BaseModel): texts: list[str] app.post(/generate) async def generate(req: Request): loop asyncio.get_event_loop() results await loop.run_in_executor(executor, inference, req.texts) return {results: results}压测我用的是locust模拟不同并发下的QPS和P99延迟。实测下来单卡16G显存7B模型8bit量化batch size 8的情况下QPS能到3到5P99延迟在2秒左右。这个数据不算好看但作为从零搭的版本已经能说明链路的完整性了。5. 常见问题与排查技巧实录5.1 显存溢出最常见的翻车现场显存溢出OOM是我遇到频率最高的问题。原因无非几个batch太大、序列太长、中间张量没释放、模型本身太大。排查顺序我一般是先看batch和序列长度再看有没有no_grad最后看量化有没有生效。现象可能原因排查方法解决加载模型就OOM模型太大或没量化看模型参数量和dtype开8bit或4bit量化推理中途OOMbatch或序列太长打印batch内最大长度减小batch或截断反复请求后OOM中间张量未释放监控显存曲线加empty_cache多卡OOMdevice_map分配不均看每卡显存占用手动指定device_map5.2 推理结果不稳定随机性和一致性的平衡有时候同样的输入两次输出不一样这不一定是bug可能是采样参数导致的。do_sampleTrue的时候本来就有随机性。如果你需要确定性输出把do_sample设成False用贪心解码。但贪心解码容易重复实际用的时候要权衡。我踩过的一个坑是batch内不同请求的padding影响了生成结果。因为padding的token也会参与attention虽然被mask了但某些实现里mask不严谨就会导致结果偏移。解决办法是确保attention_mask正确传递并且尽量让batch内长度接近。5.3 服务层超时与并发问题服务层最常见的问题是超时。一个慢请求把线程池占满后面的请求全部排队。我的做法是给每个请求设超时超时直接返回错误不拖累整体。另外线程池的worker数量要按CPU核数和任务类型调IO密集可以多CPU密集要少。提示FastAPI默认的uvicorn worker是1个生产环境记得用--workers参数开多个进程不然单进程扛不住并发。5.4 日志与监控出问题时的救命稻草日志我分三级请求级、batch级、模型级。请求级记录输入输出和耗时batch级记录batch size和显存占用模型级记录生成参数。这样出问题的时候我能快速定位是哪个环节的锅。监控我用的是最简单的方案把关键指标打到文件用tail和grep看。虽然土但够用。6. 从能跑到好用后续可以这样扩展把最小链路跑通之后我做了几件事让它更接近生产可用。第一是加了缓存对相同的输入直接返回缓存结果省掉重复推理。第二是加了限流防止突发流量打垮服务。第三是把模型加载和服务启动分离用单独的进程管理模型服务重启不影响模型。再往后可以考虑的方向有多模型路由按任务类型选不同模型、动态批处理根据负载自动调batch size、模型热更新不重启服务换模型。这些每一个都是独立的工程问题但有了从零搭的基础理解起来会快很多。我个人在实际操作中的体会是AI工程最难的不是模型本身而是模型之外的那一整套工程约束。显存、延迟、并发、一致性这些才是真正拉开差距的地方。把这条链路自己走一遍比看十篇框架文档都管用。