搞AI工程这行久了,我越来越发现一个现象:不少朋友拿着模型跑通个demo,就觉得自己已经“入行了”。但真到了生产环境,数据一乱、并发一高、模型一更新,整个系统直接“飞盘”。原因很简单——AI工程不是“跑通一个模型”,而是“让模型稳定地、可维护地、规模化地跑在业务里”。这中间隔着数据管理、特征工程、模型服务化、监控反馈、CI/CD一整套体系,而这恰恰是“ai-engineering-from-scratch”这个方向最值得深耕的东西。
写这篇东西的初衷,是我自己从2019年开始踩坑、填坑,把一套从零到一的AI工程链路走通了。今天把这些经验整理成一篇可以“抄作业”的完整指南,适合刚入门想系统学习AI工程的同学,也适合已经在做算法但总被部署、运维、数据问题折磨的工程师。我不讲那些高大上的空理论,全部落在实操上:环境怎么搭、数据管道怎么设计、模型怎么上线、上线之后怎么监控,以及LLM时代AI工程多了哪些新玩法。
1. AI工程的核心能力矩阵:别把“机器学习”和“AI工程”混为一谈
先说一个最常见的认知误区。很多人觉得AI工程就是机器学习,但这两个东西的侧重点完全不同。机器学习关注“怎么训练出一个高精度的模型”,核心指标是准确率、召回率、F1这些。AI工程关注的是“这个模型怎么在真实业务中长期稳定运行”,核心指标变成了可用性、延迟、吞吐量、可维护性、可观测性。
我打一个比方。机器学习像是厨师研究出一道拿手菜,AI工程则是把这道菜做成连锁店的标准品:食材供应链要稳定(数据管道)、每口锅的火候要可控(模型参数管理)、出菜速度要达标(推理延迟)、顾客反馈要能回流到后厨(监控与迭代)。少一环,店就开不下去。
所以AI工程的能力矩阵,我习惯拆成六层:
- 数据层:采集、清洗、校验、版本化、特征存储。数据是地基,这层做不好,上层模型再强也是空中楼阁。
- 训练层:实验跟踪、超参调优、模型评估、模型注册。训练不是简单跑一个脚本,而是要有可复现的实验体系。
- 服务层:模型推理服务的封装,包括同步接口、异步任务、批处理、流式输出等不同的服务形态。
- 部署层:容器化、镜像管理、灰度发布、弹性伸缩。模型也要像普通服务一样走完整的发布流程。
- 监控层:业务指标监控、模型性能监控(数据漂移、概念漂移)、服务健康监控、日志追踪。
- 编排层:把上面五层串起来的流水线,也就是AI系统里的CI/CD/CT,让模型从开发到上线全流程自动化。
值得强调的是,这六层不是割裂的,而是一个闭环。数据层的问题会传导到训练层,训练层的问题会暴露在服务层,服务层的问题要靠监控层发现,监控层的反馈又回流到数据层和训练层去优化。这就像人体循环系统,任何一环堵了,整个体系都会出问题。我在设计AI平台的时候,第一件事不是选模型,而是先画这个闭环,把每个环节的负责人和交接边界定清楚。
2. 从零搭建一套AI工程架构:一次生产级系统设计拆解
很多教程直接给你一段训练代码,然后就完了。但真实的AI工程系统远比这复杂。我以自己做过的一个智能内容审核系统为例,带大家拆一遍生产级AI系统长什么样,这样你再看那些零散的工具和技巧时,就能知道它们各自在架构里扮演什么角色。
这套系统的核心需求是:对用户上传的文本和图片内容进行实时审核,识别违规内容,处理峰值每秒上百次的请求,而且要支持模型每周更新迭代而不影响线上服务。
2.1 整体架构全景:从请求到响应的完整链路
整个系统的链路可以分成以下几个关键环节:
用户请求进来后,首先经过API网关做鉴权、限流和路由。网关的目的是挡住无效流量,避免下游服务被冲垮。限流的策略我习惯用令牌桶算法,每秒允许1000个请求进入,超出部分直接返回“系统繁忙”的提示,而不是让请求堆积在服务端拖垮整个系统。
请求进入推理服务之前,会先经过一个特征预处理组件。这个组件负责把原始文本清洗、分词、向量化,把图片做缩放、归一化。这里有一个容易忽略的点:训练时的预处理逻辑必须和线上推理时的预处理逻辑完全一致,否则模型拿到的数据分布不一样,效果就会断崖式下跌。为了解决这个问题,我一般会把预处理逻辑打成独立的Python包,训练和推理共用同一份代码。
推理服务本身是多个模型集成:文本模型负责文本内容识别,图像模型负责图片识别,还有一个规则引擎做兜底(比如一些明确违禁的词汇列表直接命中)。这里要强调一下,规则引擎不是多余的,它对高置信度的简单case响应速度极快,成本极低,同时还能给模型预测结果做交叉验证,降低误判率。
推理结果出来后,会写入消息队列,由后端的异步任务进行人工复审队列调度。对于模型判断为“疑似违规”的内容,不直接删除,而是进入人工审核队列。这一步在业务上是必须的——模型可以帮你筛选99%的明显case,但剩下1%的边缘case,需要人来拍板。
最后,所有的请求日志、预测结果、特征数据都会写入数据湖和监控系统,形成完整的审计链路和反馈闭环。每周的训练任务会从数据湖里抽取这一周新增的“人审确认样本”,混入历史训练集,重新训练模型,然后通过自动评估门禁后灰度上线。
2.2 关键设计决策:为什么这样选型
说完了链路,我重点讲讲几个关键的设计决策,以及背后的思考过程。
第一个是同步还是异步。内容审核这种场景,用户上传内容后在线等待审核结果,所以审核链路必须是同步的,要在一两秒内返回结果。但人工复审是异步的,不影响主流程。所以我设计了“同步推理 + 异步复审”的双轨制。如果你的业务对实时性不敏感(比如离线批处理分析历史数据),同步接口反而会成为瓶颈,这时候就改用异步任务加消息队列批处理,吞吐量能提高一个数量级。
第二个是单体服务还是微服务。我见过很多团队一上来就拆微服务:数据服务、特征服务、模型服务、监控服务各拆一套,最后维护成本比开发成本还高。我的经验是,在规模没到“必须拆”的程度之前,先用模块化单体:代码上分层清晰(路由层、业务层、模型推理层、存储层),但部署上是一个服务。这样部署成本低、调试方便,等流量确实大了,再按性能瓶颈点局部拆出独立服务,而不是一开始就全面微服务化。
第三个是模型推理用CPU还是GPU。很多人觉得AI服务必须上GPU,其实取决于模型类型和延迟要求。像BERT-base这种模型,在CPU上用ONNX Runtime做量化推理,单条延迟能到30~50ms,已经能满足大多数业务需求,而且CPU部署成本低、弹性扩容更容易。只有大模型(LLM)或者超大Batch的CV模型才建议上GPU。我在设计这套系统时,文本审核走CPU+量化,图像审核走GPU,两种资源分离部署,成本和性能都兼顾到了。
2.3 数据管道与特征工程:容易被低估的工程重心
数据是AI工程里最苦最累但价值最高的一环。我在实际项目中,数据准备和特征工程的时间往往占到整个项目周期的60%以上。很多模型效果不好,不是算法不行,而是数据没喂对。
数据管道的核心是数据版本化。训练集、验证集、测试集必须像代码一样有版本号,模型训练时明确记录用的是哪个版本的数据。不然就会出现这种情况:三个月前训练了一个模型,效果还不错,但后来想复现,已经忘了当时用的哪一批数据、哪些清洗规则。我用DVC(Data Version Control)管理数据版本,每个数据集目录下都有一个.dvc文件,记录文件哈希和存储位置,配合Git就能像管理代码一样管理数据。
特征工程这块,要点是“训练-服务一致性”。离线训练时你有整张表的数据可以做统计特征(比如用户历史行为均值),但线上推理时你只有当前这一个请求的数据,那些需要全量数据的特征根本算不出来。这个问题处理不好,线上效果一定会比离线评估差很多。我的做法是:把特征分成在线特征和离线特征两类。在线特征从特征存储(Feature Store)里读取,离线特征在训练时也是从同一个特征存储读取,两边代码和生产逻辑完全一致,杜绝“训练服务不一致”的问题。
3. 环境搭建:从Python版本到GPU驱动一次讲清楚
这一章节写给从零开始的朋友,老手也可以看看有没有你忽略的细节。AI工程的第一步不是写模型,而是把一个干净、可复现的开发环境搭起来。这一步踩坑的人太多了,而且坑都很低级,浪费的时间完全可以避免。
3.1 Python环境管理:别用系统Python裸奔
我这几年强烈推荐用Miniconda管理Python环境,不要直接改系统自带的Python。原因是不同项目对Python版本、依赖库版本的要求经常冲突,比如项目A需要TensorFlow 2.4(支持Python 3.7~3.9),项目B需要PyTorch 2.0(推荐Python 3.10+),用全局Python就是灾难。
创建专用环境的命令很简单:
# 安装Miniconda后执行 conda create -n ai-eng python=3.10 conda activate ai-eng # 国内用户建议先配置镜像源,速度提升明显 conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main/ conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/free/ conda config --set show_channel_urls yes # 安装核心库 pip install numpy pandas scikit-learn matplotlib jupyter pip install torch --index-url https://download.pytorch.org/whl/cu118 pip install transformers datasets accelerate这里有一个新手经常掉进去的坑:pip和conda混用导致依赖解析混乱。我的默认策略是环境创建和Python版本管理用conda,Python包安装统一用pip。不要在conda环境里频繁用conda install去装包,因为conda的依赖解析器有时候会把一些包降级,和pip装的包产生冲突。
3.2 GPU环境配置:验证CUDA和cuDNN是否真的可用
如果你要做深度学习训练,GPU是刚需。但很多人在GPU环境上反复折腾,装完驱动又装CUDA,装完CUDA又发现PyTorch用不了。问题往往出在没搞清楚驱动、CUDA Toolkit、cuDNN、深度学习框架四者之间的兼容关系。
简单来说:
- NVIDIA驱动:系统级的,负责GPU硬件通信,用
nvidia-smi查看版本。 - CUDA Toolkit:开发库,编译GPU代码用的。
- cuDNN:深度神经网络的加速库,依赖CUDA。
- PyTorch/TensorFlow:框架本身带了部分CUDA运行时,不一定需要单独装CUDA Toolkit。
我现在的做法是:先装NVIDIA驱动(一般通过sudo apt install nvidia-driver-545或从官网装),然后用nvidia-smi看驱动支持的CUDA版本,最后直接安装对应CUDA版本的PyTorch。PyTorch官方安装命令里的cu118、cu121就指定了不同的CUDA版本。
装完之后,不要只看import torch是否报错,要真正跑一个GPU计算验证:
import torch print(torch.cuda.is_available()) # 必须为True print(torch.cuda.get_device_name(0)) # 显示你的显卡型号 x = torch.randn(1000, 1000).cuda() y = torch.matmul(x, x) # 实际跑一次矩阵乘法 print(y.sum().item())如果torch.cuda.is_available()返回False,大概率是PyTorch的CUDA版本和驱动不匹配,或者CUDA运行时库缺失。可以用nvidia-smi里的“CUDA Version”作为上限参考,选择不高于这个版本的PyTorch CUDA版本。
3.3 基础设施:Docker和Git的初始化配置
环境搭建里还有一个容易被忽略但非常重要的环节:Docker。AI工程从第一天就应该把环境容器化,而不是等部署的时候再开始学Docker。因为Docker保证的是“可复现的环境”,你的代码换一台机器跑,结果应该是一样的。
我的建议是在项目第一天就写好Dockerfile,哪怕先是一个简单版本:
FROM nvidia/cuda:11.8-runtime-ubuntu20.04 # 设置环境变量,避免交互式提问 ENV DEBIAN_FRONTEND=noninteractive RUN apt-get update && apt-get install -y \ python3.10 python3.10-dev 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", "main.py"]这样,不管是在自己笔记本上开发,还是推到服务器上跑训练,环境始终是一致的。另外,Git仓库从一开始就要规范:.gitignore里加好__pycache__/、*.pyc、.env、data/、models/这些目录,别把超大模型文件和数据文件提交进Git,这些应该走DVC或者对象存储。
4. 最小可行AI系统实操:从数据处理到模型打包
理论讲多了容易飘,现在落一个具体的实操例子。我们做一个中文情感分析系统,输入一段文本,输出正面或负面判断。这个例子的完整链路是:数据处理、模型微调、评估验证、模型打包。麻雀虽小,五脏俱全。
4.1 准备数据和微调模型
中文情感分析,我用Hugging Face的transformers库加一个开源中文预训练模型来做。数据集用ChnSentiCorp,这个数据集包含7000多条酒店评论,标注了正面/负面,非常适合入门。
数据加载和预处理的代码:
from datasets import load_dataset from transformers import AutoTokenizer # 加载数据集 dataset = load_dataset("chnsenti/corpus") # 加载中文预训练模型的分词器 tokenizer = AutoTokenizer.from_pretrained("bert-base-chinese") def tokenize_function(examples): return tokenizer( examples["text"], padding="max_length", truncation=True, max_length=128 ) # 对数据集做分词处理 tokenized_datasets = dataset.map(tokenize_function, batched=True) # 划分训练集和验证集 train_dataset = tokenized_datasets["train"].shuffle(seed=42).select(range(5000)) eval_dataset = tokenized_datasets["validation"].select(range(500))这里有一个值得展开的细节:为什么选bert-base-chinese而不是更大的模型?因为对这个任务来说,BERT-base已经够用了,推理速度快、显存占用低。先在小的模型上把流程跑通,再根据效果决定是否换更大的模型,这才是工程上应该有的节奏。另外,max_length=128是因为酒店评论不会有特别长的文本,设置太长反而浪费算力。
模型训练部分,我直接用transformers的Trainer封装,不用手写训练循环,省时省力:
from transformers import AutoModelForSequenceClassification, Trainer, TrainingArguments # 加载模型,二分类任务 model = AutoModelForSequenceClassification.from_pretrained( "bert-base-chinese", num_labels=2 ) # 定义训练参数 training_args = TrainingArguments( output_dir="./results", # 输出目录 evaluation_strategy="epoch", # 每个epoch结束后做评估 learning_rate=2e-5, # 学习率 per_device_train_batch_size=16, # 每块GPU的训练batch大小 per_device_eval_batch_size=64, # 评估batch大小 num_train_epochs=3, # 训练轮数 weight_decay=0.01, # 权重衰减 save_strategy="epoch", # 每个epoch存一次检查点 load_best_model_at_end=True, # 训练结束后加载最佳模型 metric_for_best_model="accuracy", # 以准确率为选择标准 ) # 定义评估指标 from sklearn.metrics import accuracy_score def compute_metrics(eval_pred): predictions, labels = eval_pred predictions = predictions.argmax(axis=-1) return {"accuracy": accuracy_score(labels, predictions)} # 创建Trainer并开始训练 trainer = Trainer( model=model, args=training_args, train_dataset=train_dataset, eval_dataset=eval_dataset, compute_metrics=compute_metrics, ) trainer.train()训练完成后,不要急着部署。先做一次完整评估,看看模型在验证集上的表现,同时看一眼错误样本,判断是不是有数据标注噪声或者预处理问题。我用下面的代码生成评估报告:
import numpy as np from sklearn.metrics import classification_report predictions = trainer.predict(eval_dataset) y_pred = np.argmax(predictions.predictions, axis=-1) y_true = predictions.label_ids print(classification_report(y_true, y_pred, target_names=["负面", "正面"]))输出类似:
precision recall f1-score support 负面 0.93 0.95 0.94 250 正面 0.95 0.93 0.94 250准确率到94%左右,对这个任务来说已经是可用的水平。如果你的精度不够,优先检查数据而不是调模型结构。比如看错误样本是不是标签标错了、是不是有很短的无效文本混进去了,清洗几轮数据往往比换大模型效果更明显。
4.2 模型保存与打包:不只是save_pretrained
模型训练完,用model.save_pretrained("./my_sentiment_model")和tokenizer.save_pretrained("./my_sentiment_model")保存。但这里我要多说两句:保存的内容本身很简单,真正的坑在于整个“模型目录”的规范化管理。
一个专业的模型目录应该包含:
pytorch_model.bin或model.safetensors:模型权重config.json:模型配置,包括类别数、dropout等tokenizer.json、vocab.txt:分词器文件label_map.json:标签映射,比如{"0": "负面", "1": "正面"}model_card.md:模型说明文档,记录训练时间、数据版本、评估指标preprocess.py:预处理代码,保证线上和训练时用同一套逻辑
{ "id": "sentiment_analysis_v1", "base_model": "bert-base-chinese", "dataset": "chnsenti/corpus@20240901", "metrics": {"accuracy": 0.94, "f1": 0.94}, "created_at": "2024-09-15", "runtime": "torch2.1.1+cu118" }这个“模型清单”非常重要。当你有多个模型版本上线后,没有清晰的元信息,根本分不清哪个模型是哪个任务、哪个版本、效果如何。我用一个简单的JSON文件记录这些信息,配合模型注册表统一管理。后续做模型回滚、对比评测时,这些元信息能节省大量排查时间。
5. 模型服务化上线:从FastAPI到Docker部署的完整路径
模型训练出来只是第一步,让模型变成线上可以调用的服务,才是AI工程的核心环节。这一章节带大家把训练好的模型包装成HTTP服务,再容器化部署。
5.1 用FastAPI封装模型推理服务
我强烈推荐用FastAPI做模型服务,它同时自带异步支持、数据校验和交互式API文档,对工程师来说非常友好。
模型服务的核心代码:
from fastapi import FastAPI, HTTPException from pydantic import BaseModel from transformers import AutoTokenizer, AutoModelForSequenceClassification import torch app = FastAPI(title="Sentiment Analysis API") # 模型加载(放在全局作用域,进程启动时只加载一次) tokenizer = AutoTokenizer.from_pretrained("./my_sentiment_model") model = AutoModelForSequenceClassification.from_pretrained("./my_sentiment_model") model.eval() # 切换到推理模式 # 因为部署环境可能没有GPU,这里改为CPU推理 device = torch.device("cuda" if torch.cuda.is_available() else "cpu") model.to(device) class TextRequest(BaseModel): text: str class PredictionResponse(BaseModel): label: str confidence: float @app.post("/predict", response_model=PredictionResponse) def predict(request: TextRequest): # 预处理 inputs = tokenizer( request.text, return_tensors="pt", truncation=True, max_length=128, padding="max_length" ) inputs = {k: v.to(device) for k, v in inputs.items()} # 推理 with torch.no_grad(): outputs = model(**inputs) probs = torch.softmax(outputs.logits, dim=-1) pred_class = torch.argmax(probs, dim=-1).item() confidence = probs[0][pred_class].item() label_map = {0: "负面", 1: "正面"} return PredictionResponse(label=label_map[pred_class], confidence=round(confidence, 4))注意几个工程细节:
第一,模型加载放在全局作用域,而不是在函数内部。如果放在每个请求里加载一次,服务会直接被拖垮——加载一个BERT-base模型需要几秒钟,而推理只需要几十毫秒。
第二,model.eval()必须显式调用,否则模型里的Dropout和BatchNorm层会跑在训练模式下,推理结果会不稳定。
第三,torch.no_grad()包裹推理过程,避免构建计算图占用额外的显存和内存。
本地跑一下服务:
uvicorn main:app --host 0.0.0.0 --port 8000然后访问http://localhost:8000/docs就能看到Swagger交互式API文档,可以直接在页面上测试接口。调用测试:
curl -X POST "http://localhost:8000/predict" \ -H "Content-Type: application/json" \ -d '{"text": "这个酒店的服务态度很好,房间也很干净"}'返回:
{"label": "正面", "confidence": 0.9912}5.2 性能优化:模型推理延迟是如何压低到50ms的
上面这个简单版本虽然能跑,但性能还不行。我在生产环境里做了几轮优化,把单次推理延迟从200ms左右降到了50ms以内。这里分享核心思路。
第一层优化是动态Batch。对于高并发的场景,把多个请求攒在一起放到一个Batch里推理,GPU或CPU的利用率会大幅提升。FastAPI本身是异步的,可以配合asyncio队列实现请求攒批。核心思路是:请求进来后不立即推理,而是放入队列等待一个极短的窗口(比如20ms),在这个窗口内到达的请求会被拼成一个Batch统一推理。虽然单请求延迟略有增加,但系统整体吞吐量可以提升3~5倍。
第二层优化是模型量化。把PyTorch模型转换成ONNX格式,然后做INT8量化。BERT模型的大部分参数是Embedding和线性层,INT8量化后模型体积缩小到原来的1/4,CPU推理速度提升2~3倍,精度损失通常控制在1个百分点以内。转换流程如下:
import torch from transformers import AutoModelForSequenceClassification model = AutoModelForSequenceClassification.from_pretrained("./my_sentiment_model") model.eval() # 构造一个样例输入 dummy_input = torch.randint(0, 1000, (1, 128), dtype=torch.int64) # 导出ONNX torch.onnx.export( model, dummy_input, "model.onnx", input_names=["input_ids"], output_names=["logits"], dynamic_axes={"input_ids": {0: "batch_size"}}, opset_version=13 )导出时设置dynamic_axes是为了支持可变Batch。然后用ONNX Runtime加载:
import onnxruntime as ort sess = ort.InferenceSession("model.onnx", providers=["CPUExecutionProvider"]) outputs = sess.run(None, {"input_ids": input_ids.numpy()})在我的测试里,量化后的ONNX模型在CPU上推理延迟从120ms降到45ms,对大多数业务来说已经足够。如果你的服务有GPU,直接用TensorRT或者FasterTransformer做优化,延迟能压到个位数毫秒级。
5.3 Docker化部署和服务上线
模型服务写好后,接下来做容器化。这里有一个常见的坑:镜像做得太大,动辄好几个GB,导致构建和拉取都很慢。我一般这样写Dockerfile:
FROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 只拷贝模型目录和主代码,不要把训练脚本和数据打进去 COPY main.py . COPY my_sentiment_model ./my_sentiment_model EXPOSE 8000 CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]注意几点:
- 基础镜像用
python:3.10-slim而不是python:3.10,能小几百MB。 --workers 2表示启动两个进程,充分利用多核CPU的并行能力。但要注意每个worker都会加载一份模型到内存,如果你部署的是大模型,worker数量要适当下调,避免内存超限。- 模型文件不要用
COPY一层层拷,而是在构建时统一放到一个目录再整体拷贝,减少镜像层数,构建效率更高。
构建镜像并运行:
docker build -t sentiment-api:v1 . docker run -d --name sentiment-service -p 8000:8000 sentiment-api:v1到这里,一个完整的模型服务已经跑起来了。如果要上线到云服务器,还需要配置反向代理(Nginx)、SSL证书、健康检查等,这些属于常规Web工程的问题,我就不在这里占用篇幅了。
6. LLM时代的AI工程新挑战:RAG、Agent和流式输出
聊完传统模型的工程化,必须说说大语言模型(LLM)时代给AI工程带来的新挑战。这两年我明显感觉到,AI工程的重心正在从“训练模型”转向“编排模型”——大部分场景下,我们不再从零训练一个模型,而是用好已有的基础大模型,再把检索、工具调用、记忆、规划这些能力串起来解决问题。
6.1 RAG系统的工程化:不只是“向量检索拼Prompt”
RAG(检索增强生成)是目前落地最多的LLM应用形态。你要做一个企业知识库问答系统,不可能让LLM凭空“记住”你们公司的规章制度,所以正确的做法是:先把文档切成小块做向量化存储,用户提问时检索最相关的片段,连同问题一起给LLM生成答案。
工程上,一个RAG系统的核心组件包括:
- 文档解析:不同格式的文档(PDF、Word、Markdown、网页)要提取成干净的纯文本,保留必要的标题和层级结构。
- 切片策略:切块大小直接决定检索效果。切小了丢失上下文,切大了噪音太多还浪费Token。我常用的经验值是200~500个字符一个块,同时设置10%~20%的重叠量,保证跨块的上下文不断裂。
- 向量化:用Embedding模型把文本转成向量。中文环境我常用
text2vec-base-chinese或bge-large-zh,维度从768到1024不等。 - 向量数据库:Milvus、Qdrant、Chroma、pgvector都是常用选项。选型的核心看你的数据量和并发量:百万级向量以内用pgvector(直接存在PostgreSQL里,不用多维护一个组件);更大规模和高并发用Milvus。
- 检索与重排:基础检索用向量相似度,但效果往往不够精确,可以在第一轮检索后加一个重排(Rerank)模型,用交互式方法对候选片段精排,能明显提升回答质量。
RAG最隐蔽的坑是“检索到了不相关内容,但LLM一本正经地答了”。我的经验是,给LLM的Prompt里明确提示“如果检索到的内容与问题无关,请直接回答‘未找到相关信息’”,同时在检索时设定相似度阈值做硬过滤,低于阈值的片段根本不进Prompt。
6.2 Agent架构:从“一个模型”到“一群工具”
另一个新方向是Agent。所谓Agent,本质上是让LLM具备“规划-行动-观察”的循环能力——它可以根据目标拆解任务,调用外部工具(数据库查询、计算器、搜索等),根据工具返回结果调整下一步行动。
从工程视角看,Agent系统与传统AI系统最大的区别是:结果不确定。同一个输入,Agent可能走不同的路径,耗时也不固定。这就给工程化带来了挑战:原来我们习惯的“请求-响应”模式,在Agent场景下变成了“请求-多次回调-最终响应”,而且中间可能有多次外部依赖调用。
我做过一个自动化数据分析Agent的经验是,Agent工程化的关键不是把Agent本身写得多复杂,而是把“工具层”建设好。工具就是Agent能调用的API,每个工具必须有明确的输入输出Schema、错误处理能力和超时设置。Agent选错工具、调用失败、超时重试,这些都要在工具层做兜底。我甚至见过有人把整个Agent系统搞成了“人工智障”,就是因为某个工具返回了一个异常格式的数据,Agent顺着错误往下走了几步,最后给用户一个完全不靠谱的答案。工具层多做防御,Agent系统才能稳定。
另外,LLM服务本身也有工程问题。主流方案是用vLLM或TGI做推理加速,支持Continuous Batching大幅提升吞吐;但是官方和第三方加速套件有很多版本兼容问题,需要仔细验证,否则会出现各种推理错误。流式输出(Streaming)则是LLM应用里几乎必须支持的能力——用户不希望在网页上干等十几秒,而是希望文字一个接一个地蹦出来,体感会好很多。这要求后端接口改成StreamingResponse,前端用SSE或WebSocket对接,整个链路和传统REST接口差异很大。
6.3 LLM应用的评估:没有测试集的系统不是好系统
传统模型有明确的Test Set和指标,但LLM应用的输出是开放式的,怎么评估?这成了LLM工程化最大的难点之一。我的做法是分两层:
第一层是“过程指标”:检索命中率、上下文相关性、指令遵循率、工具调用成功率、单轮延迟、总Token数、成本估算。这些指标可以通过日志系统自动统计。
第二层是“结果指标”:用另一个LLM当“裁判”,对主模型的输出做打分评估(LLM-as-a-Judge),或者与人工标注的参考答案做相似度对比。实践中我发现,LLM裁判在标准客观的任务上(比如“回答是否包含关键要点”“格式是否符合要求”)和人类判断的一致性可以到80%以上,但主观偏好类任务(比如“回答是否有文采”)就不太可靠。
不管用什么方法,关键是从一开始就建立评估集。哪怕只有50条典型问题,也要先沉淀下来。很多团队做完RAG系统就急着上线,结果上线后用户反馈质量不稳定,想优化却不知道从哪里下手——因为没有基线数据。我现在的习惯是,每个LLM应用项目第一天就建一个eval_set.jsonl,里面放好典型问题、期望答案要点、评估维度,后续每次改Prompt、换模型、调参数,都用同一套评估集跑一遍,对比分数,形成完整的迭代记录。
7. AI工程实操常见问题与排查技巧
做了这么多年AI工程,踩过的坑比走过的路都多。这一章节把最典型的几类问题整理成速查表和排查心得,方便你在实际工作中快速定位。
7.1 训练与推理结果不一致:线上效果暴跌的元凶
这个问题我见过太多次了,几乎每个团队都会遇到。线下评估F1值95%,线上落地效果直接腰斩。原因通常是以下几类:
| 原因类别 | 典型表现 | 排查思路 |
|---|---|---|
| 预处理不一致 | 线上分词结果和训练时不同 | 对比训练脚本和推理服务里的预处理代码,统一为一个公共模块 |
| 特征缺失 | 线下有全量统计特征,线上只有单条数据 | 检查特征计算是否依赖全局信息,拆分在线特征和离线特征 |
| 样本分布漂移 | 线上真实数据和训练数据分布差距大 | 监控线上特征分布与训练集分布的差异,做漂移检测 |
| 模型版本错误 | 部署的是旧版本模型文件 | 检查模型注册表,确认线上模块路径指向正确版本 |
我自己吃过最狠的一次亏,就是训练时在预处理里用了“去除所有标点符号”,但推理服务里漏了这一步,结果用户文本里的句号让分词完全乱掉,模型输出直接崩了。从那以后,我把预处理逻辑抽成一个独立模块,训练和推理必须从同一个模块里导入,从根源上杜绝了这类问题。
7.2 服务内存泄漏:推理推理,内存越推越满
模型服务跑几天后内存持续上涨,最后OOM被Kill掉,这是典型的“隐藏bug”。排查路径如下:
第一步,先怀疑是不是全局变量在累积。比如有些代码会在全局缓存预测结果、特征值等,随着请求数增加,缓存无限膨胀。解决方法是检查全局数据结构,该清空的清空,该用LRU缓存策略的换掉。
第二步,排查是否模型加载了多次。有些框架在不同线程/进程里会重复加载模型,每个副本都占几百MB内存。尤其是在使用多线程处理器时,要注意模型是否被正确共享。
第三步,用内存分析工具定位。我常用tracemalloc做Python层的内存追踪:
import tracemalloc tracemalloc.start() # 跑一段服务逻辑后 snapshot = tracemalloc.take_snapshot() top_stats = snapshot.statistics('lineno') for stat in top_stats[:10]: print(stat)如果内存问题出现在原生扩展层(比如PyTorch的Tensor没释放),Python工具就查不出来了,这时候看nvidia-smi显存和系统内存的变化趋势,配合逐步注释代码来二分定位。
7.3 LLM应用的超时与重试:不要让用户等到天荒地老
LLM推理速度慢(动辄几秒到几十秒),加上调外部工具,整个请求链路很容易突破常规的HTTP超时时间。我在实践中的经验是:
第一,分清同步和异步场景。用户在线等结果的交互场景,设置接口超时为30~60秒,同时前端做“排队中”的反馈;离线批量处理的场景,走异步任务队列,不设硬超时,靠任务状态轮询通知。千万不要把异步任务塞进同步接口里等它完成,这是逼着网关把你掐死。
第二,所有外部调用(LLM、向量库、搜索API)都要设置各自的超时和重试策略。一个Agent链路调5个工具,只要一个工具没设超时,整个链路就可能无响应。我习惯的超时策略是:单次调用10秒超时,重试2次,重试退避用指数策略(1s、2s、4s),并且设置整体链路的最大时间预算(比如50秒),超过就直接返回“处理超时,请稍后再试”。
7.4 数据漂移监控:模型效果什么时候开始恶化?
很多团队只在模型上线初期关注效果,时间一长就变成“黑盒”了。实际上,真实业务的数据分布是不断变化的。比如一个内容审核系统,用户会不断发明新的表达方式来规避检测;一个电商推荐系统,季节变化会彻底改变商品分布。这些变化统称为“数据漂移”。
监控数据漂移的做法是:记录线上每个请求的特征值,周期性计算特征分布和训练集分布的差异。常用的指标是PSI(Population Stability Index),计算方式如下:
import numpy as np def calculate_psi(expected, actual, bins=10): """计算PSI,衡量两个分布的一致性""" # 把expected的边界作为分箱边界 breakpoints = np.percentile(expected, np.linspace(0, 100, bins + 1)[1:-1]) expected_counts = np.histogram(expected, bins=[-np.inf] + list(breakpoints) + [np.inf])[0] actual_counts = np.histogram(actual, bins=[-np.inf] + list(breakpoints) + [np.inf])[0] expected_pct = expected_counts / len(expected) actual_pct = actual_counts / len(actual) # 把0值替换为一个极小值,避免除零 expected_pct = np.where(expected_pct == 0, 0.0001, expected_pct) actual_pct = np.where(actual_pct == 0, 0.0001, actual_pct) psi = np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct)) return psiPSI小于0.1表示分布稳定,0.1~0.25表示温和漂移,超过0.25表示严重漂移。一旦某个特征触发漂移告警,就要触发“重新训练模型”或者“人工审查数据质量”的流程。有了这套自动监控,模型的衰减问题才能被及时发现,而不是等业务方反馈“效果变差了”才被动响应。
8. 如何系统化地提升AI工程能力:一条可执行的学习路线
写到这里,如果你真的从第一章看到了这里,说明你是真想走AI工程这条路。所以在最后,我用自己的经验给出一条可执行的学习路线,比那些塞给你一堆教程链接的“学习路线图”实在得多。
第一阶段:把Python基本功夯实。不是会写for循环和调库就行,而是要到能读懂开源项目源码的程度。重点练好装饰器(理解FastAPI的路由原理)、生成器与异步编程(理解流式输出和异步任务)、上下文管理器(理解资源释放)、类型注解(理解Pydantic校验)。这些内容看着基础,但AI工程里处处都有它们的影子。
第二阶段:完整复现一个项目的部署链路。不要满足于在Jupyter里跑通模型,要用FastAPI把它包成接口,用Docker打包,部署到一台服务器上,再通过Nginx暴露出去,加上Prometheus监控。这条链路走通一遍,你对“工程”两个字的感觉会完全不同。
第三阶段:做三天以上才能完成的小项目。比如一个完整的内容审核系统、一个RAG知识库问答、一个自动化报表Agent。有复杂度的项目才会逼你思考数据管道、任务队列、错误处理、测试策略这些真实问题。
第四阶段:阅读优秀的开源AI工程项目的源码。我个人推荐读LangChain早期的核心代码,看它怎么抽象工具调用和Agent循环;再看vLLM的服务层实现,理解Continuous Batching的工程细节。读源码不要求全懂,但能极大拓宽你对系统设计的认知边界。
最后说一点个人体会。AI工程这个领域,网上的资料往往走两个极端:要么是纯理论、讲了半天不知道怎么落地,要么是纯调包、跑通就完事。真正值钱的,是中间那一层——明白每个环节为什么这么设计、出了问题从哪里排查、业务变了要从哪里改起。希望你通过这篇内容,不只是会跑通一个demo,而是建立起一套AI工程化的思考框架,这样不管技术栈怎么变,核心能力都能迁移过去。