如果你在 GitHub 上搜过ai-engineering-from-scratch这类仓库名或者对照着招聘网站上的“AI 工程师”JD 一条条看下来大概率会陷入一个困惑网上 90% 的教程都在讲怎么训练模型但几乎没人系统的讲清楚——模型训练完之后怎么变成一个稳定运行的系统。数据从哪来、特征怎么做、实验怎么管理、模型怎么部署、上线之后怎么监控、多久要重新训练一次这些才是 AI 工程里真正吃时间、也真正体现水平的部分。这篇文章就是我对“从零开始做 AI 工程”这个主题做的一次完整复盘。不绕弯子直接讲清楚AI 工程到底是什么、需要哪些技能栈、怎么用一个端到端的小项目把整个链路跑通以及我踩过的那些坑。如果你是刚入行的算法工程师、想转 AI 工程方向的软件工程师或者正在给自己列学习计划的同学这篇文章应该能帮你省下不少试错时间。1. 先从根上理解AI 工程和算法岗根本不是一回事很多人以为 AI 工程就是把模型训练好再调个接口其实这个理解会害死人。1.1 模型训练只是 AI 工程的三分之一我习惯把 AI 工程拆成三个阶段数据与特征阶段、模型实验阶段、部署与监控阶段。模型训练只属于第二阶段里很小的一段。真实业务里数据阶段的脏活累活往往占总工作量的 60% 以上。模型选型和训练反而是流程里最“爽”的部分因为你能看到 loss 在下降、指标在上涨正反馈非常明显。但上线之后才是真正的考验。传统软件上线后只要代码不出 bug行为就是确定的模型上线后面对的是实时变化的用户行为和数据分布今天效果好的模型下个月可能就悄悄变差了。这种不确定性是 AI 工程和传统软件工程最本质的区别。另一个区别在测试环节。传统后端开发可以写几百个单测用例断言输入输出模型的“正确性”没法用硬断言只能靠评估指标和线上监控来判断。也就是说你不仅要测试代码还要测试“数据和策略”。很多从软件工程转过来的朋友第一反应是给模型接口写断言写完之后发现真正需要盯的是线上预测分布有没有漂移、业务指标有没有波动这些恰恰是传统测试思维覆盖不到的。1.2 为什么说“from scratch”值得重新定义from scratch这个词容易被理解成“从零手写神经网络”。但作为工程实践我们真正需要的是从零走通全链路而不是重复造轮子。手写反向传播是学术领域的练习工程领域更重要的是理解每一步在做什么、为什么这样做、能解决什么问题。我见过太多人把大量时间耗在啃论文和复现模型上结果到部署时连Dockerfile都没写过上线后连日志和监控都没有。真正合理的“从零开始”是先搭一个最小的完整闭环——哪怕模型简单到就是个逻辑回归只要数据、训练、部署、监控四个环节全部跑通你对 AI 工程的认识就会发生质变。后面再往这个闭环里添更复杂的模型、更完善的监控体系就只是时间问题了。2. 动手前的技能栈梳理与工具选型AI 工程需要的能力面很杂数据处理、模型训练、软件工程、容器化、服务编排、监控告警。如果每样都深学半年都出不了成果。我的建议是先用最朴素的工具跑通再逐步加深。2.1 地基Python 数据处理必须熟练这里说的熟练不是“会用 pandas 读 CSV”而是能顺手完成数据清洗、聚合、变换、可视化分析不卡壳。AI 工程里相当多时间在和数据打交道如果每次操作都要查 API 文档效率会非常低。环境管理我也建议从一开始就养成习惯用虚拟环境隔离项目依赖。我习惯用venv因为它是 Python 自带的不需要额外安装。conda在 Windows 上处理底层库依赖时更方便但体积大、启动慢平时自己开发用 venv 就够了。python -m venv .venv source .venv/bin/activate pip install --upgrade pip2.2 模型框架为什么推荐 PyTorch模型训练这块PyTorch 是我现在的主力选择。原因很简单它的调试体验是断点式的能像普通 Python 代码一样打印中间变量、逐步执行这对新手和理解复杂模型非常有帮助。TensorFlow 的 Keras 高层接口上手也快但一旦需要自定义训练逻辑或者调试深层模型PyTorch 的舒适度要明显高一个档次。另外 PyTorch 在学术界和工业界的生态优势明显HuggingFace 的 transformers、各类 diffusers 模型都以 PyTorch 为主学一套能通吃大部分场景。如果只是做表格数据或传统机器学习scikit-learn 和 XGBoost 依然是不错的选择。我的原则是传统结构化数据用 sklearn/XGBoost文本图像这类非结构化数据用 PyTorch各干各的活。2.3 工程化四件套MLflow、FastAPI、Docker、Prometheus模型训练只是中间产物整个工程需要四个工具链各司其职环节工具解决的问题实验追踪MLflow记录超参数、指标、模型产物解决“哪个实验跑出这个模型”的追溯问题模型服务FastAPI把模型包装成 HTTP 接口让业务方通过 REST API 调用容器化Docker统一开发、测试、生产环境解决“在我机器上是好的”这个经典问题监控告警Prometheus Grafana收集预测分布、延迟、流量等指标检测模型退化这四样是 AI 工程落地的最低配置。我的经验是第一次做全链路时不要贪多先把 MLflow 当实验记录本、用 FastAPI 起服务、用 Docker 打包、用 Prometheus 拉取指标后面再慢慢理解每部分的深层设计。2.4 目录结构新手最容易忽略的事工程项目的目录结构决定了你后续维护的体验。我推荐一个符合当前 AI 工程社区主流实践的最小结构. ├── data/ # 数据存放 │ ├── raw/ # 原始数据 │ └── processed/ # 清洗后数据 ├── src/ # 源码包 │ ├── data/ # 数据加载与预处理 │ ├── features/ # 特征工程 │ ├── models/ # 模型定义与训练 │ ├── api/ # FastAPI 服务 │ └── monitor/ # 监控与告警 ├── configs/ # 配置文件 ├── experiments/ # 实验记录 ├── models/ # 训练产物 ├── tests/ # 单元测试 ├── Dockerfile ├── requirements.txt └── README.md把数据、代码、模型、配置分开保证任何一个人接手你的项目十分钟内就能找对地方。这个结构的价值在项目初期看不出来等三个月后你自己回头改代码体会就深了——如果所有文件都堆在根目录光找文件就能耗掉你半天耐心。3. 实操从零做一个电商评论情感分析系统概念说再多都不如动手做一遍。下面我用“电商评论情感分析”这个经典场景带你走一遍完整链路。选这个场景是因为数据好找、业务价值直观、模型不用很大就能跑出效果非常适合做 AI 工程入门。3.1 数据准备与预处理第一步是拿数据。我用的是一个公开的中文电商评论数据集有标签、好评和差评。原始数据长这样label,text 1,物流很快东西质量也很好 0,质量太差了客服态度也不好 ...拿到数据后不要急着训练。先做基础清洗去重、去空白、处理缺失值、统一文本格式。import pandas as pd import re df pd.read_csv(data/raw/reviews.csv) df df.drop_duplicates(subset[text]) df df.dropna(subset[text]) def clean_text(text: str) - str: text re.sub(r[^], , text) # 去 HTML 标签 text re.sub(r\s, , text) # 去空白 text text.strip() return text df[text] df[text].apply(clean_text) df df[df[text].str.len() 2] # 去掉太短的样本 df.to_csv(data/processed/reviews_clean.csv, indexFalse)这里有个细节值得多说一句清洗规则必须统一。训练时怎么清洗线上推理时也必须用同一套逻辑否则模型看到的输入风格不一样效果会大打折扣。这个坑我后面会再提一次因为它真的是部署事故重灾区。3.2 构建词表与数据集中文文本不像英文天然按空格分词所以要做分词。我直接用 jieba简单稳定import jieba from collections import Counter def tokenize(text: str): return [w for w in jieba.lcut(text) if w.strip()] # 构建词表保留出现频率最高的 vocab_size 个词 vocab_size 20000 counter Counter() for text in df[text].tolist(): counter.update(tokenize(text)) vocab {pad: 0, unk: 1, cls: 2} for word, freq in counter.most_common(vocab_size - 3): vocab[word] len(vocab) # 保存词表线上推理时要原样加载 import pickle with open(models/vocab.pkl, wb) as f: pickle.dump(vocab, f)词表相当于模型和文本之间的“翻译词典”训练时构建好上线时加载同一个词表。如果训练和推理用了不同的词表模型的输入就全乱了预测结果没有任何意义。训练数据用 PyTorch 的Dataset和DataLoader封装import torch from torch.utils.data import Dataset class ReviewDataset(Dataset): def __init__(self, df, vocab, max_len128): self.labels df[label].astype(int).tolist() self.texts df[text].tolist() self.vocab vocab self.max_len max_len def __len__(self): return len(self.labels) def __getitem__(self, idx): tokens tokenize(self.texts[idx]) ids [self.vocab.get(t, 1) for t in tokens[:self.max_len]] pad_len self.max_len - len(ids) ids [0] * pad_len return torch.tensor(ids), torch.tensor(self.labels[idx]) train_df ... # 80% val_df ... # 20%注意要从整体里随机划分 train_dataset ReviewDataset(train_df, vocab) val_dataset ReviewDataset(val_df, vocab) train_loader DataLoader(train_dataset, batch_size64, shuffleTrue) val_loader DataLoader(val_dataset, batch_size64, shuffleFalse)如果数据本身带时间属性比如按天采样的交易数据或日志数据划分时要格外小心。评论场景相对简单随机划分即可一旦涉及时序数据就必须严格按时间切分否则未来的信息泄漏到训练集里模型在离线评估时表现很好上线后立刻现原形。3.3 模型选择从 TextCNN 开始模型我选了 TextCNN它在文本分类上结构简单、训练快但作为工程示范已经足够。它的思路很直观用几个不同尺寸的卷积核在文本序列上滑动类似于捕捉不同长度的局部 n-gram 模式再接池化和全连接层输出分类。import torch.nn as nn class TextCNN(nn.Module): def __init__(self, vocab_size, embed_dim100, num_class2, kernel_sizes[2, 3, 4], num_filters128): super().__init__() self.embedding nn.Embedding(vocab_size, embed_dim, padding_idx0) self.convs nn.ModuleList([ nn.Conv1d(embed_dim, num_filters, k, paddingk // 2) for k in kernel_sizes ]) self.fc nn.Linear(num_filters * len(kernel_sizes), num_class) def forward(self, x): emb self.embedding(x) # (B, L, D) emb emb.transpose(1, 2) # (B, D, L) convs [torch.relu(conv(emb)) for conv in self.convs] pools [torch.max_pool1d(c, c.size(2)).squeeze(2) for c in convs] cat torch.cat(pools, dim1) return self.fc(cat)超参数我一开始是这样设的embed_dim100、kernel_sizes[2,3,4]、num_filters128、batch_size64、lr1e-3、epochs10、max_len128。这些数值不是拍脑袋定的是结合经验的选择词向量维度太低表达力不足太高在小数据集上容易过拟合卷积核覆盖 2 到 4 个词的跨度基本能抓住中文评论里常见的短语模式batch size 64 在显存和梯度稳定性之间取了一个平衡点。3.4 训练与实验追踪训练循环本身不复杂但工程上最关键的是把每次实验记录下来。这时候 MLflow 派上用场了import mlflow with mlflow.start_run(): mlflow.log_param(model_type, TextCNN) mlflow.log_param(embed_dim, 100) mlflow.log_param(kernel_sizes, [2, 3, 4]) mlflow.log_param(batch_size, 64) mlflow.log_param(lr, 1e-3) mlflow.log_param(epochs, 10) criterion nn.CrossEntropyLoss() optimizer torch.optim.Adam(model.parameters(), lr1e-3) for epoch in range(10): # 训练逻辑... train_loss, train_acc train_one_epoch(...) val_loss, val_acc evaluate(val_loader) mlflow.log_metric(train_loss, train_loss, stepepoch) mlflow.log_metric(val_acc, val_acc, stepepoch) mlflow.pytorch.log_model(model, textcnn_model)我强烈建议从一开始就养成记录实验的习惯。原因很现实你会同时跑好几个超参组合过两天再看结果如果完全没有记录你根本不知道哪个配置产生了哪个模型也无法复现。MLflow 相当于实验的“账本”把输入、参数、输出都记得清清楚楚。训练结束后保存模型产物连同词表等元数据torch.save({ model_state_dict: model.state_dict(), vocab_size: len(vocab), embed_dim: 100, kernel_sizes: [2, 3, 4], num_filters: 128, num_class: 2, }, models/textcnn_model.pt) with open(models/vocab.pkl, wb) as f: pickle.dump(vocab, f)3.5 评估别只盯着准确率准确率是入门指标但在类别不平衡时极具欺骗性。比如 95% 好评、5% 差评的数据集模型全预测好评准确率也有 95%但它实际什么都没学会。因此至少要同时看精确率、召回率和 F1 值。对于情感分析我比较看重的是差评类别label0的召回率。原因很直观业务上最不希望看到的是把差评误判成好评那会让真正的问题客户被系统忽略。如果差评召回率偏低通常的应对方式有调整类别权重、用 Focal Loss、过采样少数类或者直接在阈值上做文章——把预测差评的阈值从默认的 0.5 降到 0.3。评估之后确认效果达标就可以进入部署环节了。3.6 FastAPI 部署把模型变成接口我用 FastAPI 写一个最小可用的推理服务。需要重点处理两件事模型加载要缓存、预处理逻辑要复用训练时的同一套代码。import pickle import torch import jieba from fastapi import FastAPI from pydantic import BaseModel app FastAPI() # 启动时加载模型和词表 model None vocab None class PredictRequest(BaseModel): text: str def load_model(): global model, vocab with open(models/vocab.pkl, rb) as f: vocab pickle.load(f) checkpoint torch.load(models/textcnn_model.pt, map_locationcpu) ... model.load_state_dict(checkpoint[model_state_dict]) model.eval() def predict(text: str) - int: tokens [w for w in jieba.lcut(text) if w.strip()] ids [vocab.get(t, 1) for t in tokens[:128]] ids [0] * (128 - len(ids)) input_tensor torch.tensor([ids]) with torch.no_grad(): logits model(input_tensor) pred torch.argmax(logits, dim1).item() return pred app.on_event(startup) def startup(): load_model() app.post(/predict) def predict_api(req: PredictRequest): pred predict(req.text) return {label: pred, sentiment: positive if pred 1 else negative}load_model放在启动事件里做是为了避免每个请求都重新加载模型权重。如果模型加载放在请求内部并发一上来GPU/CPU 内存和响应时间都会炸。模型本身就是冷启动成本最高的部分加载一次、常驻内存是 AI 服务部署的常识。推理的map_locationcpu也很关键。训练时权重可能在 GPU 上但线上服务通常是 CPU 部署显式指定加载目标能避免跑服务时因为没有 GPU 而报错的尴尬。本地起服务测试uvicorn src.api.main:app --host 0.0.0.0 --port 8000 curl -X POST http://localhost:8000/predict \ -H Content-Type: application/json \ -d {text: 这个商品性价比很高客服态度也好}测试返回正常接口就通了。3.7 Docker 打包消除环境差异接口本地跑通了还不够要让它能在任意环境里一键跑起来就要容器化。FROM python:3.10-slim as builder WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY src ./src COPY models ./models EXPOSE 8000 CMD [uvicorn, src.api.main:app, --host, 0.0.0.0, --port, 8000]这里有个实用技巧用python:3.10-slim而不是python:3.10能显著减小镜像体积。依赖较大的场景更推荐多阶段构建先用一个完整镜像安装依赖并编译最后只把运行所需内容复制到干净的 slim 镜像里可以把镜像从 2GB 降到 400MB 左右。构建并运行docker build -t sentiment-api . docker run -d -p 8000:8000 sentiment-api到这里一个模型服务就被打包成了标准化的容器在任何装有 Docker 的机器上都能跑。3.8 监控与迭代AI 工程最关键的一跃很多项目止步于“模型接口能调用”但这只是完成了软件工程部分。AI 工程的核心挑战在监控和迭代。上线后至少要监控三类信息服务健康度延迟、流量、错误率、预测分布好评差评的占比变化、业务反馈模型预测的准确率在业务侧的实际表现。服务健康度用 Prometheus 配合 FastAPI 的 prometheus-fastapi-instrumentator 暴露 metrics再接入 Grafana 画看板是当前最主流的组合。预测分布这块我通常每个小时统计一次预测结果的类别比例如果差评占比突然从正常的 20% 飙到 60%很可能不是用户集体脾气变差而是数据分布发生了漂移需要人工介入检查。监控只是发现问题的手段发现问题之后要有触发机制。我建议设置多级响应预测分布漂移超过阈值触发告警告警确认后进入人工检查流程如果业务效果显著下降就启动一次重新训练。至于是定时重训还是漂移触发重训取决于业务数据的时效性。电商评论这种变化较慢的场景定时周更就够了如果是用户行为实时建模就需要按天甚至按小时更新。4. 整个链路最常见的坑与排查实录这部分是我最想分享的因为很多坑不是从文档里能学到的都是真金白银踩出来的。4.1 环境和依赖问题问题代码在本机能跑进 Docker 就报错。最常见原因是本机依赖版本和组织进程里requirements.txt不一致。解决办法是锁定精确版本而不是用的宽松写法。我踩过numpy 1.x和numpy 2.x的兼容性坑版本一旦变动底层接口行为都可能不一样。问题PyTorch 安装后 CUDA 不可用。先在命令行里跑python -c import torch; print(torch.cuda.is_available())确认。如果返回 False通常是安装的是 CPU 版 PyTorch需要去官网按 CUDA 版本重新安装。装完之后显存溢出先看有没有别的进程占用 GPU再考虑调小 batch size。4.2 数据处理问题问题训练效果好线上疯狂出错。90% 的原因出在预处理不一致。比如训练时你写了re.sub(r\s, , text)线上推理时忘了写或者用了不同版本的分词器模型看到的特征空间直接变了。解决办法是把预处理固化成同一个 python 函数训练和推理都引用它严禁复制粘贴后各改一版。问题模型指标虚高上线打回原形。通常是数据泄漏。比如用户点击率预测把曝光后的行为特征放进模型离线 AUC 高得离谱线上却毫无作用。处理思路是保持时间切分确保评估时用的训练集和验证集没有信息交叉。4.3 模型训练问题问题loss 不下降。先检查学习率是否过大导致震荡再查数据有没有标准化。我遇到过一种很“坑”的情况语料标签写反了模型努力学习了很久最后学到的规律是完全颠倒的。排查方法很简单抽几十条数据人工验证标签是否正确。问题验证集 acc 很高测试集却一塌糊涂。检查验证集是否和训练集混在一起了。比如像评论这类数据用户本身会有重复表述如果不加drop_duplicates同一个样本可能同时出现在训练集和验证集里导致验证失效。4.4 部署问题问题模型服务第一次请求特别慢。除了启动时加载模型有可能是推理时每次请求都重新加载权重、重新导入 jieba。优化方向是预热启动时跑一条假请求让模型、分词器等所有组件完成首次初始化再去接真实流量。问题接口并发一高响应就变慢。如果预测函数里有大量 CPU 计算且是同步实现FastAPI 的异步性能优势就发挥不出来。简单场景可以先用threadpool或者把模型推理放到独立进程/独立队列升级版本再考虑 GPU 推理集群。4.5 监控问题问题监控太多变成告警风暴。新上线的同学容易把所有指标都配置告警结果每天几十条告警最后大家都免疫了。建议按“业务影响优先级”配置正确率下降和接口不可用是高优先级的延迟偶尔升高属于低优先级不必每次都触发人工。问题漂移检测报警了但不知道怎么应对。漂移告警只是提示“数据分布变了”下一步应该先对比当前预测分布和训练基线分布哪些特征变化最大再判断是业务周期性波动还是真实漂移。不要一报警就去重训很多漂移是自然波动重训反而引入新的不稳定。写在最后我自己把这条链路完整走通了好几遍每走一遍都会有新的理解。第一次做的时候模型还很粗糙但真正让我成长的是那些“工程问题”——环境怎么搭、依赖怎么锁、模型怎么部署、指标怎么监控。模型结构本身反而是最容易替换的部分。最后再分享一个小技巧第一次做全链路时所有工具都选最简单的版本能把闭环跑通比什么都重要。不要一开始就上 Kubernetes、上全链路灰度先用一个最小可行产品跑起来理解每个环节的本质关系之后再逐步把基础设施做厚。AI 工程像是盖楼模型是楼里的精装修而数据、部署、监控这些工程能力才是地基。地基不稳装修再豪华一场雨就塌了。