1. 从零搭建AI工程能力为什么“会调包”远远不够很多人对AI工程的理解停留在“会调API、会跑通一个demo”这个层面。我刚开始接触这块的时候也是这么想的——装个环境pip install几个库把模型加载进来输入一段文字输出一段结果完事。但真正到了要交付一个能用的系统时问题就全冒出来了模型加载慢、显存不够、推理延迟高、并发一上来就崩、换个模型整个代码要重写、日志看不懂、出了问题不知道从哪查。这就是“ai-engineering-from-scratch”这个项目标题背后真正要解决的问题。它不是教你从零训练一个大模型——那是研究岗的事——而是教你从零构建一套AI工程化的能力体系。换句话说模型本身可以是现成的但围绕模型的那一整套工程基础设施你得自己能搭起来、能调优、能维护。我见过太多人模型原理讲得头头是道但让他把一个模型部署到生产环境他就卡住了。反过来也见过一些后端工程师工程能力很强但对AI模型的特点不了解做出来的系统各种水土不服。AI工程这个方向恰恰是这两者的交叉地带也是目前市场上最缺人的地方。这篇文章适合三类人看第一类是有一定编程基础、想转行做AI工程的后端或全栈开发者第二类是刚入行做算法、但发现工程能力拖后腿的算法工程师第三类是对AI系统架构感兴趣、想了解一个完整AI应用是怎么从零搭起来的技术管理者。不管你属于哪一类我都会尽量把每个环节的“为什么”讲清楚而不是只丢一堆代码让你抄。接下来的内容我会按照一个AI系统从无到有的实际搭建顺序来展开先聊环境与依赖管理这个最容易被忽视但最要命的部分再讲模型加载与推理优化的核心逻辑然后是服务化与并发处理最后是监控、日志和迭代。每一块我都会给出具体的操作步骤、参数解释以及我在实际项目中踩过的坑。2. 环境与依赖管理别让“在我机器上能跑”成为口头禅2.1 为什么虚拟环境不是可选项而是必选项我见过最离谱的一个项目团队里五个人每个人电脑上的Python版本、CUDA版本、PyTorch版本都不一样代码提交上去今天他能跑明天她跑不了。最后花了两周时间统一环境比写代码的时间还长。这个教训告诉我AI工程的第一步不是写模型代码而是把环境管住。Python的虚拟环境方案有好几种venv、conda、pipenv、poetry。我的建议是如果你涉及深度学习框架直接用conda。原因很简单conda不仅能管Python包还能管CUDA、cuDNN这些非Python的底层库。你用venv装PyTorchCUDA还得自己手动装版本对不上就是一堆报错。conda一条命令就能把Python版本、CUDA版本、PyTorch版本全部锁定。具体操作上我习惯为每个项目建一个独立的conda环境conda create -n ai-eng python3.10 conda activate ai-engPython版本选3.10是有讲究的。3.8太老很多新库不支持3.12太新部分深度学习框架的wheel包还没跟上。3.10是目前兼容性最好的版本主流框架都有对应的预编译包。环境建好之后依赖安装不要一条一条pip install而是写一个requirements.txt或者environment.yml。conda的方式更彻底name: ai-eng channels: - pytorch - nvidia - defaults dependencies: - python3.10 - pytorch2.1.0 - torchvision0.16.0 - cudatoolkit11.8 - pip - pip: - transformers4.36.0 - fastapi0.104.0 - uvicorn0.24.0这个文件的好处是任何人拿到你的项目一条conda env create -f environment.yml就能复现完全一样的环境。版本号一定要写死不要用或者不写版本否则今天装的和明天装的可能就不一样。注意CUDA版本和显卡驱动版本是两回事。驱动版本决定了你最高能用哪个CUDA版本而conda里装的cudatoolkit是运行时库。如果你驱动版本太低装了高版本CUDA也跑不起来。用nvidia-smi看驱动支持的CUDA版本然后选一个不超过它的cudatoolkit版本。2.2 依赖冲突的排查思路与实战案例依赖冲突是AI工程里最常见的坑之一。典型症状是装了一个新库原来能跑的代码突然报错了。原因通常是两个库依赖了同一个底层库的不同版本。我遇到过一个经典案例项目里同时用了transformers和tokenizers某次升级transformers之后tokenizers的版本没跟着升结果加载分词器的时候直接段错误。这种错误最难查因为它不给你Python层面的报错直接进程崩溃。排查这类问题的思路是这样的先用pip list把所有包的版本列出来然后看报错信息里提到的库去查它的依赖树。更高效的方式是用pipdeptree这个工具pip install pipdeptree pipdeptree --warn silence它会以树形结构展示所有包的依赖关系冲突的地方会标红。看到冲突之后解决办法通常有两个要么升级所有相关包到兼容版本要么把冲突的包降级到大家都兼容的版本。我的经验是优先选择升级因为新版本通常修复了旧版本的bug但如果升级会导致其他核心库不兼容那就降级。还有一个更隐蔽的坑系统里同时装了conda的包和pip的包两者版本不一致。比如conda装了numpy 1.24pip又装了一个numpy 1.26Python导入的时候到底用哪个取决于sys.path的顺序非常混乱。我的原则是能用conda装的就用conda装conda没有的再用pip装并且装完之后用conda list和pip list分别检查一遍确保没有重复安装。2.3 容器化从“能跑”到“到处都能跑”虚拟环境解决了同一台机器上的依赖隔离问题但解决不了跨机器的环境差异。你本地是Ubuntu 22.04服务器是CentOS 7很多底层库的版本完全不一样。这时候就需要容器化。Docker在AI工程里的价值不仅仅是“打包”更重要的是环境一致性。你可以在本地用Docker跑通然后直接把镜像推到服务器上行为完全一致。写Dockerfile的时候AI项目有几个特殊注意点。第一基础镜像不要用python:3.10这种精简版因为深度学习框架需要很多系统库精简版里没有你得一个个装反而麻烦。推荐用nvidia/cuda:11.8-runtime-ubuntu22.04作为基础镜像CUDA运行时已经内置了。第二pip安装的时候加上--no-cache-dir否则镜像会多出几百MB的缓存。第三把requirements.txt的复制和安装放在复制代码之前这样只要依赖没变Docker构建时就能命中缓存不用每次重新装依赖。FROM nvidia/cuda:11.8-runtime-ubuntu22.04 RUN apt-get update apt-get install -y python3.10 python3-pip rm -rf /var/lib/apt/lists/* WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [python, serve.py]这个Dockerfile看起来简单但每一条都有讲究。rm -rf /var/lib/apt/lists/*是为了减小镜像体积COPY requirements.txt单独一行是为了利用Docker的层缓存机制。这些细节在普通开发里可能无所谓但在AI项目里镜像动辄几个GB每一点优化都能省下大量时间和存储。3. 模型加载与推理优化把“能推理”变成“推理得快”3.1 模型加载的几种方式与选择逻辑模型加载看起来简单不就是from_pretrained吗但实际项目里加载方式的选择直接影响启动速度和内存占用。最基础的方式是每次请求都加载模型这显然不可取加载一个BERT模型就要好几秒。常规做法是服务启动时加载一次常驻内存。但这里有个问题如果你的服务需要同时支持多个模型全部常驻内存可能装不下。这时候有几种策略。第一种是懒加载服务启动时不加载模型第一次请求某个模型时才加载加载后缓存起来。适合模型多但每个模型访问频率不高的场景。第二种是预加载LRU淘汰启动时加载最常用的几个模型内存不够时按最近最少使用原则淘汰。适合模型多且访问有明显热点的场景。代码层面我习惯封装一个ModelManager类来统一管理import torch from transformers import AutoModelForSequenceClassification, AutoTokenizer from collections import OrderedDict class ModelManager: def __init__(self, max_models3): self.max_models max_models self.models OrderedDict() self.device torch.device(cuda if torch.cuda.is_available() else cpu) def get_model(self, model_name): if model_name in self.models: self.models.move_to_end(model_name) return self.models[model_name] if len(self.models) self.max_models: self.models.popitem(lastFalse) tokenizer AutoTokenizer.from_pretrained(model_name) model AutoModelForSequenceClassification.from_pretrained(model_name) model.to(self.device) model.eval() self.models[model_name] (model, tokenizer) return model, tokenizer这个类用OrderedDict实现了LRU缓存move_to_end把最近使用的模型移到末尾popitem(lastFalse)淘汰最久未使用的模型。model.eval()这行很重要它会把dropout等训练时才用的层关掉不仅保证推理结果一致还能略微提升速度。3.2 推理加速的实用手段与效果对比模型加载好了接下来就是让它跑得快。推理加速的手段很多我按投入产出比从高到低排个序。第一用GPU而不是CPU。这是最明显的加速通常有10倍以上的提升。但要注意小模型在GPU上可能因为数据传输开销反而更慢。判断标准是如果单次推理时间在10ms以下GPU的优势就不明显了。第二半精度推理。把模型从float32转成float16显存占用减半速度提升30%到50%精度损失通常可以忽略。一行代码的事model.half()但要注意不是所有模型都支持半精度有些操作在CPU上不支持float16。所以这行代码要配合GPU使用。第三批处理。如果有多条请求攒成一批一起推理比一条一条跑快得多。因为GPU的并行能力很强batch size从1增加到8时间可能只增加50%但吞吐量提升了8倍。实现上可以用一个队列攒够一定数量或者等待一定时间就触发一次推理。第四ONNX Runtime或TensorRT。这两个是把模型转换成中间表示然后针对特定硬件做图优化。ONNX Runtime的加速比大概在1.5到2倍TensorRT能到3到5倍。但代价是转换过程可能遇到不支持的算子需要自己写插件。我的建议是先用前面三种手段如果还不够再考虑这两个。加速手段实现难度典型加速比适用场景GPU推理低10倍以上所有场景半精度低1.3-1.5倍GPU环境批处理中随batch size线性提升高并发ONNX Runtime中1.5-2倍跨平台部署TensorRT高3-5倍NVIDIA GPU专用3.3 显存不够用时的排查与解决路径显存不够是AI工程里最让人头疼的问题之一。报错信息通常是CUDA out of memory但原因可能有很多种。第一步先用nvidia-smi看当前显存占用。如果发现显存被其他进程占着先清理掉。如果是自己的进程占着但没释放可能是代码里有循环引用导致显存没回收。第二步算一下模型本身需要多少显存。一个粗略的公式是参数量乘以4字节float32或2字节float16。比如一个1亿参数的模型float32需要约400MBfloat16需要约200MB。但这只是模型权重的部分实际运行时还需要额外的显存来存中间激活值通常是权重的1到2倍。第三步如果显存确实不够有几个解决方向。减小batch size是最直接的但会影响吞吐量。用梯度检查点gradient checkpointing可以大幅减少激活值占用的显存代价是计算时间增加。用CPU offload把部分层放到内存里需要时再加载到显存速度会慢但能跑起来。我个人的经验是优先考虑换更小的模型或者用量化。量化是把float32的权重转成int8显存直接降到四分之一速度还能提升。PyTorch自带的量化工具就够用了model torch.quantization.quantize_dynamic( model, {torch.nn.Linear}, dtypetorch.qint8 )这行动态量化对Linear层生效几乎不需要改代码精度损失通常在1%以内。对于很多业务场景来说这点精度损失完全可以接受。4. 服务化与并发处理让模型真正对外提供服务4.1 从脚本到服务FastAPI的选型与基本骨架模型在脚本里跑通和把它变成一个能对外服务的API中间隔着一条鸿沟。脚本是一次性的服务是持续运行的脚本只考虑功能服务还要考虑并发、错误处理、超时、限流。Python生态里做API服务主流选择是FastAPI和Flask。我推荐FastAPI原因有三个第一它原生支持异步对于IO密集型的场景比如等待模型推理结果性能更好第二它自带数据校验用Pydantic定义请求和响应格式省去大量手写校验的代码第三它自动生成API文档调试和对接都很方便。一个最基本的AI服务骨架长这样from fastapi import FastAPI, HTTPException from pydantic import BaseModel from model_manager import ModelManager app FastAPI() manager ModelManager() class PredictRequest(BaseModel): text: str model_name: str default-model class PredictResponse(BaseModel): label: str score: float app.on_event(startup) async def startup(): manager.get_model(default-model) app.post(/predict, response_modelPredictResponse) async def predict(req: PredictRequest): try: model, tokenizer manager.get_model(req.model_name) inputs tokenizer(req.text, return_tensorspt, truncationTrue, max_length512) inputs {k: v.to(manager.device) for k, v in inputs.items()} with torch.no_grad(): outputs model(**inputs) probs torch.softmax(outputs.logits, dim-1) score, idx torch.max(probs, dim-1) return PredictResponse(labelstr(idx.item()), scorescore.item()) except Exception as e: raise HTTPException(status_code500, detailstr(e))这里有几个关键点。app.on_event(startup)在服务启动时预加载模型避免第一次请求时等待。torch.no_grad()关闭梯度计算减少显存占用和计算量。truncationTrue和max_length512防止输入过长导致显存爆炸。这些细节看起来小但少了任何一个生产环境都可能出问题。4.2 并发场景下的模型安全与性能瓶颈FastAPI默认是单进程的虽然支持异步但模型推理是CPU/GPU密集型操作异步并不能让它变快。真正的并发需要多进程或者多线程。多进程方案是用gunicorn启动多个worker每个worker是一个独立的进程各自加载一份模型。这样能充分利用多核CPU但显存占用会翻倍。如果显存不够可以用多线程方案但要注意PyTorch的线程安全问题。PyTorch模型在推理时是线程安全的前提是你不修改模型状态。model.eval()之后前向传播不会改变模型参数所以多个线程同时调用同一个模型是安全的。但tokenizer不一定线程安全有些tokenizer内部有缓存机制多线程访问可能出问题。保险的做法是每个线程用自己的tokenizer或者加锁。import threading lock threading.Lock() app.post(/predict) async def predict(req: PredictRequest): with lock: result do_inference(req) return result加锁能保证安全但会串行化请求降低并发能力。更好的方式是每个worker进程独立处理请求用gunicorn的--workers参数控制进程数gunicorn -w 4 -k uvicorn.workers.UvicornWorker serve:app --timeout 120-w 4表示4个worker进程-k uvicorn.workers.UvicornWorker指定用uvicorn的worker类来支持异步。--timeout 120把超时设成120秒因为模型推理可能比较慢默认的30秒不够。4.3 请求队列与限流防止服务被压垮即使有多进程如果请求量突然暴增服务还是会被压垮。这时候需要队列和限流。队列的思路是请求进来先放到队列里后台有固定数量的worker从队列里取任务执行。这样不管来多少请求同时执行的只有固定数量不会把GPU打满。Python的queue.Queue就能实现但更推荐用asyncio.Queue配合异步。限流的思路是限制每个客户端在单位时间内的请求数。FastAPI可以用slowapi这个库基于IP做限流from slowapi import Limiter from slowapi.util import get_remote_address limiter Limiter(key_funcget_remote_address) app.state.limiter limiter app.post(/predict) limiter.limit(10/minute) async def predict(request: Request, req: PredictRequest): ...10/minute表示每分钟最多10次请求超过返回429状态码。这个限制要根据实际业务来定太严会影响正常用户太松起不到保护作用。提示限流和队列要配合使用。限流挡住恶意请求队列缓冲突发流量。只限流不限队突发流量会被拒绝只限队不限流恶意请求会占满队列。5. 监控、日志与迭代让系统越跑越稳5.1 日志该记什么、不该记什么AI服务的日志和普通后端服务不太一样。普通后端记请求路径、状态码、耗时就够了AI服务还要记模型版本、输入长度、推理耗时、显存占用这些。但日志不是越多越好。我见过一个服务每次请求把完整的输入文本和输出结果都打到日志里结果日志文件一天涨了十几个GB磁盘直接满了。而且输入文本里可能包含用户隐私信息打到日志里是有合规风险的。我的做法是记录元数据不记录内容。具体来说记录请求ID、模型名称、输入文本的长度不是内容、推理耗时、显存峰值、输出标签和置信度。这些信息足够排查问题又不会泄露隐私。import logging import time logger logging.getLogger(ai-service) app.post(/predict) async def predict(req: PredictRequest): start time.time() request_id str(uuid.uuid4()) logger.info(frequest_id{request_id} model{req.model_name} input_len{len(req.text)}) try: result do_inference(req) elapsed time.time() - start logger.info(frequest_id{request_id} label{result.label} score{result.score:.4f} elapsed{elapsed:.3f}s) return result except Exception as e: logger.error(frequest_id{request_id} error{str(e)}) raise用request_id把同一次请求的多条日志串起来排查问题时用grep就能看到完整链路。耗时用秒为单位保留三位小数足够精确又不会太冗长。5.2 关键监控指标与告警阈值设定日志是被动的出了问题才去查。监控是主动的指标异常时自动告警。AI服务需要监控的指标分几类。性能指标QPS每秒请求数、P50/P95/P99延迟、错误率。P99延迟特别重要因为它反映了最慢的那部分请求往往就是用户体验最差的那些。如果P99延迟突然飙升可能是某个请求的输入特别长或者显存不够导致频繁换页。资源指标GPU利用率、显存占用、CPU利用率、内存占用。GPU利用率长期低于30%说明资源浪费长期高于90%说明快到瓶颈了。显存占用要留至少20%的余量否则容易OOM。业务指标各标签的分布、置信度的分布。如果某个标签的请求量突然暴增可能是上游业务出了问题如果置信度普遍下降可能是数据分布发生了变化模型需要重新训练了。告警阈值怎么定我的经验是先跑一周收集正常情况下的指标分布然后取P99值上浮20%作为告警线。比如正常P99延迟是200ms那告警线设在240ms。这样既能及时发现异常又不会因为正常波动频繁误报。指标类型具体指标建议告警阈值排查方向性能P99延迟正常值上浮20%输入长度、显存、模型大小性能错误率超过1%看错误日志、依赖服务资源GPU利用率持续低于20%或高于95%并发数、batch size资源显存占用超过80%模型大小、batch size业务标签分布与基线偏差超过30%上游数据、模型版本5.3 模型迭代与A/B测试的工程实现模型不是部署上去就完事了后续还要迭代。新模型怎么上线直接替换风险太大万一效果变差回滚都来不及。标准做法是A/B测试。A/B测试的工程实现核心是流量切分。比如新模型上线先切10%的流量过去观察一周如果各项指标都不差于老模型再逐步增加到50%、100%。切分可以按请求ID的哈希值来做保证同一个用户始终看到同一个模型。import hashlib def choose_model(request_id, new_model_ratio0.1): hash_val int(hashlib.md5(request_id.encode()).hexdigest(), 16) if hash_val % 100 new_model_ratio * 100: return new-model return old-model这个函数用MD5哈希取模把请求ID映射到0到99之间的一个数小于10的用新模型其余用老模型。new_model_ratio参数控制切分比例改这个参数就能调整流量分配。A/B测试期间日志里要记录用的是哪个模型这样才能对比两个模型的效果。对比的指标包括延迟、错误率、以及业务侧的指标比如点击率、转化率。只有所有指标都不差于老模型才值得全量上线。还有一个容易被忽视的点模型版本管理。每次上线的模型要存好包括模型文件、配置文件、训练数据版本、评估结果。出了问题要能快速回滚到上一个版本。我习惯用日期加序号来命名模型版本比如20240115-01然后在配置里指定当前使用的版本。回滚就是改一行配置的事。6. 一些让我少走弯路的实操心得上面按模块讲了AI工程从零搭建的各个环节最后再分享几个跨模块的经验都是我在实际项目里踩过坑之后总结出来的。第一先跑通再优化。很多人一上来就追求最优架构结果卡在某个细节上出不来。我的做法是先用最简单的方式跑通全流程哪怕性能很差、代码很丑只要端到端能跑就有了一个可迭代的基础。然后再逐个环节优化每次只改一个地方改完验证效果。第二把配置和代码分开。模型路径、batch size、超时时间这些参数不要硬编码在代码里放到配置文件或者环境变量里。这样换环境、调参数都不用改代码也避免了把敏感信息提交到代码仓库。第三写测试尤其是集成测试。单元测试测单个函数集成测试测整个流程。AI系统的集成测试至少要有模型加载测试、单条推理测试、批量推理测试、异常输入测试。这些测试跑一遍可能只要几十秒但能帮你发现90%的低级问题。第四文档是给自己看的。我写文档不是为了给别人看而是为了三个月后的自己。三个月后你肯定忘了当时为什么选这个参数、为什么用这个方案。把决策理由记下来比记操作步骤更重要。第五不要追求一步到位。AI工程这个领域变化很快今天的最佳实践明天可能就过时了。保持学习保持迭代但不要为了追新而追新。能解决问题的方案就是好方案哪怕它看起来不够“先进”。这个方向的内容其实还有很多可以展开比如分布式推理、模型压缩、边缘部署等等。但那些属于进阶话题先把上面这些基础打牢后面再深入会顺畅很多。我在实际项目里最大的体会是AI工程的难点往往不在AI本身而在工程。把工程基础打好了模型换了一个又一个系统依然稳如泰山。