从零搭建AI工程能力这件事,我前前后后折腾过好几轮。最早的时候我也觉得,搞AI嘛,会调个模型API、能跑通一个demo不就行了?结果真到了要把一个模型塞进业务系统里跑起来的时候,才发现坑多到离谱——显存不够、推理慢得像蜗牛、换个环境就报错、上线之后监控一片空白。后来我才慢慢意识到,AI工程和AI算法完全是两码事:算法关心的是模型效果好不好,工程关心的是这套东西能不能稳定、高效、低成本地跑在生产环境里。这篇内容就是把我从零搭建AI工程能力这条路上踩过的坑、总结的方法、以及一套可复现的落地路径完整梳理出来,适合那些已经会写Python、懂一点深度学习基础,但还没真正把AI系统跑进生产环境的同学。不管你是想转行做AI工程,还是已经在做但总觉得缺了点什么,下面的内容应该都能给你一些参考。
1. 先搞清楚AI工程到底在解决什么问题
1.1 算法和工程的分界线在哪里
很多人一开始会把AI工程理解成"把算法工程师写好的模型部署一下",这个理解不能说错,但太窄了。我自己的体会是,算法解决的是"这个模型能不能预测准",工程解决的是"这个预测能力能不能被稳定地、规模化地交付出去"。这两件事的思维方式完全不同。
举个具体的例子。算法同学在notebook里跑通一个模型,用的是固定的数据集、固定的随机种子、固定的环境,跑出来准确率95%,皆大欢喜。但工程同学要面对的是:数据是实时流进来的,格式可能随时变;模型要在一台没有GPU的机器上跑;请求量可能从每秒10个突然涨到每秒1000个;服务挂了要能自动恢复;模型更新了要能灰度发布。这些在notebook里根本不会遇到。
所以AI工程的核心命题其实是三个词:可复现、可扩展、可观测。可复现意味着任何人任何时候都能把同样的结果跑出来;可扩展意味着流量涨了、数据多了,系统还能扛住;可观测意味着出了问题你能快速定位是数据的问题、模型的问题还是服务的问题。这三个词听起来简单,但每一个背后都是一堆具体的工程决策。
1.2 从零开始需要建立的能力地图
我梳理了一下,一个完整的AI工程能力体系大概包含这么几块,你可以对照看看自己缺哪块:
| 能力模块 | 核心内容 | 常见工具/技术 |
|---|---|---|
| 数据处理 | 数据清洗、特征工程、数据版本管理 | Pandas, Spark, DVC |
| 模型训练 | 分布式训练、超参调优、实验管理 | PyTorch, MLflow, Optuna |
| 模型部署 | 模型转换、推理优化、服务化 | ONNX, TensorRT, FastAPI |
| 服务运维 | 容器化、编排、自动扩缩容 | Docker, Kubernetes |
| 监控告警 | 性能监控、数据漂移检测、日志 | Prometheus, Grafana, Evidently |
| 流水线 | CI/CD、自动化训练、自动化部署 | GitHub Actions, Airflow |
这张表不是让你每个都精通,而是让你知道整个链路长什么样。我见过太多人只盯着"模型部署"这一块,结果数据版本对不上、训练环境复现不了、上线后模型效果衰减了也不知道,最后整个系统就是个黑盒。
1.3 为什么建议从"最小可运行闭环"开始
新手最容易犯的错,是一上来就想搭一个"完美"的AI平台。我当年也是这样,花了两周时间研究各种框架,结果一行能跑的代码都没写出来。后来我换了个思路:先用最简单的技术栈,搭一个从数据到模型到服务的最小闭环,哪怕它很丑、很慢、很不优雅,但它是能跑通的。
这个最小闭环大概长这样:一个Python脚本读数据、一个简单的模型训练、把模型存成文件、用一个web框架包成API、写个脚本调用测试。就这么点东西,可能半天就能搞定。但当你把这个闭环跑通之后,你就知道每个环节的输入输出是什么、哪里容易出问题、哪里需要优化。这时候再去引入Docker、Kubernetes、MLflow这些工具,你才知道它们到底解决了什么问题,而不是为了用而用。
我的经验是:工具永远是为问题服务的。先有痛点,再找工具,顺序反了就会陷入"学了一堆工具但不知道怎么用"的困境。
2. 环境与依赖管理:别让"在我机器上能跑"成为噩梦
2.1 Python环境隔离的几种方案对比
"在我机器上能跑"这句话大概是工程领域最经典的梗了。AI项目尤其严重,因为依赖特别多、版本特别敏感。PyTorch 1.x和2.x的API不兼容,CUDA版本和驱动版本要对上,numpy版本高了低了都可能出问题。我踩过最离谱的坑是一个项目在本地跑得好好的,部署到服务器上因为numpy版本差了一个小版本,矩阵运算结果出现了微小的数值差异,导致模型输出完全不对。
环境隔离的方案主要有这么几种,我做个对比:
| 方案 | 隔离级别 | 适用场景 | 缺点 |
|---|---|---|---|
| venv | Python包级别 | 本地开发、简单项目 | 不隔离系统库和CUDA |
| conda | 包+系统库级别 | 科学计算、需要特定CUDA | 体积大、解析慢 |
| Docker | 完整系统级别 | 生产部署、团队协作 | 学习成本、镜像体积 |
| 虚拟环境+容器 | 组合方案 | 复杂项目 | 配置繁琐 |
我现在的习惯是:本地开发用conda管理Python环境,因为科学计算相关的包它处理得最好;生产部署一律用Docker,因为只有容器能保证"开发环境等于生产环境"。中间用一个environment.yml或者requirements.txt把依赖锁死,所有版本号都写精确,不用>=这种模糊约束。
2.2 依赖锁定:为什么requirements.txt不够用
很多人以为pip freeze > requirements.txt就万事大吉了,其实这里面有个大坑:pip freeze只会导出你显式安装的包和它们的直接依赖,但不会锁定依赖的依赖。也就是说,A包依赖B包,你锁了A的版本,但B的版本没锁,下次安装的时候B可能升级了,然后就和A不兼容了。
更靠谱的做法是用pip-tools或者poetry这类工具做完整的依赖解析和锁定。以pip-tools为例:
# 在requirements.in里写直接依赖,不写版本 # 然后生成锁定的requirements.txt pip-compile requirements.in --output-file requirements.txt # 安装时严格按锁定文件来 pip-sync requirements.txt这样生成的requirements.txt会包含所有传递依赖的精确版本,任何人任何时候安装都是一样的结果。对于AI项目,我还会额外记录CUDA版本、cuDNN版本、显卡驱动版本,因为这些不在Python包管理范围内,但同样会影响结果。
2.3 容器化:把整个运行环境打包带走
Docker解决的是"环境一致性"这个根本问题。但AI项目的Docker镜像有个特殊难点:GPU支持。普通的Docker镜像跑不了GPU,需要用nvidia-container-toolkit,基础镜像也要选对。
我常用的一个PyTorch训练镜像的Dockerfile大概长这样:
FROM nvidia/cuda:12.1.0-cudnn8-runtime-ubuntu22.04 # 装Python和基础工具 RUN apt-get update && apt-get install -y \ python3.10 python3-pip git curl \ && rm -rf /var/lib/apt/lists/* # 设置工作目录 WORKDIR /app # 先复制依赖文件,利用Docker层缓存 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 再复制代码 COPY . . CMD ["python", "train.py"]这里有个关键技巧:先复制依赖文件再复制代码。因为Docker是分层构建的,只要requirements.txt没变,依赖安装这一层就会用缓存,构建速度能快好几倍。我见过有人把整个项目一次性COPY进去,结果改一行代码就要重新装一遍依赖,等得想砸电脑。
注意:生产环境的镜像尽量用
runtime版本而不是devel版本,体积能小很多。devel版本包含了编译工具链,只有你需要从源码编译CUDA扩展的时候才用得上。
3. 数据处理流水线:模型效果的上限由数据决定
3.1 数据版本管理为什么比代码版本管理更重要
代码可以回滚,数据回滚起来就麻烦多了。我遇到过好几次这样的情况:模型效果突然下降,排查了半天代码没改,最后发现是上游数据源偷偷改了字段格式,或者某天的数据采集出了问题。没有数据版本管理,你连"上次效果好的时候用的是哪份数据"都说不清楚。
数据版本管理我推荐两个思路。轻量级的用DVC,它把大文件存在对象存储里,Git里只存元数据指针,用起来和Git很像:
# 初始化DVC dvc init # 添加数据文件 dvc add data/train.csv # 提交元数据 git add data/train.csv.dvc data/.gitignore git commit -m "add training data v1" # 切换数据版本 git checkout <commit> dvc checkout重量级的就用数据湖方案,比如Delta Lake或者Apache Iceberg,它们支持时间旅行、schema演进、ACID事务,适合数据量大、多人协作的场景。选哪个取决于你的数据规模和团队情况,小团队用DVC足够了。
3.2 特征工程的工程化:从notebook到生产
notebook里做特征工程和在生产环境做特征工程,最大的区别是一致性。notebook里你可以随手写个df['new_feature'] = df['a'] / df['b'],但生产环境里,训练时的特征计算逻辑和推理时的特征计算逻辑必须完全一致,否则就会出现"训练-服务偏差"(training-serving skew)。
解决这个问题的标准做法是特征存储(Feature Store)。它的核心思想是:特征的计算逻辑只写一次,训练时用批处理方式算历史特征,推理时用流式方式算实时特征,但用的是同一套代码。Feast是一个比较流行的开源方案:
from feast import FeatureStore store = FeatureStore(repo_path="./feature_repo") # 训练时获取历史特征 training_df = store.get_historical_features( entity_df=entity_df, features=["driver_stats:conv_rate", "driver_stats:acc_rate"] ).to_df() # 推理时获取在线特征 online_features = store.get_online_features( features=["driver_stats:conv_rate"], entity_rows=[{"driver_id": 1001}] ).to_dict()如果团队规模不大,不一定非要上Feature Store,但至少要保证特征计算逻辑封装成独立的函数或类,训练和推理都调用同一份代码。我见过最糟糕的情况是训练用SQL算特征、推理用Python重写一遍,两边逻辑稍微有点差异,模型效果就崩了。
3.3 数据质量校验:在脏数据进入模型之前拦住它
数据质量问题是AI系统最隐蔽的杀手。模型不会报错,它只会默默地给出错误的预测。所以必须在数据进入训练或推理之前做校验。校验的内容包括:字段是否存在、类型是否正确、取值范围是否合理、分布是否发生漂移。
我用得比较多的是Great Expectations和Pandera。Pandera更轻量,适合在代码里直接做schema校验:
import pandera as pa from pandera import Column, DataFrameSchema, Check schema = DataFrameSchema({ "age": Column(int, Check.in_range(0, 120)), "income": Column(float, Check.greater_than(0)), "category": Column(str, Check.isin(["A", "B", "C"])), }) # 校验数据,不符合就抛异常 validated_df = schema.validate(raw_df)对于数据漂移检测,可以用Evidently这类工具,它会对比训练数据和推理数据的分布,当漂移超过阈值时告警。这个在生产环境特别重要,因为模型效果衰减往往不是模型本身的问题,而是输入数据的分布变了。
4. 模型训练与实验管理:让每一次实验都可追溯
4.1 实验追踪:别再靠文件名区分模型版本了
我早期管理实验的方式极其原始:model_v1.pth、model_v2_final.pth、model_v2_final_真的最终版.pth。结果过了一个月,我自己都不知道哪个是哪个,超参数是什么、用的哪份数据、效果多少,全忘了。这种混乱在小规模实验时还能忍,一旦实验数量上去了,就是灾难。
MLflow是我用得最顺手的实验追踪工具,它的核心概念很简单:每次训练是一个run,run里记录参数、指标、产物。用起来就几行代码:
import mlflow mlflow.set_experiment("my_experiment") with mlflow.start_run(): # 记录超参数 mlflow.log_params({"lr": 0.001, "batch_size": 32, "epochs": 10}) # 训练循环 for epoch in range(10): train_loss = train_one_epoch() val_loss = validate() # 记录指标 mlflow.log_metrics({"train_loss": train_loss, "val_loss": val_loss}, step=epoch) # 记录模型产物 mlflow.pytorch.log_model(model, "model")跑完之后打开MLflow的UI,所有实验一目了然,可以按指标排序、对比不同run的参数、直接下载模型。这个投入产出比极高,强烈建议从第一个实验就开始用。
4.2 超参数调优:网格搜索之外的选择
超参数调优如果还用网格搜索,那真的是在浪费算力。网格搜索的复杂度是指数级的,5个参数各5个候选值就是3125次训练,根本跑不起。实际工作中我用得最多的是贝叶斯优化和Hyperband。
Optuna是我最推荐的框架,它支持多种采样算法,而且有个剪枝机制特别实用——如果某个试验在前几个epoch表现明显差于历史最优,直接提前终止,省下算力:
import optuna def objective(trial): lr = trial.suggest_float("lr", 1e-5, 1e-2, log=True) batch_size = trial.suggest_categorical("batch_size", [16, 32, 64]) dropout = trial.suggest_float("dropout", 0.1, 0.5) model = build_model(dropout) optimizer = torch.optim.Adam(model.parameters(), lr=lr) for epoch in range(20): val_loss = train_and_validate(model, optimizer, batch_size) # 报告中间结果,支持剪枝 trial.report(val_loss, epoch) if trial.should_prune(): raise optuna.TrialPruned() return val_loss study = optuna.create_study( direction="minimize", pruner=optuna.pruners.MedianPruner() ) study.optimize(objective, n_trials=100)实测下来,同样的算力预算,Optuna找到的最优解通常比网格搜索好一截,而且时间省一半以上。
4.3 训练的可复现性:随机种子只是第一步
要让训练结果可复现,设置随机种子只是最基础的一步。完整的可复现需要控制这些变量:
- Python的
random.seed() - NumPy的
np.random.seed() - PyTorch的
torch.manual_seed()和torch.cuda.manual_seed_all() - cuDNN的确定性设置:
torch.backends.cudnn.deterministic = True - 数据加载的shuffle顺序
- 多卡训练的梯度同步顺序
但即使这些都设了,GPU上的浮点运算仍然可能有微小的非确定性,因为CUDA的某些操作(比如atomicAdd)的执行顺序是不确定的。所以严格意义上的"完全可复现"在GPU上很难做到,工程上追求的是"统计意义上的可复现"——同样的配置跑多次,指标波动在可接受范围内。
我的做法是:把随机种子、环境版本、数据版本、代码commit hash全部记录到MLflow的run里。这样即使不能100%复现,至少能追溯到当时的所有条件。
5. 模型部署与推理优化:从实验室到生产的关键一跃
5.1 模型格式转换:为什么不能直接用PyTorch模型部署
PyTorch训练出来的.pth文件包含了完整的模型结构和权重,但它依赖PyTorch运行时,部署起来又重又慢。生产环境通常会把模型转换成更轻量的格式,常见的有ONNX和TensorRT。
ONNX是一个开放的模型交换格式,几乎所有主流框架都支持导出。它的好处是跨平台、跨框架,而且有很多推理引擎支持(ONNX Runtime、TensorRT、OpenVINO等):
import torch import torch.onnx # 加载训练好的模型 model = MyModel() model.load_state_dict(torch.load("model.pth")) model.eval() # 构造一个示例输入 dummy_input = torch.randn(1, 3, 224, 224) # 导出为ONNX torch.onnx.export( model, dummy_input, "model.onnx", input_names=["input"], output_names=["output"], dynamic_axes={"input": {0: "batch_size"}, "output": {0: "batch_size"}}, opset_version=13 )这里有个关键点:dynamic_axes参数。如果不设置,导出的ONNX模型会固定batch size,推理时只能接受那个尺寸的输入。设置成动态之后,batch size可以变化,灵活性高很多。
TensorRT是NVIDIA的推理优化引擎,它会对模型做层融合、精度校准、kernel自动调优,在NVIDIA GPU上能比原生PyTorch快好几倍。但它的缺点是只支持NVIDIA GPU,而且转换过程比较耗时。我的建议是:如果部署在GPU上且追求极致性能,用TensorRT;如果需要跨平台或者部署在CPU上,用ONNX Runtime。
5.2 推理服务的几种架构模式
推理服务的架构选择取决于你的场景。我总结了几种常见模式:
模式一:同步请求-响应。客户端发请求,服务端推理完返回结果。适合实时性要求高、请求量不大的场景。用FastAPI就能快速搭起来:
from fastapi import FastAPI import onnxruntime as ort import numpy as np app = FastAPI() session = ort.InferenceSession("model.onnx") @app.post("/predict") async def predict(data: dict): input_array = np.array(data["features"], dtype=np.float32) outputs = session.run(None, {"input": input_array}) return {"prediction": outputs[0].tolist()}模式二:批处理。请求先攒着,攒够一批或者等一小段时间再一起推理。因为GPU的并行能力很强,batch size从1加到32,推理时间可能只增加一点点,但吞吐量提升几十倍。这个模式适合对延迟不那么敏感的场景。
模式三:异步队列。请求进来先丢到消息队列,后台worker慢慢消费。适合推理时间很长(比如大模型生成)的场景,客户端拿到一个任务ID,过一会儿再来查结果。
模式四:流式推理。边生成边返回,适合文本生成这类场景。这个模式对服务端的要求最高,需要支持Server-Sent Events或者WebSocket。
选哪种模式,核心看两个指标:延迟要求和吞吐量要求。延迟要求高就同步,吞吐量要求高就批处理,两者都高就得上更复杂的架构。
5.3 推理性能优化的几个实用手段
推理优化是个深坑,我挑几个投入产出比最高的手段说说。
量化是最直接的手段。把FP32的权重转成INT8,模型体积缩小4倍,推理速度提升2-4倍,精度损失通常在1%以内。PyTorch支持动态量化和静态量化:
# 动态量化,最简单,一行代码 quantized_model = torch.quantization.quantize_dynamic( model, {torch.nn.Linear}, dtype=torch.qint8 ) # 静态量化,需要校准数据,精度更好 model.qconfig = torch.quantization.get_default_qconfig('fbgemm') torch.quantization.prepare(model, inplace=True) # 用校准数据跑一遍 for data in calibration_loader: model(data) torch.quantization.convert(model, inplace=True)算子融合是另一个大杀器。把Conv+BN+ReLU这种连续操作融合成一个算子,减少内存访问和kernel启动开销。TensorRT和ONNX Runtime都会自动做这个优化,你不需要手动改代码。
KV Cache是针对Transformer类模型的优化。自回归生成的时候,每次生成一个新token都要重新计算前面所有token的注意力,非常浪费。KV Cache把之前算过的Key和Value缓存起来,每次只算新token的,速度能提升好几倍。这个在HuggingFace的generate方法里默认就开了。
批处理调度是服务层面的优化。与其每个请求单独推理,不如攒一批一起推。但攒批会增加延迟,所以要设一个最大等待时间,比如10毫秒,超过就立即推理。这个平衡点需要根据实际流量调。
6. 监控与运维:上线只是开始,不是结束
6.1 模型监控和普通服务监控的区别
普通服务的监控看的是CPU、内存、QPS、延迟、错误率这些指标。AI服务除了这些,还要额外监控模型层面的指标:预测分布、特征分布、置信度分布。因为模型可能服务本身很健康(延迟低、不报错),但预测结果已经悄悄变差了。
我遇到过最典型的情况:一个推荐模型上线三个月后效果明显下降,但服务监控一切正常。排查后发现是用户行为模式变了,模型训练时用的数据分布和现在的线上数据分布差异很大,模型"过时"了。这种问题只有监控模型层面的指标才能发现。
6.2 数据漂移检测的落地方法
数据漂移检测的核心是:把线上推理时的输入数据分布,和训练时的数据分布做对比。如果差异超过阈值,就告警。常用的统计指标有:
- PSI(Population Stability Index):衡量两个分布的差异,PSI < 0.1说明分布稳定,0.1-0.25说明有轻微漂移,> 0.25说明显著漂移。
- KL散度:衡量两个概率分布的差异。
- KS检验:判断两个样本是否来自同一分布。
用Evidently可以很方便地做这件事:
from evidently.report import Report from evidently.metric_preset import DataDriftPreset report = Report(metrics=[DataDriftPreset()]) report.run(reference_data=train_df, current_data=prod_df) report.save_html("drift_report.html") # 也可以拿到结构化的结果 result = report.as_dict() drift_detected = result["metrics"][0]["result"]["dataset_drift"]我一般会把这个检测做成定时任务,每天跑一次,结果推到监控面板上。一旦检测到显著漂移,就触发模型重训练的流程。
6.3 模型回滚与灰度发布
模型更新比代码更新风险更高,因为模型的行为很难用单元测试覆盖。所以模型上线一定要支持灰度发布和快速回滚。
灰度发布的常见做法是按流量比例切分:新模型先接1%的流量,观察一段时间没问题,再逐步加到10%、50%、100%。实现方式可以是在服务层做一个路由,根据请求ID的哈希值决定走哪个模型:
import hashlib def route_model(request_id: str, new_model_ratio: float = 0.01): # 根据请求ID的哈希值决定路由 hash_val = int(hashlib.md5(request_id.encode()).hexdigest(), 16) if (hash_val % 100) < (new_model_ratio * 100): return new_model return old_model回滚就更简单了,模型文件都存着,把路由切回旧模型就行。关键是要保证旧模型的服务实例还在,别一上线就把旧的删了。我一般会保留最近3个版本的模型,随时可以切回去。
一个血的教训:模型上线前一定要做shadow mode,也就是新模型接收真实流量但不返回结果,只记录预测。对比新老模型的预测差异,确认没有异常再正式切流量。这个步骤能拦住大部分低级错误。
7. 把整条链路串起来:CI/CD与自动化
7.1 AI项目的CI/CD和普通项目有什么不同
普通项目的CI/CD是:代码提交 → 跑测试 → 构建镜像 → 部署。AI项目多了几个环节:数据校验 → 模型训练 → 模型评估 → 模型注册 → 部署。而且模型训练本身可能耗时很长,不能每次提交都跑全量训练。
我的做法是分两条流水线。代码流水线跑得快,每次提交都触发,只做代码lint、单元测试、镜像构建。模型流水线跑得慢,按需触发或者定时触发,做数据校验、训练、评估、注册。
7.2 用GitHub Actions搭一条模型训练流水线
下面是一个简化的模型训练流水线示例,用GitHub Actions实现:
name: Model Training Pipeline on: schedule: - cron: '0 2 * * 0' # 每周日凌晨2点跑一次 workflow_dispatch: # 也支持手动触发 jobs: train: runs-on: [self-hosted, gpu] # 需要GPU的runner steps: - uses: actions/checkout@v3 - name: Setup Python uses: actions/setup-python@v4 with: python-version: '3.10' - name: Install dependencies run: pip install -r requirements.txt - name: Validate data run: python scripts/validate_data.py - name: Train model run: python scripts/train.py --config configs/prod.yaml - name: Evaluate model run: python scripts/evaluate.py --threshold 0.85 - name: Register model if: success() run: python scripts/register_model.py这里有几个关键点:用self-hosted runner是因为训练需要GPU,GitHub托管的runner没有GPU;评估步骤设了阈值,效果不达标就中断流水线,不会把差模型推上去;模型注册步骤只在前面都成功时才执行。
7.3 模型注册与版本管理
模型注册表是连接训练和部署的桥梁。每次训练产出的模型都注册进去,带上版本号、指标、训练数据版本、代码commit等信息。MLflow Model Registry就是干这个的:
import mlflow # 注册模型 mlflow.register_model( model_uri=f"runs:/{run_id}/model", name="my_model" ) # 把某个版本标记为生产可用 client = mlflow.tracking.MlflowClient() client.transition_model_version_stage( name="my_model", version=3, stage="Production" )部署服务从注册表拉取标记为Production的模型版本,这样训练和部署就解耦了。训练团队只管往注册表推模型,部署团队只管从注册表拉模型,中间通过stage这个状态来协调。
8. 一些踩坑之后的真心话
8.1 不要过早优化,但也不要欠太多技术债
我见过两种极端。一种是过度设计,一个日请求量几百的小服务,非要上Kubernetes加服务网格,运维复杂度高得离谱,出问题了没人会修。另一种是完全不设计,所有东西写在一个脚本里,模型文件用文件名区分版本,数据直接读本地CSV,等到要扩展的时候推倒重来。
我的建议是:架构的复杂度应该匹配当前的业务规模,但要为未来留好扩展点。比如你一开始可以用单机部署,但代码里要把模型加载、推理、后处理这些逻辑分层写好,将来要拆成微服务的时候直接拆就行。数据存储一开始可以用本地文件,但读写接口要封装好,将来换成对象存储只改一个配置。
8.2 日志和可观测性要从第一天就做
这个是我踩过最大的坑。早期项目为了赶进度,日志随便打,监控也没做。结果上线后模型效果不对,排查了整整两天,因为根本不知道是数据的问题还是模型的问题还是代码的问题。后来我强制要求:任何AI服务上线前,必须有三样东西——结构化的日志、关键指标的监控面板、模型输入输出的采样记录。
结构化日志意味着不要用print,用logging并且输出JSON格式,方便后续检索和分析。关键指标包括推理延迟的P50/P95/P99、每秒请求数、错误率、模型置信度分布。输入输出采样记录是为了出问题的时候能复现,但要注意脱敏和存储成本。
8.3 团队协作:接口约定比技术选型更重要
AI工程项目通常涉及多个角色:数据工程师、算法工程师、后端工程师、运维工程师。我见过太多团队在技术选型上吵得不可开交,却忽略了更重要的接口约定。
所谓接口约定,就是明确每个环节的输入输出格式。数据团队产出的数据长什么样、字段类型是什么、缺失值怎么表示;算法团队产出的模型接受什么格式的输入、输出什么格式的结果;后端团队怎么调用模型服务、超时时间设多少、错误怎么处理。这些约定如果一开始就定清楚,后面能省掉大量的扯皮。
我的做法是维护一份INTERFACES.md,把每个模块的输入输出、依赖关系、SLA都写清楚。每次有变更先改文档再改代码,这样所有人都能对齐。
8.4 持续学习:这个领域变化太快了
AI工程这个领域,工具和最佳实践几乎每年都在变。两年前大家还在用Flask部署模型,现在FastAPI成了标配;一年前大家还在手动管理实验,现在MLflow、WandB已经普及。保持学习的方法我觉得有几个:关注几个高质量的开源项目看它们怎么做的、定期读一些工程博客、最重要的是动手试。
但也不要盲目追新。新技术出来先问自己:它解决了什么我现在遇到的问题?如果我现在没有这个问题,那就先记下来,等遇到了再说。我见过太多人追了一堆新工具,结果每个都只懂皮毛,真正遇到问题的时候还是抓瞎。
说到底,AI工程的核心能力不是会用多少工具,而是理解整个系统的运作原理,知道每个环节可能出什么问题,以及怎么排查和解决。工具会变,但这些底层能力是通用的。把最小闭环跑通,把可复现、可扩展、可观测这三个原则刻在脑子里,剩下的就是在这个基础上不断迭代和打磨了。