1. 从零搭建AI工程体系为什么我劝你别一上来就调包“ai-engineering-from-scratch”这个标题第一次看到的时候我愣了一下。不是因为陌生恰恰相反是因为它戳中了我这几年带团队、做项目最痛的一个点太多人把“AI工程”等同于“会调几个API、会跑几个开源模型”结果一上生产环境就原形毕露——显存炸了、推理延迟飙到秒级、模型版本对不上、数据管道三天两头断流。所谓“from scratch”不是让你从零手写矩阵乘法而是让你从零理解一个AI系统到底由哪些零件拼起来、每个零件的边界在哪、它们之间怎么咬合。这件事的价值在你第一次遇到线上事故、第一次被要求把推理成本砍掉一半、第一次需要把一个实验室里的demo变成能扛住每天百万次调用的服务时会体现得淋漓尽致。这篇文章适合三类人一是刚转行做AI工程、只会调库但心里没底的开发者二是带团队做AI落地、需要统一技术认知的技术负责人三是想搞清楚“AI系统到底怎么跑起来”的产品和运维同学。我会把AI工程从零搭建的完整链路拆开讲清楚每一步为什么这么做、坑在哪里、怎么绕过去。全程不堆术语能上手抄的配置和代码我都会给出来。2. 先想清楚AI工程到底在工程什么2.1 一个被严重低估的认知AI工程不等于模型训练很多人对AI工程的想象是这样的搞一堆数据扔进GPU训练出一个模型然后上线。这个链条里训练只占整个工程量的两成不到。真正吃掉你80%精力的是训练之前的数据处理、训练之后的推理服务、以及贯穿始终的版本管理和监控。我拿一个真实场景举例。假设你要做一个文本分类服务识别用户反馈是“投诉”还是“咨询”。模型本身可能就是一个几百MB的BERT变体训练一天就能出结果。但你要让它稳定服务需要解决这些问题用户输入的长度分布是什么超长文本怎么截断批量推理的batch size设多少才能在延迟和吞吐之间平衡模型更新时怎么做到不中断服务线上出现bad case怎么快速回溯是哪一版模型、哪一批数据的问题这些问题的答案构成了AI工程的真正骨架。所以“from scratch”的第一课是把视角从“模型”拉到“系统”。2.2 核心模块拆解一个AI系统的最小完整形态一个能上生产的AI系统至少包含五个模块缺一不可数据管道负责数据的采集、清洗、标注、版本化。它的输出是训练和评估的输入。训练与实验管理负责模型训练、超参搜索、实验记录。关键是可复现。模型仓库与版本管理负责模型产物的存储、版本标记、元数据记录。模型不是文件是带血缘的资产。推理服务负责把模型变成API处理并发、批处理、超时、降级。监控与反馈闭环负责监控延迟、吞吐、准确率漂移并把线上数据回流到训练。这五个模块之间的关系不是流水线而是环。数据管道喂给训练训练产出模型进仓库仓库驱动推理服务推理服务的线上表现又通过监控回流到数据管道。任何一个环节断了整个系统就是瘸的。我见过太多团队训练脚本写得飞起但模型文件用model_final_v2_real.pth这种名字存着过两周自己都不知道哪个是哪个。这就是典型的“有模型、没工程”。2.3 技术选型的底层逻辑为什么我推荐从轻量级方案起步一提到AI工程很多人第一反应是上Kubernetes、上MLflow、上Feature Store。这些工具都好但如果你是一个人或者小团队从零开始我强烈建议你先用最轻量的方案把闭环跑通再逐步替换。原因很简单工具本身有认知成本和运维成本。你花两周搭了一套K8s集群结果发现你的推理服务QPS只有个位数那这套集群除了让你觉得自己很专业之外没有任何价值。我的建议起步方案是这样的模块轻量起步方案规模化替换方案数据管道脚本 本地文件 DVCAirflow / Dagster实验管理表格 Git commit hashMLflow / WB模型仓库对象存储 命名规范模型注册中心推理服务FastAPI UvicornTriton / TorchServe监控日志 定时脚本Prometheus Grafana这个表格的核心逻辑是先用最低成本验证“系统能跑通”再根据实际瓶颈决定升级哪个模块。不要为了架构而架构。3. 数据管道AI工程里最脏最累但最不能省的活3.1 数据版本化为什么你的模型复现不出来“这个模型上周跑出来准确率92%今天同样的代码跑出来只有87%。”——如果你遇到过这种情况大概率是数据变了而你没有记录。数据版本化是AI工程里最容易被忽略、但后果最严重的一环。代码可以用Git管理但数据往往几个GB起步不可能直接塞进Git。解决方案是记录数据的指纹而不是数据本身。具体做法每次数据清洗后计算一个哈希值比如对整个文件内容做SHA256把这个哈希值和数据存储路径记录到一张元数据表里。训练时把数据哈希值写进模型元数据。这样任何时候你都能回答“这个模型是用哪一版数据训练的”我用Python写一个最简实现import hashlib import json from datetime import datetime def compute_data_fingerprint(file_path): sha256 hashlib.sha256() with open(file_path, rb) as f: for chunk in iter(lambda: f.read(8192), b): sha256.update(chunk) return sha256.hexdigest() def log_data_version(file_path, version_tag, description): fingerprint compute_data_fingerprint(file_path) record { version_tag: version_tag, fingerprint: fingerprint, file_path: file_path, description: description, created_at: datetime.now().isoformat() } with open(data_versions.jsonl, a) as f: f.write(json.dumps(record) \n) return fingerprint这个脚本跑一次你就有了数据版本的第一块基石。别小看这几十行代码它能在你排查“模型为什么退化”时省下好几天。注意数据指纹要计算清洗后的最终版本而不是原始数据。因为原始数据可能不变但你的清洗逻辑变了结果就完全不同。3.2 数据清洗的工程化别再用Jupyter Notebook做生产数据处理Jupyter Notebook适合探索不适合生产。原因有三第一Notebook的执行顺序是隐式的你很难保证每次跑出来的结果一致第二Notebook里的代码很难做单元测试第三Notebook的依赖管理一塌糊涂。生产级的数据清洗应该是一个可调度的脚本或任务。我推荐的结构是这样的data_pipeline/ ├── config/ │ └── pipeline_config.yaml ├── steps/ │ ├── load.py │ ├── clean.py │ ├── tokenize.py │ └── split.py ├── run_pipeline.py └── tests/ └── test_clean.py每个step是一个纯函数输入输出都是明确的数据结构。run_pipeline.py负责按顺序调用这些step并把中间结果落盘。这样做的好处是任何一步出问题你可以单独重跑那一步而不用从头再来。清洗环节有几个必须处理的点我列出来去重训练集里的重复样本会导致模型过拟合。用MinHash或简单的哈希去重都行。异常值处理文本长度超过模型最大长度的样本是截断还是丢弃我的经验是如果超长样本占比低于5%直接丢弃如果高于5%说明你的截断策略需要重新设计。标签一致性检查多人标注的数据一定要做交叉验证。我见过标注员把“咨询”和“投诉”标反的情况这种数据喂进去模型学到的就是噪声。3.3 训练/验证/测试集的划分陷阱划分数据集看起来简单其实坑很多。最常见的错误是随机划分导致数据泄漏。举个例子你的数据是用户对话同一个用户的多轮对话被随机分到了训练集和验证集。结果就是模型在验证集上表现很好因为它“见过”这个用户的说话风格但上线后遇到新用户就崩了。正确的做法是按“实体”划分。如果数据是按用户组织的就按用户划分如果数据是按时间组织的就按时间划分用过去的数据训练用未来的数据验证。这个原则叫Group Split或Time-based Split。from sklearn.model_selection import GroupShuffleSplit # 假设df有user_id列 splitter GroupShuffleSplit(n_splits1, test_size0.2, random_state42) train_idx, val_idx next(splitter.split(df, groupsdf[user_id])) train_df df.iloc[train_idx] val_df df.iloc[val_idx]实操心得划分完数据集后一定要检查训练集和验证集的标签分布是否一致。如果训练集里“投诉”占30%验证集里占50%那你的评估结果就是不可信的。4. 训练与实验管理让每一次训练都有迹可循4.1 实验记录的最小必要字段实验管理不一定要上MLflow但一定要记录以下字段。我用一张表来说明字段说明为什么重要experiment_id唯一标识区分不同实验git_commit代码版本复现代码data_version数据指纹复现数据hyperparams超参数复现配置metrics评估指标对比效果model_path模型存储路径找到产物timestamp时间戳排序和追溯这七个字段用一张CSV或一个SQLite表就能管起来。关键是每次训练自动写入而不是靠人手动记。我通常在训练脚本的入口和出口各加一段代码入口记录配置出口记录指标和模型路径。4.2 超参数搜索的工程化别再用for循环了手动调参的时代应该过去了。但我不建议你一上来就用Optuna或Ray Tune因为超参搜索的前提是单次训练可复现且可并行。如果你的训练脚本还有随机性没固定搜索出来的结果没有意义。固定随机性的三件套import random import numpy as np import torch def set_seed(seed42): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) torch.backends.cudnn.deterministic True torch.backends.cudnn.benchmark False固定之后再用Optuna做搜索。我一般会先做一轮粗粒度搜索比如学习率取1e-3、1e-4、1e-5三个值锁定大致范围后再做细粒度搜索。这样比一上来就上贝叶斯优化更省时间。4.3 模型评估准确率是最没用的指标如果你的数据是不平衡的比如正样本占5%准确率会骗你。一个把所有样本都预测为负的模型准确率也有95%。所以评估指标要根据场景选分类任务看F1、AUC、PR曲线。如果业务对误报敏感看Precision对漏报敏感看Recall。生成任务看BLEU、ROUGE但更重要的是人工评估。我一般会抽样100条让标注员打分。排序任务看NDCG、MRR。还有一个容易被忽略的点评估要在多个维度上做。除了整体指标还要看分群指标。比如你的模型在“短文本”上F1是0.9在“长文本”上只有0.6那上线后长文本用户就会骂人。5. 推理服务把模型变成能扛住流量的API5.1 推理服务的核心矛盾延迟 vs 吞吐推理服务的设计本质上是在延迟和吞吐之间找平衡。批处理Batching能提高吞吐但会增加延迟因为你要等够一个batch才能推理。这个权衡怎么定我的经验公式是这样的假设你的业务要求P99延迟不超过200ms单次推理耗时20ms那么你最多能等200 - 20 180ms来攒batch。如果请求速率是每秒100个那180ms内能攒到18个请求batch size就可以设到16左右。但这是理想情况。实际中请求速率是波动的所以你需要动态批处理设置一个最大等待时间比如50ms和最大batch size比如32谁先到就触发推理。用FastAPI 一个简单的队列就能实现import asyncio from fastapi import FastAPI from pydantic import BaseModel app FastAPI() queue asyncio.Queue() MAX_BATCH_SIZE 32 MAX_WAIT_MS 50 class Request(BaseModel): text: str async def batch_worker(): while True: batch [] try: # 等待第一个请求 item await queue.get() batch.append(item) # 在MAX_WAIT_MS内尽量攒batch deadline asyncio.get_event_loop().time() MAX_WAIT_MS / 1000 while len(batch) MAX_BATCH_SIZE: timeout deadline - asyncio.get_event_loop().time() if timeout 0: break try: item await asyncio.wait_for(queue.get(), timeouttimeout) batch.append(item) except asyncio.TimeoutError: break # 执行推理 texts [b[0].text for b in batch] results model.predict(texts) # 假设model.predict支持批量 for (future, _), result in zip(batch, results): future.set_result(result) except Exception as e: for future, _ in batch: if not future.done(): future.set_exception(e) app.post(/predict) async def predict(request: Request): future asyncio.get_event_loop().create_future() await queue.put((request, future)) return await future这段代码的核心思想是请求进来后不立即推理而是放进队列由后台worker攒batch。这样单次推理能处理多个请求GPU利用率大幅提升。注意动态批处理会引入额外的延迟所以一定要监控P99延迟。如果延迟超标就调小MAX_WAIT_MS。5.2 模型加载与热更新怎么做到不重启服务换模型模型更新是常态但重启服务意味着中断。解决方案是双缓冲加载服务启动时加载模型A当有新模型B时在后台加载B加载完成后原子性地把流量切到B然后卸载A。import threading class ModelManager: def __init__(self, model_path): self.model self._load(model_path) self.lock threading.Lock() def _load(self, path): # 实际加载逻辑 return load_model(path) def update_model(self, new_path): new_model self._load(new_path) with self.lock: self.model new_model def predict(self, texts): with self.lock: model self.model return model.predict(texts)这个模式的关键是加载新模型时不持锁只有切换引用时持锁。这样加载过程中的推理请求不受影响。5.3 降级与熔断模型挂了怎么办模型服务不可能100%可用。GPU可能OOM模型可能因为输入异常抛错。这时候你需要降级策略一级降级如果模型推理超时返回缓存结果或默认结果。二级降级如果模型服务完全不可用切到规则引擎比如关键词匹配。三级降级返回友好错误并记录日志。熔断用pybreaker这类库就能实现。核心逻辑是连续N次失败后直接拒绝请求一段时间给服务恢复的机会。6. 监控与反馈闭环上线只是开始6.1 必须监控的四个指标模型上线后你至少要看这四个指标延迟P50、P95、P99。P99超标说明有长尾请求需要排查。吞吐QPS。突然下降可能是上游问题突然上升可能是被攻击。错误率包括推理错误和业务错误。推理错误看日志业务错误看用户反馈。模型指标漂移如果线上有标注回流定期计算准确率。如果没标注监控输入分布的变化比如文本平均长度突然变了。我用一个简单的定时脚本做输入分布监控import numpy as np from scipy import stats def detect_drift(reference_data, current_data, threshold0.05): # 用KS检验检测分布漂移 statistic, p_value stats.ks_2samp(reference_data, current_data) if p_value threshold: return True, f分布漂移显著p{p_value:.4f} return False, 分布正常这个脚本每天跑一次对比当天输入和训练集输入的分布。如果漂移显著就触发告警。6.2 反馈闭环让线上数据回流到训练没有反馈闭环的AI系统是死的。闭环的核心是把线上推理的输入和最终结果人工标注或用户行为关联起来定期回流到训练集。具体做法推理服务记录每条请求的request_id和输入。当用户产生反馈比如点击“这个结果不对”把request_id和反馈关联。每周把这些数据导出人工审核后加入训练集。这个闭环跑通后你的模型就能持续进化而不是上线即巅峰。6.3 常见问题速查表问题现象可能原因排查方向解决方案推理延迟突然升高GPU被其他进程占用nvidia-smi查看GPU利用率隔离GPU或限制并发模型输出乱码输入编码问题检查输入文本编码统一用UTF-8准确率下降数据漂移对比输入分布重新训练服务OOMbatch size过大查看显存占用调小batch size模型加载慢模型文件过大查看模型大小模型量化或剪枝实操心得我踩过最大的坑是“模型版本和代码版本不匹配”。有一次线上模型是v3但服务代码是v2的导致预处理逻辑不一致输出全错。后来我强制要求模型元数据里必须记录代码commit hash服务启动时校验。7. 从零搭建的实操路线图如果你现在就要开始我建议按这个顺序推进第一周搭数据管道。把数据清洗、版本化、划分跑通。产出是一份可复现的数据集和一份数据版本记录。第二周搭训练脚本。固定随机性记录实验元数据。产出是一个能一键复现的训练流程。第三周搭推理服务。用FastAPI把模型包成API实现动态批处理。产出是一个能扛住压测的服务。第四周搭监控。记录延迟、吞吐、错误率加一个简单的漂移检测。产出是一个能告警的监控面板。这四周下来你就有了一个最小可用的AI工程系统。之后的所有优化都是在这个骨架上长肉。最后分享一个我自己的习惯我会在项目根目录放一个RUNBOOK.md记录所有关键操作——怎么重跑数据管道、怎么更新模型、怎么回滚。这个文件在半夜出故障时比任何文档都管用。AI工程from scratchscratch的不只是代码更是你对整个系统的掌控力。
1. 从零搭建AI工程体系为什么我劝你别一上来就调包“ai-engineering-from-scratch”这个标题第一次看到的时候我愣了一下。不是因为陌生恰恰相反是因为它戳中了我这几年带团队、做项目最痛的一个点太多人把“AI工程”等同于“会调几个API、会跑几个开源模型”结果一上生产环境就原形毕露——显存炸了、推理延迟飙到秒级、模型版本对不上、数据管道三天两头断流。所谓“from scratch”不是让你从零手写矩阵乘法而是让你从零理解一个AI系统到底由哪些零件拼起来、每个零件的边界在哪、它们之间怎么咬合。这件事的价值在你第一次遇到线上事故、第一次被要求把推理成本砍掉一半、第一次需要把一个实验室里的demo变成能扛住每天百万次调用的服务时会体现得淋漓尽致。这篇文章适合三类人一是刚转行做AI工程、只会调库但心里没底的开发者二是带团队做AI落地、需要统一技术认知的技术负责人三是想搞清楚“AI系统到底怎么跑起来”的产品和运维同学。我会把AI工程从零搭建的完整链路拆开讲清楚每一步为什么这么做、坑在哪里、怎么绕过去。全程不堆术语能上手抄的配置和代码我都会给出来。2. 先想清楚AI工程到底在工程什么2.1 一个被严重低估的认知AI工程不等于模型训练很多人对AI工程的想象是这样的搞一堆数据扔进GPU训练出一个模型然后上线。这个链条里训练只占整个工程量的两成不到。真正吃掉你80%精力的是训练之前的数据处理、训练之后的推理服务、以及贯穿始终的版本管理和监控。我拿一个真实场景举例。假设你要做一个文本分类服务识别用户反馈是“投诉”还是“咨询”。模型本身可能就是一个几百MB的BERT变体训练一天就能出结果。但你要让它稳定服务需要解决这些问题用户输入的长度分布是什么超长文本怎么截断批量推理的batch size设多少才能在延迟和吞吐之间平衡模型更新时怎么做到不中断服务线上出现bad case怎么快速回溯是哪一版模型、哪一批数据的问题这些问题的答案构成了AI工程的真正骨架。所以“from scratch”的第一课是把视角从“模型”拉到“系统”。2.2 核心模块拆解一个AI系统的最小完整形态一个能上生产的AI系统至少包含五个模块缺一不可数据管道负责数据的采集、清洗、标注、版本化。它的输出是训练和评估的输入。训练与实验管理负责模型训练、超参搜索、实验记录。关键是可复现。模型仓库与版本管理负责模型产物的存储、版本标记、元数据记录。模型不是文件是带血缘的资产。推理服务负责把模型变成API处理并发、批处理、超时、降级。监控与反馈闭环负责监控延迟、吞吐、准确率漂移并把线上数据回流到训练。这五个模块之间的关系不是流水线而是环。数据管道喂给训练训练产出模型进仓库仓库驱动推理服务推理服务的线上表现又通过监控回流到数据管道。任何一个环节断了整个系统就是瘸的。我见过太多团队训练脚本写得飞起但模型文件用model_final_v2_real.pth这种名字存着过两周自己都不知道哪个是哪个。这就是典型的“有模型、没工程”。2.3 技术选型的底层逻辑为什么我推荐从轻量级方案起步一提到AI工程很多人第一反应是上Kubernetes、上MLflow、上Feature Store。这些工具都好但如果你是一个人或者小团队从零开始我强烈建议你先用最轻量的方案把闭环跑通再逐步替换。原因很简单工具本身有认知成本和运维成本。你花两周搭了一套K8s集群结果发现你的推理服务QPS只有个位数那这套集群除了让你觉得自己很专业之外没有任何价值。我的建议起步方案是这样的模块轻量起步方案规模化替换方案数据管道脚本 本地文件 DVCAirflow / Dagster实验管理表格 Git commit hashMLflow / WB模型仓库对象存储 命名规范模型注册中心推理服务FastAPI UvicornTriton / TorchServe监控日志 定时脚本Prometheus Grafana这个表格的核心逻辑是先用最低成本验证“系统能跑通”再根据实际瓶颈决定升级哪个模块。不要为了架构而架构。3. 数据管道AI工程里最脏最累但最不能省的活3.1 数据版本化为什么你的模型复现不出来“这个模型上周跑出来准确率92%今天同样的代码跑出来只有87%。”——如果你遇到过这种情况大概率是数据变了而你没有记录。数据版本化是AI工程里最容易被忽略、但后果最严重的一环。代码可以用Git管理但数据往往几个GB起步不可能直接塞进Git。解决方案是记录数据的指纹而不是数据本身。具体做法每次数据清洗后计算一个哈希值比如对整个文件内容做SHA256把这个哈希值和数据存储路径记录到一张元数据表里。训练时把数据哈希值写进模型元数据。这样任何时候你都能回答“这个模型是用哪一版数据训练的”我用Python写一个最简实现import hashlib import json from datetime import datetime def compute_data_fingerprint(file_path): sha256 hashlib.sha256() with open(file_path, rb) as f: for chunk in iter(lambda: f.read(8192), b): sha256.update(chunk) return sha256.hexdigest() def log_data_version(file_path, version_tag, description): fingerprint compute_data_fingerprint(file_path) record { version_tag: version_tag, fingerprint: fingerprint, file_path: file_path, description: description, created_at: datetime.now().isoformat() } with open(data_versions.jsonl, a) as f: f.write(json.dumps(record) \n) return fingerprint这个脚本跑一次你就有了数据版本的第一块基石。别小看这几十行代码它能在你排查“模型为什么退化”时省下好几天。注意数据指纹要计算清洗后的最终版本而不是原始数据。因为原始数据可能不变但你的清洗逻辑变了结果就完全不同。3.2 数据清洗的工程化别再用Jupyter Notebook做生产数据处理Jupyter Notebook适合探索不适合生产。原因有三第一Notebook的执行顺序是隐式的你很难保证每次跑出来的结果一致第二Notebook里的代码很难做单元测试第三Notebook的依赖管理一塌糊涂。生产级的数据清洗应该是一个可调度的脚本或任务。我推荐的结构是这样的data_pipeline/ ├── config/ │ └── pipeline_config.yaml ├── steps/ │ ├── load.py │ ├── clean.py │ ├── tokenize.py │ └── split.py ├── run_pipeline.py └── tests/ └── test_clean.py每个step是一个纯函数输入输出都是明确的数据结构。run_pipeline.py负责按顺序调用这些step并把中间结果落盘。这样做的好处是任何一步出问题你可以单独重跑那一步而不用从头再来。清洗环节有几个必须处理的点我列出来去重训练集里的重复样本会导致模型过拟合。用MinHash或简单的哈希去重都行。异常值处理文本长度超过模型最大长度的样本是截断还是丢弃我的经验是如果超长样本占比低于5%直接丢弃如果高于5%说明你的截断策略需要重新设计。标签一致性检查多人标注的数据一定要做交叉验证。我见过标注员把“咨询”和“投诉”标反的情况这种数据喂进去模型学到的就是噪声。3.3 训练/验证/测试集的划分陷阱划分数据集看起来简单其实坑很多。最常见的错误是随机划分导致数据泄漏。举个例子你的数据是用户对话同一个用户的多轮对话被随机分到了训练集和验证集。结果就是模型在验证集上表现很好因为它“见过”这个用户的说话风格但上线后遇到新用户就崩了。正确的做法是按“实体”划分。如果数据是按用户组织的就按用户划分如果数据是按时间组织的就按时间划分用过去的数据训练用未来的数据验证。这个原则叫Group Split或Time-based Split。from sklearn.model_selection import GroupShuffleSplit # 假设df有user_id列 splitter GroupShuffleSplit(n_splits1, test_size0.2, random_state42) train_idx, val_idx next(splitter.split(df, groupsdf[user_id])) train_df df.iloc[train_idx] val_df df.iloc[val_idx]实操心得划分完数据集后一定要检查训练集和验证集的标签分布是否一致。如果训练集里“投诉”占30%验证集里占50%那你的评估结果就是不可信的。4. 训练与实验管理让每一次训练都有迹可循4.1 实验记录的最小必要字段实验管理不一定要上MLflow但一定要记录以下字段。我用一张表来说明字段说明为什么重要experiment_id唯一标识区分不同实验git_commit代码版本复现代码data_version数据指纹复现数据hyperparams超参数复现配置metrics评估指标对比效果model_path模型存储路径找到产物timestamp时间戳排序和追溯这七个字段用一张CSV或一个SQLite表就能管起来。关键是每次训练自动写入而不是靠人手动记。我通常在训练脚本的入口和出口各加一段代码入口记录配置出口记录指标和模型路径。4.2 超参数搜索的工程化别再用for循环了手动调参的时代应该过去了。但我不建议你一上来就用Optuna或Ray Tune因为超参搜索的前提是单次训练可复现且可并行。如果你的训练脚本还有随机性没固定搜索出来的结果没有意义。固定随机性的三件套import random import numpy as np import torch def set_seed(seed42): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) torch.backends.cudnn.deterministic True torch.backends.cudnn.benchmark False固定之后再用Optuna做搜索。我一般会先做一轮粗粒度搜索比如学习率取1e-3、1e-4、1e-5三个值锁定大致范围后再做细粒度搜索。这样比一上来就上贝叶斯优化更省时间。4.3 模型评估准确率是最没用的指标如果你的数据是不平衡的比如正样本占5%准确率会骗你。一个把所有样本都预测为负的模型准确率也有95%。所以评估指标要根据场景选分类任务看F1、AUC、PR曲线。如果业务对误报敏感看Precision对漏报敏感看Recall。生成任务看BLEU、ROUGE但更重要的是人工评估。我一般会抽样100条让标注员打分。排序任务看NDCG、MRR。还有一个容易被忽略的点评估要在多个维度上做。除了整体指标还要看分群指标。比如你的模型在“短文本”上F1是0.9在“长文本”上只有0.6那上线后长文本用户就会骂人。5. 推理服务把模型变成能扛住流量的API5.1 推理服务的核心矛盾延迟 vs 吞吐推理服务的设计本质上是在延迟和吞吐之间找平衡。批处理Batching能提高吞吐但会增加延迟因为你要等够一个batch才能推理。这个权衡怎么定我的经验公式是这样的假设你的业务要求P99延迟不超过200ms单次推理耗时20ms那么你最多能等200 - 20 180ms来攒batch。如果请求速率是每秒100个那180ms内能攒到18个请求batch size就可以设到16左右。但这是理想情况。实际中请求速率是波动的所以你需要动态批处理设置一个最大等待时间比如50ms和最大batch size比如32谁先到就触发推理。用FastAPI 一个简单的队列就能实现import asyncio from fastapi import FastAPI from pydantic import BaseModel app FastAPI() queue asyncio.Queue() MAX_BATCH_SIZE 32 MAX_WAIT_MS 50 class Request(BaseModel): text: str async def batch_worker(): while True: batch [] try: # 等待第一个请求 item await queue.get() batch.append(item) # 在MAX_WAIT_MS内尽量攒batch deadline asyncio.get_event_loop().time() MAX_WAIT_MS / 1000 while len(batch) MAX_BATCH_SIZE: timeout deadline - asyncio.get_event_loop().time() if timeout 0: break try: item await asyncio.wait_for(queue.get(), timeouttimeout) batch.append(item) except asyncio.TimeoutError: break # 执行推理 texts [b[0].text for b in batch] results model.predict(texts) # 假设model.predict支持批量 for (future, _), result in zip(batch, results): future.set_result(result) except Exception as e: for future, _ in batch: if not future.done(): future.set_exception(e) app.post(/predict) async def predict(request: Request): future asyncio.get_event_loop().create_future() await queue.put((request, future)) return await future这段代码的核心思想是请求进来后不立即推理而是放进队列由后台worker攒batch。这样单次推理能处理多个请求GPU利用率大幅提升。注意动态批处理会引入额外的延迟所以一定要监控P99延迟。如果延迟超标就调小MAX_WAIT_MS。5.2 模型加载与热更新怎么做到不重启服务换模型模型更新是常态但重启服务意味着中断。解决方案是双缓冲加载服务启动时加载模型A当有新模型B时在后台加载B加载完成后原子性地把流量切到B然后卸载A。import threading class ModelManager: def __init__(self, model_path): self.model self._load(model_path) self.lock threading.Lock() def _load(self, path): # 实际加载逻辑 return load_model(path) def update_model(self, new_path): new_model self._load(new_path) with self.lock: self.model new_model def predict(self, texts): with self.lock: model self.model return model.predict(texts)这个模式的关键是加载新模型时不持锁只有切换引用时持锁。这样加载过程中的推理请求不受影响。5.3 降级与熔断模型挂了怎么办模型服务不可能100%可用。GPU可能OOM模型可能因为输入异常抛错。这时候你需要降级策略一级降级如果模型推理超时返回缓存结果或默认结果。二级降级如果模型服务完全不可用切到规则引擎比如关键词匹配。三级降级返回友好错误并记录日志。熔断用pybreaker这类库就能实现。核心逻辑是连续N次失败后直接拒绝请求一段时间给服务恢复的机会。6. 监控与反馈闭环上线只是开始6.1 必须监控的四个指标模型上线后你至少要看这四个指标延迟P50、P95、P99。P99超标说明有长尾请求需要排查。吞吐QPS。突然下降可能是上游问题突然上升可能是被攻击。错误率包括推理错误和业务错误。推理错误看日志业务错误看用户反馈。模型指标漂移如果线上有标注回流定期计算准确率。如果没标注监控输入分布的变化比如文本平均长度突然变了。我用一个简单的定时脚本做输入分布监控import numpy as np from scipy import stats def detect_drift(reference_data, current_data, threshold0.05): # 用KS检验检测分布漂移 statistic, p_value stats.ks_2samp(reference_data, current_data) if p_value threshold: return True, f分布漂移显著p{p_value:.4f} return False, 分布正常这个脚本每天跑一次对比当天输入和训练集输入的分布。如果漂移显著就触发告警。6.2 反馈闭环让线上数据回流到训练没有反馈闭环的AI系统是死的。闭环的核心是把线上推理的输入和最终结果人工标注或用户行为关联起来定期回流到训练集。具体做法推理服务记录每条请求的request_id和输入。当用户产生反馈比如点击“这个结果不对”把request_id和反馈关联。每周把这些数据导出人工审核后加入训练集。这个闭环跑通后你的模型就能持续进化而不是上线即巅峰。6.3 常见问题速查表问题现象可能原因排查方向解决方案推理延迟突然升高GPU被其他进程占用nvidia-smi查看GPU利用率隔离GPU或限制并发模型输出乱码输入编码问题检查输入文本编码统一用UTF-8准确率下降数据漂移对比输入分布重新训练服务OOMbatch size过大查看显存占用调小batch size模型加载慢模型文件过大查看模型大小模型量化或剪枝实操心得我踩过最大的坑是“模型版本和代码版本不匹配”。有一次线上模型是v3但服务代码是v2的导致预处理逻辑不一致输出全错。后来我强制要求模型元数据里必须记录代码commit hash服务启动时校验。7. 从零搭建的实操路线图如果你现在就要开始我建议按这个顺序推进第一周搭数据管道。把数据清洗、版本化、划分跑通。产出是一份可复现的数据集和一份数据版本记录。第二周搭训练脚本。固定随机性记录实验元数据。产出是一个能一键复现的训练流程。第三周搭推理服务。用FastAPI把模型包成API实现动态批处理。产出是一个能扛住压测的服务。第四周搭监控。记录延迟、吞吐、错误率加一个简单的漂移检测。产出是一个能告警的监控面板。这四周下来你就有了一个最小可用的AI工程系统。之后的所有优化都是在这个骨架上长肉。最后分享一个我自己的习惯我会在项目根目录放一个RUNBOOK.md记录所有关键操作——怎么重跑数据管道、怎么更新模型、怎么回滚。这个文件在半夜出故障时比任何文档都管用。AI工程from scratchscratch的不只是代码更是你对整个系统的掌控力。
