企业官网建设流程全解析

1. 从零搭建AI工程能力：为什么“手搓一遍”比调包更值钱

第一次看到ai-engineering-from-scratch这个标题，我脑子里蹦出来的不是某个具体项目，而是一种很熟悉的状态：手上有现成的框架、现成的API、现成的预训练权重，但一旦线上出问题，或者老板问一句“这个指标为什么掉了”，整个人就卡住了。这个标题戳中的，正是当下很多做AI应用的人共同的软肋——会用工具，但不懂工具背后的工程链路。

ai-engineering-from-scratch从字面拆解，核心是三个词：AI、Engineering、From Scratch。它不是教你调某个库的某个函数，而是把AI系统当成一个完整工程来对待，从最底层的数据处理、模型训练、推理服务、部署监控，一层一层自己搭起来。能做什么？能让你在没有任何现成脚手架的情况下，独立跑通一条从原始数据到线上服务的完整链路。解决什么问题？解决“只会调包、不会排障、不懂取舍”的工程能力断层。适合谁？适合已经会用Python、跑过几个demo、但想真正把AI系统落地到生产环境的开发者，也适合想转AI工程方向的 backend 或 data 同学。

我自己带过几个从算法转工程的同学，最常见的误区就是：以为模型精度高就万事大吉。结果一上线，QPS上不去、显存爆掉、数据管道堵死、日志里全是超时。这些问题，调包是学不会的，只有自己从零搭一遍，才会真正理解每个环节的边界在哪里。这篇就围绕这个标题，把我自己从零搭AI工程链路时踩过的坑、做过的取舍、以及那些文档里不会写的细节，完整拆一遍。

2. 整体设计思路：为什么“从零”不等于“重复造轮子”

2.1 先想清楚“从零”的边界在哪里

很多人一听“from scratch”，第一反应是什么都自己写，连矩阵乘法都手撸。这其实是误解。ai-engineering-from-scratch里的“从零”，指的是工程链路的从零，而不是数学库的从零。你完全可以用 NumPy 做数值计算、用 PyTorch 做自动求导，但数据怎么切分、特征怎么落盘、模型怎么序列化、服务怎么暴露、监控怎么埋点，这些必须自己设计一遍。

为什么这么定边界？因为AI工程的核心难点从来不是“实现一个卷积”，而是“让一个卷积在真实流量下稳定跑三个月”。框架帮你解决了算子问题，但解决不了工程问题。我见过太多团队，模型代码写得漂漂亮亮，结果数据版本对不上、训练和推理的特征处理不一致、线上模型和离线评估用的不是同一份权重，最后排查花的时间比开发还多。

所以整体设计的第一条原则：凡是涉及“状态”和“边界”的地方，必须自己掌控。数据版本、模型版本、配置版本、服务版本，这四个东西必须能对得上。框架可以帮你算，但不能帮你管。

2.2 分层设计：把AI系统拆成五层

我习惯把一条完整的AI工程链路拆成五层，从下往上依次是：

• 数据层：原始数据采集、清洗、切分、版本管理
• 特征层：特征计算、特征存储、训练/推理一致性保障
• 模型层：训练脚本、超参管理、模型评估、模型注册
• 服务层：推理接口、批处理、并发控制、资源隔离
• 观测层：日志、指标、追踪、告警

这五层不是随便分的，而是对应了AI系统最容易出问题的五个位置。数据层出问题，表现为“模型效果莫名其妙下降”；特征层出问题，表现为“离线AUC很高，线上效果很差”；模型层出问题，表现为“复现不了上次的结果”；服务层出问题，表现为“一上量就崩”；观测层出问题，表现为“出了问题不知道去哪查”。

ai-engineering-from-scratch的价值，就是让你把这五层都亲手搭一遍，知道每层的输入输出是什么、边界在哪里、失败模式是什么。下面我逐层拆。

2.3 技术选型的取舍逻辑

从零搭不代表要用最原始的工具。我的选型原则是：越靠近数据越用轻量工具，越靠近服务越用成熟组件。

数据层我用 Pandas + Parquet + DVC 的思路。Pandas 处理中小规模数据足够，Parquet 列式存储省空间且读取快，DVC 做数据版本管理，避免“数据被覆盖了但没人知道”。为什么不直接上 Spark？因为大多数团队的初期数据量根本不到需要分布式的程度，过早引入 Spark 只会增加运维负担。

特征层我用 Feast 或者自己写一套简单的特征注册表。核心是保证训练和推理用同一份特征计算逻辑。这一点极其关键，后面会展开。

模型层用 PyTorch + MLflow。PyTorch 灵活，MLflow 做实验追踪和模型注册。为什么不用更重的平台？因为从零搭的目的是理解链路，不是被平台绑架。

服务层用 FastAPI + Uvicorn + ONNX Runtime。FastAPI 写接口快，Uvicorn 做 ASGI 服务，ONNX Runtime 做推理加速。这套组合轻量、可控、容易排查。

观测层用 Prometheus + Grafana + 结构化日志。指标采集用 Prometheus 客户端库，日志用 JSON 格式，方便后续检索。

这套选型的核心逻辑是：每一层都要能单独替换、单独测试、单独观测。如果你搭出来的系统，换个模型就得改服务代码，那说明分层没做好。

3. 核心细节解析：数据、特征、模型三层的实操要点

3.1 数据层：版本管理比清洗更重要

数据层最容易犯的错，是把精力全花在清洗上，忽略了版本管理。清洗逻辑可以慢慢迭代，但版本管理一旦缺失，后面所有实验都不可复现。

我的做法是：原始数据只读，永远不覆盖。每次清洗产出新版本，用data/v1、data/v2这样的目录区分，同时记录一份manifest.json，包含：

{ "version": "v2", "created_at": "2025-01-15T10:30:00Z", "source": "data/raw/2025-01-14", "rows": 1250000, "columns": ["user_id", "item_id", "label", "ts"], "cleaning_script": "scripts/clean_v2.py", "hash": "a3f8c2..." }

这个hash是对整个数据文件算的，用来校验数据有没有被意外修改。别小看这一步，我遇到过好几次“明明没动数据但结果变了”的情况，最后发现是有人手动改了一个CSV。

切分数据时，时间序列场景一定要按时间切，不能随机切。随机切会导致未来信息泄露，离线指标虚高。我一般用 70% 训练、15% 验证、15% 测试，且验证集和测试集在时间上晚于训练集。

注意：数据切分后，训练集、验证集、测试集的分布要单独统计一遍。如果某个类别在测试集里占比异常，说明切分有问题，别急着训练。

3.2 特征层：训练/推理一致性是命门

特征层最核心的问题只有一个：训练时算的特征，和推理时算的特征，必须是同一套逻辑。

我见过最典型的翻车场景：训练时用 Pandas 做归一化，推理时用 NumPy 手写了一遍，结果均值方差算错了，线上效果直接掉一半。这种问题在离线评估里完全看不出来，因为离线用的还是训练那套逻辑。

解决办法是把特征计算逻辑抽成一个独立模块，训练和推理都调它。比如：

# features/user_features.py class UserFeatureCalculator: def __init__(self, stats): self.mean = stats["mean"] self.std = stats["std"] def compute(self, df): df["age_norm"] = (df["age"] - self.mean) / self.std return df

训练时从统计文件加载stats，推理时也从同一份统计文件加载。统计文件本身也要版本化，跟模型版本绑定。

特征存储方面，离线特征落 Parquet，在线特征落 Redis 或本地缓存。关键是要有一份特征注册表，记录每个特征的名称、类型、计算逻辑、依赖数据、负责人。这份注册表就是特征层的“合同”，谁改了特征都要更新它。

3.3 模型层：实验追踪不是可选项

模型层最容易忽略的是实验追踪。很多人跑实验靠记笔记，结果一周后回头看，不知道哪个参数对应哪个结果。ai-engineering-from-scratch要求你把实验追踪当成基础设施来搭。

我用 MLflow 记录每次实验的：

• 超参数（学习率、batch size、层数等）
• 指标（loss、AUC、F1等）
• 产物（模型权重、配置文件、特征统计）
• 代码版本（git commit hash）

这样任何一次结果都能复现。复现的时候，checkout 对应 commit，加载对应数据版本和特征统计，跑一遍就能得到一样的结果。

模型评估不能只看一个指标。分类任务我至少看 AUC、F1、Recall@K；排序任务看 NDCG、MAP。而且要分桶看，比如按用户活跃度分桶，避免模型只在头部用户上好用。

模型注册环节，每个上线的模型都要有明确的版本号、评估报告、回滚方案。我一般用model_name/v1、model_name/v2这样的命名，配合 MLflow 的 Model Registry 管理阶段（Staging、Production、Archived）。

4. 实操过程：从原始数据到线上服务的完整链路

4.1 环境准备与依赖锁定

第一步永远是环境。我习惯用pyproject.toml管理依赖，配合uv或poetry锁定版本。为什么强调锁定？因为AI工程链路长，任何一个依赖版本漂移都可能导致结果不一致。

# 初始化项目 uv init ai-engineering-from-scratch cd ai-engineering-from-scratch # 添加核心依赖 uv add numpy pandas pyarrow scikit-learn uv add torch --index-url https://download.pytorch.org/whl/cpu uv add fastapi uvicorn onnx onnxruntime uv add mlflow prometheus-client

依赖锁定后，生成uv.lock，提交到 git。任何人 clone 下来，跑uv sync就能得到完全一致的环境。

提示：PyTorch 的 CPU 版和 GPU 版依赖不同，团队内要统一。如果本地开发用 CPU、线上用 GPU，至少保证算子行为一致，避免出现“本地能跑线上报错”。

4.2 数据管道搭建：从原始文件到训练集

假设原始数据是一批 CSV，放在data/raw/下。第一步是写一个清洗脚本，产出data/processed/v1/。

# scripts/build_dataset.py import pandas as pd import hashlib import json from pathlib import Path RAW_DIR = Path("data/raw") OUT_DIR = Path("data/processed/v1") OUT_DIR.mkdir(parents=True, exist_ok=True) def load_raw(): frames = [] for f in RAW_DIR.glob("*.csv"): frames.append(pd.read_csv(f)) return pd.concat(frames, ignore_index=True) def clean(df): df = df.dropna(subset=["user_id", "item_id", "label"]) df = df.drop_duplicates(subset=["user_id", "item_id", "ts"]) df["ts"] = pd.to_datetime(df["ts"]) return df def split_by_time(df): df = df.sort_values("ts") n = len(df) train = df.iloc[: int(n * 0.7)] valid = df.iloc[int(n * 0.7) : int(n * 0.85)] test = df.iloc[int(n * 0.85) :] return train, valid, test if __name__ == "__main__": df = clean(load_raw()) train, valid, test = split_by_time(df) train.to_parquet(OUT_DIR / "train.parquet") valid.to_parquet(OUT_DIR / "valid.parquet") test.to_parquet(OUT_DIR / "test.parquet") manifest = { "version": "v1", "rows": len(df), "train_rows": len(train), "valid_rows": len(valid), "test_rows": len(test), } (OUT_DIR / "manifest.json").write_text(json.dumps(manifest, indent=2))

这个脚本跑完，数据层就有了第一个可复现的版本。注意manifest.json里记录了行数，后续任何环节发现行数对不上，都能快速定位。

4.3 特征计算与统计落盘

特征计算脚本要独立于训练脚本。我一般放在features/目录下，训练和推理都 import 它。

# features/build_features.py import pandas as pd import json from pathlib import Path class FeatureBuilder: def __init__(self, stats=None): self.stats = stats or {} def fit(self, df): self.stats["age_mean"] = float(df["age"].mean()) self.stats["age_std"] = float(df["age"].std()) return self def transform(self, df): df = df.copy() df["age_norm"] = (df["age"] - self.stats["age_mean"]) / self.stats["age_std"] df["hour"] = df["ts"].dt.hour return df def save(self, path): Path(path).write_text(json.dumps(self.stats, indent=2)) @classmethod def load(cls, path): stats = json.loads(Path(path).read_text()) return cls(stats)

训练时：

builder = FeatureBuilder().fit(train_df) builder.save("artifacts/feature_stats_v1.json") train_feat = builder.transform(train_df) valid_feat = builder.transform(valid_df)

推理时：

builder = FeatureBuilder.load("artifacts/feature_stats_v1.json") input_feat = builder.transform(input_df)

这样训练和推理用的是同一份统计、同一套逻辑，一致性有保障。

4.4 模型训练与评估

训练脚本要记录所有超参和指标。我用 MLflow 做追踪：

import mlflow import torch mlflow.set_experiment("ai-engineering-from-scratch") with mlflow.start_run(): mlflow.log_params({"lr": 1e-3, "batch_size": 256, "epochs": 10}) # ... 训练循环 ... mlflow.log_metric("valid_auc", 0.87) mlflow.pytorch.log_model(model, "model")

评估不能只看整体指标，要分桶。比如按用户活跃度分桶：

def evaluate_by_bucket(df, preds): df = df.copy() df["pred"] = preds df["bucket"] = pd.qcut(df["user_activity"], q=4, labels=["low", "mid_low", "mid_high", "high"]) return df.groupby("bucket").apply(lambda g: roc_auc_score(g["label"], g["pred"]))

如果发现某个桶的AUC明显低，说明模型在这个人群上表现差，需要针对性优化。

4.5 模型导出与推理服务

训练完的 PyTorch 模型导出成 ONNX，推理时用 ONNX Runtime，性能更好且不依赖 PyTorch。

import torch.onnx dummy_input = torch.randn(1, feature_dim) torch.onnx.export( model, dummy_input, "artifacts/model_v1.onnx", input_names=["input"], output_names=["output"], dynamic_axes={"input": {0: "batch"}, "output": {0: "batch"}}, )

服务用 FastAPI：

from fastapi import FastAPI import onnxruntime as ort import numpy as np app = FastAPI() session = ort.InferenceSession("artifacts/model_v1.onnx") builder = FeatureBuilder.load("artifacts/feature_stats_v1.json") @app.post("/predict") def predict(payload: dict): df = pd.DataFrame([payload]) feat = builder.transform(df) x = feat[FEATURE_COLUMNS].values.astype(np.float32) out = session.run(None, {"input": x})[0] return {"score": float(out[0][0])}

启动：

uvicorn app:app --host 0.0.0.0 --port 8000 --workers 4

--workers 4表示起4个进程，充分利用多核。具体起几个，看CPU核数和单次推理耗时。我一般按核数 / 2起步，压测后再调。

4.6 观测层埋点

服务起来后，必须埋指标。Prometheus 客户端库很简单：

from prometheus_client import Counter, Histogram, make_asgi_app REQUEST_COUNT = Counter("predict_requests_total", "Total predict requests") REQUEST_LATENCY = Histogram("predict_latency_seconds", "Predict latency") ERROR_COUNT = Counter("predict_errors_total", "Total predict errors") @app.post("/predict") def predict(payload: dict): REQUEST_COUNT.inc() with REQUEST_LATENCY.time(): try: # ... 推理逻辑 ... return {"score": score} except Exception: ERROR_COUNT.inc() raise

再把/metrics挂到 FastAPI：

app.mount("/metrics", make_asgi_app())

Grafana 里配几个面板：QPS、P99延迟、错误率、模型版本分布。这四个指标基本能覆盖大部分线上问题。

5. 常见问题与排查技巧实录

5.1 离线指标好、线上效果差

这是最经典的问题。排查顺序：

1. 特征一致性：训练和推理的特征计算逻辑是否完全一致？统计文件是否同一版本？
2. 数据分布：线上流量分布和训练数据分布是否差异过大？看特征均值、方差、缺失率。
3. 标签延迟：线上标签是不是延迟回传的？如果是，评估时要用延迟后的标签。
4. 模型版本：线上加载的模型是不是最新版本？看服务启动日志里的模型hash。

我遇到过一次，排查两天，最后发现是推理服务加载的特征统计文件是旧的，归一化参数不对。所以现在我的服务启动日志里一定会打印特征统计文件的hash和模型文件的hash。

5.2 服务一上量就崩

通常是资源问题。排查清单：

我的经验是，推理服务一定要设超时和限流。FastAPI 可以用asyncio.wait_for包一层，超时直接返回降级结果。限流用slowapi或自己在网关层做。

5.3 实验结果复现不了

九成是版本问题。检查：

• 代码版本（git commit）
• 数据版本（manifest hash）
• 特征统计版本
• 依赖版本（lock文件）
• 随机种子

随机种子要显式设置：

import random import numpy as np import torch def set_seed(seed=42): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed)

注意，即使设了种子，GPU上的某些算子仍可能有非确定性。如果要求严格复现，要开torch.use_deterministic_algorithms(True)，但会牺牲一些性能。

5.4 数据管道跑得越来越慢

数据量涨了之后，Pandas 单机处理会变慢。优化顺序：

1. 用 Parquet 替代 CSV，读取快很多
2. 只读需要的列，用columns=参数
3. 用category类型存低基数列
4. 分块处理，用chunksize
5. 还不行再考虑 Polars 或 DuckDB

我实测下来，同样一份数据，Parquet + 列裁剪比全量CSV读取快5到10倍。大多数场景根本不需要上分布式。

注意：数据管道优化前先profile，别凭感觉优化。用cProfile或py-spy看时间花在哪。

5.5 模型更新后如何安全回滚

模型回滚必须能在分钟级完成。我的做法是：

• 模型文件按版本命名，model_v1.onnx、model_v2.onnx
• 服务启动时从配置读模型路径
• 配置支持热更新，或者用双buffer切换
• 保留最近3个版本，旧的归档

回滚时只改配置，重启服务，加载旧版本模型。整个过程不超过5分钟。如果做不到，说明部署流程有问题。

6. 我踩过的坑和几条实在建议

搭这套链路的过程中，有几个坑我印象特别深。第一个是特征统计文件没版本化，训练用v1、推理用v2，线上效果掉了30%，排查了一整天才定位到。从那以后，我把特征统计、模型、配置三者的版本号绑在一起，任何一个变了都要重新走评估流程。

第二个坑是ONNX导出时没设dynamic_axes，结果只能处理batch size为1的请求，批量推理直接报错。这个错误在导出时不会提示，只有实际跑批量请求才暴露。所以导出后一定要用不同batch size测一遍。

第三个坑是日志打太多，把磁盘写满了。推理服务每个请求打一条详细日志，QPS一高，日志量爆炸。后来改成采样打日志，正常请求只打指标，异常请求才打详细日志。

几条实在建议：第一，任何环节都要能单独测试，数据管道、特征计算、模型推理，各自有单元测试，别等联调才发现问题。第二，版本号要贯穿始终，数据、特征、模型、配置，四个版本号能对上，排查问题就快一半。第三，观测层要提前搭，别等出问题才想起来加日志，那时候已经晚了。第四，从零搭的目的是理解链路，不是拒绝工具，该用框架的地方用框架，但要知道框架帮你做了什么、没做什么。

这套东西搭完，你对AI工程的理解会上一个台阶。以后再遇到线上问题，脑子里会有一张完整的链路图，知道从哪查、怎么查。这比会调多少个API值钱得多。