1. 为什么我要从零手搓一套AI工程流水线第一次看到ai-engineering-from-scratch这个项目名的时候我正被一堆调包式AI项目折磨得够呛。打开任何一个开源仓库清一色的pip install transformers、from langchain import ...三行代码跑通一个Demo然后呢线上流量一上来延迟飙到三秒显存直接爆掉日志里全是看不懂的报错。那一刻我意识到会用框架和懂AI工程完全是两码事。ai-engineering-from-scratch这个标题戳中的正是这个痛点——它不满足于能跑而是要求你从最底层开始把AI系统当成一个真正的工程项目来搭建。什么叫from scratch不是让你从零实现一个Transformer那是研究员的活而是让你从零搭建一套可训练、可推理、可监控、可迭代的工程骨架。这中间涉及数据管道、模型封装、服务化、性能优化、可观测性等一整套东西每一环都有坑每一环都值得单独拿出来讲。这篇文章适合谁看如果你已经能用PyTorch或TensorFlow跑通一个模型但一到部署就抓瞎如果你面试时被问到模型上线后显存怎么管理就卡壳如果你想知道一个AI工程师每天到底在跟什么东西较劲——那这篇就是写给你的。我会把整个从零搭建的过程拆成可复现的步骤配上我踩过的坑和实测有效的参数尽量让你少走弯路。2. 整体架构设计与技术选型思路2.1 为什么我不推荐一上来就用LangChain这类重型框架很多人做AI工程的第一反应是找个大框架把活全包了。我早期也这么干过用LangChain搭RAG用FastAPI做接口看起来很美。但真到了要优化的时候你发现自己根本不知道框架内部在干什么——它什么时候加载模型、什么时候释放显存、tokenizer是不是每次请求都重新初始化全是黑盒。从零搭建的第一个原则就是每一层都要可控。我的做法是把系统拆成四个独立层数据层、模型层、服务层、监控层。数据层负责原始数据的清洗、分块、向量化模型层负责模型加载、推理、批处理服务层负责HTTP接口、并发控制、限流监控层负责延迟、吞吐、显存、错误率的采集。层与层之间通过明确的接口通信任何一层出问题都能快速定位。这样做的好处是当你需要换一个embedding模型时只需要改数据层的一个配置当你需要把推理从CPU迁到GPU时只动模型层。框架帮你省的那点代码量在调试和优化阶段会加倍还回去。2.2 技术栈的取舍轻量优先但别裸奔具体选型上我的原则是轻量优先但关键环节不裸奔。下面这张表是我实际用下来比较稳的组合层级选型理由替代方案数据处理Pandas NumPy中小规模够用调试直观Polars大数据量模型推理PyTorch ONNX Runtime训练推理统一ONNX加速明显TensorRTN卡专属服务框架FastAPI Uvicorn异步支持好文档自动生成Flask简单场景向量检索FAISS轻量、快、无需额外服务Milvus大规模监控Prometheus Grafana生态成熟指标采集方便自建日志系统这里重点说两个选择。第一为什么用ONNX Runtime而不是直接PyTorch推理实测下来同一个BERT模型PyTorch推理单条延迟约45ms转成ONNX后降到28ms左右批处理场景差距更大。转换过程确实有点折腾但一次转换长期受益。第二为什么监控不自己写日志因为AI系统的指标维度太多了——QPS、P99延迟、GPU利用率、显存占用、缓存命中率自己写日志系统最后一定会变成一团乱麻Prometheus的指标模型天然适合这种场景。2.3 目录结构让项目自己说话一个清晰的目录结构能省掉大量沟通成本。我的习惯是这样组织的ai-engineering-from-scratch/ ├── configs/ # 所有配置文件按环境分 │ ├── base.yaml │ ├── dev.yaml │ └── prod.yaml ├── data/ # 数据管道 │ ├── loader.py │ ├── cleaner.py │ └── chunker.py ├── models/ # 模型相关 │ ├── loader.py │ ├── inference.py │ └── export_onnx.py ├── serving/ # 服务层 │ ├── app.py │ ├── schemas.py │ └── middleware.py ├── monitoring/ # 监控 │ ├── metrics.py │ └── dashboard.json ├── tests/ # 测试 └── scripts/ # 运维脚本这个结构的关键在于配置与代码分离。我见过太多项目把模型路径、批大小、超时时间硬编码在代码里换个环境就要改代码重新部署。用YAML管理配置配合环境变量覆盖能做到同一份代码不同环境不同行为。3. 核心模块拆解与实操要点3.1 数据管道别小看清洗和分块数据管道是AI工程里最不起眼但最容易出问题的环节。我接手过一个项目线上效果一直不稳定排查了三天才发现是数据清洗时把某些特殊字符处理错了导致tokenizer分词结果和训练时不一致。清洗阶段的核心是统一规范。我的做法是写一个cleaner.py里面定义一组可组合的清洗函数去除多余空白、统一全半角、过滤控制字符、处理编码异常。每个函数单独可测组合起来用。这里有个细节不要用正则表达式一次性处理所有规则那样出了问题根本不知道是哪条规则导致的。分开处理虽然慢一点但可调试性天差地别。分块阶段更讲究。做RAG的时候chunk大小直接决定检索质量。我试过固定长度分块、按句子分块、按语义分块三种方案。固定长度最简单但会切断语义按句子分块保留了语义但块大小不均按语义分块效果最好但需要额外模型。实测下来滑动窗口句子边界对齐是性价比最高的方案窗口大小512个token重叠128个token切分时尽量在句号或换行处断开。def chunk_text(text, window512, overlap128): tokens tokenizer.encode(text) chunks [] start 0 while start len(tokens): end min(start window, len(tokens)) chunk tokens[start:end] # 尝试在句子边界回退 chunk_text tokenizer.decode(chunk) last_period max(chunk_text.rfind(。), chunk_text.rfind(.)) if last_period window * 0.6: chunk_text chunk_text[:last_period 1] end start len(tokenizer.encode(chunk_text)) chunks.append(chunk_text) start end - overlap return chunks注意overlap不要设得太大否则检索时会返回大量重复内容反而降低效果。我一般控制在窗口的20%到25%之间。3.2 模型加载与推理显存管理的艺术模型加载这块最大的坑是重复加载。很多新手会在每个请求里AutoModel.from_pretrained(...)结果第一个请求就把显存吃满第二个请求直接OOM。正确做法是在服务启动时加载一次全局持有请求时复用。但加载一次也有讲究。如果你的服务同时提供多个模型比如一个embedding模型加一个生成模型要算好显存预算。我的一般公式是可用显存 总显存 - 系统预留(约1GB) - 其他进程占用 模型显存 ≈ 参数量 × 精度字节数 × 1.2推理额外开销比如一个1.3B参数的模型FP16精度参数量约13亿每个参数2字节基础占用约2.6GB加上推理时的中间激活值实际需要3.5GB左右。如果你只有8GB显存加载两个这样的模型就很紧张了。推理阶段的核心是批处理。单条推理GPU利用率极低批处理能显著提升吞吐。但批大小不是越大越好要找到延迟和吞吐的平衡点。我的做法是做一个动态批处理队列请求进来先入队攒够一定数量或等待超过阈值时间就一起推理。class DynamicBatcher: def __init__(self, max_batch8, max_wait0.05): self.max_batch max_batch self.max_wait max_wait self.queue [] self.last_flush time.time() def add(self, item): self.queue.append(item) if len(self.queue) self.max_batch or \ time.time() - self.last_flush self.max_wait: return self.flush() return None def flush(self): batch self.queue[:self.max_batch] self.queue self.queue[self.max_batch:] self.last_flush time.time() return batch实测下来max_batch8、max_wait50ms这组参数在大多数场景下比较均衡。延迟增加不明显吞吐能提升3到5倍。3.3 服务化FastAPI的异步陷阱用FastAPI做AI服务最大的陷阱是在异步函数里跑同步推理。PyTorch的推理是同步阻塞的如果你在async def里直接调用整个事件循环会被卡住并发能力直接归零。正确做法有两种一是用run_in_executor把推理丢到线程池二是用专门的推理进程通过队列通信。我一般用第一种简单直接from concurrent.futures import ThreadPoolExecutor import asyncio executor ThreadPoolExecutor(max_workers4) app.post(/predict) async def predict(request: PredictRequest): loop asyncio.get_event_loop() result await loop.run_in_executor( executor, model_inference, request.text ) return {result: result}线程池大小要跟GPU数量匹配。单卡的话workers设2到4比较合适太多反而会因为上下文切换降低效率。另一个容易忽略的点是请求超时和限流。AI推理延迟波动大没有超时控制的话慢请求会堆积拖垮整个服务。我在中间件里加了基于令牌桶的限流和基于asyncio.wait_for的超时控制超过阈值的请求直接返回降级结果。3.4 监控没有度量就没有优化监控这块我踩过最大的坑是只监控了系统指标没监控业务指标。CPU、内存、GPU利用率都正常但用户就是反馈结果不对。后来加了业务指标才定位到是某个模型版本更新后输出分布偏移了。我的监控体系分三层。第一层是系统指标CPU、内存、GPU利用率、显存占用用psutil和pynvml采集。第二层是服务指标QPS、P50/P95/P99延迟、错误率、超时率用Prometheus客户端埋点。第三层是业务指标输入长度分布、输出长度分布、置信度分布、缓存命中率这些需要自己定义和采集。from prometheus_client import Histogram, Counter INFERENCE_LATENCY Histogram( inference_latency_seconds, Inference latency, buckets[0.01, 0.05, 0.1, 0.5, 1.0, 2.0] ) REQUEST_COUNT Counter( inference_requests_total, Total requests, [status] ) INFERENCE_LATENCY.time() def model_inference(text): try: result model(text) REQUEST_COUNT.labels(statussuccess).inc() return result except Exception: REQUEST_COUNT.labels(statuserror).inc() raise提示Histogram的buckets要根据你的实际延迟分布来设。设得太粗看不出问题设得太细指标基数爆炸。我一般先用默认buckets跑一周看P99落在哪个区间再针对性调整。4. 完整实操流程从零到可服务4.1 环境准备与依赖锁定第一步永远是环境。我强烈建议用conda或venv建独立环境然后用pip-tools锁定依赖版本。AI项目的依赖冲突是家常便饭torch、transformers、onnxruntime之间的版本兼容性尤其要小心。python -m venv venv source venv/bin/activate pip install pip-tools # 在requirements.in里写顶层依赖 pip-compile requirements.in -o requirements.txt pip install -r requirements.txtrequirements.in里只写你直接用的包比如torch、fastapi、faiss-cpupip-compile会自动解析出所有间接依赖并锁定版本。这样换机器部署时pip install -r requirements.txt能保证环境完全一致。4.2 模型导出与优化如果你用PyTorch训练部署前建议导出ONNX。导出脚本大概长这样import torch from transformers import AutoModel, AutoTokenizer model AutoModel.from_pretrained(your-model) tokenizer AutoTokenizer.from_pretrained(your-model) model.eval() dummy_input tokenizer( test input, return_tensorspt, paddingTrue ) torch.onnx.export( model, (dummy_input[input_ids], dummy_input[attention_mask]), model.onnx, input_names[input_ids, attention_mask], output_names[last_hidden_state], dynamic_axes{ input_ids: {0: batch, 1: sequence}, attention_mask: {0: batch, 1: sequence}, last_hidden_state: {0: batch, 1: sequence} }, opset_version14 )导出后可以用onnxruntime加载开启图优化import onnxruntime as ort options ort.SessionOptions() options.graph_optimization_level ort.GraphOptimizationLevel.ORT_ENABLE_ALL options.intra_op_num_threads 4 session ort.InferenceSession(model.onnx, options)实测下来开启图优化后推理速度能再提升10%到15%。intra_op_num_threads设成CPU物理核心数比较合适设太大反而会因为线程竞争变慢。4.3 服务启动与压测服务写好后别急着上线先压测。我用locust做压力测试模拟不同并发下的表现from locust import HttpUser, task, between class AIUser(HttpUser): wait_time between(0.1, 0.5) task def predict(self): self.client.post(/predict, json{ text: 这是一条测试文本 })压测时重点关注三个指标P99延迟是否在可接受范围、错误率是否为零、GPU显存是否稳定。如果P99延迟随并发线性增长说明有资源瓶颈如果显存持续增长不释放说明有内存泄漏。我一般会跑三组压测低并发10用户看基线延迟中并发50用户看吞吐拐点高并发100用户看系统极限。找到拐点后把线上限流阈值设在拐点的70%左右留出缓冲。4.4 灰度发布与回滚AI服务上线最怕的是新模型效果变差。我的做法是灰度发布新版本先接10%流量观察24小时业务指标没问题再逐步放大。回滚机制要提前准备好一旦错误率超过阈值或业务指标异常自动切回旧版本。class ModelRouter: def __init__(self): self.models { v1: load_model(v1), v2: load_model(v2) } self.traffic_split {v1: 0.9, v2: 0.1} def route(self, request): rand random.random() if rand self.traffic_split[v2]: return self.models[v2](request) return self.models[v1](request)灰度期间要密切监控两个版本的输出差异。如果差异过大可能是模型更新引入了非预期变化需要人工介入判断。5. 常见问题与排查技巧实录5.1 显存溢出OOM的排查路径OOM是AI工程最高频的问题。我的排查顺序是这样的先看是不是批大小设太大了把batch调成1试试如果单条也OOM看模型加载时是不是没设torch.no_grad()推理时不需要计算梯度加上这个能省不少显存如果还不行检查是不是有多个模型实例被重复加载了用torch.cuda.memory_summary()看显存分配详情。还有一个隐蔽的坑缓存没清理。如果你用了KV Cache或者特征缓存长时间运行后缓存会越积越大。我一般会设一个缓存上限超过就清理最久未使用的部分。5.2 推理结果不一致的排查同一个输入两次推理结果不一样通常有三个原因。一是模型没设eval()模式Dropout还在起作用二是随机种子没固定某些操作有随机性三是浮点运算在不同硬件或不同批大小下结果有微小差异这是正常的但如果差异大到影响业务就要检查是不是有未初始化的变量。注意批处理时同一条数据在不同batch里结果可能有细微差异这是浮点累加顺序不同导致的。如果业务对一致性要求极高要么关掉批处理要么在输出后做量化对齐。5.3 延迟波动的常见原因延迟忽高忽低先看是不是GC垃圾回收导致的。Python的GC在对象多的时候会暂停整个进程我一般会调优GC阈值或者用gc.freeze()冻结长期存活的对象。其次看是不是有其他进程在抢GPU用nvidia-smi看有没有异常进程。最后看是不是请求分布不均长文本请求集中出现会拉高P99这种情况可以在入口做长度分流长文本走单独的队列。下面这张表是我整理的高频问题速查现象可能原因排查方法解决方向OOM批太大/重复加载/缓存泄漏memory_summary调小batch/单例加载/设缓存上限结果不一致未eval/随机种子/浮点差异固定种子重跑设eval/固定种子/关批处理延迟高GC/资源竞争/长请求看GC日志/nvidia-smi/长度分布调GC/隔离进程/长度分流吞吐低单条推理/同步阻塞看GPU利用率批处理/异步化服务卡死事件循环阻塞/死锁看线程栈线程池/超时控制5.4 模型更新后的效果回退模型更新后效果变差先别急着回滚按这个顺序查第一新模型的预处理逻辑和旧模型是否一致tokenizer版本、特殊token处理、截断策略都要对齐第二新模型的输出后处理是否一致比如softmax温度、阈值、映射表第三用同一批测试数据对比新旧模型的输出分布看是整体偏移还是个别case异常。我遇到过好几次都是预处理不一致导致的模型本身没问题。6. 一些让我少走弯路的实操心得做AI工程这几年最大的体会是工程问题往往比算法问题更致命。一个模型效果再好如果服务不稳定、延迟高、显存管理混乱上线就是灾难。反过来一个中等效果的模型配上扎实的工程底座能稳定服务、快速迭代长期价值反而更大。第二个体会是度量先行。没有监控的优化都是盲人摸象。我现在的习惯是任何新功能上线前先把监控指标定义好跑一周基线数据然后再做优化。这样每次改动都能量化收益而不是凭感觉说好像快了点。第三个是保持简单。我见过太多项目为了架构先进引入一堆中间件结果运维成本远超收益。从零搭建的意义不是堆技术而是让你清楚每一层的边界和代价然后在需要的时候做加法。一开始用最简单的方案跑通遇到瓶颈再针对性优化这个节奏比一上来就上重型框架要稳得多。最后分享一个我常用的调试技巧在推理函数入口和出口打上时间戳把输入长度、输出长度、耗时、显存占用一起写进结构化日志。出问题时用Pandas读日志做聚合分析比看散落的print高效得多。这个习惯帮我定位过好几次偶发的性能抖动强烈推荐你试试。