企业官网建设流程全解析

1. 为什么我要从零手搓一套AI工程化流程

第一次看到ai-engineering-from-scratch这个项目名的时候，我正被一堆散落在各处的实验脚本折磨得够呛。Jupyter Notebook 里躺着十几个版本的模型训练代码，文件名从train_final.py一路排到train_final_v3_really_final.py，每次想复现两周前跑出来的那个还不错的结果，都得花半天时间回忆当时到底改了哪个参数。这种状态持续了大概三个月，直到我下定决心，把整个AI工程化的链路从头到尾梳理一遍，用最朴素的方式重新搭一套能跑通、能复现、能交接的流程。

这个项目标题里的 "from scratch" 其实有两层意思。一层是字面上的从零开始，不依赖那些开箱即用的大平台，自己动手把数据、训练、评估、部署这几块拼起来；另一层是认知上的从零开始，把那些平时被框架封装掉的细节重新摊开来看清楚，搞明白每一步到底在干什么。我见过太多人包括我自己，调包调得很熟练，但一旦遇到框架不支持的场景就抓瞎，根本原因就是中间那层黑盒从来没打开过。

这套流程适合谁呢？如果你是一个刚入行不久、想搞清楚AI项目从数据到上线完整链路的工程师，或者是一个在中小团队里需要一个人扛起整个模型迭代流程的开发者，再或者你只是单纯好奇那些大厂里的MLOps到底在做什么，那这篇内容应该能给你一些可以直接抄作业的东西。我不打算讲太玄乎的理论，重点放在每个环节为什么这么设计、实际踩过哪些坑、以及怎么用最少的工具把事办成。

整个流程我大概花了六周时间打磨，中间推翻重来过两次，最终稳定下来的版本支撑了我后续半年的模型迭代工作。下面我把这套东西拆开来讲，从整体设计思路到每个模块的具体实现，再到实际运行中遇到的各种幺蛾子，尽量讲透。

2. 整体架构设计与技术选型思路

2.1 为什么选择"轻量级组合"而不是一体化平台

市面上做AI工程化的方案大致分两派。一派是重量级的一体化平台，功能大而全，从数据标注到模型监控全给你包圆了；另一派是轻量级的工具组合，每个环节用最合适的工具，通过约定和脚本来串联。我最终选了后者，原因很实际：我手上的项目规模不大，数据量在几十万条这个级别，模型也是中等体量的微调任务，用一体化平台属于杀鸡用牛刀，光是平台本身的运维成本就够我喝一壶的。

轻量级组合的另一个好处是可控。每个环节的输入输出都是明确定义的文件或接口，出了问题能快速定位到具体是哪个环节挂了。一体化平台虽然省事，但一旦出问题，排查起来往往要翻平台自己的日志，有时候还得等社区回复，节奏完全不在自己手里。我印象特别深的一次，某个平台的训练任务莫名其妙卡住，查了两天最后发现是平台内部的一个资源调度bug，这种问题在自建流程里根本不会出现。

当然轻量级组合也有代价，就是需要自己写不少胶水代码。但这些胶水代码写一次就能复用很久，而且写的过程中你会对整个流程的理解越来越深，这笔账算下来是划算的。

2.2 核心模块的划分与职责边界

我把整个流程拆成了五个核心模块，每个模块只干一件事，模块之间通过文件系统来传递数据。这种设计看起来有点土，但实际用下来非常稳。

这个划分的关键在于职责边界要清晰。比如数据准备模块只负责把原始数据变成模型能吃的格式，它不关心模型是什么；训练引擎只负责根据配置训练，它不关心数据是怎么来的。这种解耦带来的好处是，我想换一个模型架构，只需要改训练引擎和配置文件，数据准备那块完全不用动。

2.3 目录结构的设计与约定

目录结构这个东西看起来不起眼，但设计不好后期会非常痛苦。我最终采用的方案是按功能划分顶层目录，每个目录内部再按时间或版本分子目录。

project/ ├── data/ │ ├── raw/ # 原始数据，只读 │ ├── processed/ # 清洗后的数据 │ └── splits/ # 训练/验证/测试切分 ├── configs/ │ ├── base.yaml # 基础配置 │ └── experiments/ # 各实验的覆盖配置 ├── src/ │ ├── data/ # 数据处理代码 │ ├── train/ # 训练代码 │ ├── eval/ # 评估代码 │ └── serve/ # 服务代码 ├── experiments/ # 实验输出 │ └── exp_20240101_001/ │ ├── config.yaml # 本次实验的完整配置 │ ├── checkpoints/ # 模型权重 │ ├── logs/ # 训练日志 │ └── metrics.json # 评估指标 └── scripts/ # 各种入口脚本

这个结构里有个关键约定：experiments 目录下的每个子目录都是一次完整的实验记录，包含复现这次实验所需的一切信息。只要拿到这个目录，就能知道当时用了什么配置、什么数据、跑出了什么结果。这个约定后来救了我很多次，尤其是当老板问"上个月那个效果不错的版本是怎么跑出来的"的时候，我直接翻出对应的目录就行。

2.4 配置管理的取舍

配置管理这块我纠结了很久，试过三种方案。第一种是纯命令行参数，简单直接但参数一多就记不住；第二种是Python字典写在代码里，灵活但容易和代码耦合太深；第三种是YAML配置文件加命令行覆盖，最终选了这个。

YAML的好处是可读性好，非技术人员也能看懂大概。我采用的模式是有一个base.yaml存放所有默认配置，然后每个实验有一个小的覆盖文件，只写和默认值不一样的部分。运行时把两个文件合并，生成最终的配置。这样既避免了重复，又能清楚地看到每次实验改了什么。

# base.yaml 片段 model: name: "bert-base" max_length: 128 dropout: 0.1 training: batch_size: 32 learning_rate: 2e-5 epochs: 3 warmup_ratio: 0.1 data: train_file: "data/splits/train.jsonl" valid_file: "data/splits/valid.jsonl"

# experiments/exp_001.yaml 覆盖配置 training: learning_rate: 1e-5 epochs: 5

合并逻辑我用了一个简单的递归字典合并函数，大概二十行代码，比引入第三方库更可控。这里有个细节要注意：列表类型的配置不要做合并，直接覆盖。我一开始想当然地做了列表合并，结果有次改数据路径的时候，新旧路径被合并成了一个列表，训练脚本读到一个不存在的路径直接崩了，排查了半天才发现是合并逻辑的问题。

3. 数据准备环节的核心细节与实操

3.1 数据清洗的标准化流程

数据清洗这块看起来简单，实际上是最容易埋雷的地方。我总结了一套标准化的流程，每次拿到新数据都按这个流程走一遍。

第一步是格式统一。不管原始数据是CSV、JSON还是数据库导出的，先统一转成JSONL格式，每行一个样本。JSONL的好处是流式读取方便，不会因为文件太大把内存撑爆，而且出错了能快速定位到具体哪一行。

第二步是字段规范化。我遇到过原始数据里同一个含义的字段有五六种命名的情况，比如"文本"这个字段，有的叫text，有的叫content，有的叫sentence。这时候要统一成一个标准字段名，我一般用text和label这两个最通用的名字。

第三步是异常样本过滤。常见的异常包括空文本、超长文本、标签缺失、编码错误等。这里有个经验：过滤规则要记录日志，每次过滤掉了多少条、分别是什么原因，都要写清楚。我有次发现清洗后数据少了一大半，查日志才发现是编码检测太严格，把一堆正常的中文文本误判成了乱码。

def clean_sample(sample): """单条样本的清洗逻辑""" # 去除首尾空白 text = sample.get("text", "").strip() # 空文本过滤 if not text: return None, "empty_text" # 超长文本截断（不是过滤，是截断） if len(text) > 10000: text = text[:10000] # 标签检查 label = sample.get("label") if label is None: return None, "missing_label" return {"text": text, "label": label}, None

3.2 数据切分的坑与正确姿势

数据切分看起来就是随机分三份，但实际操作里有几个坑我踩过。

第一个坑是切分前没有打乱。如果原始数据是按类别排序的，直接按比例切分会导致训练集和验证集的类别分布差异巨大。我一开始就犯过这个错，训练集里正样本占80%，验证集里正样本只占20%，结果验证指标一直上不去，还以为是模型有问题。

第二个坑是数据泄漏。有些数据集里同一个样本会以不同形式出现多次，如果随机切分，同一个样本的变体可能同时出现在训练集和验证集里，导致验证指标虚高。解决办法是在切分前先做去重，或者按样本的来源ID来切分，保证同一个来源的样本只出现在一个集合里。

第三个坑是切分比例固定不变。小数据集上这个问题特别明显，不同的随机种子切出来的验证集指标能差好几个点。我的做法是对于小数据集做多次切分取平均，或者用交叉验证。数据量大的话就固定一个种子，保证可复现。

import random from collections import defaultdict def stratified_split(samples, ratios=(0.8, 0.1, 0.1), seed=42): """按标签分层切分，保证各集合类别分布一致""" random.seed(seed) # 按标签分组 by_label = defaultdict(list) for s in samples: by_label[s["label"]].append(s) train, valid, test = [], [], [] for label, items in by_label.items(): random.shuffle(items) n = len(items) n_train = int(n * ratios[0]) n_valid = int(n * ratios[1]) train.extend(items[:n_train]) valid.extend(items[n_train:n_train + n_valid]) test.extend(items[n_train + n_valid:]) # 最后再打乱一次 random.shuffle(train) random.shuffle(valid) random.shuffle(test) return train, valid, test

3.3 数据版本化的轻量方案

数据版本化这块我用了一个非常土但有效的方案：每次数据处理生成一个版本号，版本号由处理脚本的哈希和输入数据的哈希组合而成。处理脚本变了或者输入数据变了，版本号就会变，这样就能保证同样的版本号对应的数据是完全一样的。

具体实现上，我在数据目录下放一个manifest.json，记录每个版本的数据来源、处理脚本的哈希、样本数量、生成时间等信息。训练的时候配置里指定数据版本号，训练脚本会校验实际数据的版本号是否匹配，不匹配就直接报错。这个校验机制帮我避免了好几次"用错数据训练"的事故。

提示：数据版本号不要用时间戳，时间戳虽然唯一但不可复现。用内容哈希虽然计算稍慢，但能保证相同内容得到相同版本号，这对复现实验至关重要。

3.4 实操心得：数据质量检查清单

在数据进入训练之前，我一定会跑一遍质量检查，清单如下：

• 样本总数是否和预期一致，差异超过5%就要查原因
• 各标签的样本数量分布，是否存在严重不平衡
• 文本长度的分布，P99长度是多少，决定模型的最大长度设置
• 是否有重复样本，重复率是多少
• 随机抽20条人工看一眼，确认格式和内容正常

这个清单看起来简单，但每次都能发现点问题。有次抽检发现一批数据的标签全是0，查下来是上游导出的时候字段映射错了，如果没检查直接训练，模型会学成一个只会输出0的废物。

4. 训练引擎的搭建与关键配置

4.1 训练脚本的骨架设计

训练脚本我改过很多版，最终稳定下来的骨架包含这几个部分：配置加载、数据加载、模型初始化、优化器和调度器设置、训练循环、验证循环、检查点保存。每个部分都独立成函数，主函数只负责串联。

这种设计的好处是每个部分都能单独测试。比如我想验证数据加载是否正确，直接调用数据加载函数打印几条样本就行，不用跑整个训练。模型初始化也是，可以先初始化完打印一下参数量，确认没问题再开始训练。

def train(config): # 1. 设置随机种子 set_seed(config["seed"]) # 2. 加载数据 train_loader = build_dataloader(config, split="train") valid_loader = build_dataloader(config, split="valid") # 3. 初始化模型 model = build_model(config) model.to(config["device"]) # 4. 优化器和调度器 optimizer = build_optimizer(model, config) scheduler = build_scheduler(optimizer, config, len(train_loader)) # 5. 训练循环 best_metric = 0 for epoch in range(config["training"]["epochs"]): train_one_epoch(model, train_loader, optimizer, scheduler, epoch) metrics = evaluate(model, valid_loader) # 6. 保存最佳模型 if metrics["f1"] > best_metric: best_metric = metrics["f1"] save_checkpoint(model, config, epoch, metrics) log_metrics(epoch, metrics)

4.2 随机种子与可复现性

可复现性这块我踩的坑最多。理论上设置一个随机种子就完事了，但实际上影响结果的因素远不止一个种子。

首先是框架层面的随机性。PyTorch里除了torch.manual_seed，还有CUDA的随机性、cuDNN的算法选择等。要完全固定需要设置好几个地方：

import torch import numpy as np import random def set_seed(seed): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) # 关闭cuDNN的自动算法选择 torch.backends.cudnn.deterministic = True torch.backends.cudnn.benchmark = False

其次是数据加载的随机性。DataLoader的shuffle、worker的初始化都有随机性，需要给worker设置种子。还有就是如果用了数据增强，增强本身的随机性也要控制。

最后是浮点运算的非确定性。即使设置了所有种子，某些CUDA操作的结果仍然可能有微小差异，这是硬件层面的问题，没法完全消除。我的做法是接受这个现实，在评估的时候用足够的样本量，让这点微小差异不影响最终判断。

注意：cudnn.deterministic = True会降低训练速度，大概慢10%到20%。如果对速度敏感，可以只在需要精确复现的时候开启，日常实验可以关掉。

4.3 超参数配置的经验值

超参数这块没有万能公式，但有一些经验值可以参考。我整理了一个表格，是我在文本分类任务上常用的配置范围。

学习率是最关键的参数，我的经验是先用一个中等值跑一遍，看loss曲线。如果loss下降太慢就调大，如果loss震荡或者变成nan就调小。warmup的作用是在训练初期用较小的学习率，等模型稳定了再升到设定值，这个对微调任务特别重要，能明显提升稳定性。

4.4 检查点保存策略

检查点保存看起来简单，但策略不对会浪费大量磁盘空间或者丢失最佳模型。我的策略是保存两个检查点：最新的和最佳的。

最新的检查点用于断点续训，每次epoch结束覆盖保存。最佳的检查点用于最终部署，只在验证指标超过历史最佳时保存。这样既保证了能续训，又不会因为保存太多检查点把磁盘撑爆。

检查点里要保存的东西也有讲究，除了模型权重，还要保存优化器状态、调度器状态、当前epoch、最佳指标值。这些信息在续训的时候都需要。我一开始只保存了模型权重，结果续训的时候优化器状态丢了，学习率调度从头开始，训练效果明显变差。

def save_checkpoint(model, optimizer, scheduler, epoch, metric, path): torch.save({ "model_state": model.state_dict(), "optimizer_state": optimizer.state_dict(), "scheduler_state": scheduler.state_dict(), "epoch": epoch, "best_metric": metric, }, path)

4.5 实操心得：训练过程中的监控要点

训练过程中我主要盯三个东西：loss曲线、学习率曲线、验证指标。

loss曲线看的是训练是否正常。正常的loss应该是先快速下降然后逐渐平缓，如果loss一直不降说明学习率太小或者模型有问题，如果loss震荡剧烈说明学习率太大，如果loss突然变成nan说明梯度爆炸了。

学习率曲线看的是调度器是否正常工作。warmup阶段学习率应该从0线性升到设定值，然后逐渐衰减。如果学习率曲线不对，说明调度器配置有问题。

验证指标看的是模型是否过拟合。如果训练loss持续下降但验证指标开始下降，说明过拟合了，应该早停或者加正则化。

我一般每100个step打印一次训练loss，每个epoch结束打印一次验证指标。日志用简单的文本格式，方便后续用脚本解析。

5. 评估模块与实验追踪的实现

5.1 评估指标的选择与计算

评估指标的选择取决于任务类型。分类任务我一般看准确率、精确率、召回率、F1值这四个。准确率在类别不平衡的时候会失真，所以主要看F1。精确率和召回率能看出模型是偏保守还是偏激进，根据业务需求来权衡。

指标计算我踩过一个坑：多分类的F1要区分macro和micro。macro是每个类别算F1再平均，micro是全局算F1。类别不平衡的时候这两个差异很大，macro会被小类别拉低，micro会被大类别主导。我一般两个都报，让看的人自己判断。

from sklearn.metrics import accuracy_score, precision_recall_fscore_support def compute_metrics(preds, labels): acc = accuracy_score(labels, preds) p_macro, r_macro, f_macro, _ = precision_recall_fscore_support( labels, preds, average="macro" ) p_micro, r_micro, f_micro, _ = precision_recall_fscore_support( labels, preds, average="micro" ) return { "accuracy": acc, "precision_macro": p_macro, "recall_macro": r_macro, "f1_macro": f_macro, "f1_micro": f_micro, }

5.2 实验追踪的极简方案

实验追踪这块我用了一个极简方案：每次实验生成一个目录，目录里放一个metrics.json记录所有指标，再放一个config.yaml记录完整配置。然后写一个脚本扫描所有实验目录，把指标汇总成一个表格。

这个方案的好处是不依赖任何外部服务，所有数据都在本地文件系统里，随时可以查看和备份。缺点是没法做复杂的查询和可视化，但对于我这种实验数量在几百次这个级别的场景完全够用。

汇总脚本大概长这样：

import json from pathlib import Path import pandas as pd def collect_experiments(exp_dir): records = [] for exp_path in Path(exp_dir).iterdir(): if not exp_path.is_dir(): continue metrics_file = exp_path / "metrics.json" config_file = exp_path / "config.yaml" if not metrics_file.exists(): continue with open(metrics_file) as f: metrics = json.load(f) record = {"exp_name": exp_path.name} record.update(metrics) records.append(record) df = pd.DataFrame(records) return df.sort_values("f1_macro", ascending=False)

5.3 结果对比与版本管理

有了实验汇总表之后，对比不同实验的结果就方便了。我一般会关注几个维度：哪个配置组合效果最好、不同随机种子的方差有多大、哪些改动带来了提升。

这里有个经验：单次实验的结果不要全信，要看多次实验的稳定性。我有次调了一个参数，单次实验F1提升了2个点，高兴得不行，结果换了三个种子重跑，平均下来反而降了0.5个点。后来我养成了习惯，重要的改动至少跑三个种子，看平均值和方差。

版本管理这块，代码用git管理，数据和模型用版本号管理，配置跟着实验目录走。这三者通过实验目录里的config.yaml关联起来，里面记录了代码的commit hash、数据版本号、模型版本号。这样任何时候都能追溯到一次实验的完整信息。

5.4 实操心得：评估阶段的常见陷阱

评估阶段有几个陷阱我踩过，这里列出来提醒一下。

第一个是测试集不能反复用。如果根据测试集的结果来调参，测试集就变成了验证集，最终报告的指标会虚高。我的做法是测试集只在最终确定模型后跑一次，中间调参只看验证集。

第二个是评估脚本要和训练脚本用同一套数据预处理。我有次评估的时候忘了应用训练时的文本截断，导致评估结果和训练时的验证结果对不上，查了半天才发现是预处理不一致。

第三个是指标的计算方式要和业务对齐。比如业务上更关心高精确率，那评估的时候就要重点看精确率，而不是只看F1。这个需要和业务方提前沟通清楚。

6. 部署服务与常见问题排查

6.1 推理服务的轻量实现

部署这块我用的是FastAPI加Uvicorn的组合，简单够用。核心逻辑就是加载模型，提供一个HTTP接口接收文本返回预测结果。

from fastapi import FastAPI from pydantic import BaseModel import torch app = FastAPI() model = None tokenizer = None class PredictRequest(BaseModel): text: str class PredictResponse(BaseModel): label: int confidence: float @app.on_event("startup") def load_model(): global model, tokenizer model = build_model(config) model.load_state_dict(torch.load("best_model.pt")) model.eval() tokenizer = build_tokenizer(config) @app.post("/predict", response_model=PredictResponse) def predict(req: PredictRequest): inputs = tokenizer(req.text, return_tensors="pt", truncation=True, max_length=128) with torch.no_grad(): outputs = model(**inputs) probs = torch.softmax(outputs.logits, dim=-1) label = int(probs.argmax()) confidence = float(probs.max()) return PredictResponse(label=label, confidence=confidence)

这个服务启动后监听一个端口，用curl就能测试。生产环境的话前面再加个Nginx做反向代理和负载均衡，基本就够用了。

6.2 性能优化的几个手段

推理性能优化我做了三件事。第一件是模型量化，把FP32的权重转成INT8，模型体积缩小到四分之一，推理速度提升大概两倍，精度损失在可接受范围内。第二件是批处理，把多个请求攒一批一起推理，吞吐量能提升好几倍。第三件是缓存，对于重复的输入直接返回缓存结果，这个在有些场景下效果很明显。

量化用PyTorch自带的动态量化就行，几行代码的事：

quantized_model = torch.quantization.quantize_dynamic( model, {torch.nn.Linear}, dtype=torch.qint8 )

批处理需要在服务层做，维护一个请求队列，攒够一批或者超时了就触发推理。这个实现起来稍复杂，但收益很大，尤其是QPS高的时候。

6.3 常见问题速查表

实际运行中遇到的问题我整理成了一个速查表，方便快速定位。

6.4 实操心得：上线前的检查清单

服务上线前我一定会跑一遍检查清单，这里分享出来。

• 模型文件是否正确加载，参数量是否和预期一致
• 单条请求的延迟是否在可接受范围内
• 并发请求下结果是否正确，有没有出现错乱
• 异常输入（空文本、超长文本、特殊字符）是否能正常处理
• 服务重启后是否能自动恢复
• 日志是否完整，出问题能否快速定位

这个清单帮我避免了好几次线上事故。有次就是忘了测异常输入，结果上线后遇到一个超长文本直接把服务打挂了，后来加了长度限制才解决。

7. 我在这套流程上踩过的坑和最终体会

这套流程从最初的想法到最终稳定运行，中间踩的坑比我预想的多得多。最大的一个坑是过度设计。一开始我想把每个环节都做得尽善尽美，数据版本化要支持回滚，实验追踪要支持可视化，部署要支持灰度发布。结果花了两周时间搭架子，真正跑实验的时间反而没多少。后来我砍掉了大部分花哨功能，只保留最核心的，反而效率高了很多。

第二个坑是忽视文档。有段时间我改代码改得很勤，但懒得更新文档，结果两周后自己都忘了某个参数是干什么的。后来我强制自己每次改完代码花五分钟更新一下对应的文档，这个习惯坚持下来受益很大。

第三个坑是不重视日志。早期我的日志就是简单的print，出了问题根本查不到原因。后来改成了结构化的日志，每条日志带时间戳、模块名、日志级别，排查问题的效率提升了好几倍。

如果让我给刚开始搭这套流程的人一个建议，那就是先跑通再优化。不要一上来就追求完美，先用最土的办法把整个链路跑通，然后再逐个环节优化。跑通的过程中你会对每个环节有更实际的理解，这时候再优化才知道该往哪个方向优化。

这套流程后来我又扩展了一些东西，比如加了简单的A/B测试框架，加了模型监控的指标采集，但核心的骨架一直没变。我觉得一个好的工程化流程就应该是这样，骨架稳定，细节可以不断迭代。