1. 从零搭建AI工程体系为什么我劝你别急着调包ai-engineering-from-scratch这个标题第一次看到的时候我愣了一下。市面上讲AI的教程铺天盖地但绝大多数都是教你import torch然后跑个预训练模型或者调个API接口就完事。真正从零开始、把AI工程当作一门系统工程来拆解的少之又少。我自己在这个方向上摸索了挺长时间踩过的坑不算少。最开始我也觉得AI工程嘛不就是数据清洗、模型训练、部署上线这三板斧后来真正在项目里滚了几圈才发现事情远没有那么简单。一个能跑通的demo和一个能扛住线上流量的AI系统中间隔着的不是几行代码而是一整套工程化的思维方式和工具链。这篇内容我想聊的就是怎么从零开始搭建一套AI工程体系。不是那种教你三行代码调用大模型的快餐教程而是从环境管理、数据处理、实验追踪、模型训练、评估验证到部署监控的完整链路。适合谁看如果你已经会写Python对机器学习有基本概念但一到实际项目就不知道从哪下手那这篇内容就是写给你的。如果你是有经验的工程师想看看别人是怎么组织AI项目的也能从中找到一些可参考的实践。核心关键词就一个ai-engineering-from-scratch。我会围绕这个关键词把从零搭建AI工程体系的每个环节拆开来讲包括为什么这么选、怎么操作、容易在哪里翻车。2. 整体设计思路AI工程到底在工程什么2.1 先搞清楚AI工程和模型训练的区别很多人把AI工程等同于模型训练这是一个非常普遍的认知偏差。模型训练只是AI工程中的一个环节而且往往不是最耗时间的那个环节。根据我自己的经验一个完整的AI项目里数据处理和工程化的工作量能占到70%以上模型训练和调参可能只占20%剩下的10%是部署和监控。打个比方模型训练像是做一道菜AI工程则是经营一家餐厅。你菜做得再好如果供应链不稳定、厨房动线混乱、出餐流程没有标准化餐厅照样开不下去。AI工程要解决的就是这些开餐厅的问题数据从哪来、怎么保证质量、实验怎么管理、模型怎么上线、上线后怎么监控。所以从零搭建AI工程体系第一步不是去学什么高级模型架构而是把工程化的基础设施搭起来。这个思路很重要因为它决定了你后续所有工作的效率上限。2.2 技术选型的核心原则够用、可替换、可复现从零开始搭建的时候最容易犯的错误就是过度设计。我见过有人一上来就搞Kubernetes集群、上Feature Store、搭全套MLOps平台结果项目还没跑起来光维护基础设施就累得够呛。我的建议是遵循三个原则够用原则当前阶段需要什么就搭什么不要为了一年后的假设需求提前买单。比如你现在的数据量只有几万条用Pandas完全够用没必要上Spark。可替换原则每个组件都要有清晰的接口边界方便后续替换。比如数据加载层用抽象类定义接口底层实现可以从Pandas换成Polars上层代码不用改。可复现原则任何一次实验结果都必须能复现。这要求你固定随机种子、记录环境依赖、版本化数据和模型。这三个原则看起来简单但真正执行起来需要克制。尤其是够用原则当你看到别人用了一堆酷炫工具的时候很容易产生焦虑。但相信我大部分时候简单的方案才是对的方案。2.3 目录结构项目的地基一个清晰的目录结构是AI工程项目的地基。我试过很多种组织方式最后沉淀下来一套自己觉得比较顺手的结构project/ ├── configs/ # 配置文件 │ ├── default.yaml │ └── experiment/ ├── data/ # 数据目录 │ ├── raw/ # 原始数据只读 │ ├── interim/ # 中间处理结果 │ └── processed/ # 最终训练数据 ├── src/ # 源代码 │ ├── data/ # 数据处理模块 │ ├── features/ # 特征工程模块 │ ├── models/ # 模型定义 │ ├── training/ # 训练逻辑 │ ├── evaluation/ # 评估逻辑 │ └── serving/ # 部署服务 ├── experiments/ # 实验记录 ├── notebooks/ # 探索性分析 ├── tests/ # 测试代码 ├── Makefile # 常用命令入口 └── pyproject.toml # 项目依赖这个结构的关键在于关注点分离。数据、代码、配置、实验记录各归其位不会混在一起。特别是data/目录下的三级划分raw层永远不动interim层放中间结果processed层放最终数据这样任何时候你都能从raw重新跑一遍完整流程。注意data/raw目录一定要设为只读这是数据可复现的第一道防线。我见过太多项目因为原始数据被意外修改导致之前的实验结果全部无法复现。3. 核心细节解析每个环节的关键决策3.1 环境管理为什么我最终选了uv而不是conda环境管理是AI工程的第一步也是最容易被忽视的一步。很多人直接用系统Python加pip install跑着跑着就发现依赖冲突了。我早期用conda后来转到了uv。原因很简单conda解决依赖冲突的能力确实强但它的解析速度太慢了尤其是在需要频繁创建新环境的场景下。uv用Rust写的解析和安装速度快了一个数量级而且它兼容pip的依赖格式迁移成本很低。具体操作上我会在pyproject.toml里声明依赖然后用uv创建虚拟环境uv venv .venv source .venv/bin/activate uv pip install -e .[dev]这里有个细节-e表示可编辑安装[dev]表示同时安装开发依赖。把训练依赖和开发依赖分开声明可以避免生产环境装一堆用不到的东西。另外一定要锁定依赖版本。uv会生成uv.lock文件这个文件要提交到版本控制里。没有lock文件的项目等于没有环境管理。3.2 数据处理从原始数据到训练样本的完整链路数据处理是AI工程里最脏最累的活但也是最能体现工程水平的地方。我的做法是把数据处理拆成三个阶段清洗、转换、验证。清洗阶段处理的是原始数据里的明显问题缺失值、重复值、格式错误、编码问题。这个阶段的输出是interim数据可以接受一定程度的粗糙。转换阶段做的是特征工程和格式转换把数据转成模型需要的格式做归一化、编码、分词等操作。这个阶段的输出是processed数据必须干净、规范。验证阶段是很多人会跳过的一步但我觉得它至关重要。验证包括数据分布检查、样本数量核对、特征范围校验、标签一致性检查。我一般会写一个validate.py脚本每次生成processed数据后自动跑一遍。def validate_dataset(df, schema): 校验数据集是否符合预期schema errors [] # 检查列是否存在 missing_cols set(schema[columns]) - set(df.columns) if missing_cols: errors.append(f缺失列: {missing_cols}) # 检查样本数量 if len(df) schema[min_samples]: errors.append(f样本数不足: {len(df)} {schema[min_samples]}) # 检查标签分布 label_dist df[label].value_counts(normalizeTrue) for label, min_ratio in schema[label_ratios].items(): if label_dist.get(label, 0) min_ratio: errors.append(f标签{label}占比过低: {label_dist.get(label, 0)}) return errors这个验证脚本看起来简单但它帮我挡掉过好几次数据管道的bug。有一次上游数据源改了格式导致标签全部错位就是靠这个脚本发现的。3.3 实验追踪别再用Excel记实验结果了我见过太多人用Excel或者记事本记录实验结果然后过两周就忘了哪个配置对应哪个结果。实验追踪不是可选项是必选项。工具选择上MLflow和Weights BiasesWB是两个主流方案。MLflow可以自托管数据在自己手里WB体验更好但数据在云端。我的建议是如果公司有数据合规要求用MLflow如果是个人项目或者小团队WB的免费额度完全够用。不管用哪个工具核心是记录这几类信息记录项说明重要性超参数学习率、batch size、模型结构参数必须指标曲线loss、accuracy、F1等随训练步数的变化必须环境信息Python版本、依赖版本、硬件信息必须数据版本训练数据的hash或版本号必须代码版本git commit hash必须随机种子所有随机源的种子值必须模型文件训练好的模型权重推荐备注这次实验的假设和观察推荐前六项是硬性要求缺任何一项都会导致实验无法复现。模型文件和备注是锦上添花但有了会方便很多。3.4 配置管理把超参数从代码里赶出去超参数硬编码在代码里是另一个常见问题。每次调参都要改代码、提交、重新跑效率极低而且容易出错。我的做法是用YAML配置文件加Hydra框架。Hydra支持配置组合、命令行覆盖、多实验并行非常适合AI项目。# configs/default.yaml model: name: resnet18 num_classes: 10 dropout: 0.3 training: lr: 0.001 batch_size: 64 epochs: 50 optimizer: adam scheduler: cosine data: train_path: data/processed/train.parquet val_path: data/processed/val.parquet num_workers: 4用Hydra之后你可以在命令行直接覆盖配置python train.py training.lr0.0001 training.batch_size128这样调参就不用改代码了而且每次实验的完整配置会被Hydra自动保存到输出目录方便追溯。4. 实操过程从零到一搭建完整流程4.1 第一步初始化项目骨架假设你现在什么都没有我们从零开始。首先创建项目目录和基础文件mkdir ai-project cd ai-project git init uv init然后按照前面说的目录结构创建各个子目录。这里有个小技巧用__init__.py文件把src变成一个包这样在测试和部署时导入会更规范。mkdir -p configs data/{raw,interim,processed} src/{data,features,models,training,evaluation,serving} experiments notebooks tests touch src/__init__.py src/data/__init__.py src/models/__init__.py接下来配置pyproject.toml声明核心依赖[project] name ai-project version 0.1.0 requires-python 3.10 dependencies [ numpy, pandas, scikit-learn, torch, pyyaml, hydra-core, mlflow, polars, ] [project.optional-dependencies] dev [ pytest, ruff, mypy, jupyter, ]依赖声明完之后跑uv sync安装。这一步会生成lock文件确保环境可复现。4.2 第二步搭建数据处理管道数据处理管道我一般用Polars而不是Pandas原因是Polars在处理中等规模数据百万行级别时速度快很多而且它的惰性求值Lazy Evaluation模式可以自动优化查询计划。import polars as pl from pathlib import Path def load_raw_data(path: Path) - pl.LazyFrame: 加载原始数据返回惰性帧 return pl.scan_parquet(path) def clean_data(lf: pl.LazyFrame) - pl.LazyFrame: 清洗数据去重、处理缺失值、类型转换 return ( lf .unique(subset[id]) .filter(pl.col(label).is_not_null()) .with_columns([ pl.col(text).str.strip_chars(), pl.col(label).cast(pl.Int32), ]) ) def build_features(lf: pl.LazyFrame) - pl.LazyFrame: 特征工程 return lf.with_columns([ pl.col(text).str.len_chars().alias(text_length), pl.col(text).str.count_matches(r\w).alias(word_count), ]) def run_pipeline(raw_path: Path, output_path: Path): 执行完整管道 lf ( load_raw_data(raw_path) .pipe(clean_data) .pipe(build_features) ) df lf.collect() df.write_parquet(output_path) return df这个管道的好处是每一步都是纯函数输入输出明确方便测试和调试。如果某一步出了问题你可以单独把中间结果collect出来检查。4.3 第三步模型训练与实验追踪训练脚本的核心结构是加载配置、准备数据、构建模型、训练循环、记录指标。我用Hydra管理配置用MLflow记录实验。import hydra from omegaconf import DictConfig import mlflow import torch hydra.main(config_path../configs, config_namedefault) def train(cfg: DictConfig): # 设置随机种子 seed cfg.get(seed, 42) torch.manual_seed(seed) # 初始化MLflow mlflow.set_experiment(cfg.experiment_name) with mlflow.start_run(): # 记录配置 mlflow.log_params(flatten_config(cfg)) # 准备数据 train_loader, val_loader prepare_data(cfg.data) # 构建模型 model build_model(cfg.model) optimizer build_optimizer(model, cfg.training) # 训练循环 for epoch in range(cfg.training.epochs): train_loss train_one_epoch(model, train_loader, optimizer) val_metrics evaluate(model, val_loader) # 记录指标 mlflow.log_metrics({ train_loss: train_loss, **val_metrics }, stepepoch) # 保存最佳模型 if val_metrics[f1] best_f1: best_f1 val_metrics[f1] torch.save(model.state_dict(), best_model.pt) mlflow.log_artifact(best_model.pt)这里有几个实操细节值得注意随机种子要固定所有来源Python的random、NumPy的random、PyTorch的random、CUDA的random一个都不能漏。指标记录要带step这样MLflow才能画出曲线图方便观察训练趋势。模型保存要带配置光保存权重不够还要保存模型结构配置否则加载的时候对不上。4.4 第四步评估与验证评估不是简单地算个准确率就完事。一个完整的评估应该包括整体指标accuracy、F1、AUC等分组指标按类别、按数据来源、按时间分组的指标错误分析哪些样本预测错了错误模式是什么鲁棒性测试对输入扰动、分布偏移的敏感度我一般会写一个独立的评估脚本输入是模型文件和测试数据输出是一份完整的评估报告。def evaluate_model(model, test_loader, label_names): 完整评估 all_preds, all_labels [], [] model.eval() with torch.no_grad(): for batch in test_loader: preds model(batch[input]) all_preds.extend(preds.argmax(dim1).cpu().numpy()) all_labels.extend(batch[label].cpu().numpy()) # 整体指标 report classification_report(all_labels, all_preds, target_nameslabel_names, output_dictTrue) # 错误分析 errors analyze_errors(all_labels, all_preds, test_loader.dataset) return report, errors评估报告要存档和模型文件放在一起。这样任何时候你都能回答这个模型在测试集上表现如何这个问题。4.5 第五步部署与监控部署环节我推荐从最简单的方案开始FastAPI加Docker。不要一上来就搞Kubernetes除非你确实有弹性伸缩的需求。from fastapi import FastAPI from pydantic import BaseModel import torch app FastAPI() model load_model(best_model.pt) class PredictRequest(BaseModel): text: str class PredictResponse(BaseModel): label: int confidence: float app.post(/predict, response_modelPredictResponse) def predict(request: PredictRequest): with torch.no_grad(): logits model(preprocess(request.text)) probs torch.softmax(logits, dim-1) label probs.argmax().item() confidence probs.max().item() return PredictResponse(labellabel, confidenceconfidence)监控方面至少要记录这几类指标监控项说明告警阈值建议请求延迟P50/P95/P99延迟P99超过500ms告警请求量QPS突增突降告警错误率5xx错误占比超过1%告警预测分布各类别预测占比与训练分布偏差超过20%告警输入分布输入特征统计量与训练分布偏差超过阈值告警预测分布和输入分布的监控特别重要因为它们是模型退化的早期信号。如果线上预测的类别分布突然和训练时差很多大概率是数据分布变了模型需要重新训练。5. 常见问题与排查技巧实录5.1 训练loss不下降怎么排查这是最常见的问题排查思路要系统化不要瞎调参。我一般按这个顺序检查数据对不对把训练数据打印几条出来看看标签和输入是否对应。我遇到过标签错位的情况模型怎么训都学不会。模型能不能过拟合小样本拿10条数据训练100个epoch如果loss降不下去说明模型结构或训练逻辑有问题。学习率是否合适太大导致震荡太小导致下降缓慢。可以试试学习率扫描从1e-5到1e-1各跑几个step看看loss变化。梯度是否正常打印梯度范数如果全是0或者全是NaN说明有梯度消失或爆炸问题。损失函数是否正确分类任务用交叉熵回归任务用MSE别搞混了。实操心得我习惯在训练脚本里加一个--debug模式只跑少量数据打印每一步的中间结果。这个模式帮我省了大量排查时间。5.2 实验无法复现怎么办实验无法复现的原因通常有这几个随机种子没固定全检查所有随机源包括数据加载的shuffle、dropout、数据增强等。依赖版本不一致用lock文件锁定版本不要用pip install package这种不指定版本的方式。数据版本不一致给数据打hash每次实验记录数据hash。硬件差异GPU和CPU的结果可能有细微差异CUDA版本不同也可能导致结果不同。非确定性操作某些CUDA操作是非确定性的需要设置torch.use_deterministic_algorithms(True)。复现性检查清单检查项具体操作随机种子固定Python/NumPy/PyTorch/CUDA种子依赖版本使用lock文件记录完整依赖树数据版本计算数据文件hash并记录代码版本记录git commit hash硬件信息记录GPU型号、CUDA版本、驱动版本确定性设置开启deterministic模式5.3 线上模型效果比离线差很多这是典型的训练-服务偏差Training-Serving Skew问题。常见原因预处理不一致训练时用Polars做归一化线上用NumPy做结果对不上。解决方案是把预处理逻辑封装成共享模块训练和线上用同一份代码。特征计算时机不同训练时用的是全量数据统计量线上只能用历史数据。解决方案是用时间窗口统计确保线上线下一致。数据分布偏移线上数据分布和训练数据不同。解决方案是持续监控输入分布发现偏移及时重新训练。版本不一致线上跑的模型不是最新训练的版本。解决方案是建立模型版本管理机制部署时明确指定版本号。5.4 常见问题速查表问题现象可能原因排查方向loss为NaN学习率过大、数据有异常值降低学习率、检查数据loss不下降标签错误、模型结构问题检查数据、小样本过拟合测试显存不足batch size过大、模型过大减小batch size、梯度累积训练速度慢数据加载瓶颈、GPU利用率低增加num_workers、检查数据管道指标波动大batch size过小、学习率过大增大batch size、降低学习率过拟合模型复杂度过高、数据量不足加正则化、数据增强、早停欠拟合模型复杂度不足、训练不充分增大模型、增加训练轮数6. 工具链选型我的实际搭配方案6.1 核心工具链一览经过多个项目的迭代我目前比较稳定的工具链搭配是环节工具选择替代方案选择理由环境管理uvconda, poetry速度快兼容pip数据处理PolarsPandas惰性求值性能好深度学习PyTorchTensorFlow生态好调试方便配置管理HydraOmegaConf支持组合和覆盖实验追踪MLflowWB可自托管数据可控服务框架FastAPIFlask异步支持自动文档容器化DockerPodman生态成熟代码质量ruff mypyflake8 pylint速度快配置简单测试pytestunittest语法简洁插件丰富这套搭配的核心逻辑是每个环节选一个够用且维护活跃的工具不追求最新最酷追求稳定可靠。6.2 什么情况下该换工具工具不是一成不变的但换工具要有明确的理由。我一般在这几种情况下会考虑换性能瓶颈当前工具成为瓶颈且没有优化空间。比如Pandas处理千万行数据太慢换Polars。维护停滞工具超过一年没有更新社区活跃度下降。需求变化项目需求发生了根本变化当前工具不再适用。比如从单机训练转到分布式训练。团队协作团队新成员对某个工具更熟悉且迁移成本可接受。换工具的成本往往被低估。除了迁移代码的时间还有学习成本、调试成本、潜在的bug风险。所以我的原则是除非有明确的痛点否则不换。6.3 那些我踩过的工具坑说几个我实际踩过的坑帮你避雷坑一过早引入MLOps平台。我曾经在一个小项目上搭了完整的MLOps平台结果项目一共就三个人平台维护占了一半时间。后来全部拆掉用MLflow加几个脚本效率反而更高。坑二迷信分布式训练。以为分布式一定比单机快结果发现数据量根本不够大通信开销比计算开销还大。单机多卡都跑不满更别说多机了。坑三忽视数据版本管理。早期觉得数据不会变结果上游数据源更新了一次所有实验结果全部无法复现。后来养成了给数据打hash的习惯。坑四配置散落各处。超参数在代码里、在命令行里、在环境变量里到处都是。后来统一用Hydra管理世界清净了。7. 从零搭建的进阶方向7.1 自动化训练管道当你的实验频率变高之后手动跑训练脚本会变得很低效。这时候可以考虑搭建自动化训练管道代码提交触发训练、训练完成自动评估、评估通过自动部署。我用的是GitHub Actions加自建runner的方案。核心逻辑是push到特定分支触发训练workflow训练完成后把指标写到PR评论里人工审核后合并触发部署。这个方案的好处是把训练和代码审查结合起来每次模型更新都有记录可查。缺点是配置起来有点复杂适合实验频率比较高的团队。7.2 特征存储当特征工程变得复杂多个模型共享特征的时候特征存储Feature Store就变得有价值了。它的核心作用是统一特征定义、保证线上线下一致性、支持特征复用。不过我要泼一盆冷水特征存储的维护成本不低小团队慎入。如果你的特征不超过几十个模型不超过三个用共享的Python模块就够了没必要上Feature Store。7.3 模型监控与自动重训模型上线不是终点而是起点。线上模型会随着数据分布变化而退化需要持续监控和定期重训。我的做法是监控输入分布和预测分布当分布偏移超过阈值时触发告警同时设置定期重训任务比如每周用最新数据重新训练一次。重训后的模型先在影子模式下跑一段时间对比新旧模型的表现确认新模型更好之后再切换。这套机制的核心是自动化加人工审核。全自动切换风险太大全人工又太累半自动是比较平衡的方案。7.4 关于从零的一点个人体会最后说点个人体会。从零搭建这件事最大的价值不在于你搭出来的东西有多完美而在于搭建过程中你对每个环节的理解会深刻很多。我见过太多人直接拿开源框架跑跑通了就觉得自己会了。但一旦出了问题完全不知道从哪排查。因为你没有经历过从零搭建的过程对每个组件的边界和交互没有直觉。所以我的建议是哪怕你最终要用开源框架也先自己从零搭一个最小可用的版本。不用很完善能跑通就行。这个过程会让你对AI工程有完全不同的理解。另外不要追求一步到位。我现在的这套体系是经过好几个项目迭代出来的最开始的时候比这简陋得多。先跑起来再优化这是工程实践的基本节奏。