从零手搓AI工程流水线：避开调包陷阱的实战指南-平芜编程栈

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加速明显	TensorRT（N卡专属）
服务框架	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, window=512, overlap=128): 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_batch=8, max_wait=0.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_batch=8、max_wait=50ms这组参数在大多数场景下比较均衡。延迟增加不明显，吞吐能提升3到5倍。

3.3 服务化：FastAPI的异步陷阱

用FastAPI做AI服务，最大的陷阱是在异步函数里跑同步推理。PyTorch的推理是同步阻塞的，如果你在async def里直接调用，整个事件循环会被卡住，并发能力直接归零。

正确做法有两种：一是用run_in_executor把推理丢到线程池；二是用专门的推理进程，通过队列通信。我一般用第一种，简单直接：

from concurrent.futures import ThreadPoolExecutor import asyncio executor = ThreadPoolExecutor(max_workers=4) @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(status='success').inc() return result except Exception: REQUEST_COUNT.labels(status='error').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.txt

requirements.in里只写你直接用的包，比如torch、fastapi、faiss-cpu，pip-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_tensors="pt", padding=True ) 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_version=14 )

导出后可以用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异常。我遇到过好几次都是预处理不一致导致的，模型本身没问题。