1. AI工程到底是什么我为什么从零开始重造轮子先交代一下背景。过去几年我一直在做算法相关的工作简历上写着“机器学习工程师”平时主要任务就是训练模型、调参、跑实验。但真正把我逼到必须重新思考“AI工程”这件事的是一次极其狼狈的落地经历模型在 Notebook 里跑得好好的精确率接近 90%等我要把它变成一个线上服务的时候才发现完全不是那么回事。数据管道是临时脚本拼的模型保存格式不统一线上接口不知道怎么处理异常输入日志打印得乱七八糟性能更是惨不忍睹——单次推理要 200 毫秒并发一上来直接内存溢出。那次之后我意识到训练模型只是 AI 项目里很小的一段真正决定一个系统能不能长期稳定运行下去的是工程能力。这也是我启动ai-engineering-from-scratch这个项目的原因不借助任何重量级平台不依赖现成的 MLOps 全家桶而是从最朴素的需求出发亲手把一条完整的 AI 工程链路搭起来。这个项目适合谁我觉得至少有三类人可以从中获得参考。第一类是刚入行的算法工程师你迟早会发现代码能跑和代码能上线是两个维度的事情第二类是后端工程师想往 AI 方向靠但对模型训练、评估、部署的完整链路缺乏概念第三类是独立开发者手里有几个模型想法需要低成本地把它们变成可用、可维护的服务。什么叫“从零开始”不是说你得从实现反向传播开始写神经网络而是说每一步都不跳过关键细节。数据如何版本化、实验如何追踪、模型如何打包、服务如何监控这些问题都手把手拆开揉碎理解背后的原理而不是装个平台就万事大吉。这篇文章就是我整条路径的梳理和实操记录希望能给你省一些摸索的时间。2. 先搞清楚边界AI工程、数据科学、MLOps到底差在哪2.1 三者不是同义词分工和视角完全不同很多人把这三个词混为一谈但真正做起来会发现它们是三套思维模式。数据科学的重点是“发现”。核心任务是探索数据、构建特征、尝试模型回答的是“这件事能不能做”、“哪个算法效果更好”这类问题。它的产出物往往是分析报告、实验笔记、模型原型。AI工程的重点是“交付”。它关心的是把模型变成一项可靠的服务让外部系统能够稳定调用并且可以被监控、被迭代、被回滚。核心是软件工程的方法论需求分析、架构设计、测试、部署、运维。MLOps更偏向流程与基础设施层面是 A工工程在运维侧的延伸关注模型从训练到上线的自动化流水线、环境一致性、版本管理、CI/CD。换句话说MLOps是AI工程的一部分但AI工程不仅限于流水线还包括接口设计、数据质量保障、业务逻辑融合等更广泛的内容。我用一句生活化的比喻帮你记忆这三个角色。数据科学家像是一位研究菜谱的美食家他告诉你某种食材搭配某种酱汁会很好吃AI工程师像是把这道菜标准化、流程化的中央厨房负责人他要保证无论谁来掌勺出品都是同一味道MLOps工程师更像是厨房的设备维护与供应链管理系统——该有的烤箱、冷藏柜、食材追踪都得齐全出问题了自动报警。2.2 为什么“从零开始”这条路更值得走直接上 Kubeflow、SageMaker 这类机器学习平台看似省事但我强烈建议你至少亲手走一遍完整的 AI 工程链路哪怕最后再回到这些平台上去。原因很简单平台屏蔽了太多“魔鬼细节”。比如数据版本化为什么需要做因为训练集发生变化后你无法追溯模型性能变化的原因。平台帮你做了这件事你就不会理解为什么某一天线上指标突然崩了——可能是数据变了可能是特征逻辑变了也可能是模型本身过期了。这些因果链条只有你亲手构建过一遍才能在故障来临时迅速定位。另外从零开始意味着你对系统里的每一个组件都有掌控权。平台一旦出现故障、或者预算受限你还能退回到自建的轻量方案。我见过不少团队把整套系统建立在某个云平台之上结果平台一改版整个项目直接瘫痪。这种系统性风险用“自建 少量平台工具”的方式可以有效对冲。3. 第一阶段把地基打牢AI工程首先是软件工程3.1 项目结构规范从“一个 notebook 打天下”改为 src 布局很多人学 AI 的时候习惯开个 Jupyter Notebook数据读进来、模型训练、画图全在一个文件里。这在探索阶段没问题但一旦进入工程化阶段就会变成灾难——变量状态难以追踪、代码不可复用、测试无从下手。我建议第一步就把项目改成标准的 src 布局news-ai-service/ ├── pyproject.toml ├── src/ │ └── news_classifier/ │ ├── __init__.py │ ├── data/ │ │ ├── loader.py │ │ └── preprocess.py │ ├── models/ │ │ ├── train.py │ │ └── predict.py │ ├── serving/ │ │ └── api.py │ └── utils/ │ └── logger.py ├── tests/ │ ├── test_preprocess.py │ └── test_api.py ├── data/ │ ├── raw/ │ ├── processed/ │ └── versions/ ├── experiments/ └── models/ └── artifacts/为什么这样设计src 布局保证所有包都通过统一的导入路径使用不会出现“脚本里靠临时改 sys.path 才能 import”的野路子。data 目录区分 raw、processed、versions 三层原始数据不污染、处理结果可复用、每个版本有痕迹。experiments 用来存放实验记录models/artifacts 用来放最终落盘的模型产物。3.2 包管理与环境锁定别再裸装依赖了Python 项目依赖管理一直是个老大难问题。我见过无数人直接把 pip 安装到全局环境结果一台机器上各种库版本冲突换台机器跑不起来。我推荐使用 venv 或者更省心的 uv把依赖声明在 pyproject.toml 里。以下是我这个项目用到的核心依赖约束[project] name news-classifier version 0.1.0 requires-python 3.10 dependencies [ fastapi0.110.0, uvicorn[standard]0.29.0, pydantic2.6.0, scikit-learn1.4.0, joblib1.3.2, pandas2.2.0, ] [project.optional-dependencies] dev [ pytest8.0.0, ruff0.3.0, pre-commit3.6.0, ]很多人会问为什么不用 requirements.txt当然也可以用但它有个问题它会将传递依赖的版本一并冻结升级时容易产生一堆无关变更。pyproject.toml 直接声明顶层依赖配合uv.lock文件能实现精确的版本锁定——既能保证可重复安装又不至于锁死所有传递依赖。另外提醒一句装依赖的时候不要图快用pip install --force-reinstall去覆盖已有包很容易把系统里其他项目的依赖弄坏。项目隔离环境必须从一开始就养成习惯这算是踩过无数次坑之后的肺腑之言。3.3 日志、配置与测试是 AI 程序的“安全气囊”模型代码的核心逻辑往往只有几十行真正体现工程能力的恰恰是辅助设施。日志这一块我推荐使用 Python 自带的logging模块 JSON 格式化输出。为什么不用 print因为 print 没有级别、没有时间戳、无法按模块过滤。当系统里有十几个模块同时在跑的时候没有格式化的日志根本没法看。下面是我常用的配置import json import logging from datetime import datetime, timezone class JsonFormatter(logging.Formatter): def format(self, record): log_entry { timestamp: datetime.now(timezone.utc).isoformat(), level: record.levelname, logger: record.name, message: record.getMessage(), } if record.exc_info: log_entry[exc_info] self.formatException(record.exc_info) return json.dumps(log_entry, ensure_asciiFalse) logger logging.getLogger(app) handler logging.StreamHandler() handler.setFormatter(JsonFormatter()) logger.addHandler(handler) logger.setLevel(logging.INFO)注意时间戳统一用 UTC ISO 8601 格式。线上服务一旦有多个实例日志时间不统一会让人排查到怀疑人生。别问我是怎么知道的。配置管理建议用 pydantic-settings 或者简单的环境变量方案不要把数据库地址、模型路径这些硬编码在代码里。配置和代码分离是 12-factor 的基本要求AI 项目同样适用。测试这一块很多算法工程师会觉得写测试太浪费时间。我的经验是不需要覆盖模型本身但一定要覆盖数据处理函数、特征转换、接口校验这些逻辑。这些环节的 bug 往往比模型本身的 bug 更致命。比如文本预处理函数里一个正则没写对会导致全量线上数据被错误清洗模型再厉害也白搭。4. 第二阶段模型开发的工程化改造从“能跑”到“可复现”4.1 数据版本化没有历史就没有追责和回滚做 AI 工程最容易被忽略的就是数据版本化。我一开始也觉得“数据放在那不乱动不就行了吗”但问题在于你总会接到“再加一点数据训练看看效果”的需求总要对特征逻辑做微调。几次改动下来连你自己都说不清当前模型是用哪版数据训练的。我采用的是一套轻量方案不需要额外服务目录 元信息一个不少data/versions/ ├── 20240101_news_v1/ │ ├── train.csv │ ├── test.csv │ └── metadata.json ├── 20240115_news_v2/ │ ├── train.csv │ ├── test.csv │ └── metadata.json每个数据版本目录里的 metadata.json 记录了三样东西数据的来源与采集时间、特征处理的代码版本用 git commit hash 标识、数据行的统计信息行数、类别分布。这样任何一次训练都能明确追溯到数据源头。如果发现新版本数据导致模型变差直接回滚到旧版本重新训练即可几分钟的事情。4.2 实验跟踪用最笨的方法做到最清晰的记录MLflow 这类实验跟踪工具很好但对于个人项目来说略重。我建议先建立一套“手动实验记录”的规范每次训练时把参数、评估指标、模型产物路径统一记录到一个 experiments/ 目录下文件名带时间戳。experiments/ ├── 20240101_1430_lr0.001_embedding128/ │ ├── params.json │ ├── metrics.json │ ├── confusion_matrix.png │ └── notes.mdparams.json 里记录模型超参数、数据版本、预处理配置metrics.json 里记录准确率、精确率、召回率、F1 等指标notes.md 是给自己看的写下这个实验和上一个的差异点、发现的现象、下一步打算。这样做的价值在于两个月后你回来看自己的实验记录能快速知道“哦当时我把学习率从 0.01 调到 0.001 后 F1 提升了 2 个百分点原因是原来模型在训练后期震荡”。这种分析能力是 AI 工程师和数据科学家最核心的差异化竞争力之一。4.3 训练脚本模块化一次运行完整闭环训练脚本我强烈不建议用 Notebook 直接转 .py而是应该按职责拆分为多个函数或类让一次训练跑出一个完整的 pipeline。我以文本分类为例展示一下核心的训练骨架from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.linear_model import LogisticRegression from sklearn.pipeline import Pipeline from sklearn.metrics import classification_report import joblib # 数据加载与划分示意 X_train, X_test, y_train, y_test load_data(data/versions/20240101_news_v1) pipeline Pipeline([ (tfidf, TfidfVectorizer(max_features5000, ngram_range(1, 2))), (clf, LogisticRegression(C1.0, max_iter1000)), ]) pipeline.fit(X_train, y_train) # 评估并输出报告 y_pred pipeline.predict(X_test) report classification_report(y_test, y_pred, target_names[tech, sports, finance]) print(report) # 模型落盘 joblib.dump(pipeline, models/artifacts/model_20240101_v1.joblib)这段代码看似简单但它体现了几个工程要点。一是把预处理放到 Pipeline 里保证训练和推理时对文本做的是完全一致的转换二是使用统一的模型落盘接口方便后续加载三是评估直接打印分类报告而不是只关注一个 accuracy 数字。很多新手只看准确率结果模型对多数类表现好到爆对少数类完全失效上线后才发现线上请求大量集中在少数类直接崩盘。4.4 模型保存与加载这是最容易被“看似简单”坑到的环节模型保存无非就是 joblib.dump 和 joblib.load但这中间的坑细数起来能写一整章。我遇到过的典型问题训练环境是 Python 3.10 scikit-learn 1.4线上环境却是 Python 3.8 scikit-learn 1.0结果模型加载直接报 ValueError。原因是 pickle/joblib 对模型类的内部表示有版本依赖从一个大版本迁移到另一个或者反序列化时找不到对应类定义非常容易报错。解决思路有两个。第一尽量保持训练环境与推理环境一致把模型运行时用的依赖版本写在 requirements.txt 中且严格锁定第二如果确实存在环境差异比如用户侧就是不能升级 Python 版本那建议训练环境也用目标环境的版本专门导出一份模型。永远不要假设 pickle 的兼容性——它从来都不是一个正式的跨版本协议。另一个值得注意的坑是模型文件里包含了训练用的一些辅助对象但加载环境里没有这个类。比如我曾在模型里塞了一个自定义的文本清洗类推理侧忘记把这个类的定义拷贝过去一加载就报 “ModuleNotFoundError”。现在我的做法是所有进入模型的转换器都必须放在独立模块里并在文档中明确指出加载依赖。5. 第三阶段把模型变成一个真正能用的服务5.1 框架选择与接口设计FastAPI 是最省心的方案API 框架我一开始在 Flask 和 FastAPI 之间犹豫过。Flask 生态成熟、上手简单但 FastAPI 有几个天然优势很适合 AI 服务基于 Pydantic 的请求校验、原生自动生成 OpenAPI 文档、异步支持、性能接近 Node.js。对于一个以 JSON 数据交互为主的模型服务来说FastAPI 几乎是零思考成本的选择。接口设计上我强烈建议至少做两个端点/health和/predict。/health是给负载均衡和监控用的返回 200 表示服务存活可以顺带返回模型版本号。/predict接收业务方传来的明文数据返回预测结果。默认不提供/batch_predict之类的批量接口因为批量接口一旦设计不当很容易被当成逗着玩的循环调用接口导致后端阻塞。下面是我这个新闻分类服务最小可用的 API 实现from fastapi import FastAPI from pydantic import BaseModel, Field import joblib import numpy as np app FastAPI(titlenews-classifier-service) # 全局加载模型避免每次请求都 load with open(models/artifacts/model_20240101_v1.joblib, rb) as f: model joblib.load(f) class PredictRequest(BaseModel): text: str Field(..., min_length1, max_length2000, description新闻文本) top_k: int Field(1, ge1, le5, description返回前k个类别) class PredictResponse(BaseModel): labels: list[str] probabilities: list[float] app.get(/health) def health(): return {status: ok, model_version: 20240101_v1} app.post(/predict, response_modelPredictResponse) def predict(req: PredictRequest): probs model.predict_proba([req.text])[0] top_indices np.argsort(probs)[::-1][: req.top_k] labels [model.classes_[i] for i in top_indices] prob_values [round(float(probs[i]), 4) for i in top_indices] return PredictResponse(labelslabels, probabilitiesprob_values)注意几个细节。模型在模块导入时加载一次保存在全局变量里如果是多 worker 部署每个进程各加载一份各自持有内存副本。Pydantic 的 Field 限制文本长度和 top_k 上限既防止恶意超长输入也避免一次返回几百个类别造成流量浪费。5.2 容器化部署Dockerfile 里的几个硬核细节很多人写 Dockerfile 就是 pip install uvicorn跑起来没问题但在生产环境会有几个隐患镜像太大、构建依赖残留、以 root 身份运行。推荐用多阶段构建先装依赖再精简运行时。下面是一个示例FROM python:3.10-slim AS builder WORKDIR /app COPY pyproject.toml . RUN pip install --user --no-cache-dir . FROM python:3.10-slim WORKDIR /app COPY --frombuilder /root/.local /root/.local COPY src ./src COPY models ./models ENV PATH/root/.local/bin:$PATH RUN useradd -m appuser chown -R appuser:appuser /app USER appuser EXPOSE 8080 CMD [uvicorn, news_classifier.serving.api:app, --host, 0.0.0.0, --port, 8080]第一个 FROM 负责构建第二个 FROM 只保留运行需要的文件和依赖镜像体积从 1.2GB 直接降到 400MB 左右。非 root 用户运行降低安全风险。这里还有一个常被忽略的点模型文件必须在构建镜像时拷贝进去而不是运行时从宿主机挂载。镜像内部模型固化才能保证所有实例的一致性。5.3 性能调优从单进程到多 worker再到批处理单进程 FastAPI 一般能够支撑每秒几十次到上百次请求但实际线上流量一上来就撑不住。这时候最通用的方案是 gunicorn 多 workergunicorn -k uvicorn.workers.UvicornWorker \ -w 4 \ -b 0.0.0.0:8080 \ --timeout 120 \ news_classifier.serving.api:app-w 4是 worker 数量经验值是 CPU 核心数加 1 或者就是核心数。文本分类这种轻量模型每个 worker 占几百 MB 内存4 个 worker 大致 2GB 左右比较常见。如果是 GPU 模型切记不要设置过多 workerGPU 显存是共享的多个进程同时往上挤很容易 OOM。如果单请求延迟还是太高可以考虑请求级缓存。比如同一段文本在一段时间内被反复查询例如热门新闻用functools.lru_cache做个简单缓存效果立竿见影from functools import lru_cache lru_cache(maxsize1024) def predict_text(text: str): return model.predict_proba([text])[0]再有更进阶的方案是请求批处理把并发的多个请求攒成一个 batch 送入模型推理这在 GPU 推理场景尤其有效。但要注意批处理会引入额外延迟和复杂度对新手我并不推荐一上来就做。6. 第四阶段上线后的监控、漂移检测与模型迭代6.1 请求日志与指标不记录就无从谈起优化服务跑起来只是开始真正要命的是“跑着跑着突然不准了”。要发现这个问题前提是有完善的日志和指标。我要求每个/predict请求都记录以下信息输入文本长度、预测类别、最高置信度、处理耗时。这些信息以 JSON 日志形式输出交给日志采集系统统一处理。在日志之外至少要有两类指标QPS每秒请求数和 P99 延迟。P99 表示最慢的 1% 请求所消耗的时间比平均延迟更能暴露性能瓶颈。当多个 worker 并行时个别 worker 出现 GC 停顿或者 CPU 争抢平均延迟可能看不出什么P99 会敏锐地拉高这是一个需要关注的信号。6.2 数据漂移与概念漂移模型失效的两大元凶模型上线后效果下降原因不外乎两类。数据漂移指输入数据的统计分布发生了变化。比如模型是在简体中文新闻上训练的某一天线上开始涌入大量繁体中文内容这就会导致分布漂移预测准确率骤降。概念漂移指输入到输出的关系发生了变化比如“金融”这个词在某个时期大量出现在娱乐新闻中模型学到的映射关系不再成立。监测数据漂移不需要太复杂的工具。对文本分类来说可以用一个非常朴素的方法每隔一段时间抽取当日的请求文本统计其词频分布与训练集上的词频分布做对比计算 PSI群体稳定性指数。PSI 的计算思路不复杂把两个分布分别分箱再对每个箱子计算占比差异和占比比值的对数乘积。当 PSI 超过 0.1 时需要警惕超过 0.25 时基本可以确定分布发生了显著变化。我写了一版简化实现def calculate_psi(expected: np.ndarray, actual: np.ndarray, bins10) - float: expected np.clip(expected, 1e-6, None) actual np.clip(actual, 1e-6, None) # 基于 expected 的分位数划分箱子 percentiles np.percentile(expected, np.linspace(0, 100, bins 1)) expected_counts, _ np.histogram(expected, binspercentiles) actual_counts, _ np.histogram(actual, binspercentiles) expected_ratio expected_counts / expected_counts.sum() actual_ratio actual_counts / actual_counts.sum() psi np.sum((expected_ratio - actual_ratio) * np.log(expected_ratio / actual_ratio)) return psi一旦监控到漂移紧接着的动作就是触发重训流程。我建议建立“周级重训 月度评估”的节奏每周拿新增的已标注数据重训模型每个月在固定的测试集上做一次对比防止模型被新数据带偏。6.3 模型回滚与灰度发布不要一把梭模型更新到线上服务最忌讳的方式是直接把旧模型文件替换成新模型文件然后重启。一旦新模型效果不佳你就得再回滚而回滚本身又是一次发布操作中间还有间隔时间线上一直处于不稳定状态。我的做法是每个模型版本独立编号存放在独立的路径下服务启动时通过环境变量或配置指定加载哪个版本。新模型先在灰度环境上运行一段时间用影子模式同时对比新旧模型的预测结果确认新模型没有明显回退后再切全部流量。这里有一个简单而有效的做法接口返回预测结果时附带model_version字段方便统计切换前后各版本的调用量和置信度分布。7. 常见问题与排查技巧实录为了让你少走弯路我把自己做这个项目时踩过和见过的坑整理成了一张速查表问题现象排查方向解决方案模型加载失败启动时报 ModuleNotFoundError模型依赖了自定义类把转换器类放在独立模块确保推理环境可导入并发后内存爆掉服务运行几分钟后内存持续增长worker 内缓存了过多数据控制 lru_cache 大小限制最大并发数推理结果与训练时不一致线下 F1 0.85线上统计只有 0.80预处理逻辑不一致务必使用 Pipeline 统一预处理不要内外两套代码延迟突然升高P99 从 50ms 涨到 500ms可能是日志阻塞、锁竞争检查 worker 数量异步日志输出必要时加缓存请求文本编码报错UnicodeDecodeError上游传了非法编码字符串在 API 入口做 decode 兜底拒绝非法字符数据漂移被忽略准确率每周下降 1%缺少分布监测接入 PSI 检测设置告警阈值这里面我觉得最值得拎出来说的是“推理结果与训练时不一致”那条。很多人会首先怀疑“是不是模型文件坏了”其实绝大多数情况是训练和推理时的文本预处理逻辑没有对齐。比如训练时文本做了去停用词、小写化但推理脚本里忘了写。这类 bug 极其隐蔽却会直接拉低线上效果。排查的时候有个技巧在沙箱环境加载同一份模型对一模一样的输入做训练态推理和线上推理比对结果是否一致。不一致就把流程拆开从分词、转小写、正则清洗到特征提取逐段 debug很快就能定位。8. 我个人做这个项目的一点体会走到这一步整个从零开始的 AI 工程链路就基本成型了数据版本化、实验追踪、训练模块化、模型打包、API 服务、容器部署、监控告警、模型回滚。我不敢说这套体系有多高级但它每一个组件都经受过真实项目的检验能跑能扛。我最想分享的体会是AI 工程的本质是确定性。模型本身是概率性的输出天然带有不确定性但工程系统必须把这种不确定性控制在可管理、可观测的范围内。每多一层测试、多一条日志、多一次版本记录都是在为“确定性”添砖加瓦。如果只给一个起步建议我会说先别急着引入任何编排框架把“一个脚本完成从数据到 API 的完整流程”走通一遍感受一下每一步到底发生了什么。等你有痛感了再决定要引入什么样的工具。无数人一上来就搭复杂架构结果大部分精力消耗在工具本身而不是业务问题上反而背离了 AI 工程的初衷。