1. 从零搭建AI工程体系,为什么我劝你别急着调库
"ai-engineering-from-scratch"这个标题,第一次看到的时候我愣了一下。市面上讲AI的文章铺天盖地,但绝大多数都在教你import torch然后调个预训练模型跑个demo,真正愿意从工程地基开始讲起的内容少得可怜。我自己在这个方向上摸爬滚打了几年,踩过的坑比写过的代码还多,所以看到这个标题的时候特别有共鸣——它说的不是"从零学AI理论",而是"从零搭建AI工程体系",这两件事之间的差距,大概相当于"会炒菜"和"能开一家餐厅"的差距。
先把话说清楚:这篇内容适合谁看?如果你已经会用Python,知道什么是神经网络,能跑通一个MNIST分类,但一到真实项目就不知道从哪下手——数据怎么管、实验怎么追踪、模型怎么部署、线上出问题怎么排查——那这篇就是写给你的。如果你是完全零基础,建议先补一下Python和机器学习基础再来,不然读起来会比较吃力。如果你已经是资深MLOps工程师,这篇可能对你来说偏基础,但里面的一些实操细节和踩坑经验,说不定还是能帮你省点时间。
我打算聊的核心问题是:一个AI工程项目,从一行代码都没有,到能稳定跑在线上服务用户,中间到底需要哪些东西?这些东西的优先级怎么排?哪些环节最容易出问题?每个环节我会给出具体的工具选型、操作步骤和参数配置,尽量做到你照着做就能复现。整个内容会围绕"工程"两个字展开,不会花太多篇幅讲算法原理——那些东西网上太多了,但工程实践中的脏活累活,很少有人系统性地讲。
2. 整体架构设计:先想清楚你要建什么
2.1 为什么"从零"反而是一种优势
很多人觉得从零开始是劣势,什么都得自己搭。但我在实际项目中的体会恰恰相反:从零开始意味着你对每一层都有完全的掌控力,不会出现"这个框架封装太深,出了问题根本不知道从哪查"的情况。我见过太多团队,上来就用某个大而全的平台,结果数据格式对不上、版本冲突、性能瓶颈找不到原因,最后推倒重来的时间比从头搭还长。
从零搭建的核心思路是:先跑通最小闭环,再逐层加固。什么叫最小闭环?就是数据能进来、模型能训练、结果能输出、服务能调用。这四个环节哪怕每个都做得极其粗糙,只要闭环通了,后面就是迭代优化的问题。最怕的是一上来就追求完美架构,结果三个月过去了连个能跑的demo都没有。
具体来说,我会把整个AI工程体系分成五层:数据层、实验层、训练层、部署层、监控层。每一层都有明确的职责边界,层与层之间通过约定好的接口通信。这样设计的好处是,任何一层出问题都不会影响其他层,替换某一层的实现也不需要动其他层的代码。
2.2 五层架构的职责划分与选型逻辑
数据层负责原始数据的采集、清洗、存储和版本管理。这一层的核心挑战不是技术难度,而是数据一致性——你训练时用的数据和推理时见到的数据,分布必须尽可能一致。我见过太多模型离线指标漂亮得不行,一上线就崩,根本原因就是训练数据和线上数据不是一回事。
实验层负责管理你的每一次训练尝试。包括超参数、数据集版本、代码版本、评估指标、模型权重。这一层的核心价值是可复现性——三个月后你还能准确知道当时那个效果最好的模型是怎么训出来的。没有这一层,你的实验就是一团乱麻,全靠记忆和Excel表格,迟早出事。
训练层是实际跑模型训练的地方。这一层要考虑的是资源调度、分布式训练、断点续训、混合精度等问题。选型上,小团队用单机多卡就够了,大团队才需要上集群调度。
部署层负责把训练好的模型变成可调用的服务。这一层的核心指标是延迟和吞吐。一个模型在笔记本上跑得好好的,部署到线上可能因为并发、内存、网络等问题完全不能用。
监控层负责盯着线上服务的运行状态。包括请求量、延迟分布、错误率、模型输出分布漂移等。这一层最容易被忽视,但恰恰是保证长期稳定运行的关键。
2.3 技术选型的取舍原则
选型这件事,我的原则是:能用简单的就不用复杂的,能用成熟就不用新兴的,能用自己掌控的就不用黑盒的。具体来说:
| 层级 | 推荐方案 | 备选方案 | 选择理由 |
|---|---|---|---|
| 数据存储 | PostgreSQL + 对象存储 | 数据湖方案 | 小规模够用,运维成本低 |
| 数据版本 | DVC | LakeFS | 与Git集成好,学习曲线平缓 |
| 实验追踪 | MLflow | Weights & Biases | 开源可自托管,无供应商锁定 |
| 训练框架 | PyTorch | TensorFlow | 生态活跃,调试方便 |
| 模型服务 | FastAPI + ONNX Runtime | TorchServe | 轻量可控,性能足够 |
| 监控 | Prometheus + Grafana | 商业APM | 开源标准,社区成熟 |
这张表不是绝对的,但背后的逻辑是一致的:在满足需求的前提下,选择运维成本最低、学习成本最低、锁定风险最低的方案。很多团队一上来就上Kubernetes、上Feature Store、上各种高大上的组件,结果维护这些基础设施的人力成本比做业务还高,完全本末倒置。
3. 数据层:一切问题的根源都在这里
3.1 数据采集与清洗的实操要点
数据采集听起来简单,做起来全是坑。我拿一个实际项目举例:我们要做一个文本分类模型,数据来源是用户提交的反馈。看起来就是读数据库、导出CSV、清洗一下就行了对吧?实际上遇到的问题包括:编码不统一(有的UTF-8有的GBK)、字段缺失(用户没填分类标签)、重复数据(同一条反馈提交了三次)、噪声数据(测试账号提交的乱码)。
处理这些问题的标准流程是这样的:
import pandas as pd import hashlib def clean_text_data(raw_df): # 第一步:统一编码,强制转UTF-8 for col in raw_df.select_dtypes(include=['object']).columns: raw_df[col] = raw_df[col].apply( lambda x: x.encode('utf-8', errors='ignore').decode('utf-8') if isinstance(x, str) else x ) # 第二步:去除完全重复的行 raw_df = raw_df.drop_duplicates() # 第三步:基于内容哈希去重(处理近似重复) raw_df['content_hash'] = raw_df['content'].apply( lambda x: hashlib.md5(str(x).strip().lower().encode()).hexdigest() ) raw_df = raw_df.drop_duplicates(subset=['content_hash']) # 第四步:过滤无效样本 raw_df = raw_df[raw_df['content'].str.len() >= 5] # 太短的不要 raw_df = raw_df[raw_df['label'].notna()] # 没标签的不要 # 第五步:重置索引 raw_df = raw_df.reset_index(drop=True) return raw_df这段代码看起来平平无奇,但每一步背后都有血泪教训。比如第三步的内容哈希去重,为什么要先strip().lower()?因为用户可能只是多打了个空格或者大小写不同,内容其实是一样的。如果不做这一步,你的训练集里会有大量近似重复样本,导致模型过拟合。
注意:数据清洗的每一步都要记录日志,包括清洗前多少条、清洗后多少条、过滤掉了哪些。这些数字在后续排查问题时非常关键。我习惯把清洗报告存成JSON,和数据集版本一起管理。
3.2 数据版本管理:别再用文件名区分了
我见过太多团队用train_v2_final_真的最终版.csv这种方式管理数据版本。这种方式在单人项目里勉强能用,一旦多人协作就是灾难。你永远不知道v2和v3之间到底改了什么,也没法回滚到某个特定版本。
正确的做法是用DVC(Data Version Control)来管理数据版本。DVC的工作原理很简单:它把大文件存在别的地方(本地目录或对象存储),然后在Git里只存一个很小的.dvc文件记录元信息。这样你的Git仓库不会因为数据文件变得巨大,同时数据版本和代码版本是绑定的。
具体操作步骤:
# 初始化DVC(在Git仓库根目录) git init dvc init # 配置远程存储(这里用本地目录举例,生产环境用S3或MinIO) dvc remote add -d myremote /path/to/dvc-storage # 添加数据文件到DVC管理 dvc add data/raw/train.csv # 这一步会生成 data/raw/train.csv.dvc 文件 # 把.dvc文件加入Git git add data/raw/train.csv.dvc data/raw/.gitignore git commit -m "add training data v1" # 推送数据到远程存储 dvc push # 切换数据版本(比如回到某个历史版本) git checkout <commit-hash> -- data/raw/train.csv.dvc dvc checkout这套流程的核心价值是:任何时候你都能准确复现某个实验用的数据。git log能看到数据版本的变化历史,dvc checkout能切换到任意历史版本。这比文件名管理靠谱一万倍。
3.3 数据质量监控:上线前的最后一道防线
数据质量监控不是可选项,是必选项。我建议在数据进入训练流程之前,加一道自动化的质量检查。检查项包括:
- 完整性:关键字段有没有缺失,缺失率是否超过阈值
- 一致性:字段类型是否符合预期,取值范围是否合理
- 分布:数值特征的均值、方差是否和基线一致,类别特征的分布是否偏移
- 时效性:数据的时间戳是否在合理范围内,有没有未来数据泄露
用Great Expectations这个库可以比较方便地实现这些检查:
import great_expectations as gx context = gx.get_context() validator = context.sources.pandas_default.read_csv("data/raw/train.csv") # 检查标签列没有空值 validator.expect_column_values_to_not_be_null("label") # 检查文本长度在合理范围 validator.expect_column_value_lengths_to_be_between( "content", min_value=5, max_value=5000 ) # 检查标签分布没有严重偏移 validator.expect_column_proportion_of_unique_values_to_be_between( "label", min_value=0.01, max_value=0.99 ) results = validator.validate() if not results["success"]: raise ValueError("数据质量检查未通过,终止训练流程")实操心得:数据质量检查的阈值不要设得太死。我一开始把文本长度上限设成1000,结果发现有些正常样本就是比较长,导致大量误报。后来改成动态阈值——基于历史数据的P99分位数来设定,误报率大幅下降。
4. 实验管理与训练流程:让每一次尝试都有迹可循
4.1 实验追踪:告别Excel表格
没有实验追踪的团队,工作状态大概是这样:小王跑了个模型准确率92%,小李说他之前跑了个93%的但忘了怎么跑的,小张说他记得好像有个参数是0.001效果不错但不确定是哪个实验。这种状态下,团队的整体效率极低,大量时间浪费在重复实验和信息同步上。
MLflow是我用过最顺手的实验追踪工具,核心概念就三个:Experiment(实验组)、Run(单次运行)、Artifact(产物)。每次训练创建一个Run,把超参数、指标、模型文件都记录下来,之后在UI界面里就能直观对比不同Run的效果。
import mlflow import mlflow.pytorch mlflow.set_tracking_uri("http://localhost:5000") mlflow.set_experiment("text-classification") with mlflow.start_run(run_name="bert-base-lr2e5"): # 记录超参数 mlflow.log_params({ "model_name": "bert-base-chinese", "learning_rate": 2e-5, "batch_size": 32, "epochs": 5, "max_length": 128, "warmup_ratio": 0.1 }) # 训练循环(伪代码) for epoch in range(5): train_loss = train_one_epoch(model, train_loader) val_metrics = evaluate(model, val_loader) # 记录指标 mlflow.log_metrics({ "train_loss": train_loss, "val_accuracy": val_metrics["accuracy"], "val_f1": val_metrics["f1"] }, step=epoch) # 记录模型 mlflow.pytorch.log_model(model, "model") # 记录数据版本 mlflow.log_param("data_version", "v1.2.0")这套流程跑通之后,你打开MLflow的UI界面,就能看到所有实验的对比表格,按准确率排序,点进去能看到详细的参数和指标曲线。想复现哪个实验,直接看它的参数配置就行。
4.2 训练流程的标准化封装
每次训练都手写训练循环,不仅效率低,还容易出错。我的做法是把训练流程封装成一个标准化的Pipeline,通过配置文件来控制行为。这样换模型、换数据、换超参数都不需要改代码,只改配置就行。
配置文件用YAML格式:
# config/train_config.yaml model: name: bert-base-chinese num_labels: 10 dropout: 0.1 data: train_path: data/processed/train.csv val_path: data/processed/val.csv max_length: 128 batch_size: 32 training: learning_rate: 2e-5 epochs: 5 warmup_ratio: 0.1 weight_decay: 0.01 gradient_accumulation_steps: 1 fp16: true seed: 42 logging: experiment_name: text-classification log_interval: 50 save_top_k: 3训练脚本读取这个配置,根据配置构建模型、数据加载器、优化器,然后跑训练循环。这样做的好处是:所有实验的配置都有记录,复现时只需要找到对应的配置文件。
import yaml from dataclasses import dataclass @dataclass class TrainConfig: model_name: str learning_rate: float batch_size: int epochs: int # ... 其他字段 def load_config(path: str) -> TrainConfig: with open(path, 'r') as f: raw = yaml.safe_load(f) return TrainConfig( model_name=raw['model']['name'], learning_rate=raw['training']['learning_rate'], batch_size=raw['data']['batch_size'], epochs=raw['training']['epochs'], ) def train(config: TrainConfig): set_seed(config.seed) # 固定随机种子,保证可复现 model = build_model(config) train_loader, val_loader = build_dataloaders(config) optimizer = build_optimizer(model, config) for epoch in range(config.epochs): # 训练和验证逻辑 ...注意:随机种子的设置要覆盖所有随机源——Python的random、NumPy的random、PyTorch的random、CUDA的随机。少设一个都可能导致结果不可复现。我一般会写一个
set_seed函数统一处理。
4.3 超参数搜索的工程化做法
超参数搜索不是随便试几个值就行,需要有策略。常用的策略有三种:网格搜索、随机搜索、贝叶斯优化。小规模实验用随机搜索性价比最高,大规模实验用贝叶斯优化更省资源。
我用Optuna做超参数搜索,它的核心优势是支持剪枝——效果明显不好的试验提前终止,节省大量计算资源。
import optuna def objective(trial): # 定义搜索空间 lr = trial.suggest_float("learning_rate", 1e-6, 1e-4, log=True) batch_size = trial.suggest_categorical("batch_size", [16, 32, 64]) warmup = trial.suggest_float("warmup_ratio", 0.0, 0.3) dropout = trial.suggest_float("dropout", 0.0, 0.5) # 训练模型 config = TrainConfig( learning_rate=lr, batch_size=batch_size, warmup_ratio=warmup, dropout=dropout, epochs=5 ) # 每个epoch报告中间结果,支持剪枝 for epoch in range(config.epochs): val_acc = train_one_epoch_and_eval(config, epoch) trial.report(val_acc, epoch) if trial.should_prune(): raise optuna.TrialPruned() return val_acc # 创建研究并优化 study = optuna.create_study( direction="maximize", pruner=optuna.pruners.MedianPruner(n_startup_trials=5, n_warmup_steps=2) ) study.optimize(objective, n_trials=50, timeout=3600*8) # 输出最佳参数 print(study.best_params) print(study.best_value)这套流程跑下来,50次试验大概能覆盖大部分有希望的超参数组合,而且因为剪枝的存在,实际计算量可能只有完整跑50次的60%左右。
5. 模型部署与服务化:从notebook到线上服务
5.1 模型导出与格式转换
训练好的PyTorch模型不能直接部署,需要先导出成适合推理的格式。常见的选择有ONNX和TorchScript。ONNX的跨平台兼容性更好,TorchScript和PyTorch生态结合更紧密。我一般优先选ONNX,因为推理性能通常更好,而且可以用ONNX Runtime来跑,不依赖PyTorch。
import torch import torch.onnx # 加载训练好的模型 model = MyModel() model.load_state_dict(torch.load("best_model.pt")) model.eval() # 构造示例输入 dummy_input = torch.randint(0, 30000, (1, 128)) # 导出ONNX torch.onnx.export( model, dummy_input, "model.onnx", input_names=["input_ids"], output_names=["logits"], dynamic_axes={ "input_ids": {0: "batch_size", 1: "sequence_length"}, "logits": {0: "batch_size"} }, opset_version=14 )导出之后一定要验证ONNX模型的输出和原模型一致:
import onnxruntime as ort import numpy as np # 原模型推理 with torch.no_grad(): torch_output = model(dummy_input).numpy() # ONNX模型推理 session = ort.InferenceSession("model.onnx") onnx_output = session.run(None, {"input_ids": dummy_input.numpy()})[0] # 对比差异 diff = np.abs(torch_output - onnx_output).max() print(f"最大差异: {diff}") assert diff < 1e-4, "ONNX导出结果不一致,检查导出配置"实操心得:ONNX导出最容易出问题的地方是自定义算子。如果你的模型里用了PyTorch的非标准操作,导出时可能报错或者结果不对。解决办法是重写这部分逻辑,用ONNX支持的标准算子实现。我遇到过一次自定义的attention实现导出后结果偏差很大,后来改成标准的scaled_dot_product_attention就好了。
5.2 FastAPI服务封装
模型服务用FastAPI来写,轻量、性能好、自动生成API文档。核心要考虑的是:请求预处理、批量推理、并发控制、错误处理。
from fastapi import FastAPI, HTTPException from pydantic import BaseModel import onnxruntime as ort import numpy as np from transformers import AutoTokenizer import asyncio from concurrent.futures import ThreadPoolExecutor app = FastAPI(title="Text Classification Service") # 全局加载模型和分词器(只加载一次) session = ort.InferenceSession("model.onnx", providers=["CPUExecutionProvider"]) tokenizer = AutoTokenizer.from_pretrained("bert-base-chinese") executor = ThreadPoolExecutor(max_workers=4) class PredictRequest(BaseModel): text: str return_probabilities: bool = False class PredictResponse(BaseModel): label: str confidence: float probabilities: list = None LABELS = ["类别A", "类别B", "类别C"] # 根据实际任务定义 def preprocess(text: str): encoded = tokenizer( text, max_length=128, padding="max_length", truncation=True, return_tensors="np" ) return encoded["input_ids"].astype(np.int64) def postprocess(logits: np.ndarray): probs = softmax(logits, axis=-1)[0] pred_idx = int(np.argmax(probs)) return LABELS[pred_idx], float(probs[pred_idx]), probs.tolist() def softmax(x, axis=-1): e_x = np.exp(x - np.max(x, axis=axis, keepdims=True)) return e_x / e_x.sum(axis=axis, keepdims=True) @app.post("/predict", response_model=PredictResponse) async def predict(request: PredictRequest): if not request.text.strip(): raise HTTPException(status_code=400, detail="文本不能为空") try: input_ids = preprocess(request.text) # 在线程池中执行推理,避免阻塞事件循环 loop = asyncio.get_event_loop() logits = await loop.run_in_executor( executor, lambda: session.run(None, {"input_ids": input_ids})[0] ) label, confidence, probs = postprocess(logits) response = PredictResponse(label=label, confidence=confidence) if request.return_probabilities: response.probabilities = probs return response except Exception as e: raise HTTPException(status_code=500, detail=f"推理失败: {str(e)}") @app.get("/health") async def health(): return {"status": "healthy"}这个服务有几个关键设计点:模型在启动时加载一次,避免每次请求都重新加载;推理放在线程池里执行,避免阻塞FastAPI的事件循环;输入做了校验,空文本直接返回400;异常做了捕获,不会把内部错误暴露给调用方。
5.3 性能优化:延迟从500ms降到50ms
模型服务上线后,最常见的抱怨就是"太慢了"。我拿一个实际案例来说:一个文本分类服务,初始版本P99延迟500ms,经过优化降到了50ms。做了哪些事?
第一,模型量化。把FP32模型转成INT8,推理速度提升2-3倍,精度损失通常在1%以内。ONNX Runtime支持动态量化:
from onnxruntime.quantization import quantize_dynamic, QuantType quantize_dynamic( "model.onnx", "model_quantized.onnx", weight_type=QuantType.QUInt8 )第二,批处理。单条推理GPU利用率极低,把多个请求攒成一批一起推理,吞吐量能提升5-10倍。实现方式是在服务层加一个缓冲队列,攒够batch_size或者超时(比如10ms)就触发一次推理。
第三,输入长度截断。很多请求的文本其实很短,但为了统一都padding到128。改成动态padding,按batch内最长序列来padding,能省不少计算。
第四,ONNX Runtime配置调优。设置合适的线程数、开启图优化:
options = ort.SessionOptions() options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL options.intra_op_num_threads = 4 options.inter_op_num_threads = 2 session = ort.InferenceSession("model_quantized.onnx", options, providers=["CPUExecutionProvider"])| 优化措施 | 延迟变化 | 精度变化 | 实施难度 |
|---|---|---|---|
| 基线 | 500ms | 基准 | - |
| 动态padding | 350ms | 无 | 低 |
| INT8量化 | 150ms | -0.5% | 低 |
| 批处理 | 80ms | 无 | 中 |
| 线程调优 | 50ms | 无 | 低 |
注意:量化后的模型一定要重新评估精度。我遇到过一次量化后某个类别的F1掉了5个点,后来发现是那个类别的样本在量化后区分度下降。解决办法是对量化后的模型做一轮校准,用代表性数据调整量化参数。
6. 监控与运维:上线只是开始
6.1 服务监控的核心指标
服务上线之后,你需要盯着这些指标:
- 请求量(QPS):每秒多少请求,有没有突增突降
- 延迟分布(P50/P95/P99):不要只看平均值,尾部延迟才是用户体验的关键
- 错误率:4xx和5xx的比例,分别代表客户端错误和服务端错误
- 资源使用:CPU、内存、GPU利用率,有没有瓶颈
- 模型输出分布:各类别的预测比例,有没有异常偏移
用Prometheus采集指标,Grafana做可视化。FastAPI集成Prometheus很简单:
from prometheus_fastapi_instrumentator import Instrumentator app = FastAPI() Instrumentator().instrument(app).expose(app)这行代码会自动采集请求量、延迟、状态码等指标,暴露在/metrics端点。然后在Prometheus配置里加上这个端点,Grafana里导入对应的Dashboard模板,监控面板就搭好了。
6.2 模型漂移检测
模型漂移是指线上数据的分布发生了变化,导致模型效果下降。这是最隐蔽的问题——服务本身运行正常,延迟正常,错误率正常,但预测质量在悄悄下降。
检测漂移的方法有两种:一是监控输入特征的分布,二是监控输出预测的分布。输入分布可以用PSI(Population Stability Index)来衡量:
import numpy as np def calculate_psi(expected, actual, buckets=10): """计算PSI,衡量两个分布的差异""" # 基于expected分布分桶 breakpoints = np.percentile(expected, np.linspace(0, 100, buckets + 1)) breakpoints[0] = -np.inf breakpoints[-1] = np.inf expected_counts = np.histogram(expected, bins=breakpoints)[0] actual_counts = np.histogram(actual, bins=breakpoints)[0] expected_pct = expected_counts / len(expected) + 1e-6 actual_pct = actual_counts / len(actual) + 1e-6 psi = np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct)) return psi # PSI < 0.1: 分布稳定 # 0.1 <= PSI < 0.25: 轻微漂移,需要关注 # PSI >= 0.25: 显著漂移,需要重新训练输出分布的监控更直接:统计各类别预测比例,和训练时的基线对比。如果某个类别的比例突然从10%涨到30%,大概率是数据分布变了。
6.3 常见线上问题排查实录
问题一:服务突然变慢,延迟从50ms涨到500ms。
排查思路:先看资源监控,CPU和内存是否正常。如果资源正常,看请求量是否突增。如果请求量正常,看输入数据是否变化——比如突然来了很多长文本,导致推理时间增加。我遇到过一次是因为上游系统改了一个字段格式,导致我们的预处理逻辑走了慢路径。
问题二:错误率突然上升,大量500错误。
排查思路:先看错误日志,定位异常类型。常见原因包括:模型文件被误删、依赖库版本冲突、内存溢出。我遇到过一次是ONNX Runtime的版本升级导致API不兼容,回滚版本后恢复。
问题三:模型预测结果异常,大量请求被分到同一个类别。
排查思路:检查输入数据是否正常,有没有大量空文本或乱码。检查模型文件是否完整,有没有被覆盖。检查预处理逻辑是否和训练时一致。我遇到过一次是分词器的配置变了,导致输入ID全部错位。
| 问题现象 | 可能原因 | 排查方法 | 解决方案 |
|---|---|---|---|
| 延迟升高 | 输入变长/资源不足/请求突增 | 看监控、看日志、看输入分布 | 动态padding/扩容/限流 |
| 错误率上升 | 模型文件问题/依赖冲突/内存溢出 | 看错误日志、看资源监控 | 回滚/修复依赖/增加内存 |
| 预测偏移 | 数据漂移/预处理不一致/模型损坏 | 对比输入分布、检查预处理 | 重新训练/修复预处理/恢复模型 |
| 服务无响应 | 死锁/线程池耗尽/GC停顿 | 看线程栈、看GC日志 | 调整线程池/优化内存 |
实操心得:线上问题排查最重要的是保留现场。服务出问题时,不要急着重启,先把日志、监控快照、内存dump保存下来。重启能解决80%的问题,但如果不找到根因,剩下20%的问题会反复出现。我习惯在服务里加一个
/debug端点,能快速导出当前的状态信息。
7. 一些踩坑之后的真心话
做AI工程这几年,最大的体会是:技术选型的重要性远低于工程规范的重要性。你用PyTorch还是TensorFlow,用MLflow还是W&B,这些选择对项目成败的影响其实很小。真正决定成败的是:数据版本有没有管好、实验记录有没有做全、部署流程有没有标准化、监控有没有到位。这些看起来"不酷"的工程实践,才是保证项目长期稳定运行的关键。
另一个体会是:不要过早优化。我见过太多团队在项目初期就花大量时间搭建完美的架构,结果业务需求一变,架构全白搭。正确的做法是先跑通最小闭环,然后根据实际遇到的问题逐步优化。性能不够了再优化性能,数据量大了再考虑分布式,团队大了再考虑权限管理。每一步优化都要有明确的驱动因素,而不是"觉得应该这样做"。
最后一个建议:把踩过的坑记录下来。我维护了一个内部Wiki,每次遇到线上问题,排查完之后都会写一篇复盘,包括问题现象、排查过程、根因分析、解决方案、预防措施。这个Wiki现在有上百篇文档,新同事入职的时候先读一遍,能避开80%的常见问题。这个习惯看起来简单,但坚持下来价值巨大。
这个内容后续还可以这样扩展:如果你对模型压缩感兴趣,可以深入研究量化、剪枝、蒸馏的具体实现;如果你对大规模部署感兴趣,可以研究Kubernetes上的模型服务编排;如果你对自动化感兴趣,可以研究CI/CD流水线怎么和模型训练部署结合。每个方向都够写好几篇了,有机会再展开聊。