1. 从零搭建AI工程体系为什么我劝你别一上来就调包很多人第一次接触AI工程脑子里想的都是“我装个PyTorch找个开源模型跑个demo出来就完事了”。我刚开始也是这么想的直到真正接手一个要上线的项目才发现从“能跑通”到“能扛住”之间隔着一整套工程体系。ai-engineering-from-scratch这个标题说的不是从零训练一个大模型而是从零把AI工程化的能力搭起来——数据怎么管、实验怎么追踪、模型怎么部署、线上怎么监控这一整套东西才是AI工程师真正每天要面对的事。这篇文章适合谁看如果你已经会写Python、跑过几个notebook、对机器学习有基本概念但一到“要把模型放到生产环境”就发怵那这篇就是写给你的。如果你是完全零基础也没关系我会把每个环节为什么这么做讲清楚你跟着走一遍至少能建立起完整的工程认知。我不会只给你一堆工具名字而是把选型逻辑、踩坑经验、参数计算都摊开讲让你看完能直接动手搭一套属于自己的AI工程骨架。先说一个我踩过的坑。早期我做项目数据、代码、模型权重全塞在一个文件夹里实验记录靠Excel部署靠手动scp。结果就是三个月后我自己都复现不出当时最好的那个模型因为忘了当时用的是哪版数据、哪个超参。这不是笑话是很多小团队的真实写照。ai-engineering-from-scratch的核心价值就是让你从第一天起就避开这些坑用工程化的方式管理AI项目的全生命周期。2. 整体架构设计先想清楚数据怎么流再动手写代码2.1 为什么架构设计要放在写代码之前我见过太多人一上来就pip install然后开始写训练脚本写到一半发现数据格式不对又回头改数据加载改完发现实验没法对比又加日志加完日志发现部署时依赖冲突。这种“边写边补”的方式在小demo里没问题但一旦项目稍微复杂一点就会变成一团乱麻。正确的做法是先画一张数据流图。不用很复杂就在纸上画原始数据从哪来、经过哪些清洗步骤、存到哪里、训练时怎么读、模型产出存哪里、推理时怎么加载、线上请求怎么进来、结果怎么返回、监控指标怎么收集。这张图想清楚了后面写代码就是填空。我自己的习惯是用一个config.yaml把所有路径、参数、版本号都管起来代码里不出现任何硬编码的路径。这样做的好处是换环境只需要改配置文件不用动代码。比如数据路径、模型保存路径、日志路径、数据库连接全部走配置。这个习惯看起来小但能省掉后面无数麻烦。2.2 目录结构怎么定才能让三个月后的自己看得懂我推荐一个经过多个项目验证的目录结构你可以直接抄project/ ├── configs/ # 配置文件 │ ├── base.yaml │ └── exp001.yaml ├── data/ # 数据目录不提交到git │ ├── raw/ │ ├── processed/ │ └── interim/ ├── src/ # 源代码 │ ├── data/ # 数据加载与清洗 │ ├── features/ # 特征工程 │ ├── models/ # 模型定义 │ ├── training/ # 训练逻辑 │ ├── evaluation/ # 评估逻辑 │ └── serving/ # 推理服务 ├── experiments/ # 实验记录 ├── notebooks/ # 探索性分析 ├── tests/ # 单元测试 ├── requirements.txt └── README.md这个结构的关键在于职责分离。src/data只管读数据和清洗src/models只管模型结构src/training只管训练循环。这样当你需要换模型时只改models数据管道不动换数据源时只改data模型不动。这种解耦是工程化的基础。注意data/和experiments/一定要加到.gitignore里。数据文件大实验产出多提交到git会让仓库爆炸。但configs/必须提交因为配置是代码的一部分。2.3 工具选型别追新选社区活跃且你能hold住的AI工程领域工具更新极快今天火的东西明天可能就没人维护了。我的选型原则是社区活跃、文档齐全、有商业公司或大机构背书、上手成本低。基于这个原则我推荐一套经过验证的组合环节推荐工具选它的理由实验追踪MLflow轻量、可本地部署、API简单数据版本DVC和git无缝集成、支持多种存储后端配置管理Hydra支持配置组合、命令行覆盖模型服务FastAPI ONNX Runtime轻量、高性能、易容器化监控Prometheus Grafana事实标准、生态完善这套组合的好处是每个工具都只解决一个问题不搞大而全。你可以一个一个引入不用一次性全上。比如先上MLflow做实验追踪等数据版本问题出现了再上DVC。渐进式引入学习成本低团队接受度高。3. 数据管道AI工程里最脏最累但最重要的部分3.1 数据版本管理为什么git不够用git管代码很好但管数据不行。原因很简单数据文件太大而且经常变。你不可能每次改一条数据就commit一次仓库会爆炸。但如果不做版本管理你就无法复现实验——这是AI工程最致命的问题。DVC的思路是数据文件本身不进gitgit只存一个很小的.dvc文件里面记录了数据文件的哈希值和存储位置。数据文件存在本地磁盘或对象存储里。这样你切换git分支时dvc checkout就能把对应版本的数据拉出来。具体操作# 初始化 git init dvc init # 添加数据目录 dvc add data/raw # 这会生成 data/raw.dvc 文件把它提交到git git add data/raw.dvc .gitignore git commit -m add raw data # 切换版本时 git checkout commit dvc checkout实测下来这套流程在团队协作里非常稳。每个人拉代码后跑一下dvc pull数据就同步了。不用再问“你那份数据是哪版的”。3.2 数据清洗的工程化写法数据清洗代码最容易写成一次性脚本跑完就扔。但工程化的写法是把清洗逻辑拆成一个个纯函数每个函数只做一件事输入输出都是DataFrame。这样你可以单独测试每个函数也可以灵活组合。# src/data/clean.py import pandas as pd def drop_duplicates(df: pd.DataFrame, subset: list) - pd.DataFrame: 去重保留第一条 return df.drop_duplicates(subsetsubset, keepfirst) def fill_missing(df: pd.DataFrame, col: str, strategy: str median) - pd.DataFrame: 填充缺失值 if strategy median: df[col] df[col].fillna(df[col].median()) elif strategy mean: df[col] df[col].fillna(df[col].mean()) elif strategy zero: df[col] df[col].fillna(0) return df def clip_outliers(df: pd.DataFrame, col: str, lower_q: float 0.01, upper_q: float 0.99) - pd.DataFrame: 按分位数截断异常值 lower df[col].quantile(lower_q) upper df[col].quantile(upper_q) df[col] df[col].clip(lower, upper) return df这种写法的好处是每个函数都可以单独写单元测试。比如测试clip_outliers你造一个含极端值的小DataFrame跑一下断言结果在分位数范围内。测试通过了这个函数就可以放心用。实操心得清洗函数一定要幂等。也就是说同一个数据跑两遍清洗结果应该一样。如果做不到幂等说明你的函数有副作用比如依赖了外部状态。这种函数在管道里会出大问题。3.3 特征工程的版本控制特征工程比数据清洗更复杂因为特征会随着业务理解不断迭代。今天你觉得某个特征有用明天可能发现它泄露了标签。所以特征也需要版本控制。我的做法是把特征计算逻辑写成独立的模块每个特征一个函数函数上标注版本号和依赖的数据版本。然后用一个features.yaml记录当前使用的特征列表和版本。# configs/features.yaml features: - name: user_age version: v1 source: data/processed/users.parquet - name: order_count_7d version: v2 source: data/processed/orders.parquet params: window_days: 7这样当特征出问题时你可以快速定位是哪个特征、哪个版本、依赖哪份数据。排查效率比翻代码高十倍。4. 实验管理与模型训练让每次实验都可追溯4.1 MLflow实战从手动记录到自动追踪没有实验追踪的时候我的记录方式是在notebook里跑完把准确率抄到Excel里模型文件存成model_final_v2_real_final.pkl。这种方式的结局就是一周后完全不知道哪个文件对应哪次实验。MLflow解决的就是这个问题。你只需要在训练脚本里加几行代码import mlflow import mlflow.sklearn mlflow.set_experiment(my_project) with mlflow.start_run(run_nameexp001): # 记录参数 mlflow.log_param(learning_rate, 0.01) mlflow.log_param(n_estimators, 100) # 训练模型 model train_model(X_train, y_train) # 记录指标 mlflow.log_metric(accuracy, 0.92) mlflow.log_metric(f1, 0.89) # 保存模型 mlflow.sklearn.log_model(model, model) # 保存特征重要性图 mlflow.log_artifact(feature_importance.png)跑完之后打开MLflow UI所有实验一目了然。你可以按指标排序找到最好的那次直接下载对应的模型文件。更重要的是每个实验的参数、指标、模型、图表都绑在一起复现的时候直接看记录就行。注意MLflow的tracking server最好单独部署不要用本地文件存储。团队协作时大家连同一个server实验记录才能共享。本地文件存储只适合个人项目。4.2 超参数搜索的工程化做法超参数搜索最容易犯的错是在notebook里写个for循环跑一晚上第二天看结果。这种做法的问题是如果中间断了前面的结果全丢而且没法并行效率低。工程化的做法是用Optuna或Ray Tune配合MLflow记录。Optuna的用法很简单import optuna def objective(trial): lr trial.suggest_float(lr, 1e-4, 1e-1, logTrue) n_estimators trial.suggest_int(n_estimators, 50, 500) with mlflow.start_run(nestedTrue): mlflow.log_params({lr: lr, n_estimators: n_estimators}) model train_model(X_train, y_train, lrlr, n_estimatorsn_estimators) acc evaluate(model, X_val, y_val) mlflow.log_metric(accuracy, acc) return acc study optuna.create_study(directionmaximize) study.optimize(objective, n_trials100)Optuna会自动记录每次试验的参数和结果支持剪枝提前终止表现差的试验还支持并行。配合MLflow每次试验的细节都能追溯。4.3 训练脚本的标准化模板我习惯把训练脚本写成一个标准模板所有项目都套这个模板。模板的核心是配置驱动、日志完善、检查点自动保存。# src/training/train.py import logging import yaml import mlflow from pathlib import Path def load_config(path): with open(path) as f: return yaml.safe_load(f) def setup_logging(log_path): logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(log_path), logging.StreamHandler() ] ) def train(config): setup_logging(config[log_path]) logging.info(fStarting training with config: {config}) # 加载数据 train_data load_data(config[data][train_path]) val_data load_data(config[data][val_path]) # 训练 model build_model(config[model]) for epoch in range(config[training][epochs]): train_loss train_epoch(model, train_data) val_loss validate(model, val_data) logging.info(fEpoch {epoch}: train_loss{train_loss:.4f}, val_loss{val_loss:.4f}) # 保存检查点 if val_loss best_val_loss: best_val_loss val_loss save_checkpoint(model, config[checkpoint_path]) return model if __name__ __main__: config load_config(configs/base.yaml) train(config)这个模板的关键点是所有参数从配置来日志同时输出到文件和终端检查点自动保存最好的模型。这样你跑训练的时候可以去干别的事回来看日志就行。5. 模型部署与服务化从notebook到线上API5.1 为什么FastAPI是AI服务的好选择模型训练完下一步是让其他系统能调用它。最简单的方式是写个Flask应用但Flask是同步的并发能力弱。FastAPI是异步的性能好很多而且自动生成API文档调试方便。一个最小的模型服务长这样# src/serving/app.py from fastapi import FastAPI from pydantic import BaseModel import joblib import numpy as np app FastAPI() model joblib.load(models/model.pkl) class PredictRequest(BaseModel): features: list[float] class PredictResponse(BaseModel): prediction: float probability: float app.post(/predict, response_modelPredictResponse) async def predict(request: PredictRequest): X np.array(request.features).reshape(1, -1) pred model.predict(X)[0] prob model.predict_proba(X)[0].max() return PredictResponse(predictionfloat(pred), probabilityfloat(prob)) app.get(/health) async def health(): return {status: ok}跑起来用uvicorn src.serving.app:app --host 0.0.0.0 --port 8000然后访问/docs就能看到自动生成的API文档可以直接在浏览器里测试。实操心得模型加载一定要放在应用启动时不要放在请求处理函数里。放在请求里每次都要重新加载模型性能极差。如果模型很大加载慢可以用lazy loading第一次请求时加载之后缓存。5.2 模型格式选择pickle、ONNX还是TorchScript模型保存格式直接影响部署的灵活性和性能。我对比过几种常见格式格式优点缺点适用场景pickle简单、Python原生依赖训练时的环境、不安全内部快速原型ONNX跨框架、跨语言、性能好转换可能丢精度生产环境、多语言调用TorchScriptPyTorch原生、支持动态图只适用于PyTorchPyTorch模型部署SavedModelTensorFlow原生、生态好只适用于TFTensorFlow模型部署我的建议是如果模型要上生产优先转ONNX。ONNX Runtime的推理速度通常比原生框架快而且可以在C、Java、C#里调用不依赖Python环境。转换也不难import torch import torch.onnx # 假设model是PyTorch模型 dummy_input torch.randn(1, input_dim) torch.onnx.export( model, dummy_input, model.onnx, input_names[input], output_names[output], dynamic_axes{input: {0: batch_size}, output: {0: batch_size}} )转换后可以用ONNX Runtime加载import onnxruntime as ort session ort.InferenceSession(model.onnx) input_name session.get_inputs()[0].name output session.run(None, {input_name: X.astype(np.float32)})实测下来ONNX Runtime的推理延迟比PyTorch原生低30%左右内存占用也小。5.3 容器化部署Dockerfile怎么写才不踩坑模型服务要部署到服务器Docker是最方便的方式。但写Dockerfile有几个坑第一基础镜像不要用python:3.9这种全量镜像太大。用slim版本能小一半。第二依赖安装要分层。先复制requirements.txt安装依赖再复制代码。这样改代码时不用重装依赖构建快。第三模型文件不要打进镜像。镜像应该只包含代码和依赖模型文件通过挂载卷或启动时下载。一个经过验证的DockerfileFROM python:3.9-slim WORKDIR /app # 先装依赖 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 再复制代码 COPY src/ ./src/ COPY configs/ ./configs/ # 模型通过卷挂载 VOLUME /app/models EXPOSE 8000 CMD [uvicorn, src.serving.app:app, --host, 0.0.0.0, --port, 8000]构建和运行docker build -t my-ai-service . docker run -d -p 8000:8000 -v /path/to/models:/app/models my-ai-service注意requirements.txt里要固定版本号比如fastapi0.104.1不要写fastapi。不固定版本的话今天构建能跑明天可能就挂了因为依赖更新了。6. 监控与迭代上线只是开始不是结束6.1 线上监控要盯哪些指标模型上线后最怕的是“静默失败”——服务没挂但预测结果越来越差。所以监控要分两层系统层和模型层。系统层指标用Prometheus收集包括请求量、延迟、错误率、CPU/内存使用率。这些指标Grafana有现成模板配一下就行。模型层指标需要自己埋点包括预测分布、特征分布、置信度分布。比如你可以记录每次请求的预测值然后定期统计分布。如果发现预测分布突然偏移说明数据分布变了模型可能失效了。from prometheus_client import Histogram, Counter PREDICTION_HIST Histogram( model_prediction_value, Distribution of model predictions, buckets[0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0] ) REQUEST_COUNT Counter( model_request_total, Total model requests, [status] ) app.post(/predict) async def predict(request: PredictRequest): try: result model.predict(request.features) PREDICTION_HIST.observe(result) REQUEST_COUNT.labels(statussuccess).inc() return result except Exception as e: REQUEST_COUNT.labels(statuserror).inc() raise6.2 数据漂移检测的简单实现数据漂移是指线上数据的分布和训练数据不一致。检测方法有很多最简单的是PSIPopulation Stability Index。PSI的计算方式是把训练数据和线上数据按特征分桶计算每个桶的占比差异然后加权求和。import numpy as np def calculate_psi(expected, actual, buckets10): 计算PSIexpected是训练数据actual是线上数据 breakpoints np.percentile(expected, np.linspace(0, 100, buckets 1)) breakpoints[0] -np.inf breakpoints[-1] np.inf expected_percents np.histogram(expected, breakpoints)[0] / len(expected) actual_percents np.histogram(actual, breakpoints)[0] / len(actual) # 避免除零 expected_percents np.where(expected_percents 0, 0.0001, expected_percents) actual_percents np.where(actual_percents 0, 0.0001, actual_percents) psi np.sum((actual_percents - expected_percents) * np.log(actual_percents / expected_percents)) return psiPSI的判断标准小于0.1说明分布稳定0.1到0.25说明有轻微漂移大于0.25说明漂移严重需要重新训练模型。实操心得PSI检测要定期跑比如每天跑一次。不要等出问题了才查。我习惯把PSI计算做成定时任务结果写到数据库Grafana里画成趋势图。这样漂移是渐变还是突变一目了然。6.3 模型迭代的闭环流程模型迭代不是“重新训练一次”那么简单而是一个闭环监控发现问题、分析原因、准备数据、训练新模型、评估、上线、继续监控。这个闭环里最容易忽略的是评估环节。新模型不能只看离线指标还要做A/B测试。我的做法是新模型上线后先切10%的流量过去对比新旧模型的线上指标。如果新模型在关键指标上不差于旧模型再逐步扩大流量。A/B测试的流量切分可以在网关层做也可以在服务层做。服务层做更灵活import random MODEL_A load_model(models/model_a.onnx) MODEL_B load_model(models/model_b.onnx) app.post(/predict) async def predict(request: PredictRequest): if random.random() 0.1: model MODEL_B version B else: model MODEL_A version A result model.predict(request.features) # 记录版本方便后续分析 log_prediction(version, request.features, result) return result跑一周后对比A和B的线上指标决定是否全量切换。7. 常见问题与排查技巧实录7.1 训练时loss不下降怎么排查这是最常见的问题。排查顺序应该是先看数据再看模型最后看超参。第一步检查数据。把一批数据喂给模型看输出是否合理。如果模型输出全是同一个值说明数据可能有问题比如标签全是0或者特征全是NaN。第二步检查模型。用一个极小的数据集比如10条过拟合一下。如果模型连10条数据都拟合不了说明模型结构有问题比如层数太少、激活函数不对。第三步检查超参。学习率太大loss会震荡学习率太小loss下降极慢。可以试试用学习率finder找一个合适的初始学习率。# 简单的学习率finder lrs np.logspace(-5, -1, 100) losses [] for lr in lrs: model build_model() optimizer torch.optim.SGD(model.parameters(), lrlr) loss train_one_step(model, optimizer, data) losses.append(loss) # 画图找loss下降最快的点 import matplotlib.pyplot as plt plt.plot(lrs, losses) plt.xscale(log) plt.show()7.2 线上服务延迟高从哪里入手延迟高的问题排查思路是先定位瓶颈在哪一层再针对性优化。排查点可能原因优化方法网络层请求体太大、网络带宽不足压缩请求、升级带宽应用层同步阻塞、GIL限制用异步框架、多进程模型层模型太大、计算量大模型量化、剪枝、用ONNX数据层特征查询慢加缓存、预计算特征我遇到最多的是模型层的问题。一个BERT模型原始PyTorch推理要200ms转ONNX后降到80ms再量化到INT8后降到30ms。所以模型优化是收益最大的方向。7.3 依赖冲突怎么解决Python依赖冲突是AI工程的经典问题。项目A需要numpy 1.19项目B需要numpy 1.21装在一起就炸。解决方案有三个第一用虚拟环境隔离。每个项目一个venv互不影响。这是最基本的做法。第二用Docker。每个服务一个容器依赖完全隔离。这是生产环境的标配。第三用poetry或pipenv管理依赖。它们会生成lock文件确保每次安装的版本一致。我个人的习惯是开发时用conda创建虚拟环境部署时用Docker。conda的好处是能装非Python依赖比如CUDA、cuDNN这对GPU训练很重要。# 创建环境 conda create -n my_project python3.9 conda activate my_project # 安装PyTorch会自动装CUDA依赖 conda install pytorch torchvision torchaudio cudatoolkit11.3 -c pytorch # 其他依赖用pip pip install -r requirements.txt7.4 常见问题速查表问题现象可能原因快速排查方法训练loss为NaN学习率太大、数据有inf降低学习率、检查数据验证集指标远低于训练集过拟合加正则化、加数据、早停线上预测结果和离线不一致特征处理不一致对比线上线下特征管道服务启动报错找不到模型路径不对、模型未挂载检查挂载卷和配置路径推理速度慢模型未优化、批处理不当转ONNX、加批处理内存泄漏全局变量累积、缓存未清理用memory_profiler排查8. 我在这套流程里踩过的几个坑第一个坑是过早优化。刚开始做AI工程时我总想一步到位把MLflow、DVC、Airflow全装上。结果光是配这些工具就花了两周真正写模型的时间反而少了。后来我学乖了先跑通最小闭环数据加载、训练、保存模型、FastAPI服务。等这个闭环跑顺了再逐步加实验追踪、数据版本、监控。工具是解决问题的不是制造问题的。第二个坑是忽略测试。AI代码也需要单元测试尤其是数据清洗和特征工程部分。这些代码逻辑复杂容易出错而且出错后很难发现。我现在的习惯是每个清洗函数都写测试用pytest跑。测试数据不用多几条就够关键是覆盖边界情况。第三个坑是配置散落。早期我把数据库密码、API密钥写在代码里后来代码传到git密钥就泄露了。正确的做法是用环境变量或密钥管理服务。配置文件中只放非敏感信息敏感信息通过环境变量注入。import os DB_PASSWORD os.environ.get(DB_PASSWORD) API_KEY os.environ.get(API_KEY)第四个坑是不做回滚预案。新模型上线后效果变差想回滚却发现旧模型没保存。所以每次上线新模型一定要保留旧模型和对应的配置确保能一键回滚。我的做法是模型文件按版本号命名配置也按版本号存档上线时记录当前版本回滚时切回上一个版本。这套从零搭建AI工程体系的流程我前后迭代了两年多现在基本稳定了。新项目启动时我直接复制这套骨架改改配置就能跑。省下来的时间可以花在真正重要的事情上理解业务、优化模型、分析数据。工程化不是为了炫技是为了让你能专注于创造价值的部分。 从零搭建AI工程体系：数据管道、实验追踪与模型部署实战
1. 从零搭建AI工程体系为什么我劝你别一上来就调包很多人第一次接触AI工程脑子里想的都是“我装个PyTorch找个开源模型跑个demo出来就完事了”。我刚开始也是这么想的直到真正接手一个要上线的项目才发现从“能跑通”到“能扛住”之间隔着一整套工程体系。ai-engineering-from-scratch这个标题说的不是从零训练一个大模型而是从零把AI工程化的能力搭起来——数据怎么管、实验怎么追踪、模型怎么部署、线上怎么监控这一整套东西才是AI工程师真正每天要面对的事。这篇文章适合谁看如果你已经会写Python、跑过几个notebook、对机器学习有基本概念但一到“要把模型放到生产环境”就发怵那这篇就是写给你的。如果你是完全零基础也没关系我会把每个环节为什么这么做讲清楚你跟着走一遍至少能建立起完整的工程认知。我不会只给你一堆工具名字而是把选型逻辑、踩坑经验、参数计算都摊开讲让你看完能直接动手搭一套属于自己的AI工程骨架。先说一个我踩过的坑。早期我做项目数据、代码、模型权重全塞在一个文件夹里实验记录靠Excel部署靠手动scp。结果就是三个月后我自己都复现不出当时最好的那个模型因为忘了当时用的是哪版数据、哪个超参。这不是笑话是很多小团队的真实写照。ai-engineering-from-scratch的核心价值就是让你从第一天起就避开这些坑用工程化的方式管理AI项目的全生命周期。2. 整体架构设计先想清楚数据怎么流再动手写代码2.1 为什么架构设计要放在写代码之前我见过太多人一上来就pip install然后开始写训练脚本写到一半发现数据格式不对又回头改数据加载改完发现实验没法对比又加日志加完日志发现部署时依赖冲突。这种“边写边补”的方式在小demo里没问题但一旦项目稍微复杂一点就会变成一团乱麻。正确的做法是先画一张数据流图。不用很复杂就在纸上画原始数据从哪来、经过哪些清洗步骤、存到哪里、训练时怎么读、模型产出存哪里、推理时怎么加载、线上请求怎么进来、结果怎么返回、监控指标怎么收集。这张图想清楚了后面写代码就是填空。我自己的习惯是用一个config.yaml把所有路径、参数、版本号都管起来代码里不出现任何硬编码的路径。这样做的好处是换环境只需要改配置文件不用动代码。比如数据路径、模型保存路径、日志路径、数据库连接全部走配置。这个习惯看起来小但能省掉后面无数麻烦。2.2 目录结构怎么定才能让三个月后的自己看得懂我推荐一个经过多个项目验证的目录结构你可以直接抄project/ ├── configs/ # 配置文件 │ ├── base.yaml │ └── exp001.yaml ├── data/ # 数据目录不提交到git │ ├── raw/ │ ├── processed/ │ └── interim/ ├── src/ # 源代码 │ ├── data/ # 数据加载与清洗 │ ├── features/ # 特征工程 │ ├── models/ # 模型定义 │ ├── training/ # 训练逻辑 │ ├── evaluation/ # 评估逻辑 │ └── serving/ # 推理服务 ├── experiments/ # 实验记录 ├── notebooks/ # 探索性分析 ├── tests/ # 单元测试 ├── requirements.txt └── README.md这个结构的关键在于职责分离。src/data只管读数据和清洗src/models只管模型结构src/training只管训练循环。这样当你需要换模型时只改models数据管道不动换数据源时只改data模型不动。这种解耦是工程化的基础。注意data/和experiments/一定要加到.gitignore里。数据文件大实验产出多提交到git会让仓库爆炸。但configs/必须提交因为配置是代码的一部分。2.3 工具选型别追新选社区活跃且你能hold住的AI工程领域工具更新极快今天火的东西明天可能就没人维护了。我的选型原则是社区活跃、文档齐全、有商业公司或大机构背书、上手成本低。基于这个原则我推荐一套经过验证的组合环节推荐工具选它的理由实验追踪MLflow轻量、可本地部署、API简单数据版本DVC和git无缝集成、支持多种存储后端配置管理Hydra支持配置组合、命令行覆盖模型服务FastAPI ONNX Runtime轻量、高性能、易容器化监控Prometheus Grafana事实标准、生态完善这套组合的好处是每个工具都只解决一个问题不搞大而全。你可以一个一个引入不用一次性全上。比如先上MLflow做实验追踪等数据版本问题出现了再上DVC。渐进式引入学习成本低团队接受度高。3. 数据管道AI工程里最脏最累但最重要的部分3.1 数据版本管理为什么git不够用git管代码很好但管数据不行。原因很简单数据文件太大而且经常变。你不可能每次改一条数据就commit一次仓库会爆炸。但如果不做版本管理你就无法复现实验——这是AI工程最致命的问题。DVC的思路是数据文件本身不进gitgit只存一个很小的.dvc文件里面记录了数据文件的哈希值和存储位置。数据文件存在本地磁盘或对象存储里。这样你切换git分支时dvc checkout就能把对应版本的数据拉出来。具体操作# 初始化 git init dvc init # 添加数据目录 dvc add data/raw # 这会生成 data/raw.dvc 文件把它提交到git git add data/raw.dvc .gitignore git commit -m add raw data # 切换版本时 git checkout commit dvc checkout实测下来这套流程在团队协作里非常稳。每个人拉代码后跑一下dvc pull数据就同步了。不用再问“你那份数据是哪版的”。3.2 数据清洗的工程化写法数据清洗代码最容易写成一次性脚本跑完就扔。但工程化的写法是把清洗逻辑拆成一个个纯函数每个函数只做一件事输入输出都是DataFrame。这样你可以单独测试每个函数也可以灵活组合。# src/data/clean.py import pandas as pd def drop_duplicates(df: pd.DataFrame, subset: list) - pd.DataFrame: 去重保留第一条 return df.drop_duplicates(subsetsubset, keepfirst) def fill_missing(df: pd.DataFrame, col: str, strategy: str median) - pd.DataFrame: 填充缺失值 if strategy median: df[col] df[col].fillna(df[col].median()) elif strategy mean: df[col] df[col].fillna(df[col].mean()) elif strategy zero: df[col] df[col].fillna(0) return df def clip_outliers(df: pd.DataFrame, col: str, lower_q: float 0.01, upper_q: float 0.99) - pd.DataFrame: 按分位数截断异常值 lower df[col].quantile(lower_q) upper df[col].quantile(upper_q) df[col] df[col].clip(lower, upper) return df这种写法的好处是每个函数都可以单独写单元测试。比如测试clip_outliers你造一个含极端值的小DataFrame跑一下断言结果在分位数范围内。测试通过了这个函数就可以放心用。实操心得清洗函数一定要幂等。也就是说同一个数据跑两遍清洗结果应该一样。如果做不到幂等说明你的函数有副作用比如依赖了外部状态。这种函数在管道里会出大问题。3.3 特征工程的版本控制特征工程比数据清洗更复杂因为特征会随着业务理解不断迭代。今天你觉得某个特征有用明天可能发现它泄露了标签。所以特征也需要版本控制。我的做法是把特征计算逻辑写成独立的模块每个特征一个函数函数上标注版本号和依赖的数据版本。然后用一个features.yaml记录当前使用的特征列表和版本。# configs/features.yaml features: - name: user_age version: v1 source: data/processed/users.parquet - name: order_count_7d version: v2 source: data/processed/orders.parquet params: window_days: 7这样当特征出问题时你可以快速定位是哪个特征、哪个版本、依赖哪份数据。排查效率比翻代码高十倍。4. 实验管理与模型训练让每次实验都可追溯4.1 MLflow实战从手动记录到自动追踪没有实验追踪的时候我的记录方式是在notebook里跑完把准确率抄到Excel里模型文件存成model_final_v2_real_final.pkl。这种方式的结局就是一周后完全不知道哪个文件对应哪次实验。MLflow解决的就是这个问题。你只需要在训练脚本里加几行代码import mlflow import mlflow.sklearn mlflow.set_experiment(my_project) with mlflow.start_run(run_nameexp001): # 记录参数 mlflow.log_param(learning_rate, 0.01) mlflow.log_param(n_estimators, 100) # 训练模型 model train_model(X_train, y_train) # 记录指标 mlflow.log_metric(accuracy, 0.92) mlflow.log_metric(f1, 0.89) # 保存模型 mlflow.sklearn.log_model(model, model) # 保存特征重要性图 mlflow.log_artifact(feature_importance.png)跑完之后打开MLflow UI所有实验一目了然。你可以按指标排序找到最好的那次直接下载对应的模型文件。更重要的是每个实验的参数、指标、模型、图表都绑在一起复现的时候直接看记录就行。注意MLflow的tracking server最好单独部署不要用本地文件存储。团队协作时大家连同一个server实验记录才能共享。本地文件存储只适合个人项目。4.2 超参数搜索的工程化做法超参数搜索最容易犯的错是在notebook里写个for循环跑一晚上第二天看结果。这种做法的问题是如果中间断了前面的结果全丢而且没法并行效率低。工程化的做法是用Optuna或Ray Tune配合MLflow记录。Optuna的用法很简单import optuna def objective(trial): lr trial.suggest_float(lr, 1e-4, 1e-1, logTrue) n_estimators trial.suggest_int(n_estimators, 50, 500) with mlflow.start_run(nestedTrue): mlflow.log_params({lr: lr, n_estimators: n_estimators}) model train_model(X_train, y_train, lrlr, n_estimatorsn_estimators) acc evaluate(model, X_val, y_val) mlflow.log_metric(accuracy, acc) return acc study optuna.create_study(directionmaximize) study.optimize(objective, n_trials100)Optuna会自动记录每次试验的参数和结果支持剪枝提前终止表现差的试验还支持并行。配合MLflow每次试验的细节都能追溯。4.3 训练脚本的标准化模板我习惯把训练脚本写成一个标准模板所有项目都套这个模板。模板的核心是配置驱动、日志完善、检查点自动保存。# src/training/train.py import logging import yaml import mlflow from pathlib import Path def load_config(path): with open(path) as f: return yaml.safe_load(f) def setup_logging(log_path): logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(log_path), logging.StreamHandler() ] ) def train(config): setup_logging(config[log_path]) logging.info(fStarting training with config: {config}) # 加载数据 train_data load_data(config[data][train_path]) val_data load_data(config[data][val_path]) # 训练 model build_model(config[model]) for epoch in range(config[training][epochs]): train_loss train_epoch(model, train_data) val_loss validate(model, val_data) logging.info(fEpoch {epoch}: train_loss{train_loss:.4f}, val_loss{val_loss:.4f}) # 保存检查点 if val_loss best_val_loss: best_val_loss val_loss save_checkpoint(model, config[checkpoint_path]) return model if __name__ __main__: config load_config(configs/base.yaml) train(config)这个模板的关键点是所有参数从配置来日志同时输出到文件和终端检查点自动保存最好的模型。这样你跑训练的时候可以去干别的事回来看日志就行。5. 模型部署与服务化从notebook到线上API5.1 为什么FastAPI是AI服务的好选择模型训练完下一步是让其他系统能调用它。最简单的方式是写个Flask应用但Flask是同步的并发能力弱。FastAPI是异步的性能好很多而且自动生成API文档调试方便。一个最小的模型服务长这样# src/serving/app.py from fastapi import FastAPI from pydantic import BaseModel import joblib import numpy as np app FastAPI() model joblib.load(models/model.pkl) class PredictRequest(BaseModel): features: list[float] class PredictResponse(BaseModel): prediction: float probability: float app.post(/predict, response_modelPredictResponse) async def predict(request: PredictRequest): X np.array(request.features).reshape(1, -1) pred model.predict(X)[0] prob model.predict_proba(X)[0].max() return PredictResponse(predictionfloat(pred), probabilityfloat(prob)) app.get(/health) async def health(): return {status: ok}跑起来用uvicorn src.serving.app:app --host 0.0.0.0 --port 8000然后访问/docs就能看到自动生成的API文档可以直接在浏览器里测试。实操心得模型加载一定要放在应用启动时不要放在请求处理函数里。放在请求里每次都要重新加载模型性能极差。如果模型很大加载慢可以用lazy loading第一次请求时加载之后缓存。5.2 模型格式选择pickle、ONNX还是TorchScript模型保存格式直接影响部署的灵活性和性能。我对比过几种常见格式格式优点缺点适用场景pickle简单、Python原生依赖训练时的环境、不安全内部快速原型ONNX跨框架、跨语言、性能好转换可能丢精度生产环境、多语言调用TorchScriptPyTorch原生、支持动态图只适用于PyTorchPyTorch模型部署SavedModelTensorFlow原生、生态好只适用于TFTensorFlow模型部署我的建议是如果模型要上生产优先转ONNX。ONNX Runtime的推理速度通常比原生框架快而且可以在C、Java、C#里调用不依赖Python环境。转换也不难import torch import torch.onnx # 假设model是PyTorch模型 dummy_input torch.randn(1, input_dim) torch.onnx.export( model, dummy_input, model.onnx, input_names[input], output_names[output], dynamic_axes{input: {0: batch_size}, output: {0: batch_size}} )转换后可以用ONNX Runtime加载import onnxruntime as ort session ort.InferenceSession(model.onnx) input_name session.get_inputs()[0].name output session.run(None, {input_name: X.astype(np.float32)})实测下来ONNX Runtime的推理延迟比PyTorch原生低30%左右内存占用也小。5.3 容器化部署Dockerfile怎么写才不踩坑模型服务要部署到服务器Docker是最方便的方式。但写Dockerfile有几个坑第一基础镜像不要用python:3.9这种全量镜像太大。用slim版本能小一半。第二依赖安装要分层。先复制requirements.txt安装依赖再复制代码。这样改代码时不用重装依赖构建快。第三模型文件不要打进镜像。镜像应该只包含代码和依赖模型文件通过挂载卷或启动时下载。一个经过验证的DockerfileFROM python:3.9-slim WORKDIR /app # 先装依赖 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 再复制代码 COPY src/ ./src/ COPY configs/ ./configs/ # 模型通过卷挂载 VOLUME /app/models EXPOSE 8000 CMD [uvicorn, src.serving.app:app, --host, 0.0.0.0, --port, 8000]构建和运行docker build -t my-ai-service . docker run -d -p 8000:8000 -v /path/to/models:/app/models my-ai-service注意requirements.txt里要固定版本号比如fastapi0.104.1不要写fastapi。不固定版本的话今天构建能跑明天可能就挂了因为依赖更新了。6. 监控与迭代上线只是开始不是结束6.1 线上监控要盯哪些指标模型上线后最怕的是“静默失败”——服务没挂但预测结果越来越差。所以监控要分两层系统层和模型层。系统层指标用Prometheus收集包括请求量、延迟、错误率、CPU/内存使用率。这些指标Grafana有现成模板配一下就行。模型层指标需要自己埋点包括预测分布、特征分布、置信度分布。比如你可以记录每次请求的预测值然后定期统计分布。如果发现预测分布突然偏移说明数据分布变了模型可能失效了。from prometheus_client import Histogram, Counter PREDICTION_HIST Histogram( model_prediction_value, Distribution of model predictions, buckets[0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0] ) REQUEST_COUNT Counter( model_request_total, Total model requests, [status] ) app.post(/predict) async def predict(request: PredictRequest): try: result model.predict(request.features) PREDICTION_HIST.observe(result) REQUEST_COUNT.labels(statussuccess).inc() return result except Exception as e: REQUEST_COUNT.labels(statuserror).inc() raise6.2 数据漂移检测的简单实现数据漂移是指线上数据的分布和训练数据不一致。检测方法有很多最简单的是PSIPopulation Stability Index。PSI的计算方式是把训练数据和线上数据按特征分桶计算每个桶的占比差异然后加权求和。import numpy as np def calculate_psi(expected, actual, buckets10): 计算PSIexpected是训练数据actual是线上数据 breakpoints np.percentile(expected, np.linspace(0, 100, buckets 1)) breakpoints[0] -np.inf breakpoints[-1] np.inf expected_percents np.histogram(expected, breakpoints)[0] / len(expected) actual_percents np.histogram(actual, breakpoints)[0] / len(actual) # 避免除零 expected_percents np.where(expected_percents 0, 0.0001, expected_percents) actual_percents np.where(actual_percents 0, 0.0001, actual_percents) psi np.sum((actual_percents - expected_percents) * np.log(actual_percents / expected_percents)) return psiPSI的判断标准小于0.1说明分布稳定0.1到0.25说明有轻微漂移大于0.25说明漂移严重需要重新训练模型。实操心得PSI检测要定期跑比如每天跑一次。不要等出问题了才查。我习惯把PSI计算做成定时任务结果写到数据库Grafana里画成趋势图。这样漂移是渐变还是突变一目了然。6.3 模型迭代的闭环流程模型迭代不是“重新训练一次”那么简单而是一个闭环监控发现问题、分析原因、准备数据、训练新模型、评估、上线、继续监控。这个闭环里最容易忽略的是评估环节。新模型不能只看离线指标还要做A/B测试。我的做法是新模型上线后先切10%的流量过去对比新旧模型的线上指标。如果新模型在关键指标上不差于旧模型再逐步扩大流量。A/B测试的流量切分可以在网关层做也可以在服务层做。服务层做更灵活import random MODEL_A load_model(models/model_a.onnx) MODEL_B load_model(models/model_b.onnx) app.post(/predict) async def predict(request: PredictRequest): if random.random() 0.1: model MODEL_B version B else: model MODEL_A version A result model.predict(request.features) # 记录版本方便后续分析 log_prediction(version, request.features, result) return result跑一周后对比A和B的线上指标决定是否全量切换。7. 常见问题与排查技巧实录7.1 训练时loss不下降怎么排查这是最常见的问题。排查顺序应该是先看数据再看模型最后看超参。第一步检查数据。把一批数据喂给模型看输出是否合理。如果模型输出全是同一个值说明数据可能有问题比如标签全是0或者特征全是NaN。第二步检查模型。用一个极小的数据集比如10条过拟合一下。如果模型连10条数据都拟合不了说明模型结构有问题比如层数太少、激活函数不对。第三步检查超参。学习率太大loss会震荡学习率太小loss下降极慢。可以试试用学习率finder找一个合适的初始学习率。# 简单的学习率finder lrs np.logspace(-5, -1, 100) losses [] for lr in lrs: model build_model() optimizer torch.optim.SGD(model.parameters(), lrlr) loss train_one_step(model, optimizer, data) losses.append(loss) # 画图找loss下降最快的点 import matplotlib.pyplot as plt plt.plot(lrs, losses) plt.xscale(log) plt.show()7.2 线上服务延迟高从哪里入手延迟高的问题排查思路是先定位瓶颈在哪一层再针对性优化。排查点可能原因优化方法网络层请求体太大、网络带宽不足压缩请求、升级带宽应用层同步阻塞、GIL限制用异步框架、多进程模型层模型太大、计算量大模型量化、剪枝、用ONNX数据层特征查询慢加缓存、预计算特征我遇到最多的是模型层的问题。一个BERT模型原始PyTorch推理要200ms转ONNX后降到80ms再量化到INT8后降到30ms。所以模型优化是收益最大的方向。7.3 依赖冲突怎么解决Python依赖冲突是AI工程的经典问题。项目A需要numpy 1.19项目B需要numpy 1.21装在一起就炸。解决方案有三个第一用虚拟环境隔离。每个项目一个venv互不影响。这是最基本的做法。第二用Docker。每个服务一个容器依赖完全隔离。这是生产环境的标配。第三用poetry或pipenv管理依赖。它们会生成lock文件确保每次安装的版本一致。我个人的习惯是开发时用conda创建虚拟环境部署时用Docker。conda的好处是能装非Python依赖比如CUDA、cuDNN这对GPU训练很重要。# 创建环境 conda create -n my_project python3.9 conda activate my_project # 安装PyTorch会自动装CUDA依赖 conda install pytorch torchvision torchaudio cudatoolkit11.3 -c pytorch # 其他依赖用pip pip install -r requirements.txt7.4 常见问题速查表问题现象可能原因快速排查方法训练loss为NaN学习率太大、数据有inf降低学习率、检查数据验证集指标远低于训练集过拟合加正则化、加数据、早停线上预测结果和离线不一致特征处理不一致对比线上线下特征管道服务启动报错找不到模型路径不对、模型未挂载检查挂载卷和配置路径推理速度慢模型未优化、批处理不当转ONNX、加批处理内存泄漏全局变量累积、缓存未清理用memory_profiler排查8. 我在这套流程里踩过的几个坑第一个坑是过早优化。刚开始做AI工程时我总想一步到位把MLflow、DVC、Airflow全装上。结果光是配这些工具就花了两周真正写模型的时间反而少了。后来我学乖了先跑通最小闭环数据加载、训练、保存模型、FastAPI服务。等这个闭环跑顺了再逐步加实验追踪、数据版本、监控。工具是解决问题的不是制造问题的。第二个坑是忽略测试。AI代码也需要单元测试尤其是数据清洗和特征工程部分。这些代码逻辑复杂容易出错而且出错后很难发现。我现在的习惯是每个清洗函数都写测试用pytest跑。测试数据不用多几条就够关键是覆盖边界情况。第三个坑是配置散落。早期我把数据库密码、API密钥写在代码里后来代码传到git密钥就泄露了。正确的做法是用环境变量或密钥管理服务。配置文件中只放非敏感信息敏感信息通过环境变量注入。import os DB_PASSWORD os.environ.get(DB_PASSWORD) API_KEY os.environ.get(API_KEY)第四个坑是不做回滚预案。新模型上线后效果变差想回滚却发现旧模型没保存。所以每次上线新模型一定要保留旧模型和对应的配置确保能一键回滚。我的做法是模型文件按版本号命名配置也按版本号存档上线时记录当前版本回滚时切回上一个版本。这套从零搭建AI工程体系的流程我前后迭代了两年多现在基本稳定了。新项目启动时我直接复制这套骨架改改配置就能跑。省下来的时间可以花在真正重要的事情上理解业务、优化模型、分析数据。工程化不是为了炫技是为了让你能专注于创造价值的部分。 RELATED READING
延伸阅读
更多一线实战笔记与深度复盘，助您持续精进
