1. 从零搭建AI工程体系为什么我劝你别急着调包ai-engineering-from-scratch这个标题第一次看到的时候我愣了一下。市面上讲AI的文章十篇有八篇在教你pip install之后怎么调API剩下两篇在讲Transformer的数学推导。但真正从零把一套AI工程体系搭起来的内容少得可怜。我自己带过几个从零起步的AI项目也踩过不少坑。最开始我也觉得现在开源生态这么成熟pip install transformers、pip install langchain拼拼凑凑不就能跑起来了吗后来发现能跑起来和能上线之间隔着一整个工程体系的距离。模型加载慢、显存炸了、推理延迟高、服务不稳定、日志查不到问题、版本对不上——这些问题调包是解决不了的。所以这篇内容我想聊的是如果你要从零开始搭建一套AI工程体系应该怎么规划、怎么选型、怎么落地。适合谁看适合那些已经会写Python、懂一点机器学习基础但还没真正把AI系统跑在生产环境里的人。也适合那些调了很久的包但总觉得心里没底、想搞清楚底层到底在发生什么的人。核心关键词就一个ai-engineering-from-scratch。我会围绕这个关键词把从零搭建AI工程体系这件事拆开揉碎讲清楚每个环节为什么这么做、怎么做、做完之后怎么验证。2. 整体架构设计先想清楚你要解决什么问题2.1 从需求反推架构而不是从技术出发很多人一上来就问我该用PyTorch还是TensorFlow要不要上Kubernetes这些问题本身没错但顺序反了。你应该先问自己我要解决什么问题是离线批量推理还是在线实时服务是单模型单任务还是多模型编排是内部工具还是对外产品我见过一个团队上来就搭了一套Kubernetes集群结果他们的需求只是每天跑一次批量推理跑完就完事。Kubernetes的运维成本远高于他们省下来的那点资源。这就是典型的从技术出发而不是从需求出发。从零搭建AI工程体系我的建议是按这个顺序思考明确任务类型分类、生成、检索、排序还是多任务组合明确服务形态离线批处理、在线API、流式处理还是边缘部署明确规模预期QPS多少数据量多大模型多大明确团队能力有几个人维护有没有专职运维最后才是技术选型框架、推理引擎、服务框架、存储方案。这个顺序不能乱。乱了后面全是返工。2.2 分层设计把系统切成可独立演进的模块一套完整的AI工程体系我习惯把它切成五层层级职责典型组件数据层数据采集、清洗、存储、版本管理对象存储、数据湖、特征库训练层模型训练、调参、实验管理训练框架、实验追踪、超参搜索模型层模型转换、压缩、版本管理模型仓库、量化工具、格式转换服务层推理服务、批处理、流处理推理引擎、服务框架、消息队列应用层API网关、业务逻辑、监控告警网关、日志、指标、追踪这么切的好处是每一层可以独立演进。比如你最开始用Flask写了个简单的推理接口后来QPS上来了只需要把服务层换成Triton或vLLM其他层不用动。如果你一开始就把所有逻辑揉在一起换一个组件就是牵一发动全身。注意分层不是目的可替换才是。每一层之间的接口要定义清楚输入输出格式要稳定。这样你换实现的时候上下游不用改。2.3 从零开始的MVP应该长什么样如果你真的是从零开始我建议第一个版本越简单越好。不要一上来就搞微服务、搞消息队列、搞分布式训练。先跑通一个最小闭环一个脚本能读数据一个脚本能训练模型一个脚本能加载模型做推理一个HTTP接口能接收请求返回结果这四个东西跑通了你才算有了一个AI工程的雏形。然后再逐步替换里面的组件脚本换成流水线Flask换成专业推理服务本地文件换成对象存储。我自己的习惯是第一个版本用最土的办法实现但把接口定义好。比如推理接口的输入输出用JSON Schema定死后面换实现的时候只要Schema不变调用方就不用改。3. 核心细节解析数据、训练、推理三件事3.1 数据管道别小看清洗和版本管理数据这块很多人觉得没什么技术含量不就是读文件吗但实际项目里数据问题占了我调试时间的一半以上。数据清洗要做的几件事去重、去噪、格式统一、异常值处理。听起来简单但每一条都有坑。比如去重文本去重和图像去重策略完全不同。文本可以用SimHash或MinHash图像可以用感知哈希。如果你不做去重训练集里大量重复样本会让模型过拟合到这些样本上。数据版本管理是另一个容易被忽略的点。你今天用了一份数据训练明天数据更新了模型效果变了你根本不知道是模型改了还是数据改了。我的做法是每次训练用的数据都打一个版本号记录数据的来源、清洗规则、样本数量。可以用DVC这样的工具也可以简单点用文件名的哈希值。import hashlib import json def dataset_fingerprint(file_paths): 计算数据集指纹用于版本追踪 hasher hashlib.sha256() for path in sorted(file_paths): with open(path, rb) as f: while chunk : f.read(8192): hasher.update(chunk) return hasher.hexdigest()[:16] # 使用示例 fingerprint dataset_fingerprint([data/train.jsonl, data/val.jsonl]) print(fDataset version: {fingerprint})这个指纹可以写进模型元数据里后面排查问题的时候一看就知道这个模型是用哪份数据训练的。实操心得数据清洗的规则一定要写成代码不要手动改数据。手动改的数据没法复现出了问题查都查不到。3.2 训练流程实验追踪比调参更重要训练这块很多人把精力全花在调参上但我觉得实验追踪才是从零搭建AI工程体系时最该先做的事。为什么因为调参是个试错过程你试了十组参数最后发现第三组最好。如果没有实验追踪你根本记不住第三组用的是什么配置、什么数据、什么代码版本。我见过太多人用Excel记实验结果记着记着就乱了。实验追踪要记录的东西超参数配置数据集版本代码版本Git commit训练指标曲线最终模型文件路径环境依赖版本工具上MLflow、Weights Biases、TensorBoard都可以。如果不想引入外部依赖自己写个简单的JSON日志也行。关键是每次训练都记录不要偷懒。import json import time from pathlib import Path def log_experiment(config, metrics, model_path): 简单的实验日志记录 experiment { timestamp: time.strftime(%Y-%m-%d %H:%M:%S), config: config, metrics: metrics, model_path: str(model_path), git_commit: get_git_commit(), # 自行实现 } log_dir Path(experiments) log_dir.mkdir(exist_okTrue) log_file log_dir / fexp_{int(time.time())}.json log_file.write_text(json.dumps(experiment, indent2)) return log_file这个简单的记录后面会帮你省下大量时间。当你想复现某个结果的时候直接看日志就行。3.3 推理服务延迟和吞吐的平衡艺术推理服务是从零搭建AI工程体系时最考验工程能力的一环。核心矛盾就一个延迟和吞吐的平衡。延迟是单个请求的响应时间吞吐是单位时间能处理的请求数。这两个指标往往是矛盾的。你增大批处理大小batch size吞吐上去了但单个请求的延迟也上去了因为它要等凑够一批才处理。怎么平衡取决于你的场景在线实时服务延迟优先batch size设小一点甚至设为1离线批处理吞吐优先batch size尽量大把显存吃满流式服务折中用动态批处理dynamic batching攒一小段时间就发车动态批处理是现在推理引擎的标配。Triton、vLLM、TensorRT-LLM都支持。原理很简单请求来了不立即处理等一小段时间比如10毫秒把这段时间内的请求攒成一批一起处理。这样既不会等太久又能提高吞吐。# 动态批处理的简化逻辑示意 import asyncio from collections import deque class DynamicBatcher: def __init__(self, max_batch_size8, max_wait_ms10): self.max_batch_size max_batch_size self.max_wait_ms max_wait_ms self.queue deque() async def add_request(self, request): self.queue.append(request) if len(self.queue) self.max_batch_size: return await self._process_batch() await asyncio.sleep(self.max_wait_ms / 1000) if self.queue: return await self._process_batch() async def _process_batch(self): batch list(self.queue) self.queue.clear() # 实际推理逻辑 return await self._infer(batch)实际生产里你不会自己写这个但理解这个逻辑能帮你更好地配置推理引擎的参数。注意batch size不是越大越好。显存是有限的batch size太大直接OOM。而且有些模型对batch size敏感太大了效果会下降。一定要实测。4. 实操过程从零搭一个可用的推理服务4.1 环境准备与依赖管理从零开始第一步是把环境搞干净。我强烈建议用虚拟环境不要用系统Python。conda、venv、uv都行选一个顺手的。# 用venv创建虚拟环境 python -m venv ai-env source ai-env/bin/activate # Linux/Mac # ai-env\Scripts\activate # Windows # 安装核心依赖 pip install torch transformers fastapi uvicorn依赖管理有个坑版本锁定。你今天装的是transformers 4.35明天自动升级到4.36可能API就变了。所以一定要用requirements.txt或pyproject.toml锁定版本。pip freeze requirements.txt这个文件要提交到Git后面部署的时候用pip install -r requirements.txt保证环境一致。4.2 模型加载与推理封装模型加载这块有几个参数直接影响性能和显存torch_dtype用float16或bfloat16能省一半显存效果几乎无损device_map多卡的时候用auto自动分配low_cpu_mem_usage加载大模型时省内存import torch from transformers import AutoModelForCausalLM, AutoTokenizer class ModelWrapper: def __init__(self, model_name, devicecuda): self.tokenizer AutoTokenizer.from_pretrained(model_name) self.model AutoModelForCausalLM.from_pretrained( model_name, torch_dtypetorch.float16, device_mapauto, low_cpu_mem_usageTrue, ) self.model.eval() torch.inference_mode() def generate(self, prompt, max_new_tokens128): inputs self.tokenizer(prompt, return_tensorspt).to(self.model.device) outputs self.model.generate( **inputs, max_new_tokensmax_new_tokens, do_sampleFalse, ) return self.tokenizer.decode(outputs[0], skip_special_tokensTrue)torch.inference_mode()比torch.no_grad()更省内存推理场景优先用它。4.3 FastAPI服务封装与压测把推理逻辑包成HTTP服务FastAPI是最顺手的选择。它的异步支持好自动生成文档性能也够用。from fastapi import FastAPI from pydantic import BaseModel app FastAPI() model ModelWrapper(your-model-name) class GenerateRequest(BaseModel): prompt: str max_new_tokens: int 128 class GenerateResponse(BaseModel): text: str app.post(/generate, response_modelGenerateResponse) async def generate(req: GenerateRequest): text model.generate(req.prompt, req.max_new_tokens) return GenerateResponse(texttext)启动服务uvicorn main:app --host 0.0.0.0 --port 8000 --workers 1注意--workers参数。因为模型加载很占显存多个worker会重复加载模型显存直接爆。所以推理服务通常是一个worker靠异步和批处理提高并发。压测用wrk或locustwrk -t4 -c100 -d30s http://localhost:8000/generate看几个指标QPS、P99延迟、错误率。如果P99延迟太高考虑减小batch size或优化模型。4.4 容器化与部署容器化是为了环境一致。Dockerfile写起来简单但有几个细节要注意FROM nvidia/cuda:12.1-runtime-ubuntu22.04 RUN apt-get update apt-get install -y python3 python3-pip WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]基础镜像选runtime而不是devel体积小很多。CUDA版本要和宿主机驱动兼容这个查NVIDIA的兼容性表。踩过的坑Docker默认的共享内存只有64MB模型加载的时候可能不够。启动容器时加--shm-size2g。5. 常见问题与排查技巧实录5.1 显存不够用怎么办这是最高频的问题。排查顺序看模型大小7B模型用float16大概14GB加上KV Cache和中间激活实际要20GB左右看batch sizebatch size翻倍显存大概增加50%到80%看是否有内存泄漏长时间运行显存持续增长多半是缓存没清理解决方案按优先级用float16或int8量化减小batch size用梯度检查点训练时用模型并行多卡用CPU offload慢但能跑5.2 推理延迟忽高忽低延迟抖动大通常是这几个原因现象可能原因排查方法周期性抖动垃圾回收看GC日志调GC参数随机抖动批处理等待调小max_wait_ms持续高延迟显存不足换页看GPU显存使用率首请求慢模型冷启动预热启动时跑几次推理预热很重要。服务启动后先跑几次推理让CUDA kernel编译好、显存分配好后面就稳定了。5.3 模型效果和训练时不一致这个问题很隐蔽。训练时评估指标很好上线后效果差。常见原因预处理不一致训练时用的分词器版本和推理时不一样后处理不一致训练时解码策略和推理时不一样数据分布不一致线上数据分布和训练数据不同评估指标不一致训练时用teacher forcing推理时用自回归排查方法拿一批训练数据走一遍推理流程对比结果。如果结果不一致就是预处理或后处理的问题。5.4 服务上线后内存持续增长内存泄漏在Python里不常见但在AI服务里不少见。常见原因全局变量缓存了请求数据日志对象没释放CUDA缓存没清理异步任务没取消排查用tracemalloc或memory_profiler。定位到泄漏点后该清理的清理该加锁的加锁。import tracemalloc tracemalloc.start() # ... 跑一段时间 ... snapshot tracemalloc.take_snapshot() top_stats snapshot.statistics(lineno) for stat in top_stats[:10]: print(stat)这个能直接告诉你哪一行代码分配的内存最多。5.5 常见问题速查表问题快速排查常用解决CUDA OOMnvidia-smi看显存减batch、量化、多卡延迟高看P99、看GPU利用率动态批处理、模型优化效果差对比训练推理流程统一预处理、后处理服务崩溃看日志、看OOM加内存限制、重启策略版本冲突pip check锁版本、虚拟环境6. 工程化进阶从能跑到好用6.1 监控与告警别等用户投诉才知道挂了服务上线只是开始能持续稳定运行才是目标。监控要覆盖三个层面系统层CPU、内存、GPU、磁盘、网络服务层QPS、延迟、错误率、并发数业务层推理成功率、结果质量、用户反馈工具上Prometheus Grafana是标配。FastAPI可以很方便地暴露metricsfrom prometheus_fastapi_instrumentator import Instrumentator app FastAPI() Instrumentator().instrument(app).expose(app)这样/metrics端点就有了一堆现成的指标。Grafana配个面板延迟、QPS、错误率一目了然。告警规则要设得合理。延迟告警不要设太低否则天天误报最后没人看。我的经验是P99延迟超过正常值2倍持续5分钟才告警。6.2 模型版本管理与灰度发布模型更新不能直接替换要灰度。做法是同时加载新旧两个模型按比例分流。新模型先接1%流量观察指标没问题再逐步放大。import random class ModelRouter: def __init__(self, old_model, new_model, new_ratio0.01): self.old_model old_model self.new_model new_model self.new_ratio new_ratio def predict(self, request): if random.random() self.new_ratio: return self.new_model.predict(request) return self.old_model.predict(request)灰度期间要重点看新模型的延迟、错误率、业务指标。任何一个异常立即回滚。6.3 成本优化省下来的都是利润AI服务的成本主要在GPU。优化方向模型量化int8量化能省一半显存效果损失通常可接受模型蒸馏用大模型教小模型小模型推理快很多请求合并相似请求合并处理缓存相同输入直接返回缓存结果弹性伸缩低峰期缩容高峰期扩容缓存这块对于问答类场景特别有效。相同问题直接返回缓存省一次推理。缓存用Redis设个合理的过期时间。实操心得成本优化要先测量再优化。用profiler看时间花在哪用nvidia-smi看显存花在哪。拍脑袋优化往往适得其反。7. 我踩过的那些坑和最后的小建议从零搭建AI工程体系这件事我最大的体会是工程能力比算法能力更稀缺。算法决定效果上限工程决定能不能落地。很多项目不是模型不行是工程没做好跑不起来、跑不稳、跑不快。几个具体的建议第一先把监控搭起来再上线。没有监控的服务就是裸奔出了问题你连哪里出问题都不知道。第二接口定义要稳定。内部实现随便换但对外接口的输入输出格式一旦定了就不要轻易改。改了就要版本化。第三日志要打全。请求ID、模型版本、输入输出摘要、耗时这些都要记。排查问题的时候日志就是你的眼睛。第四不要过度设计。从零开始的时候简单能跑比架构优雅重要。先跑通再优化。第五测试要覆盖边界。空输入、超长输入、特殊字符、并发请求这些都要测。生产环境的输入永远比你想象的离谱。最后分享一个小技巧如果你不确定某个组件该不该引入先问自己没有它我会死吗。如果不会就先不加。技术栈越简单维护成本越低。等真的遇到瓶颈了再引入对应的组件。这样你的系统是长出来的不是堆出来的。