尧图网站建设 
1. 从零搭建AI工程能力为什么“会用模型”和“会做工程”是两码事很多人第一次接触AI项目时都会经历一个相似的阶段在笔记本里跑通一个模型准确率看着还不错于是觉得“AI不过如此”。可一旦要把这个东西放到真实业务里问题就全冒出来了——模型加载慢、显存不够、接口响应超时、并发一上来就崩、日志里全是看不懂的报错。这时候才会意识到训练一个模型和交付一个AI系统中间隔着一整条工程链路。“ai-engineering-from-scratch”这个标题核心讲的其实就是这条链路。它不是教你某个框架的API怎么调也不是带你复现某篇论文而是从零开始把AI工程里那些真正决定项目成败的环节一个个搭起来。关键词“ai-engineering-from-scratch”本身就点明了两个重点一是AI工程二是从零构建。这意味着内容会覆盖环境搭建、数据处理、模型服务化、性能优化、监控运维等完整流程而不是停留在算法层面。这篇文章适合三类人看。第一类是有一定Python基础、做过一些数据分析或简单模型训练但没真正把模型部署上线的开发者第二类是从后端或运维转过来想补齐AI工程能力的工程师第三类是在小团队里“一个人当一支队伍用”既要调模型又要管服务的全栈选手。如果你属于其中任何一种下面的内容应该能帮你少走不少弯路。我自己的经历是最早做AI项目时总觉得工程部分是“脏活累活”不如调参有意思。后来被线上问题反复教育之后才明白AI工程的核心不是让模型跑起来而是让模型在真实流量下稳定、可观测、可迭代地跑下去。这个认知转变是从零搭建AI工程能力的第一步。2. 环境与依赖从零搭建时最容易埋下的三个雷2.1 为什么“在我机器上能跑”是AI工程的第一大坑AI项目对环境极其敏感。Python版本、CUDA版本、PyTorch或TensorFlow版本、各种底层库的版本任意一个不匹配都可能导致模型加载失败或者推理结果异常。更麻烦的是很多AI库之间的依赖关系是隐式的pip install的时候不会报错但运行到某个特定算子时才崩。从零搭建时我的建议是先锁定运行时环境再谈代码。具体做法是用Docker定义一个基础镜像把Python版本、CUDA版本、核心框架版本全部写死在Dockerfile里。不要用latest标签不要依赖系统自带的Python不要在宿主机上直接pip install。下面是一个最小化的示例FROM nvidia/cuda:12.1.1-cudnn8-runtime-ubuntu22.04 RUN apt-get update apt-get install -y \ python3.10 \ python3-pip \ rm -rf /var/lib/apt/lists/* RUN pip3 install --no-cache-dir \ torch2.1.0 \ transformers4.35.0 \ fastapi0.104.0 \ uvicorn0.24.0这个Dockerfile看起来简单但它解决了一个关键问题环境一致性。你在本地跑通的代码推到测试环境、生产环境用的都是同一个镜像不会因为系统库差异导致行为不一致。注意CUDA版本要和宿主机驱动版本匹配。驱动版本决定了你能用的最高CUDA版本不是反过来。很多人在这里踩坑装了半天发现驱动太旧只能降CUDA版本。2.2 依赖管理的正确姿势requirements.txt不够用很多人习惯用pip freeze requirements.txt来管理依赖这在纯Python项目里勉强够用但在AI工程里问题很大。因为pip freeze会把所有间接依赖都写进去导致文件巨大且难以维护更严重的是它不区分直接依赖和间接依赖升级某个库时容易引发连锁反应。我的做法是分层管理依赖。用一个requirements.in写直接依赖然后用pip-compile生成锁定版本的requirements.txt。这样既保证了可复现性又保留了可维护性。# requirements.in torch2.1.0 transformers4.35.0 fastapi0.104.0 uvicorn0.24.0 pydantic2.4.0 # 生成锁定文件 pip-compile requirements.in -o requirements.txt另外AI项目里经常需要区分训练依赖和推理依赖。训练时需要datasets、accelerate、wandb这些推理时完全不需要。如果混在一起推理镜像会白白大出好几个GB。建议拆成requirements-train.in和requirements-serve.in分别编译。2.3 模型文件的管理别把权重当代码提交从零搭建AI工程时另一个常见错误是把模型权重文件直接放进Git仓库。一个BERT-base模型大约400MB一个7B参数的模型动辄十几个GBGit根本扛不住。而且模型文件是二进制Git的diff机制完全失效仓库会迅速膨胀到无法维护。正确的做法是模型文件与代码分离。训练产出的权重上传到对象存储或模型仓库代码里只保留模型标识和版本号。加载时通过配置指定路径或远程地址。如果是Hugging Face的模型直接用from_pretrained(模型名)即可框架会自动缓存到本地。from transformers import AutoModelForSequenceClassification, AutoTokenizer model_name bert-base-chinese tokenizer AutoTokenizer.from_pretrained(model_name) model AutoModelForSequenceClassification.from_pretrained( model_name, num_labels2, cache_dir/data/model_cache )这里cache_dir指定了缓存目录避免默认缓存到用户主目录导致磁盘爆满。在生产环境里这个目录最好挂载到独立的数据盘上。3. 模型服务化把notebook里的模型变成能扛流量的API3.1 为什么FastAPI是AI服务化的默认选择模型训练完之后下一步是把它变成一个HTTP接口。可选方案有很多Flask、FastAPI、Tornado、gRPC甚至直接写个socket服务。但在AI工程场景下FastAPI几乎是默认答案原因有三个。第一FastAPI原生支持异步。AI推理往往是IO密集和计算密集混合的场景异步能让服务在等待GPU计算时处理其他请求。第二FastAPI基于Pydantic做请求校验输入输出的数据结构定义清晰减少了很多手写校验的代码。第三FastAPI自动生成OpenAPI文档前后端联调时非常方便。下面是一个最小化的推理服务示例from fastapi import FastAPI from pydantic import BaseModel import torch from transformers import AutoModelForSequenceClassification, AutoTokenizer app FastAPI() class PredictRequest(BaseModel): text: str class PredictResponse(BaseModel): label: int score: float model None tokenizer None app.on_event(startup) def load_model(): global model, tokenizer model_name bert-base-chinese tokenizer AutoTokenizer.from_pretrained(model_name) model AutoModelForSequenceClassification.from_pretrained(model_name, num_labels2) model.eval() if torch.cuda.is_available(): model.to(cuda) app.post(/predict, response_modelPredictResponse) def predict(req: PredictRequest): inputs tokenizer(req.text, return_tensorspt, truncationTrue, max_length512) if torch.cuda.is_available(): inputs {k: v.to(cuda) for k, v in inputs.items()} with torch.no_grad(): outputs model(**inputs) probs torch.softmax(outputs.logits, dim-1) score, label torch.max(probs, dim-1) return PredictResponse(labellabel.item(), scorescore.item())这段代码看起来简单但有几个关键点值得展开。app.on_event(startup)确保模型在服务启动时加载一次而不是每次请求都加载。model.eval()切换到推理模式关闭dropout等训练专用层。torch.no_grad()关闭梯度计算减少显存占用。这些都是从零搭建时必须养成的习惯。3.2 批处理与动态合并提升GPU利用率的关键单条推理的GPU利用率通常很低因为GPU的计算单元远多于单条数据所需。如果每条请求都单独跑一次前向传播大部分算力是浪费的。解决办法是批处理把多个请求合并成一个batch一次性送进GPU。但批处理有个矛盾如果等batch凑满再推理延迟会变高如果不等batch又太小。工程上的常见做法是动态合并设置一个极短的时间窗口比如10毫秒窗口内到达的请求合并成一个batch。这样既提升了吞吐又控制了延迟。import asyncio from collections import deque class BatchProcessor: def __init__(self, model, tokenizer, max_batch_size32, window_ms10): self.model model self.tokenizer tokenizer self.max_batch_size max_batch_size self.window window_ms / 1000.0 self.queue deque() async def process(self, text): future asyncio.get_event_loop().create_future() self.queue.append((text, future)) if len(self.queue) self.max_batch_size: await self._flush() else: await asyncio.sleep(self.window) if self.queue: await self._flush() return await future async def _flush(self): batch list(self.queue) self.queue.clear() texts [item[0] for item in batch] inputs self.tokenizer(texts, return_tensorspt, paddingTrue, truncationTrue, max_length512) with torch.no_grad(): outputs self.model(**inputs) probs torch.softmax(outputs.logits, dim-1) for i, (_, future) in enumerate(batch): score, label torch.max(probs[i], dim-1) future.set_result({label: label.item(), score: score.item()})这个实现是简化版生产环境还需要考虑超时、异常处理、队列上限等。但核心思路是清晰的用微小的时间窗口换取显著的吞吐提升。实测下来在并发请求较多的场景下动态批处理能把GPU利用率从20%提升到60%以上。3.3 模型加载的冷启动问题与预热策略服务刚启动时第一次推理往往特别慢因为CUDA上下文初始化、显存分配、算子编译都需要时间。如果这时候正好有真实流量打进来用户体验会很差。解决办法是预热服务启动后先用几条假数据跑一遍推理把该初始化的都初始化好。app.on_event(startup) def warmup(): dummy_text 这是一条预热数据 inputs tokenizer(dummy_text, return_tensorspt, truncationTrue, max_length128) if torch.cuda.is_available(): inputs {k: v.to(cuda) for k, v in inputs.items()} with torch.no_grad(): for _ in range(3): model(**inputs) torch.cuda.synchronize()预热跑3到5次就够了目的是触发CUDA kernel的编译和缓存。torch.cuda.synchronize()确保所有异步操作完成避免预热还没结束服务就开始接收流量。提示如果模型特别大预热时间可能超过健康检查的超时时间。这时候需要调整健康检查的初始延迟或者把预热放到一个单独的初始化阶段健康检查通过后再接入流量。4. 性能与资源显存、并发和延迟的三角博弈4.1 显存不够时先别急着换卡做AI工程的人几乎都遇到过CUDA out of memory。第一反应往往是“显卡太小了得换大的”。但很多时候显存问题不是硬件不够而是使用方式不对。在换卡之前有几个手段可以先试。降低batch size是最直接的。但要注意推理时的batch size和训练时不是一回事。推理不需要存梯度所以同样的显存可以跑更大的batch。如果推理还OOM说明模型本身或者输入长度有问题。使用半精度是另一个有效手段。把模型从float32转成float16显存占用直接减半推理速度还能提升。大部分场景下精度损失可以忽略。model AutoModelForSequenceClassification.from_pretrained(model_name, num_labels2) model.half() # 转为float16 model.to(cuda)及时释放中间变量也很重要。Python的垃圾回收不是实时的显存里的tensor如果一直被引用就不会被释放。在推理循环里尽量用with torch.no_grad()包裹并且避免把中间结果存到全局变量里。如果以上都试过还是不够再考虑模型量化。8-bit量化能把显存占用降到原来的四分之一左右4-bit量化还能更低。Hugging Face的bitsandbytes库提供了现成的量化加载方式from transformers import BitsAndBytesConfig quant_config BitsAndBytesConfig( load_in_8bitTrue, llm_int8_threshold6.0 ) model AutoModelForSequenceClassification.from_pretrained( model_name, quantization_configquant_config, device_mapauto )量化会带来一定的精度损失和推理速度变化需要根据具体任务评估。分类任务通常影响不大生成任务可能需要更谨慎。4.2 并发模型的选择同步、异步还是多进程AI服务的并发模型和普通Web服务不太一样因为GPU是独占资源。多个线程同时往GPU上送数据不但不会加速反而会因为上下文切换和显存竞争导致性能下降。我的经验是用异步框架处理网络IO用单线程或固定数量的工作进程处理GPU推理。FastAPI本身是异步的但推理函数如果是同步的会阻塞事件循环。解决办法是把推理放到线程池里执行from concurrent.futures import ThreadPoolExecutor import asyncio executor ThreadPoolExecutor(max_workers1) app.post(/predict) async def predict_async(req: PredictRequest): loop asyncio.get_event_loop() result await loop.run_in_executor(executor, sync_predict, req.text) return result def sync_predict(text): inputs tokenizer(text, return_tensorspt, truncationTrue, max_length512) if torch.cuda.is_available(): inputs {k: v.to(cuda) for k, v in inputs.items()} with torch.no_grad(): outputs model(**inputs) probs torch.softmax(outputs.logits, dim-1) score, label torch.max(probs, dim-1) return {label: label.item(), score: score.item()}max_workers1意味着所有推理请求排队执行避免GPU竞争。如果有多张卡可以每个卡一个worker用不同的executor。这种设计下服务的吞吐取决于单次推理时间和队列长度而不是并发线程数。4.3 延迟优化的几个实用手段延迟是AI服务最敏感的指标之一。用户可接受的延迟通常在几百毫秒以内超过1秒就会明显感到卡顿。优化延迟可以从几个层面入手。输入长度控制是最有效的。Transformer的计算复杂度随序列长度平方增长把max_length从512降到128推理时间可能减少一半以上。如果任务本身不需要长文本就不要设太大的max_length。模型蒸馏或剪枝能从根本上减少计算量。用一个小的学生模型模仿大模型的行为推理速度可以提升数倍。但蒸馏需要额外的训练流程适合对延迟要求极高的场景。缓存也很关键。如果同样的输入反复出现可以把结果缓存起来直接返回。对于问答、分类这类确定性任务缓存命中率往往不低。from functools import lru_cache lru_cache(maxsize10000) def cached_predict(text): return sync_predict(text)lru_cache适合单进程场景。如果多进程部署需要用Redis之类的共享缓存。注意缓存key要包含模型版本模型更新后旧缓存要失效。5. 可观测性模型上线后你怎么知道它“还活着”5.1 日志、指标、追踪AI服务不能只看CPU和内存普通Web服务的监控主要看QPS、响应时间、错误率、CPU、内存。AI服务除了这些还需要关注GPU利用率、显存占用、推理延迟分布、输入长度分布、模型输出分布。这些指标能帮你判断服务是否健康以及模型是否在“正常思考”。GPU指标可以用pynvml采集import pynvml pynvml.nvmlInit() handle pynvml.nvmlDeviceGetHandleByIndex(0) def get_gpu_metrics(): util pynvml.nvmlDeviceGetUtilizationRates(handle) mem pynvml.nvmlDeviceGetMemoryInfo(handle) return { gpu_util: util.gpu, gpu_mem_used: mem.used, gpu_mem_total: mem.total, gpu_mem_ratio: mem.used / mem.total }这些指标可以暴露给Prometheus然后在Grafana里做面板。重点看两个**GPU利用率长期低于20%**说明资源浪费显存占用持续上涨可能意味着有内存泄漏。5.2 模型输出监控比错误率更早发现问题的信号错误率只能告诉你“请求失败了”但模型输出异常往往不会报错。比如一个分类模型突然把所有输入都预测成同一类接口返回200错误率为0但业务上已经完全不可用了。解决办法是监控模型输出的分布。对于分类任务统计各类别的预测比例对于生成任务统计输出长度、重复率、困惑度等。如果分布发生显著偏移就触发告警。from collections import Counter import numpy as np class OutputMonitor: def __init__(self, window_size1000): self.window_size window_size self.labels [] self.scores [] def record(self, label, score): self.labels.append(label) self.scores.append(score) if len(self.labels) self.window_size: self.labels.pop(0) self.scores.pop(0) def stats(self): label_counts Counter(self.labels) total len(self.labels) return { label_distribution: {k: v / total for k, v in label_counts.items()}, avg_score: np.mean(self.scores) if self.scores else 0, score_std: np.std(self.scores) if self.scores else 0 }如果某个类别的比例突然从10%跳到90%或者平均置信度从0.9掉到0.5都说明有问题。可能是数据分布变了也可能是模型加载错了版本。5.3 从告警到定位一条完整的排查链路假设你收到一条告警“推理服务P99延迟从200ms涨到2s”。接下来怎么排查我的习惯是按这个顺序走第一步看GPU利用率。如果GPU利用率接近100%说明计算是瓶颈可能是请求量涨了或者输入变长了。如果GPU利用率很低但延迟很高说明瓶颈不在GPU可能在网络、队列或者锁竞争。第二步看输入长度分布。如果平均输入长度突然变大Transformer的平方复杂度会直接反映到延迟上。这时候可以考虑截断或者拒绝超长输入。第三步看队列长度。如果请求排队严重说明处理速度跟不上到达速度。要么加资源要么限流。第四步看是否有异常请求。有些请求可能触发了模型的慢路径比如特别长的输入、特别罕见的类别。可以在日志里标记慢请求单独分析。这套链路的关键是每个环节都有对应的指标而不是靠猜。从零搭建AI工程时可观测性往往是最容易被忽略的部分但它是服务稳定运行的底线。6. 迭代与回滚模型更新不是覆盖文件那么简单6.1 模型版本管理每个线上模型都应该可追溯模型更新比代码更新更复杂因为模型是二进制文件无法像代码一样diff。如果没有版本管理出了问题根本不知道线上跑的是哪个模型、用什么数据训练的、超参数是什么。我的做法是给每个模型分配唯一版本号版本号里包含训练日期、数据版本、关键超参数。比如cls-20240115-v3-lr2e5。模型文件、配置文件、评估报告一起打包存到模型仓库里。线上服务通过配置指定版本号而不是直接指定文件路径。import mlflow mlflow.set_tracking_uri(http://mlflow-server:5000) with mlflow.start_run(run_namebert-classifier-v3): mlflow.log_params({lr: 2e-5, batch_size: 32, epochs: 3}) mlflow.log_metrics({accuracy: 0.92, f1: 0.91}) mlflow.pytorch.log_model(model, model)MLflow只是其中一种方案核心是模型、参数、指标三者绑定。这样回滚的时候你知道要回到哪个版本也知道那个版本的表现如何。6.2 灰度发布让新模型先见一小部分流量模型更新最忌讳的是全量替换。新模型可能在离线评估里表现很好但线上数据分布不同直接全量上线风险很大。灰度发布的做法是新模型先接1%的流量观察一段时间指标正常再逐步扩大比例。实现灰度发布有几种方式。简单的是在服务里根据请求ID哈希决定走哪个模型import hashlib def select_model(request_id): hash_val int(hashlib.md5(request_id.encode()).hexdigest(), 16) if hash_val % 100 1: return new_model return old_model更规范的做法是用服务网格或者网关来做流量切分把灰度逻辑从业务代码里剥离出去。但不管哪种方式核心都是新模型先小流量验证再逐步放量。6.3 回滚预案出问题时能在几分钟内切回旧版本灰度发布的同时必须准备好回滚预案。回滚不是“把旧模型文件再拷回去”而是在秒级或分钟级内把流量切回旧版本。这要求旧版本的模型和服务一直处于待命状态而不是被新版本覆盖掉。我的做法是双版本并行部署。新版本上线时旧版本的服务不销毁只是把流量切走。一旦新版本出问题改一下流量配置就能切回来。旧版本保留至少一周确认新版本稳定后再下线。# 流量配置示例 routes: - match: prefix: /predict route: - destination: host: model-service subset: v3 weight: 99 - destination: host: model-service subset: v4 weight: 1这种配置下回滚只需要把weight从1改回0把99改成100。整个过程不需要重新部署几秒钟就能完成。7. 一些踩过坑之后才明白的事做AI工程这些年有些教训是文档里不会写的只有真正踩过才知道疼。第一不要相信“离线指标好线上就一定好”。离线评估用的是历史数据线上数据分布可能完全不同。我见过一个模型离线F1是0.95上线后实际效果不到0.6原因是训练数据里某个类别的样本全是特定来源的线上这个来源的流量占比很小。所以上线前一定要做在线小流量验证哪怕只是人工看几百条真实请求的结果。第二日志里不要只记“成功”和“失败”。中间状态、输入摘要、输出摘要、耗时、模型版本这些都要记。出问题的时候这些信息就是你的救命稻草。但要注意脱敏用户输入里可能有敏感信息不能原样落盘。第三资源限制要提前设。容器的CPU和内存限制、GPU的显存上限、请求的超时时间、队列的最大长度这些都要在服务上线前配置好。否则一个异常请求可能拖垮整个服务。我吃过一次亏一个用户传了10万字的文本直接把显存打满服务挂了半小时。第四模型加载失败要有明确的错误信息。不要只抛一个RuntimeError要告诉运维人员是哪个文件找不到、哪个版本不匹配、哪个依赖缺失。错误信息越具体恢复时间越短。第五定期做故障演练。手动杀掉一个推理进程看看服务能不能自动恢复把模型文件临时移走看看健康检查能不能正确报错模拟GPU显存耗尽看看有没有优雅降级。这些演练能帮你在真实故障发生时不慌。第六文档要写给“凌晨三点被叫醒的人”看。不要写“启动服务即可”要写清楚启动命令、依赖服务、常见错误、回滚步骤。最好有一个RUNBOOK.md放在仓库根目录任何人拿到都能照着操作。这些经验听起来琐碎但AI工程的稳定性就是靠这些细节堆出来的。从零搭建AI工程能力技术选型只是开始真正难的是把每一个环节都考虑到、配置好、验证过。 从零搭建AI工程能力：模型服务化与性能优化实战
1. 从零搭建AI工程能力为什么“会用模型”和“会做工程”是两码事很多人第一次接触AI项目时都会经历一个相似的阶段在笔记本里跑通一个模型准确率看着还不错于是觉得“AI不过如此”。可一旦要把这个东西放到真实业务里问题就全冒出来了——模型加载慢、显存不够、接口响应超时、并发一上来就崩、日志里全是看不懂的报错。这时候才会意识到训练一个模型和交付一个AI系统中间隔着一整条工程链路。“ai-engineering-from-scratch”这个标题核心讲的其实就是这条链路。它不是教你某个框架的API怎么调也不是带你复现某篇论文而是从零开始把AI工程里那些真正决定项目成败的环节一个个搭起来。关键词“ai-engineering-from-scratch”本身就点明了两个重点一是AI工程二是从零构建。这意味着内容会覆盖环境搭建、数据处理、模型服务化、性能优化、监控运维等完整流程而不是停留在算法层面。这篇文章适合三类人看。第一类是有一定Python基础、做过一些数据分析或简单模型训练但没真正把模型部署上线的开发者第二类是从后端或运维转过来想补齐AI工程能力的工程师第三类是在小团队里“一个人当一支队伍用”既要调模型又要管服务的全栈选手。如果你属于其中任何一种下面的内容应该能帮你少走不少弯路。我自己的经历是最早做AI项目时总觉得工程部分是“脏活累活”不如调参有意思。后来被线上问题反复教育之后才明白AI工程的核心不是让模型跑起来而是让模型在真实流量下稳定、可观测、可迭代地跑下去。这个认知转变是从零搭建AI工程能力的第一步。2. 环境与依赖从零搭建时最容易埋下的三个雷2.1 为什么“在我机器上能跑”是AI工程的第一大坑AI项目对环境极其敏感。Python版本、CUDA版本、PyTorch或TensorFlow版本、各种底层库的版本任意一个不匹配都可能导致模型加载失败或者推理结果异常。更麻烦的是很多AI库之间的依赖关系是隐式的pip install的时候不会报错但运行到某个特定算子时才崩。从零搭建时我的建议是先锁定运行时环境再谈代码。具体做法是用Docker定义一个基础镜像把Python版本、CUDA版本、核心框架版本全部写死在Dockerfile里。不要用latest标签不要依赖系统自带的Python不要在宿主机上直接pip install。下面是一个最小化的示例FROM nvidia/cuda:12.1.1-cudnn8-runtime-ubuntu22.04 RUN apt-get update apt-get install -y \ python3.10 \ python3-pip \ rm -rf /var/lib/apt/lists/* RUN pip3 install --no-cache-dir \ torch2.1.0 \ transformers4.35.0 \ fastapi0.104.0 \ uvicorn0.24.0这个Dockerfile看起来简单但它解决了一个关键问题环境一致性。你在本地跑通的代码推到测试环境、生产环境用的都是同一个镜像不会因为系统库差异导致行为不一致。注意CUDA版本要和宿主机驱动版本匹配。驱动版本决定了你能用的最高CUDA版本不是反过来。很多人在这里踩坑装了半天发现驱动太旧只能降CUDA版本。2.2 依赖管理的正确姿势requirements.txt不够用很多人习惯用pip freeze requirements.txt来管理依赖这在纯Python项目里勉强够用但在AI工程里问题很大。因为pip freeze会把所有间接依赖都写进去导致文件巨大且难以维护更严重的是它不区分直接依赖和间接依赖升级某个库时容易引发连锁反应。我的做法是分层管理依赖。用一个requirements.in写直接依赖然后用pip-compile生成锁定版本的requirements.txt。这样既保证了可复现性又保留了可维护性。# requirements.in torch2.1.0 transformers4.35.0 fastapi0.104.0 uvicorn0.24.0 pydantic2.4.0 # 生成锁定文件 pip-compile requirements.in -o requirements.txt另外AI项目里经常需要区分训练依赖和推理依赖。训练时需要datasets、accelerate、wandb这些推理时完全不需要。如果混在一起推理镜像会白白大出好几个GB。建议拆成requirements-train.in和requirements-serve.in分别编译。2.3 模型文件的管理别把权重当代码提交从零搭建AI工程时另一个常见错误是把模型权重文件直接放进Git仓库。一个BERT-base模型大约400MB一个7B参数的模型动辄十几个GBGit根本扛不住。而且模型文件是二进制Git的diff机制完全失效仓库会迅速膨胀到无法维护。正确的做法是模型文件与代码分离。训练产出的权重上传到对象存储或模型仓库代码里只保留模型标识和版本号。加载时通过配置指定路径或远程地址。如果是Hugging Face的模型直接用from_pretrained(模型名)即可框架会自动缓存到本地。from transformers import AutoModelForSequenceClassification, AutoTokenizer model_name bert-base-chinese tokenizer AutoTokenizer.from_pretrained(model_name) model AutoModelForSequenceClassification.from_pretrained( model_name, num_labels2, cache_dir/data/model_cache )这里cache_dir指定了缓存目录避免默认缓存到用户主目录导致磁盘爆满。在生产环境里这个目录最好挂载到独立的数据盘上。3. 模型服务化把notebook里的模型变成能扛流量的API3.1 为什么FastAPI是AI服务化的默认选择模型训练完之后下一步是把它变成一个HTTP接口。可选方案有很多Flask、FastAPI、Tornado、gRPC甚至直接写个socket服务。但在AI工程场景下FastAPI几乎是默认答案原因有三个。第一FastAPI原生支持异步。AI推理往往是IO密集和计算密集混合的场景异步能让服务在等待GPU计算时处理其他请求。第二FastAPI基于Pydantic做请求校验输入输出的数据结构定义清晰减少了很多手写校验的代码。第三FastAPI自动生成OpenAPI文档前后端联调时非常方便。下面是一个最小化的推理服务示例from fastapi import FastAPI from pydantic import BaseModel import torch from transformers import AutoModelForSequenceClassification, AutoTokenizer app FastAPI() class PredictRequest(BaseModel): text: str class PredictResponse(BaseModel): label: int score: float model None tokenizer None app.on_event(startup) def load_model(): global model, tokenizer model_name bert-base-chinese tokenizer AutoTokenizer.from_pretrained(model_name) model AutoModelForSequenceClassification.from_pretrained(model_name, num_labels2) model.eval() if torch.cuda.is_available(): model.to(cuda) app.post(/predict, response_modelPredictResponse) def predict(req: PredictRequest): inputs tokenizer(req.text, return_tensorspt, truncationTrue, max_length512) if torch.cuda.is_available(): inputs {k: v.to(cuda) for k, v in inputs.items()} with torch.no_grad(): outputs model(**inputs) probs torch.softmax(outputs.logits, dim-1) score, label torch.max(probs, dim-1) return PredictResponse(labellabel.item(), scorescore.item())这段代码看起来简单但有几个关键点值得展开。app.on_event(startup)确保模型在服务启动时加载一次而不是每次请求都加载。model.eval()切换到推理模式关闭dropout等训练专用层。torch.no_grad()关闭梯度计算减少显存占用。这些都是从零搭建时必须养成的习惯。3.2 批处理与动态合并提升GPU利用率的关键单条推理的GPU利用率通常很低因为GPU的计算单元远多于单条数据所需。如果每条请求都单独跑一次前向传播大部分算力是浪费的。解决办法是批处理把多个请求合并成一个batch一次性送进GPU。但批处理有个矛盾如果等batch凑满再推理延迟会变高如果不等batch又太小。工程上的常见做法是动态合并设置一个极短的时间窗口比如10毫秒窗口内到达的请求合并成一个batch。这样既提升了吞吐又控制了延迟。import asyncio from collections import deque class BatchProcessor: def __init__(self, model, tokenizer, max_batch_size32, window_ms10): self.model model self.tokenizer tokenizer self.max_batch_size max_batch_size self.window window_ms / 1000.0 self.queue deque() async def process(self, text): future asyncio.get_event_loop().create_future() self.queue.append((text, future)) if len(self.queue) self.max_batch_size: await self._flush() else: await asyncio.sleep(self.window) if self.queue: await self._flush() return await future async def _flush(self): batch list(self.queue) self.queue.clear() texts [item[0] for item in batch] inputs self.tokenizer(texts, return_tensorspt, paddingTrue, truncationTrue, max_length512) with torch.no_grad(): outputs self.model(**inputs) probs torch.softmax(outputs.logits, dim-1) for i, (_, future) in enumerate(batch): score, label torch.max(probs[i], dim-1) future.set_result({label: label.item(), score: score.item()})这个实现是简化版生产环境还需要考虑超时、异常处理、队列上限等。但核心思路是清晰的用微小的时间窗口换取显著的吞吐提升。实测下来在并发请求较多的场景下动态批处理能把GPU利用率从20%提升到60%以上。3.3 模型加载的冷启动问题与预热策略服务刚启动时第一次推理往往特别慢因为CUDA上下文初始化、显存分配、算子编译都需要时间。如果这时候正好有真实流量打进来用户体验会很差。解决办法是预热服务启动后先用几条假数据跑一遍推理把该初始化的都初始化好。app.on_event(startup) def warmup(): dummy_text 这是一条预热数据 inputs tokenizer(dummy_text, return_tensorspt, truncationTrue, max_length128) if torch.cuda.is_available(): inputs {k: v.to(cuda) for k, v in inputs.items()} with torch.no_grad(): for _ in range(3): model(**inputs) torch.cuda.synchronize()预热跑3到5次就够了目的是触发CUDA kernel的编译和缓存。torch.cuda.synchronize()确保所有异步操作完成避免预热还没结束服务就开始接收流量。提示如果模型特别大预热时间可能超过健康检查的超时时间。这时候需要调整健康检查的初始延迟或者把预热放到一个单独的初始化阶段健康检查通过后再接入流量。4. 性能与资源显存、并发和延迟的三角博弈4.1 显存不够时先别急着换卡做AI工程的人几乎都遇到过CUDA out of memory。第一反应往往是“显卡太小了得换大的”。但很多时候显存问题不是硬件不够而是使用方式不对。在换卡之前有几个手段可以先试。降低batch size是最直接的。但要注意推理时的batch size和训练时不是一回事。推理不需要存梯度所以同样的显存可以跑更大的batch。如果推理还OOM说明模型本身或者输入长度有问题。使用半精度是另一个有效手段。把模型从float32转成float16显存占用直接减半推理速度还能提升。大部分场景下精度损失可以忽略。model AutoModelForSequenceClassification.from_pretrained(model_name, num_labels2) model.half() # 转为float16 model.to(cuda)及时释放中间变量也很重要。Python的垃圾回收不是实时的显存里的tensor如果一直被引用就不会被释放。在推理循环里尽量用with torch.no_grad()包裹并且避免把中间结果存到全局变量里。如果以上都试过还是不够再考虑模型量化。8-bit量化能把显存占用降到原来的四分之一左右4-bit量化还能更低。Hugging Face的bitsandbytes库提供了现成的量化加载方式from transformers import BitsAndBytesConfig quant_config BitsAndBytesConfig( load_in_8bitTrue, llm_int8_threshold6.0 ) model AutoModelForSequenceClassification.from_pretrained( model_name, quantization_configquant_config, device_mapauto )量化会带来一定的精度损失和推理速度变化需要根据具体任务评估。分类任务通常影响不大生成任务可能需要更谨慎。4.2 并发模型的选择同步、异步还是多进程AI服务的并发模型和普通Web服务不太一样因为GPU是独占资源。多个线程同时往GPU上送数据不但不会加速反而会因为上下文切换和显存竞争导致性能下降。我的经验是用异步框架处理网络IO用单线程或固定数量的工作进程处理GPU推理。FastAPI本身是异步的但推理函数如果是同步的会阻塞事件循环。解决办法是把推理放到线程池里执行from concurrent.futures import ThreadPoolExecutor import asyncio executor ThreadPoolExecutor(max_workers1) app.post(/predict) async def predict_async(req: PredictRequest): loop asyncio.get_event_loop() result await loop.run_in_executor(executor, sync_predict, req.text) return result def sync_predict(text): inputs tokenizer(text, return_tensorspt, truncationTrue, max_length512) if torch.cuda.is_available(): inputs {k: v.to(cuda) for k, v in inputs.items()} with torch.no_grad(): outputs model(**inputs) probs torch.softmax(outputs.logits, dim-1) score, label torch.max(probs, dim-1) return {label: label.item(), score: score.item()}max_workers1意味着所有推理请求排队执行避免GPU竞争。如果有多张卡可以每个卡一个worker用不同的executor。这种设计下服务的吞吐取决于单次推理时间和队列长度而不是并发线程数。4.3 延迟优化的几个实用手段延迟是AI服务最敏感的指标之一。用户可接受的延迟通常在几百毫秒以内超过1秒就会明显感到卡顿。优化延迟可以从几个层面入手。输入长度控制是最有效的。Transformer的计算复杂度随序列长度平方增长把max_length从512降到128推理时间可能减少一半以上。如果任务本身不需要长文本就不要设太大的max_length。模型蒸馏或剪枝能从根本上减少计算量。用一个小的学生模型模仿大模型的行为推理速度可以提升数倍。但蒸馏需要额外的训练流程适合对延迟要求极高的场景。缓存也很关键。如果同样的输入反复出现可以把结果缓存起来直接返回。对于问答、分类这类确定性任务缓存命中率往往不低。from functools import lru_cache lru_cache(maxsize10000) def cached_predict(text): return sync_predict(text)lru_cache适合单进程场景。如果多进程部署需要用Redis之类的共享缓存。注意缓存key要包含模型版本模型更新后旧缓存要失效。5. 可观测性模型上线后你怎么知道它“还活着”5.1 日志、指标、追踪AI服务不能只看CPU和内存普通Web服务的监控主要看QPS、响应时间、错误率、CPU、内存。AI服务除了这些还需要关注GPU利用率、显存占用、推理延迟分布、输入长度分布、模型输出分布。这些指标能帮你判断服务是否健康以及模型是否在“正常思考”。GPU指标可以用pynvml采集import pynvml pynvml.nvmlInit() handle pynvml.nvmlDeviceGetHandleByIndex(0) def get_gpu_metrics(): util pynvml.nvmlDeviceGetUtilizationRates(handle) mem pynvml.nvmlDeviceGetMemoryInfo(handle) return { gpu_util: util.gpu, gpu_mem_used: mem.used, gpu_mem_total: mem.total, gpu_mem_ratio: mem.used / mem.total }这些指标可以暴露给Prometheus然后在Grafana里做面板。重点看两个**GPU利用率长期低于20%**说明资源浪费显存占用持续上涨可能意味着有内存泄漏。5.2 模型输出监控比错误率更早发现问题的信号错误率只能告诉你“请求失败了”但模型输出异常往往不会报错。比如一个分类模型突然把所有输入都预测成同一类接口返回200错误率为0但业务上已经完全不可用了。解决办法是监控模型输出的分布。对于分类任务统计各类别的预测比例对于生成任务统计输出长度、重复率、困惑度等。如果分布发生显著偏移就触发告警。from collections import Counter import numpy as np class OutputMonitor: def __init__(self, window_size1000): self.window_size window_size self.labels [] self.scores [] def record(self, label, score): self.labels.append(label) self.scores.append(score) if len(self.labels) self.window_size: self.labels.pop(0) self.scores.pop(0) def stats(self): label_counts Counter(self.labels) total len(self.labels) return { label_distribution: {k: v / total for k, v in label_counts.items()}, avg_score: np.mean(self.scores) if self.scores else 0, score_std: np.std(self.scores) if self.scores else 0 }如果某个类别的比例突然从10%跳到90%或者平均置信度从0.9掉到0.5都说明有问题。可能是数据分布变了也可能是模型加载错了版本。5.3 从告警到定位一条完整的排查链路假设你收到一条告警“推理服务P99延迟从200ms涨到2s”。接下来怎么排查我的习惯是按这个顺序走第一步看GPU利用率。如果GPU利用率接近100%说明计算是瓶颈可能是请求量涨了或者输入变长了。如果GPU利用率很低但延迟很高说明瓶颈不在GPU可能在网络、队列或者锁竞争。第二步看输入长度分布。如果平均输入长度突然变大Transformer的平方复杂度会直接反映到延迟上。这时候可以考虑截断或者拒绝超长输入。第三步看队列长度。如果请求排队严重说明处理速度跟不上到达速度。要么加资源要么限流。第四步看是否有异常请求。有些请求可能触发了模型的慢路径比如特别长的输入、特别罕见的类别。可以在日志里标记慢请求单独分析。这套链路的关键是每个环节都有对应的指标而不是靠猜。从零搭建AI工程时可观测性往往是最容易被忽略的部分但它是服务稳定运行的底线。6. 迭代与回滚模型更新不是覆盖文件那么简单6.1 模型版本管理每个线上模型都应该可追溯模型更新比代码更新更复杂因为模型是二进制文件无法像代码一样diff。如果没有版本管理出了问题根本不知道线上跑的是哪个模型、用什么数据训练的、超参数是什么。我的做法是给每个模型分配唯一版本号版本号里包含训练日期、数据版本、关键超参数。比如cls-20240115-v3-lr2e5。模型文件、配置文件、评估报告一起打包存到模型仓库里。线上服务通过配置指定版本号而不是直接指定文件路径。import mlflow mlflow.set_tracking_uri(http://mlflow-server:5000) with mlflow.start_run(run_namebert-classifier-v3): mlflow.log_params({lr: 2e-5, batch_size: 32, epochs: 3}) mlflow.log_metrics({accuracy: 0.92, f1: 0.91}) mlflow.pytorch.log_model(model, model)MLflow只是其中一种方案核心是模型、参数、指标三者绑定。这样回滚的时候你知道要回到哪个版本也知道那个版本的表现如何。6.2 灰度发布让新模型先见一小部分流量模型更新最忌讳的是全量替换。新模型可能在离线评估里表现很好但线上数据分布不同直接全量上线风险很大。灰度发布的做法是新模型先接1%的流量观察一段时间指标正常再逐步扩大比例。实现灰度发布有几种方式。简单的是在服务里根据请求ID哈希决定走哪个模型import hashlib def select_model(request_id): hash_val int(hashlib.md5(request_id.encode()).hexdigest(), 16) if hash_val % 100 1: return new_model return old_model更规范的做法是用服务网格或者网关来做流量切分把灰度逻辑从业务代码里剥离出去。但不管哪种方式核心都是新模型先小流量验证再逐步放量。6.3 回滚预案出问题时能在几分钟内切回旧版本灰度发布的同时必须准备好回滚预案。回滚不是“把旧模型文件再拷回去”而是在秒级或分钟级内把流量切回旧版本。这要求旧版本的模型和服务一直处于待命状态而不是被新版本覆盖掉。我的做法是双版本并行部署。新版本上线时旧版本的服务不销毁只是把流量切走。一旦新版本出问题改一下流量配置就能切回来。旧版本保留至少一周确认新版本稳定后再下线。# 流量配置示例 routes: - match: prefix: /predict route: - destination: host: model-service subset: v3 weight: 99 - destination: host: model-service subset: v4 weight: 1这种配置下回滚只需要把weight从1改回0把99改成100。整个过程不需要重新部署几秒钟就能完成。7. 一些踩过坑之后才明白的事做AI工程这些年有些教训是文档里不会写的只有真正踩过才知道疼。第一不要相信“离线指标好线上就一定好”。离线评估用的是历史数据线上数据分布可能完全不同。我见过一个模型离线F1是0.95上线后实际效果不到0.6原因是训练数据里某个类别的样本全是特定来源的线上这个来源的流量占比很小。所以上线前一定要做在线小流量验证哪怕只是人工看几百条真实请求的结果。第二日志里不要只记“成功”和“失败”。中间状态、输入摘要、输出摘要、耗时、模型版本这些都要记。出问题的时候这些信息就是你的救命稻草。但要注意脱敏用户输入里可能有敏感信息不能原样落盘。第三资源限制要提前设。容器的CPU和内存限制、GPU的显存上限、请求的超时时间、队列的最大长度这些都要在服务上线前配置好。否则一个异常请求可能拖垮整个服务。我吃过一次亏一个用户传了10万字的文本直接把显存打满服务挂了半小时。第四模型加载失败要有明确的错误信息。不要只抛一个RuntimeError要告诉运维人员是哪个文件找不到、哪个版本不匹配、哪个依赖缺失。错误信息越具体恢复时间越短。第五定期做故障演练。手动杀掉一个推理进程看看服务能不能自动恢复把模型文件临时移走看看健康检查能不能正确报错模拟GPU显存耗尽看看有没有优雅降级。这些演练能帮你在真实故障发生时不慌。第六文档要写给“凌晨三点被叫醒的人”看。不要写“启动服务即可”要写清楚启动命令、依赖服务、常见错误、回滚步骤。最好有一个RUNBOOK.md放在仓库根目录任何人拿到都能照着操作。这些经验听起来琐碎但AI工程的稳定性就是靠这些细节堆出来的。从零搭建AI工程能力技术选型只是开始真正难的是把每一个环节都考虑到、配置好、验证过。