1. 从零搭建AI工程体系为什么我劝你别一上来就调包“ai-engineering-from-scratch”这个标题第一次看到的时候我愣了一下。市面上讲AI的教程铺天盖地但绝大多数都是教你import torch然后跑一个预训练模型或者调个API就完事。真正讲“从零开始搭建AI工程体系”的内容少得可怜。我自己在这个行业摸爬滚打了十来年带过不少新人也面试过几百个号称“会AI”的候选人。一个很普遍的现象是很多人能背出Transformer的结构图能说清楚Attention的计算公式但你让他从零搭一个能用的训练流水线他立刻卡壳。数据怎么组织显存怎么优化训练中断了怎么恢复模型上线后怎么监控这些问题调包教程里不会讲但实际工作中天天遇到。所以这个项目标题吸引我的地方在于它瞄准的是一个真实存在的巨大缺口从“会调包”到“能工程化落地”之间的鸿沟。这篇文章我想基于自己这些年的实际经验把这个标题背后的核心内容拆开揉碎讲清楚一个完整的AI工程体系到底包含哪些环节每个环节的关键决策怎么做以及那些只有踩过坑才知道的细节。适合谁看如果你是刚入行的算法工程师正在从“跑通demo”向“能交付项目”过渡这篇文章能帮你少走至少半年的弯路。如果你是有经验的开发者想转AI方向这里面的工程思路对你来说反而是优势。如果你只是好奇AI工程到底在干什么也能从中看到一个真实的行业切面。2. 整体架构设计AI工程体系到底包含什么2.1 为什么“从零”不等于“从轮子造起”先澄清一个概念。“ai-engineering-from-scratch”里的“from scratch”不是让你用NumPy手写一个Transformer——那是深度学习框架开发者干的事。这里的“从零”指的是从零搭建一套完整的、可运行的、能支撑实际业务的AI工程体系。打个比方。你要开一家餐厅“从零”不是让你去种小麦、养牛、炼铁锅而是让你从选址、菜单设计、供应链搭建、厨房动线规划、人员分工、品控流程这一整套东西开始搭。AI工程也是一样PyTorch、TensorFlow这些框架就是你的“厨具”你不需要自己造但你需要知道怎么选、怎么用、怎么把它们组织成一个高效运转的系统。一个完整的AI工程体系我习惯把它拆成六个层次数据层数据的采集、清洗、标注、版本管理、特征存储训练层实验管理、分布式训练、超参调优、模型检查点评估层离线评估、在线A/B测试、回归测试、偏差检测部署层模型压缩、推理优化、服务化、灰度发布监控层性能监控、数据漂移检测、告警机制迭代层反馈闭环、持续训练、版本回滚这六层缺一不可。很多团队的问题就在于只关注训练层觉得模型训出来就万事大吉了结果上线后各种问题层出不穷。2.2 技术选型的核心逻辑场景驱动而非技术驱动在搭建这套体系之前最重要的一步是明确你的场景需求。我见过太多团队一上来就选最时髦的技术栈结果发现根本不匹配自己的业务。选型的时候我一般会问自己四个问题第一个问题数据量级和更新频率是多少如果你的数据是TB级别的每天新增几十万条那你的数据管道必须支持流式处理特征存储要能支撑高并发读写。如果数据量不大一周更新一次那用简单的批处理脚本就够了没必要上KafkaSpark那一套。第二个问题训练频率和延迟要求是什么一天训一次模型和一个月训一次对基础设施的要求完全不同。前者需要自动化的训练流水线后者手动触发就行。推理延迟要求是10ms还是1s直接决定了你要不要做模型量化、蒸馏、TensorRT优化。第三个问题团队的技术储备如何如果团队里没人懂Kubernetes你硬上K8s做模型部署运维成本会高到离谱。选型要匹配团队能力而不是匹配招聘JD。第四个问题预算约束是什么训练用A100还是3090推理用T4还是CPU这些都要算账。我后面会给出具体的成本估算方法。基于这四个问题的答案你的技术栈选型基本就确定了。下面这张表是我总结的常见场景对应的推荐方案场景特征数据层训练层部署层监控层小数据量、低频更新文件系统SQLite单机GPUFlask API日志文件中等数据量、日更新PostgreSQLMinIO单机多卡FastAPITritonPrometheusGrafana大数据量、实时更新KafkaFeature Store分布式训练K8sTriton全链路监控边缘部署本地缓存云端训练ONNX Runtime端侧日志回传注意这张表是起点不是终点。实际选型一定要结合具体业务做调整不要照搬。2.3 目录结构设计让项目可维护的第一步一个AI工程项目最容易烂掉的地方就是目录结构。我接手过很多“祖传代码”打开一看根目录下几十个文件混在一起train.py、train_v2.py、train_final.py、train_final_真的最终版.py找代码全靠搜索。从项目第一天起就定好目录结构后面能省无数时间。我推荐的结构是这样的project/ ├── configs/ # 配置文件 │ ├── base.yaml │ ├── train.yaml │ └── deploy.yaml ├── data/ # 数据相关 │ ├── raw/ # 原始数据 │ ├── processed/ # 处理后数据 │ └── schemas/ # 数据schema定义 ├── src/ # 源代码 │ ├── data/ # 数据处理模块 │ ├── models/ # 模型定义 │ ├── training/ # 训练逻辑 │ ├── evaluation/ # 评估逻辑 │ └── serving/ # 服务化代码 ├── experiments/ # 实验记录 ├── notebooks/ # 探索性分析 ├── tests/ # 测试代码 ├── scripts/ # 运维脚本 └── docs/ # 文档这个结构的关键在于关注点分离。配置和代码分离数据和代码分离训练和服务分离。这样做的好处是当你要改一个超参数时不需要动代码当你要换一个模型时不需要动数据管道当你要部署时不需要把训练代码也打包进去。3. 数据管道搭建AI工程的地基3.1 数据版本管理别再用文件名区分了数据版本管理是AI工程里最容易被忽视的环节。我见过太多团队用data_v1.csv、data_v2_fixed.csv这种方式管理数据结果三个月后没人说得清哪个版本对应哪次实验。正确的做法是引入数据版本控制工具。DVCData Version Control是我用得最顺手的它和Git无缝集成大文件存在远端存储Git里只存元数据。基本用法# 初始化DVC dvc init # 添加数据文件到DVC管理 dvc add data/raw/train.csv # 提交DVC元数据到Git git add data/raw/train.csv.dvc data/raw/.gitignore git commit -m add training data v1 # 推送数据到远端存储 dvc remote add -d storage s3://my-bucket/dvc-store dvc push这样每次实验都能精确对应到数据版本。当你发现三个月前那个效果特别好的模型时能一键还原当时的数据。实操心得DVC的缓存机制会占用大量本地磁盘建议定期执行dvc gc清理不再需要的缓存。另外远端存储建议用对象存储而不是NFS性能和可靠性都好很多。3.2 数据清洗的工程化方法数据清洗不是写个Jupyter Notebook跑一遍就完事它需要工程化。核心原则是清洗逻辑要可复用、可测试、可追溯。我一般把清洗分成三个阶段阶段一数据探查。用Pandas Profiling或Great Expectations这类工具自动生成数据质量报告。重点看缺失值比例、异常值分布、类别不平衡程度。这一步在Notebook里做就行目的是了解数据长什么样。阶段二清洗规则定义。把探查阶段发现的每个问题转化成一条明确的规则。比如“age字段不能为负数”、“category字段只能是A/B/C三种值”。这些规则用代码写出来放在src/data/validators.py里。阶段三清洗流水线。把规则串成流水线用Airflow或Prefect这类工具调度。每次新数据进来自动跑一遍清洗输出清洗报告。# src/data/cleaners.py import pandas as pd from typing import Dict, Any class DataCleaner: def __init__(self, config: Dict[str, Any]): self.config config def remove_duplicates(self, df: pd.DataFrame) - pd.DataFrame: 去重保留第一条 before len(df) df df.drop_duplicates(subsetself.config[dedup_keys]) after len(df) print(f去重: {before} - {after}, 移除 {before - after} 条) return df def handle_missing(self, df: pd.DataFrame) - pd.DataFrame: 处理缺失值 for col, strategy in self.config[missing_strategy].items(): if strategy drop: df df.dropna(subset[col]) elif strategy mean: df[col] df[col].fillna(df[col].mean()) elif strategy median: df[col] df[col].fillna(df[col].median()) elif strategy.startswith(constant:): value strategy.split(:)[1] df[col] df[col].fillna(value) return df def clip_outliers(self, df: pd.DataFrame) - pd.DataFrame: 截断异常值 for col, (lower, upper) in self.config[clip_ranges].items(): df[col] df[col].clip(lower, upper) return df注意清洗规则一定要写单元测试。我踩过的坑是某次改了清洗逻辑导致某个字段的分布完全变了但因为没有测试直到模型效果暴跌才发现。现在我的习惯是每条清洗规则至少配一个测试用例。3.3 特征存储从混乱到有序当项目从单模型发展到多模型时特征管理就会变成噩梦。同一个特征推荐模型算一遍风控模型又算一遍两边逻辑还不一致。这时候就需要特征存储Feature Store。特征存储的核心价值是特征复用和一致性保证。训练时用的特征和推理时用的特征必须来自同一套计算逻辑否则会出现训练-服务偏差Training-Serving Skew。我推荐用Feast作为特征存储方案它轻量、开源、和主流框架集成好。基本架构是离线存储存历史特征用于训练。一般用Parquet文件或数据仓库。在线存储存最新特征用于推理。一般用Redis或DynamoDB。特征定义用Python代码定义特征的计算逻辑保证线上线下一致。# feature_repo/features.py from feast import Entity, Feature, FeatureView, FileSource from feast.types import Float32, Int64 from datetime import timedelta # 定义实体 user Entity(nameuser_id, join_keys[user_id]) # 定义数据源 user_stats_source FileSource( pathdata/user_stats.parquet, timestamp_fieldevent_timestamp, ) # 定义特征视图 user_stats_fv FeatureView( nameuser_stats, entities[user], ttltimedelta(days7), schema[ Feature(nametotal_orders, dtypeInt64), Feature(nameavg_order_value, dtypeFloat32), Feature(namedays_since_last_order, dtypeInt64), ], sourceuser_stats_source, )这样定义之后训练时用get_historical_features拿历史特征推理时用get_online_features拿实时特征两边逻辑完全一致。4. 训练流水线从手动脚本到自动化系统4.1 实验管理让每次实验都可追溯没有实验管理的团队工作状态是这样的小王跑了个模型准确率92%很高兴地跟组长汇报。组长问“用的什么超参数”小王说“等我翻一下Notebook。”翻了半小时找到了但发现那个Notebook后来又被改过已经复现不出来了。实验管理要解决的就是这个问题。核心要求是任何一次实验结果都能精确复现。这意味着要记录代码版本、数据版本、超参数、环境依赖、随机种子、硬件信息。我常用的方案是MLflow它轻量、易用、功能全。基本用法import mlflow import mlflow.pytorch # 设置实验 mlflow.set_experiment(my-ai-project) with mlflow.start_run(run_namebaseline-v1): # 记录超参数 mlflow.log_params({ learning_rate: 1e-4, batch_size: 32, epochs: 50, optimizer: adamw, }) # 记录指标 for epoch in range(50): train_loss train_one_epoch() val_loss, val_acc validate() mlflow.log_metrics({ train_loss: train_loss, val_loss: val_loss, val_acc: val_acc, }, stepepoch) # 记录模型 mlflow.pytorch.log_model(model, model) # 记录数据版本 mlflow.log_param(data_version, v1.2.0) mlflow.log_param(git_commit, get_git_commit())实操心得MLflow的Tracking Server建议单独部署不要用本地文件存储。团队协作时所有人都往同一个Server打点才能对比实验。另外模型文件不要直接存MLflow存到对象存储MLflow里只存路径。4.2 分布式训练什么时候需要怎么选分布式训练不是银弹。很多场景下单卡就够用硬上分布式反而增加复杂度和故障率。我的判断标准是单卡能训完且时间可接受比如8小时内单卡单卡训不完或时间太长上分布式模型大到单卡放不下必须分布式分布式训练有三种主流方案数据并行Data Parallelism每个GPU持有完整模型副本数据切分到各GPU梯度汇总后更新。适合模型能放进单卡显存、但数据量大的场景。PyTorch的DDP是标准方案。模型并行Model Parallelism模型切分到多个GPU每个GPU持有部分层。适合模型大到单卡放不下的场景。实现复杂一般用Megatron-LM或DeepSpeed。流水线并行Pipeline Parallelism模型按层切分成多个阶段不同阶段在不同GPU上数据像流水线一样流过。适合超大规模模型。PyTorch的Pipe是入门方案。实际项目中最常见的是数据并行混合精度。下面是一个DDP的启动脚本# 单机8卡训练 torchrun --nproc_per_node8 \ --master_port29500 \ src/training/train.py \ --config configs/train.yaml \ --batch_size 64 \ --learning_rate 1e-4# src/training/train.py 关键部分 import torch.distributed as dist from torch.nn.parallel import DistributedDataParallel as DDP def setup_ddp(): dist.init_process_group(backendnccl) local_rank int(os.environ[LOCAL_RANK]) torch.cuda.set_device(local_rank) return local_rank def main(): local_rank setup_ddp() model MyModel().to(local_rank) model DDP(model, device_ids[local_rank]) # 数据加载器要用DistributedSampler sampler DistributedSampler(dataset) dataloader DataLoader(dataset, samplersampler, batch_size...) for epoch in range(epochs): sampler.set_epoch(epoch) # 重要每个epoch要重新shuffle for batch in dataloader: ...注意DDP下只有rank 0的进程应该做日志记录和模型保存否则会重复写文件导致冲突。另外sampler.set_epoch(epoch)这行很容易漏漏了的话每个epoch的数据顺序都一样影响训练效果。4.3 超参数调优别再用网格搜索了超参数调优是很多团队的痛点。网格搜索太慢随机搜索碰运气手动调参靠经验。我推荐用贝叶斯优化工具选Optuna。Optuna的核心优势是用更少的试验次数找到更好的参数。它通过建立代理模型预测哪些参数组合可能更好从而指导下一步搜索方向。import optuna def objective(trial): # 定义搜索空间 lr trial.suggest_float(lr, 1e-5, 1e-2, logTrue) batch_size trial.suggest_categorical(batch_size, [16, 32, 64, 128]) dropout trial.suggest_float(dropout, 0.1, 0.5) num_layers trial.suggest_int(num_layers, 2, 8) # 训练模型 model build_model(num_layers, dropout) val_acc train_and_evaluate(model, lr, batch_size) return val_acc # 创建study study optuna.create_study( directionmaximize, sampleroptuna.samplers.TPESampler(seed42), pruneroptuna.pruners.MedianPruner(n_startup_trials5), ) # 开始优化 study.optimize(objective, n_trials100, timeout3600*8) # 输出最佳参数 print(fBest trial: {study.best_trial.params})实操心得Optuna的Pruner功能非常有用。它会在训练过程中自动终止那些明显不好的试验节省大量算力。我一般设置n_startup_trials5让前5个试验完整跑完之后才开始剪枝。另外搜索空间的定义很关键范围不要设太宽否则大部分试验都在无效区域浪费算力。5. 模型部署与服务化让模型真正产生价值5.1 推理优化从100ms到10ms的实战路径模型训出来只是第一步能不能高效推理才是关键。我做过一个项目原始模型推理延迟120ms经过一系列优化后降到8ms提升了15倍。下面是我用的优化路径第一步模型量化。把FP32转成FP16或INT8显存占用减半推理速度提升1.5-2倍。PyTorch的torch.quantization或ONNX Runtime的量化工具都能做。# PyTorch动态量化 import torch.quantization model MyModel() model.eval() quantized_model torch.quantization.quantize_dynamic( model, {torch.nn.Linear}, dtypetorch.qint8 ) torch.save(quantized_model.state_dict(), model_quantized.pt)第二步算子融合。把多个连续的小算子合并成一个减少kernel launch开销。TensorRT和ONNX Runtime都有自动融合功能。第三步推理引擎优化。用TensorRT或ONNX Runtime替换原生PyTorch推理。这一步通常能带来2-3倍的提升。# ONNX Runtime推理 import onnxruntime as ort import numpy as np # 导出ONNX torch.onnx.export(model, dummy_input, model.onnx, opset_version13) # 创建推理会话 session ort.InferenceSession( model.onnx, providers[CUDAExecutionProvider], ) # 推理 input_name session.get_inputs()[0].name output session.run(None, {input_name: input_data})第四步批处理。把多个请求攒成一批一起推理GPU利用率能提升好几倍。但要注意延迟和吞吐的权衡。第五步缓存。对于重复的输入直接返回缓存结果。这个在推荐系统里特别有效热门物品的embedding可以缓存。优化手段延迟降低实现难度适用场景FP16量化1.5-2x低所有GPU场景INT8量化2-4x中对精度不敏感场景算子融合1.2-1.5x低所有场景TensorRT2-3x中NVIDIA GPU批处理3-5x中高吞吐场景缓存10x低重复输入多5.2 服务化框架选型FastAPI vs Triton vs TorchServe模型服务化框架的选择直接决定了你的运维复杂度。我三个都用过说说各自的适用场景。FastAPI最灵活适合自定义逻辑多的场景。你可以完全控制请求处理流程做各种预处理和后处理。缺点是性能一般高并发下需要自己优化。from fastapi import FastAPI import torch app FastAPI() model load_model() app.post(/predict) async def predict(request: PredictRequest): # 预处理 input_tensor preprocess(request.data) # 推理 with torch.no_grad(): output model(input_tensor) # 后处理 result postprocess(output) return {prediction: result}Triton Inference ServerNVIDIA出品性能最强支持多框架、多模型、动态批处理。适合大规模部署场景。缺点是需要写配置文件灵活性不如FastAPI。TorchServePyTorch官方方案和PyTorch生态集成好。适合纯PyTorch模型的部署。缺点是社区活跃度一般遇到问题不好找资料。我的建议是小规模用FastAPI大规模用Triton纯PyTorch生态用TorchServe。不要一上来就上Triton它的学习曲线不陡但也不平小项目用它是杀鸡用牛刀。5.3 灰度发布与回滚上线不是终点模型上线最怕的是什么是新模型效果不如旧模型但你已经全量发布了。所以灰度发布是必须的。我的做法是分三个阶段阶段一影子模式Shadow Mode。新模型和旧模型同时运行但只有旧模型的结果返回给用户。新模型的结果只记录不生效。这个阶段跑一周对比两个模型的输出差异。阶段二小流量灰度。把1%的流量切给新模型观察核心指标。如果指标没有明显下降逐步扩大到5%、10%、50%。阶段三全量发布。灰度到100%后旧模型保留一周作为回滚备份确认没问题后再下线。# 灰度发布逻辑 import random def route_request(user_id: str) - str: 根据用户ID决定用哪个模型 # 用user_id的hash做分流保证同一用户始终用同一模型 bucket hash(user_id) % 100 if bucket current_gray_ratio: return new_model else: return old_model注意分流要用user_id的hash不要用随机数。否则同一用户一会儿用新模型一会儿用旧模型体验会非常割裂。另外灰度比例要动态可调不要写死在代码里。6. 监控与迭代让系统持续进化6.1 数据漂移检测模型效果下降的预警器模型上线后效果下降最常见的原因是数据漂移。输入数据的分布变了模型还按老规律预测自然就不准了。数据漂移检测的核心是对比训练数据和线上数据的分布。常用的方法有PSIPopulation Stability Index衡量两个分布的差异PSI0.1表示稳定0.1-0.25表示轻微漂移0.25表示显著漂移。KS检验统计两个分布是否有显著差异。KL散度衡量两个概率分布的差异。import numpy as np from scipy import stats def calculate_psi(expected, actual, buckets10): 计算PSI # 分桶 breakpoints np.percentile(expected, np.linspace(0, 100, buckets 1)) breakpoints[0] -np.inf breakpoints[-1] np.inf expected_percents np.histogram(expected, breakpoints)[0] / len(expected) actual_percents np.histogram(actual, breakpoints)[0] / len(actual) # 避免除零 expected_percents np.clip(expected_percents, 1e-6, None) actual_percents np.clip(actual_percents, 1e-6, None) psi np.sum((actual_percents - expected_percents) * np.log(actual_percents / expected_percents)) return psi # 使用 psi calculate_psi(train_data[feature_1], online_data[feature_1]) if psi 0.25: alert(特征feature_1发生显著漂移PSI{:.3f}.format(psi))实操心得PSI对分桶数量敏感一般用10-20个桶。另外不要对所有特征都设同样的阈值有些特征天然波动大阈值要放宽。我一般先跑一周基线看每个特征PSI的正常波动范围再定阈值。6.2 模型性能监控不只是准确率模型上线后准确率不是唯一要看的指标。我一般会监控四类指标业务指标点击率、转化率、GMV等。这是最终目标但反馈慢不能作为唯一依据。模型指标准确率、召回率、AUC等。需要标注数据才能算有延迟。系统指标QPS、延迟、错误率、GPU利用率。实时性强能快速发现问题。数据指标特征分布、缺失率、异常值比例。能提前预警数据问题。这四类指标要放在同一个Dashboard上看才能快速定位问题。比如业务指标下降同时数据指标显示某个特征漂移了那大概率是数据问题。6.3 持续训练让模型跟上变化模型不是训一次就完事需要持续迭代。持续训练的核心是自动化和安全。自动化指的是数据更新后自动触发训练训练完成后自动评估评估通过后自动部署。这个流水线用Airflow或Kubeflow Pipelines搭建。安全指的是新模型必须经过严格评估才能上线。我一般设三道关卡离线评估在留出集上新模型的核心指标必须不低于旧模型的95%。回归测试在一组固定的测试用例上新模型的输出不能有异常变化。影子测试新模型和旧模型并行跑24小时输出差异在可接受范围内。三道关卡都通过才允许灰度发布。# 持续训练流水线伪代码 def continuous_training_pipeline(): # 1. 检查数据更新 if not new_data_available(): return # 2. 数据清洗和验证 cleaned_data clean_and_validate(new_data) # 3. 训练新模型 new_model train(cleaned_data) # 4. 离线评估 metrics evaluate(new_model, test_set) if metrics[auc] baseline[auc] * 0.95: alert(新模型效果不达标终止发布) return # 5. 回归测试 if not regression_test(new_model): alert(回归测试失败终止发布) return # 6. 影子测试 shadow_test(new_model, duration_hours24) # 7. 灰度发布 gradual_rollout(new_model)7. 常见问题与排查技巧实录7.1 训练相关的高频问题问题一Loss不下降或震荡严重排查思路先看学习率是不是太大用学习率扫描找合适值。再看数据有没有问题比如标签错了、特征没归一化。最后看模型结构是不是太深或太浅。我踩过的坑有一次loss一直震荡调了一周参数都没用最后发现是数据里混了一批错误标注的样本。所以遇到问题先查数据再查模型。问题二过拟合排查思路看训练loss和验证loss的差距。差距大就是过拟合。解决方法加正则化、加dropout、加数据增强、减小模型。问题三显存不够排查思路先看batch size能不能减。再看能不能用梯度累积。再看能不能用混合精度。最后看能不能用梯度检查点。# 梯度累积示例 accumulation_steps 4 optimizer.zero_grad() for i, batch in enumerate(dataloader): output model(batch) loss criterion(output, batch[label]) loss loss / accumulation_steps # 归一化 loss.backward() if (i 1) % accumulation_steps 0: optimizer.step() optimizer.zero_grad()7.2 部署相关的高频问题问题一推理延迟高排查思路先看是不是模型太大考虑量化或蒸馏。再看是不是批处理没开开批处理。再看是不是CPU推理换GPU。最后看是不是预处理太慢优化预处理。问题二内存泄漏排查思路用memory_profiler定位泄漏点。常见原因是全局变量累积、缓存没清理、循环引用。问题三服务不稳定排查思路看日志、看监控、看资源使用。常见原因是并发太高、超时设置不合理、依赖服务挂了。7.3 问题速查表问题现象可能原因排查方法解决方案Loss不下降学习率过大/数据问题学习率扫描/数据检查调小学习率/清洗数据过拟合模型太复杂/数据太少对比训练验证loss加正则/加数据显存不够batch太大/模型太大nvidia-smi监控减batch/梯度累积/混合精度推理延迟高模型大/没批处理分阶段计时量化/批处理/换引擎内存泄漏全局变量/缓存memory_profiler清理全局变量/加缓存淘汰数据漂移线上数据分布变了PSI/KS检验重新训练/加漂移检测最后分享一个小技巧遇到任何问题先复现再定位最后解决。不要一上来就改代码那样只会引入新问题。复现的意思是用最小的例子稳定重现问题。定位的意思是用二分法或日志逐步缩小问题范围。解决的意思是改完之后要验证问题真的解决了而不是碰巧不出现了。这套AI工程体系我从零搭过三次每次都有新的体会。第一次搭的时候追求技术先进什么新用什么结果运维成本高到离谱。第二次搭的时候追求简单结果扩展性不够业务一增长就要重构。第三次才找到平衡点够用就好留好扩展接口。技术选型没有绝对的对错只有适不适合当下的场景。希望这些经验能帮你少走一些弯路。
1. 从零搭建AI工程体系为什么我劝你别一上来就调包“ai-engineering-from-scratch”这个标题第一次看到的时候我愣了一下。市面上讲AI的教程铺天盖地但绝大多数都是教你import torch然后跑一个预训练模型或者调个API就完事。真正讲“从零开始搭建AI工程体系”的内容少得可怜。我自己在这个行业摸爬滚打了十来年带过不少新人也面试过几百个号称“会AI”的候选人。一个很普遍的现象是很多人能背出Transformer的结构图能说清楚Attention的计算公式但你让他从零搭一个能用的训练流水线他立刻卡壳。数据怎么组织显存怎么优化训练中断了怎么恢复模型上线后怎么监控这些问题调包教程里不会讲但实际工作中天天遇到。所以这个项目标题吸引我的地方在于它瞄准的是一个真实存在的巨大缺口从“会调包”到“能工程化落地”之间的鸿沟。这篇文章我想基于自己这些年的实际经验把这个标题背后的核心内容拆开揉碎讲清楚一个完整的AI工程体系到底包含哪些环节每个环节的关键决策怎么做以及那些只有踩过坑才知道的细节。适合谁看如果你是刚入行的算法工程师正在从“跑通demo”向“能交付项目”过渡这篇文章能帮你少走至少半年的弯路。如果你是有经验的开发者想转AI方向这里面的工程思路对你来说反而是优势。如果你只是好奇AI工程到底在干什么也能从中看到一个真实的行业切面。2. 整体架构设计AI工程体系到底包含什么2.1 为什么“从零”不等于“从轮子造起”先澄清一个概念。“ai-engineering-from-scratch”里的“from scratch”不是让你用NumPy手写一个Transformer——那是深度学习框架开发者干的事。这里的“从零”指的是从零搭建一套完整的、可运行的、能支撑实际业务的AI工程体系。打个比方。你要开一家餐厅“从零”不是让你去种小麦、养牛、炼铁锅而是让你从选址、菜单设计、供应链搭建、厨房动线规划、人员分工、品控流程这一整套东西开始搭。AI工程也是一样PyTorch、TensorFlow这些框架就是你的“厨具”你不需要自己造但你需要知道怎么选、怎么用、怎么把它们组织成一个高效运转的系统。一个完整的AI工程体系我习惯把它拆成六个层次数据层数据的采集、清洗、标注、版本管理、特征存储训练层实验管理、分布式训练、超参调优、模型检查点评估层离线评估、在线A/B测试、回归测试、偏差检测部署层模型压缩、推理优化、服务化、灰度发布监控层性能监控、数据漂移检测、告警机制迭代层反馈闭环、持续训练、版本回滚这六层缺一不可。很多团队的问题就在于只关注训练层觉得模型训出来就万事大吉了结果上线后各种问题层出不穷。2.2 技术选型的核心逻辑场景驱动而非技术驱动在搭建这套体系之前最重要的一步是明确你的场景需求。我见过太多团队一上来就选最时髦的技术栈结果发现根本不匹配自己的业务。选型的时候我一般会问自己四个问题第一个问题数据量级和更新频率是多少如果你的数据是TB级别的每天新增几十万条那你的数据管道必须支持流式处理特征存储要能支撑高并发读写。如果数据量不大一周更新一次那用简单的批处理脚本就够了没必要上KafkaSpark那一套。第二个问题训练频率和延迟要求是什么一天训一次模型和一个月训一次对基础设施的要求完全不同。前者需要自动化的训练流水线后者手动触发就行。推理延迟要求是10ms还是1s直接决定了你要不要做模型量化、蒸馏、TensorRT优化。第三个问题团队的技术储备如何如果团队里没人懂Kubernetes你硬上K8s做模型部署运维成本会高到离谱。选型要匹配团队能力而不是匹配招聘JD。第四个问题预算约束是什么训练用A100还是3090推理用T4还是CPU这些都要算账。我后面会给出具体的成本估算方法。基于这四个问题的答案你的技术栈选型基本就确定了。下面这张表是我总结的常见场景对应的推荐方案场景特征数据层训练层部署层监控层小数据量、低频更新文件系统SQLite单机GPUFlask API日志文件中等数据量、日更新PostgreSQLMinIO单机多卡FastAPITritonPrometheusGrafana大数据量、实时更新KafkaFeature Store分布式训练K8sTriton全链路监控边缘部署本地缓存云端训练ONNX Runtime端侧日志回传注意这张表是起点不是终点。实际选型一定要结合具体业务做调整不要照搬。2.3 目录结构设计让项目可维护的第一步一个AI工程项目最容易烂掉的地方就是目录结构。我接手过很多“祖传代码”打开一看根目录下几十个文件混在一起train.py、train_v2.py、train_final.py、train_final_真的最终版.py找代码全靠搜索。从项目第一天起就定好目录结构后面能省无数时间。我推荐的结构是这样的project/ ├── configs/ # 配置文件 │ ├── base.yaml │ ├── train.yaml │ └── deploy.yaml ├── data/ # 数据相关 │ ├── raw/ # 原始数据 │ ├── processed/ # 处理后数据 │ └── schemas/ # 数据schema定义 ├── src/ # 源代码 │ ├── data/ # 数据处理模块 │ ├── models/ # 模型定义 │ ├── training/ # 训练逻辑 │ ├── evaluation/ # 评估逻辑 │ └── serving/ # 服务化代码 ├── experiments/ # 实验记录 ├── notebooks/ # 探索性分析 ├── tests/ # 测试代码 ├── scripts/ # 运维脚本 └── docs/ # 文档这个结构的关键在于关注点分离。配置和代码分离数据和代码分离训练和服务分离。这样做的好处是当你要改一个超参数时不需要动代码当你要换一个模型时不需要动数据管道当你要部署时不需要把训练代码也打包进去。3. 数据管道搭建AI工程的地基3.1 数据版本管理别再用文件名区分了数据版本管理是AI工程里最容易被忽视的环节。我见过太多团队用data_v1.csv、data_v2_fixed.csv这种方式管理数据结果三个月后没人说得清哪个版本对应哪次实验。正确的做法是引入数据版本控制工具。DVCData Version Control是我用得最顺手的它和Git无缝集成大文件存在远端存储Git里只存元数据。基本用法# 初始化DVC dvc init # 添加数据文件到DVC管理 dvc add data/raw/train.csv # 提交DVC元数据到Git git add data/raw/train.csv.dvc data/raw/.gitignore git commit -m add training data v1 # 推送数据到远端存储 dvc remote add -d storage s3://my-bucket/dvc-store dvc push这样每次实验都能精确对应到数据版本。当你发现三个月前那个效果特别好的模型时能一键还原当时的数据。实操心得DVC的缓存机制会占用大量本地磁盘建议定期执行dvc gc清理不再需要的缓存。另外远端存储建议用对象存储而不是NFS性能和可靠性都好很多。3.2 数据清洗的工程化方法数据清洗不是写个Jupyter Notebook跑一遍就完事它需要工程化。核心原则是清洗逻辑要可复用、可测试、可追溯。我一般把清洗分成三个阶段阶段一数据探查。用Pandas Profiling或Great Expectations这类工具自动生成数据质量报告。重点看缺失值比例、异常值分布、类别不平衡程度。这一步在Notebook里做就行目的是了解数据长什么样。阶段二清洗规则定义。把探查阶段发现的每个问题转化成一条明确的规则。比如“age字段不能为负数”、“category字段只能是A/B/C三种值”。这些规则用代码写出来放在src/data/validators.py里。阶段三清洗流水线。把规则串成流水线用Airflow或Prefect这类工具调度。每次新数据进来自动跑一遍清洗输出清洗报告。# src/data/cleaners.py import pandas as pd from typing import Dict, Any class DataCleaner: def __init__(self, config: Dict[str, Any]): self.config config def remove_duplicates(self, df: pd.DataFrame) - pd.DataFrame: 去重保留第一条 before len(df) df df.drop_duplicates(subsetself.config[dedup_keys]) after len(df) print(f去重: {before} - {after}, 移除 {before - after} 条) return df def handle_missing(self, df: pd.DataFrame) - pd.DataFrame: 处理缺失值 for col, strategy in self.config[missing_strategy].items(): if strategy drop: df df.dropna(subset[col]) elif strategy mean: df[col] df[col].fillna(df[col].mean()) elif strategy median: df[col] df[col].fillna(df[col].median()) elif strategy.startswith(constant:): value strategy.split(:)[1] df[col] df[col].fillna(value) return df def clip_outliers(self, df: pd.DataFrame) - pd.DataFrame: 截断异常值 for col, (lower, upper) in self.config[clip_ranges].items(): df[col] df[col].clip(lower, upper) return df注意清洗规则一定要写单元测试。我踩过的坑是某次改了清洗逻辑导致某个字段的分布完全变了但因为没有测试直到模型效果暴跌才发现。现在我的习惯是每条清洗规则至少配一个测试用例。3.3 特征存储从混乱到有序当项目从单模型发展到多模型时特征管理就会变成噩梦。同一个特征推荐模型算一遍风控模型又算一遍两边逻辑还不一致。这时候就需要特征存储Feature Store。特征存储的核心价值是特征复用和一致性保证。训练时用的特征和推理时用的特征必须来自同一套计算逻辑否则会出现训练-服务偏差Training-Serving Skew。我推荐用Feast作为特征存储方案它轻量、开源、和主流框架集成好。基本架构是离线存储存历史特征用于训练。一般用Parquet文件或数据仓库。在线存储存最新特征用于推理。一般用Redis或DynamoDB。特征定义用Python代码定义特征的计算逻辑保证线上线下一致。# feature_repo/features.py from feast import Entity, Feature, FeatureView, FileSource from feast.types import Float32, Int64 from datetime import timedelta # 定义实体 user Entity(nameuser_id, join_keys[user_id]) # 定义数据源 user_stats_source FileSource( pathdata/user_stats.parquet, timestamp_fieldevent_timestamp, ) # 定义特征视图 user_stats_fv FeatureView( nameuser_stats, entities[user], ttltimedelta(days7), schema[ Feature(nametotal_orders, dtypeInt64), Feature(nameavg_order_value, dtypeFloat32), Feature(namedays_since_last_order, dtypeInt64), ], sourceuser_stats_source, )这样定义之后训练时用get_historical_features拿历史特征推理时用get_online_features拿实时特征两边逻辑完全一致。4. 训练流水线从手动脚本到自动化系统4.1 实验管理让每次实验都可追溯没有实验管理的团队工作状态是这样的小王跑了个模型准确率92%很高兴地跟组长汇报。组长问“用的什么超参数”小王说“等我翻一下Notebook。”翻了半小时找到了但发现那个Notebook后来又被改过已经复现不出来了。实验管理要解决的就是这个问题。核心要求是任何一次实验结果都能精确复现。这意味着要记录代码版本、数据版本、超参数、环境依赖、随机种子、硬件信息。我常用的方案是MLflow它轻量、易用、功能全。基本用法import mlflow import mlflow.pytorch # 设置实验 mlflow.set_experiment(my-ai-project) with mlflow.start_run(run_namebaseline-v1): # 记录超参数 mlflow.log_params({ learning_rate: 1e-4, batch_size: 32, epochs: 50, optimizer: adamw, }) # 记录指标 for epoch in range(50): train_loss train_one_epoch() val_loss, val_acc validate() mlflow.log_metrics({ train_loss: train_loss, val_loss: val_loss, val_acc: val_acc, }, stepepoch) # 记录模型 mlflow.pytorch.log_model(model, model) # 记录数据版本 mlflow.log_param(data_version, v1.2.0) mlflow.log_param(git_commit, get_git_commit())实操心得MLflow的Tracking Server建议单独部署不要用本地文件存储。团队协作时所有人都往同一个Server打点才能对比实验。另外模型文件不要直接存MLflow存到对象存储MLflow里只存路径。4.2 分布式训练什么时候需要怎么选分布式训练不是银弹。很多场景下单卡就够用硬上分布式反而增加复杂度和故障率。我的判断标准是单卡能训完且时间可接受比如8小时内单卡单卡训不完或时间太长上分布式模型大到单卡放不下必须分布式分布式训练有三种主流方案数据并行Data Parallelism每个GPU持有完整模型副本数据切分到各GPU梯度汇总后更新。适合模型能放进单卡显存、但数据量大的场景。PyTorch的DDP是标准方案。模型并行Model Parallelism模型切分到多个GPU每个GPU持有部分层。适合模型大到单卡放不下的场景。实现复杂一般用Megatron-LM或DeepSpeed。流水线并行Pipeline Parallelism模型按层切分成多个阶段不同阶段在不同GPU上数据像流水线一样流过。适合超大规模模型。PyTorch的Pipe是入门方案。实际项目中最常见的是数据并行混合精度。下面是一个DDP的启动脚本# 单机8卡训练 torchrun --nproc_per_node8 \ --master_port29500 \ src/training/train.py \ --config configs/train.yaml \ --batch_size 64 \ --learning_rate 1e-4# src/training/train.py 关键部分 import torch.distributed as dist from torch.nn.parallel import DistributedDataParallel as DDP def setup_ddp(): dist.init_process_group(backendnccl) local_rank int(os.environ[LOCAL_RANK]) torch.cuda.set_device(local_rank) return local_rank def main(): local_rank setup_ddp() model MyModel().to(local_rank) model DDP(model, device_ids[local_rank]) # 数据加载器要用DistributedSampler sampler DistributedSampler(dataset) dataloader DataLoader(dataset, samplersampler, batch_size...) for epoch in range(epochs): sampler.set_epoch(epoch) # 重要每个epoch要重新shuffle for batch in dataloader: ...注意DDP下只有rank 0的进程应该做日志记录和模型保存否则会重复写文件导致冲突。另外sampler.set_epoch(epoch)这行很容易漏漏了的话每个epoch的数据顺序都一样影响训练效果。4.3 超参数调优别再用网格搜索了超参数调优是很多团队的痛点。网格搜索太慢随机搜索碰运气手动调参靠经验。我推荐用贝叶斯优化工具选Optuna。Optuna的核心优势是用更少的试验次数找到更好的参数。它通过建立代理模型预测哪些参数组合可能更好从而指导下一步搜索方向。import optuna def objective(trial): # 定义搜索空间 lr trial.suggest_float(lr, 1e-5, 1e-2, logTrue) batch_size trial.suggest_categorical(batch_size, [16, 32, 64, 128]) dropout trial.suggest_float(dropout, 0.1, 0.5) num_layers trial.suggest_int(num_layers, 2, 8) # 训练模型 model build_model(num_layers, dropout) val_acc train_and_evaluate(model, lr, batch_size) return val_acc # 创建study study optuna.create_study( directionmaximize, sampleroptuna.samplers.TPESampler(seed42), pruneroptuna.pruners.MedianPruner(n_startup_trials5), ) # 开始优化 study.optimize(objective, n_trials100, timeout3600*8) # 输出最佳参数 print(fBest trial: {study.best_trial.params})实操心得Optuna的Pruner功能非常有用。它会在训练过程中自动终止那些明显不好的试验节省大量算力。我一般设置n_startup_trials5让前5个试验完整跑完之后才开始剪枝。另外搜索空间的定义很关键范围不要设太宽否则大部分试验都在无效区域浪费算力。5. 模型部署与服务化让模型真正产生价值5.1 推理优化从100ms到10ms的实战路径模型训出来只是第一步能不能高效推理才是关键。我做过一个项目原始模型推理延迟120ms经过一系列优化后降到8ms提升了15倍。下面是我用的优化路径第一步模型量化。把FP32转成FP16或INT8显存占用减半推理速度提升1.5-2倍。PyTorch的torch.quantization或ONNX Runtime的量化工具都能做。# PyTorch动态量化 import torch.quantization model MyModel() model.eval() quantized_model torch.quantization.quantize_dynamic( model, {torch.nn.Linear}, dtypetorch.qint8 ) torch.save(quantized_model.state_dict(), model_quantized.pt)第二步算子融合。把多个连续的小算子合并成一个减少kernel launch开销。TensorRT和ONNX Runtime都有自动融合功能。第三步推理引擎优化。用TensorRT或ONNX Runtime替换原生PyTorch推理。这一步通常能带来2-3倍的提升。# ONNX Runtime推理 import onnxruntime as ort import numpy as np # 导出ONNX torch.onnx.export(model, dummy_input, model.onnx, opset_version13) # 创建推理会话 session ort.InferenceSession( model.onnx, providers[CUDAExecutionProvider], ) # 推理 input_name session.get_inputs()[0].name output session.run(None, {input_name: input_data})第四步批处理。把多个请求攒成一批一起推理GPU利用率能提升好几倍。但要注意延迟和吞吐的权衡。第五步缓存。对于重复的输入直接返回缓存结果。这个在推荐系统里特别有效热门物品的embedding可以缓存。优化手段延迟降低实现难度适用场景FP16量化1.5-2x低所有GPU场景INT8量化2-4x中对精度不敏感场景算子融合1.2-1.5x低所有场景TensorRT2-3x中NVIDIA GPU批处理3-5x中高吞吐场景缓存10x低重复输入多5.2 服务化框架选型FastAPI vs Triton vs TorchServe模型服务化框架的选择直接决定了你的运维复杂度。我三个都用过说说各自的适用场景。FastAPI最灵活适合自定义逻辑多的场景。你可以完全控制请求处理流程做各种预处理和后处理。缺点是性能一般高并发下需要自己优化。from fastapi import FastAPI import torch app FastAPI() model load_model() app.post(/predict) async def predict(request: PredictRequest): # 预处理 input_tensor preprocess(request.data) # 推理 with torch.no_grad(): output model(input_tensor) # 后处理 result postprocess(output) return {prediction: result}Triton Inference ServerNVIDIA出品性能最强支持多框架、多模型、动态批处理。适合大规模部署场景。缺点是需要写配置文件灵活性不如FastAPI。TorchServePyTorch官方方案和PyTorch生态集成好。适合纯PyTorch模型的部署。缺点是社区活跃度一般遇到问题不好找资料。我的建议是小规模用FastAPI大规模用Triton纯PyTorch生态用TorchServe。不要一上来就上Triton它的学习曲线不陡但也不平小项目用它是杀鸡用牛刀。5.3 灰度发布与回滚上线不是终点模型上线最怕的是什么是新模型效果不如旧模型但你已经全量发布了。所以灰度发布是必须的。我的做法是分三个阶段阶段一影子模式Shadow Mode。新模型和旧模型同时运行但只有旧模型的结果返回给用户。新模型的结果只记录不生效。这个阶段跑一周对比两个模型的输出差异。阶段二小流量灰度。把1%的流量切给新模型观察核心指标。如果指标没有明显下降逐步扩大到5%、10%、50%。阶段三全量发布。灰度到100%后旧模型保留一周作为回滚备份确认没问题后再下线。# 灰度发布逻辑 import random def route_request(user_id: str) - str: 根据用户ID决定用哪个模型 # 用user_id的hash做分流保证同一用户始终用同一模型 bucket hash(user_id) % 100 if bucket current_gray_ratio: return new_model else: return old_model注意分流要用user_id的hash不要用随机数。否则同一用户一会儿用新模型一会儿用旧模型体验会非常割裂。另外灰度比例要动态可调不要写死在代码里。6. 监控与迭代让系统持续进化6.1 数据漂移检测模型效果下降的预警器模型上线后效果下降最常见的原因是数据漂移。输入数据的分布变了模型还按老规律预测自然就不准了。数据漂移检测的核心是对比训练数据和线上数据的分布。常用的方法有PSIPopulation Stability Index衡量两个分布的差异PSI0.1表示稳定0.1-0.25表示轻微漂移0.25表示显著漂移。KS检验统计两个分布是否有显著差异。KL散度衡量两个概率分布的差异。import numpy as np from scipy import stats def calculate_psi(expected, actual, buckets10): 计算PSI # 分桶 breakpoints np.percentile(expected, np.linspace(0, 100, buckets 1)) breakpoints[0] -np.inf breakpoints[-1] np.inf expected_percents np.histogram(expected, breakpoints)[0] / len(expected) actual_percents np.histogram(actual, breakpoints)[0] / len(actual) # 避免除零 expected_percents np.clip(expected_percents, 1e-6, None) actual_percents np.clip(actual_percents, 1e-6, None) psi np.sum((actual_percents - expected_percents) * np.log(actual_percents / expected_percents)) return psi # 使用 psi calculate_psi(train_data[feature_1], online_data[feature_1]) if psi 0.25: alert(特征feature_1发生显著漂移PSI{:.3f}.format(psi))实操心得PSI对分桶数量敏感一般用10-20个桶。另外不要对所有特征都设同样的阈值有些特征天然波动大阈值要放宽。我一般先跑一周基线看每个特征PSI的正常波动范围再定阈值。6.2 模型性能监控不只是准确率模型上线后准确率不是唯一要看的指标。我一般会监控四类指标业务指标点击率、转化率、GMV等。这是最终目标但反馈慢不能作为唯一依据。模型指标准确率、召回率、AUC等。需要标注数据才能算有延迟。系统指标QPS、延迟、错误率、GPU利用率。实时性强能快速发现问题。数据指标特征分布、缺失率、异常值比例。能提前预警数据问题。这四类指标要放在同一个Dashboard上看才能快速定位问题。比如业务指标下降同时数据指标显示某个特征漂移了那大概率是数据问题。6.3 持续训练让模型跟上变化模型不是训一次就完事需要持续迭代。持续训练的核心是自动化和安全。自动化指的是数据更新后自动触发训练训练完成后自动评估评估通过后自动部署。这个流水线用Airflow或Kubeflow Pipelines搭建。安全指的是新模型必须经过严格评估才能上线。我一般设三道关卡离线评估在留出集上新模型的核心指标必须不低于旧模型的95%。回归测试在一组固定的测试用例上新模型的输出不能有异常变化。影子测试新模型和旧模型并行跑24小时输出差异在可接受范围内。三道关卡都通过才允许灰度发布。# 持续训练流水线伪代码 def continuous_training_pipeline(): # 1. 检查数据更新 if not new_data_available(): return # 2. 数据清洗和验证 cleaned_data clean_and_validate(new_data) # 3. 训练新模型 new_model train(cleaned_data) # 4. 离线评估 metrics evaluate(new_model, test_set) if metrics[auc] baseline[auc] * 0.95: alert(新模型效果不达标终止发布) return # 5. 回归测试 if not regression_test(new_model): alert(回归测试失败终止发布) return # 6. 影子测试 shadow_test(new_model, duration_hours24) # 7. 灰度发布 gradual_rollout(new_model)7. 常见问题与排查技巧实录7.1 训练相关的高频问题问题一Loss不下降或震荡严重排查思路先看学习率是不是太大用学习率扫描找合适值。再看数据有没有问题比如标签错了、特征没归一化。最后看模型结构是不是太深或太浅。我踩过的坑有一次loss一直震荡调了一周参数都没用最后发现是数据里混了一批错误标注的样本。所以遇到问题先查数据再查模型。问题二过拟合排查思路看训练loss和验证loss的差距。差距大就是过拟合。解决方法加正则化、加dropout、加数据增强、减小模型。问题三显存不够排查思路先看batch size能不能减。再看能不能用梯度累积。再看能不能用混合精度。最后看能不能用梯度检查点。# 梯度累积示例 accumulation_steps 4 optimizer.zero_grad() for i, batch in enumerate(dataloader): output model(batch) loss criterion(output, batch[label]) loss loss / accumulation_steps # 归一化 loss.backward() if (i 1) % accumulation_steps 0: optimizer.step() optimizer.zero_grad()7.2 部署相关的高频问题问题一推理延迟高排查思路先看是不是模型太大考虑量化或蒸馏。再看是不是批处理没开开批处理。再看是不是CPU推理换GPU。最后看是不是预处理太慢优化预处理。问题二内存泄漏排查思路用memory_profiler定位泄漏点。常见原因是全局变量累积、缓存没清理、循环引用。问题三服务不稳定排查思路看日志、看监控、看资源使用。常见原因是并发太高、超时设置不合理、依赖服务挂了。7.3 问题速查表问题现象可能原因排查方法解决方案Loss不下降学习率过大/数据问题学习率扫描/数据检查调小学习率/清洗数据过拟合模型太复杂/数据太少对比训练验证loss加正则/加数据显存不够batch太大/模型太大nvidia-smi监控减batch/梯度累积/混合精度推理延迟高模型大/没批处理分阶段计时量化/批处理/换引擎内存泄漏全局变量/缓存memory_profiler清理全局变量/加缓存淘汰数据漂移线上数据分布变了PSI/KS检验重新训练/加漂移检测最后分享一个小技巧遇到任何问题先复现再定位最后解决。不要一上来就改代码那样只会引入新问题。复现的意思是用最小的例子稳定重现问题。定位的意思是用二分法或日志逐步缩小问题范围。解决的意思是改完之后要验证问题真的解决了而不是碰巧不出现了。这套AI工程体系我从零搭过三次每次都有新的体会。第一次搭的时候追求技术先进什么新用什么结果运维成本高到离谱。第二次搭的时候追求简单结果扩展性不够业务一增长就要重构。第三次才找到平衡点够用就好留好扩展接口。技术选型没有绝对的对错只有适不适合当下的场景。希望这些经验能帮你少走一些弯路。
