1. 从零搭建AI工程能力为什么大多数人卡在第一步聊到AI工程很多人第一反应是“调包”——装个transformers拉个预训练模型跑通一个demo就觉得自己入门了。但真到了要自己搭一套能用的AI系统比如做一个检索增强生成的知识库问答、搭一个多模态内容审核流水线、或者部署一个能扛住并发请求的推理服务立刻就懵了。模型加载报错、显存不够、推理速度慢得离谱、输出格式不稳定、上线之后没人监控……这些问题没有一个能靠“pip install”解决。ai-engineering-from-scratch这个标题核心讲的就是这件事不依赖现成的平台和封装好的黑盒工具从最底层开始把AI工程涉及的关键环节一个一个搭起来。它适合那些已经会写Python、了解基本深度学习概念但没真正独立做过完整AI系统的人。也适合那些平时只用过云端API、想搞清楚背后到底发生了什么的人。我自己的经历是最早做AI应用的时候全部依赖在线接口觉得方便。直到有一次需要处理一批敏感数据不能出本地环境才发现自己根本不知道怎么在本地跑一个像样的模型。从那之后我开始系统性地补AI工程的基础设施知识从模型加载、推理优化、服务封装到监控告警一步一步踩过来。这篇文章就把这条路径上的关键节点拆开讲清楚每个环节都给出可操作的方案和背后的取舍逻辑。2. 模型加载与推理环境别让环境问题吃掉你一半时间2.1 为什么虚拟环境不是可选项而是必选项AI工程的依赖冲突是出了名的严重。PyTorch、CUDA、cuDNN、transformers、accelerate、bitsandbytes这几个库之间的版本兼容关系能让人抓狂。我见过太多人因为直接在系统Python里装包导致原本能跑的代码突然报undefined symbol错误。正确的做法是每个项目一个独立环境。用conda还是venv我的建议是涉及CUDA和深度学习框架的优先用conda因为conda能管理非Python的二进制依赖比如CUDA runtime。纯Python的轻量项目用venv就够了。# 创建conda环境指定Python版本 conda create -n ai-eng python3.10 -y conda activate ai-eng # 安装PyTorch时指定CUDA版本不要直接pip install torch # 去PyTorch官网查对应CUDA版本的安装命令 pip install torch2.1.0 torchvision0.16.0 --index-url https://download.pytorch.org/whl/cu118注意不要盲目装最新版。transformers对torch版本有要求bitsandbytes对CUDA版本有要求。先确定你要用的模型需要什么再倒推装什么版本。2.2 模型加载的几种姿势和显存账怎么算加载一个模型最直接的方式是用from_pretrained。但这里有几个关键参数决定了你能不能跑起来from transformers import AutoModelForCausalLM, AutoTokenizer model_name meta-llama/Llama-2-7b-hf # 方式一全精度加载7B模型约需28GB显存FP32 # model AutoModelForCausalLM.from_pretrained(model_name, torch_dtypetorch.float32) # 方式二半精度加载约14GB显存 model AutoModelForCausalLM.from_pretrained( model_name, torch_dtypetorch.float16, device_mapauto # 自动分配到可用GPU ) # 方式三8bit量化约7GB显存 # model AutoModelForCausalLM.from_pretrained( # model_name, # load_in_8bitTrue, # device_mapauto # ) # 方式四4bit量化约3.5GB显存 # model AutoModelForCausalLM.from_pretrained( # model_name, # load_in_4bitTrue, # bnb_4bit_compute_dtypetorch.float16, # device_mapauto # )显存估算有个粗略公式参数量 × 精度字节数 × 1.2额外开销。7B模型FP16就是70亿 × 2字节 × 1.2 ≈ 16.8GB。所以一张24GB的卡跑7B FP16是够的但跑13B就悬了。我实测下来4bit量化在大多数对话任务上效果损失很小但显存占用直接降到原来的四分之一。如果你的场景对精度要求不是极端苛刻量化是性价比最高的选择。2.3 推理速度优化的三个杠杆模型能加载只是第一步推理速度才是决定能不能上生产的关键。影响推理速度的核心因素有三个批处理大小、KV Cache、以及是否用了推理加速框架。批处理很好理解一次处理多个请求比一个一个处理吞吐量高得多。但批处理会增加显存占用和单次延迟需要根据业务场景权衡。KV Cache是自回归生成中的关键优化。没有KV Cache的话每生成一个token都要重新计算前面所有token的Key和Value矩阵计算量随序列长度平方增长。开启KV Cache后前面算过的直接复用计算量降到线性。# 生成时开启KV Cachetransformers默认开启 outputs model.generate( input_ids, max_new_tokens256, do_sampleTrue, temperature0.7, use_cacheTrue # 确保KV Cache开启 )再往上走就是vLLM、TGI这类推理框架了。它们实现了PagedAttention、连续批处理等高级优化吞吐量能比原生transformers高几倍到十几倍。但引入框架也意味着额外的学习成本和部署复杂度。我的建议是先用原生transformers跑通确认业务逻辑没问题再考虑上推理框架。3. 数据管道AI系统里最容易被低估的环节3.1 为什么数据管道决定了AI系统的上限模型是引擎数据是燃料。燃料质量不行引擎再好也跑不快。我见过太多项目模型选得很讲究但数据管道一塌糊涂——训练数据格式不统一、推理时的输入预处理和训练时不一致、数据版本没有管理出了问题根本不知道是哪批数据导致的。一个健壮的数据管道应该包含这几个环节数据采集、清洗、格式化、版本管理、以及训练/推理一致性校验。数据采集阶段最容易出的问题是“拿到什么用什么”。比如做文本分类正负样本比例严重失衡模型训出来全预测多数类。这时候需要做重采样或者加权损失。做生成任务训练数据里混入了大量低质量文本模型就会学会说废话。清洗环节我一般会做这几件事去除HTML标签、统一编码格式、过滤过短或过长的样本、去重。去重特别重要重复数据会让模型过拟合到特定模式。3.2 训练和推理的数据一致性怎么保证这是最隐蔽的坑。训练的时候你对文本做了某种归一化推理的时候忘了做模型效果直接崩掉。比如训练时把所有数字替换成了特殊token推理时没替换模型就懵了。我的做法是把预处理逻辑封装成一个独立的模块训练和推理都调用同一个函数class TextPreprocessor: def __init__(self, max_length512): self.max_length max_length def clean(self, text): # 去除多余空白 text re.sub(r\s, , text).strip() # 统一标点 text text.replace(“, ).replace(”, ) return text def tokenize(self, text, tokenizer): text self.clean(text) return tokenizer( text, max_lengthself.max_length, truncationTrue, paddingmax_length, return_tensorspt )训练脚本和推理服务都import这个类就不会出现不一致的问题。3.3 数据版本管理别再用文件名区分了train_data_v2_final_真的最终版.json——这种命名方式我猜很多人都用过。问题是过了一个月你自己都不记得v2和v3有什么区别。轻量级的做法是用DVC或者Git LFS管理数据版本每次数据变更对应一个commit。如果不想引入额外工具至少要做到数据文件用内容哈希命名维护一个manifest文件记录每个版本的来源、处理脚本、和对应的模型版本。import hashlib import json def get_data_hash(filepath): with open(filepath, rb) as f: return hashlib.md5(f.read()).hexdigest()[:8] # 维护manifest manifest { version: 20240115, files: { train: {path: data/train.jsonl, hash: get_data_hash(data/train.jsonl)}, val: {path: data/val.jsonl, hash: get_data_hash(data/val.jsonl)} }, preprocessing: TextPreprocessor v1.2, model: llama-7b-finetune-v3 }这样出了问题可以快速定位是哪批数据、哪个处理脚本、哪个模型版本的组合导致的。4. 服务化与部署从脚本到能用的API4.1 为什么FastAPI是AI服务化的首选把模型跑起来和把模型变成服务是两回事。脚本是一次性的服务要考虑并发、错误处理、超时、限流、日志。Python生态里做AI服务化FastAPI是目前最顺手的选择——异步支持好、自动生成文档、类型校验强。一个最简的推理服务长这样from fastapi import FastAPI, HTTPException from pydantic import BaseModel import torch app FastAPI() class GenerateRequest(BaseModel): prompt: str max_tokens: int 256 temperature: float 0.7 class GenerateResponse(BaseModel): text: str tokens_generated: int # 全局加载模型避免每次请求重新加载 model None tokenizer None app.on_event(startup) async def load_model(): global model, tokenizer # 实际加载逻辑 pass app.post(/generate, response_modelGenerateResponse) async def generate(req: GenerateRequest): if model is None: raise HTTPException(status_code503, detailModel not loaded) inputs tokenizer(req.prompt, return_tensorspt).to(model.device) with torch.no_grad(): outputs model.generate( **inputs, max_new_tokensreq.max_tokens, temperaturereq.temperature, do_sampleTrue ) text tokenizer.decode(outputs[0], skip_special_tokensTrue) return GenerateResponse(texttext, tokens_generatedoutputs.shape[1])关键点模型在startup时加载一次全局复用。千万不要在每个请求里加载模型那样延迟会高到无法接受。4.2 并发处理异步不等于并行FastAPI的async def看起来能处理并发但Python的GIL意味着同一时刻只有一个线程在执行Python字节码。如果你的推理是CPU密集型的比如在CPU上跑模型async根本帮不上忙反而可能因为事件循环阻塞导致所有请求都变慢。正确的做法是把推理放到线程池或进程池里执行from concurrent.futures import ThreadPoolExecutor import asyncio executor ThreadPoolExecutor(max_workers4) app.post(/generate) async def generate(req: GenerateRequest): loop asyncio.get_event_loop() # 把阻塞的推理放到线程池 result await loop.run_in_executor( executor, sync_generate, req.prompt, req.max_tokens ) return result如果是GPU推理情况又不一样。GPU本身能并行处理多个请求但需要框架层面的支持比如vLLM的连续批处理。自己用transformers写的话简单的做法是用一个队列把请求攒起来凑够一个batch再一起推理。4.3 健康检查、日志和优雅关闭生产服务必须有的几个东西健康检查接口、结构化日志、优雅关闭。健康检查让负载均衡器知道你的服务还活着app.get(/health) async def health(): return {status: ok, model_loaded: model is not None}日志不要用print用logging模块输出JSON格式方便后续采集和分析import logging import json logger logging.getLogger(ai-service) def log_request(request_id, prompt_len, latency_ms, status): logger.info(json.dumps({ request_id: request_id, prompt_len: prompt_len, latency_ms: latency_ms, status: status }))优雅关闭是指收到终止信号时先把正在处理的请求处理完再退出而不是直接杀掉进程import signal import sys def graceful_shutdown(signum, frame): logger.info(Received shutdown signal, waiting for ongoing requests...) # 停止接受新请求等待现有请求完成 sys.exit(0) signal.signal(signal.SIGTERM, graceful_shutdown)5. 监控与迭代上线只是开始5.1 推理服务必须监控的四个指标服务上线之后你需要知道它到底跑得怎么样。最核心的四个指标延迟P50/P95/P99、吞吐量QPS、错误率、GPU利用率。延迟不能只看平均值P99延迟才反映用户体验的下限。如果P99延迟是10秒意味着每100个请求就有1个用户等了10秒这个体验是很差的。GPU利用率反映了资源是否浪费。如果GPU利用率长期低于30%说明要么请求量不够要么批处理没做好。如果长期高于90%说明该加卡了。这些指标可以用Prometheus采集Grafana展示。如果不想搭这套至少把指标打到日志里用简单的脚本做聚合分析。5.2 模型效果监控比系统指标更难系统指标只能告诉你服务是否正常不能告诉你模型输出是否变差了。模型效果监控要复杂得多因为“好”和“坏”没有绝对标准。我的做法是维护一个小的评估集定期比如每天用当前模型跑一遍计算关键指标准确率、BLEU、ROUGE等和基线对比。如果指标下降超过阈值就告警。另外对生成任务可以监控输出长度分布、重复率、以及是否包含敏感词。这些指标异常往往意味着模型出了问题。def check_output_quality(texts): metrics {} # 平均长度 metrics[avg_length] sum(len(t.split()) for t in texts) / len(texts) # 重复率连续重复的n-gram比例 metrics[repetition_rate] calculate_repetition(texts) # 敏感词命中 metrics[sensitive_hits] count_sensitive(texts) return metrics5.3 迭代闭环从bad case到模型更新用户反馈和bad case是改进模型最宝贵的资源。我一般会做这几件事第一把所有低置信度或用户明确反馈不好的样本存下来定期人工标注。第二分析bad case的模式是数据问题、模型问题、还是prompt问题。第三针对性地补充训练数据或调整推理策略。这个闭环跑起来之后模型效果会持续提升。最怕的是上线之后就不管了效果慢慢退化都不知道。6. 一些踩过的坑和对应的解法6.1 显存碎片化导致OOM明明显存还有余量但就是报OOM。这通常是显存碎片化导致的。PyTorch的缓存分配器会预留显存但碎片化之后没有连续的大块内存可用。解法是设置PYTORCH_CUDA_ALLOC_CONFexpandable_segments:True让分配器能动态扩展内存段。或者定期调用torch.cuda.empty_cache()但这会降低性能不建议频繁调用。6.2 模型输出格式不稳定让模型输出JSON有时候能正常解析有时候多一个逗号就崩了。解法是在prompt里明确格式要求加上few-shot示例然后在解析时做容错处理——用正则提取JSON部分或者用json5这种宽松的解析器。更稳的做法是约束解码比如用outlines或guidance这类库在解码时强制模型只能输出符合语法规则的token。但这会增加推理开销需要权衡。6.3 多卡推理的坑用device_mapauto做多卡推理时模型会被切分到不同卡上。如果切分不合理卡间通信会成为瓶颈速度反而比单卡慢。我的经验是7B以下的模型尽量单卡跑用4bit量化塞进一张卡。13B以上再考虑多卡而且最好用张量并行而不是流水线并行因为流水线并行会有气泡等待。6.4 版本升级导致的兼容性问题transformers从4.30升到4.35generate函数的某些参数行为变了原本能跑的代码报warning甚至报错。AI领域的库迭代太快升级之前一定要在测试环境验证。我的做法是生产环境锁定所有依赖的版本号用pip freeze requirements.txt固定。升级时单独开分支测试确认没问题再合并。7. 从零搭建的完整路径回顾把上面这些串起来一个从零开始的AI工程能力建设路径大概是这样的第一步搭好环境确认GPU可用装好PyTorch和transformers。第二步跑通一个最小推理demo理解模型加载、tokenization、生成的基本流程。第三步构建数据管道把预处理逻辑封装成可复用的模块。第四步用FastAPI把推理封装成服务加上健康检查和日志。第五步做性能优化上量化、批处理、KV Cache。第六步搭监控采集系统指标和模型效果指标。第七步建立迭代闭环持续收集bad case并改进。每一步都不复杂但合在一起就是一个完整的AI工程系统。我自己的体会是最难的不是某个具体技术点而是把这些环节串起来时遇到的各种“意外”——环境冲突、数据不一致、性能瓶颈、线上故障。这些只有真正动手做过才会遇到也只有在解决过程中才能真正理解。如果你正在走这条路我的建议是不要追求一步到位。先把最小可用版本跑起来然后一个环节一个环节地优化。每解决一个问题你对整个系统的理解就深一层。这个过程没有捷径但每一步都算数。 从零搭建AI工程能力：模型加载、推理优化与生产部署实战
1. 从零搭建AI工程能力为什么大多数人卡在第一步聊到AI工程很多人第一反应是“调包”——装个transformers拉个预训练模型跑通一个demo就觉得自己入门了。但真到了要自己搭一套能用的AI系统比如做一个检索增强生成的知识库问答、搭一个多模态内容审核流水线、或者部署一个能扛住并发请求的推理服务立刻就懵了。模型加载报错、显存不够、推理速度慢得离谱、输出格式不稳定、上线之后没人监控……这些问题没有一个能靠“pip install”解决。ai-engineering-from-scratch这个标题核心讲的就是这件事不依赖现成的平台和封装好的黑盒工具从最底层开始把AI工程涉及的关键环节一个一个搭起来。它适合那些已经会写Python、了解基本深度学习概念但没真正独立做过完整AI系统的人。也适合那些平时只用过云端API、想搞清楚背后到底发生了什么的人。我自己的经历是最早做AI应用的时候全部依赖在线接口觉得方便。直到有一次需要处理一批敏感数据不能出本地环境才发现自己根本不知道怎么在本地跑一个像样的模型。从那之后我开始系统性地补AI工程的基础设施知识从模型加载、推理优化、服务封装到监控告警一步一步踩过来。这篇文章就把这条路径上的关键节点拆开讲清楚每个环节都给出可操作的方案和背后的取舍逻辑。2. 模型加载与推理环境别让环境问题吃掉你一半时间2.1 为什么虚拟环境不是可选项而是必选项AI工程的依赖冲突是出了名的严重。PyTorch、CUDA、cuDNN、transformers、accelerate、bitsandbytes这几个库之间的版本兼容关系能让人抓狂。我见过太多人因为直接在系统Python里装包导致原本能跑的代码突然报undefined symbol错误。正确的做法是每个项目一个独立环境。用conda还是venv我的建议是涉及CUDA和深度学习框架的优先用conda因为conda能管理非Python的二进制依赖比如CUDA runtime。纯Python的轻量项目用venv就够了。# 创建conda环境指定Python版本 conda create -n ai-eng python3.10 -y conda activate ai-eng # 安装PyTorch时指定CUDA版本不要直接pip install torch # 去PyTorch官网查对应CUDA版本的安装命令 pip install torch2.1.0 torchvision0.16.0 --index-url https://download.pytorch.org/whl/cu118注意不要盲目装最新版。transformers对torch版本有要求bitsandbytes对CUDA版本有要求。先确定你要用的模型需要什么再倒推装什么版本。2.2 模型加载的几种姿势和显存账怎么算加载一个模型最直接的方式是用from_pretrained。但这里有几个关键参数决定了你能不能跑起来from transformers import AutoModelForCausalLM, AutoTokenizer model_name meta-llama/Llama-2-7b-hf # 方式一全精度加载7B模型约需28GB显存FP32 # model AutoModelForCausalLM.from_pretrained(model_name, torch_dtypetorch.float32) # 方式二半精度加载约14GB显存 model AutoModelForCausalLM.from_pretrained( model_name, torch_dtypetorch.float16, device_mapauto # 自动分配到可用GPU ) # 方式三8bit量化约7GB显存 # model AutoModelForCausalLM.from_pretrained( # model_name, # load_in_8bitTrue, # device_mapauto # ) # 方式四4bit量化约3.5GB显存 # model AutoModelForCausalLM.from_pretrained( # model_name, # load_in_4bitTrue, # bnb_4bit_compute_dtypetorch.float16, # device_mapauto # )显存估算有个粗略公式参数量 × 精度字节数 × 1.2额外开销。7B模型FP16就是70亿 × 2字节 × 1.2 ≈ 16.8GB。所以一张24GB的卡跑7B FP16是够的但跑13B就悬了。我实测下来4bit量化在大多数对话任务上效果损失很小但显存占用直接降到原来的四分之一。如果你的场景对精度要求不是极端苛刻量化是性价比最高的选择。2.3 推理速度优化的三个杠杆模型能加载只是第一步推理速度才是决定能不能上生产的关键。影响推理速度的核心因素有三个批处理大小、KV Cache、以及是否用了推理加速框架。批处理很好理解一次处理多个请求比一个一个处理吞吐量高得多。但批处理会增加显存占用和单次延迟需要根据业务场景权衡。KV Cache是自回归生成中的关键优化。没有KV Cache的话每生成一个token都要重新计算前面所有token的Key和Value矩阵计算量随序列长度平方增长。开启KV Cache后前面算过的直接复用计算量降到线性。# 生成时开启KV Cachetransformers默认开启 outputs model.generate( input_ids, max_new_tokens256, do_sampleTrue, temperature0.7, use_cacheTrue # 确保KV Cache开启 )再往上走就是vLLM、TGI这类推理框架了。它们实现了PagedAttention、连续批处理等高级优化吞吐量能比原生transformers高几倍到十几倍。但引入框架也意味着额外的学习成本和部署复杂度。我的建议是先用原生transformers跑通确认业务逻辑没问题再考虑上推理框架。3. 数据管道AI系统里最容易被低估的环节3.1 为什么数据管道决定了AI系统的上限模型是引擎数据是燃料。燃料质量不行引擎再好也跑不快。我见过太多项目模型选得很讲究但数据管道一塌糊涂——训练数据格式不统一、推理时的输入预处理和训练时不一致、数据版本没有管理出了问题根本不知道是哪批数据导致的。一个健壮的数据管道应该包含这几个环节数据采集、清洗、格式化、版本管理、以及训练/推理一致性校验。数据采集阶段最容易出的问题是“拿到什么用什么”。比如做文本分类正负样本比例严重失衡模型训出来全预测多数类。这时候需要做重采样或者加权损失。做生成任务训练数据里混入了大量低质量文本模型就会学会说废话。清洗环节我一般会做这几件事去除HTML标签、统一编码格式、过滤过短或过长的样本、去重。去重特别重要重复数据会让模型过拟合到特定模式。3.2 训练和推理的数据一致性怎么保证这是最隐蔽的坑。训练的时候你对文本做了某种归一化推理的时候忘了做模型效果直接崩掉。比如训练时把所有数字替换成了特殊token推理时没替换模型就懵了。我的做法是把预处理逻辑封装成一个独立的模块训练和推理都调用同一个函数class TextPreprocessor: def __init__(self, max_length512): self.max_length max_length def clean(self, text): # 去除多余空白 text re.sub(r\s, , text).strip() # 统一标点 text text.replace(“, ).replace(”, ) return text def tokenize(self, text, tokenizer): text self.clean(text) return tokenizer( text, max_lengthself.max_length, truncationTrue, paddingmax_length, return_tensorspt )训练脚本和推理服务都import这个类就不会出现不一致的问题。3.3 数据版本管理别再用文件名区分了train_data_v2_final_真的最终版.json——这种命名方式我猜很多人都用过。问题是过了一个月你自己都不记得v2和v3有什么区别。轻量级的做法是用DVC或者Git LFS管理数据版本每次数据变更对应一个commit。如果不想引入额外工具至少要做到数据文件用内容哈希命名维护一个manifest文件记录每个版本的来源、处理脚本、和对应的模型版本。import hashlib import json def get_data_hash(filepath): with open(filepath, rb) as f: return hashlib.md5(f.read()).hexdigest()[:8] # 维护manifest manifest { version: 20240115, files: { train: {path: data/train.jsonl, hash: get_data_hash(data/train.jsonl)}, val: {path: data/val.jsonl, hash: get_data_hash(data/val.jsonl)} }, preprocessing: TextPreprocessor v1.2, model: llama-7b-finetune-v3 }这样出了问题可以快速定位是哪批数据、哪个处理脚本、哪个模型版本的组合导致的。4. 服务化与部署从脚本到能用的API4.1 为什么FastAPI是AI服务化的首选把模型跑起来和把模型变成服务是两回事。脚本是一次性的服务要考虑并发、错误处理、超时、限流、日志。Python生态里做AI服务化FastAPI是目前最顺手的选择——异步支持好、自动生成文档、类型校验强。一个最简的推理服务长这样from fastapi import FastAPI, HTTPException from pydantic import BaseModel import torch app FastAPI() class GenerateRequest(BaseModel): prompt: str max_tokens: int 256 temperature: float 0.7 class GenerateResponse(BaseModel): text: str tokens_generated: int # 全局加载模型避免每次请求重新加载 model None tokenizer None app.on_event(startup) async def load_model(): global model, tokenizer # 实际加载逻辑 pass app.post(/generate, response_modelGenerateResponse) async def generate(req: GenerateRequest): if model is None: raise HTTPException(status_code503, detailModel not loaded) inputs tokenizer(req.prompt, return_tensorspt).to(model.device) with torch.no_grad(): outputs model.generate( **inputs, max_new_tokensreq.max_tokens, temperaturereq.temperature, do_sampleTrue ) text tokenizer.decode(outputs[0], skip_special_tokensTrue) return GenerateResponse(texttext, tokens_generatedoutputs.shape[1])关键点模型在startup时加载一次全局复用。千万不要在每个请求里加载模型那样延迟会高到无法接受。4.2 并发处理异步不等于并行FastAPI的async def看起来能处理并发但Python的GIL意味着同一时刻只有一个线程在执行Python字节码。如果你的推理是CPU密集型的比如在CPU上跑模型async根本帮不上忙反而可能因为事件循环阻塞导致所有请求都变慢。正确的做法是把推理放到线程池或进程池里执行from concurrent.futures import ThreadPoolExecutor import asyncio executor ThreadPoolExecutor(max_workers4) app.post(/generate) async def generate(req: GenerateRequest): loop asyncio.get_event_loop() # 把阻塞的推理放到线程池 result await loop.run_in_executor( executor, sync_generate, req.prompt, req.max_tokens ) return result如果是GPU推理情况又不一样。GPU本身能并行处理多个请求但需要框架层面的支持比如vLLM的连续批处理。自己用transformers写的话简单的做法是用一个队列把请求攒起来凑够一个batch再一起推理。4.3 健康检查、日志和优雅关闭生产服务必须有的几个东西健康检查接口、结构化日志、优雅关闭。健康检查让负载均衡器知道你的服务还活着app.get(/health) async def health(): return {status: ok, model_loaded: model is not None}日志不要用print用logging模块输出JSON格式方便后续采集和分析import logging import json logger logging.getLogger(ai-service) def log_request(request_id, prompt_len, latency_ms, status): logger.info(json.dumps({ request_id: request_id, prompt_len: prompt_len, latency_ms: latency_ms, status: status }))优雅关闭是指收到终止信号时先把正在处理的请求处理完再退出而不是直接杀掉进程import signal import sys def graceful_shutdown(signum, frame): logger.info(Received shutdown signal, waiting for ongoing requests...) # 停止接受新请求等待现有请求完成 sys.exit(0) signal.signal(signal.SIGTERM, graceful_shutdown)5. 监控与迭代上线只是开始5.1 推理服务必须监控的四个指标服务上线之后你需要知道它到底跑得怎么样。最核心的四个指标延迟P50/P95/P99、吞吐量QPS、错误率、GPU利用率。延迟不能只看平均值P99延迟才反映用户体验的下限。如果P99延迟是10秒意味着每100个请求就有1个用户等了10秒这个体验是很差的。GPU利用率反映了资源是否浪费。如果GPU利用率长期低于30%说明要么请求量不够要么批处理没做好。如果长期高于90%说明该加卡了。这些指标可以用Prometheus采集Grafana展示。如果不想搭这套至少把指标打到日志里用简单的脚本做聚合分析。5.2 模型效果监控比系统指标更难系统指标只能告诉你服务是否正常不能告诉你模型输出是否变差了。模型效果监控要复杂得多因为“好”和“坏”没有绝对标准。我的做法是维护一个小的评估集定期比如每天用当前模型跑一遍计算关键指标准确率、BLEU、ROUGE等和基线对比。如果指标下降超过阈值就告警。另外对生成任务可以监控输出长度分布、重复率、以及是否包含敏感词。这些指标异常往往意味着模型出了问题。def check_output_quality(texts): metrics {} # 平均长度 metrics[avg_length] sum(len(t.split()) for t in texts) / len(texts) # 重复率连续重复的n-gram比例 metrics[repetition_rate] calculate_repetition(texts) # 敏感词命中 metrics[sensitive_hits] count_sensitive(texts) return metrics5.3 迭代闭环从bad case到模型更新用户反馈和bad case是改进模型最宝贵的资源。我一般会做这几件事第一把所有低置信度或用户明确反馈不好的样本存下来定期人工标注。第二分析bad case的模式是数据问题、模型问题、还是prompt问题。第三针对性地补充训练数据或调整推理策略。这个闭环跑起来之后模型效果会持续提升。最怕的是上线之后就不管了效果慢慢退化都不知道。6. 一些踩过的坑和对应的解法6.1 显存碎片化导致OOM明明显存还有余量但就是报OOM。这通常是显存碎片化导致的。PyTorch的缓存分配器会预留显存但碎片化之后没有连续的大块内存可用。解法是设置PYTORCH_CUDA_ALLOC_CONFexpandable_segments:True让分配器能动态扩展内存段。或者定期调用torch.cuda.empty_cache()但这会降低性能不建议频繁调用。6.2 模型输出格式不稳定让模型输出JSON有时候能正常解析有时候多一个逗号就崩了。解法是在prompt里明确格式要求加上few-shot示例然后在解析时做容错处理——用正则提取JSON部分或者用json5这种宽松的解析器。更稳的做法是约束解码比如用outlines或guidance这类库在解码时强制模型只能输出符合语法规则的token。但这会增加推理开销需要权衡。6.3 多卡推理的坑用device_mapauto做多卡推理时模型会被切分到不同卡上。如果切分不合理卡间通信会成为瓶颈速度反而比单卡慢。我的经验是7B以下的模型尽量单卡跑用4bit量化塞进一张卡。13B以上再考虑多卡而且最好用张量并行而不是流水线并行因为流水线并行会有气泡等待。6.4 版本升级导致的兼容性问题transformers从4.30升到4.35generate函数的某些参数行为变了原本能跑的代码报warning甚至报错。AI领域的库迭代太快升级之前一定要在测试环境验证。我的做法是生产环境锁定所有依赖的版本号用pip freeze requirements.txt固定。升级时单独开分支测试确认没问题再合并。7. 从零搭建的完整路径回顾把上面这些串起来一个从零开始的AI工程能力建设路径大概是这样的第一步搭好环境确认GPU可用装好PyTorch和transformers。第二步跑通一个最小推理demo理解模型加载、tokenization、生成的基本流程。第三步构建数据管道把预处理逻辑封装成可复用的模块。第四步用FastAPI把推理封装成服务加上健康检查和日志。第五步做性能优化上量化、批处理、KV Cache。第六步搭监控采集系统指标和模型效果指标。第七步建立迭代闭环持续收集bad case并改进。每一步都不复杂但合在一起就是一个完整的AI工程系统。我自己的体会是最难的不是某个具体技术点而是把这些环节串起来时遇到的各种“意外”——环境冲突、数据不一致、性能瓶颈、线上故障。这些只有真正动手做过才会遇到也只有在解决过程中才能真正理解。如果你正在走这条路我的建议是不要追求一步到位。先把最小可用版本跑起来然后一个环节一个环节地优化。每解决一个问题你对整个系统的理解就深一层。这个过程没有捷径但每一步都算数。