“ai-engineering”这个词这几年从招聘JD里的流行词慢慢变成了真实工作里的硬门槛。我为什么要把一个仓库起名叫 ai-engineering-from-scratch因为我在过去一年里反复看到同一个问题模型训练跑通了项目却交付不了。这个仓库就是我从零开始整理的一套AI工程落地路径覆盖环境搭建、数据流水线、训练微调、评估调优、部署监控一条链路上的最小必要步骤适合准备做第一个端到端AI项目的人也适合被“模型能跑但上线即崩”折磨过的工程师。我见过太多人把“AI工程化”想得很玄觉得是大厂算法团队才需要研究的事情。但真实情况是哪怕你只是做一个文本分类、一个推荐召回、一个OCR识别只要涉及模型迭代和对外提供服务工程化就是躲不掉的。很多自学AI的同学卡住的地方不在模型结构而在“代码怎么组织”“数据怎么管理”“训练完怎么部署”。这篇文章与其说是讲解不如说是我如何把这些问题挨个解决的过程记录你可以把它当成一份可对照的操作笔记。1. 为什么需要一份“从零开始”的AI工程仓库1.1 我踩过的最贵的一课模型能跑不等于方案能交付先说一段不太光彩的经历。某次给一个业务团队做内部分类工具我在 Jupyter Notebook 里调模型测下来准确率接近95%当时觉得稳了直接把 .ipynb 文件丢给同事说“代码在这”。结果上线第二天就崩了线上数据分布和训练集不完全一样预处理的逻辑散落在几个单元格里label 映射表是用 dict 硬编码写在某个角落里新数据进来不仅分类错连前处理流程都会抛异常。整件事花了三周才救回来那个月我复盘时最扎心的结论是——问题从来不是模型选得不好而是我根本没有按工程方式对待这个项目。这就是 ai-engineering-from-scratch 这个仓库最原始的动机把那些“我以为没问题”的东西显式写下来。可复现的环境、固定版本的数据处理、记录超参数的训练脚本、可重复的评估流程、能回滚的部署方式。这些东西单独看都很枯燥但它们才是模型能在生产环境里活下来的原因。1.2 这个仓库的定位不是又一个深度学习教程现在网上的教程很多但大多数教你的是“怎么用 PyTorch 写一个 Transformer”到训练结束就戛然而止。工程仓库要解决的不是“模型结构怎么设计”而是“从零开始构建一个能长期维护的AI系统”所以我给这个项目定了四个支柱。第一是可复现任何一个人在任意一台机器上按照 README 里的命令能复现出跟我一样的训练结果。第二是可扩展加一个新数据集、一个新模型不需要重写主流程。第三是可观测训练指标、推理日志、数据分布变化都能被看到而不是黑盒。第四是可部署训练产物能很方便地变成服务或者批量任务而不是躺在 .pt 文件里。这四个支柱决定了仓库里的每个模块怎么取舍。比如我不会追求最花哨的模型但我会把 config 和代码分离得足够干净我不会搞复杂的分布式框架但我会把 checkpoint、日志、实验记录这些基本功做扎实。1.3 适合谁不适合谁我很清楚这个仓库不可能适合所有人所以干脆在 README 里写明。适合下面这三类人准备从零开始做第一个AI项目的个人开发者想看到一个完整的落地路径。已经在跑模型、但每次换环境都要折腾一天的人可以从环境管理部分直接受益。团队里需要承担“算法到工程”衔接角色的工程师比如算法工程师兼着做服务化。不太适合这两类人以发论文为目标的同学工程化里的很多步骤对单点实验来说是多余负担。只打算调用成熟API、不做任何自训练的人比如直接用大模型接口做应用确实不需要这份仓库里的训练部分。还有一个很重要的提醒这个仓库刻意不走“高大全”路线。我不会为了展示能力引入几十个依赖也不会写一个抽象到看不懂的框架层。工程化最重要的能力是做减法把最少必要的事情做好。2. 项目骨架先把“怎么组织代码”想清楚再写代码2.1 顶层目录结构定生死很多从 notebook 起步的人第一次写工程代码时都不知道把文件放哪。我在仓库里用了下面这个结构看起来很简单但足够支撑绝大多数中小型AI项目ai-engineering-from-scratch/ ├── configs/ # 所有实验配置 │ ├── train.yaml │ └── eval.yaml ├── data/ │ ├── raw/ # 原始数据只读不写 │ ├── interim/ # 中间处理结果 │ ├── processed/ # 训练/验证/测试集 │ └── features/ # 特征工程产物可选 ├── src/ │ ├── data/ # 数据加载、清洗、增强 │ ├── models/ # 模型定义 │ ├── training/ # 训练循环、优化器、loss │ ├── evaluation/ # 评估指标、错误分析 │ └── serving/ # 推理服务 ├── scripts/ # 一个个入口脚本 │ ├── run_prepare.py │ ├── run_train.py │ └── run_eval.py ├── tests/ # 单元测试和冒烟测试 ├── pyproject.toml ├── uv.lock └── README.md这个结构有几个关键设计决策。raw 目录必须是只读的所有原始文件一旦放进去就不要再手动改这是为了保证数据可追溯。scripts 层保持扁平每个脚本只负责一个动作不要在脚本里写一堆 if 分支来切换功能。src 下的模块只负责被调用不直接执行。我见过很多项目把 data、model、train 揉在一个 tools 目录里刚开始觉得方便一旦任务变复杂就会变成到处 import 的意大利面。目录结构不是给人看的是给未来的你和你的同事看的。2.2 依赖与环境用 pyproject uv彻底摆脱“在我电脑上是好的”“在我电脑上是好的”大概是工程协作里最让人血压升高的一句话。根因通常只有一个依赖没有锁定。我推荐用 uv 来管理 Python 项目。uv 是一个用 Rust 写的 Python 包管理工具速度非常快而且它的 lock 文件能完整锁定依赖树。为什么不用 requirements.txt因为 requirements.txt 往往只锁定直接依赖的版本但传递依赖的版本是不确定的你在本地装到的某个库和服务器上装到的可能是不同版本训练结果自然就对不上。创建项目并安装依赖的流程很简单uv init ai-engineering-from-scratch uv add torch transformers pandas pyyaml hydra-core uv add --dev pytest ruff mypy注意这里的前两步底层的 CUDA 环境我用 Docker 或系统级安装来管理然后建一个虚拟环境专门给 Python 包。pyproject.toml 里记录项目的元数据和所有依赖声明uv.lock 把每一条依赖的具体版本钉死。提交代码时这两个文件都要进版本库。还有一个经常被忽略的文件.python-version。这个文件可以钉死 Python 的次版本比如 3.11.x避免“本地是 3.12、服务器上 3.10”这种隐形差异。2.3 三件套固定随机种子、可重复实验、数据指纹就算你锁定了环境和依赖有一件事照样会让结果漂移随机数。深度学习训练里涉及大量随机性包括模型参数初始化、DataLoader 打乱顺序、数据增强的随机操作。不固定种子的话同一个脚本跑两遍可能得到完全不同的结果。我在仓库里写了一个全局初始化模块在训练脚本启动时被调用import os import random import numpy as np import torch def set_seed(seed: int 42) - None: random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) os.environ[PYTHONHASHSEED] str(seed) torch.backends.cudnn.deterministic True torch.backends.cudnn.benchmark False这个函数看起来无聊但它能救你的命。没有它你很难判断一次效果提升到底来自你改的代码还是来自一次运气特别好的随机初始化。除了固定种子我还给数据加了指纹。每次数据清洗脚本跑完生成一个 data_fingerprint.txt内容包含输入文件路径、文件大小、脚本版本和 SHA-256 哈希。这样做的好处是当实验记录里写着“用的 processed 数据”你可以直接追溯到那一份具体的数据内容而不是靠记忆说“应该就是那批”。2.4 最小CI先让代码能测再谈其他有人会觉得一个 AI 仓库没必要做 CI/CD其实恰恰相反。模型代码比普通后端代码更容易出隐蔽问题比如数据维度不匹配、类型转换错误、空值处理。这种问题如果等训练跑起来才发现几分钟甚至几小时后才发现纯属浪费时间。我在仓库里放了一个最小可用的 GitHub Actions 工作流它只做三件事装依赖、跑 ruff 检查、跑测试。真实代码大概是这样name: ci on: [push, pull_request] jobs: lint-test: runs-on: ubuntu-latest steps: - uses: actions/checkoutv4 - uses: astral-sh/setup-uvv5 - run: uv sync - run: uv run ruff check src tests - run: uv run pytest tests这个工作流不涉及任何 GPU 和大模型训练只做静态检查和轻量测试所以几分钟内就能完成。我强烈建议在 CI 里跑一个冒烟测试比如用 5% 的数据训练 1 个 step确保主流程没有断裂然后再在本地做完整训练。CI 的价值不是抓住所有 bug而是把最低级的错误挡在提交之前让每次实验的起点更干净。3. 数据流水线决定模型上限的环节往往在训练之前3.1 原始数据到训练集的四段式处理很多 AI 新手拿到数据后直接开始训练这是个大误区。我习惯把数据流程分成四个阶段raw、interim、processed、features。每个阶段在磁盘上有独立目录每个阶段对应一个可复用的处理脚本。raw 阶段只负责收数据。格式再乱、字段再脏也无所谓反正只保存原始内容。interim 阶段做格式统一和字段重命名把 JSON、CSV、TXT 统一成标准格式。processed 阶段负责清洗、标签映射、划分训练验证测试集。features 是针对特定模型做的特征工程比如文本的 token 化结果、用户行为的统计特征。为什么要分这么多层核心目的是隔离风险。业务方换了一批原始数据你只需要重跑第一段后面每一段都可以基于新 raw 重新生成。模型要换一种 tokenizer你只需要重跑 features 阶段不用重新处理原始数据。这种分层会让改动成本大幅下降。3.2 写一个“失败大声”的数据清洗脚本数据清洗最忌讳静默处理。我见过有人写清洗逻辑时把空值直接删掉、把超长文本直接截断、把未知标签直接赋成背景类最后连日志都不打一条。这种做法会掩盖数据质量的真实问题等模型上线才发现某些类型的样本从来没被正确处理过。我在仓库里写了个clean_text.py它的核心原则是每丢弃一个样本都要记录原因。代码逻辑大致是# src/data/clean_text.py from collections import Counter from pathlib import Path def clean_samples(samples): counter Counter() clean [] for sample in samples: text sample.get(text, ) label sample.get(label, None) if not text or not text.strip(): counter[empty_text] 1 continue if label is None: counter[missing_label] 1 continue if len(text) 10: counter[too_short] 1 continue clean.append({text: text, label: label}) counter[kept] 1 return clean, counter实际使用时会把这个 counter 序列化成一个 JSON 文件存到 processed 目录旁边。训练之前先看一眼这个统计你就能知道这批数据里有多少样本是“变成噪音”的以及你的模型有哪些情况注定学不会。这一步的价值不输给调参。3.3 用 Datasets 构建缓存友好的 DataLoader数据量大了以后每次训练前重新加载和处理数据会变成巨大的开销。我用 HuggingFace 的 Datasets 库来管理训练数据核心原因是它的 map、cache、filter 接口都做了缓存处理数据不会重复处理。一个比较经典的使用方式是这样from datasets import load_dataset from src.data.tokenizer import load_tokenizer ds load_dataset(json, data_filesdata/processed/train.jsonl, splittrain) ds ds.map( lambda x: tokenizer(x[text], truncationTrue, max_length256), batchedTrue, remove_columns[text], cache_file_namedata/features/cache_train.arrow, ) ds.set_format(typetorch, columns[input_ids, attention_mask, labels])这里有一个关键细节一旦数据变更频率不高应该把这个 tokenizer 处理后的结果保存成 arrow 文件或者 parquet 文件而不是每次都重新 tokenize。我一般把处理完的数据落盘到data/features/训练脚本直接加载落盘结果。这样做之后同样的训练数据从小时级加载变成分钟级加载。3.4 防数据泄漏的检查清单数据泄漏是 AI 项目里最容易犯、又最难查的错误它不是“模型看到答案”这么简单而是信息跨过划分边界泄漏到训练集里。比如你用全量数据做标准化然后把训练集和验证集分开这样就相当于验证集的信息参与了训练。我把自己常用的防泄漏检查写成了一份清单放在仓库 docs 里时间敏感任务按时间切分而不是随机切分。文本去重在划分前做 MinHash 去重防止同一句在不同集合里出现。标签映射先建立 label 字典再拆分数据并在训练代码里固定字典顺序。特征统计量均值、方差、归一化参数只从训练集计算。数据增强增强操作必须在划分之后进行并且不能跨文件缓存。这份清单看起来像常识但我在实际项目里至少三次发现有团队在验证集上出现极高分数最后查出来的原因都是归一化用了全量数据的均值和方差。工程化的意义就是把这些“常识”变成代码里的强制检查。4. 训练与微调先跑通一个能解释的基线再追SOTA4.1 三个起点从零训练、预训练、微调怎么选模型从零开始训练还是基于预训练模型微调这个选择应该由数据量、算力预算和任务特点共同决定。我整理过一个简单的对比表方便在项目开始时快速决策。方案需要的数据量计算成本效果起点适合场景从零训练很大通常百万级以上非常高低需要从头学数据形态极其特殊预训练模型覆盖不了预训练 微调较小几百到几万条均可中高利用通用知识文本分类、情感分析、NER 等绝大多数NLP任务直接推理 后处理几乎不需要低取决于基础模型任务和模型能力高度重合如通用问答摘要我个人的经验是优先考虑微调给从零训练保留最严苛的准入条件。除非你做的任务和通用预训练任务相差太远或者你的数据量级已经大到无法被通用模型覆盖否则从零训练通常只是满足“能力强”的自我感动成本和效果都不划算。4.2 最小训练脚本的写法训练代码最容易出现的问题是把所有逻辑塞进一个两三百行的文件里中间穿插各种硬编码路径。我的仓库里把训练脚本保持得很薄真正的内容在src/training/里入口脚本只负责组装。一个典型的scripts/run_train.py大约长这样import hydra from omegaconf import DictConfig hydra.main(config_path../configs, config_nametrain) def main(cfg: DictConfig): set_seed(cfg.seed) model build_model(cfg.model) dataset load_processed_dataset(cfg.data) trainer SimpleTrainer(model, dataset, cfg) trainer.train() if __name__ __main__: main()这样做的好处是每一份配置都是一个 yaml 文件超参数改动不需要碰代码。configs/train.yaml里写着模型名称、learning rate、batch size、epochs、数据路径、输出路径等所有关键参数。一旦发现某个实验效果好你直接把这个 yaml 存成带时间戳的副本以后任何时候都能知道“好效果是谁创造出来的”。训练循环本身不必过度设计。我在SimpleTrainer里只做了这些事前向传播、计算 loss、反向传播、梯度裁剪、定期保存 checkpoint、早停。很多人喜欢一开始就上分布式、混合精度、梯度累积我建议先跑通单卡基线再用配置开关逐步打开这些功能这样每个环节出问题都能快速定位。4.3 实验追踪和权重管理我强烈建议从第一天就使用实验跟踪工具。你不需要一开始就搭一个完整的 MLflow Server先从最轻量的开始就行。我用的是 MLflow因为它开源、无额外费用且能管理模型产物。核心用法非常直接import mlflow mlflow.start_run(run_namebert-base-ft-v1) mlflow.log_params(cfg.train) # 记录超参数 mlflow.log_metric(val_f1, 0.87) # 记录指标 mlflow.log_artifact(checkpoints/best.pth) mlflow.end_run()这样每个 run 都会留下完整的超参数、指标曲线和权重文件。当你同时跑了二十个实验你不会再靠文件名区分“这个模型到底用了什么参数”。如果不方便跑服务端也可以用一个本地目录当 artifact store效果比什么都不记好得多。4.4 训练中的常见失败模式训练没跑通有很多固定套路。我把自己踩过的坑和解决方案整理成一张表现象常见原因解决思路loss 直接变成 NaN学习率太大降低学习率启用梯度裁剪训练集 loss 下降验证集上升过拟合减小模型容量、增加正则、早停验证集指标剧烈波动验证集太小增大验证集、固定 seed、多次重复评估训练速度极慢数据加载瓶颈用 arrow/parquet 缓存、调高 num_workers指标不动标签不均衡检查类别分布使用加权loss我最想强调的是遇到这些问题时不要盲目改参数。先把训练曲线和错误样本拿出来看一次只改一个变量。否则你会陷入“改了一堆东西不知道哪一步起了作用”的混乱里。这个习惯比任何技巧都重要。5. 评估与调优不要只看Accuracy一个数字5.1 从任务类型推导评估体系很多人评估模型只会看 Accuracy但真实任务里 Accuracy 往往会骗人。比如一个二分类问题95% 的样本都是正类你全预测成正类也有95%准确率但这显然不是好模型。正确的做法是从任务目标出发推导评估指标。我做了个常用参考表任务类型核心指标附加指标业务场景示例文本二分类F1、Precision、RecallAUC、PRC垃圾邮件识别、风险判断多分类Macro/Micro F1混淆矩阵、每类Precision/Recall工单分类、商品类目回归任务MAE、RMSER²、误差分布价格预估、销量预测排序任务NDCG、MRR点击率相关指标搜索、推荐生成任务BLEU、ROUGE人工评估、语义相似度摘要、对话这里还有一层经常被忽略的东西技术指标和业务指标是两张皮。技术指标再漂亮业务方关心的可能是“误判率到底是多少”“多少比例的用户受影响”。所以我在评估脚本里会额外算一组业务指标比如“坏样本率”“必须人工复核的比例”并在报告里同时呈现两个视角。5.2 错误分析怎么做才有效评估不是算完指标就结束了。真正决定你能不能再上一个台阶的是错误分析也就是明确模型在哪些样本上犯了错以及为什么犯错。我推荐的流程分四步。第一步从验证集里抽出所有预测错误的样本保存成一个单独的文件。第二步做一个简易的分布对比把错误样本在文本长度、发布时间、类别等维度上的分布和全量验证集做对比。第三步手动看 50 到 100 条错误样本按错误类型打标——是标签标错了、模型没学到足够上下文、还是存在歧义。第四步把高频错误类型反馈给数据处理流程而不是直接改模型。这套流程我实际用下来效果往往比直接换一个更大的模型要好。有一次做长文本分类错误分析发现很大比例的坏样本来自“多个话题混杂”的文本而这类文本在标签体系里本身就模棱两可。问题的解法不是换模型而是重新定义标签规则或者加入分段逻辑。5.3 调优的先后顺序数据先于模型简单先于复杂调优最怕的就是乱拳打死老师傅。我给自己定过一条明确的优先级顺序写下来供参考先确认数据质量没有问题。有没有重复样本、错标样本、边界样本处理不当。再确认训练流程稳定。loss 是否收敛是否过拟合随机种子是否固定。然后尝试特征侧优化。比如文本类任务的清洗规则、长度截断策略。接着调整模型容量和结构。比如换更大的预训练模型、加 attention 层。最后才考虑模型集成、迁移学习、复杂后处理。这个顺序的逻辑很简单数据侧改动的投入产出比通常最高模型侧改动的成本和不确定性更高。如果数据是脏的再好的模型也只是在拟合错误。如果你一次只做一件事并保留实验记录你会慢慢积累出“什么改动影响什么指标”的直觉这个直觉比任何自动化工具都值钱。6. 部署与服务化你在notebook里跑的模型到生产还要走三关6.1 离线预测、在线推理、批处理怎么选模型训练完之后真正的挑战才开始。部署形态不同工程复杂度天差地别。我习惯先把需求场景分为三类再决定技术方案。部署形态请求方式延迟要求典型场景离线批处理定时跑任务分钟到小时级日报生成、批量打标、夜间模型更新在线推理HTTP API 实时响应毫秒到秒级在线分类、搜索排序、实时风控边缘/嵌入式本地推理极低手机端OCR、端侧智能很多个人项目一上来就做 REST API其实完全没必要。如果业务是每天夜里跑一次批量标签写个脚本加定时任务就足够了。在线推理意味着你要维护一条服务链路包括负载、扩容、监控、版本切换复杂度高很多。先想清楚需求再决定要不要上服务化。6.2 把模型包成一个服务FastAPI里的最小部署当确实需要在线推理时我用 FastAPI 写了仓库里最精简的一个服务它的核心逻辑只有三块启动时加载一次模型、提供健康检查、提供预测接口。下面是src/serving/app.py的核心结构from fastapi import FastAPI from pydantic import BaseModel from src.serving.predictor import Predictor app FastAPI() predictor Predictor(model_pathcheckpoints/best.pth) class PredictRequest(BaseModel): text: str class PredictResponse(BaseModel): label: str score: float app.get(/health) def health(): return {status: ok} app.post(/predict, response_modelPredictResponse) def predict(req: PredictRequest): label, score predictor.predict(req.text) return PredictResponse(labellabel, scorescore)一些容易被忽略但很重要的点。第一模型加载一定要放在模块加载时而不是每次请求进来都 load 一次否则推理延迟会高到没法用。第二用 Pydantic 做输入校验既能让格式错误尽早暴露也能避免脏输入直接怼到模型里产生诡异异常。第三服务的进程数和模型显存/内存要做对应不能无脑开几十个 worker。6.3 性能与成本量化、批处理、缓存在线服务稳定跑起来之后下一步是压性能。我一般不追求极端优化而是先做三个常见操作。第一能静态化就静态化。如果模型和 tokenizer 都不需要实时更新直接把 PyTorch 模型导出成 ONNX或者用 TorchScript 做推理速度和稳定性都有提升。第二开启批处理。当请求量大的时候把相同迭代次数的请求攒在一起喂给模型GPU 利用率会明显上升。第三做结果缓存。对于文本分类这类任务大量请求是重复或高度相似的用一个简单的 LRU 缓存就能挡住很多穿透请求。这里要提醒一下优化必须基于监控数据而不是直觉。先看一下你的服务到底卡在网络、CPU还是GPU再决定做哪个优化。我见过有人花了很长时间做模型量化最后发现瓶颈根本不是推理时间而是数据加载那层太慢了。6.4 观测模型不只是代码也是数据部署完成后AI 系统有一个特性是普通软件没有的它的行为会随着输入数据的分布变化而劣化。你今天部署的模型很准三个月后业务数据变了它可能就开始出现大量误判。所以我给部署环节加了一组观测项至少包含这三类请求输入分布文本长度、热门词频、类别分布判断线上数据是否和训练集偏离。预测输出分布各类别预测比例、置信度均值发现异常崩塌。基础设施指标延迟、错误率、GPU内存等。这些日志我会落到单独的存储里并在每次重训后对比一次。重训的标准不应该是“每三个月固定重训一次”而应该是“数据漂移超过阈值就触发重训”。这个机制不一定很复杂哪怕只是一份每天自动出的分布报表也足够提前发现问题了。7. 把这份经验沉淀成你自己的工程习惯回到开始那个问题为什么要做 ai-engineering-from-scratch 这个项目因为工程化不是某个神秘的高阶技能而是一系列可复用的决策方式和操作习惯。如果让我重新做一遍这个仓库我会在第一天就把“可复现”摆在最高优先级。第二步是给每个脚本写清楚 usage 说明哪怕只是三行注释三个月后你都会感谢自己。第三步是学会记录坏样本它是推动数据迭代和模型迭代最有说服力的材料。把这个仓库里的目录结构、配置管理、训练脚本、错误分析清单、服务部署模板当成起点就够了。重要的是你做第一个项目的时候就用工程的方式去做而不是先跑通再补课。真正用起来之后你会慢慢发现那些写进文档的固定习惯最后都会变成你自己判断项目靠不靠谱的本能。